From 3634cee483c6b398012c84502ca8e80679d171ac Mon Sep 17 00:00:00 2001 From: "T. Tradesman" <184814242+ttradesman@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:23:59 +0200 Subject: [PATCH] fix(event/stream): require real headroom and type the options error MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Two gaps in the previous commit's guard. Strict inequality was not enough. A client timeout one millisecond above PullTimeout passed, and the test pinned that as valid — but the client ceiling also has to cover dial, TLS and the response transfer on top of the poll it outlasts, which on a cellular bearer is hundreds of milliseconds. Require minClientHeadroom (5s) above PullTimeout. The error was a bare fmt.Errorf, so callers could not tell it from the transient pull/renew/recreate failures they retry. A consumer that retries this one loops forever on a configuration that can never succeed. ErrInvalidOptions is a sentinel they can short-circuit on. Zero stays accepted: it is the SDK's default when a caller passes no client, so rejecting it would break every default consumer. The comment no longer claims that is safe — the caller interface documents that ctx cannot interrupt an in-flight SOAP call, so an unbounded client is the one case nothing can unwedge. --- event/stream/doc.go | 3 ++- event/stream/pulltimeout.go | 37 ++++++++++++++++++++------------ event/stream/pulltimeout_test.go | 30 +++++++++++++++++++++++--- event/stream/stream.go | 4 +++- 4 files changed, 55 insertions(+), 19 deletions(-) diff --git a/event/stream/doc.go b/event/stream/doc.go index ebd05aa..f9a7912 100644 --- a/event/stream/doc.go +++ b/event/stream/doc.go @@ -18,7 +18,8 @@ // } // // NewStream performs network I/O so auth and reachability failures -// surface synchronously. Events and Errors close when the Stream stops; +// surface synchronously, and rejects a client timeout that cannot +// outlast PullTimeout with ErrInvalidOptions. Events and Errors close when the Stream stops; // Errors sends are non-blocking so a stalled consumer drops older // errors rather than blocking the pull loop. After a silent reconnect, // the next batch's events carry Event.AfterReconnect=true. diff --git a/event/stream/pulltimeout.go b/event/stream/pulltimeout.go index 82be75f..03cfc55 100644 --- a/event/stream/pulltimeout.go +++ b/event/stream/pulltimeout.go @@ -1,31 +1,40 @@ package stream import ( + "errors" "fmt" "time" "github.com/kerberos-io/onvif" ) -// validateClientTimeout rejects an HTTP client ceiling that cannot -// outlast the PullMessages long-poll. +// ErrInvalidOptions marks a configuration that cannot succeed. Callers +// retry the pull/renew/recreate errors; retrying this one never helps, +// so it is a distinct sentinel they can short-circuit on. +var ErrInvalidOptions = errors.New("stream: invalid options") + +// minClientHeadroom is how far http.Client.Timeout must exceed +// PullTimeout. The client ceiling covers dial, TLS and the response +// transfer on top of the poll it has to outlast, and starts before the +// camera has parsed the request; on a cellular bearer that overhead +// runs to hundreds of milliseconds. +const minClientHeadroom = 5 * time.Second + +// validateClientTimeout rejects a client ceiling that cannot outlast +// the PullMessages long-poll plus minClientHeadroom. // -// PullMessages asks the camera to hold the connection open for up to -// PullTimeout. http.Client.Timeout bounds the entire exchange — dial, -// write, and the wait for response headers — and starts before the -// camera has parsed the request, so it always expires first when the -// two are equal. The pull then fails on every interval with no event, -// and the subscription survives only by being recreated after -// ReconnectAfterFailures, which replays the camera's whole property -// state each time. A zero client timeout means unbounded, which is safe -// here because the pull loop is already bounded by ctx. +// Zero means unbounded and is accepted: it is the SDK's default when a +// caller passes no client, so rejecting it would break every default +// consumer. Note it is not risk-free — the caller interface documents +// that ctx cannot interrupt an in-flight SOAP call, so only the client +// timeout can unwedge a stalled camera. func validateClientTimeout(clientTimeout, pullTimeout time.Duration) error { - if clientTimeout == 0 || clientTimeout > pullTimeout { + if clientTimeout == 0 || clientTimeout >= pullTimeout+minClientHeadroom { return nil } return fmt.Errorf( - "http.Client.Timeout (%s) must exceed PullTimeout (%s): PullMessages is a long-poll and the client would abort every quiet pull; raise the client timeout above PullTimeout or leave it zero", - clientTimeout, pullTimeout) + "%w: http.Client.Timeout (%s) must exceed PullTimeout (%s) by at least %s; PullMessages is a long-poll and the client would abort every quiet pull", + ErrInvalidOptions, clientTimeout, pullTimeout, minClientHeadroom) } // clientTimeoutOf reports the device's HTTP client ceiling, or 0 when diff --git a/event/stream/pulltimeout_test.go b/event/stream/pulltimeout_test.go index 4a40691..e067851 100644 --- a/event/stream/pulltimeout_test.go +++ b/event/stream/pulltimeout_test.go @@ -1,6 +1,7 @@ package stream import ( + "errors" "testing" "time" @@ -11,10 +12,14 @@ import ( // TestValidateClientTimeout — PullMessages is a long-poll: the camera // holds the connection open for PullTimeout waiting for an event. An // http.Client.Timeout covers the whole exchange and starts before the -// camera has even parsed the request, so a client ceiling at or below +// camera has parsed the request, so a client ceiling at or below // PullTimeout loses the race on every quiet interval and the pull can // only ever fail. This shipped once (both were 5s) and presented as a // slow camera rather than a misconfiguration. +// +// Strict inequality is not enough: the client also has to cover dial, +// TLS and the response transfer, which on a cellular bearer runs to +// hundreds of milliseconds. Hence a real headroom floor. func TestValidateClientTimeout(t *testing.T) { tests := []struct { name string @@ -22,9 +27,11 @@ func TestValidateClientTimeout(t *testing.T) { pull time.Duration wantErr bool }{ - {"unbounded client is fine", 0, 30 * time.Second, false}, + {"unbounded client is the caller's risk, not an error", 0, 30 * time.Second, false}, {"comfortable headroom", 40 * time.Second, 30 * time.Second, false}, - {"strictly greater is accepted", 30*time.Second + time.Millisecond, 30 * time.Second, false}, + {"exactly the minimum headroom", 30*time.Second + minClientHeadroom, 30 * time.Second, false}, + {"a hair under the minimum headroom", 30*time.Second + minClientHeadroom - time.Millisecond, 30 * time.Second, true}, + {"strictly greater but no headroom", 30*time.Second + time.Millisecond, 30 * time.Second, true}, {"equal timeouts always lose", 5 * time.Second, 5 * time.Second, true}, {"client below pull", 4 * time.Second, 30 * time.Second, true}, } @@ -34,6 +41,8 @@ func TestValidateClientTimeout(t *testing.T) { err := validateClientTimeout(tt.client, tt.pull) if tt.wantErr { require.Error(t, err, "client=%s pull=%s must be rejected", tt.client, tt.pull) + assert.ErrorIs(t, err, ErrInvalidOptions, + "callers need a sentinel to tell a permanent misconfiguration from a transient failure") assert.Contains(t, err.Error(), "PullTimeout", "the error must name the option the caller has to change") return @@ -42,3 +51,18 @@ func TestValidateClientTimeout(t *testing.T) { }) } } + +// TestErrInvalidOptions_IsDistinctFromStreamErrors — the pull/renew/ +// recreate errors are transient and callers retry them. A bad Options +// never becomes valid by retrying, so it must not be mistaken for one. +func TestErrInvalidOptions_IsDistinctFromStreamErrors(t *testing.T) { + err := validateClientTimeout(5*time.Second, 5*time.Second) + require.Error(t, err) + + var pull ErrPullFailed + var renew ErrRenewFailed + var recreate ErrRecreateFailed + assert.False(t, errors.As(err, &pull)) + assert.False(t, errors.As(err, &renew)) + assert.False(t, errors.As(err, &recreate)) +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 0eb2760..697a505 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -39,7 +39,9 @@ type Options struct { // server-side filtering is fragile across vendors and empty is // required for AXIS. RawTopicFilter string - // PullTimeout — zero means default (5s). + // PullTimeout — zero means default (5s). The device's + // http.Client.Timeout must exceed this by minClientHeadroom or + // NewStream returns ErrInvalidOptions. PullTimeout time.Duration // MessageLimit — zero means default (32). Busy AXIS cameras with // many configured rules can burst beyond 10 per pull.