amadan

carlos / platform Public

host: reverting, the pre-flip recovery, journaled in rev_phase (§7.4)

revertSource is the MoveReverting arm (J7), routed by the participant
rule: from stopping it computes the stop evidence and starts nothing
until it holds, gives up execution through the cut's giveUpExecution
(no baselines on a reverting record, J8) and calls Plan 2's revertClaim
unchanged. Then rev_phase: evidence (the row restoring under the move
id), verify (files and timers against the stopped baselines, or straight
to probing without them), probing (the recovery probe hook, inputs
recorded before the render, never verify again, fifteen minutes then
stuck), probed (reopened_at, the conditional open), opened (waits for
the lane park, ReleaseMove, MoveRevertClose). The target region's box
runs the lane park through the park engine; the source's journal goes
to the local sweep. Unarmed in production until Task 9 and Plan 5.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c7811d345fa7e1ff32d352931588014321af5d46 parent c744086
7 files changed, +1844 −14
  • approved_moves_copy_test.go +1 −0
  • internal/host/movescut.go +25 −11
  • internal/host/movesfinal.go +11 −0
  • internal/host/movespass.go +34 −0
  • internal/host/movesphase1.go +5 −3
  • internal/host/movesrevert.go +840 −0
  • internal/host/movesrevert_test.go +928 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 840338245..4803a1430 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -591,6 +591,7 @@ var movesScanFiles = []string{
"internal/host/movesrecover.go",
"internal/host/movesreport.go",
"internal/host/movesretire.go",
+ "internal/host/movesrevert.go",
"internal/host/movesrollback.go",
"internal/host/movesstage.go",
"internal/host/movesstore.go",
diff --git a/internal/host/movescut.go b/internal/host/movescut.go
index 9050524e2..fbf4d0cb8 100644
--- a/internal/host/movescut.go
+++ b/internal/host/movescut.go
@@ -510,10 +510,17 @@ func (c *cutRun) stopped(ctx context.Context) (bool, error) {
// 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).
+//
+// A revert's entry runs it too (revertSource, Plan 4 Task 6, ruling J7), on a
+// record already reverting: it then records no baselines (MoveRecordSource is
+// a cut's, at stopped; a revert from stopping has none, J8), and a files
+// directory that cannot be archived is no reason to revert again: the report
+// carries no files digest, and nothing compares one (§7.4 evidence).
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
+ reverting := c.rec.Move.State == provision.MoveReverting
j, err := c.journal()
if err != nil {
return moves.StoppedReport{}, err
@@ -591,11 +598,16 @@ func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, st
dir, envPath := cfg.FilesDir(host)
plain, _, _, err := archivePlainDigest(dir, envPath)
if why, retry, ferr := archiveFailure(host, err); ferr != nil || why != "" || retry {
- if why != "" {
+ switch {
+ case why != "" && reverting:
+ log.Printf("moves: %s: the revert's stopped report carries no files digest: %s", host, why)
+ plain = ""
+ case why != "":
_, rerr := c.revert(ctx, provision.RevertArchiveRefused)
return moves.StoppedReport{}, rerr
+ default:
+ return moves.StoppedReport{}, ferr
}
- return moves.StoppedReport{}, ferr
}
var rt registry.Route
if cfg.Route != nil {
@@ -625,16 +637,18 @@ func (c *cutRun) giveUpExecution(ctx context.Context, obs hostclaim.Observed, st
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")
+ if !reverting {
+ 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
}
- 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 {
diff --git a/internal/host/movesfinal.go b/internal/host/movesfinal.go
index cdc48f53e..a339f7694 100644
--- a/internal/host/movesfinal.go
+++ b/internal/host/movesfinal.go
@@ -117,6 +117,12 @@ func movesParticipantRecord(cfg moveCfg, rec provision.Record) string {
if cfg.OwnsRoute == nil || cfg.OwnsRoute(rec.Host) || moveJournalPresent(cfg.DataDir, rec.Host, m.ID) {
return role
}
+ // A revert never flipped: its target has neither the row nor a journal,
+ // and its one piece of work is the lane park the claim owes its region
+ // (movesrevert.go, Plan 4 Task 6).
+ if role == roleTarget && m.State == provision.MoveReverting {
+ return role
+ }
return roleNone
}
@@ -141,6 +147,8 @@ func (w *movesWorker) participantRound(ctx context.Context, s release.Store, rec
return w.finalizeSource(ctx, s, rec)
case role == roleTarget && rec.Move.State == provision.MoveRollingBack:
return w.rollbackTarget(ctx, s, rec)
+ case role == roleTarget && rec.Move.State == provision.MoveReverting:
+ return w.revertLanePark(ctx, s, rec)
}
return nil
}
@@ -703,6 +711,9 @@ func (w *movesWorker) sweepFinalJournal(dataDir, host, op string, recs []provisi
if _, ok := j.Steps[rcStepSource]; ok {
return w.sweepRecoveryJournal(dataDir, host, op, j, recs)
}
+ if _, ok := j.Steps[rvStepSource]; ok {
+ return w.sweepRevertJournal(dataDir, host, op, j, recs) // movesrevert.go, Task 6
+ }
return w.sweepRollbackJournal(dataDir, host, op, j, recs)
}
if stamp != "done" {
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index 993cfe4e5..d1215d2ab 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -325,6 +325,10 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
// The recovery's open ends the rollback's hold on the source's row,
// a plain registry call under the host's hold lock (J12).
cfg.ReleaseHeld = releaseHeld(reg, holdStageRequester(dataDir))
+ // The revert's hold goes from moving to restoring under the move
+ // id, a conditional status change under the host's hold lock
+ // (movesrevert.go, Task 6).
+ cfg.RestatusHold = restatusHold(reg, holdStageRequester(dataDir))
}
if eb != nil {
// A box with no delegated cgroup has no manager, and Reconstruct
@@ -591,6 +595,19 @@ type moveCfg struct {
ReleaseHeld func(host, op string) (bool, error)
recoveryProbe func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error)
rcFault func(point string) error
+
+ // The revert on the source (§7.4 reverting, movesrevert.go).
+ // RestatusHold changes the status of a hold the op already has, moving
+ // to restoring under the move id: registry.SetHoldIf expecting the op
+ // itself, under the host's hold lock, after the caller's validation,
+ // touching the stage-requested marker when it landed (restatusHold). The
+ // revert also uses the cut's hooks, ReleaseHeld, Timers, FilesDir,
+ // LogEvent and recoveryProbe above; recoveryProbe nil leaves it unarmed in
+ // production (revertSourceArmed), and the target region's lane park needs
+ // ParkLeases (revertLaneArmed). rvFault is its test seam, nil in
+ // production.
+ RestatusHold func(host, status, op string, validate func() error) (bool, error)
+ rvFault func(point string) error
}
// defaultMoveReadyDelta is the readiness bar when
@@ -673,6 +690,23 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
w.trouble.Delete(rec.Host)
continue
}
+ if rec.Move.State == provision.MoveReverting && movesParticipantRecord(cfg, rec) == roleSource {
+ // The revert on the source (movesrevert.go, Plan 4 Task 6):
+ // the record never flipped, and this box is the move's
+ // source by the same participant rule (J10). Its entry
+ // gives up execution through the cut, which reads the
+ // fleet; the rest does not need it.
+ var f *moveFleet
+ if loadFleet() {
+ f = fleet
+ }
+ if err := w.revertSource(ctx, s, rec, f); err != nil {
+ w.logTrouble(rec.Host, err)
+ continue
+ }
+ w.trouble.Delete(rec.Host)
+ continue
+ }
if rec.Region != cfg.Region || (cfg.OwnsRoute != nil && !cfg.OwnsRoute(rec.Host)) {
// After the flip the record's region is the target's, and
// after the source's finalization deletes its row OwnsRoute
diff --git a/internal/host/movesphase1.go b/internal/host/movesphase1.go
index 038aba1bb..d997b5f3f 100644
--- a/internal/host/movesphase1.go
+++ b/internal/host/movesphase1.go
@@ -189,9 +189,11 @@ func phase1MoveRound(ctx context.Context, s release.Store, w *movesWorker, rec p
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)
+ // The record decided the revert; the claim follows it (§7.5), and
+ // the source recovers (movesrevert.go, Plan 4 Task 6, ruling J7).
+ // movesPass routes a reverting record by the participant rule
+ // before it gets here; this arm is the same step.
+ return w.revertSource(ctx, s, rec, &fleet)
case provision.MoveLanding:
// The target's (ruling G34): after the flip the record's region is
// this box's and reconcile has minted the held row, so OwnsRoute
diff --git a/internal/host/movesrevert.go b/internal/host/movesrevert.go
new file mode 100644
index 000000000..d4b70c874
--- /dev/null
+++ b/internal/host/movesrevert.go
@@ -0,0 +1,840 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "fmt"
+ "log"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6),
+// behind CARLOS_MOVES_PHASE1: the pre-flip recovery of a cut abandoned before
+// the flip. The record decided it (MoveRevert); the source takes the host
+// back with its own evidence and runs a recovery journalled in the claim's
+// rev_phase. revertSource is phase1MoveRound's MoveReverting arm (ruling J7),
+// and movesPass reaches a reverting record through the participant rule
+// (movesParticipantRecord, J10) as Tasks 4 and 5 do.
+//
+// Entry. A revert from stopping (or from stopped before execution was given
+// up) finds the claim still moving with this box executing: the stop
+// evidence was never obtained, so something may still run. The source holds
+// the row moving and fences it again (an agent restart dropped the fence),
+// drains the timers and asks for the evidence (§5.3), and starts nothing
+// until it holds. Then the cut's own giveUpExecution publishes the complete
+// stopped-source.json and gives execution up citing it (it records no
+// baselines on a reverting record, J8), and Plan 2's revertClaim, unchanged,
+// writes the claim half with its journalled revert_bound: reverting under the
+// source, rev_phase evidence, the lane's park queued in the target region.
+// No second staging scope is ever computed (J7). A revert decided after
+// execution was given up starts at revertClaim.
+//
+// The recovery, from durable state, each step under the claim's rev_phase:
+//
+// - evidence: the evidence holds (Revert cites it). The row's hold goes
+// from moving to restoring under the move id, a conditional status change
+// under the host's hold lock with the record read fresh (RestatusHold;
+// a revert never hands to a rollback id). Then rev_phase steps to verify
+// when the record keeps the stopped baselines (Source), else straight to
+// probing (J8), op.phase following from probing on (J1).
+// - verify: the files plain digest and the timers digest equal what the
+// source recorded at stopped (Source.Files.Plain, Source.TimersSHA256).
+// A mismatch stops for an operator with the baseline-mismatch sentence
+// (MoveMarkStuck); a retry returns to verify and checks again (INV-4).
+// Passed, rev_phase and op.phase step to probing before the first attempt.
+// - probing: the recovery probe only (recoveryProbe, Task 9's; warm or cold
+// into the source's own generation, holder.generation, which a revert
+// never changed). It records the inputs it renders before it renders them
+// (MoveRecordRevertInputs). Never verify again: the probe's process may
+// have written, and its writes are the authoritative state (INV-2). The
+// first attempt of a round of probing is journalled; fifteen minutes of
+// failed probes stop for an operator (StuckRevertProbeBudget), the
+// journalled start cleared first so a retry probes for fifteen minutes
+// again. Answered, the claim steps to probed.
+// - probed: reopened_at on the record (MoveRevertReopened, J26), then the
+// conditional open on hold_op = move id under the host's hold lock with
+// the record read fresh (ReleaseHeld, a plain registry call, J12), then
+// the claim steps to opened.
+// - opened: the claim waits for the lane's park the revert queued in the
+// target region (§7.11), which the target region's box runs
+// (revertLanePark) and the park engine acknowledges after its horizon.
+// Then ReleaseMove to serving under this box, and MoveRevertClose:
+// LastMove rolled_back, offline from cut_at to the reopen. The source's
+// journal of the op is parked in the same pass; a crash before that
+// leaves it to the local journal sweep (sweepRevertJournal).
+//
+// A claim this move already released (RequireReleasedMove: a crash before the
+// record's close) only closes.
+//
+// The target region's half (revertLanePark). A revert never flipped, so its
+// target box has neither the row nor a journal of the move; the participant
+// rule answers it as the target on the state alone, and its one piece of work
+// is the lane's park-replica the claim owes its region: the park engine over
+// its own bucket archives moves/<host>/<op>/ under
+// parked/<inc>/<host>/<ret>/moves/ and acknowledges once the horizon has
+// passed (exactly as Plan 5 Task 1 dispatches an abort's).
+//
+// Every record write goes through a provision transition (the record_seq
+// rule); every claim write through a hostclaim transition.
+//
+// Arming (rulings J18, G17). Like the landing and the rollback, the revert is
+// unarmed in production until Plan 5: the recovery probe is nil there until
+// Task 9 (claim Plan 10's recovery admission), so revertSourceArmed answers
+// false and the source only writes the claim half, as Plan 2's arm did; and
+// the target's lane park needs the park engine's leases (ParkLeases), which
+// production does not wire (revertLaneArmed). LogEvent is nil in production,
+// where an event is a log line.
+
+// revertStepsPerPass bounds the steps one pass runs: evidence to the close is
+// seven.
+const revertStepsPerPass = 10
+
+// revertProbeBudget is how long a revert's probes may fail before it stops
+// for an operator (§7.4: "Fifteen minutes of failed probes: stuck").
+const revertProbeBudget = 15 * time.Minute
+
+// The revert's steps in the source's op journal (the cut's, under the move
+// id).
+const (
+ // rvStepSource marks the journal as a revert's once the claim's recovery
+ // begins (revertJournalMark), so the sweep can tell it from a cut's that
+ // is still running, and knows the incarnation to park it under.
+ rvStepSource = "rv-source"
+ // rvStepProbeSince is the first probe attempt of this round of probing
+ // (RFC 3339), the start of the fifteen minutes. Cleared before a stuck
+ // mark, so a retry starts again.
+ rvStepProbeSince = "rv-probe-since"
+ // rvStepClosed is "done" once the record closed.
+ rvStepClosed = "rv-source-closed"
+ // rvStepJournalPark is parkLocalArtifacts' step: the retirement the
+ // journal is parked into (no artifacts of its own).
+ rvStepJournalPark = "rv-source-journal-park"
+)
+
+// revertJournalMark is rvStepSource's value.
+type revertJournalMark struct {
+ Move string `json:"move"`
+ Incarnation string `json:"incarnation"`
+}
+
+// revertSourceArmed reports whether this box can run the revert's recovery:
+// every hook it calls is wired, the recovery probe included. In production
+// the probe is not (Task 9, J18), so this answers false there.
+func revertSourceArmed(cfg moveCfg) bool {
+ return cfg.Box != "" && cfg.DataDir != "" && cfg.RestatusHold != nil && cfg.ReleaseHeld != nil &&
+ cfg.Timers != nil && cfg.FilesDir != nil && cfg.recoveryProbe != nil
+}
+
+// revertLaneArmed reports whether this box can run the target region's lane
+// park: the park engine's leases and this box's own bucket. In production
+// ParkLeases is not wired (ruling G17), so this answers false there.
+func revertLaneArmed(cfg moveCfg) bool {
+ return cfg.Box != "" && cfg.Region != "" && cfg.OwnBucket != "" && cfg.ParkLeases != nil && cfg.Versioned != nil
+}
+
+// restatusHold is moveCfg.RestatusHold in production: an op's own hold
+// changes status (a revert's moving to restoring under the move id) by
+// registry.SetHoldIf expecting the op itself, under the host's hold lock,
+// after the caller's validation; written (holdStageRequester) runs when the
+// change landed, so root re-judges what it staged. A row already held status
+// by op is a replay and answers true; a row held by another op, unheld or
+// absent answers false and is not touched.
+func restatusHold(reg *registry.Registry, written func(host string)) func(host, status, op string, validate func() error) (bool, error) {
+ return func(host, status, op string, validate func() error) (bool, error) {
+ var held bool
+ var err error
+ provision.WithHostHoldLock(host, func() {
+ var rt *registry.Route
+ if rt, err = reg.Get(host); err != nil || rt == nil {
+ return
+ }
+ if validate != nil {
+ if err = validate(); err != nil {
+ return
+ }
+ }
+ if op == "" || rt.HoldOp != op {
+ return
+ }
+ if rt.Status == status {
+ held = true
+ return
+ }
+ if held, err = reg.SetHoldIf(host, status, op, op); err != nil || !held {
+ return
+ }
+ if written != nil {
+ written(host)
+ }
+ })
+ return held, err
+ }
+}
+
+// revertRun is one pass's state for one reverting record on its source.
+type revertRun struct {
+ w *movesWorker
+ s release.Store
+ rec provision.Record
+ g provision.MoveGuard
+ fleet *moveFleet // nil when the pass could not read it: the entry waits
+}
+
+// revertSource is phase1MoveRound's MoveReverting arm (J7) and the source's
+// participant step for a reverting record (movesPass).
+func (w *movesWorker) revertSource(ctx context.Context, s release.Store, rec provision.Record, fleet *moveFleet) error {
+ cfg := w.cfg
+ m := rec.Move
+ if m == nil || m.Legacy() || m.State != provision.MoveReverting || m.From != cfg.Region {
+ return nil
+ }
+ if !revertSourceArmed(cfg) {
+ // The record decided; the claim follows it (§7.5), as Plan 2's arm
+ // did. The recovery waits for Task 9's probe.
+ if err := w.revertClaim(ctx, s, rec); err != nil {
+ return err
+ }
+ w.notice(rec.Host+"#revert", "moves: %s: the revert of %s waits: this box's revert is not armed (the recovery probe is Task 9's, on claim Plan 10's recovery admission)", rec.Host, m.ID)
+ return nil
+ }
+ r := &revertRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, fleet: fleet}
+ for i := 0; i < revertStepsPerPass; i++ {
+ advanced, err := r.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 !r.same(fresh) {
+ return nil
+ }
+ r.rec = *fresh
+ }
+ return nil
+}
+
+func (r *revertRun) cfg() moveCfg { return r.w.cfg }
+func (r *revertRun) now() time.Time { return r.w.now() }
+
+// same reports whether fresh is still this move, reverting.
+func (r *revertRun) same(fresh *provision.Record) bool {
+ m := fresh.Move
+ return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveReverting
+}
+
+// fault is the test seam after a side effect.
+func (r *revertRun) fault(point string) error {
+ if f := r.cfg().rvFault; f != nil {
+ return f(point)
+ }
+ return nil
+}
+
+// stillReverting is the record check run under the host's hold lock before
+// a row write: the record, read fresh, must still be this move reverting.
+func (r *revertRun) stillReverting(ctx context.Context) error {
+ fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host)
+ if err != nil {
+ return fmt.Errorf("re-read the record: %w", err)
+ }
+ if !r.same(fresh) {
+ return fmt.Errorf("move %s: the record is no longer reverting", r.g.ID)
+ }
+ return nil
+}
+
+// update writes one claim transition over obs.
+func (r *revertRun) update(ctx context.Context, obs hostclaim.Observed, what string, step func(*hostclaim.Claim) error) error {
+ if _, err := hostclaim.Update(ctx, r.s, obs, step); err != nil {
+ return fmt.Errorf("%s: %w", what, err)
+ }
+ return nil
+}
+
+// record writes one record transition. A conflict is someone else's
+// decision: logged, and the pass ends without error.
+func (r *revertRun) record(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
+ if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host, mut); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: %s of %s refused (%v); the record moved on", r.rec.Host, what, r.g.ID, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("%s: %w", what, err)
+ }
+ return true, nil
+}
+
+// step runs the revert's current step once, from the claim. advanced is true
+// when it wrote the claim or the record.
+func (r *revertRun) step(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ obs, err := hostclaim.Read(ctx, r.s, host)
+ if err != nil {
+ return false, fmt.Errorf("read the host claim: %w", err)
+ }
+ if obs.Absent {
+ return false, fmt.Errorf("move %s has no host claim", id)
+ }
+ c := obs.Claim
+ if hostclaim.RequireReleasedMove(id)(c) == nil {
+ // A crash between the claim's release and the record's close.
+ return r.close(ctx)
+ }
+ if c.Op.ID != id {
+ return false, fmt.Errorf("move %s: the claim names op %q", id, c.Op.ID)
+ }
+ if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation {
+ return false, fmt.Errorf("move %s: the host claim names another incarnation", id)
+ }
+ if len(c.Op.Participants) == 0 || c.Op.Participants[0] != cfg.Box {
+ return false, fmt.Errorf("move %s: this box is not the move's source in the claim", id)
+ }
+ switch c.Op.Kind {
+ case hostclaim.KindMoving:
+ if c.Op.Exec == cfg.Box {
+ return r.evidence(ctx, obs)
+ }
+ // Execution given up with the source's evidence: the claim half.
+ if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil {
+ return false, err
+ }
+ fresh, err := hostclaim.Read(ctx, r.s, host)
+ if err != nil {
+ return false, fmt.Errorf("read the host claim: %w", err)
+ }
+ if fresh.Absent || fresh.Claim.Op.Kind != hostclaim.KindReverting {
+ r.w.notice(host+"#revert", "moves: %s: the revert of %s waits: the claim is moving with executor %q", host, id, c.Op.Exec)
+ return false, nil
+ }
+ return true, nil
+ case hostclaim.KindReverting:
+ default:
+ return false, fmt.Errorf("move %s: the claim is %s, not reverting", id, c.Op.Kind)
+ }
+ if c.Op.Exec != cfg.Box {
+ return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, c.Op.Exec)
+ }
+ switch phase := c.Op.Journal[hostclaim.JournalRevPhase]; phase {
+ case hostclaim.RevPhaseEvidence:
+ return r.begin(ctx, obs)
+ case hostclaim.RevPhaseVerify:
+ return r.verify(ctx, obs)
+ case hostclaim.PhaseProbing:
+ return r.probing(ctx, obs)
+ case hostclaim.PhaseProbed:
+ return r.probed(ctx, obs)
+ case hostclaim.PhaseOpened:
+ return r.opened(ctx, obs)
+ default:
+ return false, fmt.Errorf("move %s: the claim's rev_phase is %q", id, phase)
+ }
+}
+
+// evidence is the entry from stopping (or from stopped before execution was
+// given up): the hold, the fence and the drain again, the stop evidence until
+// it holds, then the cut's giveUpExecution and Plan 2's revertClaim.
+func (r *revertRun) evidence(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
+ return false, fmt.Errorf("move %s: %w", id, err)
+ }
+ if r.fleet == nil {
+ return false, fmt.Errorf("move %s: the fleet's data buckets could not be read; the revert's evidence waits", id)
+ }
+ c := &cutRun{w: r.w, s: r.s, fleet: *r.fleet, rec: r.rec, g: r.g}
+ defer c.releaseLease()
+ holder := *obs.Claim.Holder
+ if err := c.holderSupported(holder); err != nil {
+ return false, fmt.Errorf("move %s: %w", id, err)
+ }
+ 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 revert: a hook is not wired")
+ }
+ if err := cfg.HoldMoving(host, id); err != nil {
+ return false, fmt.Errorf("hold the row: %w", err)
+ }
+ gen := moveFenceGeneration(r.rec.Move)
+ if err := cfg.Fence(ctx, host, gen); err != nil {
+ return false, fmt.Errorf("fence against wakes: %w", err)
+ }
+ r.w.fenced.Store(host, gen)
+ if err := cfg.TimersDrained(ctx, host); err != nil {
+ return false, fmt.Errorf("drain the timers: %w", err)
+ }
+ j, err := c.journal()
+ if err != nil {
+ return false, err
+ }
+ var ev moves.StoppedReport
+ if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() {
+ rep, err := cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()})
+ if err != nil {
+ return false, fmt.Errorf("stop evidence: %w", err)
+ }
+ if !rep.TreeStopped() {
+ r.w.notice(host+"#rv-evidence", "moves: %s: the revert of %s starts nothing until the instance is proven stopped (%s)", host, id, failedSteps(rep))
+ return false, nil
+ }
+ body, err := json.Marshal(rep)
+ if err != nil {
+ return false, err
+ }
+ j.Mark(cutStepEvidence, string(body))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ if err := r.fault("after-revert-evidence"); err != nil {
+ return false, err
+ }
+ }
+ stoppedKey := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), "source")
+ report, err := c.giveUpExecution(ctx, obs, stoppedKey)
+ if err != nil {
+ return false, err
+ }
+ if report.V == 0 {
+ return false, nil // a file changed under the digest: the next pass
+ }
+ if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil {
+ return false, err
+ }
+ if err := r.fault("after-revert-claim"); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+// begin is rev_phase evidence on a reverting claim: the journal marked as
+// this revert's, the row's hold restoring under the move id, then verify or
+// probing.
+func (r *revertRun) begin(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ if obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource] == "" {
+ return false, fmt.Errorf("move %s: the revert cites no stop evidence", id)
+ }
+ if err := r.markJournal(); err != nil {
+ return false, err
+ }
+ held, err := cfg.RestatusHold(host, "restoring", id, func() error { return r.stillReverting(ctx) })
+ if err != nil {
+ return false, fmt.Errorf("hold the row restoring: %w", err)
+ }
+ if !held {
+ return false, fmt.Errorf("move %s: this box's row is not held by the move", id)
+ }
+ if err := r.fault("after-restatus"); err != nil {
+ return false, err
+ }
+ baseline := r.rec.Move.Source != nil
+ to := hostclaim.PhaseProbing
+ if baseline {
+ to = hostclaim.RevPhaseVerify
+ }
+ if err := r.update(ctx, obs, "the revert's "+to, hostclaim.SetRevPhase(id, hostclaim.RevPhaseEvidence, to, baseline)); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: the revert of %s is at %s", host, id, to)
+ if err := r.fault("after-rev-" + to); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+// verify (§7.4 verify, INV-4): the files and the timers against the digests
+// recorded at stopped. A mismatch stops for an operator; a retry runs this
+// again.
+func (r *revertRun) verify(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id, m := r.rec.Host, r.g.ID, r.rec.Move
+ if err := r.fault("verify"); err != nil {
+ return false, err
+ }
+ if obs.Claim.Op.Phase != hostclaim.PhaseMoving && obs.Claim.Op.Phase != hostclaim.PhaseSealed {
+ return false, fmt.Errorf("move %s: op.phase %q at rev_phase verify", id, obs.Claim.Op.Phase)
+ }
+ if m.Source == nil || m.Source.Files == nil || m.Source.TimersSHA256 == "" {
+ return false, fmt.Errorf("move %s: the record has no stopped baselines to verify against", id)
+ }
+ dir, envPath := cfg.FilesDir(host)
+ plain, _, _, err := archivePlainDigest(dir, envPath)
+ why, retry, ferr := archiveFailure(host, err)
+ switch {
+ case ferr != nil:
+ return false, fmt.Errorf("digest the source's files: %w", ferr)
+ case retry:
+ r.w.notice(host+"#rv-verify", "moves: %s: a file changed while the source's files were digested; the verification runs again", host)
+ return false, nil
+ case why != "":
+ return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the source's files cannot be digested: %s", why))
+ case plain != m.Source.Files.Plain:
+ return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the files are %s, recorded %s", plain, m.Source.Files.Plain))
+ }
+ rows, err := cfg.Timers(host)
+ if err != nil {
+ return false, fmt.Errorf("read the timers: %w", err)
+ }
+ if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 {
+ return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256))
+ }
+ log.Printf("moves: %s: the source still matches what it recorded when it stopped", host)
+ if err := r.update(ctx, obs, "the revert's probing", hostclaim.SetRevPhase(id, hostclaim.RevPhaseVerify, hostclaim.PhaseProbing, true)); err != nil {
+ return false, err
+ }
+ if err := r.fault("after-rev-probing"); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+// probing (§7.4 probing): the probe only, against a fifteen-minute budget
+// from its first attempt. Nothing earlier runs again from here.
+func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ if obs.Claim.Op.Phase != hostclaim.PhaseProbing {
+ return false, fmt.Errorf("move %s: op.phase %q at rev_phase probing", id, obs.Claim.Op.Phase)
+ }
+ j, err := moves.ReadJournal(cfg.DataDir, host, id)
+ if err != nil {
+ return false, err
+ }
+ since, perr := time.Parse(time.RFC3339Nano, j.Steps[rvStepProbeSince])
+ if perr != nil {
+ since = r.now().UTC()
+ j.Mark(rvStepProbeSince, since.Format(time.RFC3339Nano))
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ }
+ // A revert has no rollback: it probes the source's own generation, warm
+ // or cold as for serving (§7.14 reverting row), with no recovery
+ // coverage and no attempt count (its budget is the fifteen minutes,
+ // counted here). The probe records what it renders before it renders it.
+ req := recoveryProbeRequest{Clean: true, SourceGeneration: obs.Claim.Holder.Generation,
+ Record: func(ctx context.Context, in provision.MoveInputs) error {
+ _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRevertInputs(r.g, in))
+ return err
+ }}
+ out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req)
+ if err != nil {
+ log.Printf("moves: %s: the revert's probe of %s: %v", host, id, err)
+ out = recoveryProbeWaiting
+ }
+ if out == recoveryProbeAnswered {
+ if err := r.update(ctx, obs, "the revert's probed", hostclaim.SetRevPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed, r.rec.Move.Source != nil)); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: the source answered its probe; the revert of %s is probed", host, id)
+ if err := r.fault("after-rev-probed"); err != nil {
+ return false, err
+ }
+ return true, nil
+ }
+ if elapsed := r.now().Sub(since); elapsed >= revertProbeBudget {
+ // The start goes first: a retry probes for fifteen minutes again.
+ delete(j.Steps, rvStepProbeSince)
+ if err := j.Write(cfg.DataDir); err != nil {
+ return false, err
+ }
+ return r.stuck(ctx, provision.StuckRevertProbeBudget, fmt.Errorf("no answer for %s", elapsed.Round(time.Second)))
+ }
+ r.w.notice(host+"#rv-probe", "moves: %s: the revert's probe of %s has not answered yet", host, id)
+ return false, nil
+}
+
+// probed (§7.4 probed): reopened_at on the record (J26), the conditional open
+// on hold_op = move id under the host's hold lock (J12), the claim opened.
+func (r *revertRun) probed(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ if r.rec.Move.ReopenedAt == nil {
+ ok, err := r.record(ctx, "the reopen", provision.MoveRevertReopened(r.g, r.now()))
+ if !ok {
+ return false, err
+ }
+ if err := r.fault("after-reopened"); err != nil {
+ return false, err
+ }
+ }
+ var err error
+ provision.WithHostHoldLock(host, func() {
+ if err = r.stillReverting(ctx); err != nil {
+ return
+ }
+ var released bool
+ if released, err = cfg.ReleaseHeld(host, id); err != nil || released {
+ return
+ }
+ var rt *registry.Route
+ if cfg.Route == nil {
+ err = errors.New("this box cannot read its registry row")
+ return
+ }
+ if rt, err = cfg.Route(host); err != nil {
+ return
+ }
+ switch {
+ case rt == nil:
+ err = fmt.Errorf("move %s: this box has no row to open", id)
+ case rt.HoldOp != "":
+ err = fmt.Errorf("move %s: the row is held %s by %q, not by the move", id, rt.Status, rt.HoldOp)
+ }
+ })
+ if err != nil {
+ return false, fmt.Errorf("open the row: %w", err)
+ }
+ if err := r.fault("after-row-opened"); err != nil {
+ return false, err
+ }
+ if err := r.update(ctx, obs, "the revert's opened", hostclaim.SetRevPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened, r.rec.Move.Source != nil)); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: the source serves again; the revert of %s waits for its lane's park", host, id)
+ r.w.logEvent(ctx, r.rec, "source-opened")
+ nudgeSerial(ctx, r.s)
+ if err := r.fault("after-rev-opened"); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+// opened (§7.4 opened, §7.11): once no chore of the op is left in the claim
+// (the lane's park acknowledged), the claim's release to serving under this
+// box, then the record's close.
+func (r *revertRun) opened(ctx context.Context, obs hostclaim.Observed) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ if r.rec.Move.ReopenedAt == nil {
+ return false, fmt.Errorf("move %s: the claim is opened and the record has no reopen", id)
+ }
+ owed := 0
+ for _, o := range obs.Claim.Obligations {
+ if o.Op == id {
+ owed++
+ }
+ }
+ if owed > 0 {
+ r.w.notice(host+"#rv-opened", "moves: %s: the revert of %s waits for %d chore(s) of the move, its lane's park, to be acknowledged", host, id, owed)
+ return false, nil
+ }
+ r.w.logEvent(ctx, r.rec, "rolled_back")
+ if err := r.update(ctx, obs, "release the claim", hostclaim.ReleaseMove(id, cfg.Box)); err != nil {
+ return false, err
+ }
+ log.Printf("moves: %s: the claim is serving again after the revert of %s", host, id)
+ if err := r.fault("after-release-move"); err != nil {
+ return false, err
+ }
+ return r.close(ctx)
+}
+
+// close is the record's close (MoveRevertClose), then the source's journal of
+// the op parked.
+func (r *revertRun) close(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ ok, err := r.record(ctx, "the revert's close", provision.MoveRevertClose(r.g, r.now()))
+ if !ok {
+ return false, err
+ }
+ log.Printf("moves: %s: move %s is reverted and closed", host, id)
+ nudgeSerial(ctx, r.s)
+ if err := r.fault("after-close"); err != nil {
+ return true, err
+ }
+ if err := parkRevertJournal(cfg.DataDir, host, id, true); err != nil {
+ return true, fmt.Errorf("park the source's journal of %s: %w", id, err)
+ }
+ return true, nil
+}
+
+// stuck stops the revert for an operator (MoveMarkStuck) with a recorded
+// reason; the row stays held and move --retry resumes the rev_phase.
+func (r *revertRun) stuck(ctx context.Context, reason string, why error) (bool, error) {
+ host := r.rec.Host
+ log.Printf("moves: %s: the revert of %s stops for an operator (%v): %s", host, r.g.ID, why, reason)
+ ok, err := r.record(ctx, "the stuck mark", provision.MoveMarkStuck(r.g, reason))
+ if !ok {
+ return false, err
+ }
+ r.w.logEvent(ctx, r.rec, "source-stuck")
+ nudgeSerial(ctx, r.s)
+ return true, nil
+}
+
+// markJournal marks the source's journal of the op as this revert's. A mark
+// already there is kept.
+func (r *revertRun) markJournal() error {
+ dd, host := r.cfg().DataDir, r.rec.Host
+ j, err := moves.ReadJournal(dd, host, r.g.ID)
+ if err != nil {
+ return err
+ }
+ if j.Done(rvStepSource) {
+ return nil
+ }
+ body, err := json.Marshal(revertJournalMark{Move: r.g.ID, Incarnation: r.g.Incarnation})
+ if err != nil {
+ return err
+ }
+ j.Mark(rvStepSource, string(body))
+ return j.Write(dd)
+}
+
+// parkRevertJournal parks the source's journal of a closed revert: closed
+// (true when the caller has seen the close) is journalled first, then a
+// moves-owned retirement with no artifacts is made for it (its receipt names
+// the op) and the journal, with the cut's archive inside it, is renamed into
+// it. A journal that is not there was parked already; one with no revert mark
+// is not this revert's and is left.
+func parkRevertJournal(dataDir, host, op string, closed bool) error {
+ if !moveJournalPresent(dataDir, host, op) {
+ return nil
+ }
+ j, err := moves.ReadJournal(dataDir, host, op)
+ if err != nil {
+ return err
+ }
+ raw, ok := j.Steps[rvStepSource]
+ if !ok {
+ return nil
+ }
+ var mark revertJournalMark
+ if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) {
+ return fmt.Errorf("the journal of %s has an unreadable revert mark", op)
+ }
+ write := func(step, value string) error {
+ j.Mark(step, value)
+ return j.Write(dataDir)
+ }
+ if closed && j.Steps[rvStepClosed] != "done" {
+ if err := write(rvStepClosed, "done"); err != nil {
+ return err
+ }
+ }
+ retirement, err := parkLocalArtifacts(context.Background(), j, write, localArtifactPark{DataDir: dataDir, Incarnation: mark.Incarnation,
+ Host: host, Op: op, Step: rvStepJournalPark, NewID: obligationID})
+ if err != nil {
+ return err
+ }
+ if err := parkOpJournal(dataDir, mark.Incarnation, host, op, retirement); err != nil {
+ return err
+ }
+ log.Printf("moves: %s: the source's journal of the revert %s is parked", host, op)
+ return nil
+}
+
+// sweepRevertJournal is the local journal sweep's revert half
+// (sweepFinalJournal): a journal the revert marked (rvStepSource) is parked
+// once the close is shown by durable evidence:
+// - the journal records the close (rvStepClosed "done");
+// - no listed record carries the move: a reverting or stuck move is
+// cleared only by provision.MoveRevertClose (MoveRetry returns a stuck
+// one to reverting; no abort or completion takes either state), and the
+// mark is written while the record is reverting, so the close landed.
+//
+// A host the listing does not name, or a record that still carries the move,
+// is left.
+func (w *movesWorker) sweepRevertJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error {
+ var mark revertJournalMark
+ if err := json.Unmarshal([]byte(j.Steps[rvStepSource]), &mark); err != nil || mark.Move == "" {
+ return fmt.Errorf("the journal of %s has an unreadable revert mark", op)
+ }
+ if j.Steps[rvStepClosed] != "done" {
+ if len(recs) == 0 {
+ w.notice(host+"#rv-sweep-"+op, "moves: %s: the source's journal of %s is a revert's, and no listed record names the host; it is left", host, op)
+ return nil
+ }
+ for _, rec := range recs {
+ if m := rec.Move; m != nil && m.ID == mark.Move {
+ return nil
+ }
+ }
+ }
+ return parkRevertJournal(dataDir, host, op, true)
+}
+
+// revertLanePark is the target region's half of a revert (movesPass through
+// participantRound): the lane's park-replica the revert queued in this
+// region, run by the park engine over this box's own bucket until it is
+// acknowledged after its horizon. The view and the authorizer are Plan 5's
+// when wired, else the stand-ins: the source's region authoritative (a revert
+// never flipped), and only this move's reverting lane park.
+func (w *movesWorker) revertLanePark(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m := rec.Move
+ if m == nil || m.Legacy() || m.State != provision.MoveReverting || m.To != cfg.Region || m.From == cfg.Region {
+ return nil
+ }
+ if !revertLaneArmed(cfg) {
+ w.notice(rec.Host+"#rv-lane", "moves: %s: the park of move %s's lane waits: this box's park engine is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.ID)
+ return nil
+ }
+ host, id := rec.Host, m.ID
+ obs, err := hostclaim.Read(ctx, s, host)
+ if err != nil {
+ return fmt.Errorf("read the host claim: %w", err)
+ }
+ if obs.Absent || obs.Claim.Op.ID != id || obs.Claim.Op.Kind != hostclaim.KindReverting {
+ return nil // the claim half has not followed yet, or it is over
+ }
+ if h := obs.Claim.Holder; h == nil || h.Incarnation != m.Incarnation {
+ return fmt.Errorf("move %s: the host claim names another incarnation", id)
+ }
+ lane := moves.ScopeLocation(host, id)
+ var chores []hostclaim.Obligation
+ for _, o := range obs.Claim.Obligations {
+ if o.Op == id && o.Kind == "park-replica" && o.Region == cfg.Region && o.Park != nil && o.Park.Location == lane {
+ chores = append(chores, o)
+ }
+ }
+ if len(chores) == 0 {
+ return nil
+ }
+ vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
+ if err != nil {
+ return err
+ }
+ view, auth := cfg.ParkView, cfg.ParkAuthorize
+ if view == nil {
+ view = standInView(cfg.Region, roleTarget, m)
+ }
+ if auth == nil {
+ auth = func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
+ c := obs.Claim
+ if obs.Absent || c == nil || c.Op.ID != id || c.Op.Kind != hostclaim.KindReverting || c.Op.Exec == cfg.Box ||
+ o.Op != id || o.Kind != "park-replica" || o.Region != cfg.Region || o.Park == nil || o.Park.Location != lane {
+ return fmt.Errorf("%w: not this revert's lane park", park.ErrUnsafe)
+ }
+ return nil
+ }
+ }
+ e := park.Engine{Store: vs, Claims: s, Bucket: cfg.OwnBucket, Region: cfg.Region, Leases: cfg.ParkLeases,
+ Now: w.now, NewID: obligationID, View: view, Authorize: auth, After: cfg.parkFault}
+ for _, o := range chores {
+ err := e.Run(ctx, park.Work{Host: host, ID: o.ID, Op: id})
+ switch {
+ case errors.Is(err, park.ErrPending):
+ w.notice(host+"#rv-lane", "moves: %s: move %s's lane is archived; its park is acknowledged after its horizon", host, id)
+ case errors.Is(err, park.ErrChanged):
+ case err != nil:
+ return fmt.Errorf("park move %s's lane: %w", id, err)
+ default:
+ log.Printf("moves: %s: move %s's lane in %s is parked", host, id, cfg.Region)
+ }
+ }
+ return nil
+}
diff --git a/internal/host/movesrevert_test.go b/internal/host/movesrevert_test.go
new file mode 100644
index 000000000..088158c84
--- /dev/null
+++ b/internal/host/movesrevert_test.go
@@ -0,0 +1,928 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/activator"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6). The
+// fixture is the cut harness driven to a revert, either from stopping (the
+// stop evidence never held: the claim still moving under the source, no
+// baselines on the record) or from stopped after execution was given up
+// (the claim reverting, the baselines recorded). Its row follows the holds
+// the cut and the revert write; the hold's status change, the open and the
+// probe (Task 9, BLOCKED-ON-CLAIM C1) are faked; the target region's box
+// (uk-box) runs the lane park through the real park engine over the same
+// buckets with a fake lease table. Everything else is the real revert: the
+// cut's evidence and giveUpExecution, revertClaim, hostclaim's and
+// provision's own transitions.
+
+type revertHarness struct {
+ *cutHarness
+ tdir string
+ tleases *fakeParkLeases
+ tw *movesWorker
+ probe func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
+ probes []recoveryProbeRequest
+ restatus []string // status:op of each status change asked for
+ opens []string // the op of each open asked for
+ events []string
+ rvAt string
+ rvDone bool
+ rvHits []string
+}
+
+func newRevertHarness(t *testing.T) *revertHarness {
+ t.Helper()
+ h := &revertHarness{cutHarness: newCutHarness(t), tdir: t.TempDir(), tleases: &fakeParkLeases{}}
+ // The row follows the cut's hold, as holdMoving keeps it: held moving
+ // under the op, a row held by another op refused.
+ h.cfg.HoldMoving = func(host, op string) error {
+ h.holds = append(h.holds, host+"="+op)
+ if h.route.HoldOp != "" && h.route.HoldOp != op {
+ return fmt.Errorf("%s is held %s by %s", host, h.route.Status, h.route.HoldOp)
+ }
+ if h.route.HoldOp == "" {
+ h.route.Status, h.route.HoldOp = "moving", op
+ }
+ return nil
+ }
+ return h
+}
+
+// fromStopping drives the cut to a revert at stopping: the evidence never
+// held, the deadline passed, the record reverting with no baselines and the
+// claim still moving under the source.
+func (h *revertHarness) fromStopping(t *testing.T) {
+ t.Helper()
+ 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.CutPhase != provision.CutStopping || m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
+ t.Fatalf("the fixture did not revert at stopping: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
+ t.Fatalf("the fixture's claim: %+v", c.Op)
+ }
+ if h.route.Status != "moving" || h.route.HoldOp != p1ID {
+ t.Fatalf("the fixture's row: %+v", h.route)
+ }
+}
+
+// fromStopped drives the cut to a revert at stopped after execution was
+// given up: the baselines recorded and the claim reverting at evidence.
+func (h *revertHarness) fromStopped(t *testing.T) {
+ t.Helper()
+ 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.CutPhase != provision.CutStopped || m.Source == nil || m.Source.Files == nil {
+ t.Fatalf("the fixture did not revert at stopped: %+v", m)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence {
+ t.Fatalf("the fixture's claim: %+v", c.Op)
+ }
+}
+
+// arm wires the revert's hooks on the source box.
+func (h *revertHarness) arm(t *testing.T) {
+ t.Helper()
+ h.cfg.RestatusHold = func(host, status, op string, validate func() error) (bool, error) {
+ h.restatus = append(h.restatus, status+":"+op)
+ if validate != nil {
+ if err := validate(); err != nil {
+ return false, err
+ }
+ }
+ if host != p1Host || h.route.HoldOp != op {
+ return false, nil
+ }
+ h.route.Status = status
+ return true, nil
+ }
+ h.cfg.ReleaseHeld = func(host, op string) (bool, error) {
+ h.opens = append(h.opens, op)
+ if op == "" || h.route.HoldOp != op {
+ return false, nil
+ }
+ h.route.Status, h.route.HoldOp = "ready", ""
+ return true, nil
+ }
+ h.cfg.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.events = append(h.events, event) }
+ h.cfg.recoveryProbe = func(ctx context.Context, _ release.Store, _ provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
+ h.probes = append(h.probes, req)
+ if h.probe != nil {
+ return h.probe(req)
+ }
+ if err := req.Record(ctx, revertRendered()); err != nil {
+ return recoveryProbeWaiting, err
+ }
+ return recoveryProbeAnswered, nil
+ }
+ h.cfg.rvFault = func(point string) error {
+ h.rvHits = append(h.rvHits, point)
+ if point == h.rvAt && !h.rvDone {
+ h.rvDone = true
+ return errCrash
+ }
+ return nil
+ }
+}
+
+// revertRendered are the inputs the fake probe renders: not the pinned ones.
+func revertRendered() provision.MoveInputs {
+ return provision.MoveInputs{BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9", Release: "v7"}
+}
+
+// laneCfg is the target region's box, its lane park armed.
+func (h *revertHarness) laneCfg() moveCfg {
+ c := h.target()
+ c.DataDir = h.tdir
+ c.ParkLeases = h.tleases
+ return c
+}
+
+// tPass runs one pass as the target region's box.
+func (h *revertHarness) tPass(t *testing.T) {
+ t.Helper()
+ if h.tw == nil {
+ h.tw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
+ }
+ h.tw.cfg = h.laneCfg()
+ if err := movesPass(context.Background(), h.s, h.tw); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (h *revertHarness) verifications() int {
+ n := 0
+ for _, p := range h.rvHits {
+ if p == "verify" {
+ n++
+ }
+ }
+ return n
+}
+
+func (h *revertHarness) revPhase(t *testing.T) string {
+ t.Helper()
+ return h.claim(t).Op.Journal[hostclaim.JournalRevPhase]
+}
+
+// laneChore is the lane's park the revert queued in the target region.
+func (h *revertHarness) laneChore(t *testing.T) *hostclaim.Obligation {
+ t.Helper()
+ for _, o := range h.claim(t).Obligations {
+ if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, p1ID) {
+ return &o
+ }
+ }
+ return nil
+}
+
+// laneKeys are the objects still in the move's lane in the target bucket.
+func (h *revertHarness) laneKeys() []string {
+ var out []string
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, moves.ScopeLocation(p1Host, p1ID)+"/") {
+ out = append(out, k)
+ }
+ }
+ return out
+}
+
+// pastHorizon moves the clock past the lane park's acknowledgement horizon.
+func (h *revertHarness) pastHorizon(t *testing.T, o *hostclaim.Obligation) {
+ t.Helper()
+ until := o.Park.EvidenceAt.Add(park.SignatureAge + o.Park.RequestLimit)
+ if d := until.Sub(h.clock.now); d > 0 {
+ h.clock.Advance(d + time.Second)
+ }
+}
+
+// reopened asserts the source reopened: the claim reverting/opened under
+// the source, the record still reverting with ReopenedAt, the row open.
+func (h *revertHarness) reopened(t *testing.T) {
+ t.Helper()
+ m := h.move(t)
+ if m == nil || m.State != provision.MoveReverting || m.ReopenedAt == nil {
+ t.Fatalf("the source did not reopen: %+v", m)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseOpened {
+ t.Fatalf("the claim after the reopen: %+v", c.Op)
+ }
+ if h.route.HoldOp != "" || h.route.Status != "ready" {
+ t.Fatalf("the row is not open: %+v", h.route)
+ }
+}
+
+// drive runs the source's and the target's passes, moving the clock past the
+// lane park's horizon once the source has opened, until the record closes.
+func (h *revertHarness) drive(t *testing.T) {
+ t.Helper()
+ for i := 0; i < 12; i++ {
+ h.pass(t)
+ h.tPass(t)
+ if h.read(t, p1Host).Move == nil {
+ return
+ }
+ if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
+ if o := h.laneChore(t); o != nil {
+ h.pastHorizon(t, o)
+ }
+ }
+ }
+ t.Fatalf("the revert never closed: %+v, claim %+v, hits %v", h.move(t), h.claim(t).Op, h.rvHits)
+}
+
+// closed asserts the end: the record at rest in ie with LastMove rolled_back,
+// the claim serving under the source with this move recorded as released,
+// the row open, the source's journal parked.
+func (h *revertHarness) closed(t *testing.T) {
+ t.Helper()
+ rec := h.read(t, p1Host)
+ if rec.Move != nil || rec.Region != "ie" || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" {
+ t.Fatalf("the record after the close: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" || hostclaim.RequireReleasedMove(p1ID)(c) != nil {
+ t.Fatalf("the claim after the close: %+v", c.Op)
+ }
+ if h.route.HoldOp != "" || h.route.Status != "ready" {
+ t.Fatalf("the row after the close: %+v", h.route)
+ }
+ if moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
+ t.Fatal("the source's journal of the move is still live")
+ }
+ parked, _ := filepath.Glob(filepath.Join(h.cfg.DataDir, ".parked", h.inc, p1Host, "*", "artifacts", ".moves", p1Host, p1ID))
+ if len(parked) != 1 {
+ t.Fatalf("the source's journal is not parked once: %v", parked)
+ }
+ for _, s := range h.restatus {
+ if s != "restoring:"+p1ID {
+ t.Fatalf("a status change other than restoring under the move id: %v", h.restatus)
+ }
+ }
+}
+
+// TestRevertFromStoppingComputesEvidenceFirst (§7.4 evidence, J7, J8): a
+// revert that began at stopping first computes the stop evidence, and starts
+// nothing until it holds: the row stays held moving under the move id and is
+// fenced again, the claim stays moving under the source, no report is
+// written and nothing is probed. Once the tree is proven stopped, the cut's
+// giveUpExecution publishes the complete stopped-source.json and gives up
+// execution citing it, revertClaim writes the claim half (reverting, rev_phase
+// evidence, the lane's park queued), the row goes restoring under the move id,
+// and rev_phase steps evidence to probing (no baselines, so no verify), op.phase
+// following (J1).
+func TestRevertFromStoppingComputesEvidenceFirst(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopping(t)
+ h.arm(t)
+ h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
+ fences, calls := h.fences, h.ev.calls
+ h.pass(t)
+ if h.ev.calls != calls+1 || h.fences != fences+1 {
+ t.Fatalf("the evidence was not asked again behind a fresh fence: evidence %d->%d fences %d->%d", calls, h.ev.calls, fences, h.fences)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != "" {
+ t.Fatalf("the claim moved before the evidence held: %+v", c.Op)
+ }
+ if _, _, err := h.s.Get(context.Background(), h.stoppedKey()); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a stopped report before the evidence held: %v", err)
+ }
+ if len(h.probes) != 0 || len(h.restatus) != 0 || h.route.Status != "moving" || h.route.HoldOp != p1ID {
+ t.Fatalf("something started before the evidence held: probes %d restatus %v row %+v", len(h.probes), h.restatus, h.route)
+ }
+ if m := h.move(t); m.State != provision.MoveReverting || m.Source != nil {
+ t.Fatalf("the record: %+v", m)
+ }
+
+ // The tree is proven stopped; a crash right after the report leaves the
+ // claim still moving under the source: nothing has started.
+ h.ev.stopped = true
+ h.crashAt, h.crashed = "after-report", false
+ h.crashPass(t)
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
+ t.Fatalf("the claim after the report alone: %+v", c.Op)
+ }
+ if len(h.probes) != 0 || len(h.restatus) != 0 {
+ t.Fatalf("something started before execution was given up: probes %d restatus %v", len(h.probes), h.restatus)
+ }
+ h.freshWorker()
+ h.pass(t)
+ rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
+ if err != nil || !rep.Complete() || rep.OpID != p1ID || rep.Role != "source" {
+ t.Fatalf("the stopped report: %v %+v", err, rep)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() {
+ t.Fatalf("the claim half: %+v", c.Op)
+ }
+ if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
+ t.Fatalf("rev_phase %q op.phase %q, want probing for both (no baselines)", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
+ }
+ if o := h.laneChore(t); o == nil || o.Park.Incarnation != h.inc || o.Park.RequestLimit != cutRequestLimit || o.Owner != "region:uk" {
+ t.Fatalf("the lane's park: %+v", o)
+ }
+ if h.verifications() != 0 {
+ t.Fatalf("a revert from stopping verified %d times: no baseline was recorded", h.verifications())
+ }
+ if m := h.move(t); m.Source != nil {
+ t.Fatalf("the revert recorded baselines: %+v", m.Source)
+ }
+ if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
+ t.Fatalf("the row: %+v, want restoring under the move id", h.route)
+ }
+ if len(h.probes) != 1 {
+ t.Fatalf("%d probes, want 1", len(h.probes))
+ }
+ req := h.probes[0]
+ if req.RollbackID != "" || !req.Clean || req.Generation != "" || req.SourceGeneration != "cur" || req.Coverage != "" {
+ t.Fatalf("the probe request: %+v", req)
+ }
+
+ // The probe answers: probed, reopened, the row opened, the claim opened.
+ h.probe = nil
+ h.pass(t)
+ h.reopened(t)
+ m := h.move(t)
+ if m.RecoveryInputs == nil || m.RecoveryInputs.BundleETags["default"] != "e9" || m.Inputs == nil || m.Inputs.BundleETags["default"] != "e1" {
+ t.Fatalf("recovery inputs %+v, move.inputs %+v", m.RecoveryInputs, m.Inputs)
+ }
+ if len(h.opens) != 1 || h.opens[0] != p1ID {
+ t.Fatalf("the open: %v", h.opens)
+ }
+}
+
+// TestRevertVerifyThenProbeNeverVerifiesAgain (§7.4 verify and probing;
+// witness wRollbackJournal for the revert, INV-2): a revert from stopped
+// verifies its files and timers against the stopped baselines, then journals
+// probing before the first attempt. The probe's process writes (the files and
+// the timers change) and the agent restarts: the next pass runs the probe
+// only, never the verification, and the revert is not stuck.
+func TestRevertVerifyThenProbeNeverVerifiesAgain(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
+ h.pass(t)
+ if h.verifications() != 1 {
+ t.Fatalf("verified %d times, want once", h.verifications())
+ }
+ c := h.claim(t)
+ if c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.PhaseProbing || c.Op.Phase != hostclaim.PhaseProbing {
+ t.Fatalf("rev_phase %q op.phase %q, want probing", c.Op.Journal[hostclaim.JournalRevPhase], c.Op.Phase)
+ }
+ if len(h.probes) != 1 || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
+ t.Fatalf("probes %d row %+v", len(h.probes), h.route)
+ }
+ // The probe's process writes, and the agent restarts.
+ if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("written by the probe's process"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.timers = append(h.timers, registry.Timer{Host: p1Host, Name: "added", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
+ h.freshWorker()
+ h.pass(t)
+ if h.verifications() != 1 || len(h.probes) != 2 {
+ t.Fatalf("after the restart: verified %d times, %d probes; want 1 and 2", h.verifications(), len(h.probes))
+ }
+ if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
+ t.Fatalf("the revert stopped on its own probe's writes: %+v", m)
+ }
+ h.probe = nil
+ h.freshWorker()
+ h.pass(t)
+ h.reopened(t)
+ if h.verifications() != 1 || len(h.probes) != 3 {
+ t.Fatalf("verified %d times, %d probes; want 1 and 3", h.verifications(), len(h.probes))
+ }
+}
+
+// TestRevertVerifyMismatchIsStuck (INV-4): a revert from stopped whose files
+// or timers no longer match the stopped baselines stops for an operator with
+// the baseline-mismatch sentence before anything is probed; rev_phase stays
+// verify, the row stays held restoring.
+func TestRevertVerifyMismatchIsStuck(t *testing.T) {
+ for _, what := range []string{"files", "timers"} {
+ t.Run(what, func(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ if what == "files" {
+ if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "upload.txt"), []byte("changed while stopped"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ } else {
+ h.timers = nil
+ }
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Reason != provision.RevertCutDeadline {
+ t.Fatalf("not stuck with the mismatch: %+v", m)
+ }
+ if h.revPhase(t) != hostclaim.RevPhaseVerify || len(h.probes) != 0 {
+ t.Fatalf("rev_phase %q, %d probes", h.revPhase(t), len(h.probes))
+ }
+ if h.route.Status != "restoring" || h.route.HoldOp != p1ID {
+ t.Fatalf("the row: %+v", h.route)
+ }
+ h.pass(t)
+ if h.verifications() != 1 || h.move(t).State != provision.MoveStuck {
+ t.Fatalf("a stuck revert ran again: %d verifications", h.verifications())
+ }
+ })
+ }
+}
+
+// retry is the operator's move resolve --retry: provision.MoveRetry through
+// the record writer, as the console door writes it (Task 7).
+func (h *revertHarness) retry(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.MoveRetry(g)); err != nil {
+ t.Fatal(err)
+ }
+ if m := h.move(t); m.State != provision.MoveReverting || m.StuckReason != "" {
+ t.Fatalf("after the retry: %+v", m)
+ }
+}
+
+// TestRetryResumesARevertsRevPhase (Task 7's box side for a revert, J20): a
+// stuck revert resumes its rev_phase. Stuck at verify, each retry verifies
+// again (never skipped), unrepaired it is stuck again, repaired it goes on.
+// Stuck at probing after fifteen minutes of failed probes, a retry probes
+// again with a fresh fifteen minutes and never verifies again.
+func TestRetryResumesARevertsRevPhase(t *testing.T) {
+ t.Run("verify", func(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ upload := filepath.Join(h.cfg.DataDir, p1Host, "upload.txt")
+ if err := os.WriteFile(upload, []byte("changed while stopped"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveStuck {
+ t.Fatalf("not stuck: %+v", m)
+ }
+ h.retry(t)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || h.verifications() != 2 || len(h.probes) != 0 {
+ t.Fatalf("an unrepaired retry: %+v, %d verifications, %d probes", m, h.verifications(), len(h.probes))
+ }
+ if err := os.WriteFile(upload, []byte("hello"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.retry(t)
+ h.pass(t)
+ h.reopened(t)
+ if h.verifications() != 3 || len(h.probes) != 1 {
+ t.Fatalf("%d verifications, %d probes; want 3 and 1", h.verifications(), len(h.probes))
+ }
+ if got := h.move(t).Reason; got != provision.RevertCutDeadline {
+ t.Fatalf("the retry changed the revert's cause: %q", got)
+ }
+ })
+ t.Run("probing", func(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeWaiting, nil }
+ h.pass(t)
+ h.clock.Advance(10 * time.Minute)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveReverting {
+ t.Fatalf("stuck before fifteen minutes: %+v", m)
+ }
+ h.clock.Advance(6 * time.Minute)
+ h.pass(t)
+ m := h.move(t)
+ if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
+ t.Fatalf("not stuck after fifteen minutes of failed probes: %+v", m)
+ }
+ if h.revPhase(t) != hostclaim.PhaseProbing || h.route.Status != "restoring" || h.route.HoldOp != p1ID {
+ t.Fatalf("rev_phase %q row %+v", h.revPhase(t), h.route)
+ }
+ probes := len(h.probes)
+ h.pass(t)
+ if len(h.probes) != probes {
+ t.Fatal("a stuck revert probed")
+ }
+ // The retry gives the probe fifteen minutes again.
+ h.retry(t)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveReverting || len(h.probes) != probes+1 {
+ t.Fatalf("after the retry: %+v, %d probes", m, len(h.probes))
+ }
+ h.clock.Advance(14 * time.Minute)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveReverting {
+ t.Fatalf("the retry's budget did not start again: %+v", m)
+ }
+ h.clock.Advance(2 * time.Minute)
+ h.pass(t)
+ if m := h.move(t); m.State != provision.MoveStuck || m.StuckReason != provision.StuckRevertProbeBudget {
+ t.Fatalf("not stuck again: %+v", m)
+ }
+ h.retry(t)
+ h.probe = nil
+ h.pass(t)
+ h.reopened(t)
+ if h.verifications() != 1 {
+ t.Fatalf("verified %d times across the retries, want once", h.verifications())
+ }
+ })
+}
+
+// TestRevertOpensAndReleases (§7.4 opened, §7.11): after the reopen the
+// source waits for the lane park the revert queued in the target region. The
+// target region's box archives the lane under parked/<inc>/<host>/<ret>/moves/
+// through the park engine; the obligation is acknowledged only after its
+// horizon; and only then the claim goes to serving under the source and the
+// record closes with LastMove rolled_back, the offline time counted from the
+// cut to the reopen, and the source's journal parked.
+func TestRevertOpensAndReleases(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.pass(t)
+ h.reopened(t)
+ o := h.laneChore(t)
+ if o == nil {
+ t.Fatal("no lane park queued")
+ }
+ lane := h.laneKeys()
+ if len(lane) == 0 {
+ t.Fatal("the fixture has no lane objects")
+ }
+ h.pass(t)
+ h.reopened(t)
+ if h.laneChore(t) == nil || len(h.events) == 0 || h.events[len(h.events)-1] == "rolled_back" {
+ t.Fatalf("the source closed before the lane was parked: %v", h.events)
+ }
+ // The target region's box archives the lane; the horizon has not passed.
+ h.tPass(t)
+ if left := h.laneKeys(); len(left) != 0 {
+ t.Fatalf("the lane still holds %v", left)
+ }
+ prefix := "parked/" + o.Park.Incarnation + "/" + p1Host + "/" + o.Retirement + "/moves/"
+ archived := 0
+ for k := range h.f.buckets["data-uk"] {
+ if strings.HasPrefix(k, prefix) {
+ archived++
+ }
+ }
+ if archived < len(lane) {
+ t.Fatalf("%d archived under %s, the lane had %d", archived, prefix, len(lane))
+ }
+ if h.laneChore(t) == nil {
+ t.Fatal("the lane park was acknowledged before its horizon")
+ }
+ h.pass(t)
+ h.reopened(t)
+ // Past the horizon the target acknowledges, and the source closes.
+ h.pastHorizon(t, o)
+ h.tPass(t)
+ if h.laneChore(t) != nil {
+ t.Fatal("the lane park was not acknowledged after its horizon")
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting || c.Op.Phase != hostclaim.PhaseOpened {
+ t.Fatalf("the claim before the source's pass: %+v", c.Op)
+ }
+ reopenedAt, cutAt := *h.move(t).ReopenedAt, *h.move(t).CutAt
+ h.pass(t)
+ h.closed(t)
+ if got, want := h.read(t, p1Host).LastMove.OfflineS, reopenedAt.Sub(cutAt).Seconds(); got != want || got <= 0 {
+ t.Fatalf("offline %v, want %v", got, want)
+ }
+ if n := countString(h.events, "rolled_back"); n != 1 {
+ t.Fatalf("the rolled_back event %d times: %v", n, h.events)
+ }
+ // Another pass on either box changes nothing.
+ before := h.claim(t)
+ h.pass(t)
+ h.tPass(t)
+ if after := h.claim(t); after.Gen != before.Gen {
+ t.Fatalf("a pass after the close wrote the claim (%d to %d)", before.Gen, after.Gen)
+ }
+}
+
+func countString(xs []string, s string) int {
+ n := 0
+ for _, x := range xs {
+ if x == s {
+ n++
+ }
+ }
+ return n
+}
+
+// TestRevertReplays: a crash after any of the revert's side effects, then a
+// restarted agent, still reaches the close, with the evidence asked once it
+// is journalled, the hold only ever restoring under the move id, and the
+// verification never run again once probing is journalled.
+func TestRevertReplays(t *testing.T) {
+ type entry struct {
+ name string
+ from func(*revertHarness, *testing.T)
+ cut []string // the cut's fault points (giveUpExecution)
+ points []string // the revert's
+ }
+ for _, e := range []entry{
+ {"from stopping", (*revertHarness).fromStopping, []string{"after-checkpoint", "after-report", "after-exec-none"},
+ []string{"after-revert-evidence", "after-revert-claim", "after-restatus", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
+ {"from stopped", (*revertHarness).fromStopped, nil,
+ []string{"after-restatus", "verify", "after-rev-verify", "after-rev-probing", "after-rev-probed", "after-reopened", "after-row-opened", "after-rev-opened", "after-release-move", "after-close"}},
+ } {
+ for _, point := range append(append([]string{}, e.cut...), e.points...) {
+ t.Run(e.name+"/"+point, func(t *testing.T) {
+ h := newRevertHarness(t)
+ e.from(h, t)
+ if e.name == "from stopping" {
+ h.ev.stopped = true
+ }
+ h.arm(t)
+ isCut := false
+ for _, p := range e.cut {
+ isCut = isCut || p == point
+ }
+ // The fixture's own crash (fromStopped) has used the cut's
+ // seam: each case watches only its own.
+ crashed := func() bool { return h.rvDone }
+ if isCut {
+ h.crashAt, h.crashed = point, false
+ crashed = func() bool { return h.crashed }
+ } else {
+ h.rvAt = point
+ }
+ for i := 0; i < 8 && !crashed(); i++ {
+ h.pass(t)
+ h.tPass(t)
+ if c := h.claim(t); c.Op.Kind == hostclaim.KindReverting && c.Op.Phase == hostclaim.PhaseOpened {
+ if o := h.laneChore(t); o != nil {
+ h.pastHorizon(t, o)
+ }
+ }
+ }
+ if !crashed() {
+ t.Fatalf("never reached %s (revert %v, cut %v)", point, h.rvHits, h.hits)
+ }
+ h.freshWorker()
+ h.drive(t)
+ h.closed(t)
+ switch {
+ case e.name == "from stopping" && h.verifications() != 0:
+ t.Fatalf("a revert from stopping verified %d times", h.verifications())
+ case e.name == "from stopped" && point == "verify" && h.verifications() != 2:
+ t.Fatalf("verified %d times after a crash before the verification, want 2 (the crashed one counted)", h.verifications())
+ case e.name == "from stopped" && point != "verify" && h.verifications() != 1:
+ t.Fatalf("verified %d times, want once", h.verifications())
+ }
+ if len(h.probes) == 0 {
+ t.Fatal("never probed")
+ }
+ })
+ }
+ }
+}
+
+// TestRevertJournalSweptAfterTheClose (the local journal sweep, Tasks 4 and
+// 5): a crash after the close and before the source's journal is parked
+// leaves it to the sweep, which parks it once no record carries the move; a
+// journal whose revert is still in flight is left.
+func TestRevertJournalSweptAfterTheClose(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ h.arm(t)
+ h.pass(t)
+ h.reopened(t)
+ // In flight: the sweep leaves the marked journal.
+ h.w().sweepFinalJournals([]provision.Record{h.read(t, p1Host)})
+ if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
+ t.Fatal("the sweep parked a journal whose revert is in flight")
+ }
+ h.rvAt = "after-close"
+ o := h.laneChore(t)
+ h.tPass(t)
+ h.pastHorizon(t, o)
+ h.tPass(t)
+ h.pass(t)
+ if !h.rvDone {
+ t.Fatalf("never reached the close: %v", h.rvHits)
+ }
+ if h.read(t, p1Host).Move != nil {
+ t.Fatal("the close did not land")
+ }
+ if !moveJournalPresent(h.cfg.DataDir, p1Host, p1ID) {
+ t.Fatal("the crash after the close parked the journal")
+ }
+ h.freshWorker()
+ h.pass(t)
+ h.closed(t)
+}
+
+// TestRevertUnarmedInProduction (J18): production wires no recovery probe
+// until Task 9 and no park engine leases until Plan 5, so the revert's
+// recovery and the target's lane park are unarmed there. The claim half
+// still follows the record (Plan 2's revertClaim); nothing else moves.
+func TestRevertUnarmedInProduction(t *testing.T) {
+ if revertSourceArmed(moveCfg{Box: "ie-box", DataDir: "/d", RestatusHold: func(string, string, string, func() error) (bool, error) { return false, nil },
+ ReleaseHeld: func(string, string) (bool, error) { return false, nil }, Timers: func(string) ([]registry.Timer, error) { return nil, nil },
+ FilesDir: func(string) (string, string) { return "", "" }}) {
+ t.Fatal("the revert is armed without a recovery probe")
+ }
+ if revertLaneArmed(moveCfg{Box: "uk-box", Region: "uk", OwnBucket: "data-uk", Versioned: func(context.Context, string, string) (lease.VersionedStore, error) { return nil, nil }}) {
+ t.Fatal("the lane park is armed without the park engine's leases")
+ }
+ t.Run("from stopping", func(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopping(t)
+ h.ev.stopped = true
+ h.pass(t)
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
+ t.Fatalf("an unarmed revert gave up execution: %+v", c.Op)
+ }
+ })
+ t.Run("from stopped", func(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopped(t)
+ lane := len(h.laneKeys())
+ h.pass(t)
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindReverting || c.Op.Journal[hostclaim.JournalRevPhase] != hostclaim.RevPhaseEvidence || h.route.Status != "moving" {
+ t.Fatalf("an unarmed revert went on: %+v row %+v", c.Op, h.route)
+ }
+ tc := h.target()
+ tc.DataDir = h.tdir
+ h.tw = &movesWorker{cfg: tc, mirrors: map[string]lease.MirrorStore{}}
+ if err := movesPass(context.Background(), h.s, h.tw); err != nil {
+ t.Fatal(err)
+ }
+ if len(h.laneKeys()) != lane || len(h.tleases.acquired) != 0 {
+ t.Fatal("an unarmed target parked the lane")
+ }
+ })
+ t.Run("the claim half", func(t *testing.T) {
+ // The record reverts and the claim half is lost: the unarmed pass
+ // still writes it, as Plan 2's reverting arm did.
+ h := newRevertHarness(t)
+ h.ready(t)
+ h.crashAt = "after-exec-none"
+ h.cut(t)
+ h.crashPass(t)
+ g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
+ if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
+ t.Fatal(err)
+ }
+ h.pass(t)
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
+ t.Fatalf("the claim half: %+v", c.Op)
+ }
+ })
+}
+
+// A reverting record's target region box has neither the row nor a journal
+// of the move (the revert never flipped), so the participant rule answers
+// it as the target on the state alone: its one piece of work is the lane park
+// the claim owes its region. Any other state still needs the row or the
+// journal.
+func TestRevertingTargetIsAParticipant(t *testing.T) {
+ cfg := moveCfg{Region: "uk", DataDir: t.TempDir(), OwnsRoute: func(string) bool { return false }}
+ rec := func(state string) provision.Record {
+ return provision.Record{Host: p1Host, Region: "ie", Move: &provision.MoveState{ID: p1ID, State: state, From: "ie", To: "uk"}}
+ }
+ if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleTarget {
+ t.Fatalf("a reverting record's target: %q", got)
+ }
+ for _, state := range []string{provision.MoveCommitted, provision.MoveRollingBack, provision.MoveStuck, provision.MoveCutting} {
+ if got := movesParticipantRecord(cfg, rec(state)); got != roleNone {
+ t.Fatalf("%s with no row and no journal: %q", state, got)
+ }
+ }
+ cfg.Region = "fr"
+ if got := movesParticipantRecord(cfg, rec(provision.MoveReverting)); got != roleNone {
+ t.Fatalf("a third region: %q", got)
+ }
+}
+
+// restatusHold (production's RestatusHold): an op's own hold changes status
+// under the host's hold lock, after the caller's validation, touching the
+// stage-requested marker only when the change landed. A replay answers true
+// and touches nothing; a row held by another op, an unheld row and a missing
+// row answer false and are left; a validation error moves nothing.
+func TestRestatusHoldIsConditional(t *testing.T) {
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatal(err)
+ }
+ defer reg.Close()
+ for _, rt := range []registry.Route{
+ {Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
+ {Host: "b.go.example.com", Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true},
+ } {
+ if err := reg.Upsert(rt); err != nil {
+ t.Fatal(err)
+ }
+ }
+ var written []string
+ restatus := restatusHold(reg, func(host string) { written = append(written, host) })
+ if ok, err := restatus(p1Host, "restoring", p1ID, nil); ok || err != nil {
+ t.Fatalf("an unheld row: %v %v", ok, err)
+ }
+ if err := holdMoving(reg, nil)(p1Host, p1ID); err != nil {
+ t.Fatal(err)
+ }
+ refused := errors.New("the record moved on")
+ if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return refused }); ok || !errors.Is(err, refused) {
+ t.Fatalf("a refused validation: %v %v", ok, err)
+ }
+ if rt, _ := reg.Get(p1Host); rt.Status != "moving" || len(written) != 0 {
+ t.Fatalf("a refused validation moved the row: %+v, marker %d", rt, len(written))
+ }
+ for i := 0; i < 2; i++ {
+ if ok, err := restatus(p1Host, "restoring", p1ID, func() error { return nil }); !ok || err != nil {
+ t.Fatalf("restatus %d: %v %v", i, ok, err)
+ }
+ }
+ if rt, _ := reg.Get(p1Host); rt.Status != "restoring" || rt.HoldOp != p1ID || len(written) != 1 {
+ t.Fatalf("row %+v; marker touched %d times", rt, len(written))
+ }
+ if ok, err := restatus(p1Host, "restoring", p1ID2, nil); ok || err != nil {
+ t.Fatalf("another op's hold: %v %v", ok, err)
+ }
+ if err := holdMoving(reg, nil)("b.go.example.com", p1ID2); err != nil {
+ t.Fatal(err)
+ }
+ if ok, err := restatus("b.go.example.com", "restoring", p1ID, nil); ok || err != nil {
+ t.Fatalf("a row held by another op: %v %v", ok, err)
+ }
+ if rt, _ := reg.Get("b.go.example.com"); rt.Status != "moving" || rt.HoldOp != p1ID2 {
+ t.Fatalf("another op's row moved: %+v", rt)
+ }
+ if ok, err := restatus("absent.example.com", "restoring", p1ID, nil); ok || err != nil {
+ t.Fatalf("a missing row: %v %v", ok, err)
+ }
+ // Production wires it, and leaves the revert and the lane park unarmed.
+ var prod moveCfg
+ hib := activator.Config{Bucket: "data-ie", Region: "eu-west-1", Prefix: "instances"}
+ armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, t.TempDir(), reg, nil, nil, hostwrite.Config{}, func(string) string { return "" })
+ if prod.RestatusHold == nil || prod.ReleaseHeld == nil {
+ t.Fatal("production does not wire the revert's row hooks")
+ }
+ prod.Box, prod.Region, prod.OwnBucket = "ie-box", "ie", "data-ie"
+ if revertSourceArmed(prod) || revertLaneArmed(prod) {
+ t.Fatal("production armed the revert")
+ }
+}
+
+// A revert from stopping whose files directory cannot be archived (a special
+// file) is not refused again: nothing compares a files digest on this path
+// (§7.4 evidence), so giveUpExecution publishes the report without one,
+// records no baselines, and the revert goes on.
+func TestRevertFromStoppingWithUnarchivableFiles(t *testing.T) {
+ h := newRevertHarness(t)
+ h.fromStopping(t)
+ if err := os.Symlink("/etc/passwd", filepath.Join(h.cfg.DataDir, p1Host, "link")); err != nil {
+ t.Fatal(err)
+ }
+ h.arm(t)
+ h.ev.stopped = true
+ h.pass(t)
+ h.reopened(t)
+ rep, err := readStoppedReport(context.Background(), h.s, h.stoppedKey())
+ if err != nil || !rep.Complete() || rep.FilesPlain != "" {
+ t.Fatalf("the stopped report: %v %+v", err, rep)
+ }
+ if m := h.move(t); m.Source != nil || m.Reason != provision.RevertStopEvidenceBudget {
+ t.Fatalf("the record: %+v", m)
+ }
+ if h.verifications() != 0 || len(h.probes) != 1 {
+ t.Fatalf("%d verifications, %d probes", h.verifications(), len(h.probes))
+ }
+}