fix(api): a busy consumer is not an idle stream (#46)

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

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,13 +552,56 @@ 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())
 		}
 	}
 }
 
+// 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())