amadan
console: the orphan sweep writes a run's run-end row before deleting it (final review I2)
A run a box finished with no follower got its row only from a housekeeping pass over the record; a new admission that replaced it first left it to the orphan pass, which deleted it with no row, and a run whose record let it go never got one. sweepOrphanRuns now reads the result first and audits it, or writes a create-only lost result and audits that. The audited marker keeps it to one row, and a run whose marker did not land is left for a later pass.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
2 files changed,
+135
−4
internal/console/runsweep.go+58 −4internal/console/runsweep_test.go+77 −0
diff --git a/internal/console/runsweep.go b/internal/console/runsweep.go| index 55c351be2..311d93b55 100644 |
| --- a/internal/console/runsweep.go |
| +++ b/internal/console/runsweep.go |
| @@ -213,16 +213,19 @@ func (s *Server) sweepTerminalRun(ctx context.Context, c runCall, r *provision.R |
| // runs whose delete was cut short, runs replaced by a later admission, |
| // admissions that lost their CAS or whose outcome was unknown, runs of |
| // deleted hosts. With a request.json the directory goes once the request is |
| -// a day old. Without one it is a stray (request.json is written first and |
| -// deleted last but for the audited marker, so it is neither a create in |
| -// flight nor a run anyone can read): it goes once its newest object is a day |
| -// old, and stays when the store cannot say. |
| +// a day old, after its run-end row (orphanRunEnd). Without one it is a stray |
| +// (request.json is written first and deleted last but for the audited |
| +// marker, so it is neither a create in flight nor a run anyone can read): it |
| +// goes once its newest object is a day old, and stays when the store cannot |
| +// say. |
| func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named map[string]bool, now time.Time) { |
| rel := s.cfg.Releases |
| dirs, err := rel.ListDirs(ctx, runs.RunsPrefix(acct, app)) |
| if err != nil { |
| return |
| } |
| + var account *Account |
| + accountRead := false |
| for _, dir := range dirs { |
| if named[dir] { |
| continue |
| @@ -237,6 +240,16 @@ func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named ma |
| if now.Sub(req.RequestedAt) <= runs.KeepFor { |
| continue |
| } |
| + if !accountRead { |
| + if account, err = s.store.AccountBySqid(acct); err != nil { |
| + log.Printf("console: run sweep: account %s: %v", acct, err) |
| + return // no run of it is deleted before its row can be written |
| + } |
| + accountRead = true |
| + } |
| + if account != nil && !s.orphanRunEnd(ctx, account, app, req.Host, id, now) { |
| + continue |
| + } |
| case errors.Is(err, os.ErrNotExist): |
| if !s.strayIsOld(ctx, dir, now) { |
| continue |
| @@ -250,6 +263,47 @@ func (s *Server) sweepOrphanRuns(ctx context.Context, acct, app string, named ma |
| } |
| } |
| +// orphanRunEnd writes an orphan run's run-end row before its objects go |
| +// (run spec §6). A run that ended with no follower and was replaced by a |
| +// later admission before a housekeeping pass, or whose record let it go, |
| +// has none yet. With a result the row says what it says; without one the |
| +// run never came back, so it gets a lost result (create-only: a result the |
| +// box writes first stands) and the row says that. The audited marker makes |
| +// this a no-op for a run that has its row. It reports whether the run may |
| +// be deleted. |
| +func (s *Server) orphanRunEnd(ctx context.Context, acct *Account, app, host, id string, now time.Time) bool { |
| + rel := s.cfg.Releases |
| + res, err := runs.ReadResult(ctx, rel, acct.Sqid, app, id) |
| + if err != nil { |
| + log.Printf("console: run sweep: %s/%s %s: read orphan result: %v", acct.Sqid, app, id, err) |
| + return false |
| + } |
| + if res == nil { |
| + last, err := runs.LastChunkSeq(ctx, rel, acct.Sqid, app, id) |
| + if err != nil { |
| + log.Printf("console: run sweep: %s/%s %s: list orphan chunks: %v", acct.Sqid, app, id, err) |
| + return false |
| + } |
| + lost := runs.Result{Interrupted: runs.InterruptedLost, LastSeq: last, Ended: now.UTC()} |
| + if err := runs.WriteResult(ctx, rel, acct.Sqid, app, id, lost); err != nil { |
| + log.Printf("console: run sweep: %s/%s %s: orphan lost result: %v", acct.Sqid, app, id, err) |
| + return false |
| + } |
| + if res, err = runs.ReadResult(ctx, rel, acct.Sqid, app, id); err != nil || res == nil { |
| + log.Printf("console: run sweep: %s/%s %s: read orphan result back: %v", acct.Sqid, app, id, err) |
| + return false |
| + } |
| + } |
| + s.auditRunEnd(ctx, acct, app, host, id, runEndOutcome(res), res) |
| + // The marker stands only once the row has had its chance; a marker the |
| + // bucket refused leaves the run for the next pass. |
| + if _, _, err := rel.Get(ctx, runs.AuditedKey(acct.Sqid, app, id)); err != nil { |
| + log.Printf("console: run sweep: %s/%s %s: no audit marker yet, delete left for a later pass: %v", acct.Sqid, app, id, err) |
| + return false |
| + } |
| + return true |
| +} |
| + |
| // strayIsOld reports whether a run directory with no request.json has had |
| // nothing written to it for a day. A store that cannot list modification |
| // times (both real stores can) leaves strays alone. |
diff --git a/internal/console/runsweep_test.go b/internal/console/runsweep_test.go| index ecc9c2c23..8a3364420 100644 |
| --- a/internal/console/runsweep_test.go |
| +++ b/internal/console/runsweep_test.go |
| @@ -448,3 +448,80 @@ func TestRunSweepPassIsBoundedByATimeout(t *testing.T) { |
| t.Fatal("a hung store stalled the pass") |
| } |
| } |
| + |
| +// A run a box finished with no follower, replaced by a new admission before |
| +// a housekeeping pass, is an orphan when the sweep comes to it. Its row is |
| +// written before its objects go. |
| +func TestRunSweepWritesTheRowOfAReplacedRunBeforeDeletingIt(t *testing.T) { |
| + fx := runFixture(t) |
| + id := fx.finishedRun(t) |
| + fx.startRun(t) // the admission replaces the finished run on the record |
| + if ends := fx.runEnds(t, id); len(ends) != 0 { |
| + t.Fatalf("run-end rows before the sweep %q", ends) |
| + } |
| + fx.srv.runSweep(context.Background(), time.Now().Add(25*time.Hour)) |
| + if left := fx.left(id); len(left) != 0 { |
| + t.Fatalf("the orphan survived: %q", left) |
| + } |
| + if ends := fx.runEnds(t, id); len(ends) != 1 || !strings.Contains(ends[0], `"result"`) { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| +} |
| + |
| +// An orphan with no result (its record let it go before any box wrote one) |
| +// gets a lost result and its row says lost. A run that already has its row |
| +// does not get a second. |
| +func TestRunSweepWritesALostRowForAnOrphanWithNoResult(t *testing.T) { |
| + fx := runFixture(t) |
| + ctx := context.Background() |
| + old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, By: "alice@keymail.dev", RequestedAt: time.Now().Add(-25 * time.Hour)} |
| + fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old) |
| + fx.chunk(t, old.ID, 1, runs.StreamOut, "a") |
| + rowed := old |
| + rowed.ID = "run_dddddddddddddddddddddddddd" |
| + fx.putRunJSON(t, runs.RequestKey(rowed.Account, rowed.App, rowed.ID), rowed) |
| + if _, err := fx.rel.Put(ctx, runs.AuditedKey(rowed.Account, rowed.App, rowed.ID), []byte("result"), ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + fx.srv.runSweep(ctx, time.Now()) |
| + for _, id := range []string{old.ID, rowed.ID} { |
| + if left := fx.left(id); len(left) != 0 { |
| + t.Fatalf("%s survived: %q", id, left) |
| + } |
| + } |
| + ends := fx.runEnds(t, old.ID) |
| + if len(ends) != 1 || !strings.Contains(ends[0], `"lost"`) || !strings.Contains(ends[0], `"last_seq":1`) { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| + if ends := fx.runEnds(t, rowed.ID); len(ends) != 0 { |
| + t.Fatalf("a run with its row got another: %q", ends) |
| + } |
| +} |
| + |
| +// A marker the bucket refuses leaves the orphan for a later pass, so its |
| +// row is never lost to the delete. |
| +func TestRunSweepKeepsAnOrphanWhoseRowCouldNotBeMarked(t *testing.T) { |
| + fx := runFixture(t) |
| + ctx := context.Background() |
| + old := runs.Request{Schema: 1, ID: "run_cccccccccccccccccccccccccc", Account: release.OpsAccount, App: "shop", Host: runTestHost, RequestedAt: time.Now().Add(-25 * time.Hour)} |
| + fx.putRunJSON(t, runs.RequestKey(old.Account, old.App, old.ID), old) |
| + refuse := true |
| + fx.srv.cfg.Releases = &faultStore{Store: fx.rel, put: func(key string) (bool, error) { |
| + if refuse && strings.HasSuffix(key, "/audited") { |
| + return false, errors.New("access denied") |
| + } |
| + return false, nil |
| + }} |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if left := fx.left(old.ID); len(left) == 0 { |
| + t.Fatal("the orphan was deleted with no row") |
| + } |
| + refuse = false |
| + fx.srv.runSweep(ctx, time.Now()) |
| + if left := fx.left(old.ID); len(left) != 0 { |
| + t.Fatalf("the orphan survived: %q", left) |
| + } |
| + if ends := fx.runEnds(t, old.ID); len(ends) != 1 { |
| + t.Fatalf("run-end rows %q", ends) |
| + } |
| +} |