Merge remote-tracking branch 'origin/v5' into scanner/fix-137

Signed-off-by: scanner <280983584+hivecommons-hive[bot]@users.noreply.github.com>

Author: hivecommons-hive[bot]

changelog.d/fixed-17410-cmdlog-next-seq-pin.md

@@ -0,0 +1 @@
+- Crux's command log (`GET /v1/runs/{id}/cmdlog`) no longer pins a follower's `next_seq` forever ([#17410](https://github.com/hivecommons/hive/issues/17410)). A command that ended `unfinished` (the launcher failed without an exit the runtime could read) is now terminal, so `next_seq` moves past it; a command left in flight by a runner that died is ended `unfinished` when its log is next opened, which `crux serve` now does for every run it reconciles at start; and `more` is never true on a page whose `next_seq` did not move, so a "page while `more`" loop such as `crux cmdlog list --all` can no longer spin on the same request.

src/crux/api/cmdlog.go

@@ -67,8 +67,9 @@ type CommandOutput struct {
 	Truncated bool   `json:"truncated,omitempty"`
 }
 
-// CommandRecord is one dispatched command. ExitCode and EndedAt are
-// absent while the command is unfinished.
+// CommandRecord is one dispatched command. EndedAt is absent while the
+// command is in flight; ExitCode is absent whenever the exit is not known,
+// so an unfinished record can carry EndedAt without ExitCode.
 type CommandRecord struct {
 	Seq       int64      `json:"seq"`
 	Command   []string   `json:"command"`
@@ -88,8 +89,11 @@ 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 with no EndedAt, so a follower
+// re-receives an in-flight command until it ends (idempotent by seq); a
+// record with EndedAt is terminal whatever its Status. More is never set
+// on a page whose NextSeq equals the since_seq it answered, so paging
+// while More is true always makes progress.
 type CommandPage struct {
 	Run      string          `json:"run"`
 	Commands []CommandRecord `json:"commands"`

src/crux/cli/serve.go

@@ -214,7 +214,8 @@ func (e *Env) openRuns(workDir, path string) (*runs.Service, error) {
 	}
 	svc := runs.New(store, backend, cfg)
 	// S4: every dispatched command is appended to <run dir>/commands.jsonl
-	// by the runner, with the default head/tail output policy.
+	// by the runner, with the default head/tail output policy. It is set
+	// before Reconcile so a command left in flight gets its ended line.
 	svc.CmdLog = cmdlog.FileOpener(cmdlog.Policy{})
 	// S5: turn signals go to <run dir>/signals.jsonl. The book is set
 	// before Reconcile so a run left in flight gets its done signal.

src/crux/cmdlog/cmdlog.go

@@ -5,8 +5,10 @@
 //
 // The log is one commands.jsonl file in the run's directory. Each command
 // contributes a "started" line when it is dispatched and an "ended" line
-// when its exit is known; a command with no "ended" line is unfinished,
-// never guessed. Readers fold the two into one api.CommandRecord per seq.
+// once the runtime stops waiting on it, carrying the exit when it is
+// known; a command with no "ended" line is in flight and reads as
+// unfinished, never guessed. Readers fold the two into one
+// api.CommandRecord per seq.
 //
 // Honest scope: a record says how the command was observed in its
 // capture field. The S2 runner records what it hands to the sandbox
@@ -34,6 +36,10 @@ const (
 	EventEnded   = "ended"
 )
 
+// OrphanReason is the reason on the unfinished ended line Open appends
+// for a command whose started line an earlier writer never ended.
+const OrphanReason = "the log was reopened with no exit recorded for this command (the runtime restarted, or its ended write failed); the exit is not knowable"
+
 // Output policy defaults: how much of each stream a record keeps. The
 // full output stays with the run (runs.StdoutArtifact); the record keeps
 // enough of both ends to tell what happened without re-reading it.

src/crux/cmdlog/cmdlog_test.go

@@ -120,12 +120,57 @@ func TestOpenResumesTheSeqCounter(t *testing.T) {
 	if err != nil || seq != 4 {
 		t.Fatalf("resumed seq = %d, %v", seq, err)
 	}
-	// The earlier log's open commands are not the new log's to end.
+	// Ended still refuses a seq this log never started, even one Open ended.
 	if err := second.Ended(1, End{}); !errors.Is(err, ErrUnknownSeq) {
 		t.Fatalf("Ended(1) on resumed = %v", err)
 	}
 }
 
+// A started line no writer will ever end (its runner died) is ended
+// unfinished by the next Open, so it cannot pin NextSeq for the rest of
+// the run (#17410).
+func TestOpenEndsACommandAnEarlierWriterLeftOpen(t *testing.T) {
+	dir := t.TempDir()
+	c := newClock()
+	first := openLog(t, dir, Policy{Now: c.tick})
+	orphan, err := first.Started(Start{Command: []string{"make", "test"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	seq, err := first.Started(Start{Command: []string{"echo"}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := first.Ended(seq, End{ExitCode: intp(0)}); err != nil {
+		t.Fatal(err)
+	}
+	second := openLog(t, dir, Policy{Now: c.tick})
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 2 || page.NextSeq != seq || page.More {
+		t.Fatalf("page = %+v, %v", page, err)
+	}
+	rec := page.Commands[0]
+	if rec.Seq != orphan || rec.Status != api.CommandUnfinished || rec.EndedAt == nil || rec.ExitCode != nil || rec.Reason != OrphanReason || rec.DurationMs == nil || *rec.DurationMs != 750 {
+		t.Fatalf("orphan = %+v", rec)
+	}
+	if page.Commands[1].Status != api.CommandCompleted {
+		t.Fatalf("ended command = %+v", page.Commands[1])
+	}
+	// The orphan is ended once: reopening again appends nothing.
+	path := filepath.Join(dir, FileName)
+	before, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	openLog(t, dir, Policy{})
+	if after, err := os.ReadFile(path); err != nil || string(after) != string(before) {
+		t.Fatalf("second reopen rewrote the log: %q, %v", after, err)
+	}
+	if next, err := second.Started(Start{Command: []string{"y"}}); err != nil || next != seq+1 {
+		t.Fatalf("resumed seq = %d, %v", next, err)
+	}
+}
+
 func TestReadPagesWithSinceSeqAndLimit(t *testing.T) {
 	dir := t.TempDir()
 	l := openLog(t, dir, Policy{})
@@ -202,6 +247,71 @@ func TestReadNextSeqStopsBeforeFirstUnfinished(t *testing.T) {
 	}
 }
 
+// An ended line makes a record terminal whatever its status: a command
+// ended unfinished (exit never known) must not hold NextSeq back (#17410).
+func TestReadNextSeqPassesAnEndedUnfinishedRecord(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	for i := range 4 {
+		seq, err := l.Started(Start{Command: []string{"cmd", string(rune('a' + i))}})
+		if err != nil {
+			t.Fatal(err)
+		}
+		end := End{ExitCode: intp(0)}
+		if i == 1 {
+			end = End{Reason: "launcher error"}
+		}
+		if err := l.Ended(seq, end); err != nil {
+			t.Fatal(err)
+		}
+	}
+	page, err := Read(dir, 0, 0)
+	if err != nil || len(page.Commands) != 4 || page.NextSeq != 4 || page.More {
+		t.Fatalf("page = %+v, %v", page, err)
+	}
+	if rec := page.Commands[1]; rec.Status != api.CommandUnfinished || rec.EndedAt == nil || rec.ExitCode != nil {
+		t.Fatalf("ended unfinished = %+v", rec)
+	}
+	page, err = Read(dir, 0, 2)
+	if err != nil || len(page.Commands) != 2 || page.NextSeq != 2 || !page.More {
+		t.Fatalf("page 1 = %+v, %v", page, err)
+	}
+	page, err = Read(dir, page.NextSeq, 2)
+	if err != nil || len(page.Commands) != 2 || page.NextSeq != 4 || page.More {
+		t.Fatalf("page 2 = %+v, %v", page, err)
+	}
+}
+
+// More is set only on a page whose cursor moved: with an in-flight command
+// first on the page, the documented "page while More" loop would otherwise
+// re-request the same since_seq forever (#17410).
+func TestReadMoreOnlyWhenNextSeqAdvances(t *testing.T) {
+	dir := t.TempDir()
+	l := openLog(t, dir, Policy{})
+	for i := range 6 {
+		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)
+		}
+	}
+	// Seq 2 is in flight. The first page still moves the cursor past seq 1.
+	page, err := Read(dir, 0, 2)
+	if err != nil || len(page.Commands) != 2 || page.NextSeq != 1 || !page.More {
+		t.Fatalf("page 1 = %+v, %v", page, err)
+	}
+	// From there seq 2 leads the page and the cursor cannot move.
+	page, err = Read(dir, page.NextSeq, 2)
+	if err != nil || len(page.Commands) != 2 || page.Commands[0].Seq != 2 || page.NextSeq != 1 || page.More {
+		t.Fatalf("pinned 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

@@ -5,8 +5,10 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
+	"maps"
 	"os"
 	"path/filepath"
+	"slices"
 	"sort"
 	"sync"
 	"time"
@@ -39,16 +41,43 @@ type Log struct {
 
 // Open returns the Log for runDir, resuming the seq counter from a file
 // that already exists so a restarted runner never reuses a number.
+//
+// A run has one live writer at a time (a contained run dispatches one
+// command; a session's steps run one at a time), so a started seq with no
+// ended line on disk belongs to a writer that is gone and will never end
+// it. Open appends an unfinished ended line with OrphanReason for each,
+// so a crash leaves a terminal record instead of one that pins every
+// follower's NextSeq behind it for the rest of the run.
 func Open(runDir string, pol Policy) (*Log, error) {
 	l := &Log{path: filepath.Join(runDir, FileName), pol: pol.defaults(), next: 1, open: map[int64]time.Time{}}
 	lines, err := readLines(l.path)
 	if err != nil && !errors.Is(err, ErrNotFound) {
 		return nil, err
 	}
+	// orphans folds the lines the way Read does: a started line opens its
+	// seq, a later ended line closes it.
+	orphans := map[int64]time.Time{}
 	for _, ln := range lines {
 		if ln.Seq >= l.next {
 			l.next = ln.Seq + 1
 		}
+		switch ln.Event {
+		case EventStarted:
+			orphans[ln.Seq] = ln.At
+		case EventEnded:
+			delete(orphans, ln.Seq)
+		}
+	}
+	if len(orphans) == 0 {
+		return l, nil
+	}
+	now := l.pol.Now()
+	reason := l.pol.Scrub(OrphanReason)
+	for _, seq := range slices.Sorted(maps.Keys(orphans)) {
+		ms := now.Sub(orphans[seq]).Milliseconds()
+		if err := l.append(line{Seq: seq, Event: EventEnded, At: now, DurationMs: &ms, Status: api.CommandUnfinished, Reason: reason}); err != nil {
+			return nil, err
+		}
 	}
 	return l, nil
 }
@@ -118,9 +147,13 @@ 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).
+// A started line with no ended line is an unfinished record still in
+// flight. NextSeq never advances past an in-flight record, so a follower
+// re-receives it until its ended line lands (idempotent by seq); a record
+// with an ended line is terminal whatever its status, an unfinished one
+// whose exit was never known included. More is false on a page whose
+// NextSeq could not move past sinceSeq, so a pager never re-requests the
+// same page in a loop.
 func Read(runDir string, sinceSeq int64, limit int) (api.CommandPage, error) {
 	lines, err := readLines(filepath.Join(runDir, FileName))
 	if err != nil {
@@ -165,12 +198,13 @@ 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
 			}
 		}
 	}
+	page.More = page.More && page.NextSeq > sinceSeq
 	return page, nil
 }
 

src/crux/runs/cmdlog_test.go

@@ -87,6 +87,10 @@ func TestLogStatusFollowsTheOutcome(t *testing.T) {
 			if rec.Status != tc.want || (tc.exit == nil) != (rec.ExitCode == nil) || (tc.exit != nil && *tc.exit != *rec.ExitCode) || rec.EndedAt == nil {
 				t.Fatalf("record = %+v want %s exit=%v", rec, tc.want, tc.exit)
 			}
+			// Every outcome is terminal, unfinished included: a follower moves on.
+			if page.NextSeq != rec.Seq || page.More {
+				t.Fatalf("page = %+v", page)
+			}
 		})
 	}
 }
@@ -106,6 +110,34 @@ func TestCommandsWithoutALogIsAnEmptyPage(t *testing.T) {
 	}
 }
 
+// A command the previous process dispatched and never ended is ended
+// unfinished at restart, so a follower's next_seq is not pinned on it
+// forever (#17410).
+func TestReconcileEndsTheCommandARunLeftInFlight(t *testing.T) {
+	f := withCmdLog(newFixture(t, blocking()))
+	rec, _, err := f.svc.Store.Create(runstate.Record{Owner: "operator"})
+	if err != nil {
+		t.Fatal(err)
+	}
+	log, err := f.svc.CmdLog(f.svc.Store.Path(rec.ID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := log.Started(cmdlog.Start{Command: []string{"make"}}); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := f.svc.Reconcile(); err != nil {
+		t.Fatal(err)
+	}
+	page, err := f.svc.Commands(rec.ID, 0, 0)
+	if err != nil || len(page.Commands) != 1 || page.NextSeq != 1 || page.More {
+		t.Fatalf("Commands = %+v, %v", page, err)
+	}
+	if got := page.Commands[0]; got.Status != api.CommandUnfinished || got.EndedAt == nil || got.ExitCode != nil || got.Reason != cmdlog.OrphanReason {
+		t.Fatalf("record = %+v", got)
+	}
+}
+
 func TestAnOpenerFailureDoesNotStopTheRun(t *testing.T) {
 	f := newFixture(t, newFakeLauncher(func(context.Context, sandbox.LaunchSpec) (sandbox.Result, error) {
 		return sandbox.Result{Stdout: "ran"}, nil

src/crux/runs/service.go

@@ -176,7 +176,8 @@ func (s *Service) Configured() bool { return s.Backend != nil }
 
 // Reconcile marks every run left accepted, running or waiting by a previous
 // process unfinished, and with it any step (S6) that run still records as
-// running. Call it once at serve start, before accepting requests.
+// running and any command (S4) its log still records as in flight. Call it
+// once at serve start, before accepting requests.
 func (s *Service) Reconcile() ([]runstate.Record, error) {
 	recs, err := s.Store.Reconcile(ReconcileReason)
 	// A run the previous process left in flight ended unobserved; its done
@@ -185,6 +186,13 @@ func (s *Service) Reconcile() ([]runstate.Record, error) {
 		if _, serr := (agentstep.Store{RunDir: s.Store.Path(rec.ID)}).Reconcile(ReconcileReason, s.Store.Now()); serr != nil && err == nil {
 			err = serr
 		}
+		// Reopening its command log ends the command it left in flight
+		// (cmdlog.Open), so a follower's next_seq is not pinned behind it
+		// forever. Like every other cmdlog failure, an unopenable log is
+		// skipped rather than holding up the restart.
+		if s.CmdLog != nil {
+			_, _ = s.CmdLog(s.Store.Path(rec.ID))
+		}
 		s.signalDone(rec.ID, api.SignalDoneData{Status: rec.Status}, rec.Reason)
 	}
 	return recs, err

src/docs/design/crux-work-services.md

@@ -296,9 +296,14 @@ started_at, ended_at?, duration_ms?, exit_code?, stdout?, stderr?}`;
 count always, and up to 4 KiB from each end of the text the runner kept.
 `status` is one of `completed` (exit 0), `failed` (non-zero exit,
 refusal or timeout), `unfinished` (the sandbox never reported an exit)
-or `cancelled`. A `started` line with no `ended` line is returned as
-`unfinished` with no `ended_at`; a status is never guessed. Paging:
-pass the previous page's `next_seq` as `since_seq` while `more` is true.
+or `cancelled`. A `started` line with no `ended` line is an in-flight
+command, returned as `unfinished` with no `ended_at`; a status is never
+guessed. Paging: pass the previous page's `next_seq` as `since_seq` while
+`more` is true. `next_seq` stops before the first record with no
+`ended_at`, so a follower re-receives an in-flight command until it ends;
+a record with `ended_at` is terminal whatever its `status`, `unfinished`
+included. `more` is never true on a page whose `next_seq` equals the
+`since_seq` it answered, so that loop always makes progress.
 Visibility follows the run — a project reads only its own runs' logs
 (403 otherwise), unknown run 404, bad `since_seq`/`limit` 400, a run that
 kept no log (or an older server) is an empty page, a corrupt line 500
@@ -321,8 +326,12 @@ and tails pass through `logscrub.ScrubString` before they are written.
 two lines per command: `{"event":"started","seq",…}` and
 `{"event":"ended","seq",…}`. Readers fold the pair; a stray `ended`
 without its `started` is dropped. Re-opening a run's log resumes `seq`
-after the highest one on disk and cannot end a command an earlier
-process left open — it stays `unfinished`, which is the truth.
+after the highest one on disk and ends every command an earlier writer
+left open with an `ended` line of status `unfinished` (reason
+`cmdlog.OrphanReason`, no exit code): a run has one live writer at a
+time, so that command's exit is never coming. `crux serve` reopens the
+log of every run it reconciles at start, so a crash mid-command leaves a
+terminal record rather than one that pins `next_seq` forever.
 
 **Under-counting, stated plainly.** Every record carries `capture`; in
 S4 the only value written is `"runner"`, which means *the command the