From 9ae498354a0e6d7826bca7566f6c775bfc4ce5cc Mon Sep 17 00:00:00 2001 From: "T. Tradesman" <184814242+ttradesman@users.noreply.github.com> Date: Thu, 23 Jul 2026 14:28:06 +0200 Subject: [PATCH 1/2] fix(event/stream): scale the close drain bound with PullTimeout MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit closeDrainTimeout was a fixed 5s while the pull loop can be parked inside a non-ctx-aware SendSoap for as long as the caller's HTTP client allows. That was fine when PullTimeout was also 5s, but a caller raising the poll made the drain the shorter of the two: Close then times out on every shutdown landing mid-poll, and the Unsubscribe below it is skipped, orphaning the pull-point at the camera until its termination expires. Derive the default as PullTimeout + closeDrainSlack so it outlasts a poll, and expose CloseDrainTimeout for callers whose client ceiling is higher still — that ceiling, not PullTimeout, is the real worst case for a stalled camera, and only the caller knows it. The hung-HTTP test now sets it explicitly; it was relying on the constant being small. --- event/stream/stream.go | 32 ++++++++++++++++++++++---------- event/stream/stream_test.go | 37 ++++++++++++++++++++++++++++++++++--- 2 files changed, 56 insertions(+), 13 deletions(-) diff --git a/event/stream/stream.go b/event/stream/stream.go index 697a5053..8a9530aa 100644 --- a/event/stream/stream.go +++ b/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 @@ -66,6 +66,12 @@ 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 as PullTimeout + closeDrainSlack, + // so it outlasts an in-flight poll by default. Raise it when the + // device's http.Client.Timeout is higher still — that ceiling, not + // PullTimeout, is the real worst case for a stalled camera. + CloseDrainTimeout time.Duration } func defaultOptions() Options { @@ -109,6 +115,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 } @@ -137,7 +149,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 { @@ -275,7 +287,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. // @@ -289,8 +301,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 } diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go index f0c8df7d..7c0a1ef6 100644 --- a/event/stream/stream_test.go +++ b/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,36 @@ 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") + }) + } } From 9d56f10ebe0b554d99a3a2e41efa6a9cf034ae42 Mon Sep 17 00:00:00 2001 From: "T. Tradesman" <184814242+ttradesman@users.noreply.github.com> Date: Thu, 23 Jul 2026 15:15:29 +0200 Subject: [PATCH 2/2] fix(event/stream): derive the drain bound from the client ceiling too MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The drain default was PullTimeout + slack, which bounds the poll but not the call. PullTimeout is how long the camera holds a poll; http.Client.Timeout is what actually caps the request, and callers set it higher — it has to cover dial, TLS and the response transfer on top of the poll it outlasts. So a caller with a 30s poll and a 40s ceiling got a 40s drain against a call that can also run 40s: a dead heat, and Close still times out and skips Unsubscribe on the shutdowns that land mid-poll. The Options escape hatch covered it, but a default that every caller has to override is not much of a default. Take whichever of the two is longer. A zero ceiling means unbounded, where no finite drain helps, so the poll stays the best bound available. Derived in NewStream because only that entry point sees the device; newStream keeps the PullTimeout-only bound for callers wiring their own caller implementation. --- event/stream/stream.go | 27 +++++++++++++++++++++++---- event/stream/stream_test.go | 31 +++++++++++++++++++++++++++++++ 2 files changed, 54 insertions(+), 4 deletions(-) diff --git a/event/stream/stream.go b/event/stream/stream.go index 8a9530aa..9e883d9e 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -67,10 +67,11 @@ type Options struct { // 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 as PullTimeout + closeDrainSlack, - // so it outlasts an in-flight poll by default. Raise it when the - // device's http.Client.Timeout is higher still — that ceiling, not - // PullTimeout, is the real worst case for a stalled camera. + // 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 } @@ -254,9 +255,27 @@ func NewStream(ctx context.Context, dev *onvif.Device, opts Options) (*Stream, e if err := validateClientTimeout(clientTimeoutOf(dev), opts.withDefaults().PullTimeout); err != nil { return nil, err } + // 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, clientTimeoutOf(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 +} + func newStream(ctx context.Context, c caller, opts Options) (*Stream, error) { opts = opts.withDefaults() ref, err := createPullPoint(c, opts) diff --git a/event/stream/stream_test.go b/event/stream/stream_test.go index 7c0a1ef6..48865d0d 100644 --- a/event/stream/stream_test.go +++ b/event/stream/stream_test.go @@ -541,3 +541,34 @@ func TestCloseDrainTimeout_ScalesWithPullTimeout(t *testing.T) { }) } } + +// 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") + } + }) + } +}