From 373509e67da7dde5fdd5f7a565ced6a47b518533 Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 17 Jul 2026 12:44:30 -0400 Subject: [PATCH] fix: Flush once per available batch of replayed events instead of per event When a Repository replay delivers a large batch of events (for example, a proxy replaying a full data set to a newly connected SDK), the handler was calling flusher.Flush() after every event, producing one HTTP chunk write per event per connection. The handler now encodes the batch events that are immediately available and flushes once when the batch channel goes momentarily quiet, the batch ends, or maxEventsPerFlush (512) is reached. Timeliness is unchanged: a flush still happens before the handler blocks waiting for more data. Memory is bounded regardless of batch size -- the ResponseWriter's buffer writes through to the connection as it fills and a slow client backpressures the encode -- and the per-flush cap additionally guarantees the handler returns to its main select loop regularly, so connection close and MaxConnTime are honored even while an arbitrarily long batch is streaming. New BenchmarkReplayBatch (5000 events x 300B over a real HTTP connection, events preloaded in a buffered channel): 14.4ms -> 3.3ms per replay (-77%). Callers feeding the replay channel one unbuffered send at a time do not regress, but also do not benefit; the channel must be buffered for the drain to batch (BenchmarkReplayUnbuffered). --- server.go | 63 +++++++++++++++- server_replay_benchmark_test.go | 128 ++++++++++++++++++++++++++++++++ 2 files changed, 189 insertions(+), 2 deletions(-) create mode 100644 server_replay_benchmark_test.go diff --git a/server.go b/server.go index 78cbeea..ed7bc4a 100644 --- a/server.go +++ b/server.go @@ -39,6 +39,13 @@ type eventBatch struct { events <-chan Event } +// maxEventsPerFlush caps how many batch events the handler encodes between flushes. +// The value is large enough that the per-flush cost is fully amortized for big +// replays, while keeping a deterministic bound on how long the handler can stay away +// from its main select loop (which watches for connection close and MaxConnTime) +// while a batch is streaming. +const maxEventsPerFlush = 512 + // Server manages any number of event-publishing channels and allows subscribers to consume them. // To use it within an HTTP server, create a handler for each channel with Handler(). type Server struct { @@ -137,7 +144,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { flusher.Flush() enc := NewEncoder(w, useGzip) - writeEventOrComment := func(ec eventOrComment) bool { + encodeEventOrComment := func(ec eventOrComment) bool { if err := enc.Encode(ec); err != nil { srv.unsubs <- sub if srv.Logger != nil { @@ -145,6 +152,13 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { } return false // if this happens, we'll end the handler early because something's clearly broken } + return true + } + + writeEventOrComment := func(ec eventOrComment) bool { + if !encodeEventOrComment(ec) { + return false + } flusher.Flush() return true } @@ -265,7 +279,15 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { if !ok { // end of batch readBatchCh = nil readMainCh = eventCh - } else if !writeEventOrComment(ev) { + continue + } + encodeOK, batchDone := encodeAvailableBatchEvents(ev, readBatchCh, encodeEventOrComment) + if batchDone { + readBatchCh = nil + readMainCh = eventCh + } + flusher.Flush() + if !encodeOK { break ReadLoop } } @@ -412,6 +434,43 @@ func (srv *Server) markServerClosed() { srv.isClosed = true } +// Encodes ev plus any further events that are immediately available from batch, +// without flushing in between. Batches can contain many events that are already +// available (for example, a replay of a large data set), and flushing the connection +// after every one of them costs an HTTP chunk write per event; the caller flushes +// once after this returns. Encoding stops when the channel goes momentarily quiet, +// the batch ends, or maxEventsPerFlush events have been written. Memory stays +// bounded either way -- the ResponseWriter's buffer writes through to the connection +// as it fills, and a slow client blocks the encode -- but the cap guarantees the +// handler returns to its select loop regularly, so connection shutdown and +// MaxConnTime are still honored while a long batch is streaming. +// +// encodeOK is false if an encode failed (the handler should exit). batchDone is true +// if the batch channel was closed. +func encodeAvailableBatchEvents( + ev eventOrComment, + batch <-chan Event, + encode func(eventOrComment) bool, +) (encodeOK, batchDone bool) { + for encoded := 1; ; encoded++ { + if !encode(ev) { + return false, false + } + if encoded >= maxEventsPerFlush { + return true, false // budget spent; do not pull another event before flushing + } + select { + case next, more := <-batch: + if !more { + return true, true + } + ev = next + default: + return true, false + } + } +} + // Attempts to send an event or comment to the subscription's channel. // // We do not want to block the main Server goroutine, so this is a non-blocking send. If it fails, diff --git a/server_replay_benchmark_test.go b/server_replay_benchmark_test.go new file mode 100644 index 0000000..27ec91f --- /dev/null +++ b/server_replay_benchmark_test.go @@ -0,0 +1,128 @@ +package eventsource + +import ( + "bufio" + "fmt" + "net/http" + "net/http/httptest" + "strings" + "testing" +) + +type prerenderedEvent struct { + name string + data string +} + +func (e prerenderedEvent) Id() string { return "" } //nolint:revive +func (e prerenderedEvent) Event() string { return e.name } +func (e prerenderedEvent) Data() string { return e.data } + +// batchRepository replays a fixed set of events through a buffered channel, so every +// event is immediately available to the handler. This models a server-side proxy +// replaying a large pre-rendered data set to a newly connected client. +type batchRepository struct { + events []Event +} + +func (r batchRepository) Replay(channel, id string) chan Event { + out := make(chan Event, len(r.events)) + for _, e := range r.events { + out <- e + } + close(out) + return out +} + +// unbufferedRepository replays the same events through an unbuffered channel fed by a +// goroutine, one rendezvous per event. This models callers (like ld-relay today) that +// stream replay events instead of preloading them. +type unbufferedRepository struct { + events []Event +} + +func (r unbufferedRepository) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(out) + for _, e := range r.events { + out <- e + } + }() + return out +} + +func benchmarkReplayBatch(b *testing.B, numEvents, dataSize int) { + benchmarkReplay(b, numEvents, dataSize, func(events []Event) Repository { + return batchRepository{events: events} + }) +} + +func benchmarkReplayUnbuffered(b *testing.B, numEvents, dataSize int) { + benchmarkReplay(b, numEvents, dataSize, func(events []Event) Repository { + return unbufferedRepository{events: events} + }) +} + +func benchmarkReplay(b *testing.B, numEvents, dataSize int, makeRepo func([]Event) Repository) { + events := make([]Event, 0, numEvents) + data := strings.Repeat("x", dataSize) + for i := 0; i < numEvents; i++ { + events = append(events, prerenderedEvent{name: "put-object", data: data}) + } + + server := NewServer() + server.ReplayAll = true + defer server.Close() + server.Register("test", makeRepo(events)) + + httpServer := httptest.NewServer(server.Handler("test")) + defer httpServer.Close() + + b.SetBytes(int64(numEvents * dataSize)) + b.ResetTimer() + + for i := 0; i < b.N; i++ { + resp, err := http.Get(httpServer.URL) + if err != nil { + b.Fatal(err) + } + scanner := bufio.NewScanner(resp.Body) + count := 0 + for scanner.Scan() { + if strings.HasPrefix(scanner.Text(), "data: ") { + count++ + if count == numEvents { + break + } + } + } + _ = resp.Body.Close() + if count != numEvents { + b.Fatalf("expected %d events, got %d", numEvents, count) + } + } +} + +func BenchmarkReplayBatch(b *testing.B) { + for _, bc := range []struct{ numEvents, dataSize int }{ + {100, 300}, + {1000, 300}, + {5000, 300}, + {5000, 2000}, + } { + b.Run(fmt.Sprintf("%devents_%dB", bc.numEvents, bc.dataSize), func(b *testing.B) { + benchmarkReplayBatch(b, bc.numEvents, bc.dataSize) + }) + } +} + +func BenchmarkReplayUnbuffered(b *testing.B) { + for _, bc := range []struct{ numEvents, dataSize int }{ + {5000, 300}, + } { + b.Run(fmt.Sprintf("%devents_%dB", bc.numEvents, bc.dataSize), func(b *testing.B) { + benchmarkReplayUnbuffered(b, bc.numEvents, bc.dataSize) + }) + } +}