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 new file mode 100644 index 0000000..03cfc55 --- /dev/null +++ b/event/stream/pulltimeout.go @@ -0,0 +1,51 @@ +package stream + +import ( + "errors" + "fmt" + "time" + + "github.com/kerberos-io/onvif" +) + +// 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. +// +// 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+minClientHeadroom { + return nil + } + return fmt.Errorf( + "%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 +// 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 0000000..e067851 --- /dev/null +++ b/event/stream/pulltimeout_test.go @@ -0,0 +1,68 @@ +package stream + +import ( + "errors" + "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 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 + client time.Duration + pull time.Duration + wantErr bool + }{ + {"unbounded client is the caller's risk, not an error", 0, 30 * time.Second, false}, + {"comfortable headroom", 40 * time.Second, 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}, + } + + 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.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 + } + assert.NoError(t, err, "client=%s pull=%s must be accepted", tt.client, tt.pull) + }) + } +} + +// 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 9dac55b..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. @@ -234,6 +236,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) }