fix(event/stream): address round-2 review findings

Critical - Renew busy-loop on persistent failure: after a failed renewPullPoint, the loop re-read the (now in-past) GrantedTermination and nextRenewInterval floored to 1s, hammering the camera at 1 Hz until reconnect. nextRenewIntervalAfterError decouples the failure path from the stale grant and backs off at opts.RetryBackoff. renewLoop also routes through s.now() so test clocks can drive it deterministically. - Lost-update race on GrantedTermination: a renew result for an old subscription could overwrite the grant on a new one if attemptRecreate swapped pullPoint mid-flight. A generation counter on Stream tracks subscription rotation; the renew loop captures the generation before the SOAP call and discards the result if the subscription was rotated. - Credential leak when <Security> straddled the 64 KiB error cap: the non-greedy regex required a closing tag and missed the truncated case. wsseSecurityRE now matches close-tag-or-EOF. Belt-and-braces wssePasswordRE redacts <Password> elements outside any Security wrapper. - addHeaderChildren accepted well-formed-but-element-free input and produced a header-less request. Now errors out. Important - Wrapper detection in buildRefParamsHeader was HasSuffix-based and misfired on children named *ReferenceParameters. Replaced with the unambiguous wrapper-only contract: input must be the full <*:ReferenceParameters> element returned by extractReferenceParameters. - Renamed SoapOption → SendSoapOption and WithHeader → WithSOAPHeader to disambiguate at call sites. - Renamed WithHeader's `xml` parameter to headerContent to avoid shadowing the encoding/xml package name. - Documented terminationTimeRE's "first match only" semantics so nobody re-uses it from PullMessages context where multiple TerminationTime elements appear. - goleak is now a direct require (go mod tidy). Performance - Added gosoap.AddStringHeaderContents (plural) for multi-root header content. AddStringHeaderContent remains as-is for backwards compatibility with external consumers. Device.go's addHeaderChildren workaround is gone — one etree parse per SendSoapWithOptions call instead of two. Migration - Device.go's own CallOnvifFunction and the three examples now call SendSoapWithOptions, modelling the canonical path. Docs / tests - Trust-boundary warning on SendSoapWithHeader/Options godoc. - Comments on createPullPointResp/Alt explain why TerminationTime is intentionally omitted (renew timing fixtures). - New tests: digest retry strips WS-Security, duplicate WithSOAPHeader is last-wins, malformed TerminationTime yields zero, margin==base falls to base/2, empty SubscriptionReference yields empty ref params, Security straddling cap is redacted, bare Password redacted, wrapper-only contract is enforced.
2026-08-23 15:08:33 +00:00 · 2026-05-27 18:07:56 +02:00
parent c7ef445d6a
commit d7cfee56a1
14 changed files with 367 additions and 94 deletions
--- a/event/stream/reconnect_test.go
+++ b/event/stream/reconnect_test.go
@@ -13,7 +13,9 @@ import (

 // createPullPointRespAlt mirrors the first fixture but returns a
 // different SubscriptionReference Address so a test can prove that
-// subsequent pulls hit the recreated endpoint.
+// subsequent pulls hit the recreated endpoint. Like createPullPointResp,
+// it intentionally omits <TerminationTime> so renew scheduling stays
+// driven by opts.
 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"
--- a/event/stream/renew.go
+++ b/event/stream/renew.go
@@ -11,23 +11,29 @@ import (
 )

 // renewLoop sleeps until the next deadline (camera-granted termination
-// minus RenewMargin), renews, and repeats. A permanently failing
-// renew lets the subscription die at the camera; the pull loop's
-// reconnect path then recreates it — recreate is the only reliable
-// recovery once a subscription is GC'd.
+// minus RenewMargin), renews, and repeats. On failure it backs off via
+// nextRenewIntervalAfterError so the loop doesn't busy-loop against
+// the 1s floor when the previous grant has just expired. A permanently
+// failing renew lets the subscription die at the camera; the pull
+// loop's reconnect path then recreates it — recreate is the only
+// reliable recovery once a subscription is GC'd.
 func (s *Stream) renewLoop(ctx context.Context) {
 	for {
 		ref := s.getPullPoint()
-		if !sleepCtx(ctx, nextRenewInterval(ref.GrantedTermination, s.opts, time.Now())) {
+		gen := s.pullPointGen()
+		if !sleepCtx(ctx, nextRenewInterval(ref.GrantedTermination, s.opts, s.now())) {
 			return
 		}
 		granted, err := renewPullPoint(s.caller, s.getPullPoint(), s.opts)
 		if err != nil {
 			s.surfaceError(ErrRenewFailed{Err: err})
+			if !sleepCtx(ctx, nextRenewIntervalAfterError(ref.GrantedTermination, s.opts, s.now())) {
+				return
+			}
 			continue
 		}
 		if !granted.IsZero() {
-			s.updateGrantedTermination(granted)
+			s.updateGrantedTerminationIfGen(gen, granted)
 		}
 	}
 }
@@ -52,6 +58,18 @@ func nextRenewInterval(granted time.Time, opts Options, now time.Time) time.Dura
 	return d
 }

+// nextRenewIntervalAfterError ignores GrantedTermination — by the time
+// renew has failed once, the grant is typically already in the past
+// and nextRenewInterval would floor to 1s, hammering the camera.
+// Recovery is the pull loop's reconnect path; we just need to not
+// accelerate retries past the configured RetryBackoff.
+func nextRenewIntervalAfterError(_ time.Time, opts Options, _ time.Time) time.Duration {
+	if opts.RetryBackoff > 0 {
+		return opts.RetryBackoff
+	}
+	return time.Second
+}
+
 // renewPullPoint sends Renew with an absolute UTC TerminationTime.
 // WS-BaseNotification §6.1.1 also allows xsd:duration but older
 // Hikvision, some Dahua and some Bosch firmwares reject the
--- a/event/stream/renew_test.go
+++ b/event/stream/renew_test.go
@@ -191,7 +191,7 @@ const renewFaultBody = `<env:Envelope xmlns:env="http://www.w3.org/2003/05/soap-
 func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) {
 	ref := subscriptionRef{
 		Address:      "http://192.168.1.10/onvif/services",
-		RefParamsXML: `<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>`,
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
 	}
 	fc := newFakeCaller()
 	_, err := renewPullPoint(fc, ref, defaultOptions())
@@ -202,3 +202,56 @@ func TestRenewPullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) {
 	assert.Contains(t, hdr, "297")
 	assert.Contains(t, hdr, `IsReferenceParameter="true"`)
 }
+
+// Regression: when GrantedTermination has just passed (renew failed
+// at or after the deadline), nextRenewInterval floors to one second
+// and the loop hammers the camera at 1 Hz until reconnect. Original
+// ticker design retried at the configured cadence regardless. After
+// a failure the loop must use a backoff decoupled from the stale
+// grant.
+func TestNextRenewIntervalAfterError_IgnoresStaleGrantedTermination(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	stale := now.Add(-500 * time.Millisecond)
+	opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second, RetryBackoff: time.Second}
+	got := nextRenewIntervalAfterError(stale, opts, now)
+	assert.GreaterOrEqual(t, got, opts.RetryBackoff,
+		"failure path must back off at least RetryBackoff, not 1s floor on stale grant")
+}
+
+func TestNextRenewIntervalAfterError_FallsBackToRetryBackoff(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	opts := Options{InitialTermination: 60 * time.Second, RenewMargin: 10 * time.Second, RetryBackoff: 5 * time.Second}
+	got := nextRenewIntervalAfterError(time.Time{}, opts, now)
+	assert.Equal(t, opts.RetryBackoff, got)
+}
+
+// Lost-update race: renew snapshots the ref, the SOAP call returns,
+// and meanwhile attemptRecreate replaced pullPoint with a fresh
+// subscription. If renew blindly writes the OLD subscription's
+// granted time onto the NEW subscription, the new schedule is wrong.
+// Update must be conditioned on "same subscription as when I read."
+func TestUpdateGrantedTermination_DropsWriteWhenSubscriptionRotated(t *testing.T) {
+	s := &Stream{}
+	original := subscriptionRef{Address: "http://camera/sub-A"}
+	s.setPullPoint(original)
+	gen := s.pullPointGen()
+
+	// Simulate recreate happening between snapshot and write.
+	s.setPullPoint(subscriptionRef{Address: "http://camera/sub-B"})
+
+	// Old generation's renew result must NOT overwrite sub-B's grant.
+	bogus := time.Date(1999, 1, 1, 0, 0, 0, 0, time.UTC)
+	s.updateGrantedTerminationIfGen(gen, bogus)
+
+	assert.True(t, s.getPullPoint().GrantedTermination.IsZero(),
+		"stale renew result must be discarded after a subscription rotation")
+}
+
+func TestUpdateGrantedTermination_AppliesWhenGenMatches(t *testing.T) {
+	s := &Stream{}
+	s.setPullPoint(subscriptionRef{Address: "http://camera/sub"})
+	gen := s.pullPointGen()
+	t1 := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	s.updateGrantedTerminationIfGen(gen, t1)
+	assert.Equal(t, t1, s.getPullPoint().GrantedTermination)
+}
--- a/event/stream/soap.go
+++ b/event/stream/soap.go
@@ -78,6 +78,12 @@ func extractTerminationTime(body string) time.Time {
 	return t
 }

+// terminationTimeRE matches the first <*:TerminationTime> in the body.
+// Only safe on responses that contain exactly one — currently
+// CreatePullPointSubscriptionResponse and RenewResponse via
+// extractTerminationTime. PullMessagesResponse also has a
+// TerminationTime element; do not call extractTerminationTime on pull
+// bodies.
 var terminationTimeRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?TerminationTime\b[^>]*>(.*?)</(?:[^:>\s]+:)?TerminationTime>`)

 // pullMessages returns an empty slice (no error) when the camera had
@@ -188,8 +194,14 @@ var (
 	soap12SubcodeRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Subcode\b[^>]*>.*?<(?:[^:>\s]+:)?Value[^>]*>(.*?)</(?:[^:>\s]+:)?Value>`)

 	// WS-Security blocks may carry our Username/Password if the camera
-	// echoes the request in a fault; scrub before logging.
-	wsseSecurityRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Security\b[^>]*>.*?</(?:[^:>\s]+:)?Security>`)
+	// echoes the request in a fault; scrub before logging. The
+	// alternation handles the truncated case where the read cap fell
+	// between <Security> and </Security>: in that case nothing past
+	// the opening tag is safe to retain. wssePasswordRE is the
+	// belt-and-braces fallback for non-conformant cameras emitting
+	// Password / UsernameToken outside a Security wrapper.
+	wsseSecurityRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Security\b[^>]*>(?:.*?</(?:[^:>\s]+:)?Security>|.*)`)
+	wssePasswordRE = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?Password\b[^>]*>.*?</(?:[^:>\s]+:)?Password>`)
 )

 // extractSOAPFault returns the reason text from a SOAP fault, falling
@@ -220,39 +232,28 @@ var (
 	refParamsRE       = regexp.MustCompile(`(?s)<(?:[^:>\s]+:)?ReferenceParameters\b[^>]*>(.*?)</(?:[^:>\s]+:)?ReferenceParameters>`)
 )

-// buildRefParamsHeader produces the SOAP <Header> inner XML for a set
-// of WS-Addressing ReferenceParameters: each ref-param element is
-// re-emitted with wsa:IsReferenceParameter="true" and any xmlns:*
-// it inherited from the parent <ReferenceParameters> element. Input
-// may be either the raw children or the full <*:ReferenceParameters>
-// wrapper — extractReferenceParameters returns the wrapper so parent-
-// scoped namespace declarations survive into the rebuild.
-func buildRefParamsHeader(rawXML string) (string, error) {
-	if strings.TrimSpace(rawXML) == "" {
+// buildRefParamsHeader produces the SOAP <Header> inner XML from the
+// full <*:ReferenceParameters> element returned by
+// extractReferenceParameters. Each child element is re-emitted with
+// wsa:IsReferenceParameter="true" added and any xmlns:* declared on
+// the parent inherited onto it (so the standalone child stays valid).
+// Empty input yields empty output.
+func buildRefParamsHeader(refParamsXML string) (string, error) {
+	if strings.TrimSpace(refParamsXML) == "" {
 		return "", nil
 	}
 	doc := etree.NewDocument()
-	if err := doc.ReadFromString("<wrap>" + rawXML + "</wrap>"); err != nil {
+	if err := doc.ReadFromString(refParamsXML); err != nil {
 		return "", fmt.Errorf("parse ref params: %w", err)
 	}
-	wrap := doc.SelectElement("wrap")
-	if wrap == nil {
-		return "", errors.New("parse ref params: missing wrap root")
+	wrapper := doc.Root()
+	if wrapper == nil || wrapper.Tag != "ReferenceParameters" {
+		return "", fmt.Errorf("ref params root must be <*:ReferenceParameters>, got <%s>", rootTag(wrapper))
 	}
-
-	children := wrap.ChildElements()
-	var ambient *etree.Element
-	if len(children) == 1 && strings.HasSuffix(children[0].Tag, "ReferenceParameters") {
-		ambient = children[0]
-		children = ambient.ChildElements()
-	}
-
 	var out strings.Builder
-	for _, child := range children {
+	for _, child := range wrapper.ChildElements() {
 		c := child.Copy()
-		if ambient != nil {
-			inheritXmlns(c, ambient)
-		}
+		inheritXmlns(c, wrapper)
 		c.CreateAttr("wsa:IsReferenceParameter", "true")
 		d := etree.NewDocument()
 		d.SetRoot(c)
@@ -265,6 +266,13 @@ func buildRefParamsHeader(rawXML string) (string, error) {
 	return out.String(), nil
 }

+func rootTag(e *etree.Element) string {
+	if e == nil {
+		return ""
+	}
+	return e.Tag
+}
+
 // inheritXmlns copies xmlns / xmlns:* declarations from src onto dst
 // when dst doesn't already declare them, so a child whose namespace
 // prefix was declared on an ancestor stays valid in isolation.
@@ -332,6 +340,7 @@ func enrichSOAPErr(resp *http.Response, err error) error {
 		return err
 	}
 	body := wsseSecurityRE.ReplaceAllString(string(b), "<Security>[REDACTED]</Security>")
+	body = wssePasswordRE.ReplaceAllString(body, "<Password>[REDACTED]</Password>")
 	if reason := extractSOAPFault(body); reason != "" {
 		return fmt.Errorf("SOAP fault: %s: %w", reason, err)
 	}
--- a/event/stream/soap_test.go
+++ b/event/stream/soap_test.go
@@ -315,7 +315,7 @@ func TestCreatePullPoint_VendorWithoutRefParams_RefParamsEmpty(t *testing.T) {
 func TestPullMessages_EchoesRefParamsWithIsReferenceParameterAttribute(t *testing.T) {
 	ref := subscriptionRef{
 		Address:      "http://192.168.1.10/onvif/services",
-		RefParamsXML: `<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>`,
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
 	}
 	fc := newFakeCaller()
 	_, err := pullMessages(fc, ref, defaultOptions())
@@ -349,7 +349,9 @@ func TestPullMessages_PostsToAddressFromRef(t *testing.T) {
 // --- Building the header XML from raw ref params ----------------------

 func TestBuildRefParamsHeader_AddsIsReferenceParameter(t *testing.T) {
-	raw := `<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>`
+	raw := `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+		`<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>` +
+		`</wsa:ReferenceParameters>`
 	got, err := buildRefParamsHeader(raw)
 	require.NoError(t, err)
 	assert.Contains(t, got, "SubscriptionId")
@@ -360,7 +362,9 @@ func TestBuildRefParamsHeader_AddsIsReferenceParameter(t *testing.T) {
 }

 func TestBuildRefParamsHeader_MultipleTopLevelChildren(t *testing.T) {
-	raw := `<a:Foo xmlns:a="ns/a">1</a:Foo><b:Bar xmlns:b="ns/b">2</b:Bar>`
+	raw := `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+		`<a:Foo xmlns:a="ns/a">1</a:Foo><b:Bar xmlns:b="ns/b">2</b:Bar>` +
+		`</wsa:ReferenceParameters>`
 	got, err := buildRefParamsHeader(raw)
 	require.NoError(t, err)
 	assert.Equal(t, 2, strings.Count(got, `IsReferenceParameter="true"`),
@@ -378,7 +382,7 @@ func TestBuildRefParamsHeader_EmptyInputReturnsEmpty(t *testing.T) {
 func TestUnsubscribePullPoint_EchoesRefParamsWithIsReferenceParameter(t *testing.T) {
 	ref := subscriptionRef{
 		Address:      "http://192.168.1.10/onvif/services",
-		RefParamsXML: `<dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId>`,
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing"><dom0:SubscriptionId xmlns:dom0="http://www.axis.com/2009/event">297</dom0:SubscriptionId></wsa:ReferenceParameters>`,
 	}
 	fc := newFakeCaller()
 	require.NoError(t, unsubscribePullPoint(fc, ref))
@@ -401,8 +405,10 @@ func TestUnsubscribePullPoint_EmptyAddressIsNoOp(t *testing.T) {
 func TestPullMessages_TwoRefParamsEachLandsOnTheWire(t *testing.T) {
 	ref := subscriptionRef{
 		Address: "http://camera/sub",
-		RefParamsXML: `<a:Foo xmlns:a="ns/a">1</a:Foo>` +
-			`<b:Bar xmlns:b="ns/b">2</b:Bar>`,
+		RefParamsXML: `<wsa:ReferenceParameters xmlns:wsa="http://www.w3.org/2005/08/addressing">` +
+			`<a:Foo xmlns:a="ns/a">1</a:Foo>` +
+			`<b:Bar xmlns:b="ns/b">2</b:Bar>` +
+			`</wsa:ReferenceParameters>`,
 	}
 	fc := newFakeCaller()
 	_, err := pullMessages(fc, ref, defaultOptions())
@@ -601,3 +607,76 @@ func TestBuildRefParamsHeader_WhitespaceOnlyReturnsEmpty(t *testing.T) {
 	require.NoError(t, err)
 	assert.Empty(t, got)
 }
+
+// Worst case: the response body's <Security> block starts within the
+// 64 KiB error cap but its </Security> is past it. The non-greedy
+// regex needs a close tag — without one the redaction misses and
+// raw Username/Password reaches the excerpt. Verify the helper
+// strips from <Security to EOF when no close tag is present.
+func TestEnrichSOAPErr_RedactsSecurityBlockMissingCloseTag(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x"><env:Header>` +
+		`<wsse:Security xmlns:wsse="y">` +
+		`<wsse:Username>admin</wsse:Username>` +
+		`<wsse:Password>hunter2</wsse:Password>` +
+		// no </wsse:Security> — simulates a Security block truncated
+		// at the 64 KiB read cap.
+		strings.Repeat("padding ", 1000)
+	got := enrichSOAPErr(fakeResponse(body), errors.New("500"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2",
+		"truncated Security block must not leak Password to the excerpt")
+	assert.NotContains(t, got.Error(), "admin",
+		"truncated Security block must not leak Username to the excerpt")
+}
+
+// The wrapper-only contract means an element whose local name merely
+// ends in "ReferenceParameters" cannot be mistaken for the wrapper —
+// the root must be exactly <*:ReferenceParameters>. Anything else is
+// a contract violation by the caller and surfaces as an error.
+func TestBuildRefParamsHeader_RejectsNonReferenceParametersRoot(t *testing.T) {
+	raw := `<my:MyReferenceParameters xmlns:my="urn:vendor:my">X</my:MyReferenceParameters>`
+	_, err := buildRefParamsHeader(raw)
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "ReferenceParameters")
+}
+
+// A non-conformant camera echoing UsernameToken/Password outside a
+// <Security> wrapper would still leak credentials through the body
+// excerpt. Belt-and-braces: redact Password elements directly too.
+func TestEnrichSOAPErr_RedactsBarePasswordElement(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x"><env:Body>` +
+		`<wsse:UsernameToken xmlns:wsse="y">` +
+		`<wsse:Username>admin</wsse:Username>` +
+		`<wsse:Password>hunter2</wsse:Password>` +
+		`</wsse:UsernameToken>` +
+		`</env:Body></env:Envelope>`
+	got := enrichSOAPErr(fakeResponse(body), errors.New("400"))
+	require.Error(t, got)
+	assert.NotContains(t, got.Error(), "hunter2",
+		"Password must be redacted regardless of whether it's wrapped in Security")
+}
+
+// --- Edge-case coverage flagged in review -----------------------------
+
+func TestExtractTerminationTime_MalformedDateYieldsZero(t *testing.T) {
+	body := `<env:Body><wsnt:TerminationTime>not-a-date</wsnt:TerminationTime></env:Body>`
+	assert.True(t, extractTerminationTime(body).IsZero(),
+		"unparseable datetime must not panic and must not return a garbage time — fall back to opts")
+}
+
+func TestNextRenewInterval_MarginEqualsBaseFallsToHalf(t *testing.T) {
+	now := time.Date(2026, 5, 27, 13, 0, 0, 0, time.UTC)
+	opts := Options{InitialTermination: 30 * time.Second, RenewMargin: 30 * time.Second}
+	got := nextRenewInterval(time.Time{}, opts, now)
+	assert.Equal(t, 15*time.Second, got,
+		"when margin == base, the helper must fall through to base/2 rather than the 1s floor")
+}
+
+func TestExtractReferenceParameters_EmptySubscriptionReferenceReturnsEmpty(t *testing.T) {
+	body := `<env:Envelope xmlns:env="x" xmlns:tev="y">
+  <env:Body><tev:CreatePullPointSubscriptionResponse>
+    <tev:SubscriptionReference/>
+  </tev:CreatePullPointSubscriptionResponse></env:Body>
+</env:Envelope>`
+	assert.Empty(t, extractReferenceParameters(body))
+}
--- a/event/stream/stream.go
+++ b/event/stream/stream.go
@@ -165,8 +165,9 @@ type Stream struct {
 	caller caller
 	opts   Options

-	pullPointMu sync.Mutex
-	pullPoint   subscriptionRef
+	pullPointMu  sync.Mutex
+	pullPoint    subscriptionRef
+	gen uint64 // bumped on every setPullPoint so renews can detect a recreate

 	events chan Event
 	errors chan error
@@ -191,11 +192,29 @@ func (s *Stream) setPullPoint(ref subscriptionRef) {
 	s.pullPointMu.Lock()
 	defer s.pullPointMu.Unlock()
 	s.pullPoint = ref
+	s.gen++
 }

-func (s *Stream) updateGrantedTermination(t time.Time) {
+// pullPointGen returns the current generation. Pair with
+// updateGrantedTerminationIfGen so a renew result issued against a
+// subscription that was rotated mid-flight (recreate path) is
+// discarded instead of overwriting the new subscription's grant.
+func (s *Stream) pullPointGen() uint64 {
 	s.pullPointMu.Lock()
 	defer s.pullPointMu.Unlock()
+	return s.gen
+}
+
+// updateGrantedTerminationIfGen writes the granted time only when the
+// caller's snapshot is still current. Use setPullPoint to replace the
+// full ref; this updates GrantedTermination in place after a renew
+// response.
+func (s *Stream) updateGrantedTerminationIfGen(gen uint64, t time.Time) {
+	s.pullPointMu.Lock()
+	defer s.pullPointMu.Unlock()
+	if s.gen != gen {
+		return
+	}
 	s.pullPoint.GrantedTermination = t
 }

--- a/event/stream/stream_test.go
+++ b/event/stream/stream_test.go
@@ -125,6 +125,10 @@ func (f *fakeCaller) sendSoapCallCount() int {

 // createPullPointResp is the minimal SOAP envelope the lib's existing
 // xml.Decoder + getXMLNode path can extract a pull-point address from.
+// Intentionally omits <TerminationTime> so renewLoop falls back to
+// opts.InitialTermination — tests that drive renew timing depend on
+// that path. Tests that need the camera-granted termination capture
+// path use a dedicated fixture instead.
 const createPullPointResp = `<?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"