test(discovery): cover the event-stream leak invariant directly

The 47b445d fix landed with three tests against consumeEventStream, but
the actual leak lived one layer up — eventLoop re-calling cli.Events
per message. Those tests would not have caught a regression at that
exact layer.

Refactor eventLoop into a thin glue + runEventLoop(ctx, source,
consume). source and consume are injectable, so the new test
TestRunEventLoop_OnlyReopensAfterStreamEnds can count source calls
directly: 5 messages must produce exactly 1 source call, then a stream
close must produce a 2nd. The pre-fix shape would have requested a
fresh stream after every message and failed this loudly.

Also:

- consumeEventStream now takes errBackoff as a parameter (production
  passes a new streamErrBackoff = 2*time.Second constant; tests pass
  1ms). TestConsumeEventStream_ErrSignalRequestsRetry no longer relies
  on a racy ctx-timeout — it asserts retry=true && err==nil cleanly.
- Doc comment dropped the misleading "F-46 regression" framing. The
  bug predates F-46 (lived in eventLoop since the initial skeleton);
  the doc now points at the fixing commit 47b445d instead.

Verified: go test ./internal/discovery/... -race -count=3 clean.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

Author: AlexCherrypi

internal/discovery/discovery.go

@@ -166,22 +166,37 @@ func (d *Discoverer) pollLoop(ctx context.Context) {
 	}
 }
 
+// streamErrBackoff is the pause after the docker event stream emits
+// an error before we reopen. Production value; tests pass shorter
+// durations directly into consumeEventStream.
+const streamErrBackoff = 2 * time.Second
+
 func (d *Discoverer) eventLoop(ctx context.Context) error {
 	f := buildEventFilter(d.discriminator)
+	source := func(ctx context.Context) (<-chan events.Message, <-chan error) {
+		return d.cli.Events(ctx, events.ListOptions{Filters: f})
+	}
+	return runEventLoop(ctx, source, d.consumeEvents)
+}
 
+// runEventLoop is the reconnect-loop: open a stream via source, drain
+// it via consume until ctx cancels or the stream ends, reopen only
+// then. Extracted from eventLoop so tests can verify the central
+// invariant — source is called once per real disconnect, NOT once per
+// event. The pre-fix shape placed source() inside the inner select
+// and re-called it on every message, opening a fresh long-poll HTTP
+// request to docker(-proxy) per event while the prior request's
+// goroutine stayed parked on the old (still-valid-ctx) connection.
+// On busy event sources (Frigate watchdog cycling ffmpeg subprocesses)
+// the leaked sockets exhaust the kernel's tcp_mem.
+func runEventLoop(
+	ctx context.Context,
+	source func(context.Context) (<-chan events.Message, <-chan error),
+	consume func(context.Context, <-chan events.Message, <-chan error) (bool, error),
+) error {
 	for {
-		msgs, errs := d.cli.Events(ctx, events.ListOptions{Filters: f})
-		// consumeEvents stays on the SAME (msgs, errs) pair until the
-		// stream ends (errs/closed) or ctx is cancelled. Earlier
-		// versions exited the inner select after every single message
-		// and re-called cli.Events, which opens a fresh long-poll HTTP
-		// request to the docker daemon each time but never closes the
-		// previous one — the prior goroutine stays parked on the old
-		// connection. On busy event sources (e.g. a Frigate watchdog
-		// restarting ffmpeg subprocesses) anchord then accumulates
-		// hundreds of ESTAB sockets to docker(-proxy) until the kernel
-		// hits its tcp_mem ceiling.
-		retry, err := d.consumeEvents(ctx, msgs, errs)
+		msgs, errs := source(ctx)
+		retry, err := consume(ctx, msgs, errs)
 		if err != nil {
 			return err
 		}
@@ -202,20 +217,30 @@ func (d *Discoverer) consumeEvents(ctx context.Context, msgs <-chan events.Messa
 		if err := d.snapshot(ctx); err != nil {
 			slog.Warn("event-driven snapshot failed", "err", err)
 		}
-	})
+	}, streamErrBackoff)
 }
 
 // consumeEventStream is the pure consume-loop. Behaviour contract:
 //   - ctx cancelled    → returns (false, ctx.Err())
-//   - errs delivers    → returns (true, nil) after a 2s backoff
+//   - errs delivers    → returns (true, nil) after errBackoff
 //   - msgs is closed   → returns (true, nil) immediately
 //   - msgs message     → onMessage() invoked, loop continues on SAME stream
 //
 // The invariant: while messages keep flowing, the function does NOT
 // return — callers must not re-enter to reopen a fresh stream per
 // message. That pattern leaked the long-poll HTTP request to
-// docker(-proxy) on every event (F-46 regression).
-func consumeEventStream(ctx context.Context, msgs <-chan events.Message, errs <-chan error, onMessage func()) (retry bool, err error) {
+// docker(-proxy) on every event (see commit 47b445d).
+//
+// errBackoff is a parameter, not a constant, so tests can pass a tiny
+// duration and assert retry semantics deterministically without
+// waiting two real seconds.
+func consumeEventStream(
+	ctx context.Context,
+	msgs <-chan events.Message,
+	errs <-chan error,
+	onMessage func(),
+	errBackoff time.Duration,
+) (retry bool, err error) {
 	for {
 		select {
 		case <-ctx.Done():
@@ -228,7 +253,7 @@ func consumeEventStream(ctx context.Context, msgs <-chan events.Message, errs <-
 			select {
 			case <-ctx.Done():
 				return false, ctx.Err()
-			case <-time.After(2 * time.Second):
+			case <-time.After(errBackoff):
 			}
 			return true, nil
 		case msg, ok := <-msgs:

internal/discovery/discovery_test.go

@@ -407,6 +407,10 @@ func TestParseIP(t *testing.T) {
 	}
 }
 
+// testBackoff is short enough that ErrSignalRequestsRetry can wait the
+// full backoff and assert retry=true deterministically.
+const testBackoff = 1 * time.Millisecond
+
 // Regression for the docker-socket-proxy connection leak: a single
 // open stream must serve many messages — consumeEventStream must not
 // return between messages, only on stream-end or ctx cancellation.
@@ -431,7 +435,7 @@ func TestConsumeEventStream_StaysOnSameStreamAcrossMessages(t *testing.T) {
 
 	retry, err := consumeEventStream(ctx, msgs, errs, func() {
 		calls.Add(1)
-	})
+	}, testBackoff)
 	if err != nil {
 		t.Fatalf("unexpected error: %v", err)
 	}
@@ -450,7 +454,7 @@ func TestConsumeEventStream_CtxCancelStopsLoop(t *testing.T) {
 	ctx, cancel := context.WithCancel(context.Background())
 	cancel() // pre-cancelled
 
-	retry, err := consumeEventStream(ctx, msgs, errs, func() {})
+	retry, err := consumeEventStream(ctx, msgs, errs, func() {}, testBackoff)
 	if retry {
 		t.Error("retry should be false when ctx cancelled — caller should not reopen")
 	}
@@ -462,21 +466,103 @@ func TestConsumeEventStream_CtxCancelStopsLoop(t *testing.T) {
 func TestConsumeEventStream_ErrSignalRequestsRetry(t *testing.T) {
 	msgs := make(chan events.Message)
 	errs := make(chan error, 1)
-
-	// Use a context with a small timeout so the post-error backoff
-	// returns quickly without holding the test up for 2 s.
-	ctx, cancel := context.WithTimeout(context.Background(), 100*time.Millisecond)
+	ctx, cancel := context.WithCancel(context.Background())
 	defer cancel()
 
 	errs <- errors.New("stream interrupted")
 
 	retry, err := consumeEventStream(ctx, msgs, errs, func() {
 		t.Error("onMessage must not be called on errs path")
-	})
-	// Backoff hit ctx-cancel rather than completing — that's fine, the
-	// real-world caller already received the "retry after backoff"
-	// instruction and will simply observe ctx.Err() next.
-	if err == nil && !retry {
-		t.Error("on err with no ctx-cancel mid-backoff we expect retry=true")
+	}, testBackoff)
+	if err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !retry {
+		t.Error("retry should be true after errs signal so caller reopens")
+	}
+}
+
+// Regression for the docker-socket-proxy connection leak at the
+// reconnect-loop layer: source must be called once per *real*
+// disconnect, NOT once per event. The pre-fix shape opened a fresh
+// long-poll HTTP request on every message and leaked the previous
+// one's goroutine. This test would have failed loudly before commit
+// 47b445d (a 2nd stream would have been requested after the first
+// message).
+func TestRunEventLoop_OnlyReopensAfterStreamEnds(t *testing.T) {
+	type stream struct {
+		msgs chan events.Message
+		errs chan error
+	}
+	streams := make(chan *stream, 4)
+
+	source := func(ctx context.Context) (<-chan events.Message, <-chan error) {
+		s := &stream{
+			msgs: make(chan events.Message, 16),
+			errs: make(chan error, 1),
+		}
+		streams <- s
+		return s.msgs, s.errs
+	}
+
+	var eventCalls atomic.Int32
+	consume := func(ctx context.Context, msgs <-chan events.Message, errs <-chan error) (bool, error) {
+		for {
+			select {
+			case <-ctx.Done():
+				return false, ctx.Err()
+			case _, ok := <-msgs:
+				if !ok {
+					return true, nil // signal reopen on stream end
+				}
+				eventCalls.Add(1)
+			}
+		}
+	}
+
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	done := make(chan error, 1)
+	go func() { done <- runEventLoop(ctx, source, consume) }()
+
+	// First stream is opened eagerly.
+	s1 := <-streams
+
+	// Pump 5 events through the SAME stream.
+	for i := 0; i < 5; i++ {
+		s1.msgs <- events.Message{Action: "start", Actor: events.Actor{ID: "abcdef0123456789"}}
+	}
+	waitFor(t, 2*time.Second, func() bool { return eventCalls.Load() == 5 })
+
+	// Central assertion: no new stream was requested while events were flowing.
+	select {
+	case <-streams:
+		t.Fatal("source was called a second time mid-stream — leak fix regressed")
+	case <-time.After(50 * time.Millisecond):
+	}
+
+	// Close the stream → source MUST be called a second time.
+	close(s1.msgs)
+	select {
+	case <-streams:
+		// expected
+	case <-time.After(2 * time.Second):
+		t.Fatal("source was not called again after stream close — caller failed to reopen")
+	}
+
+	cancel()
+	<-done
+}
+
+func waitFor(t *testing.T, timeout time.Duration, cond func() bool) {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if cond() {
+			return
+		}
+		time.Sleep(1 * time.Millisecond)
 	}
+	t.Fatalf("timed out waiting for condition after %s", timeout)
 }