package host

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"log"
	"reflect"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The revert on the source (moves spec §7.4 reverting; Plan 4 Task 6),
// behind CARLOS_MOVES_PHASE1: the pre-flip recovery of a cut abandoned before
// the flip. The record decided it (MoveRevert); the source takes the host
// back with its own evidence and runs a recovery journalled in the claim's
// rev_phase. revertSource is phase1MoveRound's MoveReverting arm (ruling J7),
// and movesPass reaches a reverting record through the participant rule
// (movesParticipantRecord, J10) as Tasks 4 and 5 do.
//
// Entry. A revert from stopping (or from stopped before execution was given
// up) finds the claim still moving with this box executing: the stop
// evidence was never obtained, so something may still run. The source holds
// the row moving and fences it again (an agent restart dropped the fence),
// drains the timers and asks for the evidence (§5.3), and starts nothing
// until it holds. Then the cut's own giveUpExecution publishes the complete
// stopped-source.json and gives execution up citing it (it records no
// baselines on a reverting record, J8), and Plan 2's revertClaim, unchanged,
// writes the claim half with its journalled revert_bound: reverting under the
// source, rev_phase evidence, the lane's park queued in the target region.
// No second staging scope is ever computed (J7). A revert decided after
// execution was given up starts at revertClaim.
//
// The recovery, from durable state, each step under the claim's rev_phase:
//
//   - evidence: the evidence holds (Revert cites it). The row's hold goes
//     from moving to restoring under the move id, a conditional status change
//     under the host's hold lock with the record read fresh (RestatusHold;
//     a revert never hands to a rollback id). Then rev_phase steps to verify
//     when the record keeps the stopped baselines (Source), else straight to
//     probing (J8), op.phase following from probing on (J1).
//   - verify: the files plain digest and the timers digest equal what the
//     source recorded at stopped (Source.Files.Plain, Source.TimersSHA256).
//     A mismatch stops for an operator with the baseline-mismatch sentence
//     (MoveMarkStuck); a retry returns to verify and checks again (INV-4).
//     Passed, rev_phase and op.phase step to probing before the first attempt.
//   - probing: the recovery probe only (recoveryProbe, Task 9's; warm or cold
//     into the source's own generation, holder.generation, which a revert
//     never changed). It records the inputs it renders before it renders them
//     (MoveRecordRevertInputs). Never verify again: the probe's process may
//     have written, and its writes are the authoritative state (INV-2). The
//     probe has the recovery probe's nine attempts (§7.7), and the first
//     attempt of a window is journalled: nine failed attempts
//     (StuckRecoveryProbeAttempts) or fifteen minutes of failed probes
//     (StuckRevertProbeBudget) stop for an operator. The window's end is
//     journalled before the stuck mark, so a crash or a refused write between
//     them still ends in the mark, never in a new window; only a retry starts
//     one. Answered, the claim steps to probed.
//   - probed: reopened_at on the record (MoveRevertReopened, J26), then the
//     conditional open on hold_op = move id under the host's hold lock with
//     the record read fresh (ReleaseHeld, a plain registry call, J12), then
//     the claim steps to opened.
//   - opened: the claim waits for the lane's park the revert queued in the
//     target region (§7.11), which the target region's box runs
//     (laneParkTarget) and the park engine acknowledges after its horizon.
//     Then MoveRevertClose (LastMove rolled_back, offline from cut_at to
//     the reopen), and then ReleaseMove to serving under this box: record
//     first, ruling M4; a crash between is finished by the console's
//     recovery pass from LastMove. The source's journal of the op is parked
//     in the same pass; a crash before that leaves it to the local journal
//     sweep (sweepRevertJournal).
//
// A claim this move already released (RequireReleasedMove: a binary that
// closed claim first crashed before the record's close) only closes.
//
// The target region's half (laneParkTarget). A revert never flipped, so its
// target box has neither the row nor a journal of the move; the participant
// rule answers it as the target on the state alone, and its one piece of work
// is the lane's park-replica the claim owes its region: the park engine over
// its own bucket archives moves/<host>/<op>/ under
// parked/<inc>/<host>/<ret>/moves/ and acknowledges once the horizon has
// passed (laneParkTarget dispatches an abort's the same way, Plan 5 Task 1).
//
// Every record write goes through a provision transition (the record_seq
// rule); every claim write through a hostclaim transition.
//
// Arming (rulings J18, G17). Like the landing and the rollback, the revert is
// unarmed in production until Plan 5: the recovery probe is nil there until
// Task 9 (claim Plan 10's recovery admission), so revertSourceArmed answers
// false and the source only writes the claim half, as Plan 2's arm did; and
// the target's lane park needs the claim's park leases, view and authorizer,
// which armMovesParks sets only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b; revertLaneArmed). LogEvent is nil in production,
// where an event is a log line.

// revertStepsPerPass bounds the steps one pass runs: evidence to the close is
// seven.
const revertStepsPerPass = 10

// revertProbeBudget is how long a revert's probes may fail before it stops
// for an operator (§7.4: "Fifteen minutes of failed probes: stuck").
const revertProbeBudget = 15 * time.Minute

// The revert's steps in the source's op journal (the cut's, under the move
// id).
const (
	// rvStepSource marks the journal as a revert's once the claim's recovery
	// begins (revertJournalMark), so the sweep can tell it from a cut's that
	// is still running, and knows the incarnation to park it under.
	rvStepSource = "rv-source"
	// rvStepProbeSince is the first probe attempt of this window of probing
	// (RFC 3339), the start of the fifteen minutes. Cleared, with
	// rvStepProbeExhausted, only once a retry has returned the record from
	// stuck, so a retry starts a new window and nothing else does.
	rvStepProbeSince = "rv-probe-since"
	// rvStepProbeExhausted is the window's end, journalled before the stuck
	// mark (probeExhaustion): the probe ran out of its nine attempts, or of
	// its fifteen minutes. A replay finds it and marks the move stuck without
	// probing again (Task 6 fix round 1).
	rvStepProbeExhausted = "rv-probe-exhausted"
	// rvStepClosed is "done" once the record closed.
	rvStepClosed = "rv-source-closed"
	// rvStepJournalPark is parkLocalArtifacts' step: the retirement the
	// journal is parked into (no artifacts of its own).
	rvStepJournalPark = "rv-source-journal-park"
)

// probeExhaustion is rvStepProbeExhausted's value: the recorded reason the
// window ended with, and the record's updated_at when it ended. SetRecordMove
// stamps updated_at on every transition, and once the window has ended the
// only transitions a reverting record takes are the stuck mark and then the
// operator's retry (this box records no probe inputs and no reopen after it,
// the door's abort refuses a reverting record, and MoveRevertClose needs the
// reopen). So a record still reverting at that updated_at has not been marked
// stuck, and one reverting at a later one has been retried.
type probeExhaustion struct {
	Reason    string    `json:"reason"`
	UpdatedAt time.Time `json:"updated_at"`
}

// revertJournalMark is rvStepSource's value.
type revertJournalMark struct {
	Move        string `json:"move"`
	Incarnation string `json:"incarnation"`
}

// revertSourceArmed reports whether this box can run the revert's recovery:
// every hook it calls is wired, the recovery probe included. In production
// the probe is not (Task 9, J18), so this answers false there.
func revertSourceArmed(cfg moveCfg) bool {
	return cfg.Box != "" && cfg.DataDir != "" && cfg.RestatusHold != nil && cfg.ReleaseHeld != nil &&
		cfg.Timers != nil && cfg.FilesDir != nil && cfg.recoveryProbe != nil
}

// revertLaneArmed reports whether this box can run the target region's lane
// park: an engine region (parkRegion, the home's "" included as
// park.HomeRegion; a code that is no placement is not one), the claim's
// park leases, view and authorizer, and this box's own bucket. In
// production armMovesParks sets the three only once the wake-source gate
// opens (movesWakeSourceRegistered, Plan 5 Task 9b), so this
// answers false there until it lands.
func revertLaneArmed(cfg moveCfg) bool {
	return cfg.Box != "" && cfg.parkRegion() != "" && cfg.OwnBucket != "" && parkHooksArmed(cfg) && cfg.Versioned != nil
}

// restatusHold is moveCfg.RestatusHold in production: an op's own hold
// changes status (a revert's moving to restoring under the move id) by
// registry.SetHoldIf expecting the op itself, under the host's hold lock,
// after the caller's validation; written (holdStageRequester) runs when the
// change landed, so root re-judges what it staged. A row already held status
// by op is a replay and answers true; a row held by another op, unheld or
// absent answers false and is not touched.
func restatusHold(reg *registry.Registry, written func(host string)) func(host, status, op string, validate func() error) (bool, error) {
	return func(host, status, op string, validate func() error) (bool, error) {
		var held bool
		var err error
		provision.WithHostHoldLock(host, func() {
			var rt *registry.Route
			if rt, err = reg.Get(host); err != nil || rt == nil {
				return
			}
			if validate != nil {
				if err = validate(); err != nil {
					return
				}
			}
			if op == "" || rt.HoldOp != op {
				return
			}
			if rt.Status == status {
				held = true
				return
			}
			if held, err = reg.SetHoldIf(host, status, op, op); err != nil || !held {
				return
			}
			if written != nil {
				written(host)
			}
		})
		return held, err
	}
}

// revertRun is one pass's state for one reverting record on its source.
type revertRun struct {
	w     *movesWorker
	s     release.Store
	rec   provision.Record
	g     provision.MoveGuard
	fleet *moveFleet // nil when the pass could not read it: the entry waits
}

// revertSource is phase1MoveRound's MoveReverting arm (J7) and the source's
// participant step for a reverting record (movesPass).
func (w *movesWorker) revertSource(ctx context.Context, s release.Store, rec provision.Record, fleet *moveFleet) error {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveReverting || m.From != cfg.Region {
		return nil
	}
	if !revertSourceArmed(cfg) {
		// The record decided; the claim follows it (§7.5), as Plan 2's arm
		// did. The recovery waits for Task 9's probe.
		if err := w.revertClaim(ctx, s, rec); err != nil {
			return err
		}
		w.notice(rec.Host+"#revert", "moves: %s: the revert of %s waits: this box's revert is not armed (the recovery probe is Task 9's, on claim Plan 10's recovery admission)", rec.Host, m.ID)
		return nil
	}
	r := &revertRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, fleet: fleet}
	for i := 0; i < revertStepsPerPass; i++ {
		advanced, err := r.step(ctx)
		if err != nil || !advanced {
			return err
		}
		fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
		if err != nil {
			return fmt.Errorf("re-read the record: %w", err)
		}
		if !r.same(fresh) {
			return nil
		}
		r.rec = *fresh
	}
	return nil
}

func (r *revertRun) cfg() moveCfg   { return r.w.cfg }
func (r *revertRun) now() time.Time { return r.w.now() }

// same reports whether fresh is still this move, reverting.
func (r *revertRun) same(fresh *provision.Record) bool {
	m := fresh.Move
	return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveReverting
}

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

// stillReverting is the record check run under the host's hold lock before
// a row write: the record, read fresh, must still be this move reverting.
func (r *revertRun) stillReverting(ctx context.Context) error {
	fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host)
	if err != nil {
		return fmt.Errorf("re-read the record: %w", err)
	}
	if !r.same(fresh) {
		return fmt.Errorf("move %s: the record is no longer reverting", r.g.ID)
	}
	return nil
}

// update writes one claim transition over obs.
func (r *revertRun) update(ctx context.Context, obs hostclaim.Observed, what string, step func(*hostclaim.Claim) error) error {
	after, err := hostclaim.Update(ctx, r.s, obs, step)
	if err != nil {
		return fmt.Errorf("%s: %w", what, err)
	}
	r.w.logRevPhase(ctx, r.s, r.rec, obs, after)
	return nil
}

// record writes one record transition. A conflict is someone else's
// decision: logged, and the pass ends without error.
func (r *revertRun) record(ctx context.Context, what string, mut func(*provision.Record) error) (bool, error) {
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host, observedMove(&pre, mut))
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: %s of %s refused (%v); the record moved on", r.rec.Host, what, r.g.ID, err)
			return false, nil
		}
		return false, fmt.Errorf("%s: %w", what, err)
	}
	r.w.logTransition(ctx, r.s, pre, *after)
	return true, nil
}

// step runs the revert's current step once, from the claim. advanced is true
// when it wrote the claim or the record.
func (r *revertRun) step(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	obs, err := hostclaim.Read(ctx, r.s, host)
	if err != nil {
		return false, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		return false, fmt.Errorf("move %s has no host claim", id)
	}
	c := obs.Claim
	if hostclaim.RequireReleasedMove(id)(c) == nil {
		// A binary that closed claim first crashed between the claim's
		// release and the record's close: only the record's half is left.
		return r.close(ctx, false)
	}
	if c.Op.ID != id {
		return false, fmt.Errorf("move %s: the claim names op %q", id, c.Op.ID)
	}
	if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation {
		return false, fmt.Errorf("move %s: the host claim names another incarnation", id)
	}
	if len(c.Op.Participants) == 0 || c.Op.Participants[0] != cfg.Box {
		return false, fmt.Errorf("move %s: this box is not the move's source in the claim", id)
	}
	switch c.Op.Kind {
	case hostclaim.KindMoving:
		if c.Op.Exec == cfg.Box {
			return r.evidence(ctx, obs)
		}
		// Execution given up with the source's evidence: the claim half.
		if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil {
			return false, err
		}
		fresh, err := hostclaim.Read(ctx, r.s, host)
		if err != nil {
			return false, fmt.Errorf("read the host claim: %w", err)
		}
		if fresh.Absent || fresh.Claim.Op.Kind != hostclaim.KindReverting {
			r.w.notice(host+"#revert", "moves: %s: the revert of %s waits: the claim is moving with executor %q", host, id, c.Op.Exec)
			return false, nil
		}
		return true, nil
	case hostclaim.KindReverting:
	default:
		return false, fmt.Errorf("move %s: the claim is %s, not reverting", id, c.Op.Kind)
	}
	if c.Op.Exec != cfg.Box {
		return false, fmt.Errorf("move %s: the claim's executor is %q, not this box", id, c.Op.Exec)
	}
	switch phase := c.Op.Journal[hostclaim.JournalRevPhase]; phase {
	case hostclaim.RevPhaseEvidence:
		return r.begin(ctx, obs)
	case hostclaim.RevPhaseVerify:
		return r.verify(ctx, obs)
	case hostclaim.PhaseProbing:
		return r.probing(ctx, obs)
	case hostclaim.PhaseProbed:
		return r.probed(ctx, obs)
	case hostclaim.PhaseOpened:
		return r.opened(ctx, obs)
	default:
		return false, fmt.Errorf("move %s: the claim's rev_phase is %q", id, phase)
	}
}

// evidence is the entry from stopping (or from stopped before execution was
// given up): the hold, the fence and the drain again, the stop evidence until
// it holds, then the cut's giveUpExecution and Plan 2's revertClaim.
func (r *revertRun) evidence(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if err := hostclaim.RequireMoveOp(id, hostclaim.KindMoving, hostclaim.PhaseMoving)(obs.Claim); err != nil {
		return false, fmt.Errorf("move %s: %w", id, err)
	}
	if r.fleet == nil {
		return false, fmt.Errorf("move %s: the fleet's data buckets could not be read; the revert's evidence waits", id)
	}
	c := &cutRun{w: r.w, s: r.s, fleet: *r.fleet, rec: r.rec, g: r.g}
	defer c.releaseLease()
	holder := *obs.Claim.Holder
	if err := c.holderSupported(holder); err != nil {
		return false, fmt.Errorf("move %s: %w", id, err)
	}
	if cfg.HoldMoving == nil || cfg.Fence == nil || cfg.TimersDrained == nil || cfg.Evidence == nil {
		return false, errors.New("this box cannot stop an instance for a revert: a hook is not wired")
	}
	if err := cfg.HoldMoving(host, id); err != nil {
		return false, fmt.Errorf("hold the row: %w", err)
	}
	gen := moveFenceGeneration(r.rec.Move)
	if err := cfg.Fence(ctx, host, gen); err != nil {
		return false, fmt.Errorf("fence against wakes: %w", err)
	}
	r.w.fenced.Store(host, gen)
	if err := cfg.TimersDrained(ctx, host); err != nil {
		return false, fmt.Errorf("drain the timers: %w", err)
	}
	j, err := c.journal()
	if err != nil {
		return false, err
	}
	var ev moves.StoppedReport
	if raw, ok := j.Steps[cutStepEvidence]; !ok || json.Unmarshal([]byte(raw), &ev) != nil || !ev.TreeStopped() {
		rep, err := cfg.Evidence(ctx, host, holder.Location, []string{c.dbPath()})
		if err != nil {
			return false, fmt.Errorf("stop evidence: %w", err)
		}
		if !rep.TreeStopped() {
			r.w.notice(host+"#rv-evidence", "moves: %s: the revert of %s starts nothing until the instance is proven stopped (%s)", host, id, failedSteps(rep))
			return false, nil
		}
		body, err := json.Marshal(rep)
		if err != nil {
			return false, err
		}
		j.Mark(cutStepEvidence, string(body))
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
		if err := r.fault("after-revert-evidence"); err != nil {
			return false, err
		}
	}
	stoppedKey := moves.StoppedKey(moves.ReportPrefix(r.rec.Account, r.rec.App, host, id), "source")
	report, err := c.giveUpExecution(ctx, obs, stoppedKey)
	if err != nil {
		return false, err
	}
	if report.V == 0 {
		return false, nil // a file changed under the digest: the next pass
	}
	if err := r.w.revertClaim(ctx, r.s, r.rec); err != nil {
		return false, err
	}
	if err := r.fault("after-revert-claim"); err != nil {
		return false, err
	}
	return true, nil
}

// begin is rev_phase evidence on a reverting claim: the journal marked as
// this revert's, the row's hold restoring under the move id, then verify or
// probing.
func (r *revertRun) begin(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if obs.Claim.Op.Journal[hostclaim.JournalEvidenceSource] == "" {
		return false, fmt.Errorf("move %s: the revert cites no stop evidence", id)
	}
	if err := r.markJournal(); err != nil {
		return false, err
	}
	held, err := cfg.RestatusHold(host, "restoring", id, func() error { return r.stillReverting(ctx) })
	if err != nil {
		return false, fmt.Errorf("hold the row restoring: %w", err)
	}
	if !held {
		return false, fmt.Errorf("move %s: this box's row is not held by the move", id)
	}
	if err := r.fault("after-restatus"); err != nil {
		return false, err
	}
	baseline := r.rec.Move.Source != nil
	to := hostclaim.PhaseProbing
	if baseline {
		to = hostclaim.RevPhaseVerify
	}
	if err := r.update(ctx, obs, "the revert's "+to, hostclaim.SetRevPhase(id, hostclaim.RevPhaseEvidence, to, baseline)); err != nil {
		return false, err
	}
	log.Printf("moves: %s: the revert of %s is at %s", host, id, to)
	if err := r.fault("after-rev-" + to); err != nil {
		return false, err
	}
	return true, nil
}

// verify (§7.4 verify, INV-4): the files and the timers against the digests
// recorded at stopped. A mismatch stops for an operator; a retry runs this
// again.
func (r *revertRun) verify(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := r.cfg()
	host, id, m := r.rec.Host, r.g.ID, r.rec.Move
	if err := r.fault("verify"); err != nil {
		return false, err
	}
	if obs.Claim.Op.Phase != hostclaim.PhaseMoving && obs.Claim.Op.Phase != hostclaim.PhaseSealed {
		return false, fmt.Errorf("move %s: op.phase %q at rev_phase verify", id, obs.Claim.Op.Phase)
	}
	if m.Source == nil || m.Source.Files == nil || m.Source.TimersSHA256 == "" {
		return false, fmt.Errorf("move %s: the record has no stopped baselines to verify against", id)
	}
	dir, envPath := cfg.FilesDir(host)
	plain, _, _, err := archivePlainDigest(dir, envPath)
	why, retry, ferr := archiveFailure(host, err)
	switch {
	case ferr != nil:
		return false, fmt.Errorf("digest the source's files: %w", ferr)
	case retry:
		r.w.notice(host+"#rv-verify", "moves: %s: a file changed while the source's files were digested; the verification runs again", host)
		return false, nil
	case why != "":
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the source's files cannot be digested: %s", why))
	case plain != m.Source.Files.Plain:
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the files are %s, recorded %s", plain, m.Source.Files.Plain))
	}
	rows, err := cfg.Timers(host)
	if err != nil {
		return false, fmt.Errorf("read the timers: %w", err)
	}
	if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 {
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256))
	}
	log.Printf("moves: %s: the source still matches what it recorded when it stopped", host)
	if err := r.update(ctx, obs, "the revert's probing", hostclaim.SetRevPhase(id, hostclaim.RevPhaseVerify, hostclaim.PhaseProbing, true)); err != nil {
		return false, err
	}
	if err := r.fault("after-rev-probing"); err != nil {
		return false, err
	}
	return true, nil
}

// probing (§7.4 probing): the probe only, bounded twice (§7.7): its nine
// attempts and fifteen minutes from its first attempt. Nothing earlier runs
// again from here. The window's end is journalled before the stuck mark, so
// a crash or a refused record write between the two marks the move stuck on
// the next pass without probing again; a new window starts only once a retry
// has returned the record from stuck.
func (r *revertRun) probing(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if obs.Claim.Op.Phase != hostclaim.PhaseProbing {
		return false, fmt.Errorf("move %s: op.phase %q at rev_phase probing", id, obs.Claim.Op.Phase)
	}
	j, err := moves.ReadJournal(cfg.DataDir, host, id)
	if err != nil {
		return false, err
	}
	if raw, ok := j.Steps[rvStepProbeExhausted]; ok {
		var ex probeExhaustion
		if err := json.Unmarshal([]byte(raw), &ex); err != nil {
			return false, fmt.Errorf("move %s: the journalled end of the probe's window does not read: %w", id, err)
		}
		if r.rec.Move.UpdatedAt.Equal(ex.UpdatedAt) {
			// The window ended and the stuck mark never landed.
			return r.stuck(ctx, ex.Reason, errors.New("the probe's window ended before this pass"))
		}
		// Marked stuck and retried since: a new window.
		j.Unmark(rvStepProbeExhausted)
		j.Unmark(rvStepProbeSince)
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
		log.Printf("moves: %s: the revert of %s was retried; its probe has a new window", host, id)
	}
	since, perr := time.Parse(time.RFC3339Nano, j.Steps[rvStepProbeSince])
	if perr != nil {
		since = r.now().UTC()
		j.Mark(rvStepProbeSince, since.Format(time.RFC3339Nano))
		if err := j.Commit(cfg.DataDir); err != nil {
			return false, err
		}
	}
	// A revert has no rollback: it probes the source's own generation, warm
	// or cold as for serving (§7.14 reverting row), with no recovery
	// coverage, and the recovery probe's nine attempts. The probe records
	// what it renders before it renders it.
	req := recoveryProbeRequest{Clean: true, SourceGeneration: obs.Claim.Holder.Generation, Attempts: recoveryProbeAttempts,
		Record: func(ctx context.Context, in provision.MoveInputs) error {
			_, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRevertInputs(r.g, in))
			return err
		}}
	out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req)
	var missing *ClaimDependencyError
	if errors.As(err, &missing) {
		return false, nil
	}
	if err != nil {
		log.Printf("moves: %s: the revert's probe of %s: %v", host, id, err)
		out = recoveryProbeWaiting
	}
	if out == recoveryProbeAnswered {
		// The probe's wake wrote the claim (its admission, then its
		// restore's re-tag), so the step's own read is stale: read it
		// again, as the rollback's probed does. The fresh claim is taken
		// only when it is the step's read plus those writes and nothing
		// else (revertProbeWrites, census-ruling-home-part2's condition);
		// any other change is refused as the stale CAS was.
		fresh, err := hostclaim.Read(ctx, r.s, host)
		if err != nil {
			return false, fmt.Errorf("re-read the host claim: %w", err)
		}
		if err := revertProbeWrites(cfg.Box, cfg.DataDir, obs, fresh); err != nil {
			return false, fmt.Errorf("the revert's probed: %w", err)
		}
		obs = fresh
		if err := r.update(ctx, obs, "the revert's probed", hostclaim.SetRevPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed, r.rec.Move.Source != nil)); err != nil {
			return false, err
		}
		log.Printf("moves: %s: the source answered its probe; the revert of %s is probed", host, id)
		if err := r.fault("after-rev-probed"); err != nil {
			return false, err
		}
		return true, nil
	}
	var reason string
	var why error
	elapsed := r.now().Sub(since)
	switch {
	case out == recoveryProbeExhausted:
		reason, why = provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts)
	case elapsed >= revertProbeBudget:
		reason, why = provision.StuckRevertProbeBudget, fmt.Errorf("no answer for %s", elapsed.Round(time.Second))
	default:
		r.w.notice(host+"#rv-probe", "moves: %s: the revert's probe of %s has not answered yet", host, id)
		return false, nil
	}
	// The window's end first, against the record as it stands now (the
	// probe may have recorded its inputs this pass), then the stuck mark.
	fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, host)
	if err != nil {
		return false, fmt.Errorf("re-read the record: %w", err)
	}
	if !r.same(fresh) {
		return false, nil
	}
	r.rec = *fresh
	body, err := json.Marshal(probeExhaustion{Reason: reason, UpdatedAt: fresh.Move.UpdatedAt})
	if err != nil {
		return false, err
	}
	// The probe wrote its attempt count into the same journal.
	j, err = moves.ReadJournal(cfg.DataDir, host, id)
	if err != nil {
		return false, err
	}
	j.Mark(rvStepProbeExhausted, string(body))
	if err := j.Commit(cfg.DataDir); err != nil {
		return false, err
	}
	if err := r.fault("after-probe-exhausted"); err != nil {
		return false, err
	}
	return r.stuck(ctx, reason, why)
}

// revertProbeWrites is census-ruling-home-part2's condition on the revert's
// re-read after its probe answers: fresh must be earlier, the step's own
// read, with only the probe wake's own claim writes over it. Those are the
// claim's own transitions, in order: hostclaim.AdmitWriter's cold admission
// of a new generation for box, this op's source writer (with the origin its
// CAS records), then, when the admitted stamp is new, the restore's re-tag
// of box's database paths, checked by FS-R11's claimRestoreObservation.retag.
// Each is replayed over earlier and the result must equal fresh exactly, as
// FS-R11's advance compares, so the op, its phase and journal, the holder's
// identity, every chore, fence, quiet key and every other box's inventory
// stand as the step read them, and the admission must be the first write
// after that read (its session is the claim gen it was made at). A claim no
// write touched passes. The probe's attempt reservation is not a claim
// write here: a reverting op's attempts are journalled on the box
// (movesProbeAttempts), and hostclaim.ReserveWakeRetryAttempt refuses a
// claim that is not serving. Anything else is hostclaim.ErrConflict.
func revertProbeWrites(box, dataDir string, earlier, fresh hostclaim.Observed) error {
	if sameClaimObservation(earlier, fresh) {
		return nil
	}
	conflict := func(why string) error {
		return fmt.Errorf("%w: the host claim changed under the probe other than by its own writes: %s", hostclaim.ErrConflict, why)
	}
	if earlier.Absent || fresh.Absent || earlier.Host != fresh.Host || earlier.ETag == "" || earlier.Claim == nil || fresh.Claim == nil ||
		earlier.Claim.Holder == nil || fresh.Claim.Holder == nil {
		return conflict("it is not the claim the step read")
	}
	e, f := earlier.Claim, fresh.Claim
	n := len(f.Holder.Admissions)
	if n == 0 {
		return conflict("no admission")
	}
	last := f.Holder.Admissions[n-1]
	if last.LossID != "" {
		// A loss-bound admission (as a fence-bound one, which AdmitWriter
		// refuses below) is verified only inside hostclaim, which does not
		// expose that; until a hostclaim-owned function re-verifies it
		// (amadan #300), the step stalls here as it did before the re-read.
		return conflict("a loss-bound admission, which only hostclaim can verify (amadan #300)")
	}
	if last.Box != box || last.Mode != "cold" || last.Generation == "" || last.Generation == e.Holder.Generation {
		return conflict("the last admission is not a new generation for this box's writer")
	}
	in := hostclaim.Admission{
		ObservedOpID: e.Op.ID, ObservedPhase: e.Op.Phase, ObservedSession: e.Holder.Session, ObservedGeneration: e.Holder.Generation,
		Box: box, Mode: "cold", NewGeneration: last.Generation, PredecessorCertified: last.WritesLost == nil, KeepStamp: f.Holder.Stamp != last.Session,
		Coverage: last.Coverage, Region: last.Region, PredecessorRegion: last.PredecessorRegion, AdditionalWritesLost: last.AdditionalWritesLost,
	}
	if last.WritesLost != nil {
		in.RemoteMaxTXID = last.WritesLost.RemoteMaxTXID
	}
	admitted, err := revertClaimCopy(e)
	if err != nil {
		return err
	}
	if err := hostclaim.AdmitWriter(in)(admitted); err != nil {
		if errors.Is(err, hostclaim.ErrFence) {
			return conflict("a fence-bound admission, which only hostclaim can verify (amadan #300)")
		}
		return conflict(err.Error())
	}
	admitted.Gen = e.Gen + 1
	// The admission's CAS moves the origin head by exactly one node, or
	// leaves it when the chain's last node is already this origin.
	if eh, fh := e.Origins, f.Origins; (eh == nil) != (fh == nil) || eh != nil && *eh != *fh {
		var count uint64
		if eh != nil {
			count = eh.Count
		}
		if fh == nil || fh.Count != count+1 {
			return conflict("the origin head moved by more than the admission")
		}
		head := *fh
		admitted.Origins = &head
	}
	// The copy drops the origin AdmitWriter staged for Update's CAS.
	if admitted, err = revertClaimCopy(admitted); err != nil {
		return err
	}
	switch f.Gen {
	case admitted.Gen:
		if !reflect.DeepEqual(admitted, f) {
			return conflict("the admission")
		}
	case admitted.Gen + 1:
		db := provision.DefaultDBPath(dataDir, earlier.Host)
		if e.Inventory[box].Layout == "nested" {
			db = provision.NestedDBPath(dataDir, earlier.Host)
		}
		mid := hostclaim.Observed{Host: earlier.Host, ETag: earlier.ETag + "+admitted", Claim: admitted}
		ob := &claimRestoreObservation{observed: mid, box: box, dataDir: dataDir, db: db}
		if err := ob.retag(mid, fresh); err != nil {
			return conflict("the restore's re-tag")
		}
	default:
		return conflict(fmt.Sprintf("gen %d to %d", e.Gen, f.Gen))
	}
	return nil
}

// revertClaimCopy is a deep copy of c through its JSON form, as the claim
// store decodes it.
func revertClaimCopy(c *hostclaim.Claim) (*hostclaim.Claim, error) {
	body, err := json.Marshal(c)
	if err != nil {
		return nil, err
	}
	var out hostclaim.Claim
	if err := json.Unmarshal(body, &out); err != nil {
		return nil, err
	}
	return &out, nil
}

// probed (§7.4 probed): reopened_at on the record (J26), the conditional open
// on hold_op = move id under the host's hold lock (J12), the claim opened.
func (r *revertRun) probed(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	if r.rec.Move.ReopenedAt == nil {
		ok, err := r.record(ctx, "the reopen", provision.MoveRevertReopened(r.g, r.now()))
		if !ok {
			return false, err
		}
		if err := r.fault("after-reopened"); err != nil {
			return false, err
		}
	}
	var err error
	provision.WithHostHoldLock(host, func() {
		if err = r.stillReverting(ctx); err != nil {
			return
		}
		var released bool
		if released, err = cfg.ReleaseHeld(host, id); err != nil || released {
			return
		}
		var rt *registry.Route
		if cfg.Route == nil {
			err = errors.New("this box cannot read its registry row")
			return
		}
		if rt, err = cfg.Route(host); err != nil {
			return
		}
		switch {
		case rt == nil:
			err = fmt.Errorf("move %s: this box has no row to open", id)
		case rt.HoldOp != "":
			err = fmt.Errorf("move %s: the row is held %s by %q, not by the move", id, rt.Status, rt.HoldOp)
		}
	})
	if err != nil {
		return false, fmt.Errorf("open the row: %w", err)
	}
	if err := r.fault("after-row-opened"); err != nil {
		return false, err
	}
	if err := r.update(ctx, obs, "the revert's opened", hostclaim.SetRevPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened, r.rec.Move.Source != nil)); err != nil {
		return false, err
	}
	log.Printf("moves: %s: the source serves again; the revert of %s waits for its lane's park", host, id)
	nudgeSerial(ctx, r.s)
	if err := r.fault("after-rev-opened"); err != nil {
		return false, err
	}
	return true, nil
}

// opened (§7.4 opened, §7.11): once no chore of the op is left in the claim
// (the lane's park acknowledged), the record's close, then the claim's
// release to serving under this box (record first, ruling M4).
func (r *revertRun) opened(ctx context.Context, obs hostclaim.Observed) (bool, error) {
	host, id := r.rec.Host, r.g.ID
	if r.rec.Move.ReopenedAt == nil {
		return false, fmt.Errorf("move %s: the claim is opened and the record has no reopen", id)
	}
	// Nothing owed under the op, checked before the record's close (M4), as
	// the console's grant checks it.
	if owed := owedUnder(obs.Claim, id); owed > 0 {
		r.w.notice(host+"#rv-opened", "moves: %s: the revert of %s waits for %d chore(s) of the move, its lane's park, to be acknowledged", host, id, owed)
		return false, nil
	}
	return r.close(ctx, true)
}

// close is the record's close (MoveRevertClose), then, when releaseClaim is set,
// the claim's release to serving under this box, then the source's journal
// of the op parked. A crash between the record's close and the release
// leaves the claim reverting/opened with nothing owed; the console's
// recovery pass releases it from LastMove. releaseClaim is false only for a
// claim a claim-first binary already released.
func (r *revertRun) close(ctx context.Context, releaseClaim bool) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	ok, err := r.record(ctx, "the revert's close", provision.MoveRevertClose(r.g, r.now()))
	if !ok {
		return false, err
	}
	log.Printf("moves: %s: move %s is reverted and closed", host, id)
	r.w.pageRolledBack(ctx, r.rec)
	nudgeSerial(ctx, r.s)
	if releaseClaim {
		if err := r.fault("after-close-record"); err != nil {
			return true, err
		}
		if err := releaseAfterClose(ctx, r.s, host, id, hostclaim.ReleaseMove(id, cfg.Box)); err != nil {
			return true, err
		}
		log.Printf("moves: %s: the claim is serving again after the revert of %s", host, id)
		if err := r.fault("after-release-move"); err != nil {
			return true, err
		}
	}
	if err := r.fault("after-close"); err != nil {
		return true, err
	}
	if err := parkRevertJournal(cfg.DataDir, host, id, true); err != nil {
		return true, fmt.Errorf("park the source's journal of %s: %w", id, err)
	}
	return true, nil
}

// stuck stops the revert for an operator (MoveMarkStuck) with a recorded
// reason; the row stays held and move --retry resumes the rev_phase.
func (r *revertRun) stuck(ctx context.Context, reason string, why error) (bool, error) {
	host := r.rec.Host
	log.Printf("moves: %s: the revert of %s stops for an operator (%v): %s", host, r.g.ID, why, reason)
	ok, err := r.record(ctx, "the stuck mark", provision.MoveMarkStuck(r.g, reason))
	if !ok {
		return false, err
	}
	r.w.pageStuck(ctx, r.rec, "source", reason)
	nudgeSerial(ctx, r.s)
	return true, nil
}

// markJournal marks the source's journal of the op as this revert's. A mark
// already there is kept.
func (r *revertRun) markJournal() error {
	dd, host := r.cfg().DataDir, r.rec.Host
	j, err := moves.ReadJournal(dd, host, r.g.ID)
	if err != nil {
		return err
	}
	if j.Done(rvStepSource) {
		return nil
	}
	body, err := json.Marshal(revertJournalMark{Move: r.g.ID, Incarnation: r.g.Incarnation})
	if err != nil {
		return err
	}
	j.Mark(rvStepSource, string(body))
	return j.Commit(dd)
}

// parkRevertJournal parks the source's journal of a closed revert: closed
// (true when the caller has seen the close) is journalled first, then a
// moves-owned retirement with no artifacts is made for it (its receipt names
// the op) and the journal, with the cut's archive inside it, is renamed into
// it. A journal that is not there was parked already; one with no revert mark
// is not this revert's and is left.
func parkRevertJournal(dataDir, host, op string, closed bool) error {
	if !moveJournalPresent(dataDir, host, op) {
		return nil
	}
	j, err := moves.ReadJournal(dataDir, host, op)
	if err != nil {
		return err
	}
	raw, ok := j.Steps[rvStepSource]
	if !ok {
		return nil
	}
	var mark revertJournalMark
	if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) {
		return fmt.Errorf("the journal of %s has an unreadable revert mark", op)
	}
	write := func(step, value string) error {
		j.Mark(step, value)
		return j.Commit(dataDir)
	}
	if closed && j.Steps[rvStepClosed] != "done" {
		if err := write(rvStepClosed, "done"); err != nil {
			return err
		}
	}
	retirement, err := parkLocalArtifacts(context.Background(), j, write, localArtifactPark{DataDir: dataDir, Incarnation: mark.Incarnation,
		Host: host, Op: op, Step: rvStepJournalPark, NewID: obligationID})
	if err != nil {
		return err
	}
	if err := parkOpJournal(dataDir, mark.Incarnation, host, op, retirement); err != nil {
		return err
	}
	log.Printf("moves: %s: the source's journal of the revert %s is parked", host, op)
	return nil
}

// sweepRevertJournal is the local journal sweep's revert half
// (sweepFinalJournal): a journal the revert marked (rvStepSource) is parked
// once the close is shown by durable evidence:
//   - the journal records the close (rvStepClosed "done");
//   - no listed record carries the move: a reverting or stuck move is
//     cleared only by provision.MoveRevertClose (MoveRetry returns a stuck
//     one to reverting; no abort or completion takes either state), and the
//     mark is written while the record is reverting, so the close landed.
//
// A host the listing does not name, or a record that still carries the move,
// is left.
func (w *movesWorker) sweepRevertJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error {
	var mark revertJournalMark
	if err := json.Unmarshal([]byte(j.Steps[rvStepSource]), &mark); err != nil || mark.Move == "" {
		return fmt.Errorf("the journal of %s has an unreadable revert mark", op)
	}
	if j.Steps[rvStepClosed] != "done" {
		if len(recs) == 0 {
			w.notice(host+"#rv-sweep-"+op, "moves: %s: the source's journal of %s is a revert's, and no listed record names the host; it is left", host, op)
			return nil
		}
		for _, rec := range recs {
			if m := rec.Move; m != nil && m.ID == mark.Move {
				return nil
			}
		}
	}
	return parkRevertJournal(dataDir, host, op, true)
}

// laneParkTarget is the target region's half of a move outcome that leaves
// the move's own staging lane behind in its bucket (movesPass through
// participantRound): the lane's park-replica the claim owes this region, run
// by the park engine over this box's own bucket until it is acknowledged
// after its horizon. It serves a revert (the claim reverting, any phase; the
// park hostclaim.Revert queued), an abort (moving/aborting; the park
// hostclaim.Abort queued, Plan 5 Task 1) and a committed landing
// (landing/committed; the park hostclaim.Commit queued, Plan 5 Task 3, §7.9
// step 4, called from the committed arm). No phases means any phase of
// kind. The view and the authorizer are the claim's (MoveParks), the
// authorizer behind this lane's own check: only this move's lane park under
// kind and its phases, with this box not executing the host unless the
// landing is committed (the target serves it then).
func (w *movesWorker) laneParkTarget(ctx context.Context, s release.Store, rec provision.Record, kind string, phases ...string) error {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.To != cfg.Region || m.From == cfg.Region {
		return nil
	}
	notice := rec.Host + "#lane-" + kind
	if kind == hostclaim.KindReverting {
		notice = rec.Host + "#rv-lane"
	}
	if !revertLaneArmed(cfg) {
		w.notice(notice, "moves: %s: the park of move %s's lane waits: this box's park engine 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.ID)
		return nil
	}
	host, id := rec.Host, m.ID
	inOp := func(op hostclaim.Op) bool {
		return op.ID == id && op.Kind == kind && (len(phases) == 0 || containsString(phases, op.Phase))
	}
	obs, err := hostclaim.Read(ctx, s, host)
	if err != nil {
		return fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent || !inOp(obs.Claim.Op) {
		return nil // the claim half has not followed yet, or it is over
	}
	if h := obs.Claim.Holder; h == nil || h.Incarnation != m.Incarnation {
		return fmt.Errorf("move %s: the host claim names another incarnation", id)
	}
	lane := moves.ScopeLocation(host, id)
	var chores []hostclaim.Obligation
	for _, o := range obs.Claim.Obligations {
		if o.Op == id && o.Kind == "park-replica" && park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) && o.Park != nil && o.Park.Location == lane {
			chores = append(chores, o)
		}
	}
	if len(chores) == 0 {
		return nil
	}
	vs, err := cfg.Versioned(ctx, cfg.OwnBucket, cfg.OwnBucketRegion)
	if err != nil {
		return err
	}
	committed := kind == hostclaim.KindLanding
	// The claim's view (movesParkView) names the target's region
	// authoritative after the commit, with its marker, and the source's
	// before it; the claim's authorizer runs behind this lane's own check.
	own := func(_ context.Context, obs hostclaim.Observed, o hostclaim.Obligation) error {
		c := obs.Claim
		if obs.Absent || c == nil || !inOp(c.Op) || (!committed && c.Op.Exec == cfg.Box) ||
			o.Op != id || o.Kind != "park-replica" || !park.PlacementMatches(cfg.ParkHomeRegion, cfg.Region, o.Region) || o.Park == nil || o.Park.Location != lane {
			return fmt.Errorf("%w: not this move's lane park", park.ErrUnsafe)
		}
		return nil
	}
	e := park.Engine{HomeRegion: cfg.ParkHomeRegion, Store: vs, Claims: s, Bucket: cfg.OwnBucket, Region: cfg.parkRegion(), Leases: cfg.ParkLeases,
		Now: w.now, NewID: obligationID, View: cfg.ParkView, Authorize: movesArmAuthorize(cfg.ParkAuthorize, own), After: cfg.parkFault}
	for _, o := range chores {
		err := e.Run(ctx, park.Work{Host: host, ID: o.ID, Op: id})
		switch {
		case errors.Is(err, park.ErrPending):
			w.notice(notice, "moves: %s: move %s's lane is archived; its park is acknowledged after its horizon", host, id)
		case errors.Is(err, park.ErrChanged):
		case err != nil:
			return fmt.Errorf("park move %s's lane: %w", id, err)
		default:
			log.Printf("moves: %s: move %s's lane in %s is parked", host, id, cfg.Region)
		}
	}
	return nil
}