From c439e78affd5bcf2eede027f11c39dacc3bfaad9 Mon Sep 17 00:00:00 2001 From: barckcode Date: Mon, 12 Oct 2026 20:50:40 +0200 Subject: [PATCH 1/2] fix(api): a busy consumer is not an idle stream (#46) Follow's idle timer also expires while Follow itself is busy (a slow onFrame, or the goroutine descheduled on a loaded runner). The reader is then blocked handing over a line it already read, so lastRead goes stale and select could pick the expired timer over the waiting line: a working stream was reconnected as idle. Before an oversized frame, that turned the fatal refusal into a second connection, which is the calls=2 flake in TestFollowRefusesOversizedLinesAndFrames. When the timer fires, a line already waiting now wins, like a read that already ended did. TestFollowBusyConsumerIsNotIdle reproduces the race deterministically (20/20 failures before, including the exact "frame over 4194304 bytes calls=2"). The oversize tests no longer run on the 200ms test idle limit they do not exercise, and the slowly arriving line test keeps a wide margin between chunk gaps and the limit. Fixes #46 Co-Authored-By: Claude Opus 5.5 --- internal/api/runs.go | 105 ++++++++++++++++++++------------------ internal/api/runs_test.go | 46 ++++++++++++++++- 2 files changed, 100 insertions(+), 51 deletions(-) diff --git a/internal/api/runs.go b/internal/api/runs.go index 0d19c73..d9016ea 100644 --- a/internal/api/runs.go +++ b/internal/api/runs.go @@ -698,72 +698,79 @@ func (c *RunsClient) followOnce(ctx context.Context, id string, lastSeq *int64, return nil, healthy(), err } for { + var line string select { case <-ctx.Done(): return nil, healthy(), ctx.Err() case <-timer.C: - // A read that already ended decides the outcome, whichever of - // the two select happened to pick. + // A read that already ended, or a line already read and waiting + // for us, decides the outcome whichever case select happened to + // pick: the timer also expires while we are busy (in onFrame, or + // descheduled), and a reader blocked handing us a line has + // stopped reading because of us, not because the connection went + // quiet. Calling that idle reconnected a working stream, and + // turned an oversized frame into a second connection (#46). select { case err := <-readErr: return readEnded(err) + case line = <-lines: default: + if quiet := time.Since(time.Unix(0, lastRead.Load())); quiet < idle { + timer.Reset(idle - quiet) + continue + } + return nil, healthy(), errIdle } - if quiet := time.Since(time.Unix(0, lastRead.Load())); quiet < idle { - timer.Reset(idle - quiet) - continue - } - return nil, healthy(), errIdle case err := <-readErr: return readEnded(err) - case line := <-lines: - heard = true - if !timer.Stop() { - select { - case <-timer.C: - default: - } + case line = <-lines: + } + heard = true + if !timer.Stop() { + select { + case <-timer.C: + default: } - timer.Reset(idle) + } + timer.Reset(idle) - line = strings.TrimSuffix(line, "\r") - if line == "" { - if len(data) == 0 && eventName == "" { - continue - } - payload := strings.Join(data, "\n") - name := eventName - frameID := eventID - eventName, eventID, data, dataBytes = "", "", nil, 0 - run, delivered := dispatchFrame(name, frameID, payload, lastSeq, onFrame) - if run != nil { - if run.ID != id { - return nil, true, &fatalStreamError{errors.New("the event stream ended with a different run")} - } - return run, true, nil - } - if delivered { - progressed = true - } - continue - } - if strings.HasPrefix(line, ":") { + line = strings.TrimSuffix(line, "\r") + if line == "" { + if len(data) == 0 && eventName == "" { continue } - field, value, _ := strings.Cut(line, ":") - value = strings.TrimPrefix(value, " ") - switch field { - case "event": - eventName = value - case "id": - eventID = value - case "data": - dataBytes += len(value) - if dataBytes > maxSSEFrame { - return nil, healthy(), &fatalStreamError{fmt.Errorf("the platform sent a stream frame over %d bytes", maxSSEFrame)} + payload := strings.Join(data, "\n") + name := eventName + frameID := eventID + eventName, eventID, data, dataBytes = "", "", nil, 0 + run, delivered := dispatchFrame(name, frameID, payload, lastSeq, onFrame) + if run != nil { + if run.ID != id { + return nil, true, &fatalStreamError{errors.New("the event stream ended with a different run")} } - data = append(data, value) + return run, true, nil + } + if delivered { + progressed = true + } + continue + } + if strings.HasPrefix(line, ":") { + continue + } + field, value, _ := strings.Cut(line, ":") + value = strings.TrimPrefix(value, " ") + switch field { + case "event": + eventName = value + case "id": + eventID = value + case "data": + dataBytes += len(value) + if dataBytes > maxSSEFrame { + return nil, healthy(), &fatalStreamError{fmt.Errorf("the platform sent a stream frame over %d bytes", maxSSEFrame)} } + data = append(data, value) } } } diff --git a/internal/api/runs_test.go b/internal/api/runs_test.go index 305fc00..11a7631 100644 --- a/internal/api/runs_test.go +++ b/internal/api/runs_test.go @@ -551,6 +551,9 @@ func TestFollowRefusesOversizedLinesAndFrames(t *testing.T) { w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, body) }), "t", "") + // Not a test of the idle limit: a runner stalled for longer than the + // helper's 200ms must not turn into a reconnect here (#46). + c.IdleTimeout = 30 * time.Second _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) if err == nil || errors.Is(err, ErrStreamLost) || calls.Load() != 1 { t.Errorf("%s: err=%v calls=%d (want a fatal error, no retry)", name, err, calls.Load()) @@ -558,6 +561,43 @@ func TestFollowRefusesOversizedLinesAndFrames(t *testing.T) { } } +// The idle limit is about the connection, not about us: when the caller (or +// the scheduler, on a loaded CI runner) keeps Follow busy for longer than +// IdleTimeout while the next line is already read and waiting, the stream +// was not quiet and must not be reconnected. A reconnect here also turned an +// oversized frame into a second connection before it was refused (#46). +func TestFollowBusyConsumerIsNotIdle(t *testing.T) { + const events = 8 + for name, tail := range map[string]string{ + "end": sseEnd("succeeded"), + "frame": strings.Repeat("data: "+strings.Repeat("a", maxSSELine/2)+"\n", 10) + "\n", + } { + var calls atomic.Int32 + c := testRunsClient(t, http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + calls.Add(1) + w.Header().Set("Content-Type", "text/event-stream") + for seq := int64(1); seq <= events; seq++ { + _, _ = io.WriteString(w, sseEvent(seq, "log", `{}`)) + } + _, _ = io.WriteString(w, tail) + }), "t", "") + c.IdleTimeout = 30 * time.Millisecond + // Every frame keeps Follow busy for twice the idle limit, so each one + // is a chance for the expired idle timer and the waiting line to race. + _, err := c.Follow(context.Background(), runID, 0, func(Frame) { time.Sleep(60 * time.Millisecond) }) + switch name { + case "end": + if err != nil || calls.Load() != 1 { + t.Errorf("%s: err=%v calls=%d (want the run, one connection)", name, err, calls.Load()) + } + case "frame": + if err == nil || errors.Is(err, ErrStreamLost) || !strings.Contains(err.Error(), "stream frame over") || calls.Load() != 1 { + t.Errorf("%s: err=%v calls=%d (want the fatal oversize error, no retry)", name, err, calls.Load()) + } + } + } +} + // The idle limit is about a connection that has gone quiet, not about one // slow to finish a line: an oversized line that trickles in for longer than // IdleTimeout must still end as the fatal oversize error on the first @@ -569,7 +609,7 @@ func TestFollowOversizedLineArrivingSlowlyIsNotIdle(t *testing.T) { calls.Add(1) w.Header().Set("Content-Type", "text/event-stream") _, _ = io.WriteString(w, "data: ") - chunk := strings.Repeat("a", 128<<10) + chunk := strings.Repeat("a", 32<<10) for sent := 0; sent <= maxSSELine; sent += len(chunk) { if _, err := io.WriteString(w, chunk); err != nil { return @@ -582,7 +622,9 @@ func TestFollowOversizedLineArrivingSlowlyIsNotIdle(t *testing.T) { } } }), "t", "") - c.IdleTimeout = 60 * time.Millisecond + // The line takes ~33 chunks x 20ms to arrive, well over the idle limit, + // while no single gap comes near it, so a loaded runner is not idle. + c.IdleTimeout = 250 * time.Millisecond _, err := c.Follow(context.Background(), runID, 0, func(Frame) {}) if err == nil || errors.Is(err, ErrStreamLost) || !strings.Contains(err.Error(), "stream line over") || calls.Load() != 1 { t.Errorf("err=%v calls=%d (want the fatal oversize error, no retry)", err, calls.Load()) From 8e9257a2dcceff609e47acf404e8acc7479ea0cf Mon Sep 17 00:00:00 2001 From: barckcode Date: Mon, 12 Oct 2026 20:59:58 +0200 Subject: [PATCH 2/2] test(api): wider margins in the busy-consumer regression test Review follow-up: a 100ms idle limit with a 200ms busy consumer keeps the race reproducible (10/10 failures on the old client) without the limit itself being tight enough to reconnect on a loaded runner. Co-Authored-By: Claude Opus 5.5 --- internal/api/runs_test.go | 14 +++++++++----- 1 file changed, 9 insertions(+), 5 deletions(-) diff --git a/internal/api/runs_test.go b/internal/api/runs_test.go index 11a7631..7f80913 100644 --- a/internal/api/runs_test.go +++ b/internal/api/runs_test.go @@ -542,7 +542,8 @@ func TestFollowSkipsAMalformedEnd(t *testing.T) { func TestFollowRefusesOversizedLinesAndFrames(t *testing.T) { for name, body := range map[string]string{ - "line": "data: " + strings.Repeat("a", maxSSELine+1) + "\n\n", + "line": "data: " + strings.Repeat("a", maxSSELine+1) + "\n\n", + // refused on its 9th line: 9 x 512 KiB is over the 4 MiB frame cap "frame": strings.Repeat("data: "+strings.Repeat("a", maxSSELine/2)+"\n", 10) + "\n", } { var calls atomic.Int32 @@ -567,9 +568,10 @@ func TestFollowRefusesOversizedLinesAndFrames(t *testing.T) { // was not quiet and must not be reconnected. A reconnect here also turned an // oversized frame into a second connection before it was refused (#46). func TestFollowBusyConsumerIsNotIdle(t *testing.T) { - const events = 8 + const events = 5 for name, tail := range map[string]string{ - "end": sseEnd("succeeded"), + "end": sseEnd("succeeded"), + // refused on its 9th line: 9 x 512 KiB is over the 4 MiB frame cap "frame": strings.Repeat("data: "+strings.Repeat("a", maxSSELine/2)+"\n", 10) + "\n", } { var calls atomic.Int32 @@ -581,10 +583,12 @@ func TestFollowBusyConsumerIsNotIdle(t *testing.T) { } _, _ = io.WriteString(w, tail) }), "t", "") - c.IdleTimeout = 30 * time.Millisecond // Every frame keeps Follow busy for twice the idle limit, so each one // is a chance for the expired idle timer and the waiting line to race. - _, err := c.Follow(context.Background(), runID, 0, func(Frame) { time.Sleep(60 * time.Millisecond) }) + // The limit itself stays wide enough for a loaded runner to deliver + // the headers and the next line inside it. + c.IdleTimeout = 100 * time.Millisecond + _, err := c.Follow(context.Background(), runID, 0, func(Frame) { time.Sleep(200 * time.Millisecond) }) switch name { case "end": if err != nil || calls.Load() != 1 {