From 439939716aac2fd62c96a1fb7ff229255cc3b407 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 24 Jul 2026 09:29:23 -0700 Subject: [PATCH 1/6] fix: unblock Repository producers when a subscriber disconnects mid-replay 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. --- interface.go | 44 ++- server.go | 33 +- server_replay_disconnect_test.go | 519 +++++++++++++++++++++++++++++++ 3 files changed, 594 insertions(+), 2 deletions(-) create mode 100644 server_replay_disconnect_test.go diff --git a/interface.go b/interface.go index 112dd5a..7a5a5ba 100644 --- a/interface.go +++ b/interface.go @@ -5,7 +5,10 @@ // If the Repository interface is implemented on the server, events can be replayed in case of a network disconnection. package eventsource -import "net/http" +import ( + "context" + "net/http" +) // Event is the interface for any event received by the client or sent by the server. type Event interface { @@ -61,6 +64,45 @@ type Repository interface { Replay(channel, id string) chan Event } +// RepositoryWithContext is an optional extension of Repository that allows a Repository to be +// notified when the subscriber that requested a replay has gone away. +// +// A plain Repository.Replay receives only the channel and event id, and returns a bare channel; +// it has no way to learn that the subscriber disconnected. If the subscriber disconnects while +// the Repository's producer goroutine is still blocked sending an event, that goroutine can be +// stranded until the server drains the channel on its behalf. +// +// If a registered Repository also implements RepositoryWithContext, the Server will call +// ReplayWithContext instead of Replay, passing the subscribing request's context. That context +// is cancelled when the subscriber disconnects (or when the connection is otherwise terminated), +// so the producer can select on ctx.Done() and stop sending immediately, for example: +// +// func (r *myRepo) ReplayWithContext(ctx context.Context, channel, id string) <-chan eventsource.Event { +// out := make(chan eventsource.Event) +// go func() { +// defer close(out) +// for _, event := range r.eventsFor(channel, id) { +// select { +// case out <- event: +// case <-ctx.Done(): +// return +// } +// } +// }() +// return out +// } +// +// 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. +type RepositoryWithContext interface { + Repository + // ReplayWithContext behaves like Replay, but additionally receives a context that is cancelled + // when the subscriber goes away. It has the same channel-closing responsibilities as Replay, + // and may likewise return nil if there are no events to be sent. + ReplayWithContext(ctx context.Context, channel, id string) <-chan Event +} + // Logger is the interface for a custom logging implementation that can handle log output for a Stream. type Logger interface { Println(...interface{}) diff --git a/server.go b/server.go index 78cbeea..6281299 100644 --- a/server.go +++ b/server.go @@ -1,6 +1,7 @@ package eventsource import ( + "context" "net/http" "strings" "sync" @@ -11,6 +12,9 @@ type subscription struct { channel string lastEventID string out chan<- eventOrComment + // ctx is the subscribing request's context. It is cancelled when the subscriber disconnects, + // and is passed to a Repository that implements RepositoryWithContext. + ctx context.Context } type eventOrComment interface{} @@ -131,6 +135,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { channel: channel, lastEventID: req.Header.Get("Last-Event-ID"), out: eventCh, + ctx: req.Context(), } srv.subs <- sub flusher := w.(http.Flusher) @@ -270,6 +275,23 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } } } + if readBatchCh != nil { + // We are exiting the read loop while still in the middle of consuming a batch of replayed + // events from a Repository (e.g. the subscriber disconnected, or MaxConnTime elapsed). The + // Repository's producer goroutine may be blocked trying to send the remaining events on this + // channel. Since we hold the receiving end -- we can neither close it nor keep reading it on + // this exiting goroutine -- drain it in the background so the producer can unblock and release + // its resources promptly rather than leaking until the process exits. + // + // A Repository that implements RepositoryWithContext will already have been told to stop via + // context cancellation, but draining is harmless in that case and remains the safety net for + // repositories that only implement Replay. + go func(ch <-chan Event) { + for range ch { + // Discard any remaining events until the producer closes the channel. + } + }(readBatchCh) + } if !closedNormally { srv.unsubs <- sub // the server didn't tell us to close, so we must tell it that we're closing } @@ -383,7 +405,16 @@ func (srv *Server) run() { if srv.ReplayAll || len(sub.lastEventID) > 0 { repo, ok := repos[sub.channel] if ok { - batchCh := repo.Replay(sub.channel, sub.lastEventID) + // If the repository supports it, pass the subscriber's context so its producer can + // stop sending promptly when the subscriber disconnects. Otherwise fall back to the + // original context-less Replay; the handler's background drain (see Handler) still + // ensures such a producer eventually unblocks. + var batchCh <-chan Event + if repoCtx, ok := repo.(RepositoryWithContext); ok { + batchCh = repoCtx.ReplayWithContext(sub.ctx, sub.channel, sub.lastEventID) + } else { + batchCh = repo.Replay(sub.channel, sub.lastEventID) + } if batchCh != nil { trySend(sub, eventBatch{events: batchCh}) } diff --git a/server_replay_disconnect_test.go b/server_replay_disconnect_test.go new file mode 100644 index 0000000..86d034c --- /dev/null +++ b/server_replay_disconnect_test.go @@ -0,0 +1,519 @@ +package eventsource + +import ( + "bufio" + "context" + "fmt" + "net" + "net/http" + "net/http/httptest" + neturl "net/url" + "strconv" + "strings" + "sync" + "sync/atomic" + "testing" + "time" + + "github.com/stretchr/testify/assert" + "github.com/stretchr/testify/require" +) + +// These tests cover the behavior of a registered Repository whose producer goroutine may be +// blocked sending replayed events when a subscriber disconnects. They exercise both the +// unconditional background drain performed by the handler (which unblocks any Repository) and the +// optional RepositoryWithContext extension (which lets a Repository observe cancellation directly). + +const replayTestDeadline = 3 * time.Second + +// plainReplayRepo implements only the original Repository interface. Its producer sends one event, +// waits for the test to release it, then sends a series of events on an unbuffered channel. This +// lets the test position the producer so that it becomes blocked on a channel send exactly when the +// subscriber disconnects. +type plainReplayRepo struct { + started chan struct{} + release chan struct{} + finished chan struct{} +} + +func newPlainReplayRepo() *plainReplayRepo { + return &plainReplayRepo{ + started: make(chan struct{}), + release: make(chan struct{}), + finished: make(chan struct{}), + } +} + +func (r *plainReplayRepo) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(r.finished) + defer close(out) + close(r.started) + out <- &publication{id: "0", data: "first"} + <-r.release + for i := 1; i < 50; i++ { + out <- &publication{id: strconv.Itoa(i), data: "more"} + } + }() + return out +} + +// ctxReplayRepo implements both Replay and ReplayWithContext. It records which method was called +// and (for ReplayWithContext) whether it observed context cancellation. +type ctxReplayRepo struct { + started chan struct{} + release chan struct{} + finished chan struct{} + ctxObserved chan struct{} + replayCalled int32 + replayCtxCalled int32 + // blockOnSend, when true, makes the producer block on a channel send after the first event; + // otherwise it waits directly on ctx.Done() after the first event. + blockOnSend bool +} + +func newCtxReplayRepo() *ctxReplayRepo { + return &ctxReplayRepo{ + started: make(chan struct{}), + release: make(chan struct{}), + finished: make(chan struct{}), + ctxObserved: make(chan struct{}), + } +} + +func (r *ctxReplayRepo) Replay(channel, id string) chan Event { + atomic.StoreInt32(&r.replayCalled, 1) + out := make(chan Event) + go func() { + defer close(out) + out <- &publication{id: "0", data: "first"} + }() + return out +} + +func (r *ctxReplayRepo) ReplayWithContext(ctx context.Context, channel, id string) <-chan Event { + atomic.StoreInt32(&r.replayCtxCalled, 1) + out := make(chan Event) + go func() { + defer close(r.finished) + defer close(out) + close(r.started) + select { + case out <- &publication{id: "0", data: "first"}: + case <-ctx.Done(): + close(r.ctxObserved) + return + } + if r.blockOnSend { + <-r.release + for i := 1; i < 50; i++ { + select { + case out <- &publication{id: strconv.Itoa(i), data: "more"}: + case <-ctx.Done(): + close(r.ctxObserved) + return + } + } + return + } + // Wait purely on cancellation so the drain does not race the context observation. + <-ctx.Done() + close(r.ctxObserved) + }() + return out +} + +// rawSSEConn dials the server directly so the test can control exactly how the connection is closed +// (clean FIN vs. abrupt RST). It returns the connection and a reader positioned at the response body. +func rawSSEConn(t *testing.T, url string) *net.TCPConn { + t.Helper() + u, err := neturl.Parse(url) + require.NoError(t, err) + path := u.Path + if path == "" { + path = "/" + } + conn, err := net.Dial("tcp", u.Host) + require.NoError(t, err) + _, err = fmt.Fprintf(conn, "GET %s HTTP/1.1\r\nHost: %s\r\n\r\n", path, u.Host) + require.NoError(t, err) + return conn.(*net.TCPConn) +} + +// sseReader consumes an SSE response body on a single background goroutine, exposing the lines it +// reads via a channel. Using one goroutine per response avoids concurrent reads of the same reader. +type sseReader struct { + lines chan string + done chan struct{} +} + +func newSSEReader(t *testing.T, r interface{ Read([]byte) (int, error) }) *sseReader { + t.Helper() + s := &sseReader{lines: make(chan string), done: make(chan struct{})} + go func() { + rd := bufio.NewReader(r) + for { + line, err := rd.ReadString('\n') + if line != "" { + select { + case s.lines <- line: + case <-s.done: + return + } + } + if err != nil { + close(s.lines) + return + } + } + }() + t.Cleanup(func() { close(s.done) }) + return s +} + +// waitFor reads lines until one contains the wanted substring, tolerating chunked framing. +func (s *sseReader) waitFor(t *testing.T, want string) { + t.Helper() + deadline := time.After(replayTestDeadline) + for { + select { + case line, ok := <-s.lines: + if !ok { + t.Fatalf("stream ended before receiving %q", want) + } + if strings.Contains(line, want) { + return + } + case <-deadline: + t.Fatalf("timed out waiting to receive %q", want) + } + } +} + +func assertClosedWithin(t *testing.T, ch <-chan struct{}, what string) { + t.Helper() + select { + case <-ch: + case <-time.After(replayTestDeadline): + t.Fatalf("%s did not happen within %s", what, replayTestDeadline) + } +} + +// TestReplayProducerUnblocksOnCleanDisconnectPlainRepository verifies that a plain Repository whose +// producer is blocked on a channel send is unblocked (via the handler's background drain) when the +// subscriber cleanly disconnects (FIN). +func TestReplayProducerUnblocksOnCleanDisconnectPlainRepository(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newPlainReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL, nil) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + <-repo.started + + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + + cancel() + _ = resp.Body.Close() + time.Sleep(100 * time.Millisecond) + close(repo.release) + + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// TestReplayProducerUnblocksOnAbruptResetPlainRepository is the same as above but the subscriber +// aborts the connection with a TCP reset (RST) rather than a clean close. +func TestReplayProducerUnblocksOnAbruptResetPlainRepository(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newPlainReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + conn := rawSSEConn(t, httpServer.URL) + <-repo.started + rd := newSSEReader(t, conn) + rd.waitFor(t, "data: first") + + // Force an abrupt RST instead of a clean FIN. + require.NoError(t, conn.SetLinger(0)) + require.NoError(t, conn.Close()) + time.Sleep(100 * time.Millisecond) + close(repo.release) + + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// TestReplayWithContextObservesCancellationOnDisconnect verifies that a RepositoryWithContext is +// given a context that is cancelled when the subscriber disconnects, and that the producer can +// observe that cancellation directly. +func TestReplayWithContextObservesCancellationOnDisconnect(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newCtxReplayRepo() // blockOnSend == false: producer waits on ctx.Done() after first event + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL, nil) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + <-repo.started + + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + + cancel() + _ = resp.Body.Close() + + assertClosedWithin(t, repo.ctxObserved, "producer observing ctx cancellation") + assertClosedWithin(t, repo.finished, "producer goroutine exit") + assert.Equal(t, int32(1), atomic.LoadInt32(&repo.replayCtxCalled), "ReplayWithContext should have been called") + assert.Equal(t, int32(0), atomic.LoadInt32(&repo.replayCalled), "plain Replay should not have been called") +} + +// TestReplayWithContextProducerUnblocksWhenBlockedOnSend verifies that a RepositoryWithContext +// producer that is blocked on a channel send at disconnect time still unblocks promptly (via either +// context cancellation or the background drain). +func TestReplayWithContextProducerUnblocksWhenBlockedOnSend(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newCtxReplayRepo() + repo.blockOnSend = true + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL, nil) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + <-repo.started + + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + + cancel() + _ = resp.Body.Close() + time.Sleep(100 * time.Millisecond) + close(repo.release) + + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// TestReplayWithContextIsPreferredOverReplay verifies that when a Repository implements both methods, +// the server uses ReplayWithContext. +func TestReplayWithContextIsPreferredOverReplay(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newCtxReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + <-repo.started + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + + assert.Equal(t, int32(1), atomic.LoadInt32(&repo.replayCtxCalled)) + assert.Equal(t, int32(0), atomic.LoadInt32(&repo.replayCalled)) +} + +// TestReplayNormalDeliveryPlainRepository verifies that when the subscriber stays connected, all +// replayed events from a plain Repository are delivered in order. +func TestReplayNormalDeliveryPlainRepository(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newPlainReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + <-repo.started + + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + close(repo.release) + // Events 1..49 should all arrive; confirm the last one. + rd.waitFor(t, "id: 49") + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// TestReplayNormalDeliveryContextRepository verifies full ordered delivery for a RepositoryWithContext. +func TestReplayNormalDeliveryContextRepository(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newCtxReplayRepo() + repo.blockOnSend = true + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + <-repo.started + + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + close(repo.release) + rd.waitFor(t, "id: 49") + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// TestReplayEmptyBatchNilChannel verifies that a Repository returning nil (no events) does not hang, +// and that normal publishing still works afterward. +func TestReplayEmptyBatchNilChannel(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + server.Register(channel, nilRepo{}) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + + // After the (empty) replay, a normally published event should still be delivered. + go func() { + time.Sleep(50 * time.Millisecond) + server.Publish([]string{channel}, &publication{id: "live", data: "published"}) + }() + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: published") +} + +// TestReplayEmptyBatchClosedChannel verifies that a Repository returning an already-closed channel +// does not hang and normal publishing still works afterward. +func TestReplayEmptyBatchClosedChannel(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + server.Register(channel, closedRepo{}) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + + go func() { + time.Sleep(50 * time.Millisecond) + server.Publish([]string{channel}, &publication{id: "live", data: "published"}) + }() + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: published") +} + +type nilRepo struct{} + +func (nilRepo) Replay(channel, id string) chan Event { return nil } + +type closedRepo struct{} + +func (closedRepo) Replay(channel, id string) chan Event { + out := make(chan Event) + close(out) + return out +} + +// TestReplayMultipleConcurrentSubscriptionsUnblock verifies that several concurrent subscribers that +// each disconnect mid-replay all have their producer goroutines unblocked. +func TestReplayMultipleConcurrentSubscriptionsUnblock(t *testing.T) { + const n = 5 + mux := http.NewServeMux() + server := NewServer() + server.ReplayAll = true + defer server.Close() + + repos := make([]*plainReplayRepo, n) + for i := 0; i < n; i++ { + channel := "chan-" + strconv.Itoa(i) + repos[i] = newPlainReplayRepo() + server.Register(channel, repos[i]) + mux.HandleFunc("/"+channel, server.Handler(channel)) + } + httpServer := httptest.NewServer(mux) + defer httpServer.Close() + + var wg sync.WaitGroup + for i := 0; i < n; i++ { + wg.Add(1) + go func(i int) { + defer wg.Done() + ctx, cancel := context.WithCancel(context.Background()) + req, _ := http.NewRequestWithContext(ctx, "GET", httpServer.URL+"/chan-"+strconv.Itoa(i), nil) + resp, err := http.DefaultClient.Do(req) + require.NoError(t, err) + <-repos[i].started + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + cancel() + _ = resp.Body.Close() + time.Sleep(50 * time.Millisecond) + close(repos[i].release) + }(i) + } + wg.Wait() + + for i := 0; i < n; i++ { + assertClosedWithin(t, repos[i].finished, fmt.Sprintf("producer %d exit", i)) + } +} + +// TestReplayProducerUnblocksWhenServerCloses verifies that closing the server does not strand a +// Repository producer that is mid-replay. +func TestReplayProducerUnblocksWhenServerCloses(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + repo := newPlainReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + <-repo.started + rd := newSSEReader(t, resp.Body) + rd.waitFor(t, "data: first") + + // Release the producer so it is actively delivering, then close the server and the client. + close(repo.release) + _ = resp.Body.Close() + server.Close() + + assertClosedWithin(t, repo.finished, "producer goroutine exit after server close") +} From 78e6726ded8d277b4e14f2fc2ae7c6e0e3a71170 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 24 Jul 2026 09:51:09 -0700 Subject: [PATCH 2/6] refactor: extend replay drain to undelivered-batch path and factor out 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. --- server.go | 22 +++++++++++++++------- 1 file changed, 15 insertions(+), 7 deletions(-) diff --git a/server.go b/server.go index 6281299..65d5d0a 100644 --- a/server.go +++ b/server.go @@ -286,11 +286,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // A Repository that implements RepositoryWithContext will already have been told to stop via // context cancellation, but draining is harmless in that case and remains the safety net for // repositories that only implement Replay. - go func(ch <-chan Event) { - for range ch { - // Discard any remaining events until the producer closes the channel. - } - }(readBatchCh) + go drainReplayedEvents(readBatchCh) } if !closedNormally { srv.unsubs <- sub // the server didn't tell us to close, so we must tell it that we're closing @@ -298,6 +294,14 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } } +// drainReplayedEvents consumes and discards the events from a Repository replay batch channel that +// no subscriber will read again, so the goroutine producing those events can complete and release +// its resources instead of blocking forever on a send. +func drainReplayedEvents(ch <-chan Event) { + for range ch { + } +} + // Register registers a Repository to be used for the specified channel. The Repository will be used to // determine whether new subscribers should receive data that was generated before they subscribed. // @@ -415,8 +419,12 @@ func (srv *Server) run() { } else { batchCh = repo.Replay(sub.channel, sub.lastEventID) } - if batchCh != nil { - trySend(sub, eventBatch{events: batchCh}) + if batchCh != nil && !sub.send(eventBatch{events: batchCh}) { + // The subscriber was already closed, so it will never consume this batch and + // its producer would otherwise block forever; drain it in the background. + sub.close() + delete(subs[sub.channel], sub) + go drainReplayedEvents(batchCh) } } } From 75eaaaea29866c0a228f9e000665b24e7d0e2c09 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Fri, 24 Jul 2026 10:10:45 -0700 Subject: [PATCH 3/6] chore: silence revive empty-block on the drain loop Match the existing channel-drain convention (stream.go) with a //nolint:revive directive; golangci-lint run is clean. --- server.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/server.go b/server.go index 65d5d0a..77a6165 100644 --- a/server.go +++ b/server.go @@ -298,7 +298,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // no subscriber will read again, so the goroutine producing those events can complete and release // its resources instead of blocking forever on a send. func drainReplayedEvents(ch <-chan Event) { - for range ch { + for range ch { //nolint:revive // draining until the channel is closed } } From fc098a778e799ec81fd372bad06782544afdc3ef Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:25:41 -0700 Subject: [PATCH 4/6] fix: drain replay batches stranded by Server shutdown or undequeued handoffs 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) --- server.go | 82 +++++++++++- server_replay_disconnect_test.go | 223 +++++++++++++++++++++++++++++++ 2 files changed, 299 insertions(+), 6 deletions(-) diff --git a/server.go b/server.go index 77a6165..303ad5c 100644 --- a/server.go +++ b/server.go @@ -15,6 +15,11 @@ type subscription struct { // ctx is the subscribing request's context. It is cancelled when the subscriber disconnects, // and is passed to a Repository that implements RepositoryWithContext. ctx context.Context + // batch is the replay batch channel that was handed to this subscription, if any. It is + // recorded so that if the handler exits without ever dequeuing the batch from its buffered + // event channel, the unsubscribe path can still drain it and unblock the Repository's + // producer. Accessed only from the Server.run() goroutine. + batch <-chan Event } type eventOrComment interface{} @@ -288,6 +293,26 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // repositories that only implement Replay. go drainReplayedEvents(readBatchCh) } + // A replay batch that the Server queued on eventCh but that the loop above never dequeued + // would strand its producer the same way. Server.run() drains such a batch when it + // processes our unsubscription (see the unsubs case there); this sweep of the values + // already buffered additionally covers the case where the Server has shut down and will + // never process it. Anything the Server enqueues concurrently with this sweep is still + // handled by the unsubs path. + SweepPending: + for { + select { + case ev, ok := <-eventCh: + if !ok { + break SweepPending + } + if batch, isBatch := ev.(eventBatch); isBatch { + go drainReplayedEvents(batch.events) + } + default: + break SweepPending + } + } if !closedNormally { srv.unsubs <- sub // the server didn't tell us to close, so we must tell it that we're closing } @@ -387,6 +412,16 @@ func (srv *Server) run() { } case sub := <-srv.unsubs: delete(subs[sub.channel], sub) + if sub.batch != nil { + // The handler has exited. If it never dequeued the replay batch from its event + // channel -- or exited partway through consuming it -- the Repository's producer + // may still be blocked sending on it; drain it so the producer can finish. If the + // batch was fully consumed, the channel is already closed and this goroutine exits + // immediately. Draining concurrently with the handler's own exit-time drain is + // safe: both simply receive until the channel is closed. + go drainReplayedEvents(sub.batch) + sub.batch = nil + } case pub := <-srv.pub: for _, c := range pub.channels { for s := range subs[c] { @@ -419,19 +454,54 @@ func (srv *Server) run() { } else { batchCh = repo.Replay(sub.channel, sub.lastEventID) } - if batchCh != nil && !sub.send(eventBatch{events: batchCh}) { - // The subscriber was already closed, so it will never consume this batch and - // its producer would otherwise block forever; drain it in the background. - sub.close() - delete(subs[sub.channel], sub) - go drainReplayedEvents(batchCh) + if batchCh != nil { + if sub.send(eventBatch{events: batchCh}) { + // Remember the batch so that if the subscriber goes away before its + // handler dequeues it, the unsubs case below can still drain it. + sub.batch = batchCh + } else { + // The send failed because the subscription's buffer was full (send + // closes the subscription in that case). The batch will never be + // consumed and its producer would otherwise block forever; drain it + // in the background. + delete(subs[sub.channel], sub) + go drainReplayedEvents(batchCh) + } } } } case <-srv.quit: + // We are about to stop processing unsubscriptions, so first handle any that are + // already queued: their handlers have exited, and a handler that swept its event + // channel before the replay batch was enqueued is relying on this path to drain it. + // Subscriptions handled here are deliberately not removed from the map -- the loop + // below revisits them, but with batch already nil and close() being idempotent + // that revisit is a no-op. + DrainUnsubs: + for { + select { + case sub := <-srv.unsubs: + if sub.batch != nil { + go drainReplayedEvents(sub.batch) + sub.batch = nil + } + default: + break DrainUnsubs + } + } for _, sub := range subs { for s := range sub { s.close() + // If the subscriber is already gone, its handler can no longer be relied on + // to consume or drain a batch, and its unsubscription may never be seen + // (we are exiting); drain the batch here. If the subscriber is still + // connected, we must NOT drain: its handler keeps delivering the buffered + // batch to the client even after close(s.out), and draining would steal + // events from that delivery. + if s.batch != nil && s.ctx.Err() != nil { + go drainReplayedEvents(s.batch) + s.batch = nil + } } } return diff --git a/server_replay_disconnect_test.go b/server_replay_disconnect_test.go index 86d034c..dfc17d3 100644 --- a/server_replay_disconnect_test.go +++ b/server_replay_disconnect_test.go @@ -493,6 +493,229 @@ func TestReplayMultipleConcurrentSubscriptionsUnblock(t *testing.T) { } } +// TestReplayProducerUnblocksOnImmediateDisconnectPlainRepository verifies that a plain Repository +// producer is unblocked even when the subscriber disconnects immediately, before reading any of the +// stream -- i.e. before the handler has dequeued the replay batch from its buffered event channel. +// This exercises a different window than the tests above, which all wait for "data: first" (and +// therefore guarantee the handler is already consuming the batch) before disconnecting. Because the +// disconnect races the handler's dequeue, a single trial can land on either side of the window, so +// the test runs many trials and requires every producer to finish; on code without the +// unsubscribe-path drain, most trials strand the producer. +func TestReplayProducerUnblocksOnImmediateDisconnectPlainRepository(t *testing.T) { + const trials = 20 + for i := 0; i < trials; i++ { + func() { + channel := "test" + server := NewServer() + server.ReplayAll = true + defer server.Close() + repo := newPlainReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + conn := rawSSEConn(t, httpServer.URL) + // Wait only until the producer is running (which happens as the subscription is + // registered), NOT until any event has been received, then disconnect at once -- + // alternating between an abrupt RST and a clean FIN. + <-repo.started + if i%2 == 0 { + require.NoError(t, conn.SetLinger(0)) + } + require.NoError(t, conn.Close()) + close(repo.release) + + assertClosedWithin(t, repo.finished, + fmt.Sprintf("producer goroutine exit (trial %d)", i)) + }() + } +} + +// slowReplayRepo delays returning from Replay. This lets a test arrange for the subscriber's +// handler to have exited (and swept its still-empty event channel) before the batch is ever +// enqueued, so that only the Server's unsubscription and shutdown paths can dispose of the batch. +type slowReplayRepo struct { + started chan struct{} + finished chan struct{} + delay time.Duration +} + +func newSlowReplayRepo(delay time.Duration) *slowReplayRepo { + return &slowReplayRepo{ + started: make(chan struct{}), + finished: make(chan struct{}), + delay: delay, + } +} + +func (r *slowReplayRepo) Replay(channel, id string) chan Event { + close(r.started) + time.Sleep(r.delay) + out := make(chan Event) + go func() { + defer close(r.finished) + defer close(out) + for i := 0; i < 50; i++ { + out <- &publication{id: strconv.Itoa(i), data: "replayed"} + } + }() + return out +} + +// TestReplayProducerUnblocksWhenServerClosesDuringImmediateDisconnect verifies that a producer is +// not stranded when Server.Close races a subscriber that disconnected before its replay batch was +// even produced. In that ordering the handler has already exited (its exit-time sweep found +// nothing, because the batch had not been enqueued yet), so the batch can only be disposed of by +// the Server -- and the Server may select the quit case over the handler's queued unsubscription, +// so the shutdown path itself must drain it. Whether quit or the unsubscription is selected first +// is a coin flip per trial, so the test runs many trials and requires every producer to finish. +func TestReplayProducerUnblocksWhenServerClosesDuringImmediateDisconnect(t *testing.T) { + const trials = 20 + for i := 0; i < trials; i++ { + func() { + channel := "test" + server := NewServer() + server.ReplayAll = true + repo := newSlowReplayRepo(80 * time.Millisecond) + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + conn := rawSSEConn(t, httpServer.URL) + // Replay is now sleeping inside the Server's goroutine. Disconnect and give the + // handler time to observe it, exit, and queue its unsubscription -- all before + // Replay returns and the batch is enqueued. + <-repo.started + require.NoError(t, conn.SetLinger(0)) + require.NoError(t, conn.Close()) + time.Sleep(40 * time.Millisecond) + // Close the server; the quit signal and the queued unsubscription are now both + // ready when the Server finishes the registration it is still processing. + server.Close() + + assertClosedWithin(t, repo.finished, + fmt.Sprintf("producer goroutine exit (trial %d)", i)) + }() + } +} + +// TestReplayServerCloseDrainsBatchOfDepartedSubscriber pins the narrowest shutdown window +// directly: a subscriber whose context is already cancelled (the client is gone) and whose +// handler will never act again -- it has already swept its then-empty event channel, and its +// unsubscription will never be processed. It builds that state explicitly, registering a +// subscription with a cancelled context and no handler behind it, so the only thing that can +// dispose of the replay batch is the shutdown path's drain of departed subscribers still in the +// subscription map. This is a white-box test by necessity; from outside the process the window +// between a handler's sweep and its unsubscription send is microseconds wide. +func TestReplayServerCloseDrainsBatchOfDepartedSubscriber(t *testing.T) { + channel := "test" + server := NewServer() + server.ReplayAll = true + repo := newPlainReplayRepo() + server.Register(channel, repo) + + ctx, cancel := context.WithCancel(context.Background()) + cancel() // the subscriber is already gone + eventCh := make(chan eventOrComment, server.BufferSize) + sub := &subscription{ + channel: channel, + out: eventCh, + ctx: ctx, + } + server.subs <- sub // registration enqueues the batch; no handler will ever consume it + <-repo.started + close(repo.release) + + server.Close() + assertClosedWithin(t, repo.finished, "producer goroutine exit") +} + +// pacedReplayRepo produces a fixed series of events with small gaps between them, so a test can +// reliably interleave another action (such as Server.Close) with an in-progress delivery. +type pacedReplayRepo struct { + started chan struct{} + finished chan struct{} +} + +func newPacedReplayRepo() *pacedReplayRepo { + return &pacedReplayRepo{started: make(chan struct{}), finished: make(chan struct{})} +} + +func (r *pacedReplayRepo) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(r.finished) + defer close(out) + close(r.started) + for i := 0; i < 30; i++ { + out <- &publication{id: strconv.Itoa(i), data: "replayed"} + time.Sleep(time.Millisecond) + } + }() + return out +} + +// TestReplayCloseDoesNotTruncateDeliveryToConnectedSubscriber verifies that closing the server +// while a still-connected subscriber is consuming a replay batch does not steal any of the +// batch's events from that subscriber: the handler must be left to deliver every remaining +// event, so the client's Last-Event-ID resume point stays accurate. This is the guard rail for +// the shutdown path's decision NOT to drain the batch of a subscriber whose request context is +// still live. The interleaving is racy, so the test repeats the scenario several times. +func TestReplayCloseDoesNotTruncateDeliveryToConnectedSubscriber(t *testing.T) { + const trials = 8 + for i := 0; i < trials; i++ { + func() { + channel := "test" + server := NewServer() + server.ReplayAll = true + repo := newPacedReplayRepo() + server.Register(channel, repo) + httpServer := httptest.NewServer(server.Handler(channel)) + defer httpServer.Close() + + resp, err := http.Get(httpServer.URL) + require.NoError(t, err) + defer resp.Body.Close() + <-repo.started + + // Collect every id until the stream ends (the handler exits normally once the + // batch is done and its channel has been closed by the shutdown), closing the + // server as soon as the first event is seen so the shutdown lands mid-batch. + // Every one of the 30 events must reach the subscriber; a shutdown path that + // drained this batch would swallow a subset of them. + rd := newSSEReader(t, resp.Body) + seen := make(map[string]bool) + closed := false + deadline := time.After(replayTestDeadline) + Collect: + for { + select { + case line, ok := <-rd.lines: + if !ok { + break Collect + } + if strings.HasPrefix(line, "id: ") { + seen[strings.TrimSpace(strings.TrimPrefix(line, "id: "))] = true + } + if !closed && strings.Contains(line, "data: replayed") { + server.Close() // close mid-batch, subscriber still connected + closed = true + } + case <-deadline: + t.Fatalf("timed out waiting for the stream to end (trial %d)", i) + } + } + require.True(t, closed, "stream ended before delivering any event (trial %d)", i) + for id := 0; id < 30; id++ { + assert.True(t, seen[strconv.Itoa(id)], + "event %d was not delivered to the connected subscriber (trial %d)", id, i) + } + assertClosedWithin(t, repo.finished, + fmt.Sprintf("producer goroutine exit (trial %d)", i)) + }() + } +} + // TestReplayProducerUnblocksWhenServerCloses verifies that closing the server does not strand a // Repository producer that is mid-replay. func TestReplayProducerUnblocksWhenServerCloses(t *testing.T) { From b85741dc75f8880f97298c7c61a21a5d7d79b0d1 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:29:03 -0700 Subject: [PATCH 5/6] fix: don't strand late-exiting handlers on the unsubscription send after 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. --- server.go | 32 +++++++++++++++--- server_replay_disconnect_test.go | 58 ++++++++++++++++++++++++++++++++ 2 files changed, 85 insertions(+), 5 deletions(-) diff --git a/server.go b/server.go index 303ad5c..d922c35 100644 --- a/server.go +++ b/server.go @@ -63,9 +63,13 @@ type Server struct { subs chan *subscription unsubs chan *subscription quit chan bool - isClosed bool - isClosedMutex sync.RWMutex - jitter time.Duration + // stopped is closed when run() exits, so that handlers which outlive the Server + // (e.g. a write error detected after Close) do not block forever sending an + // unsubscription that nothing will ever consume. + stopped chan struct{} + isClosed bool + isClosedMutex sync.RWMutex + jitter time.Duration } // NewServer creates a new Server instance. @@ -89,6 +93,7 @@ func NewServerWithJitter(jitter time.Duration) *Server { subs: make(chan *subscription), unsubs: make(chan *subscription, 2), quit: make(chan bool), + stopped: make(chan struct{}), BufferSize: 128, jitter: jitter, } @@ -147,9 +152,19 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { flusher.Flush() enc := NewEncoder(w, useGzip) + // unsubscribe tells the Server this handler is going away. After the Server has + // shut down nothing consumes unsubs (and its small buffer may already be full), + // so a handler that exits late must not block forever on the send. + unsubscribe := func() { + select { + case srv.unsubs <- sub: + case <-srv.stopped: + } + } + writeEventOrComment := func(ec eventOrComment) bool { if err := enc.Encode(ec); err != nil { - srv.unsubs <- sub + unsubscribe() if srv.Logger != nil { srv.Logger.Println(err) } @@ -314,7 +329,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } } if !closedNormally { - srv.unsubs <- sub // the server didn't tell us to close, so we must tell it that we're closing + unsubscribe() // the server didn't tell us to close, so we must tell it that we're closing } } } @@ -388,6 +403,7 @@ func (srv *Server) PublishComment(channels []string, text string) { } func (srv *Server) run() { + defer close(srv.stopped) // All access to the subs and repos maps is done from the same goroutine, so modifications are safe. subs := make(map[string]map[*subscription]struct{}) repos := make(map[string]Repository) @@ -408,6 +424,12 @@ func (srv *Server) run() { if unreg.forceDisconnect { for s := range previousSubs { s.close() + // Unlike the unsubscription and shutdown cases, no batch drain is needed + // here: the server keeps running. A buffered channel delivers its queued + // values before reporting closed, so the handler still dequeues and fully + // consumes an in-flight batch after close(out); if the handler instead + // exits abnormally, its own exit paths and its unsubscription (which run() + // is still alive to process) drain the batch. } } case sub := <-srv.unsubs: diff --git a/server_replay_disconnect_test.go b/server_replay_disconnect_test.go index dfc17d3..366ead4 100644 --- a/server_replay_disconnect_test.go +++ b/server_replay_disconnect_test.go @@ -740,3 +740,61 @@ func TestReplayProducerUnblocksWhenServerCloses(t *testing.T) { assertClosedWithin(t, repo.finished, "producer goroutine exit after server close") } + +// TestLateHandlerExitsDoNotBlockAfterServerClose verifies that handlers whose connections +// fail after the Server has shut down do not block forever sending an unsubscription that +// nothing will consume. More than two such handlers used to deadlock on the unsubs +// channel's small buffer, pinning their connections (and any in-flight batch) open. +func TestLateHandlerExitsDoNotBlockAfterServerClose(t *testing.T) { + const n = 4 + mux := http.NewServeMux() + server := NewServer() + server.ReplayAll = true + + repos := make([]*plainReplayRepo, n) + conns := make([]*net.TCPConn, n) + readers := make([]*sseReader, n) + for i := 0; i < n; i++ { + channel := "chan-" + strconv.Itoa(i) + repos[i] = newPlainReplayRepo() + server.Register(channel, repos[i]) + mux.HandleFunc("/"+channel, server.Handler(channel)) + } + httpServer := httptest.NewServer(mux) + + // Position every handler mid-batch, so none of them observes the shutdown's + // close(out) (they are reading the batch channel, not the event channel). + for i := 0; i < n; i++ { + conns[i] = rawSSEConn(t, httpServer.URL+"/chan-"+strconv.Itoa(i)) + <-repos[i].started + readers[i] = newSSEReader(t, conns[i]) + readers[i].waitFor(t, "data: first") + } + + server.Close() + + // Now fail every connection abruptly and let the producers proceed. Each handler + // exits abnormally (write error or disconnect notification) and unsubscribes -- + // with the Server gone, nothing consumes those sends. + for i := 0; i < n; i++ { + require.NoError(t, conns[i].SetLinger(0)) + require.NoError(t, conns[i].Close()) + } + time.Sleep(100 * time.Millisecond) + for i := 0; i < n; i++ { + close(repos[i].release) + } + + for i := 0; i < n; i++ { + assertClosedWithin(t, repos[i].finished, fmt.Sprintf("producer %d exit", i)) + } + + // A handler blocked on the unsubscription send keeps its request active, which + // makes httptest's Close hang; all handlers exiting is what lets this complete. + closed := make(chan struct{}) + go func() { + httpServer.Close() + close(closed) + }() + assertClosedWithin(t, closed, "http server shutdown (all handlers exited)") +} From 074ff330529b63200c09c369fe9d6bb6d01baca3 Mon Sep 17 00:00:00 2001 From: Ryan Lamb <4955475+kinyoklion@users.noreply.github.com> Date: Mon, 27 Jul 2026 14:43:18 -0700 Subject: [PATCH 6/6] refactor: extract exit-time batch draining from Handler 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. --- server.go | 70 +++++++++++++++++++++++++++++-------------------------- 1 file changed, 37 insertions(+), 33 deletions(-) diff --git a/server.go b/server.go index d922c35..669493d 100644 --- a/server.go +++ b/server.go @@ -295,45 +295,49 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } } } - if readBatchCh != nil { - // We are exiting the read loop while still in the middle of consuming a batch of replayed - // events from a Repository (e.g. the subscriber disconnected, or MaxConnTime elapsed). The - // Repository's producer goroutine may be blocked trying to send the remaining events on this - // channel. Since we hold the receiving end -- we can neither close it nor keep reading it on - // this exiting goroutine -- drain it in the background so the producer can unblock and release - // its resources promptly rather than leaking until the process exits. - // - // A Repository that implements RepositoryWithContext will already have been told to stop via - // context cancellation, but draining is harmless in that case and remains the safety net for - // repositories that only implement Replay. - go drainReplayedEvents(readBatchCh) - } - // A replay batch that the Server queued on eventCh but that the loop above never dequeued - // would strand its producer the same way. Server.run() drains such a batch when it - // processes our unsubscription (see the unsubs case there); this sweep of the values - // already buffered additionally covers the case where the Server has shut down and will - // never process it. Anything the Server enqueues concurrently with this sweep is still - // handled by the unsubs path. - SweepPending: - for { - select { - case ev, ok := <-eventCh: - if !ok { - break SweepPending - } - if batch, isBatch := ev.(eventBatch); isBatch { - go drainReplayedEvents(batch.events) - } - default: - break SweepPending - } - } + drainAbandonedBatches(readBatchCh, eventCh) if !closedNormally { unsubscribe() // the server didn't tell us to close, so we must tell it that we're closing } } } +// drainAbandonedBatches unblocks Repository producers whose replay batches an exiting handler +// will never consume: the batch it was reading when it exited, and any batch still queued +// unread on its buffered event channel. +// +// current is the batch the handler was mid-way through consuming, or nil. Its producer may be +// blocked sending the remaining events; since the handler holds the receiving end -- it can +// neither close the channel nor keep reading it -- the batch is drained in the background so +// the producer can unblock and release its resources promptly. A Repository that implements +// RepositoryWithContext will already have been told to stop via context cancellation, but +// draining is harmless in that case and remains the safety net for repositories that only +// implement Replay. +// +// A batch the Server queued on the buffered event channel that the handler never dequeued +// would strand its producer the same way. Server.run() drains such a batch when it processes +// the handler's unsubscription; the sweep of the already-buffered values here additionally +// covers the case where the Server has shut down and will never process it. Anything the +// Server enqueues concurrently with this sweep is still handled by the unsubscription path. +func drainAbandonedBatches(current <-chan Event, eventCh <-chan eventOrComment) { + if current != nil { + go drainReplayedEvents(current) + } + for { + select { + case ev, ok := <-eventCh: + if !ok { + return + } + if batch, isBatch := ev.(eventBatch); isBatch { + go drainReplayedEvents(batch.events) + } + default: + return + } + } +} + // drainReplayedEvents consumes and discards the events from a Repository replay batch channel that // no subscriber will read again, so the goroutine producing those events can complete and release // its resources instead of blocking forever on a send.