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..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 @@ -551,6 +552,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 +562,46 @@ 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 = 5 + for name, tail := range map[string]string{ + "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 + 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", "") + // 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. + // 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 { + 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 +613,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 +626,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())