amadan

carlos / platform Public

settings-4 task 8: fix round 1: the build state is sealed; abort and return clean up

- R-SEAL-STATE: moves.Seal gains State []SealedObject (json "state",
  omitempty; at most one object, at the lane's state/ key, validated like
  the other classes; a seal without the field decodes with none).
  mirrorBuildState answers the lane copy by the version its round copied;
  the cut's final round seals it (another round when it has no usable
  version). ReturnSeal takes the state too. The installer's lane check
  covers the class, and placeBuildState copies exactly the sealed version;
  a seal with no state places nothing. A later lane version written by a
  stale round is never placed.
- The return lane's state: once the target's install has built (its
  landing journal records the install), the state beside the target's
  lease goes back; before that, the forward lane's sealed state goes back
  as given, by its exact version.
- buildStates.read: when the staging and the own prefix both hold a state,
  an own state with a strictly later At wins (a promoted staging whose
  drop failed); a tie is the staging's, as at restore.
- Abort retires the build-state lane's receipts
  (moves.BuildStateReceiptsKey) with the replica lane's.
- activator comments now say what promoteStagedReplica does: staging wins
  when present (ruling R-STAGEDREAD-2).
- Test fake: fakeMoveMirror.keepHistory keeps superseded versions readable
  by id, as a versioned bucket does (opt-in).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude db9f9ea2fae6613b39197f3cafa0b99cced1deb3 parent a1bb791
16 files changed, +434 −75
  • internal/activator/backend_exec.go +11 −9
  • internal/host/buildstate.go +24 −2
  • internal/host/buildstate_moves_test.go +159 −8
  • internal/host/movesabort.go +1 −1
  • internal/host/movesabort_test.go +1 −1
  • internal/host/movescut.go +14 −2
  • internal/host/movesinstall.go +17 −22
  • internal/host/movespass_test.go +6 −0
  • internal/host/movesphase1.go +24 −9
  • internal/host/movesphase1_harness_test.go +23 −1
  • internal/host/movesrollback.go +65 −9
  • internal/moves/returnlane.go +4 −3
  • internal/moves/returnlane_test.go +3 −3
  • internal/moves/seal.go +13 −4
  • internal/moves/seal_state_test.go +61 −0
  • internal/moves/staging.go +8 −1
diff --git a/internal/activator/backend_exec.go b/internal/activator/backend_exec.go
index c6b381240..5205acc22 100644
--- a/internal/activator/backend_exec.go
+++ b/internal/activator/backend_exec.go
@@ -687,13 +687,13 @@ func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, err
// package's activator-scoped ".activator-<db>-litestream".
PurgeLocalDBArtifacts(dbPath, false)
- // The staging fallback (issue #331): when the ordinary prefix is empty
- // and a move staged a replica here, promote it — server-side, own
- // bucket — and restore from the promoted copy. Precedence is the
- // ordinary prefix ALWAYS: a host that re-lands here after moving away
- // and back must restore its own lineage, never regress to an old
- // staged copy (the source's cut deleted the own prefix it left behind,
- // so a non-empty own prefix is by construction the newer story).
+ // The staging fallback (issue #331): when a move staged a replica
+ // here, promote it (server-side, own bucket) and restore from the
+ // promoted copy. Staging wins when present (promoteStagedReplica,
+ // review round 1): a staged copy exists only while an inbound move is
+ // landing, so it is newer than any lineage the own prefix kept from an
+ // earlier stay on this box (the legacy cut does not delete the own
+ // prefix it leaves behind), and the promotion's mirror replaces it.
var promoted map[string]lease.MirrorObject
if b.Mirror != nil && !b.Phase1 {
var err error
@@ -2551,8 +2551,10 @@ func PurgeLocalDBArtifacts(dbPath string, full bool) {
}
// promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this
-// box's OWN bucket when — and only when — the ordinary prefix is empty. A
-// no-op (nil error) when there is nothing staged.
+// box's OWN bucket whenever anything is staged: staging wins when present,
+// whatever the ordinary prefix holds (an exact mirror, so own keys the
+// staging lacks are deleted). A no-op (nil error) when there is nothing
+// staged.
func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]lease.MirrorObject, error) {
// STAGING WINS when present (review round 1). A staged replica exists
// only while an inbound move is landing or has just landed: it was
diff --git a/internal/host/buildstate.go b/internal/host/buildstate.go
index fec24a364..b56a011ff 100644
--- a/internal/host/buildstate.go
+++ b/internal/host/buildstate.go
@@ -220,10 +220,32 @@ func hasReplica(ctx context.Context, store release.Store, prefix string) (bool,
// staging lacks). So the staging is what the instance's state will be. A
// held instance must not read as new in that window, or root installs the
// head. update still writes only the own prefix: the wake promotes first.
+//
+// When both hold a state, an own state with a strictly later At wins (a
+// tie is the staging's, as at restore): a wake that promoted the
+// staging and then failed to drop it (dropStagedReplica is best-effort)
+// leaves a pre-move staged state behind its own newer record, and that
+// record is the instance's state.
func (b *buildStates) read(ctx context.Context, host string) (*release.BuildState, bool, error) {
- if st, replica, ok, err := b.readStaged(ctx, host); err != nil || ok {
- return st, replica, err
+ staged, stagedReplica, ok, err := b.readStaged(ctx, host)
+ if err != nil {
+ return nil, false, err
+ }
+ if ok && staged == nil {
+ return nil, stagedReplica, nil
+ }
+ st, replica, err := b.readOwn(ctx, host)
+ if err != nil {
+ return nil, false, err
+ }
+ if ok && (st == nil || !st.At.After(staged.At)) {
+ return staged, stagedReplica || replica, nil
}
+ return st, replica || (ok && stagedReplica), nil
+}
+
+// readOwn reads host's build state and replica at its own location.
+func (b *buildStates) readOwn(ctx context.Context, host string) (*release.BuildState, bool, error) {
store, prefix, err := b.locate(ctx, host)
if err != nil {
return nil, false, err
diff --git a/internal/host/buildstate_moves_test.go b/internal/host/buildstate_moves_test.go
index a0f343c24..8bd455b6f 100644
--- a/internal/host/buildstate_moves_test.go
+++ b/internal/host/buildstate_moves_test.go
@@ -221,9 +221,16 @@ func TestTheCutsFinalRoundCopiesTheBuildState(t *testing.T) {
if m := h.move(t); m.State != provision.MoveLanding {
t.Fatalf("cut did not finish: %+v", m)
}
- if got := h.f.buckets["data-uk"][moves.BuildStateKey(p1Host, p1ID)].body; got != p1StateBody("v7") {
+ key := moves.BuildStateKey(p1Host, p1ID)
+ if got := h.f.buckets["data-uk"][key].body; got != p1StateBody("v7") {
t.Fatalf("the cut staged %q, want the final state", got)
}
+ // The seal names that copy by its version (ruling R-SEAL-STATE).
+ seal := h.seal(t)
+ if len(seal.State) != 1 || seal.State[0].Key != key || seal.State[0].VersionID != h.f.versionOf("data-uk", key) ||
+ !lease.UsableVersionID(seal.State[0].VersionID) || seal.State[0].Size != int64(len(p1StateBody("v7"))) {
+ t.Fatalf("seal state %+v, lane version %s", seal.State, h.f.versionOf("data-uk", key))
+ }
}
// The landing places the staged build state beside the location's lease,
@@ -264,10 +271,43 @@ func TestTheLandingPlacesTheBuildStateBeforeTheProbe(t *testing.T) {
}
}
-// A landing whose source had no build state stages none and places none.
+// A landing places exactly the sealed version: a request a stale round
+// signed may land a newer version in the lane until the seal's bound, and
+// that version is never placed (ruling R-SEAL-STATE).
+func TestTheLandingPlacesTheSealedVersionNotALaterOne(t *testing.T) {
+ h := newLandingHarness(t)
+ h.f.keepHistory = true
+ storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
+ h.flip(t)
+ key := moves.BuildStateKey(p1Host, p1ID)
+ sealed := h.seal(t).State
+ if len(sealed) != 1 || sealed[0].VersionID != h.f.versionOf("data-uk", key) {
+ t.Fatalf("seal state %+v", sealed)
+ }
+ if _, err := h.f.PutVersionedAt(context.Background(), h.clock.Now(), "data-uk", key, []byte(p1StateBody("v5"))); err != nil {
+ t.Fatal(err)
+ }
+ if h.f.versionOf("data-uk", key) == sealed[0].VersionID {
+ t.Fatal("the late write made no new version")
+ }
+ h.land(t)
+ h.installed(t)
+ if got := h.f.buckets["data-uk"][cutRoot+"/build-state.json"].body; got != p1StateBody("v6") {
+ t.Fatalf("the location holds %q, want the sealed v6", got)
+ }
+}
+
+// A landing whose source had no build state seals none and places none,
+// even if a lane object appears after the seal.
func TestALandingWithNoBuildStatePlacesNone(t *testing.T) {
h := newLandingHarness(t)
h.flip(t)
+ if st := h.seal(t).State; len(st) != 0 {
+ t.Fatalf("seal state %+v, want none", st)
+ }
+ if _, err := h.f.PutVersionedAt(context.Background(), h.clock.Now(), "data-uk", moves.BuildStateKey(p1Host, p1ID), []byte(p1StateBody("v5"))); err != nil {
+ t.Fatal(err)
+ }
h.land(t)
h.installed(t)
if _, ok := h.f.buckets["data-uk"][cutRoot+"/build-state.json"]; ok {
@@ -275,20 +315,71 @@ func TestALandingWithNoBuildStatePlacesNone(t *testing.T) {
}
}
+// returnSealState is the return seal's state class, checked against the
+// lane object it names.
+func returnSealState(t *testing.T, h *rollbackHarness, rbID string) (moves.SealedObject, string) {
+ t.Helper()
+ seal, err := moves.DecodeSeal(h.claim(t).Op.Journal[moves.JournalReturnSeal])
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := moves.BuildStateKey(p1Host, rbID)
+ if len(seal.State) != 1 || seal.State[0].Key != key || seal.State[0].VersionID != h.f.versionOf("data-ie", key) {
+ t.Fatalf("return seal state %+v, lane version %s", seal.State, h.f.versionOf("data-ie", key))
+ }
+ return seal.State[0], h.f.buckets["data-ie"][key].body
+}
+
// The rollback's return lane carries the target's build state back, under
-// the lane's state/ prefix in the source's bucket.
+// the lane's state/ prefix in the source's bucket, sealed by version. Once
+// the target's install has run, the state beside its lease is the one that
+// goes back: here the probe recorded v8 over the moved v6.
func TestTheReturnLaneCarriesTheBuildState(t *testing.T) {
- h := newRollbackHarness(t)
+ h := &rollbackHarness{landingHarness: newLandingHarness(t)}
+ storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
+ h.flip(t)
+ h.land(t)
+ h.installed(t)
+ h.step(t, provision.LandRestored, provision.LandProbing)
+ h.tpos = h.n
storeState(h.phase1Harness, "data-uk", cutRoot+"/build-state.json", "v8")
h.uncover(t)
h.rollBack(t)
h.rollbackPass(t)
rb := h.rb(t)
- if rb.TargetPhase != provision.RbTargetReturned {
+ if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
+ t.Fatalf("rollback %+v", rb)
+ }
+ if _, body := returnSealState(t, h, rb.ID); body != p1StateBody("v8") {
+ t.Fatalf("the return lane holds %q, want the target's state", body)
+ }
+}
+
+// A return before the target's install built anything: the target's
+// location holds no state of this move, so the forward lane's sealed state
+// goes back as given (by its exact version, not the lane's latest).
+func TestAReturnBeforeTheInstallCarriesTheSealedState(t *testing.T) {
+ h := &rollbackHarness{landingHarness: newLandingHarness(t)}
+ h.f.keepHistory = true
+ storeState(h.phase1Harness, "data-ie", cutRoot+"/build-state.json", "v6")
+ h.flip(t)
+ if _, err := h.f.PutVersionedAt(context.Background(), h.clock.Now(), "data-uk", moves.BuildStateKey(p1Host, p1ID), []byte(p1StateBody("v5"))); err != nil {
+ t.Fatal(err)
+ }
+ h.landAt = "after-record-" + provision.LandBuilding
+ h.crashLand(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
+ t.Fatalf("landing phase %s", m.LandingPhase)
+ }
+ h.uncover(t)
+ h.rollBack(t)
+ h.rollbackPass(t)
+ rb := h.rb(t)
+ if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
t.Fatalf("rollback %+v", rb)
}
- if got := h.f.buckets["data-ie"][moves.BuildStateKey(p1Host, rb.ID)].body; got != p1StateBody("v8") {
- t.Fatalf("the return lane holds %q, want the target's state", got)
+ if _, body := returnSealState(t, h, rb.ID); body != p1StateBody("v6") {
+ t.Fatalf("the return lane holds %q, want the forward lane's sealed v6", body)
}
}
@@ -296,7 +387,18 @@ func TestTheReturnLaneCarriesTheBuildState(t *testing.T) {
// source's location lease, as the landing does.
func TestTheReturnInstallPlacesTheBuildState(t *testing.T) {
h := newReturnHarness(t, true)
- if _, err := h.vs.PutIfAbsent(context.Background(), "data-ie", moves.BuildStateKey(p1Host, p1RB), []byte(p1StateBody("v8"))); err != nil {
+ h.f.keepHistory = true
+ key := moves.BuildStateKey(p1Host, p1RB)
+ ver, err := h.vs.PutIfAbsent(context.Background(), "data-ie", key, []byte(p1StateBody("v8")))
+ if err != nil {
+ t.Fatal(err)
+ }
+ h.seal.State = []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(p1StateBody("v8")))}}
+ if err := h.seal.Validate(); err != nil {
+ t.Fatal(err)
+ }
+ // A later lane version is not the sealed one and is not placed.
+ if _, err := h.f.PutVersionedAt(context.Background(), h.clock.Now(), "data-ie", key, []byte(p1StateBody("v5"))); err != nil {
t.Fatal(err)
}
pinned := h.read(t, p1Host).Move.Inputs
@@ -308,3 +410,52 @@ func TestTheReturnInstallPlacesTheBuildState(t *testing.T) {
t.Fatalf("the source's location holds %q, want the returned state", got)
}
}
+
+// A wake that promoted the staging and then failed to drop it
+// (dropStagedReplica is best-effort) leaves the pre-move staged state
+// behind its own newer record: when both hold a state, the later At wins.
+func TestALaterOwnStateOutranksALeftoverStaging(t *testing.T) {
+ b, s, _ := testBuildStates(t)
+ b.staged = legacyStaged
+ ctx := context.Background()
+ putObject(t, s, "moves/"+stagedStateHost+"/generations/0000/x.ltx", "x")
+ putObject(t, s, "moves/"+stagedStateHost+"/build-state.json", stagedStateBody(stagedStateHost, "v1"))
+ own := "inst/" + stagedStateHost
+ putObject(t, s, own+"/generations/0000/x.ltx", "x")
+ putObject(t, s, own+"/build-state.json",
+ `{"schema":1,"host":"`+stagedStateHost+`","last_executed":"v2","at":"2026-10-08T12:00:00Z"}`)
+ if st, replica, err := b.read(ctx, stagedStateHost); err != nil || !replica || st == nil || st.LastExecuted != "v2" {
+ t.Fatalf("read = %+v, %v, %v; want the own, later v2", st, replica, err)
+ }
+ // An older own state (an earlier stay's) loses to the staging.
+ _, etag, err := s.Get(ctx, own+"/build-state.json")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := s.Put(ctx, own+"/build-state.json",
+ []byte(`{"schema":1,"host":"`+stagedStateHost+`","last_executed":"v0","at":"2026-10-01T12:00:00Z"}`), etag); err != nil {
+ t.Fatal(err)
+ }
+ if st, _, err := b.read(ctx, stagedStateHost); err != nil || st == nil || st.LastExecuted != "v1" {
+ t.Fatalf("read = %+v, %v; want the staged v1", st, err)
+ }
+}
+
+// An abort retires the build-state lane's receipts with the replica lane's.
+func TestAbortRetiresTheBuildStateReceipts(t *testing.T) {
+ h := newAbortHarness(t)
+ loc := strings.TrimSuffix(h.claim(t).Holder.Location, "/")
+ storeState(h.phase1Harness, "data-ie", loc+"/build-state.json", "v6")
+ h.pass(t)
+ key := moves.BuildStateReceiptsKey(p1Host, p1ID)
+ if _, ok := h.f.buckets["data-ie"][key]; !ok {
+ t.Fatalf("the state lane wrote no receipts at %s", key)
+ }
+ h.abort(t)
+ h.pass(t)
+ h.waiting(t)
+ h.ownStateGone(t)
+ if _, ok := h.f.buckets["data-ie"][key]; ok {
+ t.Fatalf("%s survived the abort", key)
+ }
+}
diff --git a/internal/host/movesabort.go b/internal/host/movesabort.go
index 23a9a25f9..ef6bd9321 100644
--- a/internal/host/movesabort.go
+++ b/internal/host/movesabort.go
@@ -288,7 +288,7 @@ func (w *movesWorker) retireOwnMoveState(ctx context.Context, rec provision.Reco
if err != nil {
return err
}
- keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)}
+ keys := []string{moves.ReceiptsKey(rec.Host, rec.Move.ID), moves.BuildStateReceiptsKey(rec.Host, rec.Move.ID), moves.LedgerKey(rec.Host, rec.Move.ID)}
sort.Strings(keys)
for _, k := range keys {
if err := own.Delete(ctx, cfg.OwnBucket, k); err != nil && !errors.Is(err, lease.ErrMirrorNotFound) {
diff --git a/internal/host/movesabort_test.go b/internal/host/movesabort_test.go
index 871feb728..91b51bd7f 100644
--- a/internal/host/movesabort_test.go
+++ b/internal/host/movesabort_test.go
@@ -175,7 +175,7 @@ func (h *abortHarness) closed(t *testing.T) {
func (h *abortHarness) ownStateGone(t *testing.T) {
t.Helper()
for k := range h.f.buckets["data-ie"] {
- if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) || k == moves.ReceiptsKey(p1Host, p1ID) || k == moves.LedgerKey(p1Host, p1ID) {
+ if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) || k == moves.ReceiptsKey(p1Host, p1ID) || k == moves.BuildStateReceiptsKey(p1Host, p1ID) || k == moves.LedgerKey(p1Host, p1ID) {
t.Fatalf("%s survived the abort", k)
}
}
diff --git a/internal/host/movescut.go b/internal/host/movescut.go
index 99dab6d3e..df432dd33 100644
--- a/internal/host/movescut.go
+++ b/internal/host/movescut.go
@@ -916,9 +916,12 @@ func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report
// The build state of the stopped instance, so the lane carries the
// final state and not a round-old one (settings spec rev 11 §4.3a). A
// state that changed under the copy is not final: the pass retries.
- if st, err := mirrorBuildState(ctx, c.gm, cfg.OwnBucket, holder.Location, dst, host, id); err != nil {
+ // The seal names the version this round copied (ruling R-SEAL-STATE).
+ stRound, stateObj, err := mirrorBuildState(ctx, c.gm, cfg.OwnBucket, holder.Location, dst, host, id)
+ if err != nil {
return false, fmt.Errorf("final round: build state lane: %w", err)
- } else if st.Deferred > 0 {
+ }
+ if stRound.Deferred > 0 {
return false, fmt.Errorf("final round: the build state changed during the copy; retrying the cut")
}
if err := c.fault("after-final-round"); err != nil {
@@ -962,6 +965,14 @@ func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report
}
replica = append(replica, moves.SealedObject{Key: dstPrefix + rel, VersionID: ver, Size: o.Size})
}
+ var state []moves.SealedObject
+ if stateObj != nil {
+ if !lease.UsableVersionID(stateObj.VersionID) {
+ c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for the build state; another round", host)
+ return false, nil
+ }
+ state = []moves.SealedObject{*stateObj}
+ }
// Coverage starts false (ruling 3): a database still here is asked for
// its position against N; one that is gone is covered only if the
// checkpoint journalled that it was, at this N (giveUpExecution).
@@ -990,6 +1001,7 @@ func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report
now := cfg.Now().UTC()
seal := moves.Seal{V: moves.SealVersion, N: report.FinalTXID, Replica: replica,
Timers: []moves.SealedObject{{Key: stagedTimers, VersionID: tver, Size: int64(len(timersBody))}},
+ State: state,
SrcCovered: covered, FromDB: fromDB, SealedAt: now}
seal.LeaseUntil = c.bound(now)
if err := seal.Validate(); err != nil {
diff --git a/internal/host/movesinstall.go b/internal/host/movesinstall.go
index 941a71c0e..a75bab95d 100644
--- a/internal/host/movesinstall.go
+++ b/internal/host/movesinstall.go
@@ -272,7 +272,7 @@ func (ir *installRun) seal(c *hostclaim.Claim) (moves.Seal, error) {
if err != nil {
return seal, err
}
- for _, class := range [][]moves.SealedObject{seal.Replica, seal.Files, seal.Timers} {
+ for _, class := range [][]moves.SealedObject{seal.Replica, seal.Files, seal.Timers, seal.State} {
for _, o := range class {
if !strings.HasPrefix(o.Key, ir.src.Lane) {
return seal, fmt.Errorf("%w: %s is not in the %s's lane %s", moves.ErrBadSeal, o.Key, ir.src.Name, ir.src.Lane)
@@ -577,7 +577,7 @@ func (ir *installRun) building(ctx context.Context) (bool, error) {
if err != nil {
return false, fmt.Errorf("build the generation: %w", err)
}
- if err := ir.placeBuildState(leaseContext(ctx, held), leasedStore{VersionedStore: vs, check: held.Check}, holder); err != nil {
+ if err := ir.placeBuildState(leaseContext(ctx, held), leasedStore{VersionedStore: vs, check: held.Check}, seal, holder); err != nil {
return false, err
}
if err := held.Check(ctx); err != nil {
@@ -603,33 +603,28 @@ func (ir *installRun) building(ctx context.Context) (bool, error) {
return ir.advance(ctx, phase, next, installStamp{})
}
-// placeBuildState copies the lane's build state (moves.BuildStateKey, the
-// only object under the lane's state/ prefix) beside the location's lease,
+// placeBuildState copies the sealed build state (the seal's State class,
+// the lane's state/ key) beside the location's lease,
// holder.Location/build-state.json in this box's own bucket (settings spec
// rev 11 §4.3a, ruling R-LANDING). building calls it under the location's
// lease, so it lands before the probe and the first wake: their record
// merges onto the moved state, and a held instance never reads as new
-// here. The object is copied by its current version, as the source's
-// rounds left it; the cut's final round wrote the last one. A lane with no
-// build state (the source had none) places nothing. A replay copies again.
-func (ir *installRun) placeBuildState(ctx context.Context, vs lease.VersionedStore, holder hostclaim.Holder) error {
- bucket := ir.cfg().OwnBucket
- src := ir.src.Lane + "state/" + buildStateName
- objs, err := vs.ListVersions(ctx, bucket, src)
- if err != nil {
- return fmt.Errorf("list the lane's build state: %w", err)
- }
- var cur *lease.ObjectVersion
- for i := range objs {
- if o := &objs[i]; o.Key == src && o.IsLatest {
- cur = o
- }
- }
- if cur == nil || cur.DeleteMarker {
+// here. It copies exactly the sealed version (ruling R-SEAL-STATE): a
+// request a stale round signed can land a newer version in the lane until
+// the seal's bound, and that version is never placed. A seal with no state
+// (the source had none, or a seal from before the class existed) places
+// nothing. A replay copies again.
+func (ir *installRun) placeBuildState(ctx context.Context, vs lease.VersionedStore, seal moves.Seal, holder hostclaim.Holder) error {
+ if len(seal.State) == 0 {
return nil
}
+ o := seal.State[0]
+ if o.Key != ir.src.Lane+moves.BuildStateSuffix {
+ return fmt.Errorf("%w: the sealed build state %s is not the %s's lane key", moves.ErrBadSeal, o.Key, ir.src.Name)
+ }
+ bucket := ir.cfg().OwnBucket
dst := release.BuildStateKey(strings.TrimSuffix(holder.Location, "/"))
- if _, _, err := vs.CopyVersion(ctx, bucket, src, cur.VersionID, bucket, dst, false); err != nil {
+ if _, _, err := vs.CopyVersion(ctx, bucket, o.Key, o.VersionID, bucket, dst, false); err != nil {
return fmt.Errorf("place the build state at %s: %w", dst, err)
}
log.Printf("moves: %s: the %s placed the moved build state at %s", ir.rec.Host, ir.src.Name, dst)
diff --git a/internal/host/movespass_test.go b/internal/host/movespass_test.go
index e8b6ab3ac..9e6257db9 100644
--- a/internal/host/movespass_test.go
+++ b/internal/host/movespass_test.go
@@ -37,6 +37,12 @@ type fakeMoveMirror struct {
failGet func(bucket, key string) error // a plain read that fails (the ledger, movescut_test.go)
versions int
vers map[string]string
+ // keepHistory makes the fake keep every version a later write
+ // superseded, readable and copyable by its id, as a versioned bucket
+ // does (off by default: the fake's older tests read the current
+ // version only).
+ keepHistory bool
+ history map[string]fakeMoveObject
}
type fakeMoveObject struct {
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 09284c180..6434338e2 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -390,7 +390,7 @@ func (w *movesWorker) sourceRound(ctx context.Context, s release.Store, rec prov
return nil, nil, fmt.Errorf("replica lane: %w", err)
}
// The build state travels with the replica, in every round.
- if _, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, obs.Claim.Holder.Location, tgt.DataBucket, rec.Host, id); err != nil {
+ if _, _, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, obs.Claim.Holder.Location, tgt.DataBucket, rec.Host, id); err != nil {
return nil, nil, fmt.Errorf("build state lane: %w", err)
}
if round.BytesCopied >= crossMinBytes {
@@ -440,16 +440,31 @@ const buildStateName = "build-state.json"
// srcBucket, mirrored to opID's lane state/ prefix in dstBucket, and
// nothing else. The destination must know which build last ran against
// this data. A source with no build state leaves none in the lane (the
-// mirror deletes a stale copy). The preparing rounds, the cut's final round
-// and a rollback's return lane all call it, so the lane holds the state of
-// the round the replica is from. Receipts live in srcBucket beside the
-// replica lane's, under their own key.
-func mirrorBuildState(ctx context.Context, m lease.MirrorStore, srcBucket, location, dstBucket, host, opID string) (lease.MirrorRound, error) {
+// mirror deletes a stale copy) and answers a nil object. The preparing
+// rounds, the cut's final round and a rollback's return lane all call it.
+// Receipts live in srcBucket beside the replica lane's, under their own
+// key. The object it answers is the lane copy by the version this round
+// copied or vouched for (round.Versions, a versioned mirror only), for the
+// seal: a later write by a stale round is never what a seal names (ruling
+// R-SEAL-STATE). Its VersionID is "" on a mirror that records none.
+func mirrorBuildState(ctx context.Context, m lease.MirrorStore, srcBucket, location, dstBucket, host, opID string) (lease.MirrorRound, *moves.SealedObject, error) {
src := strings.TrimSuffix(location, "/") + "/"
- dst := strings.TrimSuffix(moves.BuildStateKey(host, opID), buildStateName)
+ dst := moves.BuildStateKey(host, opID)
keep := func(rel string) bool { return rel == buildStateName }
- return lease.MirrorPrefixKeep(ctx, m, srcBucket, src, dstBucket, dst,
- lease.Receipts{Bucket: srcBucket, Key: moves.ReceiptsKey(host, opID+"-state")}, keep)
+ round, err := lease.MirrorPrefixKeep(ctx, m, srcBucket, src, dstBucket, strings.TrimSuffix(dst, buildStateName),
+ lease.Receipts{Bucket: srcBucket, Key: moves.BuildStateReceiptsKey(host, opID)}, keep)
+ if err != nil {
+ return round, nil, err
+ }
+ listed, err := m.List(ctx, srcBucket, src+buildStateName)
+ if err != nil {
+ return round, nil, fmt.Errorf("list the build state: %w", err)
+ }
+ o, ok := listed[src+buildStateName]
+ if !ok {
+ return round, nil, nil
+ }
+ return round, &moves.SealedObject{Key: dst, VersionID: round.Versions[buildStateName], Size: o.Size}, nil
}
// oneBoxRefusal is the one-box rule (D15), re-checked from the edges
diff --git a/internal/host/movesphase1_harness_test.go b/internal/host/movesphase1_harness_test.go
index 0aea92faf..a817eee7e 100644
--- a/internal/host/movesphase1_harness_test.go
+++ b/internal/host/movesphase1_harness_test.go
@@ -92,6 +92,21 @@ func (f *fakeMoveMirror) setVersion(bucket, key, ver string) {
f.vers = map[string]string{}
}
f.vers[bucket+"/"+key] = ver
+ if f.keepHistory {
+ if f.history == nil {
+ f.history = map[string]fakeMoveObject{}
+ }
+ f.history[bucket+"/"+key+"@"+ver] = f.buckets[bucket][key]
+ }
+}
+
+// older is a non-current version of bucket/key the fake kept (keepHistory),
+// as a versioned bucket keeps every version a later write superseded.
+func (f *fakeMoveMirror) older(bucket, key, ver string) (fakeMoveObject, bool) {
+ f.mu.Lock()
+ defer f.mu.Unlock()
+ o, ok := f.history[bucket+"/"+key+"@"+ver]
+ return o, ok && f.keepHistory
}
func (f *fakeMoveMirror) versionOf(bucket, key string) string {
@@ -191,6 +206,9 @@ func (v *fakeVersioned) GetVersion(_ context.Context, bucket, key, versionID str
}
o, ok := v.f.buckets[bucket][key]
if !ok || versionID == "" || v.versionID(bucket, key) != versionID {
+ if old, kept := v.f.older(bucket, key, versionID); kept && versionID != "" {
+ return []byte(old.body), nil
+ }
return nil, lease.ErrMirrorNotFound
}
return []byte(o.body), nil
@@ -213,7 +231,11 @@ func (v *fakeVersioned) CopyVersion(_ context.Context, srcBucket, srcKey, srcVer
}
o, ok := v.f.buckets[srcBucket][srcKey]
if !ok || srcVersionID == "" || v.versionID(srcBucket, srcKey) != srcVersionID {
- return "", "", lease.ErrMirrorNotFound
+ old, kept := v.f.older(srcBucket, srcKey, srcVersionID)
+ if !kept || srcVersionID == "" {
+ return "", "", lease.ErrMirrorNotFound
+ }
+ o = old
}
if _, exists := v.f.buckets[dstBucket][dstKey]; exists && createOnly {
return "", "", lease.ErrDestinationExists
diff --git a/internal/host/movesrollback.go b/internal/host/movesrollback.go
index 36b3e41fa..db473e3a0 100644
--- a/internal/host/movesrollback.go
+++ b/internal/host/movesrollback.go
@@ -625,7 +625,10 @@ func (r *rollbackRun) toCleaning(ctx context.Context, from string) (bool, error)
type returnedLane struct {
n uint64
replica, files, timers []moves.SealedObject
- baseline provision.MoveSource
+ // state is the returned build state (settings spec rev 11 §4.3a), nil
+ // when there is none.
+ state []moves.SealedObject
+ baseline provision.MoveSource
}
// returning (§7.8 target returning): the return lane under the source's
@@ -697,6 +700,9 @@ func (r *rollbackRun) returning(ctx context.Context) (bool, error) {
} else {
lane, err = r.snapshotStopped(ctx, gm, report)
}
+ if err == nil {
+ lane.state, err = r.returnState(ctx, gm, obs.Claim.Holder, staged)
+ }
var refused *returnRefusal
if errors.As(err, &refused) {
return r.stuck(ctx, refused)
@@ -704,20 +710,13 @@ func (r *rollbackRun) returning(ctx context.Context) (bool, error) {
if err != nil {
return false, err
}
- // The target's build state goes back with its data (settings spec rev 11
- // §4.3a): the source's install places it beside its own lease.
- if h := obs.Claim.Holder; h == nil {
- return false, fmt.Errorf("move %s: the claim has no holder to return the build state from", id)
- } else if _, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, h.Location, gm.dst, host, r.rb); err != nil {
- return false, fmt.Errorf("return the build state: %w", err)
- }
// The bound is read after the last lane write: every request the guard
// let a writer sign expires before it.
bound, err := held.SafeUntil()
if err != nil {
return false, fmt.Errorf("the staging lease's bound: %w", err)
}
- seal := moves.ReturnSeal(lane.n, lane.replica, lane.files, lane.timers, r.now(), bound)
+ seal := moves.ReturnSeal(lane.n, lane.replica, lane.files, lane.timers, lane.state, r.now(), bound)
if err := seal.Validate(); err != nil {
return false, err
}
@@ -846,6 +845,63 @@ func (r *rollbackRun) copyAsGiven(ctx context.Context, gm *guardedMirror, staged
return out, nil
}
+// returnState is a dirty return's build state (settings spec rev 11
+// §4.3a), through the return lane's state/ key and sealed by the version
+// written there. Once the target's install has built (its landing journal
+// records the install), the state beside the target's lease at the holder's
+// location is the instance's: the moved state the install placed, with
+// whatever the probe and the target's wakes recorded over it. Before that
+// the target's location holds no state of this move (a return before the
+// building step), so the forward lane's sealed state goes back as given,
+// read by its exact version. Neither holds one: none goes back.
+func (r *rollbackRun) returnState(ctx context.Context, gm *guardedMirror, h *hostclaim.Holder, staged moves.Seal) ([]moves.SealedObject, error) {
+ cfg := r.cfg()
+ host := r.rec.Host
+ j, err := r.journal()
+ if err != nil {
+ return nil, err
+ }
+ if !j.Done(landStepInstall) {
+ if len(staged.State) == 0 {
+ // A lane object an earlier attempt left is not sealed, so the
+ // source's install never places it.
+ return nil, nil
+ }
+ o := staged.State[0]
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return nil, err
+ }
+ body, err := vs.GetVersion(ctx, cfg.OwnBucket, o.Key, o.VersionID)
+ if err != nil {
+ return nil, fmt.Errorf("read the sealed build state %s: %w", o.Key, err)
+ }
+ key := moves.BuildStateKey(host, r.rb)
+ ver, err := gm.PutVersioned(ctx, gm.dst, key, body)
+ if err != nil {
+ return nil, fmt.Errorf("return the build state: %w", err)
+ }
+ if !lease.UsableVersionID(ver) {
+ return nil, fmt.Errorf("the returned build state has no usable version id %q", ver)
+ }
+ return []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(body))}}, nil
+ }
+ if h == nil {
+ return nil, fmt.Errorf("move %s: the claim has no holder to return the build state from", r.g.ID)
+ }
+ _, o, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, h.Location, gm.dst, host, r.rb)
+ if err != nil {
+ return nil, fmt.Errorf("return the build state: %w", err)
+ }
+ if o == nil {
+ return nil, nil
+ }
+ if !lease.UsableVersionID(o.VersionID) {
+ return nil, fmt.Errorf("the returned build state has no usable version id %q", o.VersionID)
+ }
+ return []moves.SealedObject{*o}, nil
+}
+
// downloadExact reads one exact version of size bytes into path.
func downloadExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64, path string) error {
rc, err := vs.OpenVersion(ctx, bucket, key, version)
diff --git a/internal/moves/returnlane.go b/internal/moves/returnlane.go
index d98580d12..64b46fbf2 100644
--- a/internal/moves/returnlane.go
+++ b/internal/moves/returnlane.go
@@ -42,9 +42,10 @@ func ReturnLane(host, rbID string) string { return StagingPrefix(host, rbID) }
// or every level file of the staged seal copied as given (J22). The lane is
// built from the target's stopped state and covers it exactly, so FromDB
// and SrcCovered are both true. leaseUntil is the staging lease's bound
-// when the last lane object was written. The caller's slices are copied.
-func ReturnSeal(n uint64, replicas []SealedObject, files, timers []SealedObject, now time.Time, leaseUntil time.Time) Seal {
- return Seal{V: SealVersion, N: n, Replica: sortObjects(replicas), Files: sortObjects(files), Timers: sortObjects(timers),
+// when the last lane object was written. state is the lane's build state,
+// nil when the target had none. The caller's slices are copied.
+func ReturnSeal(n uint64, replicas []SealedObject, files, timers, state []SealedObject, now time.Time, leaseUntil time.Time) Seal {
+ return Seal{V: SealVersion, N: n, Replica: sortObjects(replicas), Files: sortObjects(files), Timers: sortObjects(timers), State: sortObjects(state),
SrcCovered: true, FromDB: true, SealedAt: now.UTC(), LeaseUntil: leaseUntil.UTC()}
}
diff --git a/internal/moves/returnlane_test.go b/internal/moves/returnlane_test.go
index eed5778b0..7b5c8b1f1 100644
--- a/internal/moves/returnlane_test.go
+++ b/internal/moves/returnlane_test.go
@@ -41,7 +41,7 @@ func TestReturnSealOfTheStoppedSnapshot(t *testing.T) {
until := at.Add(40 * time.Minute)
snap := SealedObject{Key: ReturnLane(rlHost, rlRB) + "replica/0009/" + ltx.FormatFilename(1, 9), VersionID: "v1", Size: ltx.HeaderSize + 4096}
files, timers := returnFiles()
- s := ReturnSeal(9, []SealedObject{snap}, files, timers, at, until)
+ s := ReturnSeal(9, []SealedObject{snap}, files, timers, nil, at, until)
if s.V != SealVersion || s.N != 9 || !s.FromDB || !s.SrcCovered || !s.SealedAt.Equal(at) || !s.Bound().Equal(until) {
t.Fatalf("seal %+v", s)
}
@@ -70,7 +70,7 @@ func TestReturnSealAsGivenTakesManyReplicaObjects(t *testing.T) {
{Key: lane + "replica/0009/" + ltx.FormatFilename(1, 2), VersionID: "v5", Size: ltx.HeaderSize + 10},
}
files, timers := returnFiles()
- s := ReturnSeal(3, replica, files, timers, at, at.Add(time.Hour))
+ s := ReturnSeal(3, replica, files, timers, nil, at, at.Add(time.Hour))
if err := s.Validate(); err != nil {
t.Fatal(err)
}
@@ -82,7 +82,7 @@ func TestReturnSealAsGivenTakesManyReplicaObjects(t *testing.T) {
t.Fatalf("ReturnSeal sorted the caller's slice: %+v", replica)
}
// A bound before the seal pins nothing: Validate refuses it.
- if err := ReturnSeal(3, replica, files, timers, at, at.Add(-time.Second)).Validate(); !errors.Is(err, ErrBadSeal) {
+ if err := ReturnSeal(3, replica, files, timers, nil, at, at.Add(-time.Second)).Validate(); !errors.Is(err, ErrBadSeal) {
t.Fatalf("a bound before the seal: %v", err)
}
}
diff --git a/internal/moves/seal.go b/internal/moves/seal.go
index 6e4d1abc4..c0446b24c 100644
--- a/internal/moves/seal.go
+++ b/internal/moves/seal.go
@@ -7,6 +7,7 @@ import (
"fmt"
"io"
"sort"
+ "strings"
"time"
"github.com/carlosframework/platform/internal/lease"
@@ -14,7 +15,7 @@ import (
)
// The seal (§7.3 "seal by version", §7.4 staged): the exact versions of the
-// staging lane's replica/, files/ and timers/ objects the cut wrote, the
+// staging lane's replica/, files/, timers/ and state/ objects the cut wrote, the
// required final TXID N, and the two facts a recovery reads, in the claim's
// op journal under hostclaim.JournalSeal. Landing copies these versions and
// nothing else, so a stale round's later write (a newer version, or a
@@ -30,6 +31,11 @@ type Seal struct {
Replica []SealedObject `json:"replica"`
Files []SealedObject `json:"files,omitempty"`
Timers []SealedObject `json:"timers,omitempty"`
+ // State is the lane's copy of the instance's build state (settings spec
+ // rev 11 §4.3a), at most one object at the lane's state/ key: the
+ // version the round that wrote this seal copied. Absent when the source
+ // had none; a seal written before this field existed decodes with none.
+ State []SealedObject `json:"state,omitempty"`
// SrcCovered is whether the source's own replica of holder.generation
// reached the stopped database's position with no pending write (§7.4,
// rev 7.11): it decides a rollback's clean return (§7.8).
@@ -91,7 +97,10 @@ func (s Seal) Validate() error {
return fmt.Errorf("%w: no seal time or no writers' bound", ErrBadSeal)
}
seen := map[string]bool{}
- for _, class := range [][]SealedObject{s.Replica, s.Files, s.Timers} {
+ if len(s.State) > 1 || (len(s.State) == 1 && !strings.HasSuffix(s.State[0].Key, BuildStateSuffix)) {
+ return fmt.Errorf("%w: the state class names %d objects, want one at %s", ErrBadSeal, len(s.State), BuildStateSuffix)
+ }
+ for _, class := range [][]SealedObject{s.Replica, s.Files, s.Timers, s.State} {
for _, o := range class {
if o.Key == "" || !lease.UsableVersionID(o.VersionID) || o.Size < 0 || seen[o.Key] {
return fmt.Errorf("%w: object %q version %q size %d", ErrBadSeal, o.Key, o.VersionID, o.Size)
@@ -120,7 +129,7 @@ func sortObjects(objs []SealedObject) []SealedObject {
// object list sorted by key, so one seal always encodes to one string.
func (s Seal) Encode() string {
c := s
- c.Replica, c.Files, c.Timers = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers)
+ c.Replica, c.Files, c.Timers, c.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State)
c.SealedAt, c.LeaseUntil = s.SealedAt.UTC(), s.LeaseUntil.UTC()
b, err := json.Marshal(c)
if err != nil {
@@ -145,6 +154,6 @@ func DecodeSeal(raw string) (Seal, error) {
if err := s.Validate(); err != nil {
return Seal{}, err
}
- s.Replica, s.Files, s.Timers = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers)
+ s.Replica, s.Files, s.Timers, s.State = sortObjects(s.Replica), sortObjects(s.Files), sortObjects(s.Timers), sortObjects(s.State)
return s, nil
}
diff --git a/internal/moves/seal_state_test.go b/internal/moves/seal_state_test.go
new file mode 100644
index 000000000..aa924b6d2
--- /dev/null
+++ b/internal/moves/seal_state_test.go
@@ -0,0 +1,61 @@
+package moves
+
+import (
+ "errors"
+ "strings"
+ "testing"
+ "time"
+)
+
+// The seal's state class (settings spec rev 11 §4.3a, ruling R-SEAL-STATE):
+// the lane's build state by exact version, at most one object, at the
+// lane's state/ key, validated like every other class. A seal written
+// before the class existed has no "state" field and still decodes, with
+// none; a seal with none encodes without the field.
+func TestSealStateClass(t *testing.T) {
+ old := sealForTest().Encode()
+ if strings.Contains(old, `"state"`) {
+ t.Fatalf("a seal with no state encodes the field: %s", old)
+ }
+ got, err := DecodeSeal(old)
+ if err != nil || len(got.State) != 0 {
+ t.Fatalf("a seal without the state field: %+v %v", got.State, err)
+ }
+
+ s := sealForTest()
+ s.State = []SealedObject{{Key: BuildStateKey("h", "mv"), VersionID: "v6", Size: 90}}
+ got, err = DecodeSeal(s.Encode())
+ if err != nil || len(got.State) != 1 || got.State[0] != s.State[0] {
+ t.Fatalf("round trip: %+v %v", got.State, err)
+ }
+
+ for name, edit := range map[string]func(*Seal){
+ "two states": func(s *Seal) {
+ s.State = append(s.State, SealedObject{Key: BuildStateKey("h", "rb"), VersionID: "v7", Size: 1})
+ },
+ "another key": func(s *Seal) { s.State[0].Key = "moves/h/mv/replica/build-state.json" },
+ "no version": func(s *Seal) { s.State[0].VersionID = "" },
+ "a null version": func(s *Seal) { s.State[0].VersionID = "null" },
+ "negative size": func(s *Seal) { s.State[0].Size = -1 },
+ "a key twice": func(s *Seal) { s.Timers = append(s.Timers, s.State[0]) },
+ } {
+ t.Run(name, func(t *testing.T) {
+ s := sealForTest()
+ s.State = []SealedObject{{Key: BuildStateKey("h", "mv"), VersionID: "v6", Size: 90}}
+ edit(&s)
+ if err := s.Validate(); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("Validate = %v", err)
+ }
+ if _, err := DecodeSeal(s.Encode()); !errors.Is(err, ErrBadSeal) {
+ t.Fatalf("DecodeSeal = %v", err)
+ }
+ })
+ }
+
+ at := time.Date(2026, 10, 7, 9, 0, 0, 0, time.UTC)
+ st := []SealedObject{{Key: BuildStateKey("h", "rb"), VersionID: "v9", Size: 9}}
+ r := ReturnSeal(7, sealForTest().Replica, nil, nil, st, at, at.Add(time.Hour))
+ if len(r.State) != 1 || r.State[0] != st[0] {
+ t.Fatalf("ReturnSeal state %+v", r.State)
+ }
+}
diff --git a/internal/moves/staging.go b/internal/moves/staging.go
index 364969a4b..05eb8e82b 100644
--- a/internal/moves/staging.go
+++ b/internal/moves/staging.go
@@ -17,9 +17,12 @@ func MoveJSONKey(host, opID string) string { return StagingPrefix(host, opID)
// the installer places it beside the destination's lease before the probe.
// It is the only object under the lane's state/ prefix.
func BuildStateKey(host, opID string) string {
- return StagingPrefix(host, opID) + "state/build-state.json"
+ return StagingPrefix(host, opID) + BuildStateSuffix
}
+// BuildStateSuffix is BuildStateKey relative to the lane.
+const BuildStateSuffix = "state/build-state.json"
+
// ScopeLocation is the lane as a park scope location: no trailing slash
// (claim rev 6.23 §2 grammar, INV-23).
func ScopeLocation(host, opID string) string { return "moves/" + host + "/" + opID }
@@ -41,6 +44,10 @@ func LaneKey(opID string) string { return "moves/_lane/" + opID }
func ReceiptsKey(host, opID string) string { return "move-receipts/" + host + "/" + opID + ".json" }
func SnapshotsPrefix(host, opID string) string { return "move-snapshots/" + host + "/" + opID + "/" }
+// BuildStateReceiptsKey is the build-state lane's own copy receipts, beside
+// the replica lane's (settings spec rev 11 §4.3a).
+func BuildStateReceiptsKey(host, opID string) string { return ReceiptsKey(host, opID+"-state") }
+
// LedgerKey is the per-operation ledger (the move_json_version and archive
// seal facts the rounds record), beside the copy receipts and a different
// object from them.