amadan

carlos / platform Public

host: the journaled cut, requested to flipped, with stop evidence and the seal by version (§7.4, §7.5)

runCut drives a cutting record through every phase it can in one pass,
each phase's work before its record CAS (G8), resuming from the record,
the claim, the buckets and the op journal alone. requested runs a fresh
source round and Ready and compares the pin (G1, G13); admitted compares
again; stopping holds, fences, drains and records Reconstruct's evidence;
stopped checkpoints under the claim's lineage (G3), publishes the
complete stopped-source.json create-only, records the source, gives up
execution, runs the final round under the staging lease (G6) and seals by
version (G27); staged publishes the one sealed archive with its identity
journalled first (codex 19, G14); archived flips the record, then the
claim. Past the deadline: back to ready before stopping, a revert after,
whose claim half follows once execution is none (G7's writers' bound).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 96ee1c3906775fc56eb37fdc2a255052ce6c7626 parent f43c5d8
5 files changed, +2487 −2
  • approved_moves_copy_test.go +1 −0
  • internal/host/movescut.go +1506 −0
  • internal/host/movescut_test.go +921 −0
  • internal/host/movespass.go +51 −0
  • internal/host/movesphase1.go +8 −2
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 88dbbe0bf..c9f84f8c5 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -549,6 +549,7 @@ var movesScanFiles = []string{
"internal/host/movesfloor.go",
"internal/host/movesfloor_linux.go",
"internal/host/movesfloor_other.go",
+ "internal/host/movescut.go",
"internal/host/movesguard.go",
"internal/host/movesheldenv.go",
"internal/host/movesheldlock.go",
diff --git a/internal/host/movescut.go b/internal/host/movescut.go
new file mode 100644
index 000000000..243ae4922
--- /dev/null
+++ b/internal/host/movescut.go
@@ -0,0 +1,1506 @@
+package host
+
+import (
+ "context"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "io"
+ "log"
+ "os"
+ "path/filepath"
+ "strconv"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/moves/archive"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+ "github.com/superfly/ltx"
+)
+
+// The cut on the source (moves spec §5.3, §7.4, §7.5), behind
+// CARLOS_MOVES_PHASE1. One pass drives one cutting record as far as it can:
+// each phase does its work and then advances cut_phase by one guarded CAS,
+// and the next phase runs in the same pass on the re-read record. Nothing
+// is carried in memory between passes: every phase resumes from the
+// record, the claim, the deployment bucket, this box's own bucket and the
+// op journal (moves.Journal, <DataDir>/.moves/<host>/<op>/).
+//
+// Phase names (ruling G8). A phase is written when the previous phase's
+// work is complete, before its own: `stopping` before the hold and the
+// stop, `stopped` once the tree is proven stopped, `staged` once the final
+// round is sealed, `archived` once the seal names the archive. So a record
+// at X means "X's work may have begun, and everything before X is done",
+// and a replay of X redoes X's work from durable state. Each phase's
+// comment says what a replay finds.
+//
+// The deadline (§7.4): before any work, a phase past cut_deadline stops
+// the cut. Before anything stopped (admitted) that is back to ready; from
+// stopping on it is a revert, whose claim half follows the record once the
+// source has given up execution (revertClaim).
+
+// cutLandingWindow is the landing deadline the flip writes (now + 15 min).
+const cutLandingWindow = 15 * time.Minute
+
+// cutRequestLimit is the staging park's request limit, the longest a
+// staging request runs (lease's stagingRequestTimeout).
+const cutRequestLimit = 120 * time.Second
+
+// cutMaxDeferred is how many final rounds may defer before the cut reverts
+// (§7.4 staged: three deferred rounds).
+const cutMaxDeferred = 3
+
+// cutStepsPerPass bounds the phases one pass runs.
+const cutStepsPerPass = 8
+
+// Op journal steps the cut writes (moves.Journal).
+const (
+ cutStepEvidence = "evidence" // §5.3 steps 1 to 4, the StoppedReport JSON
+ cutStepArchive = "archive" // the cut archive's identity, before any PUT
+ cutStepRevertBound = "revert_bound" // the revert's staging park evidence_at
+)
+
+// cutRun is one pass's state for one cutting record.
+type cutRun struct {
+ w *movesWorker
+ s release.Store
+ fleet moveFleet
+ rec provision.Record
+ g provision.MoveGuard
+ // The staging lease and the guarded mirror, taken once per pass by the
+ // first phase that writes staging (ruling G6) and released at its end.
+ gm *guardedMirror
+ release func(context.Context) error
+}
+
+// runCut is the MoveCutting arm of phase1MoveRound.
+func (w *movesWorker) runCut(ctx context.Context, s release.Store, rec provision.Record, fleet moveFleet) error {
+ c := &cutRun{w: w, s: s, fleet: fleet, rec: rec, g: provision.MoveGuard{ID: rec.Move.ID, Incarnation: rec.Move.Incarnation}}
+ defer c.releaseLease()
+ for i := 0; i < cutStepsPerPass; i++ {
+ advanced, err := c.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 != c.g.ID || fresh.Move.Incarnation != c.g.Incarnation || fresh.Move.State != provision.MoveCutting {
+ return nil
+ }
+ c.rec = *fresh
+ }
+ return nil
+}
+
+func (c *cutRun) cfg() moveCfg { return c.w.cfg }
+
+// fault is the test seam after a side effect.
+func (c *cutRun) fault(point string) error {
+ if f := c.w.cfg.cutFault; 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, back to ready, a revert or the flip).
+func (c *cutRun) step(ctx context.Context) (bool, error) {
+ m := c.rec.Move
+ if m.State != provision.MoveCutting {
+ return false, nil
+ }
+ if m.CutDeadline != nil && c.cfg().Now().After(*m.CutDeadline) {
+ switch m.CutPhase {
+ case provision.CutAdmitted:
+ return c.backToReady(ctx, "the cut's deadline passed before anything stopped")
+ case provision.CutStopping:
+ return c.revert(ctx, provision.RevertStopEvidenceBudget)
+ case provision.CutStopped, provision.CutStaged, provision.CutArchived:
+ return c.revert(ctx, provision.RevertCutDeadline)
+ }
+ }
+ switch m.CutPhase {
+ case provision.CutRequested:
+ return c.requested(ctx)
+ case provision.CutAdmitted:
+ return c.admitted(ctx)
+ case provision.CutStopping:
+ return c.stopping(ctx)
+ case provision.CutStopped:
+ return c.stopped(ctx)
+ case provision.CutStaged:
+ return c.staged(ctx)
+ case provision.CutArchived:
+ return c.archived(ctx)
+ }
+ return false, nil
+}
+
+// cutTransition writes one guarded record transition. A conflict is
+// someone else's decision (an abort, a console write): logged, and the
+// pass ends without error.
+func (c *cutRun) cutTransition(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
+ if _, err := provision.SetRecordMove(ctx, c.s, c.rec.Account, c.rec.App, c.rec.Host, mut); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: %s of %s refused (%v); the record moved on", c.rec.Host, what, c.g.ID, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("%s: %w", what, err)
+ }
+ return true, nil
+}
+
+func (c *cutRun) advance(ctx context.Context, from, to string, stamp func(*provision.MoveState)) (bool, error) {
+ ok, err := c.cutTransition(ctx, "the cut "+from+" to "+to, provision.MoveCutPhase(c.g, from, to, stamp))
+ if ok {
+ log.Printf("moves: %s: cut %s is %s", c.rec.Host, c.g.ID, to)
+ }
+ return ok, err
+}
+
+// backToReady undoes an admission whose checks are not fresh or whose
+// inputs moved (§7.4 admitted): nothing has stopped. The pin is cleared and
+// the next passing readiness round pins again (G1).
+func (c *cutRun) backToReady(ctx context.Context, why string) (bool, error) {
+ ok, err := c.cutTransition(ctx, "back to ready", provision.MoveBackToReady(c.g))
+ if ok {
+ log.Printf("moves: %s: cut %s goes back to ready: %s", c.rec.Host, c.g.ID, why)
+ nudgeSerial(ctx, c.s)
+ }
+ return ok, err
+}
+
+// revert starts the pre-flip recovery (§7.4 reverting): the record first,
+// then the claim half once execution is none (revertClaim).
+func (c *cutRun) revert(ctx context.Context, reason string) (bool, error) {
+ ok, err := c.cutTransition(ctx, "revert", provision.MoveRevert(c.g, reason))
+ if !ok {
+ return ok, err
+ }
+ log.Printf("moves: %s: cut %s reverts: %s", c.rec.Host, c.g.ID, reason)
+ nudgeSerial(ctx, c.s)
+ if err := c.w.revertClaim(ctx, c.s, c.rec); err != nil {
+ // The record decided; the next pass's reverting arm writes the
+ // claim half (phase1MoveRound).
+ log.Printf("moves: %s: the claim half of the revert waits: %v", c.rec.Host, err)
+ }
+ return true, nil
+}
+
+// requested (§7.4): one source round under the staging lease, with a fresh
+// source report (G13), then Ready on that round and the target's report,
+// then the pinned inputs against the current ones; the admission stamps
+// cut_at, the budget and the deadline. Any check that fails goes back to
+// ready, so readiness reruns. A replay finds requested and runs the round
+// again: nothing here is written but the reports and the lane, which every
+// round rewrites.
+func (c *cutRun) requested(ctx context.Context) (bool, error) {
+ cfg := c.cfg()
+ sr, published, err := c.w.sourceRound(ctx, c.s, c.rec, c.fleet)
+ if err != nil || published == nil {
+ return false, err
+ }
+ if !sourcePassed(published) {
+ return c.backToReady(ctx, "this round's source report does not pass")
+ }
+ target, err := readTargetReport(ctx, c.s, c.rec)
+ if err != nil {
+ return false, err
+ }
+ if ok, why := moves.Ready(sr.deltaBytes, cfg.ReadyDelta, target, published, cfg.Now()); !ok {
+ return c.backToReady(ctx, why)
+ }
+ if same, err := c.inputsHold(ctx); err != nil || !same {
+ if err != nil {
+ return false, err
+ }
+ return c.backToReady(ctx, "the inputs pinned at readiness have changed")
+ }
+ now := cfg.Now()
+ budget := target.CutBudgetS
+ return c.cutTransition(ctx, "the admission", provision.MoveAdmit(c.g, now, budget, moves.CutDeadline(now, float64(budget))))
+}
+
+// admitted (§7.4): the pinned inputs still match the current bundle and
+// overlay versions, the release and the launch report, or back to ready.
+// A replay finds admitted and compares again; nothing has stopped.
+func (c *cutRun) admitted(ctx context.Context) (bool, error) {
+ same, err := c.inputsHold(ctx)
+ if err != nil {
+ return false, err
+ }
+ if !same {
+ return c.backToReady(ctx, "the inputs pinned at readiness have changed")
+ }
+ return c.advance(ctx, provision.CutAdmitted, provision.CutStopping, nil)
+}
+
+// inputsHold compares the record's pin with the current inputs.
+func (c *cutRun) inputsHold(ctx context.Context) (bool, error) {
+ cur, err := c.w.currentInputs(ctx, c.s, c.rec)
+ if err != nil {
+ return false, err
+ }
+ return sameInputs(c.rec.Move.Inputs, cur), nil
+}
+
+// currentInputs are the inputs the admission compares (§7.4 admitted): the
+// bundle and overlay versions, the live release and the launch report's
+// digest, as this box would publish them now.
+func (w *movesWorker) currentInputs(ctx context.Context, s release.Store, rec provision.Record) (provision.MoveInputs, error) {
+ cfg := w.cfg
+ if cfg.InputVersions == nil || cfg.LaunchEnv == nil {
+ return provision.MoveInputs{}, errors.New("no way to read the current inputs on this box")
+ }
+ bundle, overlay, err := cfg.InputVersions(ctx, s, rec)
+ if err != nil {
+ return provision.MoveInputs{}, fmt.Errorf("read the configuration versions: %w", err)
+ }
+ live, err := release.LiveVersion(cfg.OptDir, rec.Host)
+ if err != nil {
+ return provision.MoveInputs{}, fmt.Errorf("read the live release: %w", err)
+ }
+ key, err := moves.ReportKeyFromSecrets(cfg.SecretsKey)
+ if err != nil {
+ return provision.MoveInputs{}, err
+ }
+ env, _, err := cfg.LaunchEnv(rec.Host)
+ if err != nil {
+ return provision.MoveInputs{}, fmt.Errorf("compose the launch environment: %w", err)
+ }
+ return provision.MoveInputs{BundleETags: bundle, OverlayETag: overlay, Release: live, EnvReport: moves.ReportEnv(key, env).Digest}, nil
+}
+
+// sameInputs is the admission's comparison: every bundle etag, the overlay
+// etag, the release and the launch report. The files digest is not one of
+// them: a file written before the stop is carried by the cut's own archive.
+func sameInputs(pin *provision.MoveInputs, cur provision.MoveInputs) bool {
+ if pin == nil || pin.OverlayETag != cur.OverlayETag || pin.Release != cur.Release || pin.EnvReport != cur.EnvReport {
+ return false
+ }
+ if len(pin.BundleETags) != len(cur.BundleETags) {
+ return false
+ }
+ for k, v := range pin.BundleETags {
+ if cv, ok := cur.BundleETags[k]; !ok || cv != v {
+ return false
+ }
+ }
+ return true
+}
+
+// claim reads the host claim and checks it is this move's moving op for
+// the record's incarnation.
+func (c *cutRun) claim(ctx context.Context) (hostclaim.Observed, error) {
+ obs, err := hostclaim.Read(ctx, c.s, c.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", c.g.ID)
+ }
+ if err := hostclaim.RequireOp(c.g.ID)(obs.Claim); err != nil {
+ return obs, fmt.Errorf("move %s: %w", c.g.ID, err)
+ }
+ if h := obs.Claim.Holder; h == nil || h.Incarnation != c.rec.Move.Incarnation {
+ return obs, fmt.Errorf("move %s: the host claim names another incarnation", c.g.ID)
+ }
+ return obs, nil
+}
+
+// dbPath is the row's database path, or the flat default.
+func (c *cutRun) dbPath() string {
+ cfg := c.cfg()
+ if cfg.Route != nil {
+ if rt, err := cfg.Route(c.rec.Host); err == nil && rt != nil && rt.DBPath != "" {
+ return rt.DBPath
+ }
+ }
+ return provision.DefaultDBPath(cfg.DataDir, c.rec.Host)
+}
+
+func (c *cutRun) journal() (*moves.Journal, error) {
+ return moves.ReadJournal(c.cfg().DataDir, c.rec.Host, c.g.ID)
+}
+
+// stopping (§5.3, §7.4): the row held moving under the move id, the wake
+// fence, the timer drain, then the evidence of steps 1 to 4. A tree not
+// proven stopped waits for the next pass, until the deadline reverts with
+// "stop evidence budget". Proven, the evidence goes into the op journal
+// (local: the report's replica step, N and digests come at stopped) and
+// the phase advances.
+//
+// A replay finds stopping and runs it all again from the box's state:
+// every step is idempotent (the hold under the same op, the fence at the
+// same generation, the drain, and Evidence, which reads only durable
+// state). No in-memory state is needed (Review Focus 1).
+func (c *cutRun) stopping(ctx context.Context) (bool, error) {
+ cfg := c.cfg()
+ host := c.rec.Host
+ obs, err := c.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
+ return false, fmt.Errorf("move %s: %w", c.g.ID, 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", c.g.ID, obs.Claim.Op.Exec)
+ }
+ if cfg.HoldMoving == nil || cfg.Fence == nil || cfg.TimersDrained == nil || cfg.Evidence == nil {
+ return false, errors.New("this box cannot stop an instance for a move: a hook is not wired")
+ }
+ if err := cfg.HoldMoving(host, c.g.ID); err != nil {
+ return false, fmt.Errorf("hold the row: %w", err)
+ }
+ gen := moveFenceGeneration(c.rec.Move)
+ if err := cfg.Fence(ctx, host, gen); err != nil {
+ return false, fmt.Errorf("fence against wakes: %w", err)
+ }
+ c.w.fenced.Store(host, gen)
+ if err := cfg.TimersDrained(ctx, host); err != nil {
+ return false, fmt.Errorf("drain the timers: %w", err)
+ }
+ rep, err := cfg.Evidence(ctx, host, obs.Claim.Holder.Location, []string{c.dbPath()})
+ if err != nil {
+ return false, fmt.Errorf("stop evidence: %w", err)
+ }
+ if !rep.TreeStopped() {
+ c.w.notice(host+"#evidence", "moves: %s: the instance is not proven stopped yet (%s); retrying until the cut's deadline", host, failedSteps(rep))
+ return false, nil
+ }
+ j, err := c.journal()
+ if err != nil {
+ return false, err
+ }
+ body, err := json.Marshal(rep)
+ if err != nil {
+ return false, err
+ }
+ j.Phase = provision.CutStopping
+ j.Mark(cutStepEvidence, string(body))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ if err := c.fault("after-evidence"); err != nil {
+ return false, err
+ }
+ return c.advance(ctx, provision.CutStopping, provision.CutStopped, nil)
+}
+
+// failedSteps names the evidence steps that did not pass, for the log.
+func failedSteps(r moves.StoppedReport) string {
+ var out []string
+ for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
+ if c, ok := r.Steps[id]; !ok || !c.OK {
+ out = append(out, id)
+ }
+ }
+ return strings.Join(out, ", ")
+}
+
+// stopped (§5.3 step 5, §7.4): the stopped source's report, its execution
+// given up, then the final round and the seal.
+//
+// While the claim's executor is still this box: the final checkpoint under
+// the claim's lineage (G3), N, the database and timers digests, the files'
+// plain digest, the complete stopped-source.json (create-only, K13), the
+// record's source baselines, and then exec to none with the cut's session
+// in one claim write (K3). Once exec is none, a replay reads the published
+// report instead: nothing on the source's replica is written again.
+//
+// Then, under the staging lease: the timers snapshot into timers/ and the
+// final mirror round into replica/. A round that deferred or moved more
+// than the bar is counted in the ledger (three revert); a round that
+// cannot name a usable version for every current key runs again. The seal
+// goes into the claim with the op's phase sealed in one CAS, and only then
+// does the record say staged (G8).
+//
+// A replay finds stopped and repeats from the claim's state: the
+// checkpoint is idempotent on a stopped database, the report and the
+// snapshots are create-only and read back, every claim write is
+// conditional, and the final round rewrites staging, which the seal pins by
+// version.
+func (c *cutRun) stopped(ctx context.Context) (bool, error) {
+ cfg := c.cfg()
+ obs, err := c.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ op := obs.Claim.Op
+ if op.Kind != hostclaim.KindMoving || (op.Phase != hostclaim.PhaseMoving && op.Phase != hostclaim.PhaseSealed) {
+ return false, fmt.Errorf("move %s: the claim is %s/%q", c.g.ID, op.Kind, op.Phase)
+ }
+ holder := *obs.Claim.Holder
+ stoppedKey := moves.StoppedKey(moves.ReportPrefix(c.rec.Account, c.rec.App, c.rec.Host, c.g.ID), "source")
+ var report moves.StoppedReport
+ switch {
+ case cfg.Box != "" && op.Exec == cfg.Box:
+ if report, err = c.giveUpExecution(ctx, obs, stoppedKey); err != nil {
+ return false, err
+ }
+ if report.V == 0 {
+ return false, nil // reverted, or waiting
+ }
+ case op.Exec == "" && op.Journal[hostclaim.JournalEvidenceSource] == stoppedKey:
+ if report, err = readStoppedReport(ctx, c.s, stoppedKey); err != nil {
+ return false, err
+ }
+ default:
+ return false, fmt.Errorf("move %s: the claim's executor is %q and cites %q", c.g.ID, op.Exec, op.Journal[hostclaim.JournalEvidenceSource])
+ }
+ return c.finalRound(ctx, holder, report, stoppedKey)
+}
+
+// giveUpExecution is stopped's first half, run while this box executes.
+// It returns a zero report when the phase ended without one (a revert, or
+// a file that changed).
+func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, stoppedKey string) (moves.StoppedReport, error) {
+ cfg := c.cfg()
+ host := c.rec.Host
+ holder := *obs.Claim.Holder
+ j, err := c.journal()
+ if err != nil {
+ return moves.StoppedReport{}, err
+ }
+ var ev moves.StoppedReport
+ if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() {
+ // The journal lost the evidence (a rebuilt data directory): the row
+ // is still held and fenced, so ask the box again.
+ if cfg.Evidence == nil {
+ return moves.StoppedReport{}, errors.New("no stop evidence on this box")
+ }
+ if ev, err = cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()}); err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("stop evidence: %w", err)
+ }
+ if !ev.TreeStopped() {
+ return moves.StoppedReport{}, fmt.Errorf("the instance is no longer proven stopped (%s)", failedSteps(ev))
+ }
+ }
+ own, vs, err := c.ownStores(ctx)
+ if err != nil {
+ return moves.StoppedReport{}, err
+ }
+ prefix, keep := moves.ReplicaSource(holder)
+ dbPath := c.dbPath()
+ fromDB := regularFile(dbPath)
+ if fromDB {
+ if cfg.FinalCheckpoint == nil {
+ return moves.StoppedReport{}, errors.New("no final checkpoint on this box")
+ }
+ check := func(ctx context.Context) error { return c.stillExecuting(ctx) }
+ if err := cfg.FinalCheckpoint(ctx, dbPath, holder, check); err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("final checkpoint: %w", err)
+ }
+ }
+ if err := c.fault("after-checkpoint"); err != nil {
+ return moves.StoppedReport{}, err
+ }
+ n, err := c.finalTXID(ctx, own, prefix, keep)
+ if err != nil {
+ return moves.StoppedReport{}, err
+ }
+ var dbSHA string
+ if fromDB {
+ if dbSHA, err = fileSHA256(dbPath); err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("digest the stopped database: %w", err)
+ }
+ }
+ timersSHA, err := c.snapshotTimers(ctx, own, vs)
+ if err != nil {
+ return moves.StoppedReport{}, err
+ }
+ dir, envPath := cfg.FilesDir(host)
+ plain, _, _, err := archivePlainDigest(dir, envPath)
+ if why, retry, ferr := archiveFailure(host, err); ferr != nil || why != "" || retry {
+ if why != "" {
+ _, rerr := c.revert(ctx, provision.RevertArchiveRefused)
+ return moves.StoppedReport{}, rerr
+ }
+ return moves.StoppedReport{}, ferr
+ }
+ var rt registry.Route
+ if cfg.Route != nil {
+ if r, err := cfg.Route(host); err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("read the registry row: %w", err)
+ } else if r != nil {
+ rt = *r
+ }
+ }
+ report := ev
+ report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, "source", cfg.Box, holder.Incarnation, obs.Claim.Gen
+ report.Host, report.OpID = host, c.g.ID
+ report.Steps = map[string]moves.Check{}
+ for k, v := range ev.Steps {
+ report.Steps[k] = v
+ }
+ report.Steps[moves.StepReplica] = moves.Check{OK: true}
+ report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, timersSHA
+ if !report.Complete() {
+ return moves.StoppedReport{}, fmt.Errorf("the stopped report is not complete (%s)", failedSteps(report))
+ }
+ // K13: create-only. A report already there is this cut's own replay
+ // when it names the same host, op, role and N; it is then the report.
+ if report, err = publishStoppedReport(ctx, c.s, stoppedKey, report); err != nil {
+ return moves.StoppedReport{}, err
+ }
+ if err := c.fault("after-report"); err != nil {
+ return moves.StoppedReport{}, err
+ }
+ src := provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: report.FinalTXID, DBSHA256: report.DBSHA256,
+ Files: &provision.MoveFilesDigest{Plain: report.FilesPlain}, TimersSHA256: report.TimersSHA256, Evidence: stoppedKey}
+ if ok, err := c.cutTransition(ctx, "the source baselines", provision.MoveRecordSource(c.g, src)); !ok {
+ if err == nil {
+ err = errors.New("the record moved on")
+ }
+ return moves.StoppedReport{}, err
+ }
+ if err := c.fault("after-record-source"); err != nil {
+ return moves.StoppedReport{}, err
+ }
+ session := strconv.FormatUint(holder.Session, 10)
+ if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
+ if err := hostclaim.ExecToNone(c.g.ID, cfg.Box, stoppedKey, hostclaim.JournalEvidenceSource)(cl); err != nil {
+ return err
+ }
+ return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalCutSession: session})(cl)
+ }); err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("give up execution: %w", err)
+ }
+ log.Printf("moves: %s: stopped at TXID %d; execution given up, citing %s", host, report.FinalTXID, stoppedKey)
+ if err := c.fault("after-exec-none"); err != nil {
+ return moves.StoppedReport{}, err
+ }
+ return report, nil
+}
+
+// stillExecuting is the final checkpoint's authority: every replica
+// request it makes first re-reads the claim, which must still name this
+// move with this box executing.
+func (c *cutRun) stillExecuting(ctx context.Context) error {
+ obs, err := c.claim(ctx)
+ if err != nil {
+ return err
+ }
+ if obs.Claim.Op.Kind != hostclaim.KindMoving || obs.Claim.Op.Exec == "" || obs.Claim.Op.Exec != c.cfg().Box {
+ return fmt.Errorf("move %s: this box no longer executes the host", c.g.ID)
+ }
+ return nil
+}
+
+// publishStoppedReport writes the report create-only. On a conflict the
+// existing report is read: the same host, op, role and N is a replay and
+// the existing report is returned; anything else is an error.
+func publishStoppedReport(ctx context.Context, s release.Store, key string, r moves.StoppedReport) (moves.StoppedReport, error) {
+ body, err := json.Marshal(r)
+ if err != nil {
+ return r, err
+ }
+ if _, err := s.Put(ctx, key, body, ""); err == nil {
+ return r, nil
+ } else if !errors.Is(err, release.ErrCASConflict) {
+ return r, fmt.Errorf("write %s: %w", key, err)
+ }
+ prev, err := readStoppedReport(ctx, s, key)
+ if err != nil {
+ return r, err
+ }
+ if prev.Host != r.Host || prev.OpID != r.OpID || prev.Role != r.Role || prev.FinalTXID != r.FinalTXID {
+ return r, fmt.Errorf("%s already names %s %s %s at TXID %d, not this stop at %d", key, prev.Host, prev.OpID, prev.Role, prev.FinalTXID, r.FinalTXID)
+ }
+ return prev, nil
+}
+
+func readStoppedReport(ctx context.Context, s release.Store, key string) (moves.StoppedReport, error) {
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ return moves.StoppedReport{}, fmt.Errorf("read %s: %w", key, err)
+ }
+ var r moves.StoppedReport
+ if err := json.Unmarshal(body, &r); err != nil || r.V != 1 || !r.Complete() {
+ return moves.StoppedReport{}, fmt.Errorf("%s is not a complete stopped report", key)
+ }
+ return r, nil
+}
+
+// ownStores are this box's own bucket as a mirror (reads and plain writes)
+// and by version (create-only writes).
+func (c *cutRun) ownStores(ctx context.Context) (lease.MirrorStore, lease.VersionedStore, error) {
+ cfg := c.cfg()
+ tgt, ok := c.fleet.targets[c.rec.Move.To]
+ if !ok {
+ return nil, nil, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
+ }
+ own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ return nil, nil, err
+ }
+ if cfg.Versioned == nil {
+ return nil, nil, errors.New("no versioned store on this box")
+ }
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return nil, nil, err
+ }
+ return own, vs, nil
+}
+
+// finalTXID is N (§5.1 step 1): the highest max TXID among the claim
+// generation's level files in this box's own replica, the set the lane
+// mirrors (G3). After the final checkpoint it is the newest level-0
+// file's; a compaction only ever rewrites below it.
+func (c *cutRun) finalTXID(ctx context.Context, own lease.MirrorStore, prefix string, keep func(string) bool) (uint64, error) {
+ objs, err := own.List(ctx, c.cfg().OwnBucket, prefix)
+ if err != nil {
+ return 0, fmt.Errorf("list the replica: %w", err)
+ }
+ newest := newestLTX(objs, prefix, keep)
+ if newest == "" {
+ return 0, errors.New("the replica holds no level file")
+ }
+ _, maxTX, err := ltx.ParseFilename(filepath.Base(newest))
+ if err != nil {
+ return 0, err
+ }
+ return uint64(maxTX), nil
+}
+
+// snapshotTimers writes the host's timer rows create-only to
+// move-snapshots/<host>/<op>/timers.json in this box's bucket and returns
+// their digest. A snapshot already there is this cut's own (a replay): it
+// is read back and its rows are the ones digested.
+func (c *cutRun) snapshotTimers(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore) (string, error) {
+ cfg := c.cfg()
+ if cfg.Timers == nil {
+ return "", errors.New("no timer rows on this box")
+ }
+ rows, err := cfg.Timers(c.rec.Host)
+ if err != nil {
+ return "", fmt.Errorf("read the timers: %w", err)
+ }
+ body, err := moves.TimersSnapshot{V: 1, Host: c.rec.Host, Rows: rows, TakenAt: cfg.Now().UTC()}.Encode()
+ if err != nil {
+ return "", err
+ }
+ key := moves.SnapshotsPrefix(c.rec.Host, c.g.ID) + "timers.json"
+ if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, key, body); errors.Is(err, lease.ErrExists) {
+ prev, err := own.Get(ctx, cfg.OwnBucket, key)
+ if err != nil {
+ return "", fmt.Errorf("read %s: %w", key, err)
+ }
+ snap, err := moves.DecodeTimers(prev)
+ if err != nil {
+ return "", err
+ }
+ rows = snap.Rows
+ } else if err != nil {
+ return "", fmt.Errorf("snapshot the timers: %w", err)
+ }
+ return moves.TimersDigest(rows), nil
+}
+
+// lease takes the staging lease once per pass (G6) and revalidates the
+// claim under it (K5). waiting is true when another worker holds it.
+func (c *cutRun) lease(ctx context.Context) (waiting bool, err error) {
+ if c.gm != nil {
+ return false, nil
+ }
+ cfg := c.cfg()
+ if cfg.StagingLease == nil {
+ return false, errors.New("no deployment bucket for the staging lease")
+ }
+ tgt, ok := c.fleet.targets[c.rec.Move.To]
+ if !ok {
+ return false, fmt.Errorf("target region %s publishes no data bucket yet. waiting", c.rec.Move.To)
+ }
+ held, rel, err := cfg.StagingLease(ctx, c.rec.Move.To, c.rec.Host)
+ if err != nil {
+ var he *lease.HeldError
+ if errors.As(err, &he) {
+ log.Printf("moves: %s: another worker holds the staging lease; the cut waits", c.rec.Host)
+ return true, nil
+ }
+ return false, fmt.Errorf("staging lease: %w", err)
+ }
+ stg, err := c.w.staging(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ rel(context.Background())
+ return false, err
+ }
+ own, err := c.w.mirror(ctx, tgt.DataBucketRegion)
+ if err != nil {
+ rel(context.Background())
+ return false, err
+ }
+ c.gm = &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: cfg.Now}, dst: tgt.DataBucket, own: own}
+ c.release = rel
+ return false, nil
+}
+
+func (c *cutRun) releaseLease() {
+ if c.release == nil {
+ return
+ }
+ rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := c.release(rctx); err != nil {
+ log.Printf("moves: %s: release the staging lease: %v", c.rec.Host, err)
+ }
+ c.release, c.gm = nil, nil
+}
+
+// sealedClaim revalidates the claim under the lease for the phases after
+// execution was given up: this move's moving op with exec none and this
+// stop's evidence, at one of the phases given.
+func (c *cutRun) sealedClaim(ctx context.Context, stoppedKey string, phases ...string) (hostclaim.Observed, error) {
+ obs, err := c.claim(ctx)
+ if err != nil {
+ return obs, err
+ }
+ op := obs.Claim.Op
+ if op.Kind != hostclaim.KindMoving || !containsString(phases, op.Phase) || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
+ return obs, fmt.Errorf("move %s: the claim is %s/%q exec %q, not given up for this cut", c.g.ID, op.Kind, op.Phase, op.Exec)
+ }
+ return obs, nil
+}
+
+// bound is the writers' horizon for the seal (G7): the latest moment a
+// request this pass signed can still land, and never earlier than the
+// bound already recorded.
+func (c *cutRun) bound(prev time.Time) time.Time {
+ b := prev
+ if last := c.gm.lastSigned(); !last.IsZero() {
+ if t := last.Add(moves.SignatureLife + moves.DefaultSkew); t.After(b) {
+ b = t
+ }
+ }
+ return b.UTC()
+}
+
+// finalRound is stopped's second half: the timers and the replica into
+// staging under the lease, then the seal.
+func (c *cutRun) finalRound(ctx context.Context, holder hostclaim.Holder, report moves.StoppedReport, stoppedKey string) (bool, error) {
+ cfg := c.cfg()
+ host, id := c.rec.Host, c.g.ID
+ if waiting, err := c.lease(ctx); waiting || err != nil {
+ return false, err
+ }
+ obs, err := c.sealedClaim(ctx, stoppedKey, hostclaim.PhaseMoving, hostclaim.PhaseSealed)
+ if err != nil {
+ return false, err
+ }
+ own, _, err := c.ownStores(ctx)
+ if err != nil {
+ return false, err
+ }
+ dst := c.gm.dst
+ timersKey := moves.SnapshotsPrefix(host, id) + "timers.json"
+ timersBody, err := own.Get(ctx, cfg.OwnBucket, timersKey)
+ if err != nil {
+ return false, fmt.Errorf("read %s: %w", timersKey, err)
+ }
+ stagedTimers := moves.TimersPrefix(host, id) + "timers.json"
+ tver, err := c.gm.PutVersioned(ctx, dst, stagedTimers, timersBody)
+ if err != nil {
+ return false, fmt.Errorf("stage the timers: %w", err)
+ }
+ if !lease.UsableVersionID(tver) {
+ return false, fmt.Errorf("the staged timers have no usable version id %q", tver)
+ }
+ prefix, keep := moves.ReplicaSource(holder)
+ dstPrefix := moves.ReplicaPrefix(host, id)
+ receipts := lease.Receipts{Bucket: cfg.OwnBucket, Key: moves.ReceiptsKey(host, id)}
+ var round lease.MirrorRound
+ for attempt := 0; ; attempt++ {
+ round, err = lease.MirrorPrefixKeep(ctx, c.gm, cfg.OwnBucket, prefix, dst, dstPrefix, receipts, keep)
+ if err == nil || attempt == 1 || errors.Is(err, moves.ErrLeaseUnsafe) || errors.Is(err, moves.ErrInvalidSkew) {
+ break
+ }
+ log.Printf("moves: %s: final round failed, retrying once: %v", host, err)
+ }
+ if err != nil {
+ return false, fmt.Errorf("final round: %w", err)
+ }
+ if err := c.fault("after-final-round"); err != nil {
+ return false, err
+ }
+ ledgerKey := moves.LedgerKey(host, id)
+ ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey)
+ if err != nil {
+ return false, err
+ }
+ // §7.4 staged: a round that deferred is not byte-final; nor is one that
+ // moved more than the bar the cut was priced for (G27).
+ if round.Deferred > 0 || round.DeltaBytes() > cfg.ReadyDelta {
+ ledger.DeferredRounds++
+ if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: final round %d deferred %d object(s) and moved %d bytes", host, ledger.DeferredRounds, round.Deferred, round.DeltaBytes())
+ if ledger.DeferredRounds >= cutMaxDeferred {
+ return c.revert(ctx, provision.RevertDeferredRounds)
+ }
+ nudgeSerial(ctx, c.s)
+ return false, nil
+ }
+ // G27: every kept current source key by a usable version, else another
+ // round.
+ src, err := own.List(ctx, cfg.OwnBucket, prefix)
+ if err != nil {
+ return false, fmt.Errorf("list the replica: %w", err)
+ }
+ var replica []moves.SealedObject
+ for k, o := range src {
+ rel := strings.TrimPrefix(k, prefix)
+ if !keep(rel) {
+ continue
+ }
+ ver := round.Versions[rel]
+ if !lease.UsableVersionID(ver) {
+ c.w.notice(host+"#seal", "moves: %s: the final round has no usable version for %s; another round", host, rel)
+ return false, nil
+ }
+ replica = append(replica, moves.SealedObject{Key: dstPrefix + rel, VersionID: ver, Size: o.Size})
+ }
+ fromDB := regularFile(c.dbPath())
+ covered := true
+ if fromDB {
+ if cfg.LocalPosition == nil {
+ return false, errors.New("no local position on this box")
+ }
+ pos, err := cfg.LocalPosition(c.dbPath())
+ if err != nil {
+ return false, fmt.Errorf("read the stopped database's position: %w", err)
+ }
+ covered = pos == report.FinalTXID
+ }
+ now := cfg.Now().UTC()
+ seal := moves.Seal{V: moves.SealVersion, N: report.FinalTXID, Replica: replica,
+ Timers: []moves.SealedObject{{Key: stagedTimers, VersionID: tver, Size: int64(len(timersBody))}},
+ SrcCovered: covered, FromDB: fromDB, SealedAt: now}
+ seal.LeaseUntil = c.bound(now)
+ if err := seal.Validate(); err != nil {
+ return false, err
+ }
+ if !seal.ReachesN() {
+ return false, fmt.Errorf("the sealed replica does not end at TXID %d", seal.N)
+ }
+ if _, err := hostclaim.Update(ctx, c.s, obs, sealStopped(id, stoppedKey, seal)); err != nil {
+ return false, fmt.Errorf("seal the cut: %w", err)
+ }
+ if err := c.fault("after-seal"); err != nil {
+ return false, err
+ }
+ ledger.LeaseUntil = seal.LeaseUntil
+ if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: sealed %d replica objects at TXID %d", host, len(replica), seal.N)
+ return c.advance(ctx, provision.CutStopped, provision.CutStaged, nil)
+}
+
+// sealStopped is the stopped seal's claim write: the op's phase "" to
+// sealed (or already sealed, a replay) and the seal, in one CAS, under
+// exec none and this stop's evidence.
+func sealStopped(id, stoppedKey string, seal moves.Seal) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if c.Op.Exec != "" || c.Op.Journal[hostclaim.JournalEvidenceSource] != stoppedKey {
+ return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
+ }
+ if c.Op.Phase == hostclaim.PhaseMoving {
+ if err := hostclaim.SetMovePhase(id, hostclaim.KindMoving, hostclaim.PhaseMoving, hostclaim.PhaseSealed)(c); err != nil {
+ return err
+ }
+ } else if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseSealed)(c); err != nil {
+ return err
+ }
+ return hostclaim.SetJournal(id, map[string]string{hostclaim.JournalSeal: seal.Encode()})(c)
+ }
+}
+
+// journalledSeal reads the seal the claim carries.
+func journalledSeal(c *hostclaim.Claim) (moves.Seal, error) {
+ raw, ok := c.Op.Journal[hostclaim.JournalSeal]
+ if !ok {
+ return moves.Seal{}, fmt.Errorf("%w: the claim carries no seal", moves.ErrBadSeal)
+ }
+ return moves.DecodeSeal(raw)
+}
+
+// staged (§7.4, §8.3, codex 19, G14): the final archive of the stopped
+// files directory, its identity fixed before any PUT, written create-only
+// to move-snapshots/ and staged into files/, then the seal with the
+// archive's two versions, and only then the record's archived with the
+// archive's digests. A replay finds staged and redoes the archive from its
+// journalled identity (cutArchive), then the seal and the phase.
+func (c *cutRun) staged(ctx context.Context) (bool, error) {
+ a, ok, err := c.sealArchive(ctx)
+ if err != nil || !ok {
+ return false, err
+ }
+ return c.advance(ctx, provision.CutStaged, provision.CutArchived, func(m *provision.MoveState) {
+ if m.Source == nil {
+ m.Source = &provision.MoveSource{}
+ }
+ m.Source.Files = &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries}
+ })
+}
+
+// sealArchive is staged's work, shared with archived's replay: the archive
+// published and the claim's seal naming it. ok is false when the phase
+// ended without it (a revert, the lease held elsewhere).
+func (c *cutRun) sealArchive(ctx context.Context) (builtArchive, bool, error) {
+ cfg := c.cfg()
+ src := c.rec.Move.Source
+ if src == nil || src.Files == nil || src.Evidence == "" {
+ return builtArchive{}, false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
+ }
+ if waiting, err := c.lease(ctx); waiting || err != nil {
+ return builtArchive{}, false, err
+ }
+ obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
+ if err != nil {
+ return builtArchive{}, false, err
+ }
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return builtArchive{}, false, err
+ }
+ if seal.N != src.FinalTXID {
+ return builtArchive{}, false, fmt.Errorf("move %s: the seal is at TXID %d and the record at %d", c.g.ID, seal.N, src.FinalTXID)
+ }
+ a, files, why, err := c.cutArchive(ctx, src.Files.Plain)
+ if err != nil {
+ return builtArchive{}, false, err
+ }
+ if why != "" {
+ _, err := c.revert(ctx, provision.RevertArchiveRefused)
+ return builtArchive{}, false, err
+ }
+ seal.Files = files
+ seal.LeaseUntil = c.bound(seal.LeaseUntil)
+ if err := seal.Validate(); err != nil {
+ return builtArchive{}, false, err
+ }
+ if _, err := hostclaim.Update(ctx, c.s, obs, func(cl *hostclaim.Claim) error {
+ if cl.Op.Exec != "" || cl.Op.Journal[hostclaim.JournalEvidenceSource] != src.Evidence {
+ return fmt.Errorf("%w: the seal needs exec none and this stop's evidence", hostclaim.ErrWrongState)
+ }
+ if err := hostclaim.RequireMoveOp(c.g.ID, hostclaim.KindMoving, hostclaim.PhaseSealed)(cl); err != nil {
+ return err
+ }
+ return hostclaim.SetJournal(c.g.ID, map[string]string{hostclaim.JournalSeal: seal.Encode()})(cl)
+ }); err != nil {
+ return builtArchive{}, false, fmt.Errorf("seal the archive: %w", err)
+ }
+ if err := c.fault("after-archive-seal"); err != nil {
+ return builtArchive{}, false, err
+ }
+ own, _, err := c.ownStores(ctx)
+ if err != nil {
+ return builtArchive{}, false, err
+ }
+ ledgerKey := moves.LedgerKey(c.rec.Host, c.g.ID)
+ if ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey); err == nil {
+ ledger.LeaseUntil = seal.LeaseUntil
+ if err := writeMoveLedger(ctx, own, cfg.OwnBucket, ledgerKey, ledger); err != nil {
+ return builtArchive{}, false, err
+ }
+ }
+ return a, true, nil
+}
+
+// cutArchiveStep is the archive's identity in the op journal, written
+// before any PUT so a replay publishes the same object (codex 19). Reuse
+// marks the preparing archive the target verified (G14): the seal names
+// its staged versions and nothing is staged again.
+type cutArchiveStep struct {
+ builtArchive
+ Reuse bool `json:"reuse,omitempty"`
+}
+
+// cutArchive publishes the cut's one sealed archive (§8.3) and returns the
+// seal's files: the sealed object and its manifest by version.
+//
+// Its identity is fixed before any PUT (sealing is randomized), in the op
+// journal. The first run reuses the preparing archive when its plaintext
+// digest is the stopped files' and its staged versions are pinned and its
+// sealed file is still on this disk (G14); otherwise it builds and seals
+// a new one. Then, from the journalled identity:
+// - the sealed file on disk: create-only into move-snapshots/ (an existing
+// object is this cut's earlier PUT), then staged into files/ unless it
+// is the reused archive, whose staged versions the ledger holds;
+// - no file, but both objects in move-snapshots/: staged into files/ by a
+// guarded server-side copy from there (or, reused, nothing to stage);
+// - neither: nothing was published under that identity, so the archive
+// is rebuilt and resealed and the journal entry overwritten.
+//
+// why is a recorded reason the stopped files cannot be archived.
+func (c *cutRun) cutArchive(ctx context.Context, plain string) (builtArchive, []moves.SealedObject, string, error) {
+ cfg := c.cfg()
+ host, id := c.rec.Host, c.g.ID
+ own, vs, err := c.ownStores(ctx)
+ if err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ ledger, err := readMoveLedger(ctx, own, cfg.OwnBucket, moves.LedgerKey(host, id))
+ if err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ j, err := c.journal()
+ if err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ adir := archiveDir(cfg.DataDir, host, id)
+ var step cutArchiveStep
+ journalled := false
+ if raw, ok := j.Steps[cutStepArchive]; ok && json.Unmarshal([]byte(raw), &step) == nil && step.Name == archiveName(step.Sealed) && step.Plain == plain {
+ step.dir = adir
+ journalled = true
+ }
+ snap := moves.SnapshotsPrefix(host, id)
+ if journalled {
+ switch onDisk := sealedOnDisk(step.builtArchive); {
+ case onDisk:
+ case c.snapshotPresent(ctx, vs, snap+step.Name):
+ default:
+ journalled = false // nothing published under it: rebuild
+ }
+ }
+ if !journalled {
+ reuse := ledger.ArchivePlain == plain && ledger.ArchiveVersions != nil && ledger.ArchiveVersions.Pinned() && ledger.ArchiveSealed != ""
+ if reuse {
+ kept := builtArchive{Name: archiveName(ledger.ArchiveSealed), Sealed: ledger.ArchiveSealed, Plain: plain, dir: adir}
+ if m, ok := keptManifest(kept); ok && sealedOnDisk(kept) {
+ kept.Bytes, kept.Entries = m.Bytes, len(m.Entries)
+ if st, err := os.Stat(kept.sealedPath()); err == nil {
+ kept.SealedSize = st.Size()
+ }
+ step = cutArchiveStep{builtArchive: kept, Reuse: true}
+ } else {
+ reuse = false
+ }
+ }
+ if !reuse {
+ a, why, retry, err := c.w.buildArchive(c.rec)
+ if err != nil || why != "" {
+ return builtArchive{}, nil, why, err
+ }
+ if retry {
+ return builtArchive{}, nil, "", errors.New("a stopped file changed while it was archived")
+ }
+ if a.Plain != plain {
+ return builtArchive{}, nil, "", fmt.Errorf("the stopped files' digest is %s, recorded %s", a.Plain, plain)
+ }
+ step = cutArchiveStep{builtArchive: a}
+ }
+ body, err := json.Marshal(step)
+ if err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ j.Mark(cutStepArchive, string(body))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ if err := c.fault("archive-journaled"); err != nil {
+ return builtArchive{}, nil, "", err
+ }
+ }
+ a := step.builtArchive
+ files, err := c.publishCutArchive(ctx, own, vs, step, ledger)
+ return a, files, "", err
+}
+
+// publishCutArchive is cutArchive's PUTs from a journalled identity.
+func (c *cutRun) publishCutArchive(ctx context.Context, own lease.MirrorStore, vs lease.VersionedStore, step cutArchiveStep, ledger moveLedger) ([]moves.SealedObject, error) {
+ cfg := c.cfg()
+ host, id := c.rec.Host, c.g.ID
+ a := step.builtArchive
+ snap := moves.SnapshotsPrefix(host, id)
+ sealedSnap, manifestSnap := snap+a.Name+".sealed", snap+a.Name+".json"
+ onDisk := sealedOnDisk(a)
+ var manifest []byte
+ if onDisk {
+ var err error
+ if manifest, err = os.ReadFile(a.manifestPath()); err != nil {
+ return nil, err
+ }
+ fp, ok := vs.(lease.FilePutIfAbsent)
+ if !ok {
+ return nil, errors.New("this box cannot write the archive create-only")
+ }
+ if _, err := fp.PutFileIfAbsent(ctx, cfg.OwnBucket, sealedSnap, a.sealedPath()); err != nil && !errors.Is(err, lease.ErrExists) {
+ return nil, fmt.Errorf("snapshot the sealed archive: %w", err)
+ }
+ if _, err := vs.PutIfAbsent(ctx, cfg.OwnBucket, manifestSnap, manifest); err != nil && !errors.Is(err, lease.ErrExists) {
+ return nil, fmt.Errorf("snapshot the archive manifest: %w", err)
+ }
+ if err := c.fault("archive-own-put"); err != nil {
+ return nil, err
+ }
+ }
+ if step.Reuse {
+ ref := ledger.ArchiveVersions
+ if ref == nil || !ref.Pinned() || ledger.ArchiveSealed != a.Sealed {
+ return nil, fmt.Errorf("the ledger no longer names the archive %s", a.Name)
+ }
+ msize := int64(len(manifest))
+ if !onDisk {
+ msize = c.snapshotSize(ctx, vs, manifestSnap)
+ }
+ return []moves.SealedObject{
+ {Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize},
+ {Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: msize},
+ }, nil
+ }
+ prefix := moves.FilesPrefix(host, id)
+ sealedKey, manifestKey := prefix+a.Name+".sealed", prefix+a.Name+".json"
+ var sv, mv string
+ var msize int64
+ var err error
+ if onDisk {
+ if sv, err = c.gm.PutFileVersioned(ctx, c.gm.dst, sealedKey, a.sealedPath(), a.SealedSize, digestHex(a.Sealed)); err != nil {
+ return nil, fmt.Errorf("stage the sealed archive: %w", err)
+ }
+ if err := c.fault("archive-first-copy"); err != nil {
+ return nil, err
+ }
+ if mv, err = c.gm.PutVersioned(ctx, c.gm.dst, manifestKey, manifest); err != nil {
+ return nil, fmt.Errorf("stage the archive manifest: %w", err)
+ }
+ msize = int64(len(manifest))
+ } else {
+ // The file is gone, the snapshot is there: a server-side copy from
+ // it, guarded, as the replica lane copies.
+ objs, err := own.List(ctx, cfg.OwnBucket, snap)
+ if err != nil {
+ return nil, fmt.Errorf("list the snapshots: %w", err)
+ }
+ so, sok := objs[sealedSnap]
+ mo, mok := objs[manifestSnap]
+ if !sok || !mok {
+ return nil, fmt.Errorf("the snapshot of %s is incomplete", a.Name)
+ }
+ if _, sv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, sealedSnap, so.ETag, c.gm.dst, sealedKey); err != nil {
+ return nil, fmt.Errorf("stage the sealed archive: %w", err)
+ }
+ if err := c.fault("archive-first-copy"); err != nil {
+ return nil, err
+ }
+ if _, mv, err = c.gm.CopyVersioned(ctx, cfg.OwnBucket, manifestSnap, mo.ETag, c.gm.dst, manifestKey); err != nil {
+ return nil, fmt.Errorf("stage the archive manifest: %w", err)
+ }
+ a.SealedSize, msize = so.Size, mo.Size
+ }
+ if err := c.fault("archive-second-copy"); err != nil {
+ return nil, err
+ }
+ if !lease.UsableVersionID(sv) || !lease.UsableVersionID(mv) {
+ return nil, fmt.Errorf("the staged archive has no usable version ids (%q, %q)", sv, mv)
+ }
+ log.Printf("moves: %s: staged the cut's archive %s", host, a.Name)
+ return []moves.SealedObject{{Key: sealedKey, VersionID: sv, Size: a.SealedSize}, {Key: manifestKey, VersionID: mv, Size: msize}}, nil
+}
+
+// sealedOnDisk reports whether the archive's sealed file and manifest are
+// on this disk and the sealed file hashes to its name.
+func sealedOnDisk(a builtArchive) bool {
+ if _, err := os.Stat(a.manifestPath()); err != nil {
+ return false
+ }
+ sum, err := fileSHA256(a.sealedPath())
+ return err == nil && sum == a.Sealed
+}
+
+// keptManifest reads the manifest of an archive kept on disk; it must
+// describe that archive.
+func keptManifest(a builtArchive) (archive.Manifest, bool) {
+ var m archive.Manifest
+ body, err := os.ReadFile(a.manifestPath())
+ if err != nil || json.Unmarshal(body, &m) != nil || m.Sealed != a.Sealed || m.Plain != a.Plain {
+ return m, false
+ }
+ return m, true
+}
+
+// snapshotPresent reports whether move-snapshots/ holds both objects of the
+// archive named base (key without suffix), current and not deleted.
+func (c *cutRun) snapshotPresent(ctx context.Context, vs lease.VersionedStore, base string) bool {
+ return c.snapshotSize(ctx, vs, base+".sealed") >= 0 && c.snapshotSize(ctx, vs, base+".json") >= 0
+}
+
+// snapshotSize is the current size of key in this box's bucket, or -1.
+func (c *cutRun) snapshotSize(ctx context.Context, vs lease.VersionedStore, key string) int64 {
+ vers, err := vs.ListVersions(ctx, c.cfg().OwnBucket, key)
+ if err != nil {
+ return -1
+ }
+ for _, v := range vers {
+ if v.Key == key && v.IsLatest && !v.DeleteMarker {
+ return v.Size
+ }
+ }
+ return -1
+}
+
+// archived (§7.4, §7.5): the flip. Its precondition is re-read from the
+// claim: the seal names the archive at the record's N, and the record's
+// digests are the seal's archive; otherwise this is a replay of staged's
+// tail, which is redone (never a flip on a seal without the archive). Then
+// the record flips (one CAS: region, state landing, the landing deadline)
+// and the claim follows: landing/checking with the target executing,
+// citing the source's evidence (hostclaim.Flip). The claim half is tried
+// once here; anyone who sees the record ahead writes it (the console's
+// recovery pass, Plan 5).
+func (c *cutRun) archived(ctx context.Context) (bool, error) {
+ cfg := c.cfg()
+ src := c.rec.Move.Source
+ if src == nil || src.Files == nil {
+ return false, fmt.Errorf("move %s: the record has no source baselines", c.g.ID)
+ }
+ obs, err := c.sealedClaim(ctx, src.Evidence, hostclaim.PhaseSealed)
+ if err != nil {
+ return false, err
+ }
+ seal, err := journalledSeal(obs.Claim)
+ if err != nil {
+ return false, err
+ }
+ if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
+ log.Printf("moves: %s: archived without a seal naming the archive; redoing the archive", c.rec.Host)
+ if _, ok, err := c.sealArchive(ctx); err != nil || !ok {
+ return false, err
+ }
+ if obs, err = c.claim(ctx); err != nil {
+ return false, err
+ }
+ if seal, err = journalledSeal(obs.Claim); err != nil {
+ return false, err
+ }
+ if !seal.HasFiles() || seal.N != src.FinalTXID || !sealNamesArchive(seal, src.Files.Sealed) {
+ return false, fmt.Errorf("move %s: the record's archive %q is not the one the seal names", c.g.ID, src.Files.Sealed)
+ }
+ }
+ targetBox := obs.Claim.Op.Intent["target"]
+ holder := *obs.Claim.Holder
+ now := cfg.Now()
+ ok, err := c.cutTransition(ctx, "the flip", provision.MoveFlip(c.g, now, now.Add(cutLandingWindow)))
+ if !ok {
+ return ok, err
+ }
+ log.Printf("moves: %s: flipped to %s", c.rec.Host, c.rec.Move.To)
+ if err := c.fault("after-record-flip"); err != nil {
+ return true, err
+ }
+ if _, err := hostclaim.Update(ctx, c.s, obs, hostclaim.Flip(c.g.ID, targetBox, holder.Stamp, holder.Generation)); err != nil {
+ log.Printf("moves: %s: the claim half of the flip waits for the recovery pass: %v", c.rec.Host, err)
+ }
+ nudgeSerial(ctx, c.s)
+ return true, nil
+}
+
+// sealNamesArchive reports whether the seal's files are the archive with
+// the sealed digest given.
+func sealNamesArchive(seal moves.Seal, sealed string) bool {
+ if sealed == "" {
+ return false
+ }
+ name := archiveName(sealed)
+ n := 0
+ for _, o := range seal.Files {
+ base := filepath.Base(o.Key)
+ if base == name+".sealed" || base == name+".json" {
+ n++
+ }
+ }
+ return n == 2 && len(seal.Files) == 2
+}
+
+// revertClaim is the claim half of a revert (§7.4 reverting, §7.5): once
+// the source has given up execution, the claim goes to reverting under
+// the source with the staging lane's park queued (hostclaim.Revert). While
+// the source still executes (a revert from stopping, or from stopped
+// before exec went to none) there is nothing to hand back, and the claim
+// stays moving: the recovery (Plan 4) starts from there. A claim already
+// reverting for this move is done.
+//
+// The park's evidence_at is the writers' horizon (G7): the later of the
+// seal's bound and this decision's time plus a signature's life and the
+// skew, journalled once so a replay names the same scope.
+func (w *movesWorker) revertClaim(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ id := rec.Move.ID
+ obs, err := hostclaim.Read(ctx, s, rec.Host)
+ if err != nil {
+ return fmt.Errorf("read the host claim: %w", err)
+ }
+ if obs.Absent || obs.Claim.Op.ID != id {
+ return nil
+ }
+ op := obs.Claim.Op
+ if op.Kind == hostclaim.KindReverting || op.Kind != hostclaim.KindMoving || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceSource] == "" {
+ return nil
+ }
+ if obs.Claim.Holder == nil || cfg.Box == "" {
+ return errors.New("the revert needs the holder and this box's label")
+ }
+ j, err := moves.ReadJournal(cfg.DataDir, rec.Host, id)
+ if err != nil {
+ return err
+ }
+ var bound time.Time
+ if raw, ok := j.Steps[cutStepRevertBound]; ok {
+ if bound, err = time.Parse(time.RFC3339Nano, raw); err != nil {
+ bound = time.Time{}
+ }
+ }
+ if bound.IsZero() {
+ bound = cfg.Now().Add(moves.SignatureLife + moves.DefaultSkew).UTC()
+ if seal, err := journalledSeal(obs.Claim); err == nil && seal.Bound().After(bound) {
+ bound = seal.Bound().UTC()
+ }
+ j.Mark(cutStepRevertBound, bound.Format(time.RFC3339Nano))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return err
+ }
+ }
+ scope := hostclaim.ParkScope{Location: moves.ScopeLocation(rec.Host, id), Incarnation: obs.Claim.Holder.Incarnation, EvidenceAt: bound, RequestLimit: cutRequestLimit}
+ if _, err := hostclaim.Update(ctx, s, obs, hostclaim.Revert(id, cfg.Box, rec.Move.To, &scope, obligationID)); err != nil {
+ return fmt.Errorf("revert the claim: %w", err)
+ }
+ log.Printf("moves: %s: the claim reverts to this box; the staging lane in %s is queued for its park", rec.Host, rec.Move.To)
+ return nil
+}
+
+// obligationID mints a claim obligation or retirement id.
+func obligationID() string {
+ var b [16]byte
+ if _, err := rand.Read(b[:]); err != nil {
+ return ""
+ }
+ return hex.EncodeToString(b[:])
+}
+
+// regularFile reports whether path is a regular file.
+func regularFile(path string) bool {
+ st, err := os.Lstat(path)
+ return err == nil && st.Mode().IsRegular()
+}
+
+// fileSHA256 is "sha256:<hex>" of the file at path.
+func fileSHA256(path string) (string, error) {
+ f, err := os.Open(path)
+ if err != nil {
+ return "", err
+ }
+ defer f.Close()
+ h := sha256.New()
+ if _, err := io.Copy(h, f); err != nil {
+ return "", err
+ }
+ return "sha256:" + hex.EncodeToString(h.Sum(nil)), nil
+}
+
+// holdMoving is moveCfg.HoldMoving in production: the row held "moving"
+// under op, conditional on its current hold (SetHoldIf) and under the
+// host's hold lock, which reconcile's hold decision and the landing's open
+// share (Task 6). A row already held moving by op is left as it is; a row
+// held by another op is refused. SQLITE_BUSY while root freezes holds is
+// an error the next pass retries (G30). A write that landed touches the
+// stage-requested marker.
+func holdMoving(reg *registry.Registry, written func(host string)) func(host, op string) error {
+ return func(host, op string) error {
+ var err error
+ provision.WithHostHoldLock(host, func() {
+ var rt *registry.Route
+ if rt, err = reg.Get(host); err != nil {
+ return
+ }
+ if rt == nil {
+ err = fmt.Errorf("%s has no registry row on this box", host)
+ return
+ }
+ if rt.HoldOp == op && rt.Status == "moving" {
+ return
+ }
+ if rt.HoldOp != "" && rt.HoldOp != op {
+ err = fmt.Errorf("%s is held %s by %s", host, rt.Status, rt.HoldOp)
+ return
+ }
+ var held bool
+ if held, err = reg.SetHoldIf(host, "moving", op, rt.HoldOp); err != nil {
+ return
+ }
+ if !held {
+ err = fmt.Errorf("%s: the row's hold changed under the cut", host)
+ return
+ }
+ if written != nil {
+ written(host)
+ }
+ })
+ return err
+ }
+}
+
+// localLTXPosition is moveCfg.LocalPosition in production: the stopped
+// database's TXID, the newest level-0 file in the activator's litestream
+// meta directory (lease.Held's SetMetaPath, litestream's DB.MaxLTX). No
+// directory is position 0.
+func localLTXPosition(dbPath string) (uint64, error) {
+ dir := filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream", "ltx", "0")
+ ents, err := os.ReadDir(dir)
+ if errors.Is(err, os.ErrNotExist) {
+ return 0, nil
+ }
+ if err != nil {
+ return 0, err
+ }
+ var newest ltx.TXID
+ for _, e := range ents {
+ if _, maxTX, err := ltx.ParseFilename(e.Name()); err == nil && maxTX > newest {
+ newest = maxTX
+ }
+ }
+ return uint64(newest), nil
+}
+
+// finalCheckpoint is moveCfg.FinalCheckpoint in production (§5.3 step 5,
+// G3): the lease at the claim holder's location, bound to its generation's
+// prefix with check as the authority read before every replica request,
+// one checkpoint, and the lease released.
+func finalCheckpoint(bucket, region string) func(ctx context.Context, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
+ return func(ctx context.Context, dbPath string, h hostclaim.Holder, check func(context.Context) error) error {
+ held, err := lease.Acquire(ctx, bucket, strings.TrimSuffix(h.Location, "/"), region)
+ if err != nil {
+ return err
+ }
+ defer func() {
+ rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
+ defer cancel()
+ if err := held.Release(rctx); err != nil {
+ log.Printf("moves: release the checkpoint lease at %s: %v", h.Location, err)
+ }
+ }()
+ if g := h.Generation; g != "" && g != hostclaim.RootGeneration {
+ prefix, _ := moves.ReplicaSource(h)
+ if err := held.BindGeneration(prefix, check); err != nil {
+ return err
+ }
+ } else if err := check(ctx); err != nil {
+ return err
+ }
+ _, err = held.Checkpoint(ctx, dbPath)
+ return err
+ }
+}
diff --git a/internal/host/movescut_test.go b/internal/host/movescut_test.go
new file mode 100644
index 000000000..1b5adc465
--- /dev/null
+++ b/internal/host/movescut_test.go
@@ -0,0 +1,921 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "path/filepath"
+ "sort"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The cut harness: the Phase 1 harness driven to ready (manual cut) and
+// cut as the door does, with the cut's hooks faked. fakeEvidence answers a
+// configurable report; the final checkpoint writes the stopped database's
+// last transaction (TXID 3) into the source's replica; cutFault is the
+// crash seam, crashing once at the point named.
+
+var errCrash = errors.New("injected crash")
+
+type fakeEvidence struct {
+ calls int
+ stopped bool
+ err error
+ onCall func()
+}
+
+func (e *fakeEvidence) report() moves.StoppedReport {
+ r := moves.StoppedReport{V: 1, Host: p1Host, Steps: map[string]moves.Check{}, CheckedAt: time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)}
+ for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
+ r.Steps[id] = moves.Check{OK: true}
+ }
+ if !e.stopped {
+ r.Steps[moves.StepFDs] = moves.Check{Reason: "1 process(es) hold the database open"}
+ }
+ return r
+}
+
+type cutHarness struct {
+ *phase1Harness
+ ev *fakeEvidence
+ holds []string
+ fences int
+ drains int
+ checkpoints int
+ locations []string
+ bundle map[string]string
+ localPos uint64
+ timers []registry.Timer
+ crashAt string
+ crashed bool
+ hits []string
+}
+
+// PutFileIfAbsent makes the fake a lease.FilePutIfAbsent.
+func (v *fakeVersioned) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) {
+ body, err := os.ReadFile(path)
+ if err != nil {
+ return "", err
+ }
+ return v.PutIfAbsent(ctx, bucket, key, body)
+}
+
+var _ lease.FilePutIfAbsent = (*fakeVersioned)(nil)
+
+const lastTXKey = "0000/0000000000000003-0000000000000003.ltx"
+
+func newCutHarness(t *testing.T) *cutHarness {
+ t.Helper()
+ h := &cutHarness{phase1Harness: newPhase1Harness(t), ev: &fakeEvidence{stopped: true}, localPos: 3,
+ bundle: map[string]string{"default": "e1", "environment": ""}}
+ h.timers = []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}}
+ loc := "instances/" + p1Host + "/" + h.inc + "/"
+ last := ltxForTest(t, 4096, 4, 3, 3)
+ if err := os.WriteFile(h.route.DBPath, []byte("the stopped database"), 0o600); err != nil {
+ t.Fatal(err)
+ }
+ h.cfg.HoldMoving = func(host, op string) error { h.holds = append(h.holds, host+"="+op); return nil }
+ h.cfg.Fence = func(context.Context, string, uint64) error { h.fences++; return nil }
+ h.cfg.TimersDrained = func(context.Context, string) error { h.drains++; return nil }
+ h.cfg.Evidence = func(_ context.Context, host, location string, _ []string) (moves.StoppedReport, error) {
+ h.ev.calls++
+ h.locations = append(h.locations, location)
+ if h.ev.onCall != nil {
+ h.ev.onCall()
+ }
+ return h.ev.report(), h.ev.err
+ }
+ h.cfg.FinalCheckpoint = func(ctx context.Context, _ string, holder hostclaim.Holder, check func(context.Context) error) error {
+ if err := check(ctx); err != nil {
+ return err
+ }
+ h.checkpoints++
+ prefix, _ := moves.ReplicaSource(holder)
+ if prefix != loc+"g-cur/" {
+ t.Errorf("checkpoint at %s, want the claim's lineage %sg-cur/", prefix, loc)
+ }
+ h.f.store("data-ie", prefix+lastTXKey, fakeMoveObject{lease.MirrorObject{ETag: "cur-c", Size: int64(len(last))}, string(last)})
+ return nil
+ }
+ h.cfg.LocalPosition = func(string) (uint64, error) { return h.localPos, nil }
+ h.cfg.Timers = func(string) ([]registry.Timer, error) { return h.timers, nil }
+ h.cfg.InputVersions = func(context.Context, release.Store, provision.Record) (map[string]string, string, error) {
+ out := map[string]string{}
+ for k, v := range h.bundle {
+ out[k] = v
+ }
+ return out, "", nil
+ }
+ h.cfg.cutFault = func(point string) error {
+ h.hits = append(h.hits, point)
+ if point == h.crashAt && !h.crashed {
+ h.crashed = true
+ return errCrash
+ }
+ return nil
+ }
+ return h
+}
+
+// targetCfg composes with the harness's current bundle versions.
+func (h *cutHarness) target() moveCfg {
+ c := h.targetCfg()
+ c.ComposeHeldEnv = func(context.Context, release.Store, provision.Record, []byte) (map[string]string, []string, provision.MoveInputs, error) {
+ env := map[string]string{}
+ for k, v := range h.env {
+ env[k] = v
+ }
+ env["CARLOS_REGION"] = "uk"
+ b := map[string]string{}
+ for k, v := range h.bundle {
+ b[k] = v
+ }
+ return env, nil, provision.MoveInputs{BundleETags: b}, nil
+ }
+ return c
+}
+
+// ready drives the move to a persisted ready with its inputs pinned.
+func (h *cutHarness) ready(t *testing.T) {
+ t.Helper()
+ h.start(t, "manual")
+ h.pass(t)
+ h.targetPass(t, h.target())
+ h.pass(t)
+ if got := h.read(t, p1Host).Move; got.State != provision.MoveReady || got.Inputs == nil {
+ t.Fatalf("the fixture did not reach ready: %+v", got)
+ }
+}
+
+// cut is the console's cut door: ready to cutting/requested.
+func (h *cutHarness) cut(t *testing.T) {
+ t.Helper()
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (h *cutHarness) move(t *testing.T) *provision.MoveState {
+ t.Helper()
+ return h.read(t, p1Host).Move
+}
+
+// crashPass runs one pass that must crash at h.crashAt.
+func (h *cutHarness) crashPass(t *testing.T) {
+ t.Helper()
+ h.pass(t)
+ if !h.crashed {
+ t.Fatalf("the pass never reached %s (points reached: %v)", h.crashAt, h.hits)
+ }
+}
+
+// freshWorker drops the source worker: the next pass starts with no
+// in-memory state, as a restarted agent does.
+func (h *cutHarness) freshWorker() { h.worker1 = nil }
+
+func (h *cutHarness) stoppedKey() string {
+ return moves.StoppedKey(moves.ReportPrefix("bab", "app", p1Host, p1ID), "source")
+}
+
+func (h *cutHarness) seal(t *testing.T) moves.Seal {
+ t.Helper()
+ s, err := journalledSeal(h.claim(t))
+ if err != nil {
+ t.Fatal(err)
+ }
+ return s
+}
+
+func (h *cutHarness) journalArchive(t *testing.T) cutArchiveStep {
+ t.Helper()
+ j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var step cutArchiveStep
+ if err := json.Unmarshal([]byte(j.Steps[cutStepArchive]), &step); err != nil {
+ t.Fatalf("no archive identity journalled: %v", err)
+ }
+ return step
+}
+
+// Admission needs this round's own passing source report, a fresh passing
+// target report, and the pin to match (G1, G13). With all of them the
+// record is admitted with cut_at, the budget and the deadline; a stale
+// target report or a failing source sends it back to ready with the pin
+// cleared, and a passing round pins again.
+func TestCutAdmitsOnlyFreshPassingReports(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.ev.stopped = false // stop at stopping, to look at the admission
+ h.cut(t)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping || m.CutAt == nil || m.CutBudgetS != int(moves.CutWindowDefault) || m.CutDeadline == nil {
+ t.Fatalf("after the admission: %+v", m)
+ }
+ if want := moves.CutDeadline(*m.CutAt, moves.CutWindowDefault); !m.CutDeadline.Equal(want) {
+ t.Fatalf("deadline %v, want %v", m.CutDeadline, want)
+ }
+
+ for name, spoil := range map[string]func(h *cutHarness){
+ "the target's report is stale": func(h *cutHarness) { h.clock.Advance(moves.ReportFreshness + time.Minute) },
+ "the source does not answer": func(h *cutHarness) { h.probe = moves.Check{Reason: s1NoAnswer(p1Host)} },
+ "a second box in the target region": func(h *cutHarness) {
+ if _, err := h.s.Put(context.Background(), "control/edges/uk-box2.json",
+ []byte(`{"v":1,"box":"uk-box2","eip":"1.2.3.9","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
+ panic(err)
+ }
+ },
+ } {
+ t.Run(name, func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.cut(t)
+ spoil(h)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.CutAt != nil {
+ t.Fatalf("not back to ready: %+v", m)
+ }
+ if h.ev.calls != 0 || len(h.holds) != 0 {
+ t.Fatalf("something stopped: %d evidence calls, holds %v", h.ev.calls, h.holds)
+ }
+ })
+ }
+}
+
+// The inputs pinned at readiness are compared at requested and again at
+// admitted (§7.4). A drift goes back to ready; the target reports the new
+// versions, a passing round pins them (MoveGuardedRepin), and the next cut
+// admits: no loop.
+func TestCutGoesBackToReadyWhenInputsMove(t *testing.T) {
+ ctx := context.Background()
+ h := newCutHarness(t)
+ h.ready(t)
+ h.ev.stopped = false
+ h.bundle["environment"] = "e2"
+ h.cut(t)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil {
+ t.Fatalf("a drift at requested did not go back to ready: %+v", m)
+ }
+ // Ready without a pin is never cut: the door refuses.
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); !errors.Is(err, provision.ErrMoveConflict) {
+ t.Fatalf("cut without a pin: %v", err)
+ }
+ h.targetPass(t, h.target())
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveReady || m.Inputs == nil || m.Inputs.BundleETags["environment"] != "e2" {
+ t.Fatalf("not pinned again: %+v", m)
+ }
+ h.cut(t)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping {
+ t.Fatalf("the repinned cut did not admit: %+v", m)
+ }
+
+ // At admitted: a record admitted before the drift goes back to ready
+ // on the replay of admitted.
+ h2 := newCutHarness(t)
+ h2.ready(t)
+ h2.cut(t)
+ now := h2.clock.Now()
+ if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120))); err != nil {
+ t.Fatal(err)
+ }
+ h2.bundle["default"] = "e9"
+ h2.pass(t)
+ if m := h2.move(t); m.State != provision.MoveReady || m.Inputs != nil || h2.ev.calls != 0 {
+ t.Fatalf("a drift at admitted: %+v, %d evidence calls", m, h2.ev.calls)
+ }
+}
+
+// Review Focus 1: a fresh worker on the same store, with the record at
+// stopping and nothing in memory, holds, fences, drains, asks for the
+// evidence and advances; the cut then runs to the flip in one pass.
+func TestCutResumesStoppingFromEvidenceAlone(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.ev.stopped = false
+ h.cut(t)
+ h.pass(t)
+ h.pass(t)
+ if m := h.move(t); m.CutPhase != provision.CutStopping || h.ev.calls != 2 {
+ t.Fatalf("an unproven stop advanced: %+v after %d calls", m, h.ev.calls)
+ }
+ if h.claim(t).Op.Exec != "ie-box" {
+ t.Fatal("execution moved before the evidence")
+ }
+ h.freshWorker()
+ h.ev.stopped = true
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped {
+ t.Fatalf("the resumed cut: %+v", m)
+ }
+ if h.ev.calls != 3 || h.fences != 3 || h.drains != 3 || len(h.holds) != 3 || h.holds[0] != p1Host+"="+p1ID {
+ t.Fatalf("evidence %d, fences %d, drains %d, holds %v", h.ev.calls, h.fences, h.drains, h.holds)
+ }
+ for _, loc := range h.locations {
+ if loc != "instances/"+p1Host+"/"+h.inc {
+ t.Fatalf("the lease step read %s, not the claim holder's location", loc)
+ }
+ }
+}
+
+// K3 and K13: the stopped report exists, complete with the replica step
+// and N, before the claim's execution is none. A crash after the
+// checkpoint leaves execution with the source and no report; a crash after
+// the report leaves one report, which the replay reuses as it is.
+func TestStoppedReportIsCompleteBeforeExecToNone(t *testing.T) {
+ ctx := context.Background()
+ for _, point := range []string{"after-checkpoint", "after-report", "after-record-source", "after-exec-none"} {
+ t.Run(point, func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.crashAt = point
+ h.cut(t)
+ h.crashPass(t)
+ if m := h.move(t); m.CutPhase != provision.CutStopped {
+ t.Fatalf("after the crash: %+v", m)
+ }
+ body, _, err := h.s.Get(ctx, h.stoppedKey())
+ exec := h.claim(t).Op.Exec
+ switch point {
+ case "after-checkpoint":
+ if !errors.Is(err, os.ErrNotExist) || exec != "ie-box" {
+ t.Fatalf("report %v, exec %q", err, exec)
+ }
+ default:
+ var r moves.StoppedReport
+ if err != nil || json.Unmarshal(body, &r) != nil || !r.Complete() || r.FinalTXID != 3 || r.DBSHA256 == "" || r.TimersSHA256 == "" || r.FilesPlain == "" {
+ t.Fatalf("report %v: %+v", err, r)
+ }
+ if want := point == "after-exec-none"; (exec == "") != want {
+ t.Fatalf("exec %q at %s", exec, point)
+ }
+ }
+ h.freshWorker()
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveLanding {
+ t.Fatalf("the replay did not finish: %+v", m)
+ }
+ after, _, err := h.s.Get(ctx, h.stoppedKey())
+ if err != nil {
+ t.Fatal(err)
+ }
+ if point != "after-checkpoint" && string(after) != string(body) {
+ t.Fatalf("the replay published a second report:\n%s\n%s", body, after)
+ }
+ keys, _ := h.s.List(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID))
+ n := 0
+ for _, k := range keys {
+ if strings.Contains(k, "stopped-") {
+ n++
+ }
+ }
+ if n != 1 {
+ t.Fatalf("%d stopped reports: %v", n, keys)
+ }
+ src := h.move(t).Source
+ if src == nil || src.FinalTXID != 3 || src.Evidence != h.stoppedKey() || src.Files == nil || src.Files.Plain == "" || src.Files.Sealed == "" {
+ t.Fatalf("source baselines: %+v", src)
+ }
+ c := h.claim(t)
+ if c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() || c.Op.Journal[hostclaim.JournalCutSession] == "" {
+ t.Fatalf("claim journal: %v", c.Op.Journal)
+ }
+ })
+ }
+}
+
+// Review Focus 2: three final rounds that defer revert the cut, and the
+// claim follows once execution is none: reverting under the source, the
+// lane's park queued with the writers' horizon.
+func TestCutRevertsAfterThreeDeferredRounds(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ n := 0
+ h.f.beforeCopy = func(srcKey string) {
+ if strings.HasSuffix(srcKey, lastTXKey) {
+ // The source changes under the copy: the conditional copy is
+ // refused and the object deferred.
+ n++
+ o := h.f.buckets["data-ie"][srcKey]
+ o.ETag = "changed-" + string(rune('a'+n))
+ h.f.buckets["data-ie"][srcKey] = o
+ }
+ }
+ h.cut(t)
+ for i := 1; i <= 2; i++ {
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopped {
+ t.Fatalf("after deferred round %d: %+v", i, m)
+ }
+ if got := h.ledger(t).DeferredRounds; got != i {
+ t.Fatalf("ledger counts %d deferred rounds, want %d", got, i)
+ }
+ }
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveReverting || m.Reason != provision.RevertDeferredRounds {
+ t.Fatalf("after three deferred rounds: %+v", m)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalRevPhase] != "evidence" {
+ t.Fatalf("claim: %+v", c.Op)
+ }
+ var park *hostclaim.ParkScope
+ for _, o := range c.Obligations {
+ if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" {
+ park = o.Park
+ }
+ }
+ if park == nil || park.Location != moves.ScopeLocation(p1Host, p1ID) || park.Incarnation != h.inc || park.RequestLimit != cutRequestLimit {
+ t.Fatalf("staging park: %+v", park)
+ }
+ if !park.EvidenceAt.After(h.clock.now.Add(moves.SignatureLife)) {
+ t.Fatalf("evidence_at %v is not past every signature's life", park.EvidenceAt)
+ }
+ // A replay changes nothing: the claim is reverting already.
+ h.pass(t)
+ if c2 := h.claim(t); c2.Gen != c.Gen {
+ t.Fatalf("a replay wrote the claim again (%d to %d)", c.Gen, c2.Gen)
+ }
+}
+
+// §7.3 seal by version: the seal names N and the exact version of every
+// current replica object, of the timers and of both archive objects. The
+// witness (wSealByVersion): a stale copy that lands after the seal makes a
+// version the seal does not name, and the seal still names the earlier one.
+func TestSealNamesEveryCurrentVersionAndN(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.cut(t)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveLanding {
+ t.Fatalf("cut did not finish: %+v", m)
+ }
+ seal := h.seal(t)
+ if seal.N != 3 || !seal.ReachesN() || !seal.FromDB || !seal.SrcCovered || len(seal.Timers) != 1 || len(seal.Files) != 2 {
+ t.Fatalf("seal: %+v", seal)
+ }
+ lane := moves.ReplicaPrefix(p1Host, p1ID)
+ want := []string{lane + "0000/0000000000000001-0000000000000002.ltx", lane + "0001/0000000000000001-0000000000000002.ltx", lane + lastTXKey}
+ sort.Strings(want)
+ var got []string
+ for _, o := range seal.Replica {
+ got = append(got, o.Key)
+ if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID {
+ t.Fatalf("%s: sealed %s, current %s", o.Key, o.VersionID, cur)
+ }
+ }
+ sort.Strings(got)
+ if strings.Join(got, ",") != strings.Join(want, ",") {
+ t.Fatalf("sealed %v, want %v", got, want)
+ }
+ for _, o := range append(append([]moves.SealedObject{}, seal.Timers...), seal.Files...) {
+ if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID || !lease.UsableVersionID(o.VersionID) {
+ t.Fatalf("%s: sealed %q, current %q", o.Key, o.VersionID, cur)
+ }
+ }
+ if !seal.Bound().After(seal.SealedAt) {
+ t.Fatalf("bound %v is not after the seal %v", seal.Bound(), seal.SealedAt)
+ }
+ // The witness.
+ stale := seal.Replica[0]
+ if _, _, err := h.f.CopyVersionedAt(context.Background(), h.clock.Now(), "data-ie",
+ "instances/"+p1Host+"/"+h.inc+"/g-cur/0000/0000000000000001-0000000000000002.ltx", "", "data-uk", stale.Key); err != nil {
+ t.Fatal(err)
+ }
+ if cur := h.f.versionOf("data-uk", stale.Key); cur == stale.VersionID {
+ t.Fatal("the stale copy did not make a new version")
+ }
+ if again := h.seal(t); again.Replica[0].VersionID != stale.VersionID {
+ t.Fatal("the seal moved with the stale copy")
+ }
+ // Every staging write of the cut went through the guard.
+ for _, op := range h.f.ops {
+ if strings.Contains(op, "data-uk/") && !strings.HasPrefix(op, "copy ") {
+ t.Fatalf("an unguarded write reached the target bucket: %s", op)
+ }
+ }
+ for _, c := range h.f.stagingCalls() {
+ if c.SignAt.IsZero() || c.Bucket != "data-uk" {
+ t.Fatalf("unguarded or misdirected request: %+v", c)
+ }
+ }
+}
+
+// writeOnStop makes the instance write a file just before it is stopped:
+// the stopped files differ from the archive the target verified, so the
+// cut builds its own.
+func (h *cutHarness) writeOnStop(t *testing.T) {
+ h.ev.onCall = func() {
+ if h.ev.calls == 1 {
+ if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "late.txt"), []byte("written before the stop"), 0o644); err != nil {
+ t.Error(err)
+ }
+ }
+ }
+}
+
+func snapshotSealed(h *cutHarness) []string {
+ var out []string
+ for k := range h.f.buckets["data-ie"] {
+ if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) && strings.HasSuffix(k, ".sealed") {
+ out = append(out, k)
+ }
+ }
+ sort.Strings(out)
+ return out
+}
+
+// codex 19, with the real randomized sealer: the archive's identity is
+// journalled before any PUT, and every replay publishes that same key. A
+// replay with the sealed file gone and nothing published reseals under a
+// new key, and the record names the new one; one with the file gone and
+// the snapshot published reuses it.
+func TestCutArchiveIsCreateOnlyAndReused(t *testing.T) {
+ for _, point := range []string{"archive-journaled", "archive-own-put", "archive-first-copy"} {
+ t.Run("crash "+point, func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.writeOnStop(t)
+ h.crashAt = point
+ h.cut(t)
+ h.crashPass(t)
+ if m := h.move(t); m.CutPhase != provision.CutStaged {
+ t.Fatalf("after the crash: %+v", m)
+ }
+ id := h.journalArchive(t)
+ if id.Reuse {
+ t.Fatal("the stopped files changed, yet the preparing archive was reused")
+ }
+ h.freshWorker()
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveLanding || m.Source.Files.Sealed != id.Sealed {
+ t.Fatalf("the replay: %+v, files %+v, journalled %s", m, m.Source.Files, id.Sealed)
+ }
+ if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], id.Name+".sealed") {
+ t.Fatalf("snapshots %v, want only %s", got, id.Name)
+ }
+ if !sealNamesArchive(h.seal(t), id.Sealed) {
+ t.Fatalf("the seal names %+v, not %s", h.seal(t).Files, id.Name)
+ }
+ })
+ }
+ t.Run("file gone, nothing published", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.writeOnStop(t)
+ h.crashAt = "archive-journaled"
+ h.cut(t)
+ h.crashPass(t)
+ first := h.journalArchive(t)
+ if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveLanding || m.Source.Files.Sealed == first.Sealed {
+ t.Fatalf("not resealed: %+v files %+v", m, m.Source.Files)
+ }
+ second := h.journalArchive(t)
+ if second.Sealed != m.Source.Files.Sealed || !sealNamesArchive(h.seal(t), second.Sealed) {
+ t.Fatalf("journal %s, record %s", second.Sealed, m.Source.Files.Sealed)
+ }
+ if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], second.Name+".sealed") {
+ t.Fatalf("snapshots %v", got)
+ }
+ })
+ t.Run("file gone, snapshot published", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.writeOnStop(t)
+ h.crashAt = "archive-own-put"
+ h.cut(t)
+ h.crashPass(t)
+ first := h.journalArchive(t)
+ if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
+ t.Fatalf("not reused: %+v files %+v", m, m.Source.Files)
+ }
+ // Staged by a server-side copy from the snapshot.
+ copied := 0
+ for _, c := range h.f.stagingCalls() {
+ if c.Op == "copy" && strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)+first.Name) {
+ copied++
+ }
+ }
+ if copied != 2 {
+ t.Fatalf("%d copies from the snapshot, want 2", copied)
+ }
+ })
+ t.Run("unchanged files reuse the verified archive", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ staged := h.ledger(t)
+ h.cut(t)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveLanding || m.Source.Files.Sealed != staged.ArchiveSealed {
+ t.Fatalf("the verified archive was not reused: %+v (ledger %s)", m.Source.Files, staged.ArchiveSealed)
+ }
+ seal := h.seal(t)
+ for _, o := range seal.Files {
+ if o.VersionID != staged.ArchiveVersions.SealedVersion && o.VersionID != staged.ArchiveVersions.ManifestVersion {
+ t.Fatalf("the seal names %s@%s, not the verified versions", o.Key, o.VersionID)
+ }
+ }
+ if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], archiveName(staged.ArchiveSealed)+".sealed") {
+ t.Fatalf("snapshots %v", got)
+ }
+ })
+}
+
+// G8: archived is written only after the seal names the archive. A crash
+// after each of the archive's PUTs leaves the record staged; the replay
+// completes and the seal names both files/ versions before archived. A
+// record hand-edited to archived whose seal has no files does not flip.
+func TestArchivedNeverPrecedesTheSealedArchive(t *testing.T) {
+ for _, point := range []string{"archive-own-put", "archive-first-copy", "archive-second-copy", "after-archive-seal"} {
+ t.Run(point, func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.writeOnStop(t)
+ h.crashAt = point
+ h.cut(t)
+ h.crashPass(t)
+ if m := h.move(t); m.CutPhase != provision.CutStaged {
+ t.Fatalf("archived before the seal named the archive: %+v", m)
+ }
+ if point != "after-archive-seal" && h.seal(t).HasFiles() {
+ t.Fatal("the seal names files before the copies finished")
+ }
+ h.pass(t)
+ m := h.move(t)
+ seal := h.seal(t)
+ if m.State != provision.MoveLanding || !sealNamesArchive(seal, m.Source.Files.Sealed) {
+ t.Fatalf("after the replay: %+v, seal files %+v", m, seal.Files)
+ }
+ for _, o := range seal.Files {
+ if !strings.HasPrefix(o.Key, moves.FilesPrefix(p1Host, p1ID)) || h.f.versionOf("data-uk", o.Key) != o.VersionID {
+ t.Fatalf("seal names %s@%s", o.Key, o.VersionID)
+ }
+ }
+ })
+ }
+ t.Run("hand-edited to archived", func(t *testing.T) {
+ ctx := context.Background()
+ h := newCutHarness(t)
+ h.ready(t)
+ h.crashAt = "archive-journaled"
+ h.cut(t)
+ h.crashPass(t)
+ if h.seal(t).HasFiles() {
+ t.Fatal("fixture: the seal already names files")
+ }
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
+ rec.Move.CutPhase = provision.CutArchived
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveCutting || m.CutPhase != provision.CutArchived || h.read(t, p1Host).Region != "ie" {
+ t.Fatalf("a hand-edited archived flipped: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
+ t.Fatalf("the claim flipped: %+v", c.Op)
+ }
+ })
+}
+
+// §7.5: the record flips first (region, landing, the landing deadline,
+// checking), then the claim follows: landing/checking with the target
+// executing, the source's stamp and generation journalled, the holder's
+// generation untouched. A crash between the two leaves the record ahead
+// and the claim moving/sealed with nobody executing, which anyone who sees
+// it may finish with the same Flip.
+func TestFlipWritesRecordThenClaim(t *testing.T) {
+ ctx := context.Background()
+ h := newCutHarness(t)
+ h.ready(t)
+ h.cut(t)
+ h.pass(t)
+ rec := h.read(t, p1Host)
+ m := rec.Move
+ if rec.Region != "uk" || m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped || m.LandingPhase != provision.LandChecking || m.FlippedAt == nil || m.LandingDeadline == nil {
+ t.Fatalf("record: region %s %+v", rec.Region, m)
+ }
+ if d := m.LandingDeadline.Sub(*m.FlippedAt); d != cutLandingWindow {
+ t.Fatalf("landing window %v", d)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != "checking" || c.Op.Exec != "uk-box" ||
+ c.Op.Journal[hostclaim.JournalSourceStamp] != "1" || c.Op.Journal[hostclaim.JournalSrcGen] != "cur" || c.Holder.Generation != "cur" {
+ t.Fatalf("claim: %+v holder %+v", c.Op, c.Holder)
+ }
+
+ h2 := newCutHarness(t)
+ h2.ready(t)
+ h2.crashAt = "after-record-flip"
+ h2.cut(t)
+ h2.crashPass(t)
+ if r := h2.read(t, p1Host); r.Region != "uk" || r.Move.State != provision.MoveLanding {
+ t.Fatalf("record not flipped: %+v", r.Move)
+ }
+ c2 := h2.claim(t)
+ if c2.Op.Kind != hostclaim.KindMoving || c2.Op.Phase != hostclaim.PhaseSealed || c2.Op.Exec != "" {
+ t.Fatalf("claim after the crash: %+v", c2.Op)
+ }
+ obs, err := hostclaim.Read(ctx, h2.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(ctx, h2.s, obs, hostclaim.Flip(p1ID, "uk-box", obs.Claim.Holder.Stamp, obs.Claim.Holder.Generation)); err != nil {
+ t.Fatalf("the recovery's flip: %v", err)
+ }
+ if c3 := h2.claim(t); c3.Op.Kind != hostclaim.KindLanding || c3.Op.Exec != "uk-box" {
+ t.Fatalf("claim after the recovery: %+v", c3.Op)
+ }
+}
+
+// The deadline (§7.4): at admitted it goes back to ready; at stopping it
+// reverts with "stop evidence budget" and the claim stays with the source,
+// which still executes; after execution was given up it reverts with the
+// deadline's reason and the claim follows to reverting.
+func TestCutDeadlinePassedReverts(t *testing.T) {
+ ctx := context.Background()
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h0inc(t)}
+
+ t.Run("admitted", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.cut(t)
+ now := h.clock.Now()
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120))); err != nil {
+ t.Fatal(err)
+ }
+ h.clock.Advance(2 * time.Minute)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveReady || h.ev.calls != 0 {
+ t.Fatalf("admitted past the deadline: %+v", m)
+ }
+ })
+ t.Run("stopping", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.ev.stopped = false
+ h.cut(t)
+ h.pass(t)
+ h.clock.Advance(2 * time.Minute)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveReverting || m.Reason != provision.RevertStopEvidenceBudget {
+ t.Fatalf("stopping past the deadline: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
+ t.Fatalf("the claim moved without the source's evidence: %+v", c.Op)
+ }
+ })
+ t.Run("stopped, execution given up", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.crashAt = "after-exec-none"
+ h.cut(t)
+ h.crashPass(t)
+ h.clock.Advance(2 * time.Minute)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveReverting || m.Reason != provision.RevertCutDeadline {
+ t.Fatalf("stopped past the deadline: %+v", m)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" {
+ t.Fatalf("claim: %+v", c.Op)
+ }
+ j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
+ if err != nil || j.Steps[cutStepRevertBound] == "" {
+ t.Fatalf("the revert's bound is not journalled: %v %+v", err, j)
+ }
+ })
+ t.Run("the claim half after a crash", func(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.crashAt = "after-exec-none"
+ h.cut(t)
+ h.crashPass(t)
+ // The record decides the revert; the claim half is lost.
+ if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
+ t.Fatal(err)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
+ t.Fatalf("fixture: %+v", c.Op)
+ }
+ h.pass(t)
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
+ t.Fatalf("the reverting arm did not write the claim half: %+v", c.Op)
+ }
+ })
+}
+
+// With the switch off, a cutting record is never touched.
+func TestCutIsInertWithTheSwitchOff(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ h.cut(t)
+ before := h.read(t, p1Host)
+ claimBefore := h.claim(t)
+ h.cfg.Enabled = false
+ h.pass(t)
+ h.pass(t)
+ if after := h.read(t, p1Host); after.Move.CutPhase != before.Move.CutPhase || after.Move.UpdatedAt != before.Move.UpdatedAt {
+ t.Fatalf("the record moved: %+v", after.Move)
+ }
+ if c := h.claim(t); c.Gen != claimBefore.Gen {
+ t.Fatal("the claim was written")
+ }
+ if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 {
+ t.Fatal("the cut ran with the switch off")
+ }
+}
+
+func h0inc(t *testing.T) string {
+ t.Helper()
+ inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return inc
+}
+
+// nullVersions is a staging writer whose copies of matching keys answer
+// "null", as a versioning-suspended bucket does.
+type nullVersions struct {
+ lease.StagingWriter
+ match func(key string) bool
+}
+
+func (n nullVersions) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) {
+ etag, ver, err := n.StagingWriter.CopyVersionedAt(ctx, signAt, srcBucket, srcKey, srcETag, dstBucket, dstKey)
+ if err == nil && n.match(dstKey) {
+ ver = "null"
+ }
+ return etag, ver, err
+}
+
+// G27: the seal needs a usable version for every current replica key. A
+// final round that copied one without (a "null") seals nothing and runs
+// again; that is not a deferred round.
+func TestSealRefusesAnObjectWithoutAUsableVersion(t *testing.T) {
+ h := newCutHarness(t)
+ h.ready(t)
+ suspended := true
+ h.cfg.Staging = func(context.Context, string) (lease.StagingWriter, error) {
+ return nullVersions{StagingWriter: h.f, match: func(key string) bool { return suspended && strings.HasSuffix(key, lastTXKey) }}, nil
+ }
+ h.freshWorker() // the worker memoizes its staging writer
+ h.cut(t)
+ h.pass(t)
+ if m := h.move(t); m.CutPhase != provision.CutStopped {
+ t.Fatalf("sealed over a null version: %+v", m)
+ }
+ if _, ok := h.claim(t).Op.Journal[hostclaim.JournalSeal]; ok {
+ t.Fatal("a seal was journalled")
+ }
+ if n := h.ledger(t).DeferredRounds; n != 0 {
+ t.Fatalf("counted as %d deferred rounds", n)
+ }
+ suspended = false
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveLanding {
+ t.Fatalf("the next round did not seal: %+v", m)
+ }
+ for _, o := range h.seal(t).Replica {
+ if !lease.UsableVersionID(o.VersionID) {
+ t.Fatalf("sealed %s@%q", o.Key, o.VersionID)
+ }
+ }
+}
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 2fb108a30..16b45018b 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -24,6 +24,7 @@ import (
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
"github.com/carlosframework/platform/internal/release"
+ "github.com/carlosframework/platform/internal/stopevidence"
)
// The residence-move worker (issue #331, instance-move spec): everything a
@@ -227,6 +228,7 @@ func armMovesPass(hib activator.Config, dataDir string, reg *registry.Registry,
if cfg.Enabled {
armPhase1(&cfg, w, hib, dataDir, reg, svc, eb, os.Getenv)
w.holds = newMovesHoldView(cfg.Box, cfg.Region, hostclaim.Read, holdStageRequester(dataDir))
+ cfg.TimersDrained = w.holds.drainTimers
}
w.cfg = cfg
armedMovesWorker.Store(w)
@@ -265,6 +267,32 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
cfg.FilesDir = func(host string) (string, string) {
return filepath.Join(dataDir, host), filepath.Join(dataDir, host+".env")
}
+ // The cut (movescut.go).
+ cfg.FinalCheckpoint = finalCheckpoint(hib.Bucket, hib.Region)
+ cfg.LocalPosition = localLTXPosition
+ cfg.InputVersions = func(ctx context.Context, s release.Store, rec provision.Record) (map[string]string, string, error) {
+ _, v, err := appconfig.LoadMergedHostVersions(ctx, s, rec.Account, rec.App, rec.Environment, rec.Host)
+ if err != nil {
+ return nil, "", err
+ }
+ return map[string]string{"default": v.Default, "environment": v.Environment}, v.Overlay, nil
+ }
+ if reg != nil {
+ cfg.HoldMoving = holdMoving(reg, holdStageRequester(dataDir))
+ cfg.Timers = reg.TimersFor
+ }
+ 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
+ // cuts (Task 3).
+ cfg.Evidence = func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error) {
+ return stopevidence.Reconstruct(ctx, stopevidence.ReconstructInputs{Host: host, Cgroups: eb.Cgroups, DBPaths: dbPaths,
+ LeaseFree: func(ctx context.Context) (bool, error) {
+ held, _, _, err := lease.LockState(ctx, hib.Bucket, location, hib.Region)
+ return !held, err
+ }})
+ }
+ }
cfg.DiskFree = diskBytes
cfg.ProbeStore = probeStores
cfg.LaneProbe = laneProbe(cfg.Versioned)
@@ -416,6 +444,29 @@ type moveCfg struct {
LaneProbe func(ctx context.Context, bucket, region, moveID string) error
MeasureS3 func(ctx context.Context, key string) (moves.MeasuredRate, error)
MeasureDisk func(ctx context.Context) (moves.MeasuredRate, moves.FilesRate, error)
+
+ // The cut's hooks (§5.3, §7.4, movescut.go). HoldMoving holds the row
+ // "moving" under the move id (holdMoving). TimersDrained waits until
+ // no timer delivery or convergence that could have missed the hold is
+ // still running (the hold view's drainTimers, Task 6). Evidence is
+ // §5.3 steps 1 to 4 (stopevidence.Reconstruct, its lease step read at
+ // the claim holder's location). FinalCheckpoint is step 5's checkpoint
+ // under the claim's lineage, check being the authority read before
+ // each replica request (G3). LocalPosition is the stopped database's
+ // TXID (the brief's OwnReplicaPosition, G22). Timers reads the host's
+ // timer rows. InputVersions reads the current bundle etags (default,
+ // environment) and overlay etag, as T3 pinned them.
+ HoldMoving func(host, op string) error
+ TimersDrained func(ctx context.Context, host string) error
+ Evidence func(ctx context.Context, host, location string, dbPaths []string) (moves.StoppedReport, error)
+ FinalCheckpoint func(ctx context.Context, dbPath string, holder hostclaim.Holder, check func(context.Context) error) error
+ LocalPosition func(dbPath string) (uint64, error)
+ Timers func(host string) ([]registry.Timer, error)
+ InputVersions func(ctx context.Context, s release.Store, rec provision.Record) (bundle map[string]string, overlay string, err error)
+ // cutFault is a test seam: called after each side effect of the cut
+ // with the point's name, an error being a crash there. Nil in
+ // production.
+ cutFault func(point string) error
}
// defaultMoveReadyDelta is the readiness bar when
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 00079e33c..05dd2991d 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -186,8 +186,14 @@ func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec p
return w.preparingRound(ctx, s, rec, fleet)
case provision.MoveAborting:
return w.interimAbort(ctx, s, rec)
- }
- return nil // cutting and later: Plan 2
+ case provision.MoveCutting:
+ return w.runCut(ctx, s, rec, fleet)
+ case provision.MoveReverting:
+ // 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
}
// srcRound is what one preparing round learned for the source report.