🐛 fix(crux): cmdlog next_seq never skips an unfinished command's ended line

changelog.d/fixed-cmdlog-follow-unfinished.md

@@ -0,0 +1 @@
+- crux cmdlog: a follower polling with next_seq now receives a command's ended record instead of seeing it as unfinished forever (#17398).

src/crux/api/cmdlog.go

@@ -88,6 +88,8 @@ type CommandRecord struct {
 // CommandPage is GET /v1/runs/{id}/cmdlog. NextSeq is the seq to pass as
 // since_seq for the following page; More says one exists now. A caller
 // following a live run keeps polling with NextSeq after More is false.
+// NextSeq never advances past an unfinished record, so a follower
+// re-receives it until it is terminal (idempotent by seq).
 type CommandPage struct {
 	Run      string          `json:"run"`
 	Commands []CommandRecord `json:"commands"`

src/crux/cmdlog/cmdlog_test.go

@@ -161,6 +161,47 @@ func TestReadPagesWithSinceSeqAndLimit(t *testing.T) {
 	}
 }
 
+func TestReadFollowerSeesEndedLineOfUnfinishedCommand(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	seq, err := l.Started(Start{Command: []string{"make", "test"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 1 || page.Commands[0].Status != api.CommandUnfinished || page.NextSeq != 0 {
+		t.Fatalf("unfinished page = %+v, %v", page, err)
+	}
+	if err := l.Ended(seq, End{ExitCode: intp(0)}); err != nil {
+		t.Fatal(err)
+	}
+	page, err = Read(dir, page.NextSeq, 0)
+	if err != nil || len(page.Commands) != 1 || page.Commands[0].Status != api.CommandCompleted || page.NextSeq != seq {
+		t.Fatalf("terminal page = %+v, %v", page, err)
+	}
+}
+
+func TestReadNextSeqStopsBeforeFirstUnfinished(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	for i := range 3 {
+		seq, err := l.Started(Start{Command: []string{"cmd", string(rune('a' + i))}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		if i == 1 {
+			continue
+		}
+		if err := l.Ended(seq, End{ExitCode: intp(0)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 3 || page.NextSeq != 1 || page.Commands[1].Status != api.CommandUnfinished {
+		t.Fatalf("mixed page = %+v, %v", page, err)
+	}
+}
+
 func TestOutputKeepsHeadAndTailPerPolicy(t *testing.T) {
 	dir := t.TempDir()
 	l := openLog(t, dir, Policy{HeadBytes: 4, TailBytes: 3})

src/crux/cmdlog/log.go

@@ -118,7 +118,9 @@ func (l *Log) append(ln line) error {
 
 // Read folds runDir's log into records after sinceSeq (exclusive), at
 // most limit of them (DefaultPageSize when zero, MaxPageSize at most).
-// A started line with no ended line is an unfinished record.
+// A started line with no ended line is an unfinished record. NextSeq never
+// advances past an unfinished record, so a follower re-receives it until
+// it is terminal (idempotent by seq).
 func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	lines, err := readLines(filepath.Join(runDir, FileName))
 	if err != nil {
@@ -132,9 +134,6 @@ func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	}
 	bySeq := map[int64]*api.CommandRecord{}
 	for _, ln := range lines {
-		if ln.Seq <= sinceSeq {
-			continue
-		}
 		switch ln.Event {
 		case EventStarted:
 			bySeq[ln.Seq] = &api.CommandRecord{
@@ -153,6 +152,9 @@ func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	}
 	all := make([]api.CommandRecord, 0, len(bySeq))
 	for _, rec := range bySeq {
+		if rec.Seq <= sinceSeq {
+			continue
+		}
 		all = append(all, *rec)
 	}
 	sort.Slice(all, func(i, j int) bool { return all[i].Seq < all[j].Seq })
@@ -162,6 +164,12 @@ func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	}
 	if n := len(page.Commands); n > 0 {
 		page.NextSeq = page.Commands[n-1].Seq
+		for _, rec := range page.Commands {
+			if rec.Status == api.CommandUnfinished {
+				page.NextSeq = max(rec.Seq-1, sinceSeq)
+				break
+			}
+		}
 	}
 	return page, nil
 }