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

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

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

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

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