fix(event/stream): bound Close drain to survive a hung HTTP caller

Concurrency audit (third review) flagged that caller.SendSoap is not
ctx-aware: cancelling ctx does not unblock a pull or renew goroutine
parked in the underlying http.Client.Do. The previous Close()
unconditionally did <-s.done before its 5s unsubscribe timeout, so a
wedged SendSoap could hang Close indefinitely — taking the agent's
shutdown down with it.

Adds closeDrainTimeout (5s) to bound the wait for the run goroutines
to exit. When the drain times out:
  * Close returns a 'did not drain' error so the caller can move on.
  * Unsubscribe is skipped; the subscription expires at the camera
    once InitialTermination elapses without a Renew.
  * The wedged goroutines exit later, when the HTTP transport
    eventually gives up. They are effectively leaked until then —
    documented in the caller interface comment as the contract
    callers must accept (or fix, by configuring an http.Client.Timeout).

The caller interface doc-comment now states both invariants
explicitly: must be goroutine-safe AND must enforce its own per-
request timeout, because we cannot from here.

Test
----
TestClose_BoundedWhenLoopsStuckOnHungHTTP: drives the fakeCaller
with blockAllSendSoap (new flag) so every SendSoap parks. Waits for
pullLoop to actually reach the blocked SendSoap before calling
Close (a race the previous attempt had: Close raced the loop and
exited via the ctx pre-check). Asserts Close returns within
closeDrainTimeout + 2s slack with a drain-timeout error.

Other concurrency audit findings disposition
--------------------------------------------
* unsubscribe goroutine leaks past 5s: intentional, already
  documented at closeUnsubscribeTimeout.
* now func() time.Time data race: written once before goroutines
  start; safe by happens-before. Tests do not swap it today.
* closeOnce self-deadlock if Close called from inside a loop:
  no path exists; not exposed via the API.

event/stream/stream.go

@@ -10,6 +10,14 @@ import (
 	"github.com/kerberos-io/onvif"
 )
 
+// closeDrainTimeout bounds Close's wait for the pull and renew
+// goroutines to exit. The loops block in caller.SendSoap which is not
+// ctx-aware (the underlying http.Client is the only thing that can
+// unblock them — see caller below). On a hung HTTP transport Close
+// would otherwise wait forever; instead it returns an error and lets
+// the calling agent move on.
+const closeDrainTimeout = 5 * time.Second
+
 // closeUnsubscribeTimeout bounds the Unsubscribe SOAP call issued by
 // Close. A subscription expires at the camera once InitialTermination
 // elapses without a renew, so a missed unsubscribe is at worst
@@ -103,8 +111,17 @@ func (o Options) withDefaults() Options {
 }
 
 // caller is the *onvif.Device subset Stream depends on. Implementations
-// must be safe for concurrent use — pull and renew goroutines call in
-// from separate goroutines. *onvif.Device satisfies this via http.Client.
+// must:
+//
+//   - Be safe for concurrent use — pull and renew goroutines call in
+//     from separate goroutines. *onvif.Device satisfies this via
+//     http.Client.
+//   - Enforce a per-request timeout via the underlying HTTP client.
+//     The methods do not take a ctx, so ctx-cancel cannot interrupt a
+//     hung request; only the HTTP client's own timeout can. Close
+//     bounds its drain wait at closeDrainTimeout to survive a misbehaving
+//     caller, but a leaking goroutine remains until the HTTP call
+//     eventually returns.
 type caller interface {
 	CallMethod(method any) (*http.Response, error)
 	SendSoap(endpoint, body string) (*http.Response, error)
@@ -194,15 +211,24 @@ func (s *Stream) Events() <-chan Event { return s.events }
 // when the Stream stops.
 func (s *Stream) Errors() <-chan error { return s.errors }
 
-// Close stops the background goroutines, waits for them to exit and
-// Unsubscribes from the camera. Subsequent calls are no-ops.
+// Close stops the background goroutines, waits up to closeDrainTimeout
+// for them to exit, and then Unsubscribes from the camera (also bounded,
+// by closeUnsubscribeTimeout). Subsequent calls are no-ops.
 //
-// Unsubscribe is bounded by closeUnsubscribeTimeout so a hung camera
-// connection cannot wedge the caller.
+// If the drain times out the goroutines are likely wedged inside a
+// non-ctx-aware caller.SendSoap; they will exit on their own once the
+// HTTP call returns. Unsubscribe is skipped in that case — the
+// subscription expires at the camera anyway.
 func (s *Stream) Close() error {
 	s.closeOnce.Do(func() {
 		s.cancel()
-		<-s.done
+
+		select {
+		case <-s.done:
+		case <-time.After(closeDrainTimeout):
+			s.closeErr = fmt.Errorf("close: pull/renew loops did not drain within %s (likely stuck in caller HTTP)", closeDrainTimeout)
+			return
+		}
 
 		errCh := make(chan error, 1)
 		go func() {

event/stream/stream_test.go

@@ -22,8 +22,9 @@ import (
 // require tests to enumerate every call.
 //
 // blockUnsubscribe, when non-nil, causes SendSoap calls whose body
-// contains "Unsubscribe" to block until the channel is closed. Used to
-// verify Close's timeout path.
+// contains "Unsubscribe" to block until the channel is closed.
+// blockAllSendSoap, when non-nil, blocks every SendSoap call until
+// closed (simulates a hung HTTP transport).
 type fakeCaller struct {
 	mu               sync.Mutex
 	callMethodResps  []fakeResp
@@ -33,6 +34,7 @@ type fakeCaller struct {
 	callMethodCalls  []any
 	sendSoapCalls    [][2]string
 	blockUnsubscribe chan struct{}
+	blockAllSendSoap chan struct{}
 }
 
 type fakeResp struct {
@@ -84,8 +86,12 @@ func (f *fakeCaller) SendSoap(endpoint, body string) (*http.Response, error) {
 		f.sendSoapResps = f.sendSoapResps[1:]
 	}
 	block := f.blockUnsubscribe
+	blockAll := f.blockAllSendSoap
 	f.mu.Unlock()
 
+	if blockAll != nil {
+		<-blockAll
+	}
 	if block != nil && strings.Contains(body, "Unsubscribe") {
 		<-block
 	}
@@ -444,3 +450,48 @@ func TestFakeCaller_QueueThenDefaultFallback(t *testing.T) {
 	assert.Contains(t, string(b3[:n]), "PullMessagesResponse",
 		"default SendSoap should be an empty PullMessagesResponse envelope")
 }
+
+func TestClose_BoundedWhenLoopsStuckOnHungHTTP(t *testing.T) {
+	// Simulates a hung HTTP transport: every SendSoap blocks
+	// indefinitely. The pull and renew loops are wedged inside
+	// SendSoap and ctx-cancel cannot unblock them. Close must still
+	// return within its bounded budget so the agent's shutdown does
+	// not hang.
+	fc := newFakeCaller()
+	fc.queueCallMethod(createPullPointResp, nil)
+	blockAll := make(chan struct{})
+	defer close(blockAll)
+	fc.mu.Lock()
+	fc.blockAllSendSoap = blockAll
+	fc.mu.Unlock()
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+	s, err := newStream(ctx, fc, Options{
+		PullTimeout:        100 * time.Millisecond,
+		InitialTermination: 30 * time.Second,
+	})
+	require.NoError(t, err)
+
+	// Wait until pullLoop is actually parked inside the blocked
+	// SendSoap. Without this, Close races with the loop's first
+	// iteration and exits via the ctx pre-check instead of
+	// exercising the drain-timeout path.
+	deadline := time.Now().Add(time.Second)
+	for time.Now().Before(deadline) && fc.sendSoapCallCount() == 0 {
+		time.Sleep(10 * time.Millisecond)
+	}
+	require.GreaterOrEqual(t, fc.sendSoapCallCount(), 1, "pullLoop never reached SendSoap")
+
+	start := time.Now()
+	err = s.Close()
+	elapsed := time.Since(start)
+
+	require.Error(t, err)
+	assert.Contains(t, err.Error(), "drain", "expected a drain-timeout error")
+	// Total budget is closeDrainTimeout for the wait + ~0 for unsubscribe
+	// (which is skipped when drain times out). Give plenty of slack for
+	// scheduling on a loaded CI machine.
+	assert.Less(t, elapsed, closeDrainTimeout+2*time.Second,
+		"Close exceeded bound (%s); expected ~%s", elapsed, closeDrainTimeout)
+}