From 43bc40babd874d1554050bd0e1906d3085c0b3f2 Mon Sep 17 00:00:00 2001 From: "T. Tradesman" <184814242+ttradesman@users.noreply.github.com> Date: Thu, 23 Jul 2026 13:04:27 +0200 Subject: [PATCH 1/2] fix(event/stream): reject a client timeout that cannot outlast the pull MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit PullMessages is a long-poll: the camera holds the connection open for up to PullTimeout waiting for an event. http.Client.Timeout bounds the whole exchange — dial, write, wait-for-headers — and starts before the camera has parsed the request, so a client ceiling equal to or below PullTimeout expires first on every interval with no event. The failure mode is quiet and easy to misread. Pulls fail continuously, but the stream stays alive because ReconnectAfterFailures recreates the subscription, and each recreate makes the camera replay its full property state. Events keep arriving, in bursts, on the reconnect cadence rather than when they happen — so it reads as a slow camera rather than a misconfiguration. Observed in the field with both values at 5s: every pull timed out, recovery landed after exactly 3 failures, and ~90 property-state events were replayed every 18s. Validated in NewStream, before the subscription call, since the config can only fail. A zero client timeout stays legal — unbounded is safe because the pull loop is already bounded by ctx. --- event/stream/pulltimeout.go | 42 ++++++++++++++++++++++++++++++ event/stream/pulltimeout_test.go | 44 ++++++++++++++++++++++++++++++++ event/stream/stream.go | 6 +++++ 3 files changed, 92 insertions(+) create mode 100644 event/stream/pulltimeout.go create mode 100644 event/stream/pulltimeout_test.go diff --git a/event/stream/pulltimeout.go b/event/stream/pulltimeout.go new file mode 100644 index 00000000..82be75f2 --- /dev/null +++ b/event/stream/pulltimeout.go @@ -0,0 +1,42 @@ +package stream + +import ( + "fmt" + "time" + + "github.com/kerberos-io/onvif" +) + +// validateClientTimeout rejects an HTTP client ceiling that cannot +// outlast the PullMessages long-poll. +// +// 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. +func validateClientTimeout(clientTimeout, pullTimeout time.Duration) error { + if clientTimeout == 0 || clientTimeout > pullTimeout { + 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) +} + +// clientTimeoutOf reports the device's HTTP client ceiling, or 0 when +// the SDK is using its own default (unbounded) client. +func clientTimeoutOf(dev *onvif.Device) time.Duration { + if dev == nil { + return 0 + } + c := dev.GetDeviceParams().HttpClient + if c == nil { + return 0 + } + return c.Timeout +} diff --git a/event/stream/pulltimeout_test.go b/event/stream/pulltimeout_test.go new file mode 100644 index 00000000..4a406912 --- /dev/null +++ b/event/stream/pulltimeout_test.go @@ -0,0 +1,44 @@ +package stream + +import ( + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// 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 +// 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. +func TestValidateClientTimeout(t *testing.T) { + tests := []struct { + name string + client time.Duration + pull time.Duration + wantErr bool + }{ + {"unbounded client is fine", 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}, + {"equal timeouts always lose", 5 * time.Second, 5 * time.Second, true}, + {"client below pull", 4 * time.Second, 30 * time.Second, true}, + } + + for _, tt := range tests { + t.Run(tt.name, func(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.Contains(t, err.Error(), "PullTimeout", + "the error must name the option the caller has to change") + return + } + assert.NoError(t, err, "client=%s pull=%s must be accepted", tt.client, tt.pull) + }) + } +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 9dac55b2..0eb2760a 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -234,6 +234,12 @@ func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) { // // The returned Stream stops when ctx is cancelled or Close is called. func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, error) { + // Checked before the subscription call: this config can only fail, + // so surfacing it here beats a stream that appears to work and + // silently survives on reconnects alone. + if err := validateClientTimeout(clientTimeoutOf(dev), opts.withDefaults().PullTimeout); err != nil { + return nil, err + } return newStream(ctx, deviceCaller{dev: dev}, opts) } 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 2/2] 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 ebd05aaf..f9a7912a 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 82be75f2..03cfc55f 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 4a406912..e067851a 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 0eb2760a..697a5053 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.