amadan

carlos / platform Public

host: the source's recovery, clean and dirty, through opened and the rolled_back finalization (§7.8)

The source box takes a rolled-back host back (Plan 4 Task 5): waiting for
the target's evidence; the claim's return at S (src_gen for a clean
return) and the guard's activation; the row handed from the move's hold to
the rollback's (J3); the clean verification against the stopped
baselines, stuck on a mismatch (INV-4), or the dirty return's install
through the shared installer from the return lane; probing behind the
recoveryProbe hook (Task 9, nil in production); probed with the return
lane's park; recovered; the open (J12); and the finalization: the
terminal event, ReleaseMove and MoveRolledBackClose once every chore is
acknowledged. Routed by movesParticipantRecord; unarmed in production
until Plan 5.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 745578ab1fe2ced910814d071618749a27a42b43 parent 8e9dcb3
4 files changed, +1499 −0
  • approved_moves_copy_test.go +1 −0
  • internal/host/movespass.go +29 −0
  • internal/host/movesrecover.go +783 −0
  • internal/host/movesrecover_test.go +686 −0
diff --git a/approved_moves_copy_test.go b/approved_moves_copy_test.go
index 29d2a52fb..840338245 100644
--- a/approved_moves_copy_test.go
+++ b/approved_moves_copy_test.go
@@ -588,6 +588,7 @@ var movesScanFiles = []string{
"internal/host/movesphase1.go",
"internal/host/movespolicy.go",
"internal/host/movesprobe.go",
+ "internal/host/movesrecover.go",
"internal/host/movesreport.go",
"internal/host/movesretire.go",
"internal/host/movesrollback.go",
diff --git a/internal/host/movespass.go b/internal/host/movespass.go
index c7888a25b..993cfe4e5 100644
--- a/internal/host/movespass.go
+++ b/internal/host/movespass.go
@@ -322,6 +322,9 @@ func armPhase1(cfg *moveCfg, w *movesWorker, hib activator.Config, dataDir strin
// The rollback's release hands the row's hold from the move to the
// rollback, touching the stage-requested marker (ruling J3).
cfg.HandHold = handHold(reg, holdStageRequester(dataDir))
+ // 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))
}
if eb != nil {
// A box with no delegated cgroup has no manager, and Reconstruct
@@ -574,6 +577,20 @@ type moveCfg struct {
HandHold func(host, status, toOp, fromOp string, validate func() error) (bool, error)
LogEvent func(ctx context.Context, rec provision.Record, event string)
rbFault func(point string) error
+
+ // The source's recovery after a rollback (§7.8, movesrecover.go).
+ // ReleaseHeld opens the row: registry.ReleaseHold under the op that
+ // holds it, touching the stage-requested marker when it landed
+ // (releaseHeld); the caller holds provision.WithHostHoldLock (J12). The
+ // recovery also uses HandHold, LogEvent, the landing's hooks (the dirty
+ // return's install) and Timers above; ParkLeases nil leaves it unarmed
+ // in production (recoverySourceArmed). recoveryProbe is the recovery's
+ // probe (Task 9, BLOCKED-ON-CLAIM C1, rulings J2 and J18): nil in
+ // production, where probing logs and waits, and faked in the tests.
+ // rcFault is the recovery's test seam, nil in production.
+ 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
}
// defaultMoveReadyDelta is the readiness bar when
@@ -644,6 +661,18 @@ func movesPass(ctx context.Context, s release.Store, w *movesWorker) error {
}
continue
}
+ if rec.Move.State == provision.MoveRollingBack && movesParticipantRecord(cfg, rec) == roleSource {
+ // The source's recovery after a rollback (movesrecover.go,
+ // Plan 4 Task 5): the record's region is the source's again,
+ // and this box is the move's source by the post-flip
+ // participant rule (J10), whether or not it still has the row.
+ if err := w.recoverSource(ctx, s, rec); 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/movesrecover.go b/internal/host/movesrecover.go
new file mode 100644
index 000000000..ecf858090
--- /dev/null
+++ b/internal/host/movesrecover.go
@@ -0,0 +1,783 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "log"
+ "strconv"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "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 source's recovery after a rollback (moves spec §7.8 source table; Plan
+// 4 Task 5), behind CARLOS_MOVES_PHASE1: the target has released, and the
+// source box takes the host back, clean or dirty, through opened, and then
+// the rolled_back finalization.
+//
+// Routing. MoveRollBack moved the record's region back to From, so the
+// source reaches a rolling_back record through the post-flip participant
+// rule (movesParticipantRecord, J10) in movesPass, whether or not it still
+// has its row. One pass drives it as far as it can, each phase from durable
+// state, under the record's source_phase. A record at X means X's work may
+// have begun (Plan 2's convention, G8; provision's "each is written before
+// its side effects"):
+//
+// - waiting: until the target's evidence, released clean or returned
+// dirty on the record, and the claim restoring/released with nobody
+// executing ("" after the target's evidence, "none" after a
+// declaration). Then the record says claimed.
+// - claimed: the claim's return (hostclaim.ReturnToSource: this box
+// executing at the source's own stamp S; a clean return also names the
+// source's generation, src_gen, again), then the guard's activation
+// (the fresh read finds holder.stamp == S, matching the installation
+// record, or re-initialises a lost one to S), then, for a dirty return
+// whose inventory lists paths of S, the source's park-local (J13: the
+// installer's local retires S against it). The claim's phase is
+// claimed from here through the install (J1).
+// - restoring: the row handed from the move's moving hold to the
+// rollback's restoring hold (handHold, J3), with the record read fresh
+// under the host's hold lock.
+// - clean verified: the files plain digest and the timers digest equal
+// the source's stopped baselines (Source.Files.Plain,
+// Source.TimersSHA256), and the warm database's <dbPath>.gen names
+// src_gen (plain text, H11). A mismatch stops for an operator with the
+// baseline-mismatch sentence at verified (MoveMarkStuck, INV-4); a
+// retry verifies again, never skips it.
+// - dirty parking to configured: the shared installer (installGeneration,
+// Task 3) bound by recoveryInstallSource to the return seal and the
+// return baseline the target journalled in the claim (Task 4), from the
+// return lane moves/<host>/<rb id>/ in this box's own bucket, under S;
+// configured records the inputs it rendered as recovery_inputs, never
+// compared with move.inputs. restored hands over to probing.
+// - probing: the claim steps claimed to probing first (J1), then the
+// recovery probe: recoveryProbe, Task 9's (BLOCKED-ON-CLAIM C1, rulings
+// J2 and J18), a cold admission of kind recovery with nine attempts,
+// into the return generation (dirty, coverage sealed) or a fresh one
+// restored from src_gen (clean, coverage clean-return; J14: both
+// returns are cold and never keep S). The probe records the inputs it
+// renders before it renders them (MoveRecordRecoveryInputs). Once the
+// record says probing nothing earlier runs again: a restarted probe
+// restarts only the probe (INV-2, INV-7). Nine failed attempts stop
+// for an operator at probing.
+// - probed: the claim steps to probed and, for a dirty return, queues the
+// return lane's park in this region in the same write
+// (QueueReturnLanePark, its horizon the return seal's bound); then
+// recovered_at, and the serial is nudged (the target's cleaning waits
+// on it).
+// - recovered: the open: the rollback's hold on the row ends
+// (ReleaseHeld, a plain registry call under the host's hold lock with
+// the record read fresh, J12), the claim steps to opened, and the record
+// says opened with opened_at.
+// - opened: the finalization. Once recovered_at, opened_at and
+// target_parked_at are stamped and no obligation of the op remains in
+// the claim, the terminal event (LogEvent, Plan 6's moves.LogEvent),
+// the claim's release to serving (ReleaseMove) and the record's close
+// (MoveRolledBackClose). Whichever actor stamps last tries; the
+// source's next pass, and Plan 5's console pass, finish what a crash
+// leaves.
+//
+// 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 target's half of the
+// rollback, the source's recovery is unarmed in production until Plan 5:
+// it needs the park engine's leases (ParkLeases, for the dirty return's
+// residue parks), which production does not wire, so recoverySourceArmed
+// answers false there and the pass logs a standing notice. The probe hook is
+// nil in production too, until Task 9; probing then logs and waits. LogEvent
+// is nil in production, where the event is a log line.
+
+// recoverStepsPerPass bounds the phases one pass runs: the dirty path from
+// waiting to the finalization is fifteen.
+const recoverStepsPerPass = 16
+
+// recoveryProbeAttempts is the recovery probe's budget (§7.8: three times
+// the landing's three).
+const recoveryProbeAttempts = 9
+
+// The recovery probe's admission coverage (§7.14): a dirty return admits
+// the return generation from its sealed lane, a clean one restores the
+// source's own generation.
+const (
+ recoveryCoverageSealed = "sealed"
+ recoveryCoverageClean = "clean-return"
+)
+
+// recoveryProbeOutcome is what one call of the recovery probe found.
+type recoveryProbeOutcome int
+
+const (
+ // recoveryProbeWaiting: no answer yet; the next pass calls the probe
+ // again (and only the probe).
+ recoveryProbeWaiting recoveryProbeOutcome = iota
+ // recoveryProbeAnswered: the app answered; the recovery goes to probed.
+ recoveryProbeAnswered
+ // recoveryProbeExhausted: every attempt failed; the recovery stops for
+ // an operator (StuckRecoveryProbeAttempts).
+ recoveryProbeExhausted
+)
+
+// recoveryProbeRequest is what the source asks of the recovery probe (Task
+// 9 builds it against claim Group D for recovery, J2).
+type recoveryProbeRequest struct {
+ // RollbackID names the rollback, whose id names a dirty return's
+ // generation and journal.
+ RollbackID string
+ // Clean is the return's kind; Coverage is the admission's coverage,
+ // "sealed" (dirty) or "clean-return".
+ Clean bool
+ Coverage string
+ // Generation is the generation the dirty return installed (the
+ // rollback id); empty for a clean return, whose probe admits a fresh
+ // one restored from SourceGeneration (src_gen, J14).
+ Generation, SourceGeneration string
+ // Attempts is the budget, nine.
+ Attempts int
+ // Record records the inputs the probe is about to render, before it
+ // renders them (MoveRecordRecoveryInputs; Task 3b concern 3).
+ Record func(context.Context, provision.MoveInputs) error
+}
+
+// recoverySourceArmed reports whether this box can run the source's
+// recovery: every hook a dirty return's install and the open call is wired,
+// the park engine's leases included. In production ParkLeases is not
+// (ruling G17), so this answers false there (J18).
+func recoverySourceArmed(cfg moveCfg) bool {
+ return landingArmed(cfg) && cfg.DataDir != "" && cfg.HandHold != nil && cfg.ReleaseHeld != nil && cfg.Timers != nil
+}
+
+// releaseHeld is the recovery's open in production: registry.ReleaseHold
+// under the op that holds the row, then written (holdStageRequester) when
+// it landed, so root re-judges what it staged now the hold has ended. The
+// caller holds provision.WithHostHoldLock.
+func releaseHeld(reg *registry.Registry, written func(host string)) func(host, op string) (bool, error) {
+ return func(host, op string) (bool, error) {
+ released, err := reg.ReleaseHold(host, op)
+ if err == nil && released && written != nil {
+ written(host)
+ }
+ return released, err
+ }
+}
+
+// recoverRun is one pass's state for one rolling_back record on its source.
+type recoverRun struct {
+ w *movesWorker
+ s release.Store
+ rec provision.Record
+ g provision.MoveGuard
+ rb string
+}
+
+// recoverSource is the participant step for a rolling_back record on its
+// source (movesPass).
+func (w *movesWorker) recoverSource(ctx context.Context, s release.Store, rec provision.Record) error {
+ cfg := w.cfg
+ m := rec.Move
+ if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || m.From != cfg.Region || m.To == cfg.Region {
+ return nil
+ }
+ if !recoverySourceArmed(cfg) {
+ w.notice(rec.Host+"#recover", "moves: %s: the source's recovery of rollback %s waits: this box's recovery is not armed (Plan 5's park view and the park engine's leases are needed)", rec.Host, m.Rollback.ID)
+ return nil
+ }
+ r := &recoverRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID}
+ defer r.settleLease()
+ for i := 0; i < recoverStepsPerPass; 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 *recoverRun) cfg() moveCfg { return r.w.cfg }
+func (r *recoverRun) now() time.Time { return r.w.now() }
+func (r *recoverRun) rollback() *provision.MoveRollback { return r.rec.Move.Rollback }
+func (r *recoverRun) dbPath() string { return provision.DefaultDBPath(r.cfg().DataDir, r.rec.Host) }
+
+// same reports whether fresh is still this rollback, rolling back.
+func (r *recoverRun) same(fresh *provision.Record) bool {
+ m := fresh.Move
+ return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.ID == r.rb
+}
+
+// fault is the test seam after a side effect.
+func (r *recoverRun) fault(point string) error {
+ if f := r.cfg().rcFault; f != nil {
+ return f(point)
+ }
+ return nil
+}
+
+// settleLease releases the own-location lease a dirty return's build holds
+// from building through published, once the record is past them or no
+// longer this return: the lease is the build's alone.
+func (r *recoverRun) settleLease() {
+ m := r.rec.Move
+ if m != nil && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.Dirty &&
+ (m.Rollback.SourcePhase == provision.RbSourceBuilding || m.Rollback.SourcePhase == provision.RbSourcePublished) {
+ return
+ }
+ releaseOwnLease(r.w, r.rec.Host)
+}
+
+// step runs the source's current recovery phase once. advanced is true when
+// it wrote the record.
+func (r *recoverRun) step(ctx context.Context) (bool, error) {
+ rb := r.rollback()
+ switch rb.SourcePhase {
+ case provision.RbSourceWaiting:
+ return r.waiting(ctx)
+ case provision.RbSourceClaimed:
+ return r.claimed(ctx)
+ case provision.RbSourceRestoring:
+ return r.restoring(ctx)
+ case provision.RbSourceVerified:
+ return r.verified(ctx)
+ case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured:
+ return r.install(ctx)
+ case provision.RbSourceRestored:
+ return r.phase(ctx, provision.RbSourceRestored, provision.RbSourceProbing, nil, "")
+ case provision.RbSourceProbing:
+ return r.probing(ctx)
+ case provision.RbSourceProbed:
+ return r.probed(ctx)
+ case provision.RbSourceRecovered:
+ return r.open(ctx)
+ case provision.RbSourceOpened:
+ return r.finalize(ctx)
+ }
+ return false, nil
+}
+
+// phase writes one source phase step, then logs the event, if any, and
+// passes the step's fault point. A conflict is someone else's decision.
+func (r *recoverRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), event string) (bool, error) {
+ host := r.rec.Host
+ if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRollbackSourcePhase(r.g, from, to, stamp)); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: the source's recovery %s to %s refused (%v); the record moved on", host, from, to, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("the source's recovery %s to %s: %w", from, to, err)
+ }
+ log.Printf("moves: %s: the source of rollback %s is %s", host, r.rb, to)
+ if event != "" {
+ r.w.logEvent(ctx, r.rec, event)
+ }
+ if err := r.fault("after-record-" + to); err != nil {
+ return true, err
+ }
+ return true, nil
+}
+
+// stuck stops the recovery for an operator (MoveMarkStuck) with a recorded
+// reason; the row stays held and move --retry returns to the phase.
+func (r *recoverRun) stuck(ctx context.Context, reason string, why error) (bool, error) {
+ host := r.rec.Host
+ log.Printf("moves: %s: the recovery of %s stops for an operator (%v): %s", host, r.rb, why, reason)
+ if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveMarkStuck(r.g, reason)); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: the stuck mark of %s refused (%v); the record moved on", host, r.rb, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("mark the recovery stuck: %w", err)
+ }
+ r.w.logEvent(ctx, r.rec, "source-stuck")
+ nudgeSerial(ctx, r.s)
+ return true, nil
+}
+
+// claim reads the host claim and requires this rollback's restoring op,
+// with this box the move's source, for the record's incarnation.
+func (r *recoverRun) claim(ctx context.Context) (hostclaim.Observed, error) {
+ obs, err := hostclaim.Read(ctx, r.s, r.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", r.g.ID)
+ }
+ if err := r.ours(obs.Claim); err != nil {
+ return obs, fmt.Errorf("move %s: %w", r.g.ID, err)
+ }
+ return obs, nil
+}
+
+// ours is the claim check of every source phase before the finalization:
+// this rollback's restoring op, this box the move's source, the record's
+// incarnation.
+func (r *recoverRun) ours(c *hostclaim.Claim) error {
+ box := r.cfg().Box
+ if c == nil || c.Op.ID != r.g.ID || c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != r.rb {
+ return fmt.Errorf("%w: the claim is not rollback %s", hostclaim.ErrWrongState, r.rb)
+ }
+ if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation {
+ return fmt.Errorf("%w: the claim names another incarnation", hostclaim.ErrWrongState)
+ }
+ if len(c.Op.Participants) == 0 || c.Op.Participants[0] != box || c.Op.Intent["source"] != box {
+ return fmt.Errorf("%w: this box is not the move's source in the claim", hostclaim.ErrWrongState)
+ }
+ return nil
+}
+
+// executing requires the claim at phase with this box executing.
+func (r *recoverRun) executing(c *hostclaim.Claim, phase string) error {
+ if c.Op.Phase != phase || c.Op.Exec != r.cfg().Box {
+ return fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not %s on this box", r.g.ID, c.Op.Phase, c.Op.Exec, phase)
+ }
+ return nil
+}
+
+// sourceStamp is S, the source installation's stamp the flip journalled.
+func sourceStamp(c *hostclaim.Claim) (uint64, error) {
+ s, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
+ if err != nil || s == 0 {
+ return 0, fmt.Errorf("%w: the flip's source stamp is not journalled", hostclaim.ErrWrongState)
+ }
+ return s, nil
+}
+
+// stillAt is the record check run under the host's hold lock before a row
+// write: the record, read fresh, must still be this rollback with the
+// source at phase. It reads the record only.
+func (r *recoverRun) stillAt(ctx context.Context, phase string) 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) || fresh.Move.Rollback.SourcePhase != phase {
+ return fmt.Errorf("move %s: the record is no longer rollback %s at %s", r.g.ID, r.rb, phase)
+ }
+ return nil
+}
+
+// waiting (§7.8 source waiting): the target's evidence, released clean or
+// returned dirty, and the claim released with nobody executing.
+func (r *recoverRun) waiting(ctx context.Context) (bool, error) {
+ host, rb := r.rec.Host, r.rollback()
+ if !(rb.TargetPhase == provision.RbTargetReleased && !rb.Dirty) && !(rb.TargetPhase == provision.RbTargetReturned && rb.Dirty) {
+ r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the target of rollback %s (%s, dirty %t)", host, r.rb, rb.TargetPhase, rb.Dirty)
+ return false, nil
+ }
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || (op.Exec != "" && op.Exec != "none") {
+ r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the claim's release of rollback %s (restoring/%q, executor %q)", host, r.rb, op.Phase, op.Exec)
+ return false, nil
+ }
+ return r.phase(ctx, provision.RbSourceWaiting, provision.RbSourceClaimed, nil, "")
+}
+
+// claimed (§7.8 source claimed): the claim's return to this box at S (and
+// src_gen for a clean return), the guard's activation at S, and a dirty
+// return's park-local for its registered pre-move installation (J13).
+func (r *recoverRun) claimed(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id, rb := r.rec.Host, r.g.ID, r.rollback()
+ clean := !rb.Dirty
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if op := obs.Claim.Op; op.Phase != hostclaim.PhaseClaimed || op.Exec != cfg.Box {
+ if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.ReturnToSource(id, cfg.Box, clean, hostclaim.ReturnFence{})); err != nil {
+ return false, fmt.Errorf("the claim's return to the source: %w", err)
+ }
+ log.Printf("moves: %s: the source executes the host again for rollback %s (clean %t)", host, r.rb, clean)
+ }
+ if err := r.fault("after-return-claimed"); err != nil {
+ return false, err
+ }
+ c := obs.Claim
+ s, err := sourceStamp(c)
+ if err != nil {
+ return false, err
+ }
+ if c.Holder.Stamp != s {
+ return false, fmt.Errorf("move %s: the holder is at stamp %d, not the source's %d", id, c.Holder.Stamp, s)
+ }
+ if clean && c.Holder.Generation != c.Op.Journal[hostclaim.JournalSrcGen] {
+ return false, fmt.Errorf("move %s: a clean return's holder names generation %q, not src_gen", id, c.Holder.Generation)
+ }
+ g := cfg.Guard
+ if g.Claims == nil {
+ g.Claims = hostguard.StoreReader{Store: r.s}
+ }
+ if err := g.Activate(ctx, host, cfg.Box, nil); err != nil {
+ return false, fmt.Errorf("activate the source's installation at stamp %d: %w", s, err)
+ }
+ if err := r.fault("after-activate"); err != nil {
+ return false, err
+ }
+ if rb.Dirty && hostclaim.InventoryHasStamp(c, cfg.Box, s) {
+ if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil {
+ return false, fmt.Errorf("queue the source's local park: %w", err)
+ }
+ }
+ return r.phase(ctx, provision.RbSourceClaimed, provision.RbSourceRestoring, nil, "")
+}
+
+// restoring (§7.8 source restoring): the row handed from the move's hold to
+// the rollback's, as restoring (J3), with the record checked under the
+// host's hold lock; then the branch, clean or dirty.
+func (r *recoverRun) restoring(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ handed, err := cfg.HandHold(host, "restoring", r.rb, id, func() error { return r.stillAt(ctx, provision.RbSourceRestoring) })
+ if err != nil {
+ return false, fmt.Errorf("hand the row's hold to the rollback: %w", err)
+ }
+ if !handed {
+ return false, fmt.Errorf("move %s: this box's row is held by neither the move nor rollback %s", id, r.rb)
+ }
+ if err := r.fault("after-hand"); err != nil {
+ return false, err
+ }
+ next := provision.RbSourceVerified
+ if r.rollback().Dirty {
+ next = provision.RbSourceParking
+ }
+ return r.phase(ctx, provision.RbSourceRestoring, next, nil, "")
+}
+
+// verified (§7.8 clean verified, INV-4): the source's files and timers hash
+// to what it recorded when it stopped, and its warm database's marker names
+// src_gen. A mismatch stops for an operator; a retry runs this again.
+func (r *recoverRun) verified(ctx context.Context) (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
+ }
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil {
+ return false, err
+ }
+ srcGen, ok := obs.Claim.Op.Journal[hostclaim.JournalSrcGen]
+ if !ok {
+ return false, fmt.Errorf("move %s: the flip's source generation is not journalled", id)
+ }
+ if m.Source == nil || m.Source.Files == nil || m.Source.Files.Plain == "" || 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+"#rc-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))
+ }
+ gen, err := hostbox.ReadGenerationMarker(r.dbPath() + ".gen")
+ if err != nil {
+ return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the warm database's marker: %w", err))
+ }
+ if gen != srcGen {
+ return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the warm database's marker names %q, the pre-move generation is %q", gen, srcGen))
+ }
+ log.Printf("moves: %s: the source still matches what it recorded when it stopped", host)
+ return r.phase(ctx, provision.RbSourceVerified, provision.RbSourceProbing, nil, "")
+}
+
+// returnSeal reads the return seal the target journalled in the claim.
+func returnSeal(c *hostclaim.Claim) (moves.Seal, error) {
+ raw, ok := c.Op.Journal[moves.JournalReturnSeal]
+ if !ok {
+ return moves.Seal{}, fmt.Errorf("%w: the claim carries no return seal", moves.ErrBadSeal)
+ }
+ return moves.DecodeSeal(raw)
+}
+
+// install (§7.8 dirty parking to configured): one phase of the shared
+// installer over the return lane, its seal and baseline the target's
+// journalled return. A refusal it records stops the return for an operator.
+func (r *recoverRun) install(ctx context.Context) (bool, error) {
+ host := r.rec.Host
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ raw, ok := obs.Claim.Op.Journal[moves.JournalReturnBaseline]
+ if !ok {
+ return false, fmt.Errorf("move %s: the claim carries no return baseline", r.g.ID)
+ }
+ baseline, err := moves.DecodeReturnBaseline(raw)
+ if err != nil {
+ return false, err
+ }
+ src, err := recoveryInstallSource(r.w, r.s, r.rec, returnSeal, &baseline)
+ if err != nil {
+ return false, err
+ }
+ advanced, refused, err := installGeneration(ctx, r.s, r.w, r.rec, src)
+ if refused != "" {
+ log.Printf("moves: %s: the return of %s stopped for an operator: %s", host, r.rb, refused)
+ r.w.logEvent(ctx, r.rec, "source-stuck")
+ }
+ return advanced, err
+}
+
+// probing (§7.8 source probing): the claim steps to probing before the
+// first attempt (J1), then the recovery probe. Nothing earlier is run again
+// from here.
+func (r *recoverRun) probing(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id, rb := r.rec.Host, r.g.ID, r.rollback()
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if obs.Claim.Op.Phase == hostclaim.PhaseClaimed {
+ if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil {
+ return false, err
+ }
+ if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseClaimed, hostclaim.PhaseProbing)); err != nil {
+ return false, fmt.Errorf("the claim's probing: %w", err)
+ }
+ if err := r.fault("after-claim-probing"); err != nil {
+ return false, err
+ }
+ }
+ if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil {
+ return false, err
+ }
+ if cfg.recoveryProbe == nil {
+ r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: it is Task 9's, on claim Plan 10's recovery admission", host, r.rb)
+ return false, nil
+ }
+ req := recoveryProbeRequest{RollbackID: r.rb, Clean: !rb.Dirty, Attempts: recoveryProbeAttempts,
+ SourceGeneration: obs.Claim.Op.Journal[hostclaim.JournalSrcGen],
+ Record: func(ctx context.Context, in provision.MoveInputs) error {
+ _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRecoveryInputs(r.g, r.rb, in))
+ return err
+ }}
+ if rb.Dirty {
+ req.Coverage, req.Generation = recoveryCoverageSealed, r.rb
+ } else {
+ req.Coverage = recoveryCoverageClean
+ }
+ out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req)
+ if err != nil {
+ return false, fmt.Errorf("the recovery probe: %w", err)
+ }
+ switch out {
+ case recoveryProbeAnswered:
+ return r.phase(ctx, provision.RbSourceProbing, provision.RbSourceProbed, nil, "")
+ case recoveryProbeExhausted:
+ return r.stuck(ctx, provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts))
+ }
+ r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s has not answered yet", host, r.rb)
+ return false, nil
+}
+
+// probed (§7.8 source probed): the claim steps to probed and, for a dirty
+// return, queues the return lane's park in the same write; then recovered
+// with recovered_at, and the serial is nudged for the target's cleaning.
+func (r *recoverRun) probed(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id, rb := r.rec.Host, r.g.ID, r.rollback()
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ var lane func(*hostclaim.Claim) error
+ if rb.Dirty {
+ seal, err := returnSeal(obs.Claim)
+ if err != nil {
+ return false, err
+ }
+ scope := hostclaim.ParkScope{Location: moves.ScopeLocation(host, r.rb), Incarnation: obs.Claim.Holder.Incarnation,
+ EvidenceAt: seal.Bound().UTC(), RequestLimit: cutRequestLimit}
+ lane = hostclaim.QueueReturnLanePark(id, cfg.Region, scope, obligationID)
+ }
+ switch obs.Claim.Op.Phase {
+ case hostclaim.PhaseProbing:
+ if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil {
+ return false, err
+ }
+ step := hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed)
+ if _, err := hostclaim.Update(ctx, r.s, obs, func(c *hostclaim.Claim) error {
+ if err := step(c); err != nil {
+ return err
+ }
+ if lane != nil {
+ return lane(c)
+ }
+ return nil
+ }); err != nil {
+ return false, fmt.Errorf("the claim's probed: %w", err)
+ }
+ case hostclaim.PhaseProbed:
+ // A replay after the claim's write: the lane park is idempotent,
+ // and once the park engine completed it it is not queued again.
+ if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil {
+ return false, err
+ }
+ if lane != nil {
+ if _, err := hostclaim.Update(ctx, r.s, obs, lane); err != nil {
+ return false, fmt.Errorf("queue the return lane's park: %w", err)
+ }
+ }
+ default:
+ return false, r.executing(obs.Claim, hostclaim.PhaseProbing)
+ }
+ if err := r.fault("after-claim-probed"); err != nil {
+ return false, err
+ }
+ ok, err := r.phase(ctx, provision.RbSourceProbed, provision.RbSourceRecovered, func(rb *provision.MoveRollback) {
+ t := r.now().UTC()
+ rb.RecoveredAt = &t
+ }, "source-recovered")
+ if ok {
+ nudgeSerial(ctx, r.s)
+ }
+ return ok, err
+}
+
+// open (§7.8 source opened): the rollback's hold on the row ends, under the
+// host's hold lock with the record read fresh (J12); the claim steps to
+// opened; the record says opened with opened_at. A replay that finds the
+// row already open goes on.
+func (r *recoverRun) open(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id := r.rec.Host, r.g.ID
+ var err error
+ provision.WithHostHoldLock(host, func() {
+ if err = r.stillAt(ctx, provision.RbSourceRecovered); err != nil {
+ return
+ }
+ var released bool
+ if released, err = cfg.ReleaseHeld(host, r.rb); err != nil || released {
+ return
+ }
+ var rt *registry.Route
+ 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 rollback %s", id, rt.Status, rt.HoldOp, r.rb)
+ }
+ })
+ if err != nil {
+ return false, fmt.Errorf("open the row: %w", err)
+ }
+ if err := r.fault("after-row-opened"); err != nil {
+ return false, err
+ }
+ obs, err := r.claim(ctx)
+ if err != nil {
+ return false, err
+ }
+ if obs.Claim.Op.Phase != hostclaim.PhaseOpened {
+ if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil {
+ return false, err
+ }
+ if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened)); err != nil {
+ return false, fmt.Errorf("the claim's opened: %w", err)
+ }
+ }
+ if err := r.fault("after-claim-opened"); err != nil {
+ return false, err
+ }
+ return r.phase(ctx, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) {
+ t := r.now().UTC()
+ rb.OpenedAt = &t
+ }, "source-opened")
+}
+
+// finalize (§7.8 rolled_back): with the source opened and recovered and the
+// target parked, and no obligation of the op left in the claim, the
+// terminal event, the claim's release to serving and the record's close. A
+// claim already released by this move (a crash before the close) only
+// closes.
+func (r *recoverRun) finalize(ctx context.Context) (bool, error) {
+ cfg := r.cfg()
+ host, id, rb := r.rec.Host, r.g.ID, r.rollback()
+ if rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil {
+ r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for the target's park before it closes", host, r.rb)
+ return false, nil
+ }
+ 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)
+ }
+ if hostclaim.RequireReleasedMove(id)(obs.Claim) != nil {
+ if err := r.ours(obs.Claim); err != nil {
+ return false, fmt.Errorf("move %s: %w", id, err)
+ }
+ if err := r.executing(obs.Claim, hostclaim.PhaseOpened); err != nil {
+ return false, err
+ }
+ owed := 0
+ for _, o := range obs.Claim.Obligations {
+ if o.Op == id {
+ owed++
+ }
+ }
+ if owed > 0 {
+ r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for %d chore(s) of the move to be acknowledged before it closes", host, r.rb, owed)
+ return false, nil
+ }
+ r.w.logEvent(ctx, r.rec, "rolled_back")
+ if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.ReleaseMove(id, cfg.Box)); err != nil {
+ return false, fmt.Errorf("release the claim: %w", err)
+ }
+ log.Printf("moves: %s: the claim is serving again after rollback %s", host, r.rb)
+ if err := r.fault("after-release-move"); err != nil {
+ return false, err
+ }
+ }
+ if _, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRolledBackClose(r.g, r.now())); err != nil {
+ if errors.Is(err, provision.ErrMoveConflict) {
+ log.Printf("moves: %s: the close of rollback %s refused (%v); the record moved on", host, r.rb, err)
+ return false, nil
+ }
+ return false, fmt.Errorf("close the rolled-back move: %w", err)
+ }
+ log.Printf("moves: %s: move %s is rolled back and closed", host, id)
+ nudgeSerial(ctx, r.s)
+ return true, nil
+}
diff --git a/internal/host/movesrecover_test.go b/internal/host/movesrecover_test.go
new file mode 100644
index 000000000..fad6afa2c
--- /dev/null
+++ b/internal/host/movesrecover_test.go
@@ -0,0 +1,686 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "os"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/appconfig"
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/moves"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The source's recovery after a rollback (moves spec §7.8 source table; Plan
+// 4 Task 5). The fixture is Task 4's: the landing harness landed through
+// restored and stepped to probing, then rolled back and released by the
+// target's own pass (clean, or dirty with the target's rows changed, so its
+// return lane holds a snapshot of the stopped database). The source box
+// (ie-box) then runs its own passes over the same store and buckets: its
+// row is the cut's, held moving by the move id; its warm database's marker
+// names the pre-move generation; its hold hand-over, its open and the probe
+// (Task 9, BLOCKED-ON-CLAIM C1) are faked; everything else is the real
+// recovery, the shared installer (Task 3) and provision's and hostclaim's own
+// transitions.
+
+type recoverHarness struct {
+ *rollbackHarness
+ sdir string // the source's data directory
+ lockDir string
+ srcRoute registry.Route
+ sw *movesWorker
+ srcRender appconfig.HeldRender
+ srcRends int // the installer's held renders on the source
+ srcAsked []string // the releases the source's gate was asked about
+ srcHanded []string // status:to:from of each hand-over the source asked for
+ sEvents []string
+ // probe answers each recovery probe; nil records the render's inputs
+ // and answers.
+ probe func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
+ probes []recoveryProbeRequest
+ rcAt string
+ rcDone bool
+ rcHits []string
+ rcHook func(point string)
+}
+
+// newRecoverHarness builds the fixture up to the rollback decision's
+// release: dirty makes the target's state differ from what it installed, so
+// its release is dirty and its return a snapshot. The caller rolls back and
+// runs the target's pass.
+func newRecoverHarness(t *testing.T, dirty bool) *recoverHarness {
+ t.Helper()
+ h := &recoverHarness{rollbackHarness: newRollbackHarness(t)}
+ if dirty {
+ h.targetChanged(t)
+ }
+ h.sdir = h.cfg.DataDir
+ h.lockDir = t.TempDir()
+ for _, d := range []string{"hostguard", ".parked"} {
+ if err := os.MkdirAll(filepath.Join(h.sdir, d), 0o770); err != nil {
+ t.Fatal(err)
+ }
+ }
+ // The source's warm database (the cut's stopped database) carries the
+ // marker of its pre-move generation, src_gen.
+ db := provision.DefaultDBPath(h.sdir, p1Host)
+ if err := os.WriteFile(db+".gen", []byte("cur\n"), 0o644); err != nil {
+ t.Fatal(err)
+ }
+ h.srcRoute = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "moving", HoldOp: p1ID,
+ Socket: provision.DefaultSocket(p1Host), DBPath: db, Provisioned: true}
+ // The current bundle is not the pinned one: a recovery renders and
+ // records what is current, never compared with move.inputs.
+ h.srcRender = appconfig.HeldRender{Path: filepath.Join(h.sdir, p1Host+".json"), SHA256: "y", BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9"}
+ return h
+}
+
+// released rolls the landing back and runs the target's pass: released
+// (clean) or returned (dirty).
+func (h *recoverHarness) released(t *testing.T) *provision.MoveRollback {
+ t.Helper()
+ h.rollBack(t)
+ h.rollbackPass(t)
+ rb := h.rb(t)
+ if want := map[bool]string{false: provision.RbTargetReleased, true: provision.RbTargetReturned}[rb.Dirty]; rb.TargetPhase != want {
+ t.Fatalf("the target did not release: %+v", rb)
+ }
+ return rb
+}
+
+// sourceCfg is the source box's configuration with the recovery armed.
+func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg {
+ t.Helper()
+ c := h.landingCfg(t)
+ c.Region, c.Box, c.OwnBucket, c.OwnBucketRegion = "ie", "ie-box", "data-ie", "eu-west-1"
+ c.DataDir = h.sdir
+ c.OptDir = h.optDir
+ c.OwnsRoute = func(host string) bool { return host == p1Host }
+ c.Route = func(host string) (*registry.Route, error) {
+ if host != p1Host {
+ return nil, nil
+ }
+ rt := h.srcRoute
+ return &rt, nil
+ }
+ c.FilesDir = func(host string) (string, string) {
+ return filepath.Join(h.sdir, host), filepath.Join(h.sdir, host+".env")
+ }
+ c.Guard = hostguard.Guard{DataDir: h.sdir, LockDir: h.lockDir}
+ c.Local = park.Local{DataDir: h.sdir, Box: "ie-box"}
+ c.RenderHeld = func(_ context.Context, _ release.Store, rt registry.Route, op string) (appconfig.HeldRender, error) {
+ h.srcRends++
+ if rb := h.read(t, p1Host).Move.Rollback; rb == nil || op != rb.ID || rt.HoldOp != rb.ID {
+ t.Errorf("rendered for %q over a row held by %q, want the rollback", op, rt.HoldOp)
+ }
+ return h.srcRender, nil
+ }
+ c.ReleaseReady = func(_ context.Context, _ release.Store, _ provision.Record, v string) moves.Check {
+ h.srcAsked = append(h.srcAsked, v)
+ return moves.Check{OK: true}
+ }
+ c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) {
+ h.srcHanded = append(h.srcHanded, status+":"+to+":"+from)
+ if validate != nil {
+ if err := validate(); err != nil {
+ return false, err
+ }
+ }
+ if h.srcRoute.HoldOp == to && h.srcRoute.Status == status {
+ return true, nil
+ }
+ if h.srcRoute.HoldOp != from {
+ return false, nil
+ }
+ h.srcRoute.Status, h.srcRoute.HoldOp = status, to
+ return true, nil
+ }
+ c.ReleaseHeld = func(host, op string) (bool, error) {
+ if op == "" || h.srcRoute.HoldOp != op {
+ return false, nil
+ }
+ h.srcRoute.Status, h.srcRoute.HoldOp = "ready", ""
+ return true, nil
+ }
+ c.LogEvent = func(_ context.Context, _ provision.Record, event string) { h.sEvents = append(h.sEvents, event) }
+ c.recoveryProbe = func(ctx context.Context, _ release.Store, rec 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, h.renderedInputs()); err != nil {
+ return recoveryProbeWaiting, err
+ }
+ return recoveryProbeAnswered, nil
+ }
+ c.rcFault = func(point string) error {
+ h.rcHits = append(h.rcHits, point)
+ if h.rcHook != nil {
+ h.rcHook(point)
+ }
+ if point == h.rcAt && !h.rcDone {
+ h.rcDone = true
+ return errCrash
+ }
+ return nil
+ }
+ return c
+}
+
+// renderedInputs are what the source renders now: the render's etags and
+// the live release.
+func (h *recoverHarness) renderedInputs() provision.MoveInputs {
+ b := map[string]string{}
+ for k, v := range h.srcRender.BundleETags {
+ b[k] = v
+ }
+ return provision.MoveInputs{BundleETags: b, OverlayETag: h.srcRender.OverlayETag, Release: "v7"}
+}
+
+// srcPass runs one pass as the source box, its worker kept across passes.
+func (h *recoverHarness) srcPass(t *testing.T) {
+ t.Helper()
+ if h.sw == nil {
+ h.sw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
+ }
+ h.sw.cfg = h.sourceCfg(t)
+ if err := movesPass(context.Background(), h.s, h.sw); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (h *recoverHarness) sourceAt(t *testing.T, phase string) *provision.MoveRollback {
+ t.Helper()
+ m := h.read(t, p1Host).Move
+ if m == nil || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.SourcePhase != phase {
+ if m != nil {
+ t.Fatalf("the source is not at %s: %+v %+v", phase, m, m.Rollback)
+ }
+ t.Fatalf("the source is not at %s: no move", phase)
+ }
+ return m.Rollback
+}
+
+// opened asserts the end of a recovery: the record opened with RecoveredAt
+// and OpenedAt, the claim restoring/opened with the source executing, the
+// row open.
+func (h *recoverHarness) opened(t *testing.T) *provision.MoveRollback {
+ t.Helper()
+ rb := h.sourceAt(t, provision.RbSourceOpened)
+ if rb.RecoveredAt == nil || rb.OpenedAt == nil {
+ t.Fatalf("opened without its stamps: %+v", rb)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != "ie-box" || c.Op.Intent["rollback"] != rb.ID {
+ t.Fatalf("claim: %+v", c.Op)
+ }
+ if h.srcRoute.Status != "ready" || h.srcRoute.HoldOp != "" {
+ t.Fatalf("the row: %+v", h.srcRoute)
+ }
+ return rb
+}
+
+// TestSourceWaitsForTargetEvidence: the source's recovery starts only on the
+// target's evidence (§7.8 waiting): the record released clean or returned
+// dirty, and the claim's execution holder none. Until then nothing is
+// claimed, held or probed.
+func TestSourceWaitsForTargetEvidence(t *testing.T) {
+ untouched := func(t *testing.T, h *recoverHarness) {
+ t.Helper()
+ h.sourceAt(t, provision.RbSourceWaiting)
+ if h.srcRoute.Status != "moving" || h.srcRoute.HoldOp != p1ID || len(h.srcHanded) != 0 || len(h.probes) != 0 {
+ t.Fatalf("the source acted: row %+v handed %v probes %d", h.srcRoute, h.srcHanded, len(h.probes))
+ }
+ if c := h.claim(t); c.Op.Exec == "ie-box" || c.Op.Phase == hostclaim.PhaseClaimed {
+ t.Fatalf("the source claimed: %+v", c.Op)
+ }
+ }
+ t.Run("the target is releasing", func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.rollBack(t)
+ h.srcPass(t)
+ untouched(t, h)
+ })
+ t.Run("released clean on the record, the target still executing", func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.rollBack(t)
+ h.rbAt = "after-record-released"
+ h.rollbackPass(t)
+ if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased || rb.Dirty {
+ t.Fatalf("fixture: %+v", rb)
+ }
+ if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Exec != "uk-box" {
+ t.Fatalf("fixture claim: %+v", c.Op)
+ }
+ h.srcPass(t)
+ untouched(t, h)
+ // The target's next pass writes the claim half; the source then
+ // recovers.
+ h.rollbackPass(t)
+ h.srcPass(t)
+ h.opened(t)
+ })
+ t.Run("dirty, released but not returned", func(t *testing.T) {
+ h := newRecoverHarness(t, true)
+ h.rollBack(t)
+ h.rbAt = "after-record-returning"
+ h.rollbackPass(t)
+ if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturning || !rb.Dirty {
+ t.Fatalf("fixture: %+v", rb)
+ }
+ if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleased || c.Op.Exec != "" {
+ t.Fatalf("fixture claim: %+v", c.Op)
+ }
+ h.srcPass(t)
+ untouched(t, h)
+ })
+ t.Run("unarmed in production", func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.released(t)
+ // h.cfg is the source box as production wires it: no park
+ // engine leases (ruling G17), so the recovery waits for Plan 5.
+ h.pass(t)
+ untouched(t, h)
+ })
+}
+
+// TestCleanReturnVerifiesAndOpens: a clean return claims the host back with
+// the source's own generation named again (holder.generation = src_gen), its
+// row handed from the move to the rollback as restoring, its files, timers
+// and warm database marker verified against what it recorded when it
+// stopped, then the probe (faked until Task 9), probed, recovered and the
+// open. No install phase runs.
+func TestCleanReturnVerifiesAndOpens(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ pinned := h.read(t, p1Host).Move.Inputs
+ rb := h.released(t)
+ srcGen := h.claim(t).Op.Journal[hostclaim.JournalSrcGen]
+ // The landing's probe admission named the landing's generation; the
+ // return must name the source's again.
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = p1ID })
+ var atClaimed *hostclaim.Holder
+ h.rcHook = func(point string) {
+ if point == "after-activate" {
+ c := h.claim(t)
+ hc := *c.Holder
+ atClaimed = &hc
+ }
+ }
+ h.srcPass(t)
+ got := h.opened(t)
+ if atClaimed == nil || atClaimed.Generation != srcGen || srcGen != "cur" || atClaimed.Stamp != 1 {
+ t.Fatalf("the holder after claimed: %+v (src_gen %q)", atClaimed, srcGen)
+ }
+ if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
+ t.Fatalf("the source's installation record: %+v %v", captured, err)
+ }
+ if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
+ t.Fatalf("hand-overs: %v", h.srcHanded)
+ }
+ // No install phase ran: no render, no return journal, no generation.
+ if h.srcRends != 0 {
+ t.Fatalf("a clean return rendered through the installer %d times", h.srcRends)
+ }
+ if _, err := os.Lstat(moves.JournalDir(h.sdir, p1Host, rb.ID)); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("a clean return wrote an install journal: %v", err)
+ }
+ for k := range h.f.buckets["data-ie"] {
+ if strings.HasPrefix(k, moves.GenerationPrefix(cutRoot, rb.ID)) {
+ t.Fatalf("a clean return built a generation: %s", k)
+ }
+ }
+ for _, p := range h.rcHits {
+ if strings.HasPrefix(p, "after-record-") {
+ switch strings.TrimPrefix(p, "after-record-") {
+ case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored:
+ t.Fatalf("a clean return reached %s", p)
+ }
+ }
+ }
+ // The probe (Task 9's): clean, coverage clean-return, nine attempts; its
+ // inputs recorded, never compared with move.inputs.
+ if len(h.probes) != 1 || !h.probes[0].Clean || h.probes[0].Coverage != "clean-return" || h.probes[0].Attempts != 9 || h.probes[0].RollbackID != rb.ID ||
+ h.probes[0].SourceGeneration != "cur" {
+ t.Fatalf("probes: %+v", h.probes)
+ }
+ m := h.read(t, p1Host).Move
+ if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e9" || in.OverlayETag != "o9" || in.Release != "v7" {
+ t.Fatalf("recovery inputs: %+v", got.RecoveryInputs)
+ }
+ if m.Inputs == nil || pinned == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
+ t.Fatalf("move.inputs changed: %+v was %+v", m.Inputs, pinned)
+ }
+ // A clean return queues no park the source owes: no return lane, no
+ // park-local of its own.
+ for _, o := range h.claim(t).Obligations {
+ if o.Region == "ie" || o.Box == "ie-box" {
+ t.Fatalf("a clean return queued %+v", o)
+ }
+ }
+ if !strings.Contains(strings.Join(h.sEvents, " "), "source-recovered source-opened") {
+ t.Fatalf("events %v", h.sEvents)
+ }
+ // The record waits for the target's cleaning: a second pass changes
+ // nothing.
+ h.srcPass(t)
+ h.opened(t)
+}
+
+// TestDirtyReturnInstallsFromTheLane: a dirty return installs the target's
+// snapshot from the return lane through the shared installer: the return
+// generation g-<rb>/ (snapshot and anchor) in the source's own bucket, the
+// database restored to what the target wrote, .gen naming the rollback;
+// recovery inputs recorded at configured and again by the probe, never
+// move.inputs (witness wRecoveryInputs: a bundle published after the
+// configured render does not block the probe); then probed, which queues
+// the return lane's park in the source's region, recovered and opened.
+func TestDirtyReturnInstallsFromTheLane(t *testing.T) {
+ for _, registered := range []bool{false, true} {
+ name := "unregistered"
+ if registered {
+ name = "registered"
+ }
+ t.Run(name, func(t *testing.T) {
+ h := newRecoverHarness(t, true)
+ pinned := h.read(t, p1Host).Move.Inputs
+ rb := h.released(t)
+ c := h.claim(t)
+ returnSeal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
+ if err != nil {
+ t.Fatal(err)
+ }
+ if registered {
+ rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
+ c.Inventory = map[string]hostclaim.InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
+ })
+ }
+ var queuedLocal bool
+ var atConfigured *provision.MoveInputs
+ h.landFault = func(point string) error {
+ if point == "after-record-"+provision.RbSourceRestored {
+ atConfigured = h.read(t, p1Host).Move.Rollback.RecoveryInputs
+ // A bundle published after the configured render.
+ h.srcRender.BundleETags = map[string]string{"default": "e10", "environment": ""}
+ }
+ return nil
+ }
+ h.rcHook = func(point string) {
+ if point == "after-record-"+provision.RbSourceRestoring {
+ for _, o := range h.claim(t).Obligations {
+ if o.Kind == "park-local" && o.Box == "ie-box" {
+ queuedLocal = true
+ }
+ }
+ }
+ }
+ h.srcPass(t)
+ got := h.opened(t)
+ if queuedLocal != registered {
+ t.Fatalf("the source's park-local queued at claimed: %v, registered %v", queuedLocal, registered)
+ }
+ if atConfigured == nil || atConfigured.BundleETags["default"] != "e9" || atConfigured.Release != "v7" {
+ t.Fatalf("configured recorded %+v", atConfigured)
+ }
+ if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e10" || in.Release != "v7" {
+ t.Fatalf("the probe's recovery inputs: %+v", got.RecoveryInputs)
+ }
+ m := h.read(t, p1Host).Move
+ if m.Inputs == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
+ t.Fatalf("move.inputs changed: %+v", m.Inputs)
+ }
+ // The return generation: a snapshot and an anchor at N'.
+ prefix := moves.GenerationPrefix(cutRoot, rb.ID)
+ for _, k := range []string{moves.SnapshotKey(prefix, returnSeal.N), moves.AnchorKey(prefix, returnSeal.N)} {
+ if _, ok := h.f.buckets["data-ie"][k]; !ok {
+ t.Fatalf("the return generation has no %s", k)
+ }
+ }
+ dbPath := provision.DefaultDBPath(h.sdir, p1Host)
+ if n := countRowsForTest(t, dbPath); n != h.rows+1 {
+ t.Fatalf("the returned database holds %d rows, the target wrote %d", n, h.rows+1)
+ }
+ if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != rb.ID {
+ t.Fatalf(".gen %q %v", gen, err)
+ }
+ // The probe: dirty, into the return generation, coverage sealed.
+ if len(h.probes) != 1 || h.probes[0].Clean || h.probes[0].Coverage != "sealed" || h.probes[0].Generation != rb.ID || h.probes[0].Attempts != 9 {
+ t.Fatalf("probes: %+v", h.probes)
+ }
+ // probed queued the return lane's park, in the source's region,
+ // its horizon the return seal's bound.
+ c = h.claim(t)
+ lane := 0
+ for _, o := range c.Obligations {
+ if o.Kind == "park-replica" && o.Region == "ie" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, rb.ID) {
+ lane++
+ if !o.Park.EvidenceAt.Equal(returnSeal.Bound()) || o.Park.RequestLimit != cutRequestLimit || o.Park.Incarnation != h.inc {
+ t.Fatalf("the lane park's scope: %+v", o.Park)
+ }
+ }
+ if o.Kind == "park-local" && o.Box == "ie-box" {
+ t.Fatalf("the source's park-local is still owed: %+v", o)
+ }
+ }
+ if lane != 1 {
+ t.Fatalf("return lane parks: %d (%+v)", lane, c.Obligations)
+ }
+ if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
+ t.Fatalf("hand-overs: %v", h.srcHanded)
+ }
+ if held := h.leases.holding(); len(held) != 0 {
+ t.Fatalf("the source still holds %v", held)
+ }
+ })
+ }
+}
+
+// TestVerifyMismatchIsStuck (INV-4): a clean return whose files, timers or
+// warm database marker no longer match what the source recorded when it
+// stopped stops for an operator with the baseline-mismatch sentence, at
+// verified, before any probe; the row stays held. A retry verifies again,
+// never skips it: unchanged, it stops again; repaired, it recovers.
+func TestVerifyMismatchIsStuck(t *testing.T) {
+ for _, c := range []struct {
+ name string
+ breakIt, mend func(h *recoverHarness)
+ }{
+ {"files", func(h *recoverHarness) {
+ os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("changed on the source"), 0o644)
+ }, func(h *recoverHarness) {
+ os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("hello"), 0o644)
+ }},
+ {"timers", func(h *recoverHarness) {
+ h.timers = append(append([]registry.Timer(nil), h.timers...), registry.Timer{Host: p1Host, Name: "extra", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
+ }, func(h *recoverHarness) {
+ h.timers = h.timers[:len(h.timers)-1]
+ }},
+ {"generation marker", func(h *recoverHarness) {
+ os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("other\n"), 0o644)
+ }, func(h *recoverHarness) {
+ os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("cur\n"), 0o644)
+ }},
+ } {
+ t.Run(c.name, func(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ rb := h.released(t)
+ c.breakIt(h)
+ stuck := func() {
+ t.Helper()
+ m := h.read(t, p1Host).Move
+ if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified {
+ t.Fatalf("not stuck at verified: %+v %+v", m, m.Rollback)
+ }
+ if len(h.probes) != 0 {
+ t.Fatalf("probed past a mismatch: %+v", h.probes)
+ }
+ if h.srcRoute.Status != "restoring" || h.srcRoute.HoldOp != rb.ID {
+ t.Fatalf("the row: %+v", h.srcRoute)
+ }
+ if cl := h.claim(t); cl.Op.Phase != hostclaim.PhaseClaimed || cl.Op.Exec != "ie-box" {
+ t.Fatalf("claim: %+v", cl.Op)
+ }
+ }
+ h.srcPass(t)
+ stuck()
+ h.srcPass(t)
+ stuck()
+ // A retry verifies again: still broken, stuck again.
+ h.retry(t)
+ h.srcPass(t)
+ stuck()
+ c.mend(h)
+ h.retry(t)
+ h.srcPass(t)
+ h.opened(t)
+ if !strings.Contains(strings.Join(h.sEvents, " "), "source-stuck") {
+ t.Fatalf("events %v", h.sEvents)
+ }
+ })
+ }
+}
+
+// TestSourceRecoveryNeverReverifiesAfterProbing (Review Focus 2; INV-2,
+// INV-7): once the record says probing, the recovery's own writes are
+// authoritative: a probe that has not answered yet, or a restart, runs only
+// the probe again, never the verification or the install, whatever the
+// files say now. Nine failed attempts stop it for an operator at probing,
+// and a retry probes again.
+func TestSourceRecoveryNeverReverifiesAfterProbing(t *testing.T) {
+ h := newRecoverHarness(t, false)
+ h.released(t)
+ waiting := true
+ h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
+ if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
+ return recoveryProbeWaiting, err
+ }
+ if waiting {
+ return recoveryProbeWaiting, nil
+ }
+ return recoveryProbeAnswered, nil
+ }
+ h.srcPass(t)
+ h.sourceAt(t, provision.RbSourceProbing)
+ if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != "ie-box" {
+ t.Fatalf("the claim is not probing before the attempt: %+v", c.Op)
+ }
+ // The source's files and marker change under the probe (its app wrote).
+ os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("written by the recovered app"), 0o644)
+ os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("fresh\n"), 0o644)
+ h.sw = nil // a restarted agent
+ h.srcPass(t)
+ h.sourceAt(t, provision.RbSourceProbing)
+ // Nine attempts used: stuck at probing; the retry probes again.
+ h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil }
+ h.srcPass(t)
+ m := h.read(t, p1Host).Move
+ if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Rollback.StuckAt != provision.RbSourceProbing {
+ t.Fatalf("not stuck at probing: %+v %+v", m, m.Rollback)
+ }
+ h.retry(t)
+ waiting = false
+ h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
+ if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
+ return recoveryProbeWaiting, err
+ }
+ return recoveryProbeAnswered, nil
+ }
+ h.srcPass(t)
+ h.opened(t)
+ if n := len(h.probes); n != 4 {
+ t.Fatalf("probe attempts: %d", n)
+ }
+ verified := 0
+ for _, p := range h.rcHits {
+ if p == "verify" {
+ verified++
+ }
+ }
+ if verified != 1 {
+ t.Fatalf("verified %d times (%v)", verified, h.rcHits)
+ }
+}
+
+// TestRolledBackFinalizationReleasesTheClaim (§7.8 rolled_back): once the
+// source has opened and the target has parked, and no obligation of the op
+// remains, the source writes the terminal event, releases the claim to
+// serving and closes the record with the rolled_back outcome. Until every
+// chore is acknowledged it waits; a crash between the claim's release and
+// the record's close is finished by the next pass.
+func TestRolledBackFinalizationReleasesTheClaim(t *testing.T) {
+ ctx := context.Background()
+ h := newRecoverHarness(t, false)
+ h.released(t)
+ h.srcPass(t)
+ h.opened(t)
+ // The target cleans now that the source has recovered.
+ h.rollbackPass(t)
+ if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
+ t.Fatalf("the target did not clean: %+v", rb)
+ }
+ // The parks are acknowledged by the regional runner (C4), not here:
+ // the source waits for them.
+ h.srcPass(t)
+ h.opened(t)
+ owed := 0
+ for _, o := range h.claim(t).Obligations {
+ if o.Op == p1ID {
+ owed++
+ }
+ }
+ if owed == 0 {
+ t.Fatal("the fixture owes no chore; the wait is not exercised")
+ }
+ if strings.Contains(strings.Join(h.sEvents, " "), "rolled_back") {
+ t.Fatalf("finalized with chores owed: %v", h.sEvents)
+ }
+ for _, o := range h.claim(t).Obligations {
+ if o.Op != p1ID {
+ continue
+ }
+ obs, err := hostclaim.Read(ctx, h.s, p1Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
+ t.Fatal(err)
+ }
+ }
+ h.rcAt = "after-release-move"
+ h.srcPass(t)
+ if !h.rcDone {
+ t.Fatalf("never reached the claim's release (%v)", h.rcHits)
+ }
+ c := h.claim(t)
+ if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
+ t.Fatalf("the claim was not released: %+v", c.Op)
+ }
+ if err := hostclaim.RequireReleasedMove(p1ID)(c); err != nil {
+ t.Fatal(err)
+ }
+ h.sourceAt(t, provision.RbSourceOpened)
+ h.srcPass(t)
+ rec := h.read(t, p1Host)
+ if rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" || rec.Region != "ie" {
+ t.Fatalf("not closed: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
+ }
+ if n := strings.Count(strings.Join(h.sEvents, " "), "rolled_back"); n != 1 {
+ t.Fatalf("the terminal event %d times: %v", n, h.sEvents)
+ }
+ if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
+ t.Fatalf("claim after the close: %+v", c.Op)
+ }
+ // Nothing routes to the source any more.
+ h.srcPass(t)
+ if rec := h.read(t, p1Host); rec.Move != nil {
+ t.Fatalf("record %+v", rec.Move)
+ }
+}