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