feat: log warning when SSE flush to client is slow (#66)

Adds slow client detection to the EventStream SSE sender. After each
flush, the elapsed time is measured using an injected quartz.Clock.
If it exceeds 500ms, a Warn log is emitted with the flush duration,
making slow clients immediately visible in logs without requiring
distributed tracing infrastructure.

quartz.Clock is injected into EventStream so tests can simulate slow
flushes by advancing a mock clock — no real sleeping required.

Author: mtojek

intercept/chatcompletions/streaming.go

@@ -28,6 +28,7 @@ import (
 	"github.com/coder/aibridge/tracing"
 
 	"cdr.dev/slog/v3"
+	"github.com/coder/quartz"
 )
 
 type StreamingInterception struct {
@@ -105,7 +106,7 @@ func (i *StreamingInterception) ProcessRequest(w http.ResponseWriter, r *http.Re
 	defer streamCancel(xerrors.New("deferred"))
 
 	// events will either terminate when shutdown after interaction with upstream completes, or when streamCtx is done.
-	events := eventstream.NewEventStream(streamCtx, logger.Named("sse-sender"), nil)
+	events := eventstream.NewEventStream(streamCtx, logger.Named("sse-sender"), nil, quartz.NewReal())
 	go events.Start(w, r)
 	defer func() {
 		_ = events.Shutdown(streamCtx) // Catch-all in case it doesn't get shutdown after stream completes.

intercept/eventstream/eventstream.go

@@ -15,17 +15,22 @@ import (
 	"golang.org/x/xerrors"
 
 	"cdr.dev/slog/v3"
+	"github.com/coder/quartz"
 )
 
 var ErrEventStreamClosed = xerrors.New("event stream closed")
 
-const pingInterval = time.Second * 10
+const (
+	pingInterval       = time.Second * 10
+	slowFlushThreshold = time.Millisecond * 500
+)
 
 type event []byte
 
 type EventStream struct {
 	ctx    context.Context
 	logger slog.Logger
+	clk    quartz.Clock
 
 	pingPayload []byte
 
@@ -43,7 +48,7 @@ type EventStream struct {
 }
 
 // NewEventStream creates a new SSE stream, with an optional payload which is used to send pings every [pingInterval].
-func NewEventStream(ctx context.Context, logger slog.Logger, pingPayload []byte) *EventStream {
+func NewEventStream(ctx context.Context, logger slog.Logger, pingPayload []byte, clk quartz.Clock) *EventStream {
 	// Send periodic pings to keep connections alive.
 	// The upstream provider may also send their own pings, but we can't rely on this.
 	tick := time.NewTicker(time.Nanosecond)
@@ -52,6 +57,7 @@ func NewEventStream(ctx context.Context, logger slog.Logger, pingPayload []byte)
 	return &EventStream{
 		ctx:    ctx,
 		logger: logger,
+		clk:    clk,
 
 		pingPayload: pingPayload,
 
@@ -131,10 +137,14 @@ func (s *EventStream) Start(w http.ResponseWriter, r *http.Request) {
 			}
 			return
 		}
+		flushStart := s.clk.Now()
 		if err := flush(w); err != nil {
 			s.logger.Warn(ctx, "failed to flush event stream", slog.Error(err))
 			return
 		}
+		if d := s.clk.Since(flushStart); d > slowFlushThreshold {
+			s.logger.Warn(ctx, "slow client detected", slog.F("flush_duration", d))
+		}
 
 		// Reset the timer once we've flushed some data to the stream, since it's already fresh.
 		// No need to ping in that case.

intercept/eventstream/eventstream_test.go

@@ -0,0 +1,132 @@
+package eventstream_test
+
+import (
+	"bufio"
+	"context"
+	"net"
+	"net/http"
+	"net/http/httptest"
+	"sync"
+	"testing"
+	"time"
+
+	"cdr.dev/slog/v3"
+	"cdr.dev/slog/v3/sloggers/slogtest"
+	"github.com/stretchr/testify/require"
+
+	"github.com/coder/aibridge/intercept/eventstream"
+	"github.com/coder/quartz"
+)
+
+// captureSink collects log entries for assertions in tests.
+type captureSink struct {
+	mu      sync.Mutex
+	entries []slog.SinkEntry
+}
+
+func (s *captureSink) LogEntry(_ context.Context, e slog.SinkEntry) {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	s.entries = append(s.entries, e)
+}
+
+func (*captureSink) Sync() {}
+
+func (s *captureSink) warns() []slog.SinkEntry {
+	s.mu.Lock()
+	defer s.mu.Unlock()
+	var out []slog.SinkEntry
+	for _, e := range s.entries {
+		if e.Level == slog.LevelWarn {
+			out = append(out, e)
+		}
+	}
+	return out
+}
+
+// clockAdvancingFlusher wraps httptest.ResponseRecorder and advances the mock
+// clock on each Flush call, simulating a slow client without real sleeping.
+type clockAdvancingFlusher struct {
+	*httptest.ResponseRecorder
+	clk     *quartz.Mock
+	advance time.Duration
+}
+
+func (f *clockAdvancingFlusher) Flush() {
+	f.clk.Advance(f.advance)
+	f.ResponseRecorder.Flush()
+}
+
+// Hijack satisfies the FullResponseWriter lint rule.
+func (f *clockAdvancingFlusher) Hijack() (net.Conn, *bufio.ReadWriter, error) {
+	return nil, nil, nil
+}
+
+func TestEventStream_LogsWarning_WhenFlushIsSlow(t *testing.T) {
+	t.Parallel()
+
+	sink := &captureSink{}
+	logger := slogtest.Make(t, nil).AppendSinks(sink).Leveled(slog.LevelWarn)
+	ctx := context.Background()
+	clk := quartz.NewMock(t)
+
+	stream := eventstream.NewEventStream(ctx, logger, nil, clk)
+
+	w := &clockAdvancingFlusher{
+		ResponseRecorder: httptest.NewRecorder(),
+		clk:              clk,
+		advance:          600 * time.Millisecond, // exceeds slowFlushThreshold (500ms)
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "/", nil)
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		stream.Start(w, req)
+	}()
+
+	stream.InitiateStream(w)
+	require.NoError(t, stream.SendRaw(ctx, []byte("data: hello\n\n")))
+	require.NoError(t, stream.Shutdown(ctx))
+	<-done
+
+	warns := sink.warns()
+	require.Len(t, warns, 1)
+	require.Equal(t, "slow client detected", warns[0].Message)
+}
+
+func TestEventStream_NoWarning_WhenFlushIsFast(t *testing.T) {
+	t.Parallel()
+
+	sink := &captureSink{}
+	logger := slogtest.Make(t, nil).AppendSinks(sink).Leveled(slog.LevelWarn)
+	ctx := context.Background()
+	clk := quartz.NewMock(t)
+
+	stream := eventstream.NewEventStream(ctx, logger, nil, clk)
+
+	// No clock advance — flush duration stays at 0, below threshold.
+	w := &clockAdvancingFlusher{
+		ResponseRecorder: httptest.NewRecorder(),
+		clk:              clk,
+		advance:          0,
+	}
+
+	req, err := http.NewRequestWithContext(ctx, http.MethodGet, "/", nil)
+	require.NoError(t, err)
+
+	done := make(chan struct{})
+	go func() {
+		defer close(done)
+		stream.Start(w, req)
+	}()
+
+	stream.InitiateStream(w)
+	require.NoError(t, stream.SendRaw(ctx, []byte("data: hello\n\n")))
+	require.NoError(t, stream.Shutdown(ctx))
+	<-done
+
+	require.Empty(t, sink.warns())
+}

intercept/messages/streaming.go

@@ -29,6 +29,7 @@ import (
 	"github.com/coder/aibridge/tracing"
 
 	"cdr.dev/slog/v3"
+	"github.com/coder/quartz"
 )
 
 type StreamingInterception struct {
@@ -140,7 +141,7 @@ func (i *StreamingInterception) ProcessRequest(w http.ResponseWriter, r *http.Re
 	}
 
 	// events will either terminate when shutdown after interaction with upstream completes, or when streamCtx is done.
-	events := eventstream.NewEventStream(streamCtx, logger.Named("sse-sender"), i.pingPayload())
+	events := eventstream.NewEventStream(streamCtx, logger.Named("sse-sender"), i.pingPayload(), quartz.NewReal())
 	go events.Start(w, r)
 	defer func() {
 		_ = events.Shutdown(streamCtx) // Catch-all in case it doesn't get shutdown after stream completes.

intercept/responses/streaming.go

@@ -18,6 +18,7 @@ import (
 	"cdr.dev/slog/v3"
 	"github.com/coder/aibridge/config"
 	aibcontext "github.com/coder/aibridge/context"
+	"github.com/coder/quartz"
 	"github.com/coder/aibridge/intercept"
 	"github.com/coder/aibridge/intercept/eventstream"
 	"github.com/coder/aibridge/mcp"
@@ -83,7 +84,7 @@ func (i *StreamingResponsesInterceptor) ProcessRequest(w http.ResponseWriter, r
 
 	i.injectTools()
 
-	events := eventstream.NewEventStream(ctx, i.logger.Named("sse-sender"), nil)
+	events := eventstream.NewEventStream(ctx, i.logger.Named("sse-sender"), nil, quartz.NewReal())
 	go events.Start(w, r)
 	defer func() {
 		shutdownCtx, shutdownCancel := context.WithTimeout(ctx, streamShutdownTimeout)