fix: unblock Repository producers when a subscriber disconnects mid-replay - #63
Merged
Merged
Conversation
…eplay When a Repository is registered for a channel, the server calls Replay on each new subscription and drains the returned channel from the per-connection handler goroutine. If the subscriber disconnects while the handler is still consuming that batch, the handler breaks out of its read loop and abandons the channel without draining it -- and it cannot close it, because it holds the receiving end. Replay itself has no cancellation hook (two strings in, a bare channel out), so the producer goroutine blocks forever on its next `out <- event` send, leaking the goroutine and its payload until the process exits. This fixes it at two levels: 1. Background drain (no API change, works for every Repository): when the handler exits its read loop while still mid-batch, it drains the remaining events in a throwaway goroutine so the producer unblocks as fast as it can emit and releases in milliseconds. 2. RepositoryWithContext (optional extension): a Repository may implement ReplayWithContext(ctx, channel, id), which the server prefers when present, passing the subscribing request's context. That context is cancelled on disconnect, so an adopting producer can select on ctx.Done() and stop promptly and cleanly. Repositories that implement only Replay are unaffected and continue to be served via the drain safety net. Replay remains for backward compatibility; the new interface embeds Repository and is selected by type assertion, so no existing consumer or exported API breaks. Tests reproduce the leak (a producer blocked on send is stranded on disconnect) and verify it is fixed for both the plain-Replay drain path and the ReplayWithContext path, across clean FIN and abrupt RST, plus normal delivery, context-cancellation observation, method preference, empty batches, multiple concurrent subscriptions, and server close during replay. Full suite passes under -race.
…t helper Factor the background drain into drainReplayedEvents, and also drain the batch channel in Server.run() when the initial trySend to a subscriber fails (the subscriber was already closed and will never consume the batch, so its producer would otherwise block forever). This broadens the safety net beyond the mid-batch-consume case already covered after the handler's read loop.
Match the existing channel-drain convention (stream.go) with a //nolint:revive directive; golangci-lint run is clean.
kinyoklion
commented
Jul 27, 2026
| // | ||
| // Implementing this interface is optional and does not change the behavior of Replay. A Repository | ||
| // that implements only Replay continues to work unchanged; a Repository that implements both must | ||
| // still provide Replay for backward compatibility. |
Member
Author
There was a problem hiding this comment.
Not sure how I feel about this. Maybe not super avoidable for now, but we should remove the non-context version and major version this library when it is reasonable to do so.
…andoffs The handler-exit and trySend-failure drains did not cover every way a replay batch can be abandoned. The Server now records the batch channel it hands to each subscription (run()-goroutine-only state) and drains it wherever the subscription can no longer be relied on to consume it: - unsubscription: the handler exited (possibly before ever dequeuing the batch from its buffered event channel, or partway through it) - handler exit: a sweep of the already-buffered event channel covers batches the Server enqueued that run() will never see unsubscribed (shutdown) - shutdown: queued unsubscriptions are drained first, then departed subscribers' batches. A subscriber that is still connected is deliberately NOT drained: its handler continues delivering the buffered batch after close(out), and a concurrent drain would steal events from that delivery (fence tests cover both directions)
…ter Close A handler that exits abnormally (write error, client disconnect) after the Server has shut down sends on unsubs, which nothing consumes anymore and whose buffer holds only two entries; each additional such handler blocked forever, pinning its connection open. Handlers now select the unsubscription send against a stopped channel closed when run() exits. Also documents why the forceDisconnect unregistration path needs no batch drain of its own: with the server still running, a buffered channel delivers queued values before reporting closed, so the handler consumes an in-flight batch itself, and abnormal exits are covered by the handler's exit paths plus the unsubscription drain.
Keeps (*Server).Handler under the gocyclo limit and gives the two abandoned- batch cases (the batch being consumed at exit, and one still queued unread on the buffered event channel) one documented home.
kinyoklion
marked this pull request as ready for review
July 27, 2026 22:07
joker23
approved these changes
Jul 28, 2026
keelerm84
approved these changes
Jul 28, 2026
kinyoklion
added a commit
that referenced
this pull request
Jul 29, 2026
CI on `main` is red for two independent reasons after #63 and #64 merged; each PR was green alone but their combination broke both the lint job and the test jobs. This PR fixes both. **1. Test deadlock (the 10-minute Windows timeouts, reproducible on Linux).** #64 changed replayed batches to flush once per batch instead of once per event. The #63 disconnect tests positioned their producer by waiting for `data: first` to reach the client mid-batch — which can never happen once mid-batch writes are no longer flushed, since the producer deliberately holds the batch open. 10 of the 16 tests in `server_replay_disconnect_test.go` fail this way, and the first one deadlocks the whole test binary: its `t.Fatalf` skips the producer-release step, so the deferred `httptest.Server.Close` waits forever on the still-running handler. (#63's CI predated #64's merge, so neither PR saw this.) The fix is test-only: the test repositories close a `firstDelivered` channel when their first unbuffered send completes — which proves the handler has consumed the event and is reading the batch — and the tests synchronize on that instead of on client-visible bytes. The normal-delivery tests still assert the full batch, including `data: first`, reaches the client once the batch completes. The production code needs no changes; the drain and context-cancellation logic composes with per-batch flushing as-is. **2. gocyclo (the Linux lint failure).** The two merges combined pushed `(*Server).Handler` to cyclomatic complexity 31, over golangci-lint's default limit of 30, and left `(*Server).run` at exactly 30. Rather than raising the limit, this extracts two self-contained pieces into named functions — pure code motion, no behavior change: - `writeStreamHeaders`: the SSE response-header setup and gzip negotiation that opened the `Handler` closure (Handler 31 -> 28). - `replay`: the choice between `ReplayWithContext` and the context-less `Replay` fallback, previously inlined at the deepest nesting level of `run` (run 30 -> 29). Verified with the repo's `make lint` (same golangci-lint version as CI) and the full suite green under `-race`, including all 10 previously hanging/failing tests.
kinyoklion
pushed a commit
that referenced
this pull request
Jul 29, 2026
🤖 I have created a release *beep* *boop* --- ## [1.11.2](v1.11.1...v1.11.2) (2026-07-29) ### Bug Fixes * Flush once per replayed batch instead of once per event ([#64](#64)) ([e505af0](e505af0)) * unblock Repository producers when a subscriber disconnects mid-replay ([#63](#63)) ([1146284](1146284)) --- This PR was generated with [Release Please](https://github.com/googleapis/release-please). See [documentation](https://github.com/googleapis/release-please#release-please). Co-authored-by: github-actions[bot] <41898282+github-actions[bot]@users.noreply.github.com>
kinyoklion
added a commit
to launchdarkly/ld-relay
that referenced
this pull request
Jul 29, 2026
…onnects (#774) ## 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 `select`s 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. <!-- CURSOR_SUMMARY --> --- > [!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 **`select`s 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. > > <sup>Reviewed by [Cursor Bugbot](https://cursor.com/bugbot) for commit 9997de9. Bugbot is set up for automated code reviews on this repo. Configure [here](https://www.cursor.com/dashboard/bugbot).</sup> <!-- /CURSOR_SUMMARY -->
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Problem
eventsourceis an SSE server. When aRepositoryis registered for a channel,Server.run()callsrepo.Replay(channel, lastEventID) <-chan Eventon each new subscription, wraps the returned channel as aneventBatch, and hands it to the per-connection HTTP handler goroutine, which drains it and writes each event to the client.The handler detects client disconnect immediately via
req.Context().Done()(both a clean FIN and an abrupt reset fire it). But on disconnect it doesbreak ReadLoopand abandons the batch channel without draining it — and it cannot close it, because it holds the receiving end.Repository.Replayhas no context/cancellation hook (two strings in, a bare channel out), so the producer goroutine insideReplayblocks forever on its nextout <- event: a clean shutdown and a dirty one are indistinguishable to it, because "my send never completes" is all the API exposes.Consequence: a mid-replay disconnect strands the
Replayproducer goroutine and its payload until the process exits. (Once events are flowing, disconnects are already caught on the write path viaenc.Encodeerrors — only the initial producer to handler hop is blind.)Fix — two levels
1. Background drain (no API change, covers every Repository). When the handler exits its read loop while still consuming a batch (
readBatchChnon-nil), it drains the remaining events in a throwaway goroutine (go func(){ for range ch {} }()). The producer unblocks as fast as it can emit and releases in milliseconds. This alone fixes the forever-leak for anyRepository, including third-party ones.2.
RepositoryWithContext(optional extension, clean propagation). ARepositorymay additionally implement:The
Servertype-asserts the registered repository for it and, when present, calls it with the subscription's request context, so the producer canselect { case out <- event: case <-ctx.Done(): return }and abort promptly and cleanly.Replayremains required (the new interface embedsRepository); repositories that implement onlyReplayare served exactly as before via the drain safety net.Together: #1 guarantees no producer blocks past the handler's exit even for repos that never adopt the context; #2 lets adopters stop immediately on disconnect.
Backward compatibility
No exported symbol changed or was removed.
Repository.Replayis unchanged;RepositoryWithContextis new and optional and is selected purely by type assertion.SliceRepositoryand any existing consumer continue to work unmodified.Tests
server_replay_disconnect_test.go:Replaydrain path and theReplayWithContextpath.SetLinger(0)) disconnects.ReplayWithContextobservesctx.Done()on disconnect, and is preferred overReplaywhen both are implemented.Full suite passes under
go test -race ./...;gofmtandgo vetclean.