From 82f98cb82479d0b995bdf43b16e2084bc418085d Mon Sep 17 00:00:00 2001 From: Sebastian Norling <1932208+Bazze@users.noreply.github.com> Date: Thu, 21 May 2026 14:40:42 +0200 Subject: [PATCH] feat(event/stream): recreate subscription after consecutive pull failures MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Adds automatic CreatePullPointSubscription recreation when the pull loop hits ReconnectAfterFailures (default 3) consecutive errors. Mirrors what production ONVIF clients (Home Assistant event_manager, Milestone integration) do because pull points die for many reasons none of which surface as a clean SOAP fault: camera reboot, NAT session timeout, subscription garbage-collected after a renew miss, firmware bug. Recreating is the only reliable recovery; Renew alone cannot save an already-dropped subscription. Two new options --------------- * ReconnectAfterFailures int (default 3) — how many consecutive pull failures trigger recreate. Conservative default; tunable for always-on cameras vs flaky NAT. * RetryBackoff time.Duration (default 1s) — base sleep between pull retries; recreate failures double this up to a 30s cap so a permanently broken camera does not hammer the network. Lifecycle changes ----------------- * Stream.pullPoint is now mutex-protected — the renew goroutine reads it concurrently with the pull loop installing a new address after recreate. getPullPoint/setPullPoint accessors keep the locking contained. * On successful recreate, failure count and backoff reset to defaults so the loop is back to its happy-path cadence. * On recreate failure, the loop continues retrying (until ctx cancel) with exponentially increasing sleep — never blocks Close. Tests cover: post-failure recreate hits a different SubscriptionRef Address and subsequent events come from the new endpoint; exponential backoff drives multiple recreate attempts when the camera stays down; defaults match production-sensible 3 failures / 1s backoff. -race clean. --- event/stream/reconnect_test.go | 120 +++++++++++++++++++++++++++++++++ event/stream/stream.go | 99 ++++++++++++++++++++++----- 2 files changed, 204 insertions(+), 15 deletions(-) create mode 100644 event/stream/reconnect_test.go diff --git a/event/stream/reconnect_test.go b/event/stream/reconnect_test.go new file mode 100644 index 0000000..fd229a5 --- /dev/null +++ b/event/stream/reconnect_test.go @@ -0,0 +1,120 @@ +package stream + +import ( + "context" + "errors" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// createPullPointRespAlt mirrors the first fixture but returns a +// different SubscriptionReference Address so a test can prove that +// subsequent pulls hit the recreated endpoint. +const createPullPointRespAlt = ` + + + + + http://camera.local/onvif/Events/PullSub_2 + + 2026-05-21T10:30:10Z + 2026-05-21T10:31:10Z + + +` + +func TestStream_RecreatesSubscriptionAfterRepeatedPullErrors(t *testing.T) { + fc := newFakeCaller() + // Initial subscription. + fc.queueCallMethod(createPullPointResp, nil) + // Recreated subscription returns a *different* endpoint. + fc.queueCallMethod(createPullPointRespAlt, nil) + + // First pull fails. With ReconnectAfterFailures=1 this triggers a + // recreate; subsequent pulls go to PullSub_2 which we'll observe. + fc.queueSendSoap("", errors.New("transient failure")) + fc.queueSendSoap(pullMessagesResp(motionMsg("true")), nil) + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + DeviceID: "cam-1", + PullTimeout: 50 * time.Millisecond, + ReconnectAfterFailures: 1, + RetryBackoff: 10 * time.Millisecond, + InitialTermination: 30 * time.Second, // keep renew quiet + }) + require.NoError(t, err) + defer s.Close() + + ev := receive(t, s.Events(), 2*time.Second) + assert.Equal(t, KindMotion, ev.Kind) + + fc.mu.Lock() + defer fc.mu.Unlock() + require.Len(t, fc.callMethodCalls, 2, + "expected exactly 2 CallMethod calls (initial + recreate)") + // The PullMessages call that delivered the motion event must + // target the new endpoint. + var newEndpointPulls int + for _, c := range fc.sendSoapCalls { + if c[0] == "http://camera.local/onvif/Events/PullSub_2" { + newEndpointPulls++ + } + } + assert.GreaterOrEqual(t, newEndpointPulls, 1, + "expected pulls against the recreated subscription endpoint") +} + +func TestStream_BackoffWhenRecreateFails(t *testing.T) { + fc := newFakeCaller() + fc.queueCallMethod(createPullPointResp, nil) + // After the initial successful create, every CallMethod (recreate) + // and SendSoap (pull) fails. The loop should keep retrying with + // exponential backoff rather than blocking forever or spinning. + fc.mu.Lock() + fc.defaultCall = fakeResp{err: errors.New("recreate fail")} + fc.defaultSendSoap = fakeResp{err: errors.New("pull fail")} + fc.mu.Unlock() + + ctx, cancel := context.WithCancel(context.Background()) + defer cancel() + s, err := newStream(ctx, fc, Options{ + PullTimeout: 10 * time.Millisecond, + ReconnectAfterFailures: 1, + RetryBackoff: 10 * time.Millisecond, + InitialTermination: 30 * time.Second, + }) + require.NoError(t, err) + defer s.Close() + + deadline := time.Now().Add(2 * time.Second) + var calls atomic.Int32 + for time.Now().Before(deadline) { + fc.mu.Lock() + calls.Store(int32(len(fc.callMethodCalls))) + fc.mu.Unlock() + if calls.Load() >= 4 { + break + } + time.Sleep(20 * time.Millisecond) + } + assert.GreaterOrEqual(t, calls.Load(), int32(4), + "expected stream to retry recreate (>=3 retries on top of the initial create)") +} + +func TestStream_ReconnectAfterFailuresDefault(t *testing.T) { + o := defaultOptions() + assert.Equal(t, 3, o.ReconnectAfterFailures) +} + +func TestStream_RetryBackoffDefault(t *testing.T) { + o := defaultOptions() + assert.Equal(t, time.Second, o.RetryBackoff) +} diff --git a/event/stream/stream.go b/event/stream/stream.go index 102f88e..a7cc06b 100644 --- a/event/stream/stream.go +++ b/event/stream/stream.go @@ -45,6 +45,17 @@ type Options struct { // renew loop fires. Larger margins tolerate slower networks at the // cost of more renew SOAP calls. Default: 10s. RenewMargin time.Duration + // ReconnectAfterFailures is the consecutive PullMessages failure + // count that triggers a CreatePullPointSubscription recreate. The + // camera or pull-point can die for many reasons (camera reboot, + // subscription garbage-collected after a renew miss, intermediate + // NAT timeout); rebuilding the subscription is the only reliable + // recovery. Default: 3. + ReconnectAfterFailures int + // RetryBackoff is the initial sleep between a pull/recreate failure + // and the next attempt. Recreate failures double this up to a 30s + // ceiling. Default: 1s. + RetryBackoff time.Duration // BufferSize is the buffer size of the Events and Errors channels. // Larger buffers absorb consumer hiccups at the cost of memory. // Default: 16. @@ -53,11 +64,13 @@ type Options struct { func defaultOptions() Options { return Options{ - PullTimeout: 5 * time.Second, - MessageLimit: 10, - InitialTermination: 60 * time.Second, - RenewMargin: 10 * time.Second, - BufferSize: 16, + PullTimeout: 5 * time.Second, + MessageLimit: 10, + InitialTermination: 60 * time.Second, + RenewMargin: 10 * time.Second, + ReconnectAfterFailures: 3, + RetryBackoff: time.Second, + BufferSize: 16, } } @@ -75,6 +88,12 @@ func (o Options) withDefaults() Options { if o.RenewMargin > 0 { d.RenewMargin = o.RenewMargin } + if o.ReconnectAfterFailures > 0 { + d.ReconnectAfterFailures = o.ReconnectAfterFailures + } + if o.RetryBackoff > 0 { + d.RetryBackoff = o.RetryBackoff + } if o.BufferSize > 0 { d.BufferSize = o.BufferSize } @@ -83,6 +102,9 @@ func (o Options) withDefaults() Options { return d } +// maxRecreateBackoff caps exponential backoff between recreate attempts. +const maxRecreateBackoff = 30 * time.Second + // caller is the subset of *onvif.Device the Stream depends on. Tests // substitute a fake; production code uses the device adapter. type caller interface { @@ -107,9 +129,11 @@ func (d deviceCaller) SendSoap(endpoint, body string) (*http.Response, error) { // A Stream is safe for concurrent use by Close from any goroutine while // readers consume Events / Errors; Close is idempotent. type Stream struct { - caller caller - opts Options - pullPoint string + caller caller + opts Options + + pullPointMu sync.Mutex + pullPoint string events chan Event errors chan error @@ -124,6 +148,18 @@ type Stream struct { now func() time.Time } +func (s *Stream) getPullPoint() string { + s.pullPointMu.Lock() + defer s.pullPointMu.Unlock() + return s.pullPoint +} + +func (s *Stream) setPullPoint(addr string) { + s.pullPointMu.Lock() + defer s.pullPointMu.Unlock() + s.pullPoint = addr +} + // NewStream creates a Stream against an ONVIF device. It performs the // CreatePullPointSubscription call synchronously so connectivity and // authentication problems surface immediately as an error rather than @@ -175,7 +211,7 @@ func (s *Stream) Close() error { // Unsubscribe is best-effort: if the camera is unreachable // the subscription will expire on its own at // InitialTermination + Renew interval anyway. - if err := unsubscribePullPoint(s.caller, s.pullPoint); err != nil { + if err := unsubscribePullPoint(s.caller, s.getPullPoint()); err != nil { s.closeErr = fmt.Errorf("unsubscribe pull point: %w", err) } }) @@ -198,21 +234,31 @@ func (s *Stream) run(ctx context.Context) { } func (s *Stream) pullLoop(ctx context.Context) { + var failures int + recreateBackoff := s.opts.RetryBackoff + for { if ctx.Err() != nil { return } - msgs, err := pullMessages(s.caller, s.pullPoint, s.opts) + msgs, err := pullMessages(s.caller, s.getPullPoint(), s.opts) if err != nil { s.surfaceError(err) - // Brief backoff before retrying; automatic - // subscription recreation lands in the reconnect - // commit and replaces this fallback. - if !sleepCtx(ctx, time.Second) { + failures++ + if failures >= s.opts.ReconnectAfterFailures { + if !s.attemptRecreate(ctx, &failures, &recreateBackoff) { + return + } + continue + } + if !sleepCtx(ctx, s.opts.RetryBackoff) { return } continue } + // Successful pull resets failure tracking. + failures = 0 + recreateBackoff = s.opts.RetryBackoff observedAt := s.now() for _, m := range msgs { ev := Decode(m, s.opts.DeviceID, observedAt) @@ -225,6 +271,29 @@ func (s *Stream) pullLoop(ctx context.Context) { } } +// attemptRecreate calls CreatePullPointSubscription and on success +// installs the new endpoint atomically. Returns false if ctx was +// cancelled while waiting for backoff (caller should exit the run +// loop). +func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *time.Duration) bool { + addr, err := createPullPoint(s.caller, s.opts) + if err != nil { + s.surfaceError(fmt.Errorf("recreate pull point: %w", err)) + if !sleepCtx(ctx, *backoff) { + return false + } + *backoff *= 2 + if *backoff > maxRecreateBackoff { + *backoff = maxRecreateBackoff + } + return true + } + s.setPullPoint(addr) + *failures = 0 + *backoff = s.opts.RetryBackoff + return true +} + // renewLoop refreshes the subscription before InitialTermination expires. // Exits when ctx is cancelled. func (s *Stream) renewLoop(ctx context.Context) { @@ -245,7 +314,7 @@ func (s *Stream) renewLoop(ctx context.Context) { case <-ctx.Done(): return case <-ticker.C: - if err := renewPullPoint(s.caller, s.pullPoint, s.opts); err != nil { + if err := renewPullPoint(s.caller, s.getPullPoint(), s.opts); err != nil { s.surfaceError(fmt.Errorf("renew pull point: %w", err)) } }