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

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

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

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