fix(event/stream): require real headroom and type the options error

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

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

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

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

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.