From 29609aa78c8af5ca65645f093294deadacf9e50c Mon Sep 17 00:00:00 2001 From: Matthew Keeler Date: Fri, 24 Jul 2026 14:09:58 -0400 Subject: [PATCH] fix: Flush once per replayed batch instead of once per event 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. --- server.go | 19 ++++- server_batch_responsiveness_test.go | 88 +++++++++++++++++++ server_replay_benchmark_test.go | 128 ++++++++++++++++++++++++++++ 3 files changed, 231 insertions(+), 4 deletions(-) create mode 100644 server_batch_responsiveness_test.go create mode 100644 server_replay_benchmark_test.go diff --git a/server.go b/server.go index 78cbeea..7cb26c6 100644 --- a/server.go +++ b/server.go @@ -145,6 +145,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 + } + + writeEventOrCommentAndFlush := func(ec eventOrComment) bool { + if !writeEventOrComment(ec) { + return false + } flusher.Flush() return true } @@ -209,7 +216,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { continue } - ok := writeEventOrComment(delayedEvent) + ok := writeEventOrCommentAndFlush(delayedEvent) delayedEvent = nil if !ok { @@ -227,7 +234,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // any event that was pending processing. if delayedEvent != nil { jitterTimer.Stop() - ok := writeEventOrComment(delayedEvent) + ok := writeEventOrCommentAndFlush(delayedEvent) delayedEvent = nil if !ok { @@ -242,7 +249,7 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { // Write immediately if we aren't using the jitter functionality. if !usingJitter { - if !writeEventOrComment(ev) { + if !writeEventOrCommentAndFlush(ev) { break ReadLoop } continue @@ -263,9 +270,13 @@ func (srv *Server) Handler(channel string) http.HandlerFunc { case ev, ok := <-readBatchCh: if !ok { // end of batch + flusher.Flush() readBatchCh = nil readMainCh = eventCh - } else if !writeEventOrComment(ev) { + continue + } + + if !writeEventOrComment(ev) { break ReadLoop } } diff --git a/server_batch_responsiveness_test.go b/server_batch_responsiveness_test.go new file mode 100644 index 0000000..d4e8719 --- /dev/null +++ b/server_batch_responsiveness_test.go @@ -0,0 +1,88 @@ +package eventsource + +import ( + "bufio" + "net/http" + "net/http/httptest" + "testing" + "time" +) + +// floodingRepo returns a batch channel that is continuously ready with events and never +// closes, until stop is signalled. It models a Repository that floods the handler during +// replay -- the case that would starve the handler's select loop if the handler drained +// the batch in a tight inner loop instead of returning to select between events. +type floodingRepo struct { + data string + stop <-chan struct{} +} + +func (r floodingRepo) Replay(channel, id string) chan Event { + out := make(chan Event) + go func() { + defer close(out) + ev := prerenderedFloodEvent{data: r.data} + for { + select { + case out <- ev: + case <-r.stop: + return + } + } + }() + return out +} + +type prerenderedFloodEvent struct{ data string } + +func (e prerenderedFloodEvent) Id() string { return "" } //nolint:revive +func (e prerenderedFloodEvent) Event() string { return "put" } +func (e prerenderedFloodEvent) Data() string { return e.data } + +// TestMaxConnTimeInterruptsFloodingReplay proves that MaxConnTime is honored even while +// a batch replay is actively flooding the handler with events. Because each batch event +// now flows through one iteration of the main select loop, the maxConnTimeCh case is +// evaluated between events and the handler exits promptly -- rather than staying away +// from the select loop until the (here, unbounded) batch drains, which would ignore +// MaxConnTime entirely. +func TestMaxConnTimeInterruptsFloodingReplay(t *testing.T) { + stop := make(chan struct{}) + defer close(stop) + + server := NewServer() + server.ReplayAll = true + server.MaxConnTime = 100 * time.Millisecond + defer server.Close() + server.Register("test", floodingRepo{data: "hello", stop: stop}) + + httpServer := httptest.NewServer(server.Handler("test")) + defer httpServer.Close() + + start := time.Now() + resp, err := http.Get(httpServer.URL) + if err != nil { + t.Fatal(err) + } + defer func() { _ = resp.Body.Close() }() + + // Drain the body until the server closes the connection at MaxConnTime. If the + // handler were stuck draining the unbounded batch, this would never return and the + // test would time out. + scanner := bufio.NewScanner(resp.Body) + scanner.Buffer(make([]byte, 0, 64*1024), 1024*1024) + events := 0 + for scanner.Scan() { + if len(scanner.Bytes()) > 0 { + events++ + } + } + elapsed := time.Since(start) + + if elapsed > 2*time.Second { + t.Fatalf("handler did not honor MaxConnTime during replay: took %v", elapsed) + } + if events == 0 { + t.Fatal("expected to receive at least some replayed events before MaxConnTime") + } + t.Logf("MaxConnTime honored mid-replay: connection closed after %v having delivered ~%d lines", elapsed, events) +} 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) + }) + } +}