fix: stop server-side stream replay producer when the SDK client disconnects

[kinyoklion]

Summary

The server-side stream replay path (serverSideEnvStreamRepository.Replay, backing /sdk/stream for FDv2 and /all for FDv1) leaks a goroutine per SDK client that disconnects mid-replay.

Replay returns an unbuffered channel and spawns a producer goroutine that sends the replay events on it. Under backpressure — a slow or stalled SDK client — the producer parks on out <- event while the eventsource connection handler is busy writing to the socket. If the client then disconnects, the handler stops reading the channel, and because Replay has no cancellation hook, the producer is stranded on that send until the process exits. The goroutine and the replay payload it holds leak.

Change

Adopt the new optional eventsource.RepositoryWithContext extension:

• serverSideEnvStreamRepository now implements ReplayWithContext(ctx, channel, id). The eventsource server calls it with the subscribing request's context, which is cancelled on disconnect. The send loop selects on ctx.Done(), so the producer returns immediately instead of blocking on a send nobody will receive.
• Replay is retained (it delegates to the same logic with a background context) to satisfy the eventsource.Repository interface; the server prefers ReplayWithContext when a repository implements it.

The shared replay helper keeps the existing IsInitialized short-circuit and singleflight behavior unchanged.

Dependency

This depends on launchdarkly/eventsource#63, which adds RepositoryWithContext (plus a handler-side background drain that unblocks any Repository producer, even those that don't adopt the context). That change is now released: go.mod points at the tagged eventsource v1.11.2, which contains #63 along with the once-per-batch replay flush (launchdarkly/eventsource#64) and the CI fixes (launchdarkly/eventsource#66). The earlier pseudo-version pin of the PR branch is gone.

Testing

• New unit test: ReplayWithContext stops producing (channel closes) promptly when the subscriber's context is cancelled without a reader — before context propagation this producer would block forever.
• go test -race ./internal/streams/... and ./relay/... pass, re-run against the released v1.11.2.
• End to end against mockld (a ~3MB, 6500-flag dataset) with a non-reading client that stalls the socket:
  • Preconditions reproduced: the producer goroutine parks on the channel send while the handler blocks in a socket write (real TCP backpressure).
  • Fixed build: the replay producer goroutine exits within ~0.6s of a client FIN.
  • Stock build (eventsource v1.11.0): the producer is still blocked on the channel send 3s after the client disconnects (handler already gone) — the leak.

---

Note

Medium Risk
Touches hot-path SSE replay for all server-side SDK connections; behavior change is limited to cleanup on disconnect, but large replay payloads and timing-sensitive send loops warrant careful review.

Overview
Fixes a goroutine leak on server-side SDK streams (/sdk/stream, /all) when a client disconnects while replay is still sending on an unbuffered channel—the producer could block forever on out <- event with no reader.

serverSideEnvStreamRepository now implements eventsource.RepositoryWithContext: shared replay logic uses the subscribe request context (cancelled on disconnect), bails before building the snapshot if already cancelled, and selects on ctx.Done() when sending events. Legacy Replay delegates to the same helper with context.Background().

Dependencies: github.com/launchdarkly/eventsource v1.11.0 → v1.11.2 (adds RepositoryWithContext); klauspost/compress patch bump in lockfile.

Adds a unit test that cancels context mid-replay with no consumer and asserts the channel closes without delivering events.

^{Reviewed by Cursor Bugbot for commit 9997de9. Bugbot is set up for automated code reviews on this repo. Configure here.}

[cursor]

Cursor Bugbot has reviewed your changes using default effort and found 1 potential issue.

^{❌ Bugbot Autofix is OFF. To automatically fix reported issues with cloud agents, have a team admin enable autofix in the Cursor dashboard.}

^{Reviewed by Cursor Bugbot for commit 3a9ff12. Configure here.}

[aaron-zeisler]

👍 The code in the production file looks good.

My Claude also reported the test that Bugbot pointed out: that specific test passes even if the code changes are reverted. Below is the recommended test from my agent:

t.Run("ReplayWithContext stops producing when the subscriber's context is cancelled", func(t *testing.T) {
    snapshotReturned := make(chan struct{}, 1)
    underlyingQuery := queryThatIncrementsFlagVersionOnEachCall()
    store := newMockStoreQueries()
    store.setupSnapshotFn(func() (map[ldstoretypes.DataKind][]ldstoretypes.KeyedItemDescriptor, subsystems.Selector, error) {
        data, selector, err := underlyingQuery()
        snapshotReturned <- struct{}{}
        return data, selector, err
    })
    repo := &serverSideEnvStreamRepository{store: store, logger: slog.Default()}

    ctx, cancel := context.WithCancel(context.Background())
    eventCh := repo.ReplayWithContext(ctx, "", "")

    // Producer has built its payload and is parked on the unbuffered, unread send.
    <-snapshotReturned
    time.Sleep(50 * time.Millisecond)
    cancel()
    // Let the producer observe cancellation and return *before* any reader engages, so the
    // send-loop select has only ctx.Done() ready (deterministic — no out<-event / ctx race).
    time.Sleep(50 * time.Millisecond)

    // The fix delivers ZERO events and closes. Reverting the send-loop ctx.Done() case makes the
    // producer deliver its event first, which AssertChannelClosed rejects.
    require.True(t, helpers.AssertChannelClosed(t, eventCh, time.Second),
        "replay channel should close with no event delivered after cancellation")
})

[aaron-zeisler]

It looks like the CI steps are failing because:

1. There's a flaky test that's unrelated to this PR: TestCapturesSimpleStreamDuration in internal/metrics/usage_test.go
2. A CVE warning in klauspost/compress v1.18.5. It can be upgraded to 1.18.7 to fix the CVE