package host

import (
	"context"
	"crypto/rand"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"time"

	"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/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// The rollback on the target (moves spec §7.8; Plan 4 Task 4), behind
// CARLOS_MOVES_PHASE1: the decision that ends a landing, then the target
// box's half of the rollback: its release, its return lane and its cleaning.
//
// The decision (rollBackLanding). A landing the target can no longer finish
// is rolled back by one record CAS, MoveRollBack, which moves the region back
// to From; the claim follows (RollBack, or RollBackFlipped over a claim the
// flip's half never reached, J16), journalling the target installation's
// stamp F (J15). It runs at the top of the landing's pass, before its arming
// gate (J10), and is the one place a landing's recorded mark (every
// markRollback site, Plans 2 and 3) becomes a rollback. The deadline that
// triggers it is landingDeadlineOf, the earlier of the cut's and the
// landing's, at every landing phase, probing and probed included (J11). The
// commit races it on the record: a commit that won makes MoveRollBack refuse
// with ErrMoveConflict, and the decision stops there with no trace, leaving
// the record to the commit and the open (Plan 3 Task 7).
//
// The target's half (rollbackTarget). Once the record is rolling_back its
// region is the source's, so the target reaches it through the post-flip
// participant rule (movesParticipantRecord, J10). One pass drives it as far
// as it can, each phase from durable state, under the record's
// target_phase (§7.8 target table; Plan 2's phase-name convention, G8: a
// record at X means X's work may have begun):
//
//   - releasing: the row's hold handed from the move to the rollback
//     (handHold, J3), the wake fence, the timer drain and the stop evidence
//     (stopevidence.Reconstruct); a target not yet proven stopped retries,
//     and the release deadline is the console's to page on. Then, if the
//     landing reached probing and the database is here, a final checkpoint
//     under the writer's lease (finalCheckpoint, writerLease), N' from the
//     local position, and the digests: the database, the files plain digest
//     and the timers. The release is clean when the landing never reached
//     probing or all three equal what was installed (installed_db_sha256,
//     source.files.plain, source.timers_sha256), and the staged seal covers
//     the source (src_covered). stopped-target.json is written create-only,
//     the record goes to released with Dirty and Evidence, and the claim
//     follows (TargetReleased: execution to none, the target region's
//     location park, its park-local only when the inventory lists F, the
//     target region's predecessors covered);
//   - returning (dirty only): the return lane, moves/<host>/<rb id>/ in the
//     source's bucket, written as a staging lane under the staging-work lease
//     for (source region, host), every object through the guarded staging
//     writer, one presigned request per call. A target that served and
//     changed sends a snapshot [1, N'] of its stopped database
//     (moves.EncodeFull, held in memory: Phase 1's size limit), its files
//     archive (bound to the move id, which the source's installer checks)
//     and its timers; one whose state is still the staged seal's (dirty only
//     for want of src_covered) sends the staged seal's versions as given.
//     The return seal (moves.ReturnSeal, bound = the lease's SafeUntil after
//     the last write) and the return baseline go into the claim's journal in
//     one write, then the record says returned;
//   - cleaning, only once the source has recovered (RecoveredAt, Review
//     Focus 4): the target region's location park and the move's lane park
//     (queued here) run through the park engine until their archives are
//     complete (D6: this region is no longer authoritative, so the engine
//     selects everything); the registered installation is retired through
//     park.Local, which completes its own park-local (J15), and everything
//     else at the flat paths is parked by the moves-owned rename (H5(b))
//     inside the host guard; the row and its timers rows go in one registry
//     transaction under the host's hold lock (J12); the record says cleaned
//     with target_parked_at, and the op's journal is parked last.
//
// Arming (ruling J18; Plan 5 Task 9a). Like the landing, the rollback's
// target half needs the claim's park leases, view and authorizer, which
// armMovesParks sets only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b), so rollbackTargetArmed
// answers false in production and the pass logs a standing notice. The decision is
// not gated (J10). The move's event log (moves.LogEvent, Plan 6) is the
// LogEvent hook: nil in production, where the event is a log line. The
// stuck mark pages an operator (pageStuck, the Page hook); the release
// deadline is the console's.

// rollbackStepsPerPass bounds the phases one pass runs.
const rollbackStepsPerPass = 8

// The target's rollback steps in its op journal (the landing's journal of
// the move id, moves.Journal). Each step is skipped on a replay once done.
const (
	rbStepTargetStamp  = "rb-target-stamp"             // F, noted from the claim before anything is retired
	rbStepLaneQueued   = "rb-lane-queued"              // the move's lane park is queued in this region
	rbStepRemote       = "rb-target-park-remote"       // both parks' archives are complete, or acknowledged
	rbStepInstallation = "rb-target-park-installation" // retired, completed or unregistered
	rbStepLocal        = "rb-target-park-local"        // parkLocalArtifacts' step: the retirement id
	rbStepRowDeleted   = "rb-target-row-deleted"       // "pending" before the delete, "done" after it
	rbStepCleaned      = "rb-target-cleaned"           // "done": the record says cleaned, only the journal's park is left
)

// stoppedTargetRole is the target's StoppedReport role.
const stoppedTargetRole = "target"

// now is the worker's clock.
func (w *movesWorker) now() time.Time {
	if n := w.cfg.Now; n != nil {
		return n()
	}
	return time.Now()
}

// stuckPage is the stuck mark's page (pageStuck, movespaging.go): the subject (MV-N313) and the body
// (MV-N314).
func stuckPage(host, id, box, role, reason string) (subject, body string) {
	subject = fmt.Sprintf("carlos move stuck: %s", host)
	body = fmt.Sprintf("The move %s of %s stopped for an operator on %s, its %s: %s. Run carlos move resolve --host %s --retry once the cause is fixed.",
		id, host, box, role, reason, host)
	return subject, body
}

// rollBackLanding is the rollback decision on the target (§7.8; J10, J11):
// a landing marked for a rollback, or past landingDeadlineOf at any phase,
// is rolled back by MoveRollBack, and the claim follows. done is true when
// the record was this decision's to take (written, or refused because the
// record moved on); the landing's pass then ends.
func (w *movesWorker) rollBackLanding(ctx context.Context, s release.Store, rec provision.Record) (done bool, err error) {
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveLanding {
		return false, nil
	}
	reason := m.Reason
	if reason == "" {
		d, which := landingDeadlineOf(m)
		if d == nil || !w.now().After(*d) {
			return false, nil
		}
		log.Printf("moves: %s: landing %s is past its %s (%s)", rec.Host, m.ID, which, d.UTC().Format(time.RFC3339))
		reason = provision.RollbackLandingDeadline
	}
	rbID, err := moves.NewRollbackID(rand.Reader)
	if err != nil {
		return true, err
	}
	g := provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}
	var pre provision.Record
	fresh, err := provision.SetRecordMove(ctx, s, rec.Account, rec.App, rec.Host, observedMove(&pre, provision.MoveRollBack(g, rbID, reason, w.now())))
	if errors.Is(err, provision.ErrMoveConflict) {
		// A commit that won (the open is the commit's path), or another
		// writer's rollback (the participant pass follows it).
		log.Printf("moves: %s: the rollback of landing %s is refused (%v); the record moved on", rec.Host, m.ID, err)
		return true, nil
	}
	if err != nil {
		return true, fmt.Errorf("roll the landing back: %w", err)
	}
	log.Printf("moves: %s: landing %s rolls back as %s: %s", rec.Host, m.ID, rbID, fresh.Move.Rollback.Reason)
	w.logTransition(ctx, s, pre, *fresh)
	if _, err := w.rollbackClaim(ctx, s, *fresh); err != nil {
		log.Printf("moves: %s: the claim half of rollback %s waits for the next pass: %v", rec.Host, rbID, err)
	}
	nudgeSerial(ctx, s)
	return true, nil
}

// rollbackClaim is the claim half of the record's rollback: the op of this
// move, for the record's incarnation, goes to restoring/releasing under the
// rollback id (RollBackFlipped writes the flip first over a claim still at
// moving/sealed). A claim already restoring for this rollback is done.
func (w *movesWorker) rollbackClaim(ctx context.Context, s release.Store, rec provision.Record) (hostclaim.Observed, error) {
	m := rec.Move
	obs, err := hostclaim.Read(ctx, s, 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", m.ID)
	}
	if err := hostclaim.RequireOp(m.ID)(obs.Claim); err != nil {
		return obs, fmt.Errorf("move %s: %w", m.ID, err)
	}
	h := obs.Claim.Holder
	if h == nil || h.Incarnation != m.Incarnation {
		return obs, fmt.Errorf("move %s: the host claim names another incarnation", m.ID)
	}
	rb := m.Rollback.ID
	if op := obs.Claim.Op; op.Kind == hostclaim.KindRestoring && op.Intent["rollback"] == rb {
		return obs, nil
	}
	if obs, err = hostclaim.Update(ctx, s, obs, hostclaim.RollBackFlipped(m.ID, rb, w.cfg.Box, h.Stamp, h.Generation)); err != nil {
		return obs, fmt.Errorf("the claim half of the rollback: %w", err)
	}
	log.Printf("moves: %s: the claim follows rollback %s", rec.Host, rb)
	return obs, nil
}

// rollbackTargetArmed reports whether this box can run the target's half of
// a rollback: every hook it calls is wired, the claim's three park hooks and
// its local authorizer (M9: the installation's park.Local) included. In
// production armMovesParks sets them only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b), so this answers false there
// (J18).
func rollbackTargetArmed(cfg moveCfg) bool {
	return cfg.Box != "" && cfg.DataDir != "" && cfg.Guard.DataDir != "" && cfg.Local.DataDir != "" && parkHooksArmed(cfg) && cfg.ParkLocalAuthorize != nil &&
		cfg.HandHold != nil && cfg.Fence != nil && cfg.TimersDrained != nil && cfg.Evidence != nil && cfg.FinalCheckpoint != nil &&
		cfg.LocalPosition != nil && cfg.Timers != nil && cfg.FilesDir != nil && cfg.Route != nil && cfg.Versioned != nil &&
		cfg.StagingLease != nil && cfg.Staging != nil && cfg.DeleteHeld != nil && cfg.SecretsPub != nil
}

// rollbackRun is one pass's state for one rolling_back record on its target.
type rollbackRun struct {
	w   *movesWorker
	s   release.Store
	rec provision.Record
	g   provision.MoveGuard
	rb  string
}

// rollbackTarget is the participant step for a rolling_back record on its
// target (movesPass, J10).
func (w *movesWorker) rollbackTarget(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.To != cfg.Region || m.From == cfg.Region {
		return nil
	}
	// The landing is abandoned: a location lease its build still holds on
	// this worker (kept across passes from building through published) is
	// released on entry, whoever decided the rollback, so the cleaning's
	// park engine can take it (the final review's finding 2). Idempotent:
	// nothing held is nothing to release.
	releaseOwnLease(w, rec.Host)
	if !rollbackTargetArmed(cfg) {
		w.notice(rec.Host+"#rollback", "moves: %s: the target's half of rollback %s waits: this box's rollback is not armed (the claim's park hooks are set from MoveParks only once this box's moves wake source is registered, Plan 5 Task 9b)", rec.Host, m.Rollback.ID)
		return nil
	}
	r := &rollbackRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID}
	for i := 0; i < rollbackStepsPerPass; 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 fm := fresh.Move; fm == nil || fm.ID != r.g.ID || fm.Incarnation != r.g.Incarnation || fm.State != provision.MoveRollingBack || fm.Rollback == nil || fm.Rollback.ID != r.rb {
			return nil
		}
		r.rec = *fresh
	}
	return nil
}

func (r *rollbackRun) cfg() moveCfg                      { return r.w.cfg }
func (r *rollbackRun) now() time.Time                    { return r.w.now() }
func (r *rollbackRun) rollback() *provision.MoveRollback { return r.rec.Move.Rollback }
func (r *rollbackRun) dbPath() string                    { return provision.DefaultDBPath(r.cfg().DataDir, r.rec.Host) }

// fault is the test seam after a side effect.
func (r *rollbackRun) fault(point string) error {
	if f := r.cfg().rbFault; f != nil {
		return f(point)
	}
	return nil
}

// step runs the target's current rollback phase once. advanced is true when
// it wrote the record.
func (r *rollbackRun) step(ctx context.Context) (bool, error) {
	switch r.rollback().TargetPhase {
	case provision.RbTargetReleasing:
		return r.releasing(ctx)
	case provision.RbTargetReleased:
		return r.released(ctx)
	case provision.RbTargetReturning:
		return r.returning(ctx)
	case provision.RbTargetReturned:
		return r.toCleaning(ctx, provision.RbTargetReturned)
	case provision.RbTargetCleaning:
		return r.cleaning(ctx)
	case provision.RbTargetCleaned:
		return false, r.cleaned()
	}
	return false, nil // lost: the declared loss's (Task 10)
}

// phase writes one target phase step, logs it (movesevents.go names it from
// the record; the last argument, the step's old event name, is unused) and
// passes the step's fault point. A conflict is someone else's decision.
func (r *rollbackRun) phase(ctx context.Context, from, to string, stamp func(*provision.MoveRollback), _ string) (bool, error) {
	host := r.rec.Host
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRollbackTargetPhase(r.g, from, to, stamp)))
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: the target's rollback %s to %s refused (%v); the record moved on", host, from, to, err)
			return false, nil
		}
		return false, fmt.Errorf("the target's rollback %s to %s: %w", from, to, err)
	}
	log.Printf("moves: %s: the target of rollback %s is %s", host, r.rb, to)
	r.w.logTransition(ctx, r.s, pre, *after)
	if err := r.fault("after-record-" + to); err != nil {
		return true, err
	}
	return true, nil
}

// claim reads the host claim and requires this rollback's restoring op,
// with this box the move's target, for the record's incarnation.
func (r *rollbackRun) 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 target phase: this rollback's restoring
// op, this box the move's target, the record's incarnation.
func (r *rollbackRun) 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) != 2 || c.Op.Participants[1] != box || c.Op.Intent["target"] != box {
		return fmt.Errorf("%w: this box is not the move's target in the claim", hostclaim.ErrWrongState)
	}
	return nil
}

// stillReleasing is the final checkpoint's authority: every replica request
// re-reads the claim, which must still name this rollback releasing with this
// box executing.
func (r *rollbackRun) stillReleasing(ctx context.Context) error {
	obs, err := r.claim(ctx)
	if err != nil {
		return err
	}
	if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != r.cfg().Box {
		return fmt.Errorf("move %s: this box no longer executes the host", r.g.ID)
	}
	return nil
}

// stillReleasingRecord is the hand-over's record check (J3, fix round 1,
// ruling 1), run by handHold under the host's hold lock: the record, read
// fresh, must still be this move and incarnation, rolling back under this
// rollback, with the target releasing. It reads the record only: the lock is
// the wrapper's, and nothing here takes it again.
func (r *rollbackRun) stillReleasingRecord(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 m := fresh.Move; m == nil || m.ID != r.g.ID || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack ||
		m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetReleasing {
		return fmt.Errorf("move %s: the record is no longer rollback %s releasing its target", r.g.ID, r.rb)
	}
	return nil
}

// reachedProbing reports whether the landing reached probing: a writer may
// have run on this box.
func (r *rollbackRun) reachedProbing() bool {
	order := provision.LandingOrder()
	return landingIndex(order, r.rec.Move.LandingPhase) >= landingIndex(order, provision.LandProbing)
}

// unchanged reports whether the stopped target's digests are what the
// landing installed: the database, the files and the timers.
func (r *rollbackRun) unchanged(report moves.StoppedReport) bool {
	m := r.rec.Move
	return m.Landing != nil && m.Landing.InstalledDBSHA256 != "" && report.DBSHA256 == m.Landing.InstalledDBSHA256 &&
		m.Source != nil && m.Source.Files != nil && report.FilesPlain == m.Source.Files.Plain && report.TimersSHA256 == m.Source.TimersSHA256
}

// releasing (§7.8 target releasing and released): the hold, the fence, the
// drain and the stop evidence; the final checkpoint and the digests; the
// stopped report; the record's released with Dirty and Evidence; the claim's
// release. A replay runs it all again: the hand-over, the fence and the drain
// are idempotent, the evidence reads durable state, the checkpoint is
// idempotent on a stopped database, and the report is create-only and read
// back, so the decision is the published report's.
func (r *rollbackRun) releasing(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if _, err := r.w.rollbackClaim(ctx, r.s, r.rec); err != nil {
		return false, err
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleasing || op.Exec != cfg.Box {
		return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not releasing on this box", id, op.Phase, op.Exec)
	}
	holder := *obs.Claim.Holder
	if _, err := writerLease(cfg.OwnPrefix, host, holder); err != nil {
		return false, fmt.Errorf("move %s: %w", id, err)
	}
	seal, err := journalledSeal(obs.Claim)
	if err != nil {
		return false, err
	}
	handed, err := cfg.HandHold(host, "releasing", r.rb, id, func() error { return r.stillReleasingRecord(ctx) })
	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)
	}
	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)
	}
	dbPath := r.dbPath()
	ev, err := cfg.Evidence(ctx, host, holder.Location, []string{dbPath})
	if err != nil {
		return false, fmt.Errorf("stop evidence: %w", err)
	}
	if !ev.TreeStopped() {
		r.w.notice(host+"#rb-evidence", "moves: %s: the target is not proven stopped yet (%s); the release retries until its deadline", host, failedSteps(ev))
		return false, nil
	}
	reached := r.reachedProbing()
	fromDB, err := regularFile(dbPath)
	if err != nil {
		return false, err
	}
	served := reached && fromDB
	if served {
		if err := cfg.FinalCheckpoint(ctx, host, dbPath, holder, r.stillReleasing); err != nil {
			return false, fmt.Errorf("final checkpoint: %w", err)
		}
	}
	if err := r.fault("after-checkpoint"); err != nil {
		return false, err
	}
	n := seal.N
	if served {
		if n, err = cfg.LocalPosition(dbPath); err != nil {
			return false, fmt.Errorf("read the stopped database's position: %w", err)
		}
	}
	var dbSHA string
	if fromDB {
		if dbSHA, err = fileSHA256(dbPath); err != nil {
			return false, fmt.Errorf("digest the stopped database: %w", err)
		}
	}
	dir, envPath := cfg.FilesDir(host)
	plain, _, _, err := archivePlainDigest(dir, envPath)
	if err != nil {
		if reached && archiveRefused(err) {
			// Files the archive can never carry: no retry by itself can
			// help, so the release stops for an operator (Plan 5 Task 9a),
			// keeping its phase and everything the target holds.
			return r.stuck(ctx, &returnRefusal{reason: provision.StuckReleaseTargetFiles,
				err: fmt.Errorf("move %s: the target's stopped files cannot be archived: %w", id, err)})
		}
		if reached {
			return false, fmt.Errorf("digest the stopped files: %w", err)
		}
		// No writer ran here: the files are not compared, and may not
		// have been extracted yet.
		plain = ""
	}
	rows, err := cfg.Timers(host)
	if err != nil {
		return false, fmt.Errorf("read the timers: %w", err)
	}
	report := ev
	report.V, report.Role, report.Box, report.Incarnation, report.ClaimGen = 1, stoppedTargetRole, cfg.Box, holder.Incarnation, obs.Claim.Gen
	report.Host, report.OpID = host, id
	report.Steps = map[string]moves.Check{}
	for k, v := range ev.Steps {
		report.Steps[k] = v
	}
	report.Steps[moves.StepReplica] = moves.Check{OK: true}
	report.FinalTXID, report.DBSHA256, report.FilesPlain, report.TimersSHA256 = n, dbSHA, plain, moves.TimersDigest(rows)
	// The report's time is after the checkpoint, the target's last write to
	// its location: it is the horizon of that location's park.
	report.CheckedAt = r.now().UTC()
	if !report.Complete() {
		return false, fmt.Errorf("the target's stopped report is not complete (%s)", failedSteps(report))
	}
	key := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), stoppedTargetRole)
	if report, err = publishStoppedReport(ctx, r.s, key, report); err != nil {
		return false, err
	}
	if err := r.fault("after-report"); err != nil {
		return false, err
	}
	clean := seal.SrcCovered && (!reached || r.unchanged(report))
	log.Printf("moves: %s: the target of rollback %s stopped at TXID %d (clean %t, src_covered %t)", host, r.rb, report.FinalTXID, clean, seal.SrcCovered)
	ok, err := r.phase(ctx, provision.RbTargetReleasing, provision.RbTargetReleased, func(rb *provision.MoveRollback) {
		rb.Dirty, rb.Evidence = !clean, key
	}, "target-released")
	if !ok || err != nil {
		return ok, err
	}
	if err := r.claimReleased(ctx, key); err != nil {
		log.Printf("moves: %s: the claim half of the target's release waits for the next pass: %v", host, err)
	}
	return true, nil
}

// claimReleased is the claim half of released (TargetReleased): execution to
// none citing the target's evidence, the target region's location park whose
// horizon is the report's time, the target's park-local when its inventory
// lists F, and the region's predecessors covered. A claim already released
// with this evidence is done.
func (r *rollbackRun) claimReleased(ctx context.Context, evidence string) error {
	cfg := r.cfg()
	obs, err := r.claim(ctx)
	if err != nil {
		return err
	}
	op := obs.Claim.Op
	if op.Phase != hostclaim.PhaseReleasing {
		if op.Journal[hostclaim.JournalEvidenceTarget] == evidence {
			return nil
		}
		return fmt.Errorf("move %s: the claim is restoring/%q and cites %q, not %s", r.g.ID, op.Phase, op.Journal[hostclaim.JournalEvidenceTarget], evidence)
	}
	report, err := readStoppedReport(ctx, r.s, evidence)
	if err != nil {
		return err
	}
	holder := obs.Claim.Holder
	scope := moves.TargetLocationScope(*holder, report.CheckedAt)
	if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.TargetReleased(r.g.ID, cfg.Box, cfg.parkRegion(), evidence, scope, obligationID)); err != nil {
		return fmt.Errorf("the claim half of the release: %w", err)
	}
	log.Printf("moves: %s: the target gave execution up for rollback %s, citing %s", r.rec.Host, r.rb, evidence)
	return nil
}

// released: the claim half, if a crash left it, then returning (dirty) or,
// once the source has recovered, cleaning (clean).
func (r *rollbackRun) released(ctx context.Context) (bool, error) {
	if err := r.claimReleased(ctx, r.rollback().Evidence); err != nil {
		return false, err
	}
	if r.rollback().Dirty {
		return r.phase(ctx, provision.RbTargetReleased, provision.RbTargetReturning, nil, "")
	}
	return r.toCleaning(ctx, provision.RbTargetReleased)
}

// toCleaning waits for the source's recovery (Review Focus 4): until then the
// target keeps everything it holds.
func (r *rollbackRun) toCleaning(ctx context.Context, from string) (bool, error) {
	if r.rollback().RecoveredAt == nil {
		r.w.notice(r.rec.Host+"#rb-clean", "moves: %s: the target keeps what it holds for rollback %s until the source has recovered", r.rec.Host, r.rb)
		return false, nil
	}
	return r.phase(ctx, from, provision.RbTargetCleaning, nil, "")
}

// returnedLane is what a return lane holds and restores to.
type returnedLane struct {
	n                      uint64
	replica, files, timers []moves.SealedObject
	// state is the returned build state (settings spec rev 11 §4.3a), nil
	// when there is none.
	state    []moves.SealedObject
	baseline provision.MoveSource
}

// returning (§7.8 target returning): the return lane under the source's
// staging-work lease, the return seal and baseline in the claim's journal,
// then the record's returned. A replay that finds the seal journalled only
// writes the record; one that does not writes the lane again (new versions;
// the seal names the ones it wrote, and the lane's park removes the rest).
func (r *rollbackRun) returning(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, m := r.rec.Host, r.g.ID, r.rec.Move
	rb := r.rollback()
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || op.Exec != "" || op.Journal[hostclaim.JournalEvidenceTarget] != rb.Evidence {
		return false, fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not released by this target", id, op.Phase, op.Exec)
	}
	if _, ok := obs.Claim.Op.Journal[moves.JournalReturnSeal]; ok {
		return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned")
	}
	report, err := readStoppedReport(ctx, r.s, rb.Evidence)
	if err != nil {
		return false, err
	}
	staged, err := journalledSeal(obs.Claim)
	if err != nil {
		return false, err
	}
	fleet, err := loadMoveFleet(ctx, r.s)
	if err != nil {
		return false, err
	}
	src, ok := fleet.targets[m.From]
	if !ok {
		return false, fmt.Errorf("source region %s publishes no data bucket. waiting", m.From)
	}
	held, release, err := cfg.StagingLease(ctx, m.From, host)
	if err != nil {
		var he *lease.HeldError
		if errors.As(err, &he) {
			r.w.notice(host+"#rb-lease", "moves: %s: another worker holds the staging lease; the return of %s waits", host, r.rb)
			return false, nil
		}
		return false, fmt.Errorf("staging lease: %w", err)
	}
	defer func() {
		rctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		if err := release(rctx); err != nil {
			log.Printf("moves: %s: release the staging lease: %v", host, err)
		}
	}()
	stg, err := r.w.staging(ctx, src.DataBucketRegion)
	if err != nil {
		return false, err
	}
	own, err := r.w.mirror(ctx, cfg.OwnBucketRegion)
	if err != nil {
		return false, err
	}
	gm := &guardedMirror{w: stg, g: moves.Guarded{Lease: held, Now: r.now}, dst: src.DataBucket, own: own}
	if err := r.laneMoveJSON(ctx, gm); err != nil {
		return false, err
	}
	var lane returnedLane
	if !r.reachedProbing() || r.unchanged(report) {
		lane, err = r.copyAsGiven(ctx, gm, staged)
	} else {
		lane, err = r.snapshotStopped(ctx, gm, report)
	}
	if err == nil {
		lane.state, err = r.returnState(ctx, gm, obs.Claim.Holder, staged)
	}
	var refused *returnRefusal
	if errors.As(err, &refused) {
		return r.stuck(ctx, refused)
	}
	if err != nil {
		return false, err
	}
	// The bound is read after the last lane write: every request the guard
	// let a writer sign expires before it.
	bound, err := held.SafeUntil()
	if err != nil {
		return false, fmt.Errorf("the staging lease's bound: %w", err)
	}
	seal := moves.ReturnSeal(lane.n, lane.replica, lane.files, lane.timers, lane.state, r.now(), bound)
	if err := seal.Validate(); err != nil {
		return false, err
	}
	if !seal.ReachesN() {
		return false, fmt.Errorf("the return lane does not end at TXID %d", seal.N)
	}
	base, err := moves.EncodeReturnBaseline(lane.baseline)
	if err != nil {
		return false, err
	}
	if obs, err = r.claim(ctx); err != nil {
		return false, err
	}
	if _, err := hostclaim.Update(ctx, r.s, obs, journalReturn(id, r.rb, rb.Evidence, seal.Encode(), base)); err != nil {
		return false, fmt.Errorf("journal the return seal: %w", err)
	}
	log.Printf("moves: %s: the return lane of %s is sealed: %d replica objects at TXID %d", host, r.rb, len(seal.Replica), seal.N)
	if err := r.fault("after-return-seal"); err != nil {
		return false, err
	}
	return r.phase(ctx, provision.RbTargetReturning, provision.RbTargetReturned, nil, "target-returned")
}

// journalReturn is the return seal's claim write: the seal and the baseline,
// in one write, on this rollback's op released by this target.
func journalReturn(opID, rbID, evidence, seal, baseline string) func(*hostclaim.Claim) error {
	return func(c *hostclaim.Claim) error {
		if err := hostclaim.RequireOp(opID)(c); err != nil {
			return err
		}
		op := c.Op
		if op.Kind != hostclaim.KindRestoring || op.Intent["rollback"] != rbID || op.Phase != hostclaim.PhaseReleased || op.Exec != "" ||
			op.Journal[hostclaim.JournalEvidenceTarget] != evidence {
			return fmt.Errorf("%w: the return seal needs this rollback released by its target", hostclaim.ErrWrongState)
		}
		return hostclaim.SetJournal(opID, map[string]string{moves.JournalReturnSeal: seal, moves.JournalReturnBaseline: baseline})(c)
	}
}

// laneMoveJSON writes the lane's move.json create-only (§7.3), naming the
// rollback; one already there must name it and this incarnation.
func (r *rollbackRun) laneMoveJSON(ctx context.Context, gm *guardedMirror) error {
	host, m := r.rec.Host, r.rec.Move
	body, err := json.Marshal(moves.MoveJSON{MoveID: r.rb, Incarnation: m.Incarnation, Host: host, From: m.To, To: m.From})
	if err != nil {
		return err
	}
	key := moves.MoveJSONKey(host, r.rb)
	if _, err := gm.PutIfAbsentVersioned(ctx, gm.dst, key, body); errors.Is(err, lease.ErrExists) {
		prev, err := gm.Get(ctx, gm.dst, key)
		if err != nil {
			return fmt.Errorf("read the lane's move.json: %w", err)
		}
		var mj moves.MoveJSON
		if err := json.Unmarshal(prev, &mj); err != nil || mj.MoveID != r.rb || mj.Incarnation != m.Incarnation {
			return fmt.Errorf("the lane's move.json names %q/%q, not rollback %s", mj.MoveID, mj.Incarnation, r.rb)
		}
	} else if err != nil {
		return fmt.Errorf("write the lane's move.json: %w", err)
	}
	return nil
}

// copyAsGiven is the return of a target whose state is the staged seal's
// (dirty only for want of src_covered): every sealed version is read from
// this box's bucket and written into the lane under the same key relative to
// the lane, through the guarded staging writer (J22: PutReaderVersionedAt
// under Guarded.Before). The baseline is the source's own, cited under the
// target's evidence.
func (r *rollbackRun) copyAsGiven(ctx context.Context, gm *guardedMirror, staged moves.Seal) (returnedLane, error) {
	cfg := r.cfg()
	host, id, m := r.rec.Host, r.g.ID, r.rec.Move
	if m.Source == nil || m.Source.Files == nil || staged.N != m.Source.FinalTXID {
		return returnedLane{}, fmt.Errorf("move %s: the record's source baselines are not the staged seal's", id)
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return returnedLane{}, err
	}
	tmp := filepath.Join(moves.JournalDir(cfg.DataDir, host, id), "return")
	if err := os.MkdirAll(tmp, 0o700); err != nil {
		return returnedLane{}, err
	}
	defer os.RemoveAll(tmp)
	from, lane := moves.StagingPrefix(host, id), moves.ReturnLane(host, r.rb)
	copyClass := func(objs []moves.SealedObject) ([]moves.SealedObject, error) {
		var out []moves.SealedObject
		for _, o := range objs {
			rel, ok := strings.CutPrefix(o.Key, from)
			if !ok || rel == "" {
				return nil, fmt.Errorf("%w: %s is outside the move's lane", moves.ErrBadSeal, o.Key)
			}
			path := filepath.Join(tmp, "object")
			if err := downloadExact(ctx, vs, cfg.OwnBucket, o.Key, o.VersionID, o.Size, path); err != nil {
				return nil, fmt.Errorf("read %s: %w", o.Key, err)
			}
			sum, err := fileSHA256(path)
			if err != nil {
				return nil, err
			}
			ver, err := gm.PutFileVersioned(ctx, gm.dst, lane+rel, path, o.Size, digestHex(sum))
			os.Remove(path)
			if err != nil {
				return nil, fmt.Errorf("return %s: %w", o.Key, err)
			}
			if !lease.UsableVersionID(ver) {
				return nil, fmt.Errorf("the returned %s has no usable version id %q", rel, ver)
			}
			out = append(out, moves.SealedObject{Key: lane + rel, VersionID: ver, Size: o.Size})
		}
		return out, nil
	}
	var out returnedLane
	if out.replica, err = copyClass(staged.Replica); err != nil {
		return out, err
	}
	if out.files, err = copyClass(staged.Files); err != nil {
		return out, err
	}
	if out.timers, err = copyClass(staged.Timers); err != nil {
		return out, err
	}
	files := *m.Source.Files
	out.n, out.baseline = staged.N, *m.Source
	out.baseline.Files, out.baseline.Evidence = &files, r.rollback().Evidence
	return out, nil
}

// returnState is a dirty return's build state (settings spec rev 11
// §4.3a), through the return lane's state/ key and sealed by the version
// written there. Once the target's install has built (its landing journal
// records the install), or the landing reached probing (reachedProbing,
// the predicate the return's lane choice uses: a writer may have run, and
// a journal that lost the install step is rebuilt by installManifest,
// which never places the build state again), the state beside the
// target's lease at the holder's location is the instance's: the moved
// state the install placed, with whatever the probe and the target's
// wakes recorded over it. Before both the target's location holds no
// state of this move that differs from the seal's (a return before the
// building step, or before any writer ran over the placed copy), so the
// forward lane's sealed state goes back as given: read at the move's own
// lane key by its exact version and its sealed size, as copyAsGiven reads
// the other classes. Neither holds one: none goes back.
func (r *rollbackRun) returnState(ctx context.Context, gm *guardedMirror, h *hostclaim.Holder, staged moves.Seal) ([]moves.SealedObject, error) {
	cfg := r.cfg()
	host := r.rec.Host
	j, err := r.journal()
	if err != nil {
		return nil, err
	}
	if !j.Done(landStepInstall) && !r.reachedProbing() {
		if len(staged.State) == 0 {
			// A lane object an earlier attempt left is not sealed, so the
			// source's install never places it.
			return nil, nil
		}
		o := staged.State[0]
		if want := moves.BuildStateKey(host, r.g.ID); o.Key != want {
			return nil, fmt.Errorf("%w: the sealed build state %s is not the move's lane key %s", moves.ErrBadSeal, o.Key, want)
		}
		vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
		if err != nil {
			return nil, err
		}
		body, err := readExact(ctx, vs, cfg.OwnBucket, o.Key, o.VersionID, o.Size)
		if err != nil {
			return nil, fmt.Errorf("read the sealed build state %s: %w", o.Key, err)
		}
		key := moves.BuildStateKey(host, r.rb)
		ver, err := gm.PutVersioned(ctx, gm.dst, key, body)
		if err != nil {
			return nil, fmt.Errorf("return the build state: %w", err)
		}
		if !lease.UsableVersionID(ver) {
			return nil, fmt.Errorf("the returned build state has no usable version id %q", ver)
		}
		return []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(body))}}, nil
	}
	if h == nil {
		return nil, fmt.Errorf("move %s: the claim has no holder to return the build state from", r.g.ID)
	}
	// This box's request record goes back too (ruling R-SPEC-43A-2): a
	// request the target closed as failed stays closed on the source.
	if cfg.CarryRequest != nil {
		if err := cfg.CarryRequest(ctx, host, cfg.OwnBucket, strings.TrimSuffix(h.Location, "/")); err != nil {
			return nil, fmt.Errorf("carry the request record into the returned build state: %w", err)
		}
	}
	_, o, err := mirrorBuildState(ctx, gm, cfg.OwnBucket, h.Location, gm.dst, host, r.rb)
	if err != nil {
		return nil, fmt.Errorf("return the build state: %w", err)
	}
	if o == nil {
		return nil, nil
	}
	if !lease.UsableVersionID(o.VersionID) {
		return nil, fmt.Errorf("the returned build state has no usable version id %q", o.VersionID)
	}
	return []moves.SealedObject{*o}, nil
}

// readExact reads one exact version of size bytes into memory, as
// downloadExact does into a file: a version of any other size refuses.
func readExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64) ([]byte, error) {
	rc, err := vs.OpenVersion(ctx, bucket, key, version)
	if err != nil {
		return nil, err
	}
	defer rc.Close()
	body, err := io.ReadAll(io.LimitReader(rc, size+1))
	if err != nil {
		return nil, err
	}
	if int64(len(body)) != size {
		return nil, fmt.Errorf("the version is %d bytes, sealed as %d", len(body), size)
	}
	return body, nil
}

// downloadExact reads one exact version of size bytes into path.
func downloadExact(ctx context.Context, vs lease.VersionedStore, bucket, key, version string, size int64, path string) error {
	rc, err := vs.OpenVersion(ctx, bucket, key, version)
	if err != nil {
		return err
	}
	defer rc.Close()
	f, err := os.OpenFile(path, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
	if err != nil {
		return err
	}
	n, err := io.Copy(f, io.LimitReader(rc, size+1))
	if cerr := f.Close(); err == nil {
		err = cerr
	}
	if err != nil {
		return err
	}
	if n != size {
		return fmt.Errorf("the version is %d bytes, sealed as %d", n, size)
	}
	return nil
}

// returnRefusal is a dirty return that cannot be built from what the target
// holds (fix round 1, ruling 2): its database data is missing, it changed
// after it stopped, or its files cannot be archived. No retry by itself can
// help, so the record stops for an operator with reason. Storage, network
// and local read errors are not refusals: they stay ordinary errors, and the
// next pass retries.
type returnRefusal struct {
	reason string
	err    error
}

func (e *returnRefusal) Error() string { return e.err.Error() }
func (e *returnRefusal) Unwrap() error { return e.err }

func refuseReturn(reason string, err error) error { return &returnRefusal{reason: reason, err: err} }

// archiveRefused reports whether err is one the archive gives for files it
// can never carry (a special file, over the cap, too many entries, a path
// too long), as opposed to a change under the walk or the disk's error,
// which a retry may get past.
func archiveRefused(err error) bool {
	return errors.Is(err, archive.ErrSpecialFile) || errors.Is(err, archive.ErrOverCap) ||
		errors.Is(err, archive.ErrTooManyEntries) || errors.Is(err, archive.ErrBadPath)
}

// databaseGone reports whether err says the database's data is not there to
// read (absent, short, or not a database), as opposed to a read that failed:
// an *os.PathError other than "does not exist" is the disk's, and transient.
func databaseGone(err error) bool {
	if errors.Is(err, os.ErrNotExist) {
		return true
	}
	var pe *os.PathError
	return !errors.As(err, &pe)
}

// stuck stops the record for an operator (MoveMarkStuck) at the target's
// releasing or returning: the phase, the row and everything the target
// holds are kept, and move --retry runs that phase again.
func (r *rollbackRun) stuck(ctx context.Context, refused *returnRefusal) (bool, error) {
	host := r.rec.Host
	log.Printf("moves: %s: the target's half of %s cannot go on (%v); it stops for an operator: %s", host, r.rb, refused.err, refused.reason)
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, refused.reason)))
	if 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 return stuck: %w", err)
	}
	r.w.logTransition(ctx, r.s, pre, *after)
	r.w.pageStuck(ctx, r.rec, "target", refused.reason)
	nudgeSerial(ctx, r.s)
	return true, nil
}

// snapshotStopped is the return of a target that served and changed: one
// snapshot [1, N'] of the stopped database (moves.EncodeFull, in memory, the
// Phase 1 size limit), the target's files archive (manifest bound to the move
// id) and its timers. Each must still be what the stopped report recorded:
// data that is gone (StuckReturnTargetDatabase), a state that differs from
// the report (StuckReturnTargetChanged) and files that cannot be archived
// (StuckReturnTargetFiles) are refusals; anything else is retried.
func (r *rollbackRun) snapshotStopped(ctx context.Context, gm *guardedMirror, report moves.StoppedReport) (returnedLane, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	dbPath := r.dbPath()
	sha, err := fileSHA256(dbPath)
	if err != nil {
		err = fmt.Errorf("the target's stopped database: %w", err)
		if errors.Is(err, os.ErrNotExist) {
			err = refuseReturn(provision.StuckReturnTargetDatabase, err)
		}
		return returnedLane{}, err
	}
	if sha != report.DBSHA256 {
		return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's database changed since its release", id))
	}
	n, err := cfg.LocalPosition(dbPath)
	if err != nil {
		return returnedLane{}, fmt.Errorf("read the stopped database's position: %w", err)
	}
	if n == 0 {
		return returnedLane{}, refuseReturn(provision.StuckReturnTargetDatabase, fmt.Errorf("move %s: the stopped database has no position", id))
	}
	if n != report.FinalTXID {
		return returnedLane{}, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the stopped database is at TXID %d, the report at %d", id, n, report.FinalTXID))
	}
	pageSize, pageCount, err := moves.ReadDBHeader(dbPath)
	if err != nil {
		if databaseGone(err) {
			err = refuseReturn(provision.StuckReturnTargetDatabase, err)
		}
		return returnedLane{}, err
	}
	f, err := os.Open(dbPath)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			err = refuseReturn(provision.StuckReturnTargetDatabase, err)
		}
		return returnedLane{}, err
	}
	data, _, err := moves.EncodeFull(f, pageSize, pageCount, 1, ltx.TXID(n), report.CheckedAt.UnixMilli(), 0)
	f.Close()
	if err != nil {
		err = fmt.Errorf("snapshot the stopped database: %w", err)
		if errors.Is(err, io.EOF) || errors.Is(err, io.ErrUnexpectedEOF) {
			// The header names pages the file does not hold.
			err = refuseReturn(provision.StuckReturnTargetDatabase, err)
		}
		return returnedLane{}, err
	}
	var out returnedLane
	key := moves.SnapshotKey(moves.ReplicaPrefix(host, r.rb), n)
	ver, err := gm.PutVersioned(ctx, gm.dst, key, data)
	if err != nil {
		return out, fmt.Errorf("return the snapshot: %w", err)
	}
	if !lease.UsableVersionID(ver) {
		return out, fmt.Errorf("the returned snapshot has no usable version id %q", ver)
	}
	out.replica = []moves.SealedObject{{Key: key, VersionID: ver, Size: int64(len(data))}}
	a, why, retry, err := r.w.buildArchive(r.rec)
	switch {
	case err != nil:
		return out, fmt.Errorf("archive the target's files: %w", err)
	case why != "":
		return out, refuseReturn(provision.StuckReturnTargetFiles, fmt.Errorf("move %s: the target's files cannot be archived (%s)", id, why))
	case retry:
		return out, errors.New("a stopped file changed while it was archived")
	case a.Plain != report.FilesPlain:
		return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's files are %s, the report %s", id, a.Plain, report.FilesPlain))
	}
	ref, err := a.stage(ctx, gm, gm.dst, host, r.rb)
	if err != nil {
		return out, err
	}
	mst, err := os.Stat(a.manifestPath())
	if err != nil {
		return out, err
	}
	if !lease.UsableVersionID(ref.SealedVersion) || !lease.UsableVersionID(ref.ManifestVersion) {
		return out, fmt.Errorf("the returned archive has no usable version ids (%q, %q)", ref.SealedVersion, ref.ManifestVersion)
	}
	out.files = []moves.SealedObject{{Key: ref.SealedObjectKey, VersionID: ref.SealedVersion, Size: a.SealedSize},
		{Key: ref.ManifestKey, VersionID: ref.ManifestVersion, Size: mst.Size()}}
	rows, err := cfg.Timers(host)
	if err != nil {
		return out, fmt.Errorf("read the timers: %w", err)
	}
	if moves.TimersDigest(rows) != report.TimersSHA256 {
		return out, refuseReturn(provision.StuckReturnTargetChanged, fmt.Errorf("move %s: the target's timers changed since its release", id))
	}
	body, err := moves.TimersSnapshot{V: 1, Host: host, Rows: rows, TakenAt: report.CheckedAt}.Encode()
	if err != nil {
		return out, err
	}
	tkey := moves.TimersPrefix(host, r.rb) + "timers.json"
	tver, err := gm.PutVersioned(ctx, gm.dst, tkey, body)
	if err != nil {
		return out, fmt.Errorf("return the timers: %w", err)
	}
	if !lease.UsableVersionID(tver) {
		return out, fmt.Errorf("the returned timers have no usable version id %q", tver)
	}
	out.timers = []moves.SealedObject{{Key: tkey, VersionID: tver, Size: int64(len(body))}}
	rt, err := cfg.Route(host)
	if err != nil {
		return out, fmt.Errorf("read the registry row: %w", err)
	}
	if rt == nil {
		return out, fmt.Errorf("%s has no registry row on this box", host)
	}
	out.n = n
	out.baseline = provision.MoveSource{Replication: rt.Replication, NextDue: rt.NextDue, FinalTXID: n, DBSHA256: sha,
		Files:        &provision.MoveFilesDigest{Plain: a.Plain, Sealed: a.Sealed, Bytes: a.Bytes, Entries: a.Entries},
		TimersSHA256: report.TimersSHA256, Evidence: r.rollback().Evidence}
	return out, nil
}

// journal is the target's op journal of the move (the landing's).
func (r *rollbackRun) journal() (*moves.Journal, error) {
	return moves.ReadJournal(r.cfg().DataDir, r.rec.Host, r.g.ID)
}

// cleaning (§7.8 target cleaning, INV-10), each step journalled and skipped
// on a replay once done: the remote parks; the registered installation; the
// moves-owned rename of everything else at the flat paths; the row; then the
// record's cleaned with target_parked_at. The source has recovered (the
// record says so), so nothing here is still needed by the move.
func (r *rollbackRun) cleaning(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	j, err := r.journal()
	if err != nil {
		return false, err
	}
	mark := func(step, value string) error {
		j.Mark(step, value)
		if err := j.Commit(cfg.DataDir); err != nil {
			return err
		}
		return r.fault("after-" + step)
	}
	obs, err := hostclaim.Read(ctx, r.s, host)
	if err != nil {
		return false, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent || obs.Claim.Holder == nil || obs.Claim.Holder.Incarnation != r.g.Incarnation {
		return false, fmt.Errorf("move %s: the host claim names no holder of this incarnation", id)
	}
	ours := r.ours(obs.Claim) == nil
	if ours && !j.Done(rbStepTargetStamp) {
		if err := mark(rbStepTargetStamp, obs.Claim.Op.Journal[hostclaim.JournalTargetStamp]); err != nil {
			return false, err
		}
	}
	if !j.Done(rbStepRemote) {
		// An op that has moved on had its chores acknowledged with it.
		if ours {
			done, err := r.remote(ctx, obs, j, mark)
			if err != nil || !done {
				return false, err
			}
		}
		if err := mark(rbStepRemote, ""); err != nil {
			return false, err
		}
	}
	if !j.Done(rbStepInstallation) {
		how, err := r.installation(ctx, ours)
		if err != nil {
			return false, err
		}
		if err := mark(rbStepInstallation, how); err != nil {
			return false, err
		}
	}
	if err := r.parkLeftovers(ctx, j, mark); err != nil {
		return false, fmt.Errorf("park the target's local artifacts: %w", err)
	}
	if err := r.deleteRow(ctx, j, mark); err != nil {
		return false, err
	}
	ok, err := r.phase(ctx, provision.RbTargetCleaning, provision.RbTargetCleaned, func(rb *provision.MoveRollback) {
		t := r.now().UTC()
		rb.TargetParkedAt = &t
	}, "target-cleaned")
	if !ok || err != nil {
		return ok, err
	}
	// The journal is parked in this same call: once target_parked_at is
	// stamped the source may close the record before this worker reads it
	// again, and then only the local sweep finds the journal.
	if err := mark(rbStepCleaned, "done"); err != nil {
		return true, err
	}
	return true, parkRollbackJournal(cfg.DataDir, host, id, j)
}

// remote parks the target region's location and the move's lane in it: the
// lane's park is queued once (its horizon the staged seal's bound), then
// every park-replica of this op in this region runs through the park engine
// until its archive is complete (park.Engine.Parked). The acknowledgement,
// after the horizon, is the engine's or the regional runner's (C4), never a
// reason to wait here; a chore no longer in the claim was acknowledged.
func (r *rollbackRun) remote(ctx context.Context, obs hostclaim.Observed, j *moves.Journal, mark func(step, value string) error) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if !j.Done(rbStepLaneQueued) {
		staged, err := journalledSeal(obs.Claim)
		if err != nil {
			return false, err
		}
		scope := moves.StagingScope(host, id, obs.Claim.Holder.Incarnation, staged.Bound())
		if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueStagingPark(id, cfg.parkRegion(), scope, obligationID)); err != nil {
			return false, fmt.Errorf("queue the move's lane park: %w", err)
		}
		if err := mark(rbStepLaneQueued, ""); err != nil {
			return false, err
		}
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return false, err
	}
	e := r.engine(vs)
	waiting := 0
	for _, o := range obs.Claim.Obligations {
		if o.Op != id || o.Kind != "park-replica" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) || o.Park == nil {
			continue
		}
		work := park.Work{Host: host, ID: o.ID, Op: id}
		if err := e.Run(ctx, work); err != nil && !errors.Is(err, park.ErrPending) && !errors.Is(err, park.ErrChanged) {
			return false, fmt.Errorf("park %s: %w", o.Park.Location, err)
		}
		parked, err := e.Parked(ctx, work)
		if errors.Is(err, park.ErrChanged) {
			continue
		}
		if err != nil {
			return false, fmt.Errorf("the park of %s: %w", o.Park.Location, err)
		}
		if !parked {
			waiting++
		}
	}
	if waiting > 0 {
		r.w.notice(host+"#rb-remote", "moves: %s: the target's cleaning of %s waits for %d park archive(s)", host, r.rb, waiting)
		return false, nil
	}
	log.Printf("moves: %s: the target region's copies of %s are parked", host, id)
	return true, nil
}

// engine is the park engine over this box's own bucket with the claim's
// leases and view, and the claim's authorizer behind this rollback's own
// check (rollbackTargetArmed requires all three; nothing stands in).
func (r *rollbackRun) engine(vs lease.VersionedStore) park.Engine {
	cfg := r.cfg()
	own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
		if obs.Absent {
			return park.ErrUnsafe
		}
		return r.authorizeChore(obs.Claim, o)
	}
	return park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: r.s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases,
		Now: r.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault}
}

// authorizeChore is this rollback's own operation check (M17): a chore is
// run only when it is this op's, the claim is still this rollback's with this
// box no longer executing, and the chore is this region's or this box's. The
// engine runs it before the claim's authorizer, and park.Local before the
// claim's local authorizer (installation, movesLocalAuthorize).
func (r *rollbackRun) authorizeChore(c *hostclaim.Claim, o hostclaim.Obligation) error {
	cfg := r.cfg()
	if r.ours(c) != nil || o.Op != r.g.ID || c.Op.Exec == cfg.Box {
		return fmt.Errorf("%w: not this rollback's chore", park.ErrUnsafe)
	}
	switch {
	case o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region):
	case o.Kind == "park-local" && o.Box == cfg.Box:
	default:
		return fmt.Errorf("%w: not this target's chore", park.ErrUnsafe)
	}
	return nil
}

// targetStamp is F: the claim's journalled target stamp, as noted in this
// box's journal when cleaning began.
func targetStamp(j *moves.Journal) uint64 {
	f, err := strconv.ParseUint(j.Steps[rbStepTargetStamp], 10, 64)
	if err != nil {
		return 0
	}
	return f
}

// guard is the host guard over this box's data directory, reading the claim
// from the pass's store.
func (r *rollbackRun) guard() hostguard.Guard {
	g := r.cfg().Guard
	if g.Claims == nil {
		g.Claims = hostguard.StoreReader{Store: r.s}
	}
	return g
}

// installation is the registered half of the local retirement (J15): with the
// target's park-local and inventory paths of F, the installation is retired
// through park.Local under the host guard, which completes the chore itself;
// a park-local with nothing of F listed is completed; neither is an
// unregistered target, whose artifacts are all the moves-owned rename's.
func (r *rollbackRun) installation(ctx context.Context, ours bool) (string, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if !ours {
		return "moved-on", nil
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return "", err
	}
	c := obs.Claim
	f, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalTargetStamp], 10, 64)
	if err != nil || f == 0 {
		return "", fmt.Errorf("move %s: the claim journals no target stamp", id)
	}
	var chore *hostclaim.Obligation
	for i := range c.Obligations {
		if o := &c.Obligations[i]; o.Op == id && o.Kind == "park-local" && o.Box == cfg.Box {
			chore = o
		}
	}
	registered := hostclaim.InventoryHasStamp(c, cfg.Box, f)
	switch {
	case chore != nil && registered:
		captured, err := hostguard.Capture(cfg.DataDir, host)
		if err != nil {
			return "", fmt.Errorf("the target's installation record: %w", err)
		}
		if captured.Incarnation != c.Holder.Incarnation || (captured.Stamp != 0 && captured.Stamp != f) {
			return "", fmt.Errorf("move %s: the target's installation record is at %s/%d, the claim's is %s/%d", id, captured.Incarnation, captured.Stamp, c.Holder.Incarnation, f)
		}
		local := cfg.Local
		local.Guard, local.Claims = r.guard(), r.s
		if local.Box == "" {
			local.Box = cfg.Box
		}
		if local.Authorize == nil {
			// park.Local's authorizer: the claim's MoveParks local authorizer
			// (ParkLocalAuthorize) behind this rollback's own check. The
			// claim's engine authorizer refuses a park-local (M9), so it is
			// never asked here. Without it the chore refuses
			// (rollbackTargetArmed requires it).
			local.Authorize = movesLocalAuthorize(cfg.ParkLocalAuthorize, func(_ context.Context, c *hostclaim.Claim, o hostclaim.Obligation) error {
				return r.authorizeChore(c, o)
			})
		}
		if err := r.fault("before-rb-local-retire"); err != nil {
			return "", err
		}
		if err := local.RetireInstallation(ctx, host, hostguard.Captured{Incarnation: captured.Incarnation, Stamp: f}, chore.ID, id); err != nil {
			return "", fmt.Errorf("retire the target's installation: %w", err)
		}
		if err := r.fault("after-rb-local-retire"); err != nil {
			return "", err
		}
		log.Printf("moves: %s: the target's registered installation (stamp %d) is retired", host, f)
		return "retired", nil
	case chore != nil:
		if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.CompleteLocalPark(id, cfg.Box, f)); err != nil {
			return "", fmt.Errorf("complete the target's local park: %w", err)
		}
		return "completed", nil
	case registered:
		return "", fmt.Errorf("move %s: this box's inventory lists the target's paths, and the release queued no park-local for them", id)
	}
	return "unregistered", nil
}

// parkLeftovers is the moves-owned rename (H5(b), parkLocalArtifacts) of
// the artifacts this op created that the target still has at the flat paths,
// its plan and its renames inside the host guard, which retires the
// installation record at F if it is still active (an unregistered target),
// with the claim read fresh there: this box no longer executes and no longer
// takes part in the claim's op, and no inventory entry claims a path the
// rename would move.
//
// The candidates are the installation's paths, the rendered config and the
// sidecar's env (sourceArtifactPaths), but a candidate is moved only on a
// record that this op created it (rollbackOwnership; the final review's
// finding 1). Anything else, a files directory or env the landing refused as
// occupied (RollbackFilesOccupied) included, is logged and left where it is.
func (r *rollbackRun) parkLeftovers(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error {
	if j.Done(rbStepLocal) {
		return nil
	}
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	paths := sourceArtifactPaths(host)
	var owns func(path, identity string) (bool, error)
	if _, planned := j.Steps[rbStepLocal+"-plan"]; !planned {
		// A journalled plan already recorded what is the op's; the evidence
		// is read (the config ack is a store read) only before the plan.
		ev, err := r.ownership(ctx, j)
		if err != nil {
			return err
		}
		owns = ev.owns
	}
	var incarnation string
	check := func(c *hostclaim.Claim) error {
		if c.Op.Exec == cfg.Box {
			return fmt.Errorf("%w: this box executes the host again", hostclaim.ErrWrongState)
		}
		if c.Op.ID == id {
			if err := r.ours(c); err != nil {
				return err
			}
		} else if containsString(c.Op.Participants, cfg.Box) {
			return fmt.Errorf("%w: this box takes part in the claim's op %s", hostclaim.ErrWrongState, c.Op.ID)
		}
		for _, p := range paths {
			if _, ok := c.Inventory[cfg.Box].Paths[p]; ok {
				return fmt.Errorf("%w: the inventory claims %s", hostclaim.ErrWrongState, p)
			}
		}
		incarnation = c.Holder.Incarnation
		return nil
	}
	work := func() error {
		if incarnation != r.g.Incarnation {
			return fmt.Errorf("move %s: the claim names another incarnation", id)
		}
		if _, rendered := j.Steps[landStepPreRender]; rendered {
			if err := removeRenderTemp(cfg.DataDir, host); err != nil {
				return err
			}
		}
		_, err := parkLocalArtifacts(ctx, j, mark, localArtifactPark{DataDir: cfg.DataDir, Incarnation: incarnation, Host: host,
			Op: id, Step: rbStepLocal, Paths: paths, NewID: obligationID, Fault: r.fault, Owns: owns})
		return err
	}
	g := r.guard()
	captured, err := hostguard.Capture(cfg.DataDir, host)
	switch {
	case errors.Is(err, hostguard.ErrNoRecord):
		return g.WithNoInstallation(ctx, host, r.g.Incarnation, check, work)
	case err != nil:
		return fmt.Errorf("the target's installation record: %w", err)
	}
	f := captured.Stamp
	if f == 0 {
		f = targetStamp(j)
	}
	if f == 0 {
		return fmt.Errorf("move %s: the target's retired installation stamp is not known", id)
	}
	return g.RetireInstallationWith(ctx, host, hostguard.Captured{Incarnation: r.g.Incarnation, Stamp: f}, check, work)
}

// removeRenderTemp removes the held renderer's temp sibling of
// <host>.config.env (Task 9a fix round 5). The flat renderer writes the
// rendered bytes to <path>.tmp and renames them over the path (appconfig
// renderExecConfigBefore), so a crash between the two leaves the temp
// holding the rendered secrets: never live config, only an unfinished write
// of this op's, whose render the journal records (pre_render). It is
// removed, never parked. Only a regular file is removed (Lstat'ed, never
// followed); anything else at the name is logged and left. A moved row is
// always flat (the landing refuses a nested row at checking), so the nested
// renderer's temp (instancefs.WriteAtomic) never holds a held render.
func removeRenderTemp(dataDir, host string) error {
	tmp := filepath.Join(dataDir, host+".config.env.tmp")
	fi, err := os.Lstat(tmp)
	switch {
	case errors.Is(err, os.ErrNotExist):
		return nil
	case err != nil:
		return err
	case !fi.Mode().IsRegular():
		log.Printf("moves: %s: %s is not a regular file (%s); it is not the renderer's temp, and is left", host, tmp, fi.Mode())
		return nil
	}
	if err := os.Remove(tmp); err != nil && !errors.Is(err, os.ErrNotExist) {
		return fmt.Errorf("remove the held render's temp: %w", err)
	}
	log.Printf("moves: %s: the held render's unfinished write %s is removed", host, tmp)
	return syncPath(dataDir)
}

// rollbackOwnership is the record that this op created a path the target's
// cleaning may park (the final review's finding 1). The registered
// installation, the paths the landing listed under F, is park.Local's and is
// retired before the moves-owned rename runs; what is left at the flat paths
// is parked only on one of these:
//
//   - the files directory and <host>.env: the landing's placement receipt
//     (files_placed), or before it the staged extraction's identities
//     (staged), which a rename keeps: the live path must have that identity;
//   - the database, its marker, WAL and shared memory: the landing installed
//     its database, as the record says (landing.installed_db_sha256, stamped
//     by the CAS that ends configured); the install refused a WAL or shared
//     memory beside an absent database, so what sits beside it since is the
//     op's instance's. Before that stamp, the database and its marker are
//     the op's only when they are the build itself: the database's sha256
//     is the journalled install manifest's, and the marker names the op's
//     generation (a crash between the install and its stamp; carry (b));
//   - the quarantine and the litestream shadow, which the install does not
//     refuse: the database installed as above, and the install's journal
//     records the path absent just before it put the database in place
//     (pre_install, notePreInstall). Present then, or no record, is never
//     the op's (the final re-review's residual);
//   - the rendered config, <host>.config.env: the file's sha256 is one this
//     op rendered, as the held render's ack on this box records it (the ack
//     names this op), or as the landing's journal records it: pre_render's
//     intended hash, journalled before the render wrote, or rendered,
//     journalled as soon as the render answered (carry (c); fix round 1,
//     I1). Any other content is not the op's;
//   - the sidecar's env: never, since a moved host has no sidecar.
type rollbackOwnership struct {
	host, dataDir string
	placed        stagedFiles // the landing's receipt, or its staged identities
	installed     bool
	builtSHA256   string          // the journalled install manifest's database, "sha256:<hex>", or ""
	generation    string          // the generation the install names (the op id)
	absentBefore  map[string]bool // the siblings pre_install recorded absent
	configSHA256  []string        // what this op rendered, "sha256:<hex>": the ack's and the journal's
}

// ownership reads the evidence: the landing's journal, the record and the
// held render's config ack on this box.
func (r *rollbackRun) ownership(ctx context.Context, j *moves.Journal) (rollbackOwnership, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	ev := rollbackOwnership{host: host, dataDir: cfg.DataDir}
	for _, step := range []string{landStepFilesPlaced, landStepStaged} {
		if _, ok := j.Steps[step]; !ok {
			continue
		}
		placed, err := journalledStaged(j, step)
		if err != nil {
			return ev, err
		}
		ev.placed = placed
		break
	}
	if l := r.rec.Move.Landing; l != nil && l.InstalledDBSHA256 != "" {
		ev.installed = true
	}
	ev.generation = id
	if raw, ok := j.Steps[landStepInstall]; ok {
		var m moves.InstallManifest
		if err := json.Unmarshal([]byte(raw), &m); err == nil && m.Op == id && m.Host == host && m.DBSHA256 != "" {
			ev.builtSHA256 = "sha256:" + m.DBSHA256
		}
	}
	if rendered := j.Steps[landStepRendered]; strings.HasPrefix(rendered, "sha256:") {
		ev.configSHA256 = append(ev.configSHA256, rendered)
	}
	if raw, ok := j.Steps[landStepPreRender]; ok {
		// The hash the render journalled before it wrote (fix round 1, I1):
		// a crash after the write and before either the ack or the
		// rendered record still leaves the file the op's by content.
		var pre preRender
		if err := json.Unmarshal([]byte(raw), &pre); err != nil {
			log.Printf("moves: %s: the journal's %s step is unreadable (%v); it vouches for no rendered config", host, landStepPreRender, err)
		} else {
			for _, h := range pre.Intended {
				if strings.HasPrefix(h, "sha256:") {
					ev.configSHA256 = append(ev.configSHA256, h)
				}
			}
		}
	}
	if raw, ok := j.Steps[landStepPreInstall]; ok {
		var pre preInstall
		if err := json.Unmarshal([]byte(raw), &pre); err != nil {
			log.Printf("moves: %s: the journal's %s step is unreadable (%v); the database's siblings are not counted as %s's", host, landStepPreInstall, err, id)
		} else {
			ev.absentBefore = map[string]bool{}
			for _, p := range pre.Absent {
				ev.absentBefore[p] = true
			}
		}
	}
	body, _, err := r.s.Get(ctx, appconfig.ConfigAckKey(cfg.Box, host))
	switch {
	case errors.Is(err, os.ErrNotExist):
	case err != nil:
		return ev, fmt.Errorf("read the held render's config ack: %w", err)
	default:
		var a appconfig.ConfigAck
		if err := json.Unmarshal(body, &a); err != nil {
			log.Printf("moves: %s: the config ack is unreadable (%v); the rendered config is not counted as %s's", host, err, id)
		} else if a.MoveOp == id && a.Box == cfg.Box && a.Host == host && a.FileSHA256 != "" {
			ev.configSHA256 = append(ev.configSHA256, "sha256:"+a.FileSHA256)
		}
	}
	return ev, nil
}

// owns answers localArtifactPark.Owns for the target's leftovers.
func (ev rollbackOwnership) owns(path, identity string) (bool, error) {
	db := ev.host + ".db"
	switch path {
	case ev.host:
		return ev.placed.Files != "" && identity == ev.placed.Files, nil
	case ev.host + ".env":
		return ev.placed.Env != "" && identity == ev.placed.Env, nil
	case ev.host + ".config.env":
		if len(ev.configSHA256) == 0 {
			return false, nil
		}
		full := filepath.Join(ev.dataDir, path)
		if fi, err := os.Lstat(full); err != nil || !fi.Mode().IsRegular() {
			return false, nil // a link or anything else is not the render's file
		}
		got, err := fileSHA256(full)
		if err != nil {
			return false, err
		}
		return containsString(ev.configSHA256, got), nil
	case db, db + ".gen":
		if ev.installed {
			return true, nil
		}
		return ev.isBuild(path)
	case db + "-wal", db + "-shm":
		return ev.installed, nil
	case db + ".corrupt", ".activator-" + db + "-litestream":
		return ev.installed && ev.absentBefore[path], nil
	}
	return false, nil
}

// isBuild reports whether path, the database or its marker at the flat
// paths, is this op's build before the record stamped it installed: the
// database's sha256 is the journalled build's, and the marker names the
// op's generation. A link or anything else is not.
func (ev rollbackOwnership) isBuild(path string) (bool, error) {
	if ev.builtSHA256 == "" {
		return false, nil
	}
	full := filepath.Join(ev.dataDir, path)
	if fi, err := os.Lstat(full); err != nil || !fi.Mode().IsRegular() {
		return false, nil
	}
	if strings.HasSuffix(path, ".gen") {
		gen, err := hostbox.ReadGenerationMarker(full)
		if err != nil {
			return false, nil
		}
		return gen == ev.generation, nil
	}
	got, err := fileSHA256(full)
	if err != nil {
		return false, err
	}
	return got == ev.builtSHA256, nil
}

// deleteRow deletes the row and its timers rows in one registry transaction
// (registry.DeleteIfHoldWithTimers, J12), only while the rollback holds the
// row, under the host's hold lock with the record re-read there; the journal
// says "pending" before and "done" after, and a replay that finds "pending"
// and no row of this rollback's treats the delete as done.
func (r *rollbackRun) deleteRow(ctx context.Context, j *moves.Journal, mark func(step, value string) error) error {
	if j.Steps[rbStepRowDeleted] == "done" {
		return nil
	}
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	var err error
	provision.WithHostHoldLock(host, func() {
		fresh, e := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host)
		if e != nil {
			err = fmt.Errorf("re-read the record: %w", e)
			return
		}
		if m := fresh.Move; m == nil || m.ID != id || m.Incarnation != r.g.Incarnation || m.State != provision.MoveRollingBack ||
			m.Rollback == nil || m.Rollback.ID != r.rb || m.Rollback.TargetPhase != provision.RbTargetCleaning {
			err = fmt.Errorf("move %s: the record is no longer this rollback's cleaning", id)
			return
		}
		if j.Steps[rbStepRowDeleted] != "pending" {
			j.Mark(rbStepRowDeleted, "pending")
			if err = j.Commit(cfg.DataDir); err != nil {
				return
			}
			if err = r.fault("after-rb-row-pending"); err != nil {
				return
			}
		}
		deleted, e := cfg.DeleteHeld(host, r.rb)
		if e != nil {
			err = fmt.Errorf("delete the target's row: %w", e)
			return
		}
		if deleted {
			log.Printf("moves: %s: the target's row and its timers are deleted", host)
		}
		if err = r.fault("after-rb-row-delete"); err != nil {
			return
		}
		err = mark(rbStepRowDeleted, "done")
	})
	return err
}

// cleaned parks the op's journal into the local retirement once the record
// says cleaned, if a crash or a quick close left it.
func (r *rollbackRun) cleaned() error {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if !moveJournalPresent(cfg.DataDir, host, id) {
		return nil
	}
	j, err := r.journal()
	if err != nil {
		return err
	}
	if j.Steps[rbStepCleaned] != "done" {
		j.Mark(rbStepCleaned, "done")
		if err := j.Commit(cfg.DataDir); err != nil {
			return err
		}
	}
	return parkRollbackJournal(cfg.DataDir, host, id, j)
}

// parkRollbackJournal parks the target's op journal (its steps, the landing's
// build and extract scratch, and the built files archive of a dirty return)
// into the retirement its own moves-owned rename recorded.
func parkRollbackJournal(dataDir, host, op string, j *moves.Journal) error {
	retirement := j.Steps[rbStepLocal]
	if retirement == "" {
		return fmt.Errorf("move %s: the journal records no local retirement for the target's park", op)
	}
	plan, err := journalledLocalPlan(j, rbStepLocal)
	if err != nil {
		return err
	}
	if err := parkOpJournal(dataDir, plan.Incarnation, host, op, retirement); err != nil {
		return fmt.Errorf("park the op journal: %w", err)
	}
	log.Printf("moves: %s: the target's half of the rollback of %s is complete", host, op)
	return nil
}

// sweepRollbackJournal is the local journal sweep's rollback half
// (sweepFinalJournals, fix round 1, ruling 3): a target's journal whose
// cleaning reached its moves-owned rename (rbStepLocal) is parked once its
// completion is shown, independent of any record that still routes here:
//   - the journal records cleaned;
//   - a listed record carries the op and its rollback shows target_parked_at
//     (the cleaned CAS landed);
//   - no listed record carries the op: the only close of a rolling_back move,
//     provision.MoveRolledBackClose, refuses unless target_parked_at is
//     stamped, and a journal at rbStepLocal is past every declared-loss
//     window, so the cleaned CAS landed before the close.
//
// A listed record that carries the op without target_parked_at is the
// cleaning's own pass, and a host the listing does not name is left. A journal
// that never reached rbStepLocal is never touched.
func (w *movesWorker) sweepRollbackJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error {
	if _, ok := j.Steps[rbStepLocal]; !ok {
		return nil
	}
	if j.Steps[rbStepCleaned] != "done" {
		if len(recs) == 0 {
			w.notice(host+"#rb-sweep-"+op, "moves: %s: the target's journal of %s reached its park, 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 == op && (m.Rollback == nil || m.Rollback.TargetParkedAt == nil) {
				return nil
			}
		}
		j.Mark(rbStepCleaned, "done")
		if err := j.Commit(dataDir); err != nil {
			return err
		}
	}
	return parkRollbackJournal(dataDir, host, op, j)
}