feat(event/stream): raise recreate backoff cap and add jitter

The previous 30-second cap meant a 1000-camera fleet recovering from
a switch reboot would generate a sustained 33 RPS of doomed
CreatePullPointSubscription traffic against still-booting cameras,
and the synchronised retries would arrive in phase. Two changes:

* Cap raised to 5 minutes. Single-camera recovery latency goes from
  '<=30s after camera comes back' to '<=300s', which is fine because
  by the time we are this deep in backoff the camera has already been
  unreachable through 6+ attempts (1s, 2s, 4s, 8s, 16s, 30s under the
  old cap) — the marginal recovery delay is acceptable to avoid the
  network melt.

* Symmetric ±25% jitter on every recreate sleep so synchronised
  drops (switch reboot, DHCP storm, NTP slew) do not cause
  synchronised reconnect surges. Standard practice — same shape AWS,
  Cloudflare and HA event_manager use.

Tests assert the jitter range, the documented cap value (so a future
maintainer flipping it back to 30s notices in CI), and that jitter
varies across calls (proves the rand source is wired).

event/stream/jitter_test.go

@@ -0,0 +1,44 @@
+package stream
+
+import (
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestJitter_StaysWithinFraction(t *testing.T) {
+	const base = time.Second
+	low := time.Duration(float64(base) * (1 - jitterFraction))
+	high := time.Duration(float64(base) * (1 + jitterFraction))
+	for i := 0; i < 200; i++ {
+		got := jitter(base)
+		assert.GreaterOrEqual(t, got, low, "iteration %d", i)
+		assert.LessOrEqual(t, got, high, "iteration %d", i)
+	}
+}
+
+func TestJitter_ZeroAndNegativeReturnPositive(t *testing.T) {
+	assert.Greater(t, jitter(0), time.Duration(0))
+	assert.Greater(t, jitter(-time.Second), time.Duration(0))
+}
+
+func TestJitter_VariesAcrossCalls(t *testing.T) {
+	// Sanity check that we're not returning a constant. Vanishingly
+	// unlikely to flake (probability ~ (1/uint64-space)^9).
+	first := jitter(time.Second)
+	allEqual := true
+	for i := 0; i < 10; i++ {
+		if jitter(time.Second) != first {
+			allEqual = false
+			break
+		}
+	}
+	assert.False(t, allEqual, "jitter is producing a constant; rand seed not working")
+}
+
+func TestMaxRecreateBackoff_Is5Minutes(t *testing.T) {
+	// Document the policy choice in a test so a future maintainer
+	// changing this notices.
+	assert.Equal(t, 5*time.Minute, maxRecreateBackoff)
+}

event/stream/stream.go

@@ -7,6 +7,7 @@ import (
 	"errors"
 	"fmt"
 	"io"
+	"math/rand"
 	"net/http"
 	"regexp"
 	"strconv"
@@ -138,7 +139,18 @@ func (o Options) withDefaults() Options {
 }
 
 // maxRecreateBackoff caps exponential backoff between recreate attempts.
-const maxRecreateBackoff = 30 * time.Second
+// Sized for fleet deployments: a 1000-camera setup recovering from a
+// switch reboot would otherwise hammer the network with one recreate
+// attempt per camera per 30s; 5 minutes gives the network time to
+// settle while still recovering promptly when a single camera comes
+// back.
+const maxRecreateBackoff = 5 * time.Minute
+
+// jitterFraction is the symmetric jitter applied to recreate backoff:
+// the actual sleep is sampled from [backoff*(1-jitter), backoff*(1+jitter)].
+// Prevents thundering-herd reconnects when many cameras drop together
+// (switch reboot, NAT timeout).
+const jitterFraction = 0.25
 
 // caller is the subset of *onvif.Device the Stream depends on. Tests
 // substitute a fake; production code uses the device adapter.
@@ -351,7 +363,7 @@ func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *ti
 	addr, err := createPullPoint(s.caller, s.opts)
 	if err != nil {
 		s.surfaceError(ErrRecreateFailed{Err: err})
-		if !sleepCtx(ctx, *backoff) {
+		if !sleepCtx(ctx, jitter(*backoff)) {
 			return false, false
 		}
 		*backoff *= 2
@@ -366,6 +378,23 @@ func (s *Stream) attemptRecreate(ctx context.Context, failures *int, backoff *ti
 	return true, true
 }
 
+// jitter returns d perturbed by ±jitterFraction. Used to spread
+// recreate attempts across a fleet so a synchronised drop (switch
+// reboot, DHCP storm) does not cause a synchronised reconnect surge.
+// Returns at least 1ns to keep sleepCtx happy.
+func jitter(d time.Duration) time.Duration {
+	if d <= 0 {
+		return time.Nanosecond
+	}
+	spread := float64(d) * jitterFraction
+	delta := (rand.Float64()*2 - 1) * spread
+	out := time.Duration(float64(d) + delta)
+	if out <= 0 {
+		out = time.Nanosecond
+	}
+	return out
+}
+
 // renewLoop refreshes the subscription before InitialTermination expires.
 // Exits when ctx is cancelled.
 func (s *Stream) renewLoop(ctx context.Context) {