amadan

carlos / platform Public

host: the landing from checking to restored: the flip's claim half, the residue parks, the generation build, the local install, the held render and the pinned release (section 7.6)

landingPass drives one landing record on the target as far as a pass can,
the cut's way: each phase does its work, advances landing_phase by one
guarded CAS and mirrors the step into the claim's op.phase; every phase
first walks a claim left behind forward along LandingOrder (section 7.5).
checking refuses under CARLOS_HOST_CLAIM and for a nested row (rulings
G16, G11), writes the flip's claim half when the source died before it
(G12), requires this box as executor with the source's complete evidence,
re-adds this box to touched (K12), notes an earlier stay's stamp before
activating the installation record at F, and queues a park-local only
when the inventory has paths at that stamp (G5). parking queues one
park-generation per residue generation (the origin as @root, the marker
never) and waits until each park's manifest exists and its deletion set
is gone. building restores the generation from the seal under the
location's lease, held through published and released there. local
retires the earlier stay through park.Local, extracts the verified
archive (a partial of its own removed first, G15), registers the four
flat paths under F, replaces the timers from the sealed snapshot with
their registrations, carries replication and next_due, removes the
restore-verification entry, records the prepared files manifest as
skipped until C8, and writes .files-gen last. configured renders the held
row and requires the pinned etags, waits on Plan 1's T7 hook (G4), then
installs the built database by rename (G34), writes <dbPath>.gen through
hostbox's marker writer (G2) and stamps installed_db_sha256 in the CAS to
restored, where Plan 3's probe takes over. landing_deadline before
restored marks the rollback.

The pass is armed only when the park engine's leases, view and
authorizer are wired, which production does not yet do (ruling G17); the
target report pass keeps a cutting or landing move's refusal and verdict
live (G26). The tests drive a real litestream replica through the cut to
the flip and land it on a target with its own data directory, the real
park.Local and host guard, and a fake lease table that refuses a second
acquisition of a held location.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
claude c8bfef4f74be449535c68594d9933da4acb3a097 parent 8673e1c
7 files changed, +2791 −5
  • approved_moves_copy_test.go +1 −0
  • internal/host/movesland.go +1444 −0
  • internal/host/movesland_test.go +1243 −0
  • internal/host/movespass.go +61 −0
  • internal/host/movesphase1.go +8 −2
  • internal/host/movesphase1_harness_test.go +26 −2
  • internal/host/movesreport.go +8 −1
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index d76463e37..a035ae96d 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -557,6 +557,7 @@ var movesScanFiles = []string{
"internal/host/movesheldenv.go",
"internal/host/movesheldlock.go",
"internal/host/moveshold.go",
+ "internal/host/movesland.go",
"internal/host/movesmeasure.go",
"internal/host/movesmeasure_linux.go",
"internal/host/movesmeasure_other.go",
diff --git a/internal/host/movesland.go b/internal/host/movesland.go
new file mode 100644
index 000000000..bce15a9d7
--- /dev/null
+++ b/internal/host/movesland.go
@@ -0,0 +1,1444 @@
+package host
+
+import (
+ "context"
+ "crypto/ecdh"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path/filepath"
+ "sort"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The landing on the target (moves spec §5.1, §7.5, §7.6), behind
+// CARLOS_MOVES_PHASE1. After the flip the record's region is the target's
+// and reconcile has minted the held row, so the target's pass visits the
+// record (OwnsRoute) and phase1MoveRound dispatches it here (ruling G34).
+// One pass drives one landing record as far as it can, as the cut does:
+// each phase does its work, then advances landing_phase by one guarded CAS,
+// then mirrors the step into the claim's op.phase (§7.5: the record is the
+// decision, the claim follows it). Nothing is carried in memory between
+// passes but the own-location lease (building through published): every
+// phase resumes from the record, the claim, this box's own bucket and the
+// op journal (moves.Journal, <DataDir>/.moves/<host>/<op>/).
+//
+// Phase names (ruling G8, the cut's convention). A record at X means every
+// phase before X is complete and X's own work may have begun; a replay of X
+// redoes X's work from durable state, and each phase's comment says what it
+// finds. The one name read the other way is `restored`: it is written by
+// the CAS that ends `configured`, after the database is installed and its
+// markers written, stamping landing.installed_db_sha256, so Plan 3's probe
+// reads `restored` as "the install is complete". Every later phase is the
+// probe's (Plan 3).
+//
+// The claim never leads the record. Every phase starts with syncClaimPhase:
+// equal is the common case; a claim behind the record (a crash between the
+// record CAS and the claim CAS) is walked forward along LandingOrder in one
+// claim CAS; a claim ahead is left alone and logged, and the phase's own
+// RequireMoveOp then refuses, so nothing is written over it.
+//
+// Nothing here ever serves from moves/: the lane is read by exact version
+// into the generation the build writes and into the files directory the
+// archive is extracted into (M14, M15). A failure that is permanent for this
+// landing is a rollback (§7.8), which Plan 4 builds: until then the record
+// carries the reason (provision.MoveMarkRollback) and this pass leaves a
+// marked record alone. landing_deadline passing before `restored` marks it
+// too.
+//
+// Arming (ruling G17). armPhase1 wires the hooks that exist on main (the
+// guard, park.Local, the own-location lease, RenderHeld, the registry
+// setters, the restore-verification record, the marker writer), and leaves
+// ParkLeases, ParkView and ParkAuthorize nil: the production park engine
+// needs claim Plan 6b's regional runner for the acknowledgement half and
+// Plan 5's view and authorizer. landingArmed therefore answers false in
+// production and the pass logs a standing notice instead of landing; the
+// tests arm everything with fakes and the stand-ins below. Under
+// CARLOS_HOST_CLAIM the landing refuses at checking with a recorded reason
+// (ruling G16): the held writes it makes have no claim selection until Task
+// 10b supplies the adapters.
+
+// landingStepsPerPass bounds the phases one pass runs.
+const landingStepsPerPass = 8
+
+// landingRequestLimit is the residue parks' request limit: the longest a
+// request against the location runs (hostbox.RequestLimit, the cut's too).
+const landingRequestLimit = cutRequestLimit
+
+// Op journal steps the landing writes (moves.Journal). Each sub-step of
+// `local` is a step, so a replay skips what is durable and finishes the
+// rest; `.files-gen` is written last and vouches for them all (§7.6).
+const (
+ landStepPrevious = "previous" // the earlier stay's stamp S' (decimal), noted before Activate
+ landStepResidue = "residue" // the generations whose residue parks were queued (JSON list)
+ landStepInstall = "install" // the completed install manifest (JSON)
+ landStepPublished = "published" // the build is complete and recorded (R10-3: no generation.json)
+ landStepRetired = "retired" // the earlier stay's artifacts are parked, or there were none
+ landStepExtracting = "extracting" // this landing's own extraction began (G15: a partial is ours)
+ landStepExtracted = "extracted" // the archive is in the files directory and <host>.env
+ landStepInventory = "inventory" // the new paths are registered under F
+ landStepTimers = "timers" // the timers table is the sealed snapshot
+ landStepColumns = "columns" // replication and next_due carried over
+ landStepVerify = "verify" // the restore-verification entry is gone
+ landStepFilesManifest = "files_manifest" // the prepared files manifest (C8; skipped until it exists)
+ landStepFilesGen = "files_gen" // .files-gen written
+)
+
+// landingFilesManifestSkipped is what the files_manifest step records while
+// C8 (the prepared-manifest build, region HA spec 1 §5.8) is not on main.
+const landingFilesManifestSkipped = "skipped: the prepared files manifest needs C8 (files.BuildPrepared)"
+
+// filesGenMarker is the .files-gen body (§5.1): the op and the incarnation.
+type filesGenMarker struct {
+ Op string `json:"op"`
+ Incarnation string `json:"incarnation"`
+}
+
+// landingRun is one pass's state for one landing record.
+type landingRun struct {
+ w *movesWorker
+ s release.Store
+ rec provision.Record
+ g provision.MoveGuard
+ guard hostguard.Guard
+ local park.Local
+}
+
+// landingArmed reports whether this box can run the landing: every hook
+// the phases call is wired. In production ParkLeases, ParkView and
+// ParkAuthorize are not (ruling G17), so this answers false there.
+func landingArmed(cfg moveCfg) bool {
+ return cfg.Box != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" && cfg.ParkLeases != nil &&
+ cfg.OwnLease != nil && cfg.RenderHeld != nil && cfg.ReplaceTimers != nil && cfg.SetReplication != nil &&
+ cfg.SetNextDue != nil && cfg.RemoveVerifyEntry != nil && cfg.WriteGen != nil && cfg.Route != nil &&
+ cfg.FilesDir != nil && cfg.Versioned != nil && cfg.ReleaseReady != nil && cfg.SecretsPriv != nil
+}
+
+// landingPass is the MoveLanding arm of phase1MoveRound.
+func (w *movesWorker) landingPass(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ if !landingArmed(cfg) {
+ w.notice(rec.Host+"#landing", "moves: %s: the landing of %s waits: this box's landing is not armed (claim Plan 6b's regional runner and Plan 5's park view are needed)", rec.Host, rec.Move.ID)
+ return nil
+ }
+ l := &landingRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}}
+ l.guard = cfg.Guard
+ if l.guard.Claims == nil {
+ l.guard.Claims = hostguard.StoreReader{Store: s}
+ }
+ l.local = cfg.Local
+ l.local.Guard, l.local.Claims = l.guard, s
+ if l.local.Box == "" {
+ l.local.Box = cfg.Box
+ }
+ if l.local.Authorize == nil {
+ l.local.Authorize = l.authorizeChore
+ }
+ defer l.settleLease()
+ for i := 0; i < landingStepsPerPass; i++ {
+ advanced, err := l.step(ctx)
+ if err != nil || !advanced {
+ return err
+ }
+ fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
+ if err != nil {
+ return fmt.Errorf("re-read the record: %w", err)
+ }
+ if fresh.Move == nil || fresh.Move.ID != l.g.ID || fresh.Move.Incarnation != l.g.Incarnation || fresh.Move.State != provision.MoveLanding {
+ return nil
+ }
+ l.rec = *fresh
+ }
+ return nil
+}
+
+func (l *landingRun) cfg() moveCfg { return l.w.cfg }
+
+// fault is the test seam after a side effect.
+func (l *landingRun) fault(point string) error {
+ if f := l.w.cfg.landFault; f != nil {
+ return f(point)
+ }
+ return nil
+}
+
+// step runs the record's current phase once. advanced is true when it
+// wrote the record (a phase or the rollback marker).
+func (l *landingRun) step(ctx context.Context) (bool, error) {
+ m := l.rec.Move
+ if m.State != provision.MoveLanding {
+ return false, nil
+ }
+ if m.Reason != "" {
+ l.w.notice(l.rec.Host+"#rollback", "moves: %s: landing %s is marked for rollback (%s); the rollback is Plan 4's", l.rec.Host, l.g.ID, m.Reason)
+ return false, nil
+ }
+ if m.LandingDeadline != nil && l.cfg().Now().After(*m.LandingDeadline) && landingBefore(m.LandingPhase, provision.LandRestored) {
+ return l.markRollback(ctx, provision.RollbackLandingDeadline)
+ }
+ switch m.LandingPhase {
+ case provision.LandChecking:
+ return l.checking(ctx)
+ case provision.LandParking:
+ return l.parking(ctx)
+ case provision.LandBuilding:
+ return l.building(ctx)
+ case provision.LandPublished:
+ return l.published(ctx)
+ case provision.LandLocal:
+ return l.local_(ctx)
+ case provision.LandConfigured:
+ return l.configured(ctx)
+ }
+ return false, nil // restored and later: the probe's (Plan 3)
+}
+
+// landingBefore reports whether phase comes before other in LandingOrder.
+func landingBefore(phase, other string) bool {
+ order := provision.LandingOrder()
+ return landingIndex(order, phase) >= 0 && landingIndex(order, phase) < landingIndex(order, other)
+}
+
+func landingIndex(order []string, phase string) int {
+ for i, p := range order {
+ if p == phase {
+ return i
+ }
+ }
+ return -1
+}
+
+// transition writes one guarded record transition. A conflict is someone
+// else's decision (a rollback, a console write): logged, and the pass ends
+// without error.
+func (l *landingRun) transition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
+ if _, err := provision.SetRecordMove(ctx, l.s, l.rec.Account, l.rec.App, l.rec.Host, mut); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: %s of %s refused (%v); the record moved on", l.rec.Host, what, l.g.ID, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("%s: %w", what, err)
+ }
+ return true, nil
+}
+
+// advance writes the record's step and then mirrors it into the claim. A
+// mirror that fails is logged: the next pass's syncClaimPhase writes it.
+func (l *landingRun) advance(ctx context.Context, from, to string, stamp func(*provision.MoveState)) (bool, error) {
+ ok, err := l.transition(ctx, "the landing "+from+" to "+to, provision.MoveLandingPhase(l.g, from, to, stamp))
+ if !ok {
+ return ok, err
+ }
+ log.Printf("moves: %s: landing %s is %s", l.rec.Host, l.g.ID, to)
+ if err := l.fault("after-record-" + to); err != nil {
+ return true, err
+ }
+ if err := l.mirrorPhase(ctx, from, to); err != nil {
+ log.Printf("moves: %s: the claim half of landing %s waits for the next pass: %v", l.rec.Host, to, err)
+ }
+ return true, nil
+}
+
+// mirrorPhase is the claim CAS that follows a record step (§7.5).
+func (l *landingRun) mirrorPhase(ctx context.Context, from, to string) error {
+ obs, err := l.claim(ctx)
+ if err != nil {
+ return err
+ }
+ _, err = hostclaim.Update(ctx, l.s, obs, hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, from, to))
+ return err
+}
+
+// markRollback records why the landing stopped (§7.6; Plan 4's rollback
+// transition replaces this one call site).
+func (l *landingRun) markRollback(ctx context.Context, reason string) (bool, error) {
+ ok, err := l.transition(ctx, "the rollback marker", provision.MoveMarkRollback(l.g, reason))
+ if ok {
+ log.Printf("moves: %s: landing %s stops for a rollback: %s", l.rec.Host, l.g.ID, reason)
+ nudgeSerial(ctx, l.s)
+ }
+ return ok, err
+}
+
+// claim reads the host claim and checks it is this move's op for the
+// record's incarnation.
+func (l *landingRun) claim(ctx context.Context) (hostclaim.Observed, error) {
+ obs, err := hostclaim.Read(ctx, l.s, l.rec.Host)
+ if err != nil {
+ return obs, fmt.Errorf("read the host claim: %w", err)
+ }
+ if obs.Absent {
+ return obs, fmt.Errorf("move %s has no host claim", l.g.ID)
+ }
+ if err := hostclaim.RequireOp(l.g.ID)(obs.Claim); err != nil {
+ return obs, fmt.Errorf("move %s: %w", l.g.ID, err)
+ }
+ if h := obs.Claim.Holder; h == nil || h.Incarnation != l.rec.Move.Incarnation {
+ return obs, fmt.Errorf("move %s: the host claim names another incarnation", l.g.ID)
+ }
+ return obs, nil
+}
+
+// syncClaimPhase is every phase's first step (§7.5): the claim's op.phase
+// follows the record's landing_phase. Behind, it is walked forward one step
+// at a time in one CAS; ahead, it is left alone and logged (the phase's own
+// RequireMoveOp then refuses); equal is the common case. The observation
+// returned is the one the phase works from, and names this box as the
+// executor.
+func (l *landingRun) syncClaimPhase(ctx context.Context) (hostclaim.Observed, error) {
+ cfg := l.cfg()
+ obs, err := l.claim(ctx)
+ if err != nil {
+ return obs, err
+ }
+ op := obs.Claim.Op
+ if op.Kind != hostclaim.KindLanding {
+ return obs, fmt.Errorf("move %s: the claim is %s/%q, not landing", l.g.ID, op.Kind, op.Phase)
+ }
+ if cfg.Box == "" || op.Exec != cfg.Box {
+ return obs, fmt.Errorf("move %s: the claim's executor is %q, not this box", l.g.ID, op.Exec)
+ }
+ order := provision.LandingOrder()
+ ci, ri := landingIndex(order, op.Phase), landingIndex(order, l.rec.Move.LandingPhase)
+ switch {
+ case ci < 0 || ri < 0:
+ return obs, fmt.Errorf("move %s: landing phases %q (claim) and %q (record) are not both known", l.g.ID, op.Phase, l.rec.Move.LandingPhase)
+ case ci == ri:
+ return obs, nil
+ case ci > ri:
+ log.Printf("moves: %s: the claim's landing phase %q is ahead of the record's %q; left alone", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)
+ return obs, nil
+ }
+ obs, err = hostclaim.Update(ctx, l.s, obs, func(c *hostclaim.Claim) error {
+ for i := ci; i < ri; i++ {
+ if err := hostclaim.SetMovePhase(l.g.ID, hostclaim.KindLanding, order[i], order[i+1])(c); err != nil {
+ return err
+ }
+ }
+ return nil
+ })
+ if err != nil {
+ return obs, fmt.Errorf("catch the claim up to %s: %w", l.rec.Move.LandingPhase, err)
+ }
+ log.Printf("moves: %s: the claim's landing phase caught up from %q to %q", l.rec.Host, op.Phase, l.rec.Move.LandingPhase)
+ return obs, nil
+}
+
+// requirePhase is the phase's K5 revalidation on the synced observation.
+func (l *landingRun) requirePhase(obs hostclaim.Observed, phase string) error {
+ if err := hostclaim.RequireMoveOp(l.g.ID, hostclaim.KindLanding, phase)(obs.Claim); err != nil {
+ return fmt.Errorf("move %s: %w", l.g.ID, err)
+ }
+ return nil
+}
+
+// landingCheck is the claim check the guard runs before a local write of
+// phase: this landing's op, at that phase, with this box executing.
+func (l *landingRun) landingCheck(phase string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if err := hostclaim.RequireMoveOp(l.g.ID, hostclaim.KindLanding, phase)(c); err != nil {
+ return err
+ }
+ if c.Op.Exec != l.cfg().Box {
+ return fmt.Errorf("%w: the executor is %q, not this box", hostclaim.ErrWrongState, c.Op.Exec)
+ }
+ return nil
+ }
+}
+
+// authorizeChore is park.Local.Authorize's stand-in until Plan 5: a chore is
+// run only when it is this landing's own and the claim still names this
+// landing with this box executing.
+func (l *landingRun) authorizeChore(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
+ if c == nil || o.Op != l.g.ID || c.Op.ID != l.g.ID || c.Op.Kind != hostclaim.KindLanding || c.Op.Exec != l.cfg().Box {
+ return fmt.Errorf("%w: not this landing's chore", park.ErrUnsafe)
+ }
+ return nil
+}
+
+// authorizeObligation is park.Engine.Authorize's stand-in (ruling G17): the
+// same rule over an observation.
+func (l *landingRun) authorizeObligation(ctx context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ if obs.Absent {
+ return park.ErrUnsafe
+ }
+ return l.authorizeChore(ctx, obs.Claim, o)
+}
+
+// landingView is park.Engine.View's stand-in (ruling G17): the location's
+// objects classified by key, and the source's region authoritative, which
+// §5.1's table says it is for a landing before the target's probe is
+// admitted (every phase here). Plan 5's provider replaces it; a test may set
+// cfg.ParkView to the view a later state has.
+func (l *landingRun) landingView(_ context.Context, obs hostclaim.Observed) (park.View, error) {
+ if obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
+ return park.View{}, park.ErrUnsafe
+ }
+ location := strings.TrimSuffix(obs.Claim.Holder.Location, "/")
+ return park.View{ClaimETag: obs.ETag, Region: l.cfg().Region, AuthoritativeRegion: l.rec.Move.From,
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(location, r.Key) }}, nil
+}
+
+// landingMarkerGeneration is what landingGenerationOf answers for the
+// location's generation.json. The marker is the one key every generation
+// reuses (§5.1): the probe's admission supersedes an earlier stay's marker
+// with If-Match rather than deleting it, and a park-generation chore whose
+// deletion set named the marker key could never be acknowledged once a new
+// marker stands there (the engine skips the marker key and then refuses a
+// recorded delete at it). So the marker belongs to no generation's chore:
+// parking never queues one for it and the engine passes it over. "@" keeps
+// it apart from any generation name, as hostclaim.RootGeneration does.
+const landingMarkerGeneration = "@marker"
+
+// landingGenerationOf classifies an object under a holder location by
+// generation, as the view does: `g-<gen>/...` and `files/g-<gen>/...` are
+// gen's; a level directory (four hex digits) at the location is the
+// origin, ""; generation.json is landingMarkerGeneration; anything else is
+// unknown, and a caller fails closed on it. The naming rule is
+// moves.GenerationPrefix's twin.
+func landingGenerationOf(location, key string) (string, bool) {
+ rel, ok := strings.CutPrefix(key, location+"/")
+ if !ok || rel == "" {
+ return "", false
+ }
+ for _, p := range []string{"g-", "files/g-"} {
+ if rest, ok := strings.CutPrefix(rel, p); ok {
+ if gen, _, ok := strings.Cut(rest, "/"); ok && gen != "" {
+ return gen, true
+ }
+ return "", false
+ }
+ }
+ if rel == "generation.json" {
+ return landingMarkerGeneration, true
+ }
+ if first, _, ok := strings.Cut(rel, "/"); ok && len(first) == 4 && isHexString(first) {
+ return "", true
+ }
+ return "", false
+}
+
+func isHexString(s string) bool {
+ for _, c := range s {
+ if (c < '0' || c > '9') && (c < 'a' || c > 'f') && (c < 'A' || c > 'F') {
+ return false
+ }
+ }
+ return s != ""
+}
+
+// route is the host's registry row on this box.
+func (l *landingRun) route() (registry.Route, error) {
+ rt, err := l.cfg().Route(l.rec.Host)
+ if err != nil {
+ return registry.Route{}, fmt.Errorf("read the registry row: %w", err)
+ }
+ if rt == nil {
+ return registry.Route{}, fmt.Errorf("%s has no registry row on this box", l.rec.Host)
+ }
+ return *rt, nil
+}
+
+func (l *landingRun) dbPath() string { return provision.DefaultDBPath(l.cfg().DataDir, l.rec.Host) }
+
+func (l *landingRun) journal() (*moves.Journal, error) {
+ return moves.ReadJournal(l.cfg().DataDir, l.rec.Host, l.g.ID)
+}
+
+func (l *landingRun) journalDir() string {
+ return moves.JournalDir(l.cfg().DataDir, l.rec.Host, l.g.ID)
+}
+
+// buildDir is BuildGeneration's scratch, <JournalDir>/build/ (ruling G34),
+// on the data disk so the result renames into DBPath.
+func (l *landingRun) buildDir() string { return filepath.Join(l.journalDir(), "build") }
+
+// versioned is this box's own bucket by exact version.
+func (l *landingRun) versioned(ctx context.Context) (lease.VersionedStore, error) {
+ cfg := l.cfg()
+ if cfg.Versioned == nil {
+ return nil, errors.New("no versioned store on this box")
+ }
+ return cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+}
+
+// ensureLocalDirs creates the two directories the host guard and park.Local
+// require under the data directory (hostguard/ for the installation record,
+// .parked/ for retired artifacts). On a box with claims on they exist
+// already; a Phase 1 landing on a box without them creates them.
+func (l *landingRun) ensureLocalDirs() error {
+ for _, d := range []string{"hostguard", ".parked"} {
+ if err := os.MkdirAll(filepath.Join(l.cfg().DataDir, d), 0o770); err != nil {
+ return err
+ }
+ }
+ return nil
+}
+
+// checking (§7.6): the refusals this box can know (G16, G11), the flip's
+// claim half if the source crashed before it (G12), then the claim's word:
+// landing/checking for this move with this box executing and a root-layout
+// holder; the source's complete stop evidence; this box in touched.boxes
+// (K12, before any local write); move.json's incarnation; the earlier
+// stay's stamp S' noted in the journal before the installation record is
+// activated at F (holder.stamp, the flip's); and a park-local queued only
+// when the inventory has paths tagged S' (ruling G5). No lease is taken
+// here: the residue parks take the location's lease themselves.
+//
+// A replay finds checking and runs it all again: every write is
+// idempotent (the flip, Touch and the park queue under the same op, the
+// journal step written once, Activate at the same stamp), and Capture after
+// Activate reads F, which is never noted as a previous stay.
+func (l *landingRun) checking(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ if hostwrite.Enabled(os.Getenv) {
+ return l.markRollback(ctx, provision.RollbackClaimWired)
+ }
+ rt, err := l.route()
+ if err != nil {
+ return false, err
+ }
+ if rt.DBPath != l.dbPath() {
+ log.Printf("moves: %s: the held row's database is at %s, not the flat %s", host, rt.DBPath, l.dbPath())
+ return l.markRollback(ctx, provision.RollbackLandingLayout)
+ }
+ if !registry.IsHold(rt.Status) || rt.HoldOp != id {
+ return false, fmt.Errorf("the registry row is %q held by %q, not held by landing %s", rt.Status, rt.HoldOp, id)
+ }
+ obs, err := l.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if op := obs.Claim.Op; op.Kind == hostclaim.KindMoving && op.Phase == hostclaim.PhaseSealed && op.Exec == "" && op.Journal[hostclaim.JournalEvidenceSource] != "" {
+ // The record flipped and the source died before the claim half
+ // (ruling G12): the target writes it, with the same preconditions
+ // the source would have met.
+ h := obs.Claim.Holder
+ if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.Flip(id, cfg.Box, h.Stamp, h.Generation)); err != nil {
+ return false, fmt.Errorf("the claim half of the flip: %w", err)
+ }
+ log.Printf("moves: %s: the claim half of the flip was written by the target", host)
+ if err := l.fault("after-target-flip"); err != nil {
+ return false, err
+ }
+ }
+ if err := l.requirePhase(obs, provision.LandChecking); err != nil {
+ return false, err
+ }
+ if cfg.Box == "" || obs.Claim.Op.Exec != cfg.Box {
+ return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, obs.Claim.Op.Exec)
+ }
+ holder := *obs.Claim.Holder
+ if _, err := writerLease(cfg.OwnPrefix, host, holder); err != nil {
+ return false, fmt.Errorf("move %s: %w", id, err)
+ }
+ src := l.rec.Move.Source
+ if src == nil || src.Evidence == "" {
+ return false, fmt.Errorf("move %s: the record has no source baselines", id)
+ }
+ if obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource] != src.Evidence {
+ return false, fmt.Errorf("move %s: the claim cites %q, the record %q", id, obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource], src.Evidence)
+ }
+ report, err := readStoppedReport(ctx, l.s, src.Evidence)
+ if err != nil {
+ return false, err
+ }
+ if report.Host != host || report.OpID != id || report.Role != "source" || report.FinalTXID != src.FinalTXID {
+ return false, fmt.Errorf("move %s: %s is %s %s %s at TXID %d, not this move's source at %d", id, src.Evidence, report.Host, report.OpID, report.Role, report.FinalTXID, src.FinalTXID)
+ }
+ if !containsString(obs.Claim.Touched.Boxes, cfg.Box) {
+ if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.Touch(id, nil, []string{cfg.Box})); err != nil {
+ return false, fmt.Errorf("touch the claim: %w", err)
+ }
+ }
+ own, err := l.w.mirror(ctx, cfg.OwnBucketRegion)
+ if err != nil {
+ return false, err
+ }
+ body, err := own.Get(ctx, cfg.OwnBucket, moves.MoveJSONKey(host, id))
+ if err != nil {
+ return false, fmt.Errorf("read move.json: %w", err)
+ }
+ var mj moves.MoveJSON
+ if err := json.Unmarshal(body, &mj); err != nil {
+ return false, fmt.Errorf("move.json: %w", err)
+ }
+ if mj.Incarnation != l.rec.Move.Incarnation || holder.Incarnation != l.rec.Move.Incarnation {
+ return false, fmt.Errorf("move %s: move.json names incarnation %q, the record %q", id, mj.Incarnation, l.rec.Move.Incarnation)
+ }
+ if err := l.ensureLocalDirs(); err != nil {
+ return false, err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ captured, err := hostguard.Capture(cfg.DataDir, host)
+ switch {
+ case errors.Is(err, hostguard.ErrNoRecord):
+ // No earlier stay on this box.
+ case err != nil:
+ return false, fmt.Errorf("read the installation record: %w", err)
+ case !j.Done(landStepPrevious) && captured.Stamp != 0 && captured.Stamp != holder.Stamp && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, captured.Stamp):
+ j.Phase = provision.LandChecking
+ j.Mark(landStepPrevious, strconv.FormatUint(captured.Stamp, 10))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: an earlier stay at stamp %d will be retired by local", host, captured.Stamp)
+ }
+ if err := l.fault("after-previous"); err != nil {
+ return false, err
+ }
+ if err := l.guard.Activate(ctx, host, cfg.Box, nil); err != nil {
+ return false, fmt.Errorf("activate the installation record: %w", err)
+ }
+ if err := l.fault("after-activate"); err != nil {
+ return false, err
+ }
+ if j.Done(landStepPrevious) {
+ if obs, err = l.claim(ctx); err != nil {
+ return false, err
+ }
+ if _, err := hostclaim.Update(ctx, l.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil {
+ return false, fmt.Errorf("queue the local park: %w", err)
+ }
+ }
+ return l.advance(ctx, provision.LandChecking, provision.LandParking, nil)
+}
+
+// parking (§7.6, D6): whatever an earlier stay of this incarnation left at
+// holder.location in this box's own bucket is parked before the build
+// writes beside it, one park-generation chore per generation (the origin as
+// @root), scoped to the location with the flip's time as the evidence
+// horizon. Each chore is run once per pass through the park engine, which
+// takes the location's lease itself (ErrPending is the expected answer: the
+// acknowledgement waits for the horizon, and is C4's runner's); the phase
+// advances once every chore's manifest exists and its deletion set is gone
+// (Engine.Parked). Per-generation chores are what let the acknowledgement
+// finish after the probe makes the new generation live here.
+//
+// A replay finds parking and continues from the claim: the residue list is
+// journalled so the chores are the same ones, the queue is idempotent per
+// generation, and Engine.Run resumes a chore from its manifest.
+func (l *landingRun) parking(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandParking); err != nil {
+ return false, err
+ }
+ holder := *obs.Claim.Holder
+ location := strings.TrimSuffix(holder.Location, "/")
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return false, err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ var gens []string
+ if raw, ok := j.Steps[landStepResidue]; ok {
+ if err := json.Unmarshal([]byte(raw), &gens); err != nil {
+ return false, fmt.Errorf("the journal's residue list: %w", err)
+ }
+ } else {
+ objs, err := vs.ListVersions(ctx, cfg.OwnBucket, location+"/")
+ if err != nil {
+ return false, fmt.Errorf("list %s: %w", location, err)
+ }
+ seen := map[string]bool{}
+ for _, o := range objs {
+ if o.Key == location+"/lock.json" {
+ continue
+ }
+ g, ok := landingGenerationOf(location, o.Key)
+ if !ok {
+ return false, fmt.Errorf("%s is not an object of a generation this landing can classify", o.Key)
+ }
+ if g == id || g == landingMarkerGeneration {
+ // The landing's own generation is never residue, and the
+ // marker is superseded by the probe, never parked.
+ continue
+ }
+ if g == "" {
+ g = hostclaim.RootGeneration
+ }
+ seen[g] = true
+ }
+ for g := range seen {
+ gens = append(gens, g)
+ }
+ sort.Strings(gens)
+ body, err := json.Marshal(gens)
+ if err != nil {
+ return false, err
+ }
+ j.Phase = provision.LandParking
+ j.Mark(landStepResidue, string(body))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ }
+ if len(gens) == 0 {
+ return l.advance(ctx, provision.LandParking, provision.LandBuilding, nil)
+ }
+ if l.rec.Move.FlippedAt == nil {
+ return false, fmt.Errorf("move %s: the record has no flip time", id)
+ }
+ scope := hostclaim.ParkScope{Location: location, Incarnation: holder.Incarnation, EvidenceAt: l.rec.Move.FlippedAt.UTC(), RequestLimit: landingRequestLimit}
+ if obs, err = hostclaim.Update(ctx, l.s, obs, hostclaim.QueueResidueParks(id, cfg.Region, scope, gens, obligationID)); err != nil {
+ return false, fmt.Errorf("queue the residue parks: %w", err)
+ }
+ if err := l.fault("after-residue-queued"); err != nil {
+ return false, err
+ }
+ eng := l.engine(vs)
+ pending := 0
+ for _, o := range obs.Claim.Obligations {
+ if o.Op != id || o.Kind != "park-generation" || o.Region != cfg.Region {
+ continue
+ }
+ w := park.Work{Host: host, ID: o.ID, Op: id}
+ if err := eng.Run(ctx, w); err != nil && !errors.Is(err, park.ErrPending) {
+ return false, fmt.Errorf("park generation %s: %w", o.Generation, err)
+ }
+ parked, err := eng.Parked(ctx, w)
+ if errors.Is(err, park.ErrChanged) {
+ continue // acknowledged already
+ }
+ if err != nil {
+ return false, fmt.Errorf("park generation %s: %w", o.Generation, err)
+ }
+ if !parked {
+ pending++
+ }
+ }
+ if pending > 0 {
+ l.w.notice(host+"#parking", "moves: %s: %d residue park(s) are not complete yet", host, pending)
+ return false, nil
+ }
+ return l.advance(ctx, provision.LandParking, provision.LandBuilding, nil)
+}
+
+// engine is the park engine over this box's own bucket: the wired leases,
+// and the view and authorizer Plan 5 supplies, or the stand-ins.
+func (l *landingRun) engine(vs lease.VersionedStore) park.Engine {
+ cfg := l.cfg()
+ view, auth := cfg.ParkView, cfg.ParkAuthorize
+ if view == nil {
+ view = l.landingView
+ }
+ if auth == nil {
+ auth = l.authorizeObligation
+ }
+ return park.Engine{Store: vs, Claims: l.s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
+ Now: cfg.Now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault}
+}
+
+// lease is the own-location lease the build holds (§7.6 checking: "hold it
+// through published"), kept in the worker's per-host map across passes and
+// re-acquired after a restart. waiting is true when another holder has it.
+func (l *landingRun) lease(ctx context.Context, holder hostclaim.Holder) (held park.HeldLease, waiting bool, err error) {
+ cfg := l.cfg()
+ host := l.rec.Host
+ l.w.mu.Lock()
+ held = l.w.landLeases[host]
+ l.w.mu.Unlock()
+ if held != nil {
+ if err := held.Check(ctx); err == nil {
+ return held, false, nil
+ }
+ l.releaseLease()
+ }
+ held, err = cfg.OwnLease(ctx, cfg.OwnBucket, strings.TrimSuffix(holder.Location, "/"))
+ if err != nil {
+ var he *lease.HeldError
+ if errors.As(err, &he) {
+ log.Printf("moves: %s: another holder has the location's lease; the build waits", host)
+ return nil, true, nil
+ }
+ return nil, false, fmt.Errorf("the location's lease: %w", err)
+ }
+ l.w.mu.Lock()
+ if l.w.landLeases == nil {
+ l.w.landLeases = map[string]park.HeldLease{}
+ }
+ l.w.landLeases[host] = held
+ l.w.mu.Unlock()
+ return held, false, nil
+}
+
+// releaseLease releases the host's own-location lease, if held.
+func (l *landingRun) releaseLease() {
+ host := l.rec.Host
+ l.w.mu.Lock()
+ held := l.w.landLeases[host]
+ delete(l.w.landLeases, host)
+ l.w.mu.Unlock()
+ if held == nil {
+ return
+ }
+ rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := held.Release(rctx); err != nil {
+ log.Printf("moves: %s: release the location's lease: %v", host, err)
+ }
+}
+
+// settleLease releases the lease once the record is past published, marked
+// for a rollback, or no longer landing: the lease is the build's alone.
+func (l *landingRun) settleLease() {
+ m := l.rec.Move
+ if m == nil || m.State != provision.MoveLanding || m.Reason != "" || !landingBefore(m.LandingPhase, provision.LandLocal) {
+ l.releaseLease()
+ }
+}
+
+// building (§7.6, §5.1 step 2): the generation built from the seal's pinned
+// replica versions under the location's lease (moves.BuildGeneration: the
+// intent, the snapshot, the anchor, the completed install manifest, every
+// one create-only), with the result left at <JournalDir>/build/<host>.db
+// for the restore. The manifest goes into the journal.
+//
+// A replay finds building and builds again: BuildGeneration reuses what it
+// finds under the generation's keys and refuses anything it did not make.
+func (l *landingRun) building(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandBuilding); err != nil {
+ return false, err
+ }
+ holder := *obs.Claim.Holder
+ held, waiting, err := l.lease(ctx, holder)
+ if waiting || err != nil {
+ return false, err
+ }
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return false, err
+ }
+ src := l.rec.Move.Source
+ if src == nil || seal.N != src.FinalTXID {
+ return false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", id, seal.N, srcTXID(src))
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := held.Check(ctx); err != nil {
+ return false, fmt.Errorf("the location's lease: %w", err)
+ }
+ m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ GenerationPrefix: moves.GenerationPrefix(holder.Location, id), Op: id, Host: host, WorkDir: l.buildDir(), Now: cfg.Now})
+ if err != nil {
+ return false, fmt.Errorf("build the generation: %w", err)
+ }
+ body, err := json.Marshal(m)
+ if err != nil {
+ return false, err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ j.Phase = provision.LandBuilding
+ j.Mark(landStepInstall, string(body))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ if err := l.fault("after-build"); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: generation %s built at TXID %d", host, id, m.N)
+ return l.advance(ctx, provision.LandBuilding, provision.LandPublished, nil)
+}
+
+func srcTXID(src *provision.MoveSource) uint64 {
+ if src == nil {
+ return 0
+ }
+ return src.FinalTXID
+}
+
+// published (§7.6): the build is complete and recorded; generation.json is
+// not written here (R10-3: the probe's cold admission publishes it). The
+// journal records the phase, the record advances to local and the location's
+// lease is released: the build was the only writer that needed it.
+func (l *landingRun) published(ctx context.Context) (bool, error) {
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandPublished); err != nil {
+ return false, err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ if !j.Done(landStepInstall) {
+ return false, fmt.Errorf("move %s: published without an install manifest in the journal", l.g.ID)
+ }
+ j.Phase = provision.LandPublished
+ j.Mark(landStepPublished, l.cfg().Now().UTC().Format(time.RFC3339Nano))
+ if err := j.Write(l.cfg().DataDir); err != nil {
+ return false, err
+ }
+ ok, err := l.advance(ctx, provision.LandPublished, provision.LandLocal, nil)
+ if ok {
+ l.releaseLease()
+ }
+ return ok, err
+}
+
+// local (§7.6), each sub-step journalled and skipped on a replay once done,
+// in this order (round-2 finding 8): (1) the installation record must be at
+// F; (2) an earlier stay's artifacts tagged S' are parked into .parked/
+// through park.Local (which refuses a path tagged with another stamp), or a
+// park-local left with nothing to retire is completed; (3) the sealed
+// archive, read by exact version and verified, is extracted into the files
+// directory and <host>.env (a partial of this landing's own is removed
+// first, G15: the journal says extraction began and the paths are not yet
+// registered); (4) the new paths are registered under F; (5) the timers
+// table is the sealed snapshot with its registrations, the carried columns
+// are set and the restore-verification entry removed; (6) the prepared
+// files manifest, skipped with a log line until C8 exists; (7) the files
+// directory is synced and .files-gen written last, so its presence vouches
+// for every step before it.
+//
+// A replay finds local and finishes what the journal does not record;
+// nothing here is read by the probe before .files-gen exists.
+func (l *landingRun) local_(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandLocal); err != nil {
+ return false, err
+ }
+ holder := *obs.Claim.Holder
+ src := l.rec.Move.Source
+ if src == nil || src.Files == nil {
+ return false, fmt.Errorf("move %s: the record has no source baselines", id)
+ }
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return false, err
+ }
+ j, err := l.journal()
+ if err != nil {
+ return false, err
+ }
+ j.Phase = provision.LandLocal
+ mark := func(step, value string) error {
+ j.Mark(step, value)
+ if err := j.Write(cfg.DataDir); err != nil {
+ return err
+ }
+ return l.fault("after-" + step)
+ }
+ captured, err := hostguard.Capture(cfg.DataDir, host)
+ if err != nil {
+ return false, fmt.Errorf("read the installation record: %w", err)
+ }
+ if captured.Incarnation != holder.Incarnation || captured.Stamp != holder.Stamp {
+ return false, fmt.Errorf("move %s: the installation record is at stamp %d of %s, not this landing's %d", id, captured.Stamp, captured.Incarnation, holder.Stamp)
+ }
+ if !j.Done(landStepRetired) {
+ if raw, ok := j.Steps[landStepPrevious]; ok {
+ earlier, err := strconv.ParseUint(raw, 10, 64)
+ if err != nil || earlier == 0 {
+ return false, fmt.Errorf("the journal's previous stamp %q", raw)
+ }
+ var chore *hostclaim.Obligation
+ for i := range obs.Claim.Obligations {
+ if o := &obs.Claim.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
+ chore = o
+ }
+ }
+ switch {
+ case chore != nil && hostclaim.InventoryHasStamp(obs.Claim, cfg.Box, earlier):
+ if err := l.local.RetireArtifacts(ctx, host, captured, chore.ID, id, earlier); err != nil {
+ return false, fmt.Errorf("retire the earlier stay's artifacts: %w", err)
+ }
+ log.Printf("moves: %s: the earlier stay's artifacts (stamp %d) are parked", host, earlier)
+ case chore != nil:
+ if _, err := hostclaim.Update(ctx, l.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, earlier)); err != nil {
+ return false, fmt.Errorf("complete the local park: %w", err)
+ }
+ }
+ }
+ if err := mark(landStepRetired, ""); err != nil {
+ return false, err
+ }
+ }
+ dir, envPath := cfg.FilesDir(host)
+ if !j.Done(landStepExtracted) {
+ if j.Done(landStepExtracting) {
+ // G15: a partial extraction of this landing's own, not yet
+ // registered. Nothing else can be at these paths: the earlier
+ // stay's were parked above.
+ if err := os.RemoveAll(dir); err != nil {
+ return false, err
+ }
+ if err := os.Remove(envPath); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return false, err
+ }
+ }
+ if err := mark(landStepExtracting, ""); err != nil {
+ return false, err
+ }
+ if err := l.extractArchive(ctx, seal, *src.Files, dir, envPath); err != nil {
+ return false, fmt.Errorf("extract the archive: %w", err)
+ }
+ if err := mark(landStepExtracted, ""); err != nil {
+ return false, err
+ }
+ }
+ if !j.Done(landStepInventory) {
+ paths := []string{host + ".db", host + ".db.gen", host, host + ".env"}
+ if err := l.local.RecordInventory(ctx, host, captured, "flat", paths, l.landingCheck(provision.LandLocal)); err != nil {
+ return false, fmt.Errorf("record the inventory: %w", err)
+ }
+ if err := mark(landStepInventory, ""); err != nil {
+ return false, err
+ }
+ }
+ if !j.Done(landStepTimers) {
+ rows, err := l.sealedTimers(ctx, seal, src.TimersSHA256)
+ if err != nil {
+ return false, err
+ }
+ if err := cfg.ReplaceTimers(host, rows); err != nil {
+ return false, fmt.Errorf("replace the timers: %w", err)
+ }
+ if err := mark(landStepTimers, strconv.Itoa(len(rows))); err != nil {
+ return false, err
+ }
+ }
+ if !j.Done(landStepColumns) {
+ if err := cfg.SetReplication(host, src.Replication); err != nil {
+ return false, fmt.Errorf("set replication: %w", err)
+ }
+ if err := cfg.SetNextDue(host, src.NextDue); err != nil {
+ return false, fmt.Errorf("set next_due: %w", err)
+ }
+ if err := mark(landStepColumns, ""); err != nil {
+ return false, err
+ }
+ }
+ if !j.Done(landStepVerify) {
+ if err := cfg.RemoveVerifyEntry(ctx, l.s, host); err != nil {
+ return false, fmt.Errorf("remove the restore-verification entry: %w", err)
+ }
+ if err := mark(landStepVerify, ""); err != nil {
+ return false, err
+ }
+ }
+ if !j.Done(landStepFilesManifest) {
+ // Spec rev 7.22 §7.6 (L17, R19-1): the prepared files manifest under
+ // files/g-<move id>/ in the location, bound to {location, g-<op>, op,
+ // F}. Its build (C8, files.BuildPrepared) is not on main; until it
+ // is, the step is recorded as skipped and the probe's admission
+ // establishes the files baseline (Plan 3).
+ log.Printf("moves: %s: %s", host, landingFilesManifestSkipped)
+ if err := mark(landStepFilesManifest, landingFilesManifestSkipped); err != nil {
+ return false, err
+ }
+ }
+ if err := l.fault("before-files-gen"); err != nil {
+ return false, err
+ }
+ if err := syncPath(dir); err != nil {
+ return false, err
+ }
+ marker, err := json.Marshal(filesGenMarker{Op: id, Incarnation: holder.Incarnation})
+ if err != nil {
+ return false, err
+ }
+ if err := writeFileAtomic(filepath.Join(dir, archive.MarkerName), marker, 0o644); err != nil {
+ return false, fmt.Errorf("write %s: %w", archive.MarkerName, err)
+ }
+ if err := mark(landStepFilesGen, ""); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: files, environment, timers and columns installed; %s written", host, archive.MarkerName)
+ return l.advance(ctx, provision.LandLocal, provision.LandConfigured, nil)
+}
+
+// extractArchive reads the seal's archive and manifest by exact version from
+// this box's own bucket, verifies the archive against the manifest and the
+// record's digests (archive.Verify: the sealed digest, every entry), and
+// extracts it into dir and envPath (archive.Open under the box's secrets
+// key, then archive.Extract, which admits nothing it did not read whole).
+func (l *landingRun) extractArchive(ctx context.Context, seal moves.Seal, want provision.MoveFilesDigest, dir, envPath string) error {
+ cfg := l.cfg()
+ var sealedObj, manifestObj *moves.SealedObject
+ for i := range seal.Files {
+ switch o := &seal.Files[i]; {
+ case strings.HasSuffix(o.Key, ".sealed"):
+ sealedObj = o
+ case strings.HasSuffix(o.Key, ".json"):
+ manifestObj = o
+ }
+ }
+ if sealedObj == nil || manifestObj == nil {
+ return errors.New("the seal does not name the archive and its manifest")
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return err
+ }
+ raw, err := vs.GetVersion(ctx, cfg.OwnBucket, manifestObj.Key, manifestObj.VersionID)
+ if err != nil {
+ return fmt.Errorf("read the archive manifest: %w", err)
+ }
+ var m archive.Manifest
+ if err := json.Unmarshal(raw, &m); err != nil {
+ return fmt.Errorf("the archive manifest: %w", err)
+ }
+ if m.Op != l.g.ID || m.Incarnation != l.rec.Move.Incarnation || m.Plain != want.Plain || m.Sealed != want.Sealed {
+ return fmt.Errorf("the archive manifest is for %s/%s (%s, %s), not this move's %s (%s, %s)", m.Op, m.Incarnation, m.Plain, m.Sealed, l.g.ID, want.Plain, want.Sealed)
+ }
+ if err := os.MkdirAll(l.journalDir(), 0o700); err != nil {
+ return err
+ }
+ sealedPath := filepath.Join(l.journalDir(), "archive.sealed")
+ if err := os.Remove(sealedPath); err != nil && !errors.Is(err, os.ErrNotExist) {
+ return err
+ }
+ defer os.Remove(sealedPath)
+ if err := downloadVersion(ctx, vs, cfg.OwnBucket, sealedObj.Key, sealedObj.VersionID, sealedPath); err != nil {
+ return fmt.Errorf("read the archive: %w", err)
+ }
+ f, err := os.Open(sealedPath)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ bind := appconfig.Binding{Account: l.rec.Account, App: l.rec.App}
+ if err := archive.Verify(f, m, cfg.SecretsPriv, bind); err != nil {
+ return fmt.Errorf("verify the archive: %w", err)
+ }
+ return openAndExtract(f, m, cfg.SecretsPriv, bind, dir, envPath)
+}
+
+// openAndExtract pipes the opened archive into Extract, as carriedEnv does.
+func openAndExtract(f *os.File, m archive.Manifest, priv *ecdh.PrivateKey, bind appconfig.Binding, dir, envPath string) error {
+ if _, err := f.Seek(0, io.SeekStart); err != nil {
+ return err
+ }
+ pr, pw := io.Pipe()
+ done := make(chan error, 1)
+ go func() {
+ err := archive.Open(f, m.SealedDataKey, priv, bind, pw)
+ pw.CloseWithError(err)
+ done <- err
+ }()
+ xerr := archive.Extract(pr, dir, envPath)
+ pr.CloseWithError(errors.New("extract stopped reading"))
+ if oerr := <-done; oerr != nil && xerr == nil {
+ xerr = oerr
+ }
+ return xerr
+}
+
+// sealedTimers reads the sealed timers snapshot by exact version and checks
+// its digest against the record's; every row is this host's.
+func (l *landingRun) sealedTimers(ctx context.Context, seal moves.Seal, wantDigest string) ([]registry.Timer, error) {
+ cfg := l.cfg()
+ if len(seal.Timers) != 1 {
+ return nil, fmt.Errorf("the seal names %d timers objects, want 1", len(seal.Timers))
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return nil, err
+ }
+ raw, err := vs.GetVersion(ctx, cfg.OwnBucket, seal.Timers[0].Key, seal.Timers[0].VersionID)
+ if err != nil {
+ return nil, fmt.Errorf("read the timers snapshot: %w", err)
+ }
+ snap, err := moves.DecodeTimers(raw)
+ if err != nil {
+ return nil, err
+ }
+ if snap.Host != l.rec.Host {
+ return nil, fmt.Errorf("the timers snapshot is %s's, not %s's", snap.Host, l.rec.Host)
+ }
+ if got := moves.TimersDigest(snap.Rows); got != wantDigest {
+ return nil, fmt.Errorf("the timers snapshot digests to %s, the record says %s", got, wantDigest)
+ }
+ for _, r := range snap.Rows {
+ if r.Host != l.rec.Host {
+ return nil, fmt.Errorf("timer %q is %s's, not %s's", r.Name, r.Host, l.rec.Host)
+ }
+ }
+ return snap.Rows, nil
+}
+
+// configured (§7.6, §9.1, §9.2): the held row rendered (RenderHeld, the
+// move-bound ack), its bundle and overlay etags required to be the ones
+// pinned at readiness (else the rollback marker, "configuration changed
+// during the move"); the pinned release live and proven for the held row
+// through Plan 1's T7 hook (ruling G4: "waiting for the root stage pass"
+// waits, anything else waits until landing_deadline marks the rollback);
+// then the install the brief lists under `restored`: the built database's
+// sha256 against the manifest, lease.CheckDB, the rename into DBPath (G34),
+// <dbPath>.gen through hostbox's marker writer naming the prepared
+// generation (ruling G2), and the CAS to restored stamping
+// landing.installed_db_sha256.
+//
+// A replay finds configured and does it all again: the render and the
+// release check are idempotent reads and writes of the same bytes; a
+// database already at DBPath with the manifest's sha256 is this landing's
+// own install, so only the marker is written again.
+func (l *landingRun) configured(ctx context.Context) (bool, error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := l.requirePhase(obs, provision.LandConfigured); err != nil {
+ return false, err
+ }
+ in := l.rec.Move.Inputs
+ if in == nil {
+ return false, fmt.Errorf("move %s: the record has no pinned inputs", id)
+ }
+ rt, err := l.route()
+ if err != nil {
+ return false, err
+ }
+ hr, err := cfg.RenderHeld(ctx, l.s, rt, id)
+ if err != nil {
+ return false, fmt.Errorf("render the held row: %w", err)
+ }
+ if !sameBundle(hr.BundleETags, in.BundleETags) || hr.OverlayETag != in.OverlayETag {
+ log.Printf("moves: %s: the render's etags (%v, overlay %q) are not the pinned ones (%v, overlay %q)", host, hr.BundleETags, hr.OverlayETag, in.BundleETags, in.OverlayETag)
+ return l.markRollback(ctx, provision.RollbackConfigChanged)
+ }
+ if c := cfg.ReleaseReady(ctx, l.s, l.rec, in.Release); !c.OK {
+ l.w.notice(host+"#release", "moves: %s: release %s is not live and proven for the held row yet (%s); retrying until the landing's deadline", host, in.Release, c.Reason)
+ return false, nil
+ }
+ if err := l.fault("after-release-ready"); err != nil {
+ return false, err
+ }
+ sha, installed, err := l.restore(ctx, obs)
+ if err != nil || !installed {
+ return installed, err
+ }
+ return l.advance(ctx, provision.LandConfigured, provision.LandRestored, func(m *provision.MoveState) {
+ if m.Landing == nil {
+ m.Landing = &provision.MoveLandingJournal{}
+ }
+ m.Landing.InstalledDBSHA256 = sha
+ })
+}
+
+// sameBundle compares two bundle etag maps.
+func sameBundle(a, b map[string]string) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ for k, v := range a {
+ if bv, ok := b[k]; !ok || bv != v {
+ return false
+ }
+ }
+ return true
+}
+
+// restore is the install of the built database (§7.6 restored, R6-3):
+// its sha256 must be the manifest's and CheckDB must pass, else the
+// rollback marker; then the rename into DBPath and the generation marker.
+// installed is false when the phase ended with the marker written. The
+// returned sha carries the record's "sha256:" prefix, as Source.DBSHA256
+// does.
+func (l *landingRun) restore(ctx context.Context, obs hostclaim.Observed) (sha string, installed bool, err error) {
+ cfg := l.cfg()
+ host, id := l.rec.Host, l.g.ID
+ m, err := l.installManifest(ctx, obs)
+ if err != nil {
+ return "", false, err
+ }
+ want := "sha256:" + m.DBSHA256
+ dbPath := l.dbPath()
+ present, err := regularFile(dbPath)
+ if err != nil {
+ return "", false, err
+ }
+ if present {
+ got, err := fileSHA256(dbPath)
+ if err != nil {
+ return "", false, err
+ }
+ if got != want {
+ return "", false, fmt.Errorf("%s holds a database that is not this landing's build (%s, want %s)", dbPath, got, want)
+ }
+ } else {
+ built := filepath.Join(l.buildDir(), host+".db")
+ if ok, err := regularFile(built); err != nil {
+ return "", false, err
+ } else if !ok {
+ // The scratch was lost (a rebuilt data directory): build again
+ // from the pinned set, which reproduces the same bytes.
+ if _, err := l.rebuild(ctx, obs); err != nil {
+ return "", false, err
+ }
+ }
+ got, err := fileSHA256(built)
+ if err != nil {
+ return "", false, err
+ }
+ if got != want {
+ log.Printf("moves: %s: the built database is %s, the manifest says %s", host, got, want)
+ ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
+ return "", ok, err
+ }
+ if err := lease.CheckDB(built); err != nil {
+ log.Printf("moves: %s: the built database does not pass its check: %v", host, err)
+ ok, err := l.markRollback(ctx, provision.RollbackInstallMismatch)
+ return "", ok, err
+ }
+ for _, side := range []string{"-wal", "-shm"} {
+ os.Remove(built + side)
+ }
+ if err := os.Rename(built, dbPath); err != nil {
+ return "", false, fmt.Errorf("install the database: %w", err)
+ }
+ if err := syncPath(filepath.Dir(dbPath)); err != nil {
+ return "", false, err
+ }
+ }
+ if err := l.fault("after-db-rename"); err != nil {
+ return "", false, err
+ }
+ if err := cfg.WriteGen(dbPath+".gen", id); err != nil {
+ return "", false, fmt.Errorf("write the generation marker: %w", err)
+ }
+ if err := syncPath(filepath.Dir(dbPath)); err != nil {
+ return "", false, err
+ }
+ log.Printf("moves: %s: database installed at %s (%s), generation %s", host, dbPath, want, id)
+ return want, true, nil
+}
+
+// installManifest is the journal's, or the build's again when the journal
+// lost it.
+func (l *landingRun) installManifest(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+ j, err := l.journal()
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ var m moves.InstallManifest
+ if raw, ok := j.Steps[landStepInstall]; ok && json.Unmarshal([]byte(raw), &m) == nil && m.Op == l.g.ID && m.Host == l.rec.Host && m.DBSHA256 != "" {
+ return m, nil
+ }
+ return l.rebuild(ctx, obs)
+}
+
+// rebuild runs BuildGeneration again over the seal: a replay that reuses
+// every object it finds and restores the same bytes into the scratch.
+func (l *landingRun) rebuild(ctx context.Context, obs hostclaim.Observed) (moves.InstallManifest, error) {
+ cfg := l.cfg()
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ vs, err := l.versioned(ctx)
+ if err != nil {
+ return moves.InstallManifest{}, err
+ }
+ holder := *obs.Claim.Holder
+ m, err := moves.BuildGeneration(ctx, moves.BuildInputs{Store: vs, Bucket: cfg.OwnBucket, Source: seal.Replica, N: seal.N,
+ GenerationPrefix: moves.GenerationPrefix(holder.Location, l.g.ID), Op: l.g.ID, Host: l.rec.Host, WorkDir: l.buildDir(), Now: cfg.Now})
+ if err != nil {
+ return moves.InstallManifest{}, fmt.Errorf("rebuild the generation: %w", err)
+ }
+ j, err := l.journal()
+ if err != nil {
+ return m, err
+ }
+ body, err := json.Marshal(m)
+ if err != nil {
+ return m, err
+ }
+ j.Mark(landStepInstall, string(body))
+ return m, j.Write(cfg.DataDir)
+}
+
+// writeFileAtomic writes body to path through a temporary file in the same
+// directory, synced and renamed, then syncs the directory.
+func writeFileAtomic(path string, body []byte, mode os.FileMode) error {
+ dir := filepath.Dir(path)
+ f, err := os.CreateTemp(dir, "."+filepath.Base(path)+".*")
+ if err != nil {
+ return err
+ }
+ tmp := f.Name()
+ if _, err := f.Write(body); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Chmod(mode); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Sync(); err != nil {
+ f.Close()
+ os.Remove(tmp)
+ return err
+ }
+ if err := f.Close(); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ if err := os.Rename(tmp, path); err != nil {
+ os.Remove(tmp)
+ return err
+ }
+ return syncPath(dir)
+}
+
+// syncPath fsyncs a file or directory.
+func syncPath(path string) error {
+ f, err := os.Open(path)
+ if err != nil {
+ return err
+ }
+ defer f.Close()
+ return f.Sync()
+}
diff --git a/internal/host/movesland_test.go b/internal/host/movesland_test.go
new file mode 100644
index 000000000..66c2bbe50
--- /dev/null
+++ b/internal/host/movesland_test.go
@@ -0,0 +1,1243 @@
+package host
+
+import (
+ "context"
+ "database/sql"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log/slog"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "sync"
+ "testing"
+ "time"
+
+ "github.com/benbjohnson/litestream"
+ lsfile "github.com/benbjohnson/litestream/file"
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/dbverify"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
+)
+
+// The landing harness: the cut harness with a REAL litestream replica in
+// the source's lineage (so the target's build restores a real SQLite file
+// that passes its check), driven to the flip, and a target box (uk-box)
+// with its own data directory, the held row reconcile would have minted,
+// the landing's hooks faked where main has no production provider yet
+// (ruling G17) and real where it has one: park.Local and hostguard.Guard
+// over the target's data directory, dbverify over the store, the hostbox
+// marker writer, moves.BuildGeneration over the fake versioned bucket.
+//
+// fakeParkLeases is one lease table for the landing's own-location lease
+// and the park engine's: a second acquisition of a held location is
+// refused, so a lease the landing still held while a residue park ran
+// would fail the test (round-2 finding 9).
+
+type landingHarness struct {
+ *cutHarness
+ tdir string // the target's data directory
+ leases *fakeParkLeases
+ lw *movesWorker
+ lcfg moveCfg
+ route registry.Route
+ rows int // rows the source database holds
+ n uint64
+ replaced [][]registry.Timer
+ repl []string
+ nextDue []int64
+ renders int
+ render func() (appconfig.HeldRender, error)
+ release moves.Check
+ landAt string
+ landHits []string
+ landDone bool
+ parkAt string
+ parkHits []string
+ parkDone bool
+ parkView func(context.Context, hostclaim.Observed) (park.View, error)
+ landFault func(point string) error
+}
+
+// fakeParkLeases refuses a second acquisition of a held location.
+type fakeParkLeases struct {
+ mu sync.Mutex
+ held map[string]bool
+ acquired []string
+}
+
+type fakeHeldLease struct {
+ l *fakeParkLeases
+ key string
+}
+
+func (l *fakeParkLeases) Acquire(_ context.Context, bucket, scope string) (park.HeldLease, error) {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ key := bucket + "/" + scope
+ if l.held[key] {
+ return nil, &lease.HeldError{Owner: "the landing"}
+ }
+ if l.held == nil {
+ l.held = map[string]bool{}
+ }
+ l.held[key] = true
+ l.acquired = append(l.acquired, key)
+ return &fakeHeldLease{l: l, key: key}, nil
+}
+
+func (l *fakeParkLeases) holding() []string {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ var out []string
+ for k, on := range l.held {
+ if on {
+ out = append(out, k)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+func (h *fakeHeldLease) Check(context.Context) error {
+ h.l.mu.Lock()
+ defer h.l.mu.Unlock()
+ if !h.l.held[h.key] {
+ return errors.New("fake lease: released")
+ }
+ return nil
+}
+func (h *fakeHeldLease) Release(context.Context) error {
+ h.l.mu.Lock()
+ defer h.l.mu.Unlock()
+ delete(h.l.held, h.key)
+ return nil
+}
+func (h *fakeHeldLease) Close() { _ = h.Release(context.Background()) }
+
+// realReplica writes a small SQLite database with rows rows over syncs
+// litestream syncs (a level-9 snapshot too) and returns the file replica
+// client and the newest level-0 position, as Task 8's test does.
+func realReplica(t *testing.T, syncs int) (litestream.ReplicaClient, uint64, int) {
+ t.Helper()
+ dir := t.TempDir()
+ path := filepath.Join(dir, "src.db")
+ rows := 0
+ write := func(n int) {
+ d, err := sql.Open("sqlite", path)
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.Close()
+ if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
+ t.Fatal(err)
+ }
+ for i := 0; i < n; i++ {
+ if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ rows += n
+ }
+ write(1)
+ client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
+ db := litestream.NewDB(path)
+ db.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
+ db.SetMetaPath(filepath.Join(dir, "meta"))
+ db.MonitorInterval = time.Hour
+ db.Replica = litestream.NewReplicaWithClient(db, client)
+ db.Replica.SyncInterval = time.Hour
+ if err := db.Open(); err != nil {
+ t.Fatal(err)
+ }
+ ctx := context.Background()
+ for i := 0; i < syncs; i++ {
+ if err := db.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ write(2)
+ }
+ if err := db.SyncAndWait(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := db.Snapshot(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if err := db.Close(ctx); err != nil {
+ t.Fatal(err)
+ }
+ var n uint64
+ for _, f := range replicaLevelFiles(t, client, 0) {
+ if uint64(f.MaxTXID) > n {
+ n = uint64(f.MaxTXID)
+ }
+ }
+ if n == 0 {
+ t.Fatal("the replica has no level-0 files")
+ }
+ return client, n, rows
+}
+
+func replicaLevelFiles(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo {
+ t.Helper()
+ itr, err := client.LTXFiles(context.Background(), level, 0, false)
+ if err != nil {
+ t.Fatal(err)
+ }
+ infos, err := ltx.SliceFileIterator(itr)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return infos
+}
+
+func countRowsForTest(t *testing.T, path string) int {
+ t.Helper()
+ d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer d.Close()
+ var n int
+ if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
+ t.Fatal(err)
+ }
+ return n
+}
+
+func newLandingHarness(t *testing.T) *landingHarness {
+ t.Helper()
+ h := &landingHarness{cutHarness: newCutHarness(t), leases: &fakeParkLeases{}, release: moves.Check{OK: true}}
+ // The real replica replaces the harness's fake level files under the
+ // source's lineage; the final checkpoint adds nothing (the database is
+ // quiet at N).
+ client, n, rows := realReplica(t, 2)
+ h.n, h.rows = n, rows
+ loc := "instances/" + p1Host + "/" + h.inc + "/"
+ for k := range h.f.buckets["data-ie"] {
+ if strings.HasPrefix(k, loc+"g-cur/") && strings.HasSuffix(k, ".ltx") {
+ delete(h.f.buckets["data-ie"], k)
+ }
+ }
+ for level := 0; level <= litestream.SnapshotLevel; level++ {
+ for _, info := range replicaLevelFiles(t, client, level) {
+ rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ body, err := io.ReadAll(rc)
+ rc.Close()
+ if err != nil {
+ t.Fatal(err)
+ }
+ key := loc + "g-cur/" + fmt.Sprintf("%04x/", level) + ltx.FormatFilename(info.MinTXID, info.MaxTXID)
+ h.f.store("data-ie", key, fakeMoveObject{lease.MirrorObject{ETag: "r-" + filepath.Base(key), Size: int64(len(body))}, string(body)})
+ }
+ }
+ h.localPos = n
+ h.cfg.FinalCheckpoint = func(ctx context.Context, _, _ string, _ hostclaim.Holder, check func(context.Context) error) error {
+ return check(ctx)
+ }
+ // The target's data directory, with the two directories the guard and
+ // park.Local need, and the held row reconcile minted.
+ h.tdir = t.TempDir()
+ for _, d := range []string{"hostguard", ".parked"} {
+ if err := os.Mkdir(filepath.Join(h.tdir, d), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ h.route = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "landing", HoldOp: p1ID,
+ Socket: provision.DefaultSocket(p1Host), DBPath: provision.DefaultDBPath(h.tdir, p1Host), Provisioned: true}
+ h.render = func() (appconfig.HeldRender, error) {
+ in := h.read(t, p1Host).Move.Inputs
+ if in == nil {
+ return appconfig.HeldRender{}, errors.New("no pinned inputs")
+ }
+ return appconfig.HeldRender{Path: filepath.Join(h.tdir, p1Host+".json"), SHA256: "x", BundleETags: in.BundleETags, OverlayETag: in.OverlayETag}, nil
+ }
+ // The restore-verification record names the host: local removes it.
+ if err := dbverify.Publish(context.Background(), h.s, &dbverify.Record{V: 1, Box: "uk-box", CheckedAt: h.clock.Now(),
+ Hosts: map[string]dbverify.Verdict{p1Host: {OK: true, VerifiedAt: h.clock.Now()}, "other.example": {OK: true, VerifiedAt: h.clock.Now()}}}); err != nil {
+ t.Fatal(err)
+ }
+ h.lcfg = h.landingCfg(t)
+ return h
+}
+
+// landingCfg is the target's configuration with the landing armed.
+func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
+ t.Helper()
+ c := h.target()
+ c.DataDir = h.tdir
+ c.OwnsRoute = func(host string) bool { return host == p1Host }
+ c.Route = func(host string) (*registry.Route, error) {
+ if host != p1Host {
+ return nil, nil
+ }
+ rt := h.route
+ return &rt, nil
+ }
+ c.FilesDir = func(host string) (string, string) {
+ return filepath.Join(h.tdir, host), filepath.Join(h.tdir, host+".env")
+ }
+ c.Guard = hostguard.Guard{DataDir: h.tdir, LockDir: t.TempDir()}
+ c.Local = park.Local{DataDir: h.tdir, Box: "uk-box"}
+ c.ParkLeases = h.leases
+ c.ParkView = func(ctx context.Context, obs hostclaim.Observed) (park.View, error) {
+ if h.parkView == nil {
+ return park.View{}, errors.New("no view override")
+ }
+ return h.parkView(ctx, obs)
+ }
+ if h.parkView == nil {
+ c.ParkView = nil
+ }
+ c.OwnLease = h.leases.Acquire
+ c.RenderHeld = func(context.Context, release.Store, registry.Route, string) (appconfig.HeldRender, error) {
+ h.renders++
+ return h.render()
+ }
+ c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check { return h.release }
+ c.ReplaceTimers = func(host string, rows []registry.Timer) error {
+ h.replaced = append(h.replaced, append([]registry.Timer(nil), rows...))
+ return nil
+ }
+ c.SetReplication = func(host, class string) error { h.repl = append(h.repl, class); return nil }
+ c.SetNextDue = func(host string, due int64) error { h.nextDue = append(h.nextDue, due); return nil }
+ c.RemoveVerifyEntry = dbverify.RemoveHost
+ c.WriteGen = hostbox.WriteGenerationMarker
+ c.landFault = func(point string) error {
+ h.landHits = append(h.landHits, point)
+ if h.landFault != nil {
+ if err := h.landFault(point); err != nil {
+ return err
+ }
+ }
+ if point == h.landAt && !h.landDone {
+ h.landDone = true
+ return errCrash
+ }
+ return nil
+ }
+ c.parkFault = func(step string) error {
+ h.parkHits = append(h.parkHits, step)
+ if step == h.parkAt && !h.parkDone {
+ h.parkDone = true
+ return errCrash
+ }
+ return nil
+ }
+ return c
+}
+
+// flip drives the cut to the flip: the record landing/checking in uk and
+// the claim landing/checking under uk-box.
+func (h *landingHarness) flip(t *testing.T) {
+ t.Helper()
+ h.ready(t)
+ h.cut(t)
+ h.pass(t)
+ if rec := h.read(t, p1Host); rec.Region != "uk" || rec.Move.State != provision.MoveLanding || rec.Move.LandingPhase != provision.LandChecking {
+ t.Fatalf("the fixture did not flip: %s %+v", rec.Region, rec.Move)
+ }
+}
+
+// land runs one pass as the target box, the worker kept across passes.
+func (h *landingHarness) land(t *testing.T) {
+ t.Helper()
+ if h.lw == nil {
+ h.lw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
+ }
+ h.lw.cfg = h.landingCfg(t)
+ if err := movesPass(context.Background(), h.s, h.lw); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// crashLand runs one target pass that must reach the landing crash point.
+func (h *landingHarness) crashLand(t *testing.T) {
+ t.Helper()
+ h.land(t)
+ if !h.landDone {
+ t.Fatalf("the pass never reached %s (points reached: %v)", h.landAt, h.landHits)
+ }
+}
+
+// freshTarget drops the target worker, as a restarted agent would, and
+// with it every lease the dead process held: a real lease lapses at its
+// TTL once nobody renews it, and the restarted worker acquires afresh.
+func (h *landingHarness) freshTarget() {
+ h.lw = nil
+ h.leases.expire()
+}
+
+// expire lapses every held lease, as the TTL does for a dead holder.
+func (l *fakeParkLeases) expire() {
+ l.mu.Lock()
+ defer l.mu.Unlock()
+ l.held = map[string]bool{}
+}
+
+func (h *landingHarness) landJournal(t *testing.T) *moves.Journal {
+ t.Helper()
+ j, err := moves.ReadJournal(h.tdir, p1Host, p1ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return j
+}
+
+func (h *landingHarness) filesGen(t *testing.T) (filesGenMarker, bool) {
+ t.Helper()
+ body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, archive.MarkerName))
+ if errors.Is(err, os.ErrNotExist) {
+ return filesGenMarker{}, false
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ var m filesGenMarker
+ if err := json.Unmarshal(body, &m); err != nil {
+ t.Fatal(err)
+ }
+ return m, true
+}
+
+// installed asserts the end state of a landing through restored.
+func (h *landingHarness) installed(t *testing.T) moves.InstallManifest {
+ t.Helper()
+ rec := h.read(t, p1Host)
+ m := rec.Move
+ if m.State != provision.MoveLanding || m.LandingPhase != provision.LandRestored || m.Reason != "" {
+ t.Fatalf("record: %+v", m)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != provision.LandRestored || c.Op.Exec != "uk-box" || !containsString(c.Touched.Boxes, "uk-box") {
+ t.Fatalf("claim: %+v touched %+v", c.Op, c.Touched)
+ }
+ j := h.landJournal(t)
+ var im moves.InstallManifest
+ if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &im); err != nil || im.Op != p1ID || im.N != h.n {
+ t.Fatalf("install manifest in the journal: %v %+v", err, im)
+ }
+ if m.Landing == nil || m.Landing.InstalledDBSHA256 != "sha256:"+im.DBSHA256 {
+ t.Fatalf("installed_db_sha256: %+v want %s", m.Landing, im.DBSHA256)
+ }
+ dbPath := provision.DefaultDBPath(h.tdir, p1Host)
+ if got, err := fileSHA256(dbPath); err != nil || got != "sha256:"+im.DBSHA256 {
+ t.Fatalf("the installed database: %v %s want %s", err, got, im.DBSHA256)
+ }
+ if got := countRowsForTest(t, dbPath); got != h.rows {
+ t.Fatalf("the installed database holds %d rows, the source wrote %d", got, h.rows)
+ }
+ if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != p1ID {
+ t.Fatalf("%s.gen: %q %v", dbPath, gen, err)
+ }
+ if fg, ok := h.filesGen(t); !ok || fg.Op != p1ID || fg.Incarnation != h.inc {
+ t.Fatalf(".files-gen: %+v present=%v", fg, ok)
+ }
+ if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil || string(body) != "hello" {
+ t.Fatalf("the extracted files: %q %v", body, err)
+ }
+ if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host+".env")); err != nil || string(body) != "APP_MODE=prod\n" {
+ t.Fatalf("the extracted env: %q %v", body, err)
+ }
+ if held := h.leases.holding(); len(held) != 0 {
+ t.Fatalf("the landing still holds %v", held)
+ }
+ if _, err := os.Stat(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build", p1Host+".db")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the build scratch was not renamed away: %v", err)
+ }
+ return im
+}
+
+// The end to end landing through restored: the generation built from the
+// seal, the archive extracted, timers replaced with their registrations,
+// the carried columns set, the verify entry gone, .files-gen and
+// <dbPath>.gen written, installed_db_sha256 recorded, the lease released.
+func TestLandingBuildsInstallsAndRestores(t *testing.T) {
+ ctx := context.Background()
+ h := newLandingHarness(t)
+ h.timers = []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000, Attempts: 2, Registration: "reg-1"},
+ {Host: p1Host, Name: "once", Kind: "once", NextAt: 2000, DueAt: 2000, Registration: "reg-2"}}
+ h.flip(t)
+ h.route.Replication = ""
+ h.land(t)
+ im := h.installed(t)
+ // The generation holds exactly the build's four objects, at usable
+ // versions, in this box's own bucket.
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ var keys []string
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ keys = append(keys, strings.TrimPrefix(k, prefix))
+ }
+ }
+ sort.Strings(keys)
+ want := []string{"0000/" + ltx.FormatFilename(ltx.TXID(h.n), ltx.TXID(h.n)), "0009/" + ltx.FormatFilename(1, ltx.TXID(h.n)), "build-intent.json", "install.json"}
+ if strings.Join(keys, " ") != strings.Join(want, " ") {
+ t.Fatalf("generation objects %v, want %v", keys, want)
+ }
+ if !lease.UsableVersionID(im.SnapshotVersion) || !lease.UsableVersionID(im.AnchorVersion) {
+ t.Fatalf("manifest versions %q %q", im.SnapshotVersion, im.AnchorVersion)
+ }
+ // Timers: the sealed snapshot's rows, registrations included, once.
+ if len(h.replaced) != 1 || len(h.replaced[0]) != 2 {
+ t.Fatalf("ReplaceTimers calls: %+v", h.replaced)
+ }
+ if moves.TimersDigest(h.replaced[0]) != h.read(t, p1Host).Move.Source.TimersSHA256 {
+ t.Fatalf("the replaced rows do not digest to the record's: %+v", h.replaced[0])
+ }
+ byName := map[string]registry.Timer{}
+ for _, r := range h.replaced[0] {
+ byName[r.Name] = r
+ }
+ if byName["nightly"].Registration != "reg-1" || byName["nightly"].Attempts != 2 || byName["once"].Registration != "reg-2" || byName["once"].DueAt != 2000 {
+ t.Fatalf("registrations and bookkeeping not preserved: %+v", h.replaced[0])
+ }
+ src := h.read(t, p1Host).Move.Source
+ if len(h.repl) != 1 || h.repl[0] != src.Replication || len(h.nextDue) != 1 || h.nextDue[0] != src.NextDue {
+ t.Fatalf("columns: replication %v next_due %v, source %+v", h.repl, h.nextDue, src)
+ }
+ dv, err := dbverify.Load(ctx, h.s)
+ if err != nil || dv == nil {
+ t.Fatal(err)
+ }
+ if _, ok := dv.Hosts[p1Host]; ok || len(dv.Hosts) != 1 {
+ t.Fatalf("the verify entry: %+v", dv.Hosts)
+ }
+ // The inventory names the four flat paths under F, the flip's stamp.
+ c := h.claim(t)
+ inv := c.Inventory["uk-box"]
+ if inv.Layout != "flat" || len(inv.Paths) != 4 {
+ t.Fatalf("inventory: %+v", inv)
+ }
+ for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host, p1Host + ".env"} {
+ if inv.Paths[p] != c.Holder.Stamp {
+ t.Fatalf("inventory path %s at %d, want F %d", p, inv.Paths[p], c.Holder.Stamp)
+ }
+ }
+ captured, err := hostguard.Capture(h.tdir, p1Host)
+ if err != nil || captured.Stamp != c.Holder.Stamp || captured.Incarnation != h.inc {
+ t.Fatalf("the installation record: %+v %v, want stamp %d", captured, err, c.Holder.Stamp)
+ }
+ // No residue, so no park chores; and no park-local (no earlier stay).
+ if len(c.Obligations) != 0 {
+ t.Fatalf("obligations: %+v", c.Obligations)
+ }
+ // The journal records every sub-step, the prepared manifest as skipped.
+ j := h.landJournal(t)
+ for _, step := range []string{landStepResidue, landStepInstall, landStepPublished, landStepRetired, landStepExtracted, landStepInventory, landStepTimers, landStepColumns, landStepVerify, landStepFilesManifest, landStepFilesGen} {
+ if !j.Done(step) {
+ t.Fatalf("journal lacks %s: %v", step, j.Steps)
+ }
+ }
+ if h.renders != 1 {
+ t.Fatalf("renders %d", h.renders)
+ }
+ // Restored is Plan 3's: another pass changes nothing.
+ before := h.read(t, p1Host).Move.UpdatedAt
+ h.land(t)
+ if after := h.read(t, p1Host).Move; after.LandingPhase != provision.LandRestored || !after.UpdatedAt.Equal(before) {
+ t.Fatalf("a pass at restored wrote the record: %+v", after)
+ }
+}
+
+// checking (§7.6): the claim must be this move's landing op with this box
+// executing, citing complete source evidence, and this box is added to
+// touched.boxes before any local write; the installation record is
+// activated at F. A claim the source left at moving/sealed (a crash after
+// the record's flip) is flipped by the target itself (ruling G12).
+func TestLandingCheckingRequiresClaimEvidenceAndTouch(t *testing.T) {
+ ctx := context.Background()
+ t.Run("the target writes the flip's claim half", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.ready(t)
+ h.crashAt = "after-record-flip"
+ h.cut(t)
+ h.crashPass(t)
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseSealed {
+ t.Fatalf("fixture: %+v", c.Op)
+ }
+ h.landAt = "after-target-flip"
+ h.crashLand(t)
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != provision.LandChecking || c.Op.Exec != "uk-box" {
+ t.Fatalf("after the target's flip: %+v", c.Op)
+ }
+ if c.Op.Journal[hostclaim.JournalSourceStamp] != "1" || c.Op.Journal[hostclaim.JournalSrcGen] != "cur" {
+ t.Fatalf("the target's flip did not journal the source's stamp and generation: %+v", c.Op.Journal)
+ }
+ h.land(t)
+ h.installed(t)
+ })
+ t.Run("another executor is refused", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ // Refused before any claim write: with uk-box taken out of touched,
+ // the K12 Touch that follows the executor check must not run (the
+ // guard would refuse the activation too, but later and locally).
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "other-box"; c.Touched.Boxes = []string{"ie-box"} })
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
+ t.Fatalf("advanced under another executor: %+v", m)
+ }
+ if _, err := hostguard.Capture(h.tdir, p1Host); !errors.Is(err, hostguard.ErrNoRecord) {
+ t.Fatalf("the installation record was written: %v", err)
+ }
+ if c := h.claim(t); containsString(c.Touched.Boxes, "uk-box") {
+ t.Fatal("a claim write under another executor")
+ }
+ })
+ t.Run("missing evidence is refused", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ if err := h.s.Delete(ctx, h.stoppedKey()); err != nil {
+ t.Fatal(err)
+ }
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
+ t.Fatalf("advanced without the source's evidence: %+v", m)
+ }
+ })
+ t.Run("touch and the installation record", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ // Plan 1's target report touched uk-box at its first staging write;
+ // checking's Touch is the idempotent K12 re-add, so take it away
+ // first to see it written.
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Touched.Boxes = []string{"ie-box"} })
+ h.landAt = "after-record-parking"
+ h.crashLand(t)
+ c := h.claim(t)
+ if !containsString(c.Touched.Boxes, "uk-box") || !containsString(c.Touched.Regions, "ie") {
+ t.Fatalf("touched: %+v", c.Touched)
+ }
+ captured, err := hostguard.Capture(h.tdir, p1Host)
+ if err != nil || captured.Stamp != c.Holder.Stamp || captured.Incarnation != h.inc {
+ t.Fatalf("active_stamp %+v %v, want F %d", captured, err, c.Holder.Stamp)
+ }
+ if len(h.leases.acquired) != 0 {
+ t.Fatalf("checking took a lease: %v", h.leases.acquired)
+ }
+ })
+ t.Run("a nested row is refused for a rollback", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.route.DBPath = filepath.Join(h.tdir, "instances", p1Host, "db", "instance.db")
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackLandingLayout || m.LandingPhase != provision.LandChecking || m.State != provision.MoveLanding {
+ t.Fatalf("a nested row: %+v", m)
+ }
+ })
+ t.Run("host claims on refuse at checking", func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ t.Setenv("CARLOS_HOST_CLAIM", "1")
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackClaimWired || m.LandingPhase != provision.LandChecking {
+ t.Fatalf("under CARLOS_HOST_CLAIM: %+v", m)
+ }
+ })
+}
+
+// residue plants an earlier stay of this incarnation at the holder's
+// location in the target's bucket: two generations (old and the origin)
+// and a generation.json, each at a usable version.
+func (h *landingHarness) residue(t *testing.T) []string {
+ t.Helper()
+ ctx := context.Background()
+ keys := []string{
+ cutRoot + "/g-old/0000/0000000000000001-0000000000000001.ltx",
+ cutRoot + "/g-old/0009/0000000000000001-0000000000000001.ltx",
+ cutRoot + "/0000/0000000000000001-0000000000000001.ltx",
+ cutRoot + "/generation.json",
+ }
+ for _, k := range keys {
+ if _, err := h.vs.PutIfAbsent(ctx, "data-uk", k, []byte("residue "+k)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ return keys
+}
+
+// parking (§7.6, D6): an earlier stay's objects under holder.location are
+// archived under parked/<inc>/<host>/<ret>/instances/... by one
+// park-generation chore per generation before building; a crash inside the
+// engine leaves the phase at parking and the replay finishes; the chores
+// stay pending until the horizon; after the probe makes the new generation
+// live here, Engine.Run on them acknowledges rather than refusing.
+func TestLandingParkingWaitsForTheManifest(t *testing.T) {
+ ctx := context.Background()
+ h := newLandingHarness(t)
+ h.flip(t)
+ keys := h.residue(t)
+ h.parkAt = "copy"
+ h.land(t)
+ if !h.parkDone {
+ t.Fatalf("the engine never copied (%v)", h.parkHits)
+ }
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
+ t.Fatalf("after the crash: %+v", m)
+ }
+ c := h.claim(t)
+ var gens []string
+ for _, o := range c.Obligations {
+ if o.Op == p1ID && o.Kind == "park-generation" && o.Region == "uk" && o.Park != nil && o.Park.Location == cutRoot && o.Park.EvidenceAt.Equal(h.read(t, p1Host).Move.FlippedAt.UTC()) {
+ gens = append(gens, o.Generation)
+ }
+ }
+ sort.Strings(gens)
+ if strings.Join(gens, " ") != hostclaim.RootGeneration+" old" {
+ t.Fatalf("residue chores: %v (%+v)", gens, c.Obligations)
+ }
+ if len(h.leases.acquired) == 0 {
+ t.Fatal("the engine took no lease")
+ }
+ // The replay archives and writes the manifest, but one delete the store
+ // accepts does not land (its listing still shows the version): the
+ // manifest exists and its deletion set is not gone, so Parked is false
+ // and the phase waits, chores queued, nothing built.
+ stuck := keys[0]
+ h.vs.skipDelete = func(key string) bool { return key == stuck }
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
+ t.Fatalf("advanced with a deletion outstanding: %+v", m)
+ }
+ if _, ok := h.f.buckets["data-uk"][stuck]; !ok {
+ t.Fatalf("fixture: %s was deleted", stuck)
+ }
+ manifests := 0
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") && strings.HasSuffix(k, "/manifest.json") {
+ manifests++
+ }
+ }
+ if manifests != 2 {
+ t.Fatalf("%d park manifests after the second pass", manifests)
+ }
+ if n := len(h.vs.callsOn(moves.GenerationPrefix(cutRoot, p1ID))); n != 0 {
+ t.Fatalf("the build ran while parking waited: %d calls", n)
+ }
+ // The deletes land: the deletion set is gone, the landing goes on to
+ // restored. The chores are still owed.
+ h.vs.skipDelete = nil
+ h.land(t)
+ h.installed(t)
+ c = h.claim(t)
+ var retirements []string
+ for _, o := range c.Obligations {
+ if o.Kind == "park-generation" {
+ retirements = append(retirements, o.Retirement)
+ }
+ }
+ if len(retirements) != 2 {
+ t.Fatalf("chores after the landing: %+v", c.Obligations)
+ }
+ // The marker is the location's one reused key: it is not parked (the
+ // probe supersedes it), every other object is.
+ marker := cutRoot + "/generation.json"
+ if _, ok := h.f.buckets["data-uk"][marker]; !ok {
+ t.Fatal("the earlier stay's generation.json was parked")
+ }
+ for _, k := range keys {
+ if _, ok := h.f.buckets["data-uk"][k]; ok && k != marker {
+ t.Fatalf("%s is still at the location", k)
+ }
+ }
+ var archived []string
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") {
+ archived = append(archived, k)
+ }
+ }
+ sort.Strings(archived)
+ if len(archived) != len(keys)-1+2 { // every object but the marker, plus two manifests
+ t.Fatalf("archived: %v", archived)
+ }
+ for _, k := range keys {
+ if k == marker {
+ continue
+ }
+ rel := strings.TrimPrefix(k, "instances/")
+ found := false
+ for _, a := range archived {
+ if strings.Contains(a, "/instances/"+strings.TrimPrefix(rel, p1Host+"/")+"@") {
+ found = true
+ }
+ }
+ if !found {
+ t.Fatalf("%s was not archived under parked/<inc>/<host>/<ret>/instances/ (%v)", k, archived)
+ }
+ }
+ // The probe's admission (Plan 3) makes g-<move id> live here: the
+ // holder's generation is the move id, the marker names it and pins the
+ // snapshot, and uk is authoritative. Past the horizon the chores
+ // acknowledge: their generations are not the live one.
+ j := h.landJournal(t)
+ var im moves.InstallManifest
+ if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &im); err != nil {
+ t.Fatal(err)
+ }
+ // The probe writes the marker If-Match over the earlier stay's: a new
+ // current version at the same key.
+ markerVer := h.f.nextVersion()
+ h.f.store("data-uk", marker, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + markerVer, Size: 20}, `{"op":"` + p1ID + `"}`})
+ h.f.setVersion("data-uk", marker, markerVer)
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = p1ID })
+ h.clock.Advance(park.SignatureAge + landingRequestLimit + time.Minute)
+ view := func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
+ return park.View{ClaimETag: obs.ETag, Region: "uk", AuthoritativeRegion: "uk", MarkerKnown: true, MarkerGeneration: p1ID,
+ Marker: park.Ref{Key: cutRoot + "/generation.json", VersionID: markerVer},
+ Pinned: map[park.Ref]bool{{Key: im.SnapshotKey, VersionID: im.SnapshotVersion}: true},
+ GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(cutRoot, r.Key) }}, nil
+ }
+ l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
+ eng := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID, View: view, Authorize: l.authorizeObligation}
+ for _, o := range c.Obligations {
+ if o.Kind != "park-generation" {
+ continue
+ }
+ if err := eng.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil {
+ t.Fatalf("acknowledge generation %s: %v", o.Generation, err)
+ }
+ }
+ if c := h.claim(t); len(c.Obligations) != 0 {
+ t.Fatalf("chores after the acknowledgement: %+v", c.Obligations)
+ }
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ n := 0
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ n++
+ }
+ }
+ if n != 4 {
+ t.Fatalf("the acknowledgement touched the live generation: %d objects left", n)
+ }
+ if held := h.leases.holding(); len(held) != 0 {
+ t.Fatalf("leases held after the acknowledgement: %v", held)
+ }
+}
+
+// Phase catch-up (§7.5, round-2 finding 11): a crash after the record's CAS
+// and before the claim's leaves the claim one step behind; the next pass
+// walks the claim forward and continues. A claim several steps behind is
+// walked step by step in one write.
+func TestLandingPhaseCatchUp(t *testing.T) {
+ ctx := context.Background()
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landAt = "after-record-parking"
+ h.crashLand(t)
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
+ t.Fatalf("record: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Phase != provision.LandChecking {
+ t.Fatalf("the claim moved with the record: %+v", c.Op)
+ }
+ h.freshTarget()
+ h.landAt = "after-residue-queued" // never reached: no residue
+ h.landDone = false
+ h.land(t)
+ h.installed(t)
+
+ // A claim left at checking with the record at restored.
+ rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandChecking })
+ l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
+ obs, err := l.syncClaimPhase(ctx)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if obs.Claim.Op.Phase != provision.LandRestored || h.claim(t).Op.Phase != provision.LandRestored {
+ t.Fatalf("after the catch-up: %+v", obs.Claim.Op)
+ }
+ // A claim ahead of the record is left alone.
+ rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandProbing })
+ if obs, err := l.syncClaimPhase(ctx); err != nil || obs.Claim.Op.Phase != provision.LandProbing {
+ t.Fatalf("a claim ahead: %v %+v", err, obs.Claim.Op)
+ }
+ // A claim at another kind is an error.
+ rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Kind = hostclaim.KindRestoring; op.Phase = "released" })
+ if _, err := l.syncClaimPhase(ctx); err == nil {
+ t.Fatal("a restoring claim was accepted")
+ }
+}
+
+// earlierStay plants an earlier stay of this incarnation on the target box:
+// the installation record at stamp 1, the inventory listing the four flat
+// paths at that stamp, and the files at the live paths.
+func (h *landingHarness) earlierStay(t *testing.T) {
+ t.Helper()
+ rec := fmt.Sprintf(`{"v":1,"host":%q,"incarnation":%q,"active_stamp":1,"retired_stamps":[]}`, p1Host, h.inc)
+ if err := os.WriteFile(filepath.Join(h.tdir, "hostguard", p1Host+".json"), []byte(rec), 0o660); err != nil {
+ t.Fatal(err)
+ }
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
+ c.Inventory = map[string]hostclaim.InventoryEntry{"uk-box": {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
+ })
+ if err := os.MkdirAll(filepath.Join(h.tdir, p1Host), 0o755); err != nil {
+ t.Fatal(err)
+ }
+ for name, body := range map[string]string{p1Host + ".db": "old database", p1Host + ".db.gen": "old\n", p1Host + ".env": "OLD=1\n", filepath.Join(p1Host, "old.txt"): "old files"} {
+ if err := os.WriteFile(filepath.Join(h.tdir, name), []byte(body), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ }
+}
+
+// local's order (round-2 finding 8, rulings G5 and G15): an earlier stay's
+// artifacts tagged S' are parked into .parked/ and gone from the live paths
+// before the new installation is registered under F; registering first is
+// what the real RecordInventory refuses. A park-local is queued at checking
+// only because the inventory has paths at S'.
+func TestLandingInventoryOrder(t *testing.T) {
+ ctx := context.Background()
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.earlierStay(t)
+ // Registering first is refused: with the record activated at F and the
+ // old paths still tagged 1, RecordInventory says ErrLocalInventory.
+ h.landAt = "after-activate"
+ h.crashLand(t)
+ c := h.claim(t)
+ if !hostclaim.InventoryHasStamp(c, "uk-box", 1) {
+ t.Fatal("fixture: no S' paths")
+ }
+ if j := h.landJournal(t); j.Steps[landStepPrevious] != "1" {
+ t.Fatalf("previous: %v", j.Steps)
+ }
+ captured, err := hostguard.Capture(h.tdir, p1Host)
+ if err != nil || captured.Stamp != c.Holder.Stamp {
+ t.Fatalf("captured %+v %v", captured, err)
+ }
+ local := park.Local{Guard: hostguard.Guard{DataDir: h.tdir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}, Claims: h.s, DataDir: h.tdir, Box: "uk-box"}
+ check := func(c *hostclaim.Claim) error { return nil }
+ if err := local.RecordInventory(ctx, p1Host, captured, "flat", []string{p1Host + ".db"}, check); !errors.Is(err, park.ErrLocalInventory) {
+ t.Fatalf("registering over S' paths: %v", err)
+ }
+ // The landing: the park-local chore is queued, local retires then registers.
+ h.land(t)
+ h.installed(t)
+ c = h.claim(t)
+ if len(c.Obligations) != 0 {
+ t.Fatalf("the park-local was not completed: %+v", c.Obligations)
+ }
+ inv := c.Inventory["uk-box"]
+ for p, s := range inv.Paths {
+ if s != c.Holder.Stamp {
+ t.Fatalf("inventory %s at %d, want F %d", p, s, c.Holder.Stamp)
+ }
+ }
+ // The old artifacts are under .parked/<inc>/<host>/<ret>/artifacts/, and
+ // the live paths hold the new installation.
+ var parked []string
+ filepath.WalkDir(filepath.Join(h.tdir, ".parked", h.inc, p1Host), func(p string, d os.DirEntry, err error) error {
+ if err == nil && !d.IsDir() && !strings.HasSuffix(p, ".journal.json") {
+ rel, _ := filepath.Rel(filepath.Join(h.tdir, ".parked", h.inc, p1Host), p)
+ parked = append(parked, strings.Join(strings.Split(rel, string(filepath.Separator))[1:], "/"))
+ }
+ return nil
+ })
+ sort.Strings(parked)
+ wantParked := []string{"artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env", "artifacts/" + p1Host + "/old.txt"}
+ if strings.Join(parked, " ") != strings.Join(wantParked, " ") {
+ t.Fatalf("parked: %v, want %v", parked, wantParked)
+ }
+ if _, err := os.Stat(filepath.Join(h.tdir, p1Host, "old.txt")); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the old files are still live: %v", err)
+ }
+ if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host+".env")); err != nil || string(body) != "APP_MODE=prod\n" {
+ t.Fatalf("env: %q %v", body, err)
+ }
+}
+
+// No earlier stay queues no park-local and runs no retirement (ruling G5),
+// and the first landing on a box completes end to end with the real
+// park.Local and hostguard.Guard.
+func TestLandingFirstStayQueuesNoLocalPark(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landAt = "after-record-parking"
+ h.crashLand(t)
+ if c := h.claim(t); len(c.Obligations) != 0 {
+ t.Fatalf("a first landing queued %+v", c.Obligations)
+ }
+ if j := h.landJournal(t); j.Done(landStepPrevious) {
+ t.Fatalf("a first landing noted a previous stay: %v", j.Steps)
+ }
+ h.land(t)
+ h.installed(t)
+}
+
+// The markers are written last (witness wMarkersLast, codex R2-6): a crash
+// before local's final sub-step leaves no .files-gen with everything else
+// durable; the replay finishes local and the landing.
+func TestLandingMarkersAreWrittenLast(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.landAt = "before-files-gen"
+ h.crashLand(t)
+ if _, ok := h.filesGen(t); ok {
+ t.Fatal(".files-gen exists before the last sub-step")
+ }
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandLocal {
+ t.Fatalf("record: %+v", m)
+ }
+ if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil || string(body) != "hello" {
+ t.Fatalf("the files were not extracted before the marker: %q %v", body, err)
+ }
+ if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host) + ".gen"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("<dbPath>.gen before restored: %v", err)
+ }
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+ if len(h.replaced) != 1 {
+ t.Fatalf("the replay replaced the timers again: %d", len(h.replaced))
+ }
+}
+
+// configured (§7.6, Review Focus 4): the held render's bundle and overlay
+// etags must be the pinned inputs; a difference marks the rollback with
+// "configuration changed during the move" and the landing stops there. The
+// release gate waits.
+func TestLandingConfiguredPinsInputs(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.render = func() (appconfig.HeldRender, error) {
+ return appconfig.HeldRender{BundleETags: map[string]string{"default": "e2", "environment": ""}}, nil
+ }
+ h.land(t)
+ m := h.read(t, p1Host).Move
+ if m.Reason != provision.RollbackConfigChanged || m.LandingPhase != provision.LandConfigured || m.State != provision.MoveLanding {
+ t.Fatalf("after the render: %+v", m)
+ }
+ if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("the database was installed under a mismatch: %v", err)
+ }
+ // A marked record is left alone.
+ before := m.UpdatedAt
+ h.land(t)
+ if after := h.read(t, p1Host).Move; !after.UpdatedAt.Equal(before) || after.Reason != provision.RollbackConfigChanged {
+ t.Fatalf("a marked record was written: %+v", after)
+ }
+
+ h2 := newLandingHarness(t)
+ h2.flip(t)
+ h2.release = moves.Check{Reason: t7Waiting}
+ h2.land(t)
+ if m := h2.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
+ t.Fatalf("waiting on the release: %+v", m)
+ }
+ h2.release = moves.Check{OK: true}
+ h2.land(t)
+ h2.installed(t)
+}
+
+// landing_deadline passed before restored marks the rollback, at whatever
+// phase the landing is; at restored it is the probe's.
+func TestLandingDeadlineMarksRollback(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.clock.Advance(cutLandingWindow + time.Minute)
+ h.land(t)
+ if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandChecking {
+ t.Fatalf("at checking: %+v", m)
+ }
+
+ h2 := newLandingHarness(t)
+ h2.flip(t)
+ h2.release = moves.Check{Reason: t7NotProven}
+ h2.land(t)
+ if m := h2.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
+ t.Fatalf("waiting on the proof: %+v", m)
+ }
+ h2.clock.Advance(cutLandingWindow + time.Minute)
+ h2.land(t)
+ if m := h2.read(t, p1Host).Move; m.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandConfigured {
+ t.Fatalf("at configured past the deadline: %+v", m)
+ }
+
+ h3 := newLandingHarness(t)
+ h3.flip(t)
+ h3.land(t)
+ h3.installed(t)
+ h3.clock.Advance(cutLandingWindow + time.Minute)
+ h3.land(t)
+ if m := h3.read(t, p1Host).Move; m.Reason != "" || m.LandingPhase != provision.LandRestored {
+ t.Fatalf("at restored past the deadline: %+v", m)
+ }
+}
+
+// Every phase is replay-safe: a crash after each durable side effect, then
+// a fresh worker to restored, yields one object per generation key, one
+// timers replacement, one install manifest, and the same end state.
+func TestLandingReplaysArePhaseIdempotent(t *testing.T) {
+ points := []string{"after-previous", "after-activate", "after-record-parking", "after-residue-queued", "after-record-building", "after-build",
+ "after-record-published", "after-record-local", "after-retired", "after-extracting", "after-extracted", "after-inventory", "after-timers",
+ "after-columns", "after-verify", "after-files_manifest", "before-files-gen", "after-files_gen", "after-record-configured", "after-release-ready", "after-db-rename"}
+ for _, point := range points {
+ t.Run(point, func(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.earlierStay(t)
+ h.residue(t)
+ h.landAt = point
+ h.crashLand(t)
+ h.freshTarget()
+ h.land(t)
+ h.installed(t)
+ if len(h.replaced) != 1 {
+ t.Fatalf("timers replaced %d times", len(h.replaced))
+ }
+ prefix := moves.GenerationPrefix(cutRoot, p1ID)
+ n := 0
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ n++
+ }
+ }
+ if n != 4 {
+ t.Fatalf("%d generation objects", n)
+ }
+ c := h.claim(t)
+ gens := 0
+ for _, o := range c.Obligations {
+ switch o.Kind {
+ case "park-generation":
+ gens++
+ default:
+ t.Fatalf("obligation left: %+v", o)
+ }
+ }
+ if gens != 2 {
+ t.Fatalf("%d residue chores", gens)
+ }
+ for p, s := range c.Inventory["uk-box"].Paths {
+ if s != c.Holder.Stamp {
+ t.Fatalf("inventory %s at %d", p, s)
+ }
+ }
+ })
+ }
+}
+
+// The prepared files manifest (spec rev 7.22 §7.6, R19-1) needs C8, which
+// is not on main: local records the step as skipped and the landing goes
+// on. When C8 lands this test becomes TestLocalBindsThePreparedManifestTo
+// TheNewGeneration: the manifest key files/g-<op>/manifest.json, bound to
+// g-<op> while holder.generation is still the source's.
+func TestLocalRecordsThePreparedManifestAsSkippedUntilC8(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.land(t)
+ h.installed(t)
+ if got := h.landJournal(t).Steps[landStepFilesManifest]; got != landingFilesManifestSkipped {
+ t.Fatalf("files_manifest: %q", got)
+ }
+ if c := h.claim(t); c.Holder.Generation != "cur" {
+ t.Fatalf("holder.generation moved before the probe: %q", c.Holder.Generation)
+ }
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, cutRoot+"/files/") {
+ t.Fatalf("a files manifest was written without C8: %s", k)
+ }
+ }
+}
+
+// Ruling G17: with the park engine's hooks unwired, as armPhase1 leaves
+// them, the landing is not armed: the pass logs and writes nothing.
+func TestLandingWaitsUnarmedInProduction(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ h.lw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
+ c := h.landingCfg(t)
+ c.ParkLeases = nil
+ h.lw.cfg = c
+ if landingArmed(c) {
+ t.Fatal("armed without park leases")
+ }
+ if err := movesPass(context.Background(), h.s, h.lw); err != nil {
+ t.Fatal(err)
+ }
+ if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
+ t.Fatalf("an unarmed landing advanced: %+v", m)
+ }
+ if _, err := hostguard.Capture(h.tdir, p1Host); !errors.Is(err, hostguard.ErrNoRecord) {
+ t.Fatalf("an unarmed landing wrote the installation record: %v", err)
+ }
+ // Production's own wiring: everything but the park engine's hooks.
+ var prod moveCfg
+ hib := activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}
+ armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, h.tdir, nil, nil, nil, func(string) string { return "" })
+ if prod.Guard.DataDir != h.tdir || prod.Local.DataDir != h.tdir || prod.OwnLease == nil || prod.RenderHeld == nil || prod.RemoveVerifyEntry == nil || prod.WriteGen == nil {
+ t.Fatalf("production wiring: %+v", prod)
+ }
+ if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || landingArmed(prod) {
+ t.Fatal("production armed the landing")
+ }
+}
+
+// The generation classifier the stand-in view and parking share.
+func TestLandingGenerationOf(t *testing.T) {
+ loc := "instances/h"
+ cases := map[string]struct {
+ gen string
+ ok bool
+ }{
+ loc + "/g-old/0000/x.ltx": {"old", true},
+ loc + "/g-mv_1/install.json": {"mv_1", true},
+ loc + "/files/g-mv_1/manifest": {"mv_1", true},
+ loc + "/0000/x.ltx": {"", true},
+ loc + "/0009/x.ltx": {"", true},
+ loc + "/generation.json": {landingMarkerGeneration, true},
+ loc + "/g-/x": {"", false},
+ loc + "/stray.json": {"", false},
+ loc + "/zzzz/x.ltx": {"", false},
+ "instances/other/0000/x.ltx": {"", false},
+ loc + "/" + strings.Repeat("a", 64) + "/0000/x.ltx": {"", false},
+ }
+ for key, want := range cases {
+ gen, ok := landingGenerationOf(loc, key)
+ if gen != want.gen || ok != want.ok {
+ t.Errorf("%s: %q %v, want %q %v", key, gen, ok, want.gen, want.ok)
+ }
+ }
+}
+
+// The target report pass keeps a landing's refusal and verdict live (G26).
+func TestTargetLiveSetCoversTheCutAndTheLanding(t *testing.T) {
+ h := newLandingHarness(t)
+ h.flip(t)
+ w := &movesWorker{cfg: h.landingCfg(t), mirrors: map[string]lease.MirrorStore{}, archives: map[string]verifiedArchive{p1ID: {}, "mv_other": {}}}
+ recs, err := provision.ListAllRecords(context.Background(), h.s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ fleet, err := loadMoveFleet(context.Background(), h.s)
+ if err != nil {
+ t.Fatal(err)
+ }
+ w.targetReportPass(context.Background(), h.s, recs, fleet)
+ if _, ok := w.archives[p1ID]; !ok {
+ t.Fatal("the landing's verdict was evicted")
+ }
+ if _, ok := w.archives["mv_other"]; ok {
+ t.Fatal("a finished move's verdict was kept")
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 7a6340391..b7d21ee68 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -17,10 +17,14 @@ import (
"github.com/carlosframework/platform/internal/activator"
"github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/dbverify"
+ "github.com/carlosframework/platform/internal/hostbox"
"github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
"github.com/carlosframework/platform/internal/layout"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
@@ -90,6 +94,9 @@ type movesWorker struct {
// box's running moves (moveshold.go), rebuilt every pass. Nil with the
// switch off.
holds *movesHoldView
+ // landLeases is the landing's own-location lease per host, held from
+ // building through published (movesland.go); guarded by mu.
+ landLeases map[string]park.HeldLease
}
// staging is the memoized staging writer for a target bucket region.
@@ -281,6 +288,28 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
cfg.HoldMoving = holdMoving(reg, holdStageRequester(dataDir))
cfg.Timers = reg.TimersFor
}
+ // The landing (movesland.go). Everything that exists on main is wired;
+ // ParkLeases, ParkView and ParkAuthorize are not (ruling G17: the
+ // residue parks need claim Plan 6b's regional runner for their
+ // acknowledgement and Plan 5's view and authorizer), so landingArmed is
+ // false in production and the landing pass only logs. Under
+ // CARLOS_HOST_CLAIM the held timers write would go through
+ // schedule.SelectedTimers with a held-timers Work (ruling G16); until
+ // Task 10b supplies hostbox.Policy.Moves the landing refuses at checking,
+ // so the registry's own ReplaceTimers is the pass-through here.
+ cfg.Guard = hostguard.Guard{DataDir: dataDir}
+ cfg.Local = park.Local{DataDir: dataDir, Box: cfg.Box}
+ cfg.OwnLease = hostbox.RenewingLeases{Region: hib.Region}.Acquire
+ cfg.RenderHeld = func(ctx context.Context, s release.Store, rt registry.Route, op string) (appconfig.HeldRender, error) {
+ return appconfig.RenderHeld(ctx, s, cfg.SecretsPriv, dataDir, rt, op, cfg.Box)
+ }
+ cfg.RemoveVerifyEntry = dbverify.RemoveHost
+ cfg.WriteGen = hostbox.WriteGenerationMarker
+ if reg != nil {
+ cfg.ReplaceTimers = reg.ReplaceTimers
+ cfg.SetReplication = reg.SetReplication
+ cfg.SetNextDue = reg.SetNextDue
+ }
if eb != nil {
// A box with no delegated cgroup has no manager, and Reconstruct
// refuses it: such a box cannot produce stop evidence, so it never
@@ -477,6 +506,38 @@ type moveCfg struct {
// with the point's name, an error being a crash there. Nil in
// production.
cutFault func(point string) error
+
+ // The landing's hooks (§7.6, movesland.go). Guard is the host guard
+ // over this box's data directory (its Claims are set per pass from the
+ // pass's store when nil); Local is park.Local over the same directory
+ // (its Guard, Claims and, when nil, Authorize are set per pass).
+ // ParkLeases, ParkView and ParkAuthorize are the park engine's over the
+ // own bucket for the residue parks: nil in production until claim Plan
+ // 6b's regional runner and Plan 5's providers (ruling G17), which is
+ // what leaves the landing unarmed there (landingArmed); a nil View or
+ // Authorize with ParkLeases set gets the landing's stand-ins. OwnLease
+ // takes the location's lease in the own bucket for the build
+ // (hostbox.RenewingLeases over lease.Acquire). RenderHeld is
+ // appconfig.RenderHeld for the held row; ReplaceTimers, SetReplication
+ // and SetNextDue are the registry's; RemoveVerifyEntry is
+ // dbverify.RemoveHost; WriteGen is hostbox.WriteGenerationMarker, the
+ // one <dbPath>.gen writer (ruling G2).
+ Guard hostguard.Guard
+ Local park.Local
+ ParkLeases park.Leases
+ ParkView func(context.Context, hostclaim.Observed) (park.View, error)
+ ParkAuthorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
+ OwnLease func(ctx context.Context, bucket, location string) (park.HeldLease, error)
+ RenderHeld func(ctx context.Context, s release.Store, rt registry.Route, op string) (appconfig.HeldRender, error)
+ ReplaceTimers func(host string, rows []registry.Timer) error
+ SetReplication func(host, class string) error
+ SetNextDue func(host string, due int64) error
+ RemoveVerifyEntry func(ctx context.Context, s release.Store, host string) error
+ WriteGen func(path, generation string) error
+ // landFault and parkFault are test seams: the landing's crash points
+ // and the park engine's (park.Engine.After). Nil in production.
+ landFault func(point string) error
+ parkFault func(step string) error
}
// defaultMoveReadyDelta is the readiness bar when
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 8bba3b79a..038aba1bb 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -192,8 +192,14 @@ func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec p
// The record decided the revert; the claim follows it (§7.5). The
// recovery itself is Plan 4's.
return w.revertClaim(ctx, s, rec)
- }
- return nil // landing and later: the target's, and Plans 3 to 5
+ case provision.MoveLanding:
+ // The target's (ruling G34): after the flip the record's region is
+ // this box's and reconcile has minted the held row, so OwnsRoute
+ // admits it here. checking to restored is Task 10's (movesland.go);
+ // the probe and the commit are Plan 3's.
+ return w.landingPass(ctx, s, rec)
+ }
+ return nil // committed and later: Plans 3 to 5
}
// srcRound is what one preparing round learned for the source report.
diff --git a/internal/host/movesphase1_harness_test.go b/internal/host/movesphase1_harness_test.go
index 97c20ce21..1e1ce03b6 100644
--- a/internal/host/movesphase1_harness_test.go
+++ b/internal/host/movesphase1_harness_test.go
@@ -145,6 +145,11 @@ type fakeVersioned struct {
mu sync.Mutex
calls []string
deny func(op, bucket, key string) error
+ // skipDelete, when it answers true for a key, makes DeleteVersion
+ // return success without removing the version: a delete the store
+ // accepted but whose listing still shows the version (Plan 2 Task 10,
+ // the landing's Parked gate).
+ skipDelete func(key string) bool
}
func (v *fakeVersioned) call(op, bucket, key string) error {
@@ -199,14 +204,33 @@ func (v *fakeVersioned) OpenVersion(ctx context.Context, bucket, key, versionID
return io.NopCloser(bytes.NewReader(b)), nil
}
-func (v *fakeVersioned) CopyVersion(context.Context, string, string, string, string, string, bool) (string, string, error) {
- return "", "", errors.New("fakeVersioned: CopyVersion is not used by Phase 1 Plan 1")
+// CopyVersion copies one exact version within the fake's buckets, as the
+// park engine archives a landing's residue (Plan 2 Task 10). createOnly
+// refuses an existing destination with lease.ErrDestinationExists.
+func (v *fakeVersioned) CopyVersion(_ context.Context, srcBucket, srcKey, srcVersionID, dstBucket, dstKey string, createOnly bool) (string, string, error) {
+ if err := v.call("copy-version", dstBucket, dstKey); err != nil {
+ return "", "", err
+ }
+ o, ok := v.f.buckets[srcBucket][srcKey]
+ if !ok || srcVersionID == "" || v.versionID(srcBucket, srcKey) != srcVersionID {
+ return "", "", lease.ErrMirrorNotFound
+ }
+ if _, exists := v.f.buckets[dstBucket][dstKey]; exists && createOnly {
+ return "", "", lease.ErrDestinationExists
+ }
+ ver := v.f.nextVersion()
+ v.f.store(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + ver, Size: o.Size}, o.body})
+ v.f.setVersion(dstBucket, dstKey, ver)
+ return ver, "etag-" + ver, nil
}
func (v *fakeVersioned) DeleteVersion(_ context.Context, bucket, key, versionID string) error {
if err := v.call("delete-version", bucket, key); err != nil {
return err
}
+ if v.skipDelete != nil && v.skipDelete(key) {
+ return nil
+ }
if _, ok := v.f.buckets[bucket][key]; ok && v.f.versionOf(bucket, key) == versionID {
delete(v.f.buckets[bucket], key)
}
diff --git a/internal/host/movesreport.go b/internal/host/movesreport.go
index 4b2f521da..147dbe336 100644
--- a/internal/host/movesreport.go
+++ b/internal/host/movesreport.go
@@ -127,10 +127,17 @@ func (w *movesWorker) targetReportPass(ctx context.Context, s release.Store, rec
defer w.retireHeldRefusals(live)
for _, rec := range recs {
m := rec.Move
- if m == nil || m.Legacy() || m.To != w.cfg.Region || (m.State != provision.MovePreparing && m.State != provision.MoveReady) {
+ if m == nil || m.Legacy() || m.To != w.cfg.Region {
continue
}
+ // A move into this box is live from preparing through the cut and
+ // the landing (ruling G26): the held refusal a landing's configured
+ // phase writes, and T5's verdict, stay until the move is over. The
+ // report itself is written only while preparing or ready.
live[m.ID] = true
+ if m.State != provision.MovePreparing && m.State != provision.MoveReady {
+ continue
+ }
if err := w.targetReport(ctx, s, rec, fleet); err != nil {
w.logTrouble(rec.Host+" (target)", err)
continue