Merge pull request #11 from sharedjourney/fix/guard-pull-timeout-vs-client-timeout

fix(event/stream): reject a client timeout that cannot outlast PullTimeout

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.

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
+}

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))
+}

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)
 }