feat(event/stream): renew subscription before termination expires

Adds a background renew loop alongside the pull loop. ONVIF pull-point
subscriptions expire at the InitialTerminationTime supplied to Create;
without periodic Renew calls the camera silently drops the
subscription and subsequent pulls start returning empty messages — the
shape the existing agent's heartbeat code in cloud/Cloud.go has been
papering over by occasionally recreating subscriptions.

Design
------
* New Options.RenewMargin (default 10s) — how far before
  InitialTermination expiry the renew fires. Smaller margins mean
  fewer SOAP round-trips; larger margins tolerate slow networks. With
  default 60s termination + 10s margin we renew every 50s, which is in
  line with what production NVRs (Milestone, Genetec) use.
* The renew loop runs in a separate goroutine sharing ctx with the
  pull loop. WaitGroup synchronisation in run() ensures both have
  exited before close()-of-channels happens, so a renew in flight
  during Close() cannot send on a closed Errors channel.
* Pathological config (RenewMargin >= InitialTermination) falls back
  to renewing at termination/2 rather than busy-looping or never
  renewing.
* renewPullPoint sends a wsnt:Renew SOAP against the SubscriptionRef
  Address with the same InitialTermination duration; renew errors
  surface on Errors non-blockingly, identically to pull errors.

Tests use very short termination/margin (80-100ms / 10ms) so a single
test run observes multiple renews within ~500ms, and assert that
renew calls target the SubscriptionReference endpoint (not the device
endpoint). -race clean.

event/stream/renew_test.go

@@ -0,0 +1,134 @@
+package stream
+
+import (
+	"context"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+	"github.com/stretchr/testify/require"
+)
+
+// countSendSoapMatching counts how many recorded SendSoap calls have a
+// body containing needle. Safe to call concurrently with the run loop.
+func countSendSoapMatching(fc *fakeCaller, needle string) int {
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	n := 0
+	for _, c := range fc.sendSoapCalls {
+		if strings.Contains(c[1], needle) {
+			n++
+		}
+	}
+	return n
+}
+
+func TestStream_RenewsSubscriptionBeforeExpiry(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	// 100 ms termination with 10 ms margin -> renew every ~90 ms.
+	s, err := newStream(ctx, fc, Options{
+		DeviceID:           "cam-1",
+		InitialTermination: 100 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(500 * time.Millisecond)
+	var renewCount int
+	for time.Now().Before(deadline) {
+		renewCount = countSendSoapMatching(fc, "Renew")
+		if renewCount >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	assert.GreaterOrEqual(t, renewCount, 1, "expected at least one Renew SendSoap call within 500ms")
+}
+
+func TestStream_RenewSendsToSubscriptionEndpoint(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s, err := newStream(ctx, fc, Options{
+		InitialTermination: 80 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	deadline := time.Now().Add(500 * time.Millisecond)
+	for time.Now().Before(deadline) {
+		if countSendSoapMatching(fc, "Renew") >= 1 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+
+	fc.mu.Lock()
+	defer fc.mu.Unlock()
+	var renewEndpoint string
+	for _, c := range fc.sendSoapCalls {
+		if strings.Contains(c[1], "Renew") {
+			renewEndpoint = c[0]
+			break
+		}
+	}
+	require.NotEmpty(t, renewEndpoint, "no Renew call found")
+	assert.Equal(t, "http://camera.local/onvif/Events/PullSub_1", renewEndpoint,
+		"Renew must target the SubscriptionReference Address")
+}
+
+func TestStream_RenewMarginAppliesDefault(t *testing.T) {
+	o := defaultOptions()
+	assert.Equal(t, 10*time.Second, o.RenewMargin)
+}
+
+func TestStream_RenewErrorSurfacedOnErrorsChannel(t *testing.T) {
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	// Defaults return empty pulls indefinitely so the pull loop is clean.
+	// Override defaultSendSoap on the fly to return a Renew error for
+	// any body that looks like a Renew. We do that by tagging the
+	// default response with an err, then resetting after capturing one.
+	// Simpler: just queue several explicit Renew-error responses; the
+	// fake's queue is consumed in FIFO and the pull body never matches
+	// 'Renew', so queued errors will land on the renew call only if
+	// queued before any pulls. To bias the order we drain via a custom
+	// default.
+	fc.mu.Lock()
+	fc.defaultSendSoap = fakeResp{err: errInjected{}}
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	s, err := newStream(ctx, fc, Options{
+		InitialTermination: 80 * time.Millisecond,
+		RenewMargin:        10 * time.Millisecond,
+	})
+	require.NoError(t, err)
+	defer s.Close()
+
+	select {
+	case e := <-s.Errors():
+		assert.Contains(t, e.Error(), "injected")
+	case <-time.After(time.Second):
+		t.Fatal("expected an error on Errors channel from failing Renew/pull")
+	}
+}
+
+// errInjected is a sentinel error type so the test message has a stable
+// substring without depending on a wrapped string match.
+type errInjected struct{}
+
+func (errInjected) Error() string { return "injected fake error" }

event/stream/stream.go

@@ -38,9 +38,13 @@ type Options struct {
 	// returned per PullMessages call. Default: 10.
 	MessageLimit int
 	// InitialTermination is the requested subscription lifetime passed
-	// to CreatePullPointSubscription. The renew loop (added in a later
-	// commit) will refresh well before this expires. Default: 60s.
+	// to CreatePullPointSubscription. The renew loop refreshes well
+	// before this expires. Default: 60s.
 	InitialTermination time.Duration
+	// RenewMargin is how long before InitialTermination expiry the
+	// renew loop fires. Larger margins tolerate slower networks at the
+	// cost of more renew SOAP calls. Default: 10s.
+	RenewMargin 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.
@@ -52,6 +56,7 @@ func defaultOptions() Options {
 		PullTimeout:        5 * time.Second,
 		MessageLimit:       10,
 		InitialTermination: 60 * time.Second,
+		RenewMargin:        10 * time.Second,
 		BufferSize:         16,
 	}
 }
@@ -67,6 +72,9 @@ func (o Options) withDefaults() Options {
 	if o.InitialTermination > 0 {
 		d.InitialTermination = o.InitialTermination
 	}
+	if o.RenewMargin > 0 {
+		d.RenewMargin = o.RenewMargin
+	}
 	if o.BufferSize > 0 {
 		d.BufferSize = o.BufferSize
 	}
@@ -179,6 +187,17 @@ func (s *Stream) run(ctx context.Context) {
 	defer close(s.events)
 	defer close(s.errors)
 
+	var wg sync.WaitGroup
+	wg.Add(1)
+	go func() {
+		defer wg.Done()
+		s.renewLoop(ctx)
+	}()
+	s.pullLoop(ctx)
+	wg.Wait()
+}
+
+func (s *Stream) pullLoop(ctx context.Context) {
 	for {
 		if ctx.Err() != nil {
 			return
@@ -186,9 +205,9 @@ func (s *Stream) run(ctx context.Context) {
 		msgs, err := pullMessages(s.caller, s.pullPoint, s.opts)
 		if err != nil {
 			s.surfaceError(err)
-			// Brief backoff before retrying; reconnect-on-error
-			// lands in a follow-up commit and replaces this with
-			// proper subscription recreation.
+			// Brief backoff before retrying; automatic
+			// subscription recreation lands in the reconnect
+			// commit and replaces this fallback.
 			if !sleepCtx(ctx, time.Second) {
 				return
 			}
@@ -206,6 +225,33 @@ func (s *Stream) run(ctx context.Context) {
 	}
 }
 
+// renewLoop refreshes the subscription before InitialTermination expires.
+// Exits when ctx is cancelled.
+func (s *Stream) renewLoop(ctx context.Context) {
+	interval := s.opts.InitialTermination - s.opts.RenewMargin
+	if interval <= 0 {
+		// Pathological config (margin >= termination): fall back to
+		// renewing at half the termination so we still refresh,
+		// rather than busy-looping or never renewing.
+		interval = s.opts.InitialTermination / 2
+		if interval <= 0 {
+			interval = time.Second
+		}
+	}
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if err := renewPullPoint(s.caller, s.pullPoint, s.opts); err != nil {
+				s.surfaceError(fmt.Errorf("renew pull point: %w", err))
+			}
+		}
+	}
+}
+
 // surfaceError sends err on the errors channel non-blockingly so a
 // stalled consumer cannot block the pull loop.
 func (s *Stream) surfaceError(err error) {
@@ -284,6 +330,20 @@ func pullMessages(c caller, endpoint string, opts Options) ([]event.Notification
 	return decoded.NotificationMessage, nil
 }
 
+func renewPullPoint(c caller, endpoint string, opts Options) error {
+	req := event.Renew{TerminationTime: xsd.String(durationToXSD(opts.InitialTermination))}
+	body, err := xml.Marshal(req)
+	if err != nil {
+		return fmt.Errorf("marshal Renew: %w", err)
+	}
+	resp, err := c.SendSoap(endpoint, string(body))
+	if err != nil {
+		return err
+	}
+	_, err = readClose(resp)
+	return err
+}
+
 func unsubscribePullPoint(c caller, endpoint string) error {
 	if endpoint == "" {
 		return nil

event/stream/topics.go

@@ -168,7 +168,7 @@ var topicRules = []struct {
 	// tns1:RuleEngine/LineDetector/Crossed — line crossing (Hikvision,
 	// Bosch IVA, others). Data: ObjectId (xsd:int); edge-triggered, no
 	// State boolean.
-	// ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
+	// https://www.onvif.org/specs/srv/analytics/ONVIF-VideoAnalytics-Service-Spec-v220.pdf §5.4
 	{"LineDetector/Crossed", KindObjectDetected},
 
 	// tns1:RuleEngine/FieldDetector/ObjectsInside — intrusion / region