diff --git a/internal/api/agent.go b/internal/api/agent.go index 89a5ee3..0480fb6 100644 --- a/internal/api/agent.go +++ b/internal/api/agent.go @@ -108,6 +108,17 @@ func (h *handlers) agentChat(c *gin.Context) { send(chatEvent{Done: turn}) } +// sseWriteTimeout bounds ONE write to the stream, not the stream itself. +// +// It is refreshed per frame, which is the only shape that satisfies both ends: +// the server's absolute WriteTimeout would cut a long turn (#78), while removing +// the deadline entirely would let a client that stops reading block a write +// forever once the socket buffer fills — pinning the run goroutine and this +// stream's mutex with it, and taking the keep-alive down too since it needs the +// same lock. Generous, because it is a backstop against a stuck peer and not a +// pacing mechanism. +const sseWriteTimeout = 30 * time.Second + // eventStream serializes writes to one SSE response. // // The mutex is load-bearing, not decoration: step events are sent from the @@ -116,6 +127,7 @@ func (h *handlers) agentChat(c *gin.Context) { // frames long before it crashes anything. type eventStream struct { c *gin.Context + rc *http.ResponseController mu sync.Mutex } @@ -125,29 +137,33 @@ type eventStream struct { // so a proxy holding the response until it looks complete can't reintroduce // exactly the silence streaming exists to remove. // -// Clearing the write deadline is what makes a turn longer than the server's -// WriteTimeout possible at all (#78). That timeout is an ABSOLUTE deadline from -// when the request header was read, not an idle timeout, so a streaming response -// has to opt out of it explicitly or it is cut mid-turn no matter how recently +// Taking the write deadline off the server's absolute WriteTimeout and onto a +// per-write one is what makes a turn longer than 30s possible at all (#78). +// WriteTimeout is an ABSOLUTE deadline from when the request header was read, +// not an idle timeout, so a streaming response is cut mid-turn however recently // it wrote. Without this the 4-minute runTimeout is unreachable and the -// keep-alive below tops out at one tick — it would be pacing a connection that -// is destroyed underneath it. +// keep-alive below tops out at one tick — pacing a connection that is destroyed +// underneath it. // -// This failure is INVISIBLE from in here: the writes after the deadline return -// err == nil and their bytes are dropped, so there is nothing to detect and -// report. Only the client sees it, as a truncated stream it reports as a dropped -// connection. That is why this is done up front rather than handled on write. +// That failure is INVISIBLE from in here: writes past the deadline return +// err == nil and their bytes are dropped, so there is nothing to detect on the +// write path. Only the client sees it, as a truncated stream it reports as a +// dropped connection. Hence a deadline set up front and refreshed per frame, +// rather than anything checked after the fact. func openEventStream(c *gin.Context) *eventStream { c.Header("Content-Type", "text/event-stream") c.Header("Cache-Control", "no-cache") c.Header("X-Accel-Buffering", "no") - // A failure here is worth saying out loud: it means this stream WILL be cut - // at WriteTimeout and we cannot stop it. - if err := http.NewResponseController(c.Writer).SetWriteDeadline(time.Time{}); err != nil { - slog.Error("api: cannot clear SSE write deadline; long turns will be truncated", "error", err) + s := &eventStream{c: c, rc: http.NewResponseController(c.Writer)} + // Probe once here rather than reporting per frame: a writer that can't take + // deadlines will fail identically on every write, and the operator needs to + // hear it once. If this fails the stream still works — it is just back to + // being cut at WriteTimeout, which is worth saying out loud. + if err := s.rc.SetWriteDeadline(time.Now().Add(sseWriteTimeout)); err != nil { + slog.Error("api: SSE write deadlines unavailable; long turns will be truncated at the server WriteTimeout", "error", err) } c.Writer.Flush() - return &eventStream{c: c} + return s } func (s *eventStream) send(ev chatEvent) { @@ -162,6 +178,11 @@ func (s *eventStream) send(ev chatEvent) { func (s *eventStream) write(frame string) { s.mu.Lock() defer s.mu.Unlock() + // Refresh for THIS write, so the stream as a whole is unbounded but no single + // write is. Error deliberately unchecked: openEventStream already reported + // whether deadlines work at all, and this call can only fail the same way, so + // checking here would log once per frame to say the same thing. + _ = s.rc.SetWriteDeadline(time.Now().Add(sseWriteTimeout)) _, _ = io.WriteString(s.c.Writer, frame) s.c.Writer.Flush() } diff --git a/internal/api/sse_deadline_test.go b/internal/api/sse_deadline_test.go index 6676fd6..61a7f24 100644 --- a/internal/api/sse_deadline_test.go +++ b/internal/api/sse_deadline_test.go @@ -22,11 +22,15 @@ import ( // after the deadline return err == nil and their bytes are silently discarded. // A test that checked the return of io.WriteString would pass against the bug. func TestEventStreamOutlivesWriteTimeout(t *testing.T) { - // Long enough that the frames below straddle it, short enough to keep the - // test quick. Everything else is scaled off it. - const writeTimeout = 300 * time.Millisecond - const tick = 200 * time.Millisecond - const frames = 4 // last one lands at ~800ms, well past the deadline + // tick > writeTimeout on purpose, so EVERY frame lands after the deadline has + // already passed and the first one clears it by a full 100ms. The margin only + // ever grows: a slow or loaded CI runner delays frames further past a deadline + // that has already expired, so scheduling jitter pushes this test toward + // passing rather than toward flaking. Sizing it the other way — frames landing + // just before the deadline — would flake exactly when CI is busiest. + const writeTimeout = 200 * time.Millisecond + const tick = 300 * time.Millisecond + const frames = 3 // ~900ms total; the last lands 700ms past the deadline gin.SetMode(gin.TestMode) r := gin.New()