carlos / platform Public

host: the evicted warm database, the source journals at close, recovery replays (Task 5 fix round 1)

Ruling 1: a clean return's verified passes when the warm database and its
marker are both gone (the probe restores src_gen cold, J14); a database
without a marker is refused unless the generation is the origin; a marker
naming another generation is refused.

Ruling 2: the source's two journals (the cut's under the move id, the
recovery's under the rollback id with a dirty return's build scratch) are
marked at claimed and parked in the pass that closes the record; the local
journal sweep parks a marked journal on durable evidence: the close
journalled, no listed record carrying the move (only MoveRolledBackClose
clears a rolling_back move, and it needs recovered_at, opened_at and
target_parked_at), or the record carrying exactly those three stamps.

Ruling 3: replay tests at every recovery crash point with a fresh worker,
and a row held by another op refused at the hand-over and at the open.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude agent, operated by paul@keymail.dev d60a9f201e46caed2dc06df0addd6603f3d73abb parent 745578a
3 files changed, +442 −12
  • internal/host/movesfinal.go +6 −2
  • internal/host/movesrecover.go +181 −6
  • internal/host/movesrecover_test.go +255 −4
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index 10706c861..cdc48f53e 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -697,8 +697,12 @@ func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provisi
}
stamp, reached := j.Steps[finalStepParked]
if !reached {
- // Not a finalization's: a rollback's target journal, perhaps
- // (movesrollback.go, Task 4 fix round 1).
+ // Not a finalization's: a rollback source's recovery journal
+ // (movesrecover.go, Task 5 fix round 1), or a rollback's target
+ // journal (movesrollback.go, Task 4 fix round 1).
+ if _, ok := j.Steps[rcStepSource]; ok {
+ return w.sweepRecoveryJournal(dataDir, host, op, j, recs)
+ }
return w.sweepRollbackJournal(dataDir, host, op, j, recs)
}
if stamp != "done" {
diff --git a/internal/host/movesrecover.go b/internal/host/movesrecover.go
index ecf858090..d51924ce3 100644
--- a/internal/host/movesrecover.go
+++ b/internal/host/movesrecover.go
@@ -2,6 +2,7 @@ package host
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"log"
@@ -396,6 +397,9 @@ func (r *recoverRun) claimed(ctx context.Context) (bool, error) {
cfg := r.cfg()
host, id, rb := r.rec.Host, r.g.ID, r.rollback()
clean := !rb.Dirty
+ if err := r.markJournals(); err != nil {
+ return false, err
+ }
obs, err := r.claim(ctx)
if err != nil {
return false, err
@@ -505,17 +509,42 @@ func (r *recoverRun) verified(ctx context.Context) (bool, error) {
if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 {
return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256))
}
- gen, err := hostbox.ReadGenerationMarker(r.dbPath() + ".gen")
- if err != nil {
- return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the warm database's marker: %w", err))
- }
- if gen != srcGen {
- return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the warm database's marker names %q, the pre-move generation is %q", gen, srcGen))
+ if why, err := r.warmMismatch(srcGen); err != nil {
+ return false, err
+ } else if why != nil {
+ return r.stuck(ctx, provision.StuckBaselineMismatch, why)
}
log.Printf("moves: %s: the source still matches what it recorded when it stopped", host)
return r.phase(ctx, provision.RbSourceVerified, provision.RbSourceProbing, nil, "")
}
+// warmMismatch is the warm database's half of the clean verification (fix
+// round 1, ruling 1). A marker that is there must name src_gen. A database
+// with no marker is refused unless src_gen is the origin (""), whose
+// database never had one. A database evicted with its marker passes: the
+// probe restores the source cold from src_gen (J14), so nothing warm is
+// read. why is the mismatch; err a read that failed.
+func (r *recoverRun) warmMismatch(srcGen string) (why, err error) {
+ db := r.dbPath()
+ gen, gerr := hostbox.ReadGenerationMarker(db + ".gen")
+ if gerr != nil {
+ return fmt.Errorf("the warm database's marker: %w", gerr), nil
+ }
+ present, err := regularFile(db)
+ if err != nil {
+ return nil, err
+ }
+ switch {
+ case gen != "" && gen != srcGen:
+ return fmt.Errorf("the warm database's marker names %q, the pre-move generation is %q", gen, srcGen), nil
+ case gen == "" && present && srcGen != "":
+ return fmt.Errorf("the warm database has no marker, the pre-move generation is %q", srcGen), nil
+ case gen == "" && !present:
+ log.Printf("moves: %s: the source's warm database was evicted; the probe restores generation %q cold", r.rec.Host, srcGen)
+ }
+ return nil, nil
+}
+
// returnSeal reads the return seal the target journalled in the claim.
func returnSeal(c *hostclaim.Claim) (moves.Seal, error) {
raw, ok := c.Op.Journal[moves.JournalReturnSeal]
@@ -779,5 +808,151 @@ func (r *recoverRun) finalize(ctx context.Context) (bool, error) {
}
log.Printf("moves: %s: move %s is rolled back and closed", host, id)
nudgeSerial(ctx, r.s)
+ if err := r.fault("after-close"); err != nil {
+ return true, err
+ }
+ // The journals go in the pass that closes; a crash before this leaves
+ // them to the local journal sweep (sweepRecoveryJournal).
+ for _, op := range []string{r.rb, id} {
+ if err := parkRecoveryJournal(cfg.DataDir, host, op, true); err != nil {
+ return true, fmt.Errorf("park the source's journal of %s: %w", op, err)
+ }
+ }
return true, nil
}
+
+// The source's recovery steps in its two op journals, the cut's under the
+// move id and the recovery's under the rollback id (§7.8: "journaled ... in
+// <DataDir>/.moves/<host>/<rollback id>/"; a dirty return's install keeps
+// its build and extract scratch there too). Fix round 1, ruling 2.
+const (
+ // rcStepSource marks a journal as a source recovery's (written at
+ // claimed): sourceJournalMark, so the sweep can tell it from a cut, a
+ // landing or a target's, and knows the incarnation to park it under.
+ rcStepSource = "rb-source"
+ // rcStepClosed is "done" once the record closed.
+ rcStepClosed = "rb-source-closed"
+ // rcStepJournalPark is parkLocalArtifacts' step: the retirement the
+ // journal is parked into (no artifacts of its own).
+ rcStepJournalPark = "rb-source-journal-park"
+)
+
+// sourceJournalMark is rcStepSource's value.
+type sourceJournalMark struct {
+ Move string `json:"move"`
+ Rollback string `json:"rollback"`
+ Incarnation string `json:"incarnation"`
+}
+
+// markJournals marks the source's journals as this recovery's: the cut's,
+// when this box holds it, and the recovery's own, made here for a clean
+// return (the installer's for a dirty one). A mark already there is kept.
+func (r *recoverRun) markJournals() error {
+ dd, host := r.cfg().DataDir, r.rec.Host
+ body, err := json.Marshal(sourceJournalMark{Move: r.g.ID, Rollback: r.rb, Incarnation: r.g.Incarnation})
+ if err != nil {
+ return err
+ }
+ for _, op := range []string{r.g.ID, r.rb} {
+ if op == r.g.ID && !moveJournalPresent(dd, host, op) {
+ continue
+ }
+ j, err := moves.ReadJournal(dd, host, op)
+ if err != nil {
+ return err
+ }
+ if j.Done(rcStepSource) {
+ continue
+ }
+ j.Mark(rcStepSource, string(body))
+ if err := j.Write(dd); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// parkRecoveryJournal parks one source journal of a closed rollback: closed
+// (true when the caller has seen the close) is journalled first, then a
+// moves-owned retirement with no artifacts is made for it (its receipt names
+// the journal's op) and the journal is renamed into it. A journal that is
+// not there was parked already; one with no recovery mark is not this
+// recovery's and is left.
+func parkRecoveryJournal(dataDir, host, op string, closed bool) error {
+ if !moveJournalPresent(dataDir, host, op) {
+ return nil
+ }
+ j, err := moves.ReadJournal(dataDir, host, op)
+ if err != nil {
+ return err
+ }
+ var mark sourceJournalMark
+ raw, ok := j.Steps[rcStepSource]
+ if !ok {
+ return nil
+ }
+ if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) {
+ return fmt.Errorf("the journal of %s has an unreadable recovery mark", op)
+ }
+ write := func(step, value string) error {
+ j.Mark(step, value)
+ return j.Write(dataDir)
+ }
+ if closed && j.Steps[rcStepClosed] != "done" {
+ if err := write(rcStepClosed, "done"); err != nil {
+ return err
+ }
+ }
+ retirement, err := parkLocalArtifacts(context.Background(), j, write, localArtifactPark{DataDir: dataDir, Incarnation: mark.Incarnation,
+ Host: host, Op: op, Step: rcStepJournalPark, NewID: obligationID})
+ if err != nil {
+ return err
+ }
+ if err := parkOpJournal(dataDir, mark.Incarnation, host, op, retirement); err != nil {
+ return err
+ }
+ log.Printf("moves: %s: the source's journal of %s is parked", host, op)
+ return nil
+}
+
+// sweepRecoveryJournal is the local journal sweep's source-recovery half
+// (sweepFinalJournal, fix round 1, ruling 2): a journal the recovery marked
+// (rcStepSource) is parked once the close is shown by durable evidence:
+// - the journal records the close (rcStepClosed "done");
+// - no listed record carries the move: a rolling_back or stuck move is
+// cleared only by provision.MoveRolledBackClose, which refuses unless
+// the rollback's recovered_at, opened_at and target_parked_at are all
+// stamped (the aborts clear only an aborting move, MoveComplete only a
+// cutting one), and the mark is written while the record is rolling
+// back, so the close landed;
+// - a listed record carries the move with exactly those three stamps on
+// this rollback: the close's preconditions hold, nothing of the
+// recovery reads its journals again, and the close follows.
+//
+// A host the listing does not name, or a record that carries the move
+// without the three stamps, is left.
+func (w *movesWorker) sweepRecoveryJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error {
+ var mark sourceJournalMark
+ if err := json.Unmarshal([]byte(j.Steps[rcStepSource]), &mark); err != nil || mark.Move == "" || mark.Rollback == "" {
+ return fmt.Errorf("the journal of %s has an unreadable recovery mark", op)
+ }
+ closed := j.Steps[rcStepClosed] == "done"
+ if !closed {
+ if len(recs) == 0 {
+ w.notice(host+"#rc-sweep-"+op, "moves: %s: the source's journal of %s is a recovery's, and no listed record names the host; it is left", host, op)
+ return nil
+ }
+ for _, rec := range recs {
+ m := rec.Move
+ if m == nil || m.ID != mark.Move {
+ continue
+ }
+ rb := m.Rollback
+ if rb == nil || rb.ID != mark.Rollback || rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil {
+ return nil
+ }
+ return parkRecoveryJournal(dataDir, host, op, false)
+ }
+ }
+ return parkRecoveryJournal(dataDir, host, op, true)
+}
diff --git a/internal/host/movesrecover_test.go b/internal/host/movesrecover_test.go
index fad6afa2c..bd270da0b 100644
--- a/internal/host/movesrecover_test.go
+++ b/internal/host/movesrecover_test.go
@@ -2,7 +2,6 @@ package host
import (
"context"
- "errors"
"os"
"path/filepath"
"strings"
@@ -326,12 +325,13 @@ func TestCleanReturnVerifiesAndOpens(t *testing.T) {
if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
t.Fatalf("hand-overs: %v", h.srcHanded)
}
- // No install phase ran: no render, no return journal, no generation.
+ // No install phase ran: no render, no install steps in the recovery's
+ // journal (which holds only its mark, fix round 1), no generation.
if h.srcRends != 0 {
t.Fatalf("a clean return rendered through the installer %d times", h.srcRends)
}
- if _, err := os.Lstat(moves.JournalDir(h.sdir, p1Host, rb.ID)); !errors.Is(err, os.ErrNotExist) {
- t.Fatalf("a clean return wrote an install journal: %v", err)
+ if j, err := moves.ReadJournal(h.sdir, p1Host, rb.ID); err != nil || len(j.Steps) != 1 || !j.Done(rcStepSource) {
+ t.Fatalf("a clean return's journal: %+v %v", j, err)
}
for k := range h.f.buckets["data-ie"] {
if strings.HasPrefix(k, moves.GenerationPrefix(cutRoot, rb.ID)) {
@@ -684,3 +684,254 @@ func TestRolledBackFinalizationReleasesTheClaim(t *testing.T) {
t.Fatalf("record %+v", rec.Move)
}
}
+
+// Fix round 1, ruling 1: an evicted warm database. A clean return whose
+// warm database and marker are both gone is restored cold from src_gen by
+// its probe (J14), so verified passes; a database with no marker is refused
+// unless the generation is the origin; a marker naming another generation
+// is refused, with or without the database.
+func TestCleanReturnWarmDatabaseMarker(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ db, marker bool
+ markerBody string
+ origin, stuck bool
+ }{
+ {name: "database and marker both evicted", db: false, marker: false},
+ {name: "database without its marker", db: true, marker: false, stuck: true},
+ {name: "database without its marker, the origin generation", db: true, marker: false, origin: true},
+ {name: "a conflicting marker beside the database", db: true, marker: true, markerBody: "other\n", stuck: true},
+ {name: "a conflicting marker, the database evicted", db: false, marker: true, markerBody: "other\n", stuck: true},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.released(t)
+ db := provision.DefaultDBPath(h.sdir, p1Host)
+ if c.origin {
+ rewriteP1Claim(t, h.s, func(cl *hostclaim.Claim) { cl.Op.Journal[hostclaim.JournalSrcGen] = "" })
+ }
+ if !c.db {
+ if err := os.Remove(db); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if !c.marker {
+ if err := os.Remove(db + ".gen"); err != nil {
+ t.Fatal(err)
+ }
+ } else if err := os.WriteFile(db+".gen", []byte(c.markerBody), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.srcPass(t)
+ if !c.stuck {
+ h.opened(t)
+ return
+ }
+ m := h.read(t, p1Host).Move
+ if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified || len(h.probes) != 0 {
+ t.Fatalf("not stuck at verified: %+v %+v probes %d", m, m.Rollback, len(h.probes))
+ }
+ })
+ }
+}
+
+// parkedJournal reports whether op's journal is live under the source's
+// .moves/, and whether it was parked into a retirement's artifacts.
+func (h *recoverHarness) parkedJournal(t *testing.T, op string) (live, parked bool) {
+ t.Helper()
+ live = moveJournalPresent(h.sdir, p1Host, op)
+ suffix := filepath.Join("artifacts", ".moves", p1Host, op)
+ filepath.WalkDir(filepath.Join(h.sdir, ".parked"), func(p string, d os.DirEntry, err error) error {
+ if err == nil && d.IsDir() && strings.HasSuffix(p, suffix) {
+ if _, err := os.Stat(filepath.Join(p, "journal.json")); err == nil {
+ parked = true
+ }
+ }
+ return nil
+ })
+ return live, parked
+}
+
+// closeOut drives a recovered source and its target to the close: the
+// target cleans, the chores are acknowledged (the regional runner's, C4),
+// and the source's next pass finalizes.
+func (h *recoverHarness) closeOut(t *testing.T) {
+ t.Helper()
+ ctx := context.Background()
+ h.rollbackPass(t)
+ if rb := h.rb(t); rb.TargetParkedAt == nil {
+ t.Fatalf("the target did not park: %+v", rb)
+ }
+ for _, o := range h.claim(t).Obligations {
+ if o.Op != p1ID {
+ continue
+ }
+ obs, err := hostclaim.Read(ctx, h.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ h.srcPass(t)
+}
+
+// Fix round 1, ruling 2: the source keeps two journals through a rollback,
+// the cut's under the move id and the recovery's under the rollback id (a
+// dirty return's holds the install's build and extract scratch). The pass
+// that closes the record parks both; a crash right after the close leaves
+// them to the local journal sweep, which parks them after a restart on the
+// close's evidence (no listed record carries the move). Until the close's
+// preconditions hold, the sweep leaves them.
+func TestSourceJournalsParkedAfterTheClose(t *testing.T) {
+ for _, dirty := range []bool{false, true} {
+ name := "clean"
+ if dirty {
+ name = "dirty"
+ }
+ t.Run(name+", in the pass that closes", func(t *testing.T) {
+ h := newRecoverHarness(t, dirty)
+ rb := h.released(t)
+ h.srcPass(t)
+ h.opened(t)
+ for _, op := range []string{p1ID, rb.ID} {
+ if live, _ := h.parkedJournal(t, op); !live {
+ t.Fatalf("the journal of %s is gone before the close's preconditions", op)
+ }
+ }
+ h.srcPass(t)
+ for _, op := range []string{p1ID, rb.ID} {
+ if live, _ := h.parkedJournal(t, op); !live {
+ t.Fatalf("the sweep parked the journal of %s while the target still holds its state", op)
+ }
+ }
+ h.closeOut(t)
+ if rec := h.read(t, p1Host); rec.Move != nil {
+ t.Fatalf("not closed: %+v", rec.Move)
+ }
+ for _, op := range []string{p1ID, rb.ID} {
+ if live, parked := h.parkedJournal(t, op); live || !parked {
+ t.Fatalf("the journal of %s: live %v parked %v", op, live, parked)
+ }
+ }
+ })
+ t.Run(name+", a crash after the close", func(t *testing.T) {
+ h := newRecoverHarness(t, dirty)
+ rb := h.released(t)
+ h.srcPass(t)
+ h.opened(t)
+ if dirty {
+ if _, err := os.Stat(filepath.Join(moves.JournalDir(h.sdir, p1Host, rb.ID), "build")); err != nil {
+ t.Fatalf("the dirty return's build scratch: %v", err)
+ }
+ }
+ // The agent dies right after the close's write: nothing else of
+ // that pass runs, the journal sweep included.
+ crashed := false
+ h.rcHook = func(point string) {
+ if point == "after-close" {
+ crashed = true
+ panic(errCrash)
+ }
+ }
+ func() {
+ defer func() {
+ if r := recover(); r != nil && r != errCrash {
+ panic(r)
+ }
+ }()
+ h.closeOut(t)
+ }()
+ h.rcHook = nil
+ if !crashed {
+ t.Fatalf("never reached the close (%v)", h.rcHits)
+ }
+ if rec := h.read(t, p1Host); rec.Move != nil {
+ t.Fatalf("not closed: %+v", rec.Move)
+ }
+ for _, op := range []string{p1ID, rb.ID} {
+ if live, _ := h.parkedJournal(t, op); !live {
+ t.Fatalf("the journal of %s went with the crash", op)
+ }
+ }
+ h.sw = nil // a restarted agent
+ h.srcPass(t)
+ for _, op := range []string{p1ID, rb.ID} {
+ if live, parked := h.parkedJournal(t, op); live || !parked {
+ t.Fatalf("the journal of %s after the sweep: live %v parked %v", op, live, parked)
+ }
+ }
+ if entries, err := os.ReadDir(filepath.Join(h.sdir, ".moves", p1Host)); err == nil && len(entries) != 0 {
+ t.Fatalf("journals left: %v", entries)
+ }
+ })
+ }
+}
+
+// Fix round 1, ruling 3: every recovery step is replay-safe. A crash after
+// each of the claim's return, the guard's activation, the hold's hand-over,
+// the claim's probing and probed, and the opening of the row and of the
+// claim (and after each record step), then a restarted worker, ends opened
+// with one hold, opened once.
+func TestSourceRecoveryReplays(t *testing.T) {
+ for _, point := range []string{
+ "after-record-claimed", "after-return-claimed", "after-activate", "after-record-restoring", "after-hand",
+ "verify", "after-record-verified", "after-claim-probing", "after-record-probing", "after-claim-probed",
+ "after-record-recovered", "after-row-opened", "after-claim-opened", "after-record-opened",
+ } {
+ t.Run(point, func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ rb := h.released(t)
+ h.rcAt = point
+ h.srcPass(t)
+ if !h.rcDone {
+ t.Fatalf("never reached %s (%v)", point, h.rcHits)
+ }
+ h.sw = nil
+ h.srcPass(t)
+ h.opened(t)
+ for _, hd := range h.srcHanded {
+ if hd != "restoring:"+rb.ID+":"+p1ID {
+ t.Fatalf("hand-overs %v", h.srcHanded)
+ }
+ }
+ if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
+ t.Fatalf("the installation record: %+v %v", captured, err)
+ }
+ })
+ }
+}
+
+// Fix round 1, ruling 3: a row held by another op is never taken: the hand-
+// over at restoring and the open at recovered each refuse, and the record
+// stays where it is.
+func TestSourceRecoveryRefusesARowHeldByAnotherOp(t *testing.T) {
+ t.Run("at restoring", func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.released(t)
+ h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
+ h.srcPass(t)
+ h.sourceAt(t, provision.RbSourceRestoring)
+ if h.srcRoute.HoldOp != "op_someoneelse" || len(h.probes) != 0 {
+ t.Fatalf("the row %+v, probes %d", h.srcRoute, len(h.probes))
+ }
+ })
+ t.Run("at recovered", func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ rb := h.released(t)
+ h.rcHook = func(point string) {
+ if point == "after-record-recovered" {
+ h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
+ }
+ }
+ h.srcPass(t)
+ h.sourceAt(t, provision.RbSourceRecovered)
+ if h.srcRoute.HoldOp != "op_someoneelse" {
+ t.Fatalf("the row %+v", h.srcRoute)
+ }
+ if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbed || c.Op.Intent["rollback"] != rb.ID {
+ t.Fatalf("the claim opened over another op's row: %+v", c.Op)
+ }
+ })
+}