feat(event/stream): recreate subscription after consecutive pull failures

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.
2026-08-23 15:08:33 +00:00 · 2026-05-21 14:40:42 +02:00
parent 9b5b626113
commit 82f98cb824
2 changed files with 204 additions and 15 deletions
--- a/event/stream/reconnect_test.go
+++ 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 = `<?xml version="1.0" encoding="UTF-8"?>
+<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-envelope"
+              xmlns:wsa="http://www.w3.org/2005/08/addressing"
+              xmlns:tev="http://www.onvif.org/ver10/events/wsdl">
+  <env:Body>
+    <tev:CreatePullPointSubscriptionResponse>
+      <tev:SubscriptionReference>
+        <wsa:Address>http://camera.local/onvif/Events/PullSub_2</wsa:Address>
+      </tev:SubscriptionReference>
+      <tev:CurrentTime>2026-05-21T10:30:10Z</tev:CurrentTime>
+      <tev:TerminationTime>2026-05-21T10:31:10Z</tev:TerminationTime>
+    </tev:CreatePullPointSubscriptionResponse>
+  </env:Body>
+</env:Envelope>`
+
+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)
+}
--- 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))
 			}
 		}