fix: Flush once per replayed batch instead of once per event - #64
Merged
Conversation
The SSE handler previously called flusher.Flush() after every replayed event, emitting one HTTP chunk (and typically one write syscall) per event. A large replay therefore cost as many network writes as it had events. Replayed batch events are now encoded one per main select-loop iteration and flushed only when the batch channel closes. net/http's chunked writer flushes its buffer through to the connection as it fills, so the batch streams out as a few large chunks rather than one chunk per event -- far fewer write syscalls for an identical payload, which cuts replay CPU and wall-clock substantially on both buffered and streaming repositories. Memory stays bounded because the response buffer writes through as it fills and a slow client blocks the encode. Encoding through the select loop keeps the handler responsive: connection close and MaxConnTime are evaluated between every event, so no cap on events-per-flush is needed to bound time away from the loop. Includes a replay benchmark and a regression test that floods an unbounded batch and asserts MaxConnTime is still honored mid-replay.
keelerm84
marked this pull request as ready for review
July 27, 2026 16:39
aaron-zeisler
approved these changes
Jul 27, 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
2 participants
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.
Summary
The SSE server handler previously called
flusher.Flush()after every replayed event. On a chunked HTTP response each flush emits its own HTTP chunk (and typically its own write syscall), so replaying a large data set to a newly connected client cost roughly one network write per event.This change encodes replayed batch events one per iteration of the handler's main
selectloop and flushes only when the batch channel closes.net/http's chunked writer already flushes its buffer through to the connection as it fills, so the batch now streams out as a handful of large chunks instead of one chunk per event -- the same payload with far fewer write syscalls. Memory stays bounded: the response buffer writes through as it fills, and a slow client blocks the encode.Because every event still passes through the
selectloop, connection close andMaxConnTimeare evaluated between events. That removes the need for any cap on how many events are encoded between flushes (there is no longer a window where the handler is "away" from the loop), so the previousmaxEventsPerFlushbound is gone.Benchmarking a replay over loopback HTTP (counting real
Flushcalls) shows flushes drop from N+1 to a small constant for an N-event replay, cutting replay CPU and wall-clock substantially on both buffered and streaming repositories, with identical payload bytes and encoder writes.A regression test floods an unbounded replay batch and asserts the handler still honors
MaxConnTimemid-replay.