🐛 fix(crux): cmdlog next_seq advances past terminal unfinished and orphaned commands

changelog.d/fixed-crux-cmdlog-next-seq-terminal-unfinished.md

@@ -0,0 +1 @@
+- crux cmdlog: `next_seq` no longer pins forever on an ended command whose exit was never learned, and a runner restart closes commands it left open, so cmdlog followers always make progress ([#17410](https://github.com/hivecommons/hive/issues/17410)).

src/crux/api/cmdlog.go

@@ -88,8 +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).
+// NextSeq never advances past a record that has no ended line yet, 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

@@ -202,6 +202,64 @@ func TestReadNextSeqStopsBeforeFirstUnfinished(t *testing.T) {
 	}
 }
 
+func TestReadNextSeqAdvancesPastEndedUnfinished(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	for i := range 2 {
+		seq, err := l.Started(Start{Command: []string{"cmd", string(rune('a' + i))}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		end := End{}
+		if i == 1 {
+			end = End{ExitCode: intp(0)}
+		}
+		if err := l.Ended(seq, end); err != nil {
+			t.Fatal(err)
+		}
+	}
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 2 || page.Commands[0].Status != api.CommandUnfinished || page.NextSeq != 2 {
+		t.Fatalf("page = %+v, %v", page, err)
+	}
+}
+
+func TestReadMoreWithUnendedFirstClampsToSinceSeq(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	if _, err := l.Started(Start{Command: []string{"open"}}); err != nil {
+		t.Fatal(err)
+	}
+	for range 2 {
+		seq, _ := l.Started(Start{Command: []string{"done"}})
+		if err := l.Ended(seq, End{ExitCode: intp(0)}); err != nil {
+			t.Fatal(err)
+		}
+	}
+	page, err := Read(dir, 0, 2)
+	if err != nil || !page.More || page.NextSeq != 0 {
+		t.Fatalf("page = %+v, %v", page, err)
+	}
+}
+
+func TestOpenClosesOrphanedStartedLines(t *testing.T) {
+	dir := t.TempDir()
+	first := openLog(t, dir, Policy{})
+	seq, err := first.Started(Start{Command: []string{"orphan"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	openLog(t, dir, Policy{})
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 1 {
+		t.Fatalf("page = %+v, %v", page, err)
+	}
+	rec := page.Commands[0]
+	if rec.Status != api.CommandUnfinished || rec.EndedAt == nil || rec.Reason == "" || page.NextSeq != seq {
+		t.Fatalf("record = %+v, next_seq = %d", rec, page.NextSeq)
+	}
+}
+
 func TestOutputKeepsHeadAndTailPerPolicy(t *testing.T) {
 	dir := t.TempDir()
 	l := openLog(t, dir, Policy{HeadBytes: 4, TailBytes: 3})

src/crux/cmdlog/log.go

@@ -22,6 +22,9 @@ const (
 	// two policy windows of output plus the command line.
 	scanBuffer   = 64 * 1024
 	maxLineBytes = 16 * 1024 * 1024
+	// orphanReason is written on the ended line Open appends for a command
+	// the previous runner left open.
+	orphanReason = "runner restarted before the command ended"
 )
 
 // Log is the Recorder over one run's commands.jsonl. It appends only;
@@ -45,10 +48,34 @@ func Open(runDir string, pol Policy) (*Log, error) {
 	if err != nil && !errors.Is(err, ErrNotFound) {
 		return nil, err
 	}
+	started := map[int64]time.Time{}
 	for _, ln := range lines {
 		if ln.Seq >= l.next {
 			l.next = ln.Seq + 1
 		}
+		switch ln.Event {
+		case EventStarted:
+			started[ln.Seq] = ln.At
+		case EventEnded:
+			delete(started, ln.Seq)
+		}
+	}
+	// A started seq with no ended line belonged to a runner that is gone;
+	// close it so it is a terminal record, not a permanent pin on NextSeq.
+	orphans := make([]int64, 0, len(started))
+	for seq := range started {
+		orphans = append(orphans, seq)
+	}
+	sort.Slice(orphans, func(i, j int) bool { return orphans[i] < orphans[j] })
+	for _, seq := range orphans {
+		now := l.pol.Now()
+		ms := now.Sub(started[seq]).Milliseconds()
+		if err := l.append(line{
+			Seq: seq, Event: EventEnded, At: now, DurationMs: &ms,
+			Status: api.CommandUnfinished, Reason: orphanReason,
+		}); err != nil {
+			return nil, err
+		}
 	}
 	return l, nil
 }
@@ -119,8 +146,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. NextSeq never
-// advances past an unfinished record, so a follower re-receives it until
-// it is terminal (idempotent by seq).
+// advances past a record that has no ended line yet, so a follower
+// re-receives it until it is terminal (idempotent by seq). A record with an
+// ended line is terminal whatever its status says.
 func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	lines, err := readLines(filepath.Join(runDir, FileName))
 	if err != nil {
@@ -165,7 +193,7 @@ 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 {
+			if rec.EndedAt == nil {
 				page.NextSeq = max(rec.Seq-1, sinceSeq)
 				break
 			}