Skip to content

fix: unblock Repository producers when a subscriber disconnects mid-replay - #63

Merged
kinyoklion merged 6 commits into
mainfrom
rlamb/eventsource-replay-disconnect
Jul 28, 2026
Merged

fix: unblock Repository producers when a subscriber disconnects mid-replay#63
kinyoklion merged 6 commits into
mainfrom
rlamb/eventsource-replay-disconnect

Conversation

@kinyoklion

Copy link
Copy Markdown
Member

Problem

eventsource is an SSE server. When a Repository is registered for a channel, Server.run() calls repo.Replay(channel, lastEventID) <-chan Event on each new subscription, wraps the returned channel as an eventBatch, 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 does break ReadLoop and abandons the batch channel without draining it — and it cannot close it, because it holds the receiving end. Repository.Replay has no context/cancellation hook (two strings in, a bare channel out), so the producer goroutine inside Replay blocks forever on its next out <- 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 Replay producer goroutine and its payload until the process exits. (Once events are flowing, disconnects are already caught on the write path via enc.Encode errors — 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 (readBatchCh non-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 any Repository, including third-party ones.

2. RepositoryWithContext (optional extension, clean propagation). A Repository may additionally implement:

ReplayWithContext(ctx context.Context, channel, id string) <-chan Event

The Server type-asserts the registered repository for it and, when present, calls it with the subscription's request context, so the producer can select { case out <- event: case <-ctx.Done(): return } and abort promptly and cleanly. Replay remains required (the new interface embeds Repository); repositories that implement only Replay are 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.Replay is unchanged; RepositoryWithContext is new and optional and is selected purely by type assertion. SliceRepository and any existing consumer continue to work unmodified.

Tests

server_replay_disconnect_test.go:

  • Repro/regression: a producer blocked on a channel send is stranded on disconnect before the fix; after the fix it unblocks well under the deadline — verified for both the plain-Replay drain path and the ReplayWithContext path.
  • Clean FIN and abrupt RST (SetLinger(0)) disconnects.
  • Normal in-order delivery still works (plain and context repos) when the subscriber stays connected.
  • ReplayWithContext observes ctx.Done() on disconnect, and is preferred over Replay when both are implemented.
  • Empty batch (nil channel and already-closed channel) does not hang and normal publishing still works afterward.
  • Multiple concurrent subscriptions all unblock.
  • Server close during replay does not strand the producer.

Full suite passes under go test -race ./...; gofmt and go vet clean.

…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.
Comment thread interface.go
//
// 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.

@kinyoklion kinyoklion Jul 27, 2026

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

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
kinyoklion marked this pull request as ready for review July 27, 2026 22:07
@kinyoklion
kinyoklion requested a review from a team as a code owner July 27, 2026 22:07
@kinyoklion
kinyoklion merged commit 1146284 into main Jul 28, 2026
8 checks passed
@kinyoklion
kinyoklion deleted the rlamb/eventsource-replay-disconnect branch July 28, 2026 21:20
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 -->
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet

Projects

None yet

Development

Successfully merging this pull request may close these issues.

3 participants