fix: Flush once per available batch of replayed events instead of per event

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,14 +144,21 @@ 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 {
 					srv.Logger.Println(err)
 				}
 				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,

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)
+		})
+	}
+}