fix(event/stream): scale the close drain bound with PullTimeout

event/stream/stream.go

@@ -10,13 +10,13 @@ import (
 	"github.com/kerberos-io/onvif"
 )
 
-// closeDrainTimeout bounds Close's wait for the pull and renew
-// goroutines to exit. The loops block in caller.SendSoap which is not
+// closeDrainSlack is how far the default drain bound sits above
+// PullTimeout. The loops block in caller.SendSoap which is not
 // ctx-aware (the underlying http.Client is the only thing that can
-// unblock them — see caller below). On a hung HTTP transport Close
-// would otherwise wait forever; instead it returns an error and lets
-// the calling agent move on.
-const closeDrainTimeout = 5 * time.Second
+// unblock them — see caller below), so a drain shorter than a pull
+// times out on every shutdown that lands mid-poll and skips the
+// Unsubscribe below it.
+const closeDrainSlack = 10 * time.Second
 
 // closeUnsubscribeTimeout bounds the Unsubscribe SOAP call issued by
 // Close. A subscription expires at the camera once InitialTermination
@@ -64,6 +64,13 @@ type Options struct {
 	RetryBackoff time.Duration
 	// BufferSize — zero means default (16); use -1 for unbuffered.
 	BufferSize int
+	// CloseDrainTimeout bounds Close's wait for the pull and renew
+	// loops to exit. Zero derives it from whichever is longer, the
+	// resolved PullTimeout or the device's http.Client.Timeout, plus
+	// closeDrainSlack — so it outlasts any call those loops can be
+	// parked in. Only NewStream can see the client ceiling; callers
+	// reaching newStream directly get the PullTimeout-only bound.
+	CloseDrainTimeout time.Duration
 }
 
 func defaultOptions() Options {
@@ -107,6 +114,12 @@ func (o Options) withDefaults() Options {
 	d.DeviceID = o.DeviceID
 	d.RawTopicFilter = o.RawTopicFilter
 	d.DisableReconnect = o.DisableReconnect
+	// Derived last: it depends on the resolved PullTimeout.
+	if o.CloseDrainTimeout > 0 {
+		d.CloseDrainTimeout = o.CloseDrainTimeout
+	} else {
+		d.CloseDrainTimeout = d.PullTimeout + closeDrainSlack
+	}
 	return d
 }
 
@@ -135,7 +148,7 @@ type subscriptionRef struct {
 //   - Enforce a per-request timeout via the underlying HTTP client.
 //     The methods do not take a ctx, so ctx-cancel cannot interrupt a
 //     hung request; only the HTTP client's own timeout can. Close
-//     bounds its drain wait at closeDrainTimeout to survive a misbehaving
+//     bounds its drain wait at Options.CloseDrainTimeout to survive a misbehaving
 //     caller, but a leaking goroutine remains until the HTTP call
 //     eventually returns.
 type caller interface {
@@ -234,9 +247,40 @@ 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) {
+	// Derived here rather than in withDefaults because only this entry
+	// point can see the device's HTTP ceiling.
+	if opts.CloseDrainTimeout == 0 {
+		opts.CloseDrainTimeout = drainFor(opts.withDefaults().PullTimeout, deviceClientTimeout(dev))
+	}
 	return newStream(ctx, deviceCaller{dev: dev}, opts)
 }
 
+// drainFor bounds Close's wait by the longest a SOAP call can run.
+// PullTimeout is how long the camera holds a poll, but the client
+// ceiling is what caps the call and callers set it higher. A zero
+// ceiling means unbounded, where no finite drain helps, so the poll
+// stays the best available bound.
+func drainFor(pullTimeout, clientTimeout time.Duration) time.Duration {
+	longest := pullTimeout
+	if clientTimeout > longest {
+		longest = clientTimeout
+	}
+	return longest + closeDrainSlack
+}
+
+// deviceClientTimeout reports the device's HTTP ceiling, or 0 when the
+// SDK's own default (unbounded) client is in use.
+func deviceClientTimeout(dev *onvif.Device) time.Duration {
+	if dev == nil {
+		return 0
+	}
+	c := dev.GetDeviceParams().HttpClient
+	if c == nil {
+		return 0
+	}
+	return c.Timeout
+}
+
 func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) {
 	opts = opts.withDefaults()
 	ref, err := createPullPoint(c, opts)
@@ -267,7 +311,7 @@ func (s *Stream) Events() <-chan Event { return s.events }
 // when the Stream stops.
 func (s *Stream) Errors() <-chan error { return s.errors }
 
-// Close stops the background goroutines, waits up to closeDrainTimeout
+// Close stops the background goroutines, waits up to Options.CloseDrainTimeout
 // for them to exit, and then Unsubscribes from the camera (also bounded,
 // by closeUnsubscribeTimeout). Subsequent calls are no-ops.
 //
@@ -281,8 +325,8 @@ func (s *Stream) Close() error {
 
 		select {
 		case <-s.done:
-		case <-time.After(closeDrainTimeout):
-			s.closeErr = fmt.Errorf("close: pull/renew loops did not drain within %s (likely stuck in caller HTTP)", closeDrainTimeout)
+		case <-time.After(s.opts.CloseDrainTimeout):
+			s.closeErr = fmt.Errorf("close: pull/renew loops did not drain within %s (likely stuck in caller HTTP)", s.opts.CloseDrainTimeout)
 			return
 		}
 

event/stream/stream_test.go

@@ -485,6 +485,10 @@ func TestClose_BoundedWhenLoopsStuckOnHungHTTP(t *testing.T) {
 	s, err := newStream(ctx, fc, Options{
 		PullTimeout:        100 * time.Millisecond,
 		InitialTermination: 30 * time.Second,
+		// Explicit so the bound stays short: the derived default is
+		// PullTimeout + closeDrainSlack, which would make this test
+		// sit for ten seconds.
+		CloseDrainTimeout: time.Second,
 	})
 	require.NoError(t, err)
 
@@ -504,9 +508,67 @@ func TestClose_BoundedWhenLoopsStuckOnHungHTTP(t *testing.T) {
 
 	require.Error(t, err)
 	assert.Contains(t, err.Error(), "drain", "expected a drain-timeout error")
-	// Total budget is closeDrainTimeout for the wait + ~0 for unsubscribe
+	// Total budget is CloseDrainTimeout for the wait + ~0 for unsubscribe
 	// (which is skipped when drain times out). Give plenty of slack for
 	// scheduling on a loaded CI machine.
-	assert.Less(t, elapsed, closeDrainTimeout+2*time.Second,
-		"Close exceeded bound (%s); expected ~%s", elapsed, closeDrainTimeout)
+	assert.Less(t, elapsed, s.opts.CloseDrainTimeout+2*time.Second,
+		"Close exceeded bound (%s); expected ~%s", elapsed, s.opts.CloseDrainTimeout)
+}
+
+// TestCloseDrainTimeout_ScalesWithPullTimeout — the drain bound has to
+// outlast an in-flight pull, otherwise Close times out deterministically
+// on every shutdown that lands mid-poll, skips Unsubscribe, and orphans
+// the pull-point at the camera until its termination expires. A fixed 5s
+// was fine only while PullTimeout was 5s; callers raising the poll made
+// the drain the shorter of the two.
+func TestCloseDrainTimeout_ScalesWithPullTimeout(t *testing.T) {
+	tests := []struct {
+		name string
+		opts Options
+		want time.Duration
+	}{
+		{"derived from the default poll", Options{}, 5*time.Second + closeDrainSlack},
+		{"derived from a long poll", Options{PullTimeout: 30 * time.Second}, 30*time.Second + closeDrainSlack},
+		{"explicit value wins", Options{PullTimeout: 30 * time.Second, CloseDrainTimeout: time.Minute}, time.Minute},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := tt.opts.withDefaults()
+			assert.Equal(t, tt.want, got.CloseDrainTimeout)
+			assert.Greater(t, got.CloseDrainTimeout, got.PullTimeout,
+				"the drain must outlast a pull or Close can never drain cleanly")
+		})
+	}
+}
+
+// TestDrainFor_BoundedByTheLongestPossibleCall — PullTimeout is how
+// long the camera holds a poll, but the client ceiling is what actually
+// caps the call, and callers routinely set it higher. Deriving the
+// drain from PullTimeout alone leaves it shorter than a call that runs
+// to the client ceiling, which is the case Close cannot survive: it
+// gives up, skips Unsubscribe, and orphans the pull-point.
+func TestDrainFor_BoundedByTheLongestPossibleCall(t *testing.T) {
+	tests := []struct {
+		name   string
+		pull   time.Duration
+		client time.Duration
+		want   time.Duration
+	}{
+		{"client ceiling above the poll wins", 30 * time.Second, 40 * time.Second, 40*time.Second + closeDrainSlack},
+		{"client ceiling below the poll is not the bound", 30 * time.Second, 10 * time.Second, 30*time.Second + closeDrainSlack},
+		{"equal leaves the poll as the bound", 30 * time.Second, 30 * time.Second, 30*time.Second + closeDrainSlack},
+		{"unbounded client falls back to the poll", 30 * time.Second, 0, 30*time.Second + closeDrainSlack},
+	}
+
+	for _, tt := range tests {
+		t.Run(tt.name, func(t *testing.T) {
+			got := drainFor(tt.pull, tt.client)
+			assert.Equal(t, tt.want, got)
+			assert.Greater(t, got, tt.pull, "the drain must outlast a poll")
+			if tt.client > 0 {
+				assert.Greater(t, got, tt.client, "the drain must outlast the client ceiling")
+			}
+		})
+	}
 }