package host

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

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

// The source's recovery after a rollback (moves spec §7.8 source table; Plan
// 4 Task 5), behind CARLOS_MOVES_PHASE1: the target has released, and the
// source box takes the host back, clean or dirty, through opened, and then
// the rolled_back finalization.
//
// Routing. MoveRollBack moved the record's region back to From, so the
// source reaches a rolling_back record through the post-flip participant
// rule (movesParticipantRecord, J10) in movesPass, whether or not it still
// has its row. One pass drives it as far as it can, each phase from durable
// state, under the record's source_phase. A record at X means X's work may
// have begun (Plan 2's convention, G8; provision's "each is written before
// its side effects"):
//
//   - waiting: until the target's evidence, released clean or returned
//     dirty on the record, and the claim restoring/released with nobody
//     executing ("" after the target's evidence, "none" after a
//     declaration). Then the record says claimed.
//   - claimed: the claim's return (hostclaim.ReturnToSource: this box
//     executing at the source's own stamp S; a clean return also names the
//     source's generation, src_gen, again), then the guard's activation
//     (the fresh read finds holder.stamp == S, matching the installation
//     record, or re-initialises a lost one to S), then, for a dirty return
//     whose inventory lists paths of S, the source's park-local (J13: the
//     installer's local retires S against it). The claim's phase is
//     claimed from here through the install (J1).
//   - restoring: the row handed from the move's moving hold to the
//     rollback's restoring hold (handHold, J3), with the record read fresh
//     under the host's hold lock.
//   - clean verified: the files plain digest and the timers digest equal
//     the source's stopped baselines (Source.Files.Plain,
//     Source.TimersSHA256), and the warm database's <dbPath>.gen names
//     src_gen (plain text, H11). A mismatch stops for an operator with the
//     baseline-mismatch sentence at verified (MoveMarkStuck, INV-4); a
//     retry verifies again, never skips it.
//   - dirty parking to configured: the shared installer (installGeneration,
//     Task 3) bound by recoveryInstallSource to the return seal and the
//     return baseline the target journalled in the claim (Task 4), from the
//     return lane moves/<host>/<rb id>/ in this box's own bucket, under S;
//     configured records the inputs it rendered as recovery_inputs, never
//     compared with move.inputs. restored hands over to probing.
//   - probing: the claim steps claimed to probing first (J1), then the
//     recovery probe: recoveryProbe, Task 9's (BLOCKED-ON-CLAIM C1, rulings
//     J2 and J18), a cold admission of kind recovery with nine attempts,
//     into the return generation (dirty, coverage sealed) or a fresh one
//     restored from src_gen (clean, coverage clean-return; J14: both
//     returns are cold and never keep S). The probe records the inputs it
//     renders before it renders them (MoveRecordRecoveryInputs, at
//     probing: recoveryProbeAdapter, movesrecoveryprobe.go, claim Plan 10
//     J2; a clean return records only here). Once the
//     record says probing nothing earlier runs again: a restarted probe
//     restarts only the probe (INV-2, INV-7). Nine failed attempts stop
//     for an operator at probing.
//   - probed: the claim steps to probed and, for a dirty return, queues the
//     return lane's park in this region in the same write
//     (QueueReturnLanePark, its horizon the return seal's bound); then
//     recovered_at, and the serial is nudged (the target's cleaning waits
//     on it).
//   - recovered: the open: the rollback's hold on the row ends
//     (ReleaseHeld, a plain registry call under the host's hold lock with
//     the record read fresh, J12), the claim steps to opened, and the record
//     says opened with opened_at.
//   - opened: the finalization. Once recovered_at, opened_at and
//     target_parked_at are stamped and no obligation of the op remains in
//     the claim, the terminal event (LogEvent, Plan 6's moves.LogEvent),
//     the record's close (MoveRolledBackClose) and then the claim's
//     release to serving (ReleaseMove): record first, ruling M4. Whichever
//     actor stamps last tries; a crash before the record's close is the
//     source's next pass, and one after it the console's recovery pass
//     (Plan 5 Task 7), which releases from LastMove.
//
// Every record write goes through a provision transition (the record_seq
// rule); every claim write through a hostclaim transition.
//
// Arming (rulings J18, G17; Plan 5 Task 9a). Like the landing and the
// target's half of the rollback, the source's recovery needs the claim's
// park hooks, RunHeld (for the dirty return's residue parks) and the local
// authorizer, which armMovesParks sets only once the wake-source gate opens
// (movesWakeSourceRegistered, Plan 5 Task 9b), so
// recoverySourceArmed answers false in production and the pass logs a
// standing notice. The probe hook is
// the recovery probe adapter (armRecoveryProbe) with CARLOS_HOST_CLAIM on,
// bound to the activator's Group D entry. Without the claim switch the
// legacy nil hook still waits, recording nothing. LogEvent
// is nil in production, where the event is a log line.

// recoverStepsPerPass bounds the phases one pass runs: the dirty path from
// waiting to the finalization is fifteen.
const recoverStepsPerPass = 16

// recoveryProbeAttempts is the recovery probe's budget (§7.8: three times
// the landing's three).
const recoveryProbeAttempts = 9

// The recovery probe's admission coverage (§7.14): a dirty return admits
// the return generation from its sealed lane, a clean one restores the
// source's own generation.
const (
	recoveryCoverageSealed = "sealed"
	recoveryCoverageClean  = "clean-return"
)

// recoveryProbeOutcome is what one call of the recovery probe found.
type recoveryProbeOutcome int

const (
	// recoveryProbeWaiting: no answer yet; the next pass calls the probe
	// again (and only the probe).
	recoveryProbeWaiting recoveryProbeOutcome = iota
	// recoveryProbeAnswered: the app answered; the recovery goes to probed.
	recoveryProbeAnswered
	// recoveryProbeExhausted: every attempt failed; the recovery stops for
	// an operator (StuckRecoveryProbeAttempts).
	recoveryProbeExhausted
)

// recoveryProbeRequest is what the source asks of the recovery probe (Task
// 9 builds it against claim Group D for recovery, J2).
type recoveryProbeRequest struct {
	// RollbackID names the rollback, whose id names a dirty return's
	// generation and journal.
	RollbackID string
	// Clean is the return's kind; Coverage is the admission's coverage,
	// "sealed" (dirty), "clean-return", or "" for a declared return.
	Clean    bool
	Coverage string
	// Declared is a clean return after a declared-lost or fenced target
	// (Q6): the record's rollback cites the loss the claim journals. It asks
	// no coverage, for every session, and carries no loss id: the claim's
	// writer derives it.
	Declared bool
	// Generation is the generation the dirty return installed (the
	// rollback id); empty for a clean return, whose probe admits a fresh
	// one restored from SourceGeneration (src_gen, J14).
	Generation, SourceGeneration string
	// Attempts is the budget, nine.
	Attempts int
	// Record records the inputs the probe is about to render, before it
	// renders them (MoveRecordRecoveryInputs; Task 3b concern 3).
	Record func(context.Context, provision.MoveInputs) error
}

// recoverySourceArmed reports whether this box can run the source's
// recovery: every hook a dirty return's install and the open call is wired,
// the claim's park hooks, RunHeld and local authorizer included
// (landingArmed). In production armMovesParks sets those only once the
// wake-source gate opens (movesWakeSourceRegistered, Plan 5 Task 9b), so this
// answers false there until then (J18).
func recoverySourceArmed(cfg moveCfg) bool {
	return landingArmed(cfg) && cfg.DataDir != "" && cfg.HandHold != nil && cfg.ReleaseHeld != nil && cfg.Timers != nil
}

// releaseHeld is the recovery's open in production: registry.ReleaseHold
// under the op that holds the row, then written (holdStageRequester) when
// it landed, so root re-judges what it staged now the hold has ended. The
// caller holds provision.WithHostHoldLock.
func releaseHeld(reg *registry.Registry, written func(host string)) func(host, op string) (bool, error) {
	return func(host, op string) (bool, error) {
		released, err := reg.ReleaseHold(host, op)
		if err == nil && released && written != nil {
			written(host)
		}
		return released, err
	}
}

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

// recoverSource is the participant step for a rolling_back record on its
// source (movesPass).
func (w *movesWorker) recoverSource(ctx context.Context, s release.Store, rec provision.Record) error {
	cfg := w.cfg
	m := rec.Move
	if m == nil || m.Legacy() || m.State != provision.MoveRollingBack || m.Rollback == nil || m.From != cfg.Region || m.To == cfg.Region {
		return nil
	}
	if !recoverySourceArmed(cfg) {
		w.notice(rec.Host+"#recover", "moves: %s: the source's recovery of rollback %s waits: this box's recovery is not armed (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 := &recoverRun{w: w, s: s, rec: rec, g: provision.MoveGuard{ID: m.ID, Incarnation: m.Incarnation}, rb: m.Rollback.ID}
	defer r.settleLease()
	for i := 0; i < recoverStepsPerPass; i++ {
		advanced, err := r.step(ctx)
		if err != nil || !advanced {
			return err
		}
		fresh, err := provision.GetRecord(ctx, s, rec.Account, rec.App, rec.Host)
		if err != nil {
			return fmt.Errorf("re-read the record: %w", err)
		}
		if !r.same(fresh) {
			return nil
		}
		r.rec = *fresh
	}
	return nil
}

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

// same reports whether fresh is still this rollback, rolling back.
func (r *recoverRun) same(fresh *provision.Record) bool {
	m := fresh.Move
	return m != nil && m.ID == r.g.ID && m.Incarnation == r.g.Incarnation && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.ID == r.rb
}

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

// settleLease releases the own-location lease a dirty return's build holds
// from building through published, once the record is past them or no
// longer this return: the lease is the build's alone.
func (r *recoverRun) settleLease() {
	m := r.rec.Move
	if m != nil && m.State == provision.MoveRollingBack && m.Rollback != nil && m.Rollback.Dirty &&
		(m.Rollback.SourcePhase == provision.RbSourceBuilding || m.Rollback.SourcePhase == provision.RbSourcePublished) {
		return
	}
	releaseOwnLease(r.w, r.rec.Host)
}

// step runs the source's current recovery phase once. advanced is true when
// it wrote the record.
func (r *recoverRun) step(ctx context.Context) (bool, error) {
	rb := r.rollback()
	switch rb.SourcePhase {
	case provision.RbSourceWaiting:
		return r.waiting(ctx)
	case provision.RbSourceClaimed:
		return r.claimed(ctx)
	case provision.RbSourceRestoring:
		return r.restoring(ctx)
	case provision.RbSourceVerified:
		return r.verified(ctx)
	case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured:
		return r.install(ctx)
	case provision.RbSourceRestored:
		return r.phase(ctx, provision.RbSourceRestored, provision.RbSourceProbing, nil, "")
	case provision.RbSourceProbing:
		return r.probing(ctx)
	case provision.RbSourceProbed:
		return r.probed(ctx)
	case provision.RbSourceRecovered:
		return r.open(ctx)
	case provision.RbSourceOpened:
		return r.finalize(ctx)
	}
	return false, nil
}

// phase writes one source phase step, 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 *recoverRun) 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.MoveRollbackSourcePhase(r.g, from, to, stamp)))
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: the source's recovery %s to %s refused (%v); the record moved on", host, from, to, err)
			return false, nil
		}
		return false, fmt.Errorf("the source's recovery %s to %s: %w", from, to, err)
	}
	log.Printf("moves: %s: the source of rollback %s is %s", host, r.rb, to)
	r.w.logTransition(ctx, r.s, pre, *after)
	if err := r.fault("after-record-" + to); err != nil {
		return true, err
	}
	return true, nil
}

// stuck stops the recovery for an operator (MoveMarkStuck) with a recorded
// reason; the row stays held and move --retry returns to the phase.
func (r *recoverRun) stuck(ctx context.Context, reason string, why error) (bool, error) {
	host := r.rec.Host
	log.Printf("moves: %s: the recovery of %s stops for an operator (%v): %s", host, r.rb, why, reason)
	var pre provision.Record
	after, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveMarkStuck(r.g, 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 recovery stuck: %w", err)
	}
	r.w.logTransition(ctx, r.s, pre, *after)
	r.w.pageStuck(ctx, r.rec, "source", reason)
	nudgeSerial(ctx, r.s)
	return true, nil
}

// claim reads the host claim and requires this rollback's restoring op,
// with this box the move's source, for the record's incarnation.
func (r *recoverRun) claim(ctx context.Context) (hostclaim.Observed, error) {
	obs, err := hostclaim.Read(ctx, r.s, r.rec.Host)
	if err != nil {
		return obs, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		return obs, fmt.Errorf("move %s has no host claim", r.g.ID)
	}
	if err := r.ours(obs.Claim); err != nil {
		return obs, fmt.Errorf("move %s: %w", r.g.ID, err)
	}
	return obs, nil
}

// ours is the claim check of every source phase before the finalization:
// this rollback's restoring op, this box the move's source, the record's
// incarnation.
func (r *recoverRun) ours(c *hostclaim.Claim) error {
	box := r.cfg().Box
	if c == nil || c.Op.ID != r.g.ID || c.Op.Kind != hostclaim.KindRestoring || c.Op.Intent["rollback"] != r.rb {
		return fmt.Errorf("%w: the claim is not rollback %s", hostclaim.ErrWrongState, r.rb)
	}
	if c.Holder == nil || c.Holder.Incarnation != r.g.Incarnation {
		return fmt.Errorf("%w: the claim names another incarnation", hostclaim.ErrWrongState)
	}
	if len(c.Op.Participants) == 0 || c.Op.Participants[0] != box || c.Op.Intent["source"] != box {
		return fmt.Errorf("%w: this box is not the move's source in the claim", hostclaim.ErrWrongState)
	}
	return nil
}

// executing requires the claim at phase with this box executing.
func (r *recoverRun) executing(c *hostclaim.Claim, phase string) error {
	if c.Op.Phase != phase || c.Op.Exec != r.cfg().Box {
		return fmt.Errorf("move %s: the claim is restoring/%q with executor %q, not %s on this box", r.g.ID, c.Op.Phase, c.Op.Exec, phase)
	}
	return nil
}

// sourceStamp is S, the source installation's stamp the flip journalled.
func sourceStamp(c *hostclaim.Claim) (uint64, error) {
	s, err := strconv.ParseUint(c.Op.Journal[hostclaim.JournalSourceStamp], 10, 64)
	if err != nil || s == 0 {
		return 0, fmt.Errorf("%w: the flip's source stamp is not journalled", hostclaim.ErrWrongState)
	}
	return s, nil
}

// stillAt is the record check run under the host's hold lock before a row
// write: the record, read fresh, must still be this rollback with the
// source at phase. It reads the record only.
func (r *recoverRun) stillAt(ctx context.Context, phase string) error {
	fresh, err := provision.GetRecord(ctx, r.s, r.rec.Account, r.rec.App, r.rec.Host)
	if err != nil {
		return fmt.Errorf("re-read the record: %w", err)
	}
	if !r.same(fresh) || fresh.Move.Rollback.SourcePhase != phase {
		return fmt.Errorf("move %s: the record is no longer rollback %s at %s", r.g.ID, r.rb, phase)
	}
	return nil
}

// waiting (§7.8 source waiting): the target's evidence, released clean or
// returned dirty, and the claim released with nobody executing.
func (r *recoverRun) waiting(ctx context.Context) (bool, error) {
	host, rb := r.rec.Host, r.rollback()
	if !(rb.TargetPhase == provision.RbTargetReleased && !rb.Dirty) && !(rb.TargetPhase == provision.RbTargetReturned && rb.Dirty) {
		r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the target of rollback %s (%s, dirty %t)", host, r.rb, rb.TargetPhase, rb.Dirty)
		return false, nil
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if op := obs.Claim.Op; op.Phase != hostclaim.PhaseReleased || (op.Exec != "" && op.Exec != "none") {
		r.w.notice(host+"#rc-wait", "moves: %s: the source waits for the claim's release of rollback %s (restoring/%q, executor %q)", host, r.rb, op.Phase, op.Exec)
		return false, nil
	}
	return r.phase(ctx, provision.RbSourceWaiting, provision.RbSourceClaimed, nil, "")
}

// claimed (§7.8 source claimed): the claim's return to this box at S (and
// src_gen for a clean return), the guard's activation at S, and a dirty
// return's park-local for its registered pre-move installation (J13).
func (r *recoverRun) claimed(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, rb := r.rec.Host, r.g.ID, r.rollback()
	clean := !rb.Dirty
	if err := r.markJournals(); err != nil {
		return false, err
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if op := obs.Claim.Op; op.Phase != hostclaim.PhaseClaimed || op.Exec != cfg.Box {
		if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.ReturnToSource(id, cfg.Box, clean, hostclaim.ReturnFence{})); err != nil {
			return false, fmt.Errorf("the claim's return to the source: %w", err)
		}
		log.Printf("moves: %s: the source executes the host again for rollback %s (clean %t)", host, r.rb, clean)
	}
	if err := r.fault("after-return-claimed"); err != nil {
		return false, err
	}
	c := obs.Claim
	s, err := sourceStamp(c)
	if err != nil {
		return false, err
	}
	if c.Holder.Stamp != s {
		return false, fmt.Errorf("move %s: the holder is at stamp %d, not the source's %d", id, c.Holder.Stamp, s)
	}
	if clean && c.Holder.Generation != c.Op.Journal[hostclaim.JournalSrcGen] {
		return false, fmt.Errorf("move %s: a clean return's holder names generation %q, not src_gen", id, c.Holder.Generation)
	}
	g := cfg.Guard
	if g.Claims == nil {
		g.Claims = hostguard.StoreReader{Store: r.s}
	}
	if err := g.Activate(ctx, host, cfg.Box, nil); err != nil {
		return false, fmt.Errorf("activate the source's installation at stamp %d: %w", s, err)
	}
	if err := r.fault("after-activate"); err != nil {
		return false, err
	}
	if rb.Dirty && hostclaim.InventoryHasStamp(c, cfg.Box, s) {
		if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.QueueLocalPark(id, cfg.Box, obligationID)); err != nil {
			return false, fmt.Errorf("queue the source's local park: %w", err)
		}
	}
	return r.phase(ctx, provision.RbSourceClaimed, provision.RbSourceRestoring, nil, "")
}

// restoring (§7.8 source restoring): the row handed from the move's hold to
// the rollback's, as restoring (J3), with the record checked under the
// host's hold lock; then the branch, clean or dirty.
func (r *recoverRun) restoring(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	handed, err := cfg.HandHold(host, "restoring", r.rb, id, func() error { return r.stillAt(ctx, provision.RbSourceRestoring) })
	if err != nil {
		return false, fmt.Errorf("hand the row's hold to the rollback: %w", err)
	}
	if !handed {
		return false, fmt.Errorf("move %s: this box's row is held by neither the move nor rollback %s", id, r.rb)
	}
	if err := r.fault("after-hand"); err != nil {
		return false, err
	}
	next := provision.RbSourceVerified
	if r.rollback().Dirty {
		next = provision.RbSourceParking
	}
	return r.phase(ctx, provision.RbSourceRestoring, next, nil, "")
}

// verified (§7.8 clean verified, INV-4): the source's files and timers hash
// to what it recorded when it stopped, and its warm database's marker names
// src_gen. A mismatch stops for an operator; a retry runs this again.
func (r *recoverRun) verified(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, m := r.rec.Host, r.g.ID, r.rec.Move
	if err := r.fault("verify"); err != nil {
		return false, err
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil {
		return false, err
	}
	srcGen, ok := obs.Claim.Op.Journal[hostclaim.JournalSrcGen]
	if !ok {
		return false, fmt.Errorf("move %s: the flip's source generation is not journalled", id)
	}
	if m.Source == nil || m.Source.Files == nil || m.Source.Files.Plain == "" || m.Source.TimersSHA256 == "" {
		return false, fmt.Errorf("move %s: the record has no stopped baselines to verify against", id)
	}
	dir, envPath := cfg.FilesDir(host)
	plain, _, _, err := archivePlainDigest(dir, envPath)
	why, retry, ferr := archiveFailure(host, err)
	switch {
	case ferr != nil:
		return false, fmt.Errorf("digest the source's files: %w", ferr)
	case retry:
		r.w.notice(host+"#rc-verify", "moves: %s: a file changed while the source's files were digested; the verification runs again", host)
		return false, nil
	case why != "":
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the source's files cannot be digested: %s", why))
	case plain != m.Source.Files.Plain:
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the files are %s, recorded %s", plain, m.Source.Files.Plain))
	}
	rows, err := cfg.Timers(host)
	if err != nil {
		return false, fmt.Errorf("read the timers: %w", err)
	}
	if got := moves.TimersDigest(rows); got != m.Source.TimersSHA256 {
		return r.stuck(ctx, provision.StuckBaselineMismatch, fmt.Errorf("the timers are %s, recorded %s", got, m.Source.TimersSHA256))
	}
	if why, err := r.warmMismatch(srcGen); err != nil {
		return false, err
	} else if why != nil {
		return r.stuck(ctx, provision.StuckBaselineMismatch, why)
	}
	log.Printf("moves: %s: the source still matches what it recorded when it stopped", host)
	return r.phase(ctx, provision.RbSourceVerified, provision.RbSourceProbing, nil, "")
}

// warmMismatch is the warm database's half of the clean verification (fix
// round 1, ruling 1). A marker that is there must name src_gen. A database
// with no marker is refused unless src_gen is the origin (""), whose
// database never had one. A database evicted with its marker passes: the
// probe restores the source cold from src_gen (J14), so nothing warm is
// read. why is the mismatch; err a read that failed.
func (r *recoverRun) warmMismatch(srcGen string) (why, err error) {
	db := r.dbPath()
	gen, gerr := hostbox.ReadGenerationMarker(db + ".gen")
	if gerr != nil {
		return fmt.Errorf("the warm database's marker: %w", gerr), nil
	}
	present, err := regularFile(db)
	if err != nil {
		return nil, err
	}
	switch {
	case gen != "" && gen != srcGen:
		return fmt.Errorf("the warm database's marker names %q, the pre-move generation is %q", gen, srcGen), nil
	case gen == "" && present && srcGen != "":
		return fmt.Errorf("the warm database has no marker, the pre-move generation is %q", srcGen), nil
	case gen == "" && !present:
		log.Printf("moves: %s: the source's warm database was evicted; the probe restores generation %q cold", r.rec.Host, srcGen)
	}
	return nil, nil
}

// returnSeal reads the return seal the target journalled in the claim.
func returnSeal(c *hostclaim.Claim) (moves.Seal, error) {
	raw, ok := c.Op.Journal[moves.JournalReturnSeal]
	if !ok {
		return moves.Seal{}, fmt.Errorf("%w: the claim carries no return seal", moves.ErrBadSeal)
	}
	return moves.DecodeSeal(raw)
}

// install (§7.8 dirty parking to configured): one phase of the shared
// installer over the return lane, its seal and baseline the target's
// journalled return. A refusal it records stops the return for an operator.
func (r *recoverRun) install(ctx context.Context) (bool, error) {
	host := r.rec.Host
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	raw, ok := obs.Claim.Op.Journal[moves.JournalReturnBaseline]
	if !ok {
		return false, fmt.Errorf("move %s: the claim carries no return baseline", r.g.ID)
	}
	baseline, err := moves.DecodeReturnBaseline(raw)
	if err != nil {
		return false, err
	}
	src, err := recoveryInstallSource(r.w, r.s, r.rec, returnSeal, &baseline)
	if err != nil {
		return false, err
	}
	advanced, refused, err := installGeneration(ctx, r.s, r.w, r.rec, src)
	if refused != "" {
		log.Printf("moves: %s: the return of %s stopped for an operator: %s", host, r.rb, refused)
		// The stuck mark is installGeneration's write, logged there.
		r.w.pageStuck(ctx, r.rec, "source", refused)
	}
	return advanced, err
}

// probing (§7.8 source probing): the claim steps to probing before the
// first attempt (J1), then the recovery probe. Nothing earlier is run again
// from here.
func (r *recoverRun) probing(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, rb := r.rec.Host, r.g.ID, r.rollback()
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if obs.Claim.Op.Phase == hostclaim.PhaseClaimed {
		if err := r.executing(obs.Claim, hostclaim.PhaseClaimed); err != nil {
			return false, err
		}
		if obs, err = hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseClaimed, hostclaim.PhaseProbing)); err != nil {
			return false, fmt.Errorf("the claim's probing: %w", err)
		}
		if err := r.fault("after-claim-probing"); err != nil {
			return false, err
		}
	}
	if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil {
		return false, err
	}
	if cfg.recoveryProbe == nil {
		r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: it is Task 9's, on claim Plan 10's recovery admission", host, r.rb)
		return false, nil
	}
	req := recoveryProbeRequest{RollbackID: r.rb, Clean: !rb.Dirty, Attempts: recoveryProbeAttempts,
		SourceGeneration: obs.Claim.Op.Journal[hostclaim.JournalSrcGen],
		Record: func(ctx context.Context, in provision.MoveInputs) error {
			_, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, provision.MoveRecordRecoveryInputs(r.g, r.rb, in))
			return err
		}}
	// A declared or fenced return asks no coverage (Q6); while the record
	// and the claim disagree on the target's loss the probe waits.
	declared, err := recoveryDeclared(obs.Claim, rb.LossRef)
	if err != nil {
		r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: %v", host, r.rb, err)
		return false, nil
	}
	switch {
	case rb.Dirty:
		req.Coverage, req.Generation = recoveryCoverageSealed, r.rb
	case declared:
		req.Declared = true
	default:
		req.Coverage = recoveryCoverageClean
	}
	out, err := cfg.recoveryProbe(ctx, r.s, r.rec, req)
	var missing *ClaimDependencyError
	if errors.As(err, &missing) {
		// The adapter refused before recording or waking (Plan 10 J2):
		// probing waits, as it did with no hook.
		r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s waits: %v", host, r.rb, err)
		return false, nil
	}
	if err != nil {
		return false, fmt.Errorf("the recovery probe: %w", err)
	}
	switch out {
	case recoveryProbeAnswered:
		return r.phase(ctx, provision.RbSourceProbing, provision.RbSourceProbed, nil, "")
	case recoveryProbeExhausted:
		return r.stuck(ctx, provision.StuckRecoveryProbeAttempts, fmt.Errorf("%d attempts", recoveryProbeAttempts))
	}
	r.w.notice(host+"#rc-probe", "moves: %s: the recovery probe of rollback %s has not answered yet", host, r.rb)
	return false, nil
}

// probed (§7.8 source probed): the claim steps to probed and, for a dirty
// return, queues the return lane's park in the same write; then recovered
// with recovered_at, and the serial is nudged for the target's cleaning.
func (r *recoverRun) probed(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, rb := r.rec.Host, r.g.ID, r.rollback()
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	var lane func(*hostclaim.Claim) error
	if rb.Dirty {
		seal, err := returnSeal(obs.Claim)
		if err != nil {
			return false, err
		}
		scope := moves.ReturnLaneScope(host, r.rb, obs.Claim.Holder.Incarnation, seal.Bound())
		lane = hostclaim.QueueReturnLanePark(id, cfg.parkRegion(), scope, obligationID)
	}
	switch obs.Claim.Op.Phase {
	case hostclaim.PhaseProbing:
		if err := r.executing(obs.Claim, hostclaim.PhaseProbing); err != nil {
			return false, err
		}
		step := hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbing, hostclaim.PhaseProbed)
		if _, err := hostclaim.Update(ctx, r.s, obs, func(c *hostclaim.Claim) error {
			if err := step(c); err != nil {
				return err
			}
			if lane != nil {
				return lane(c)
			}
			return nil
		}); err != nil {
			return false, fmt.Errorf("the claim's probed: %w", err)
		}
	case hostclaim.PhaseProbed:
		// A replay after the claim's write: the lane park is idempotent,
		// and once the park engine completed it it is not queued again.
		if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil {
			return false, err
		}
		if lane != nil {
			if _, err := hostclaim.Update(ctx, r.s, obs, lane); err != nil {
				return false, fmt.Errorf("queue the return lane's park: %w", err)
			}
		}
	default:
		return false, r.executing(obs.Claim, hostclaim.PhaseProbing)
	}
	if err := r.fault("after-claim-probed"); err != nil {
		return false, err
	}
	ok, err := r.phase(ctx, provision.RbSourceProbed, provision.RbSourceRecovered, func(rb *provision.MoveRollback) {
		t := r.now().UTC()
		rb.RecoveredAt = &t
	}, "source-recovered")
	if ok {
		nudgeSerial(ctx, r.s)
	}
	return ok, err
}

// open (§7.8 source opened): the rollback's hold on the row ends, under the
// host's hold lock with the record read fresh (J12); the claim steps to
// opened; the record says opened with opened_at. A replay that finds the
// row already open goes on.
func (r *recoverRun) open(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id := r.rec.Host, r.g.ID
	var err error
	provision.WithHostHoldLock(host, func() {
		if err = r.stillAt(ctx, provision.RbSourceRecovered); err != nil {
			return
		}
		var released bool
		if released, err = cfg.ReleaseHeld(host, r.rb); err != nil || released {
			return
		}
		var rt *registry.Route
		if rt, err = cfg.Route(host); err != nil {
			return
		}
		switch {
		case rt == nil:
			err = fmt.Errorf("move %s: this box has no row to open", id)
		case rt.HoldOp != "":
			err = fmt.Errorf("move %s: the row is held %s by %q, not by rollback %s", id, rt.Status, rt.HoldOp, r.rb)
		}
	})
	if err != nil {
		return false, fmt.Errorf("open the row: %w", err)
	}
	if err := r.fault("after-row-opened"); err != nil {
		return false, err
	}
	obs, err := r.claim(ctx)
	if err != nil {
		return false, err
	}
	if obs.Claim.Op.Phase != hostclaim.PhaseOpened {
		if err := r.executing(obs.Claim, hostclaim.PhaseProbed); err != nil {
			return false, err
		}
		if _, err := hostclaim.Update(ctx, r.s, obs, hostclaim.SetRestoringPhase(id, hostclaim.PhaseProbed, hostclaim.PhaseOpened)); err != nil {
			return false, fmt.Errorf("the claim's opened: %w", err)
		}
	}
	if err := r.fault("after-claim-opened"); err != nil {
		return false, err
	}
	return r.phase(ctx, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) {
		t := r.now().UTC()
		rb.OpenedAt = &t
	}, "source-opened")
}

// finalize (§7.8 rolled_back): with the source opened and recovered and the
// target parked, and no obligation of the op left in the claim, the
// terminal event, the record's close, then the claim's release to serving:
// record first (§7.5 and §19 K8 as Paul's ruling M4 amends them). A crash
// between the halves leaves the record at rest and the claim
// restoring/opened with nothing owed; the console's recovery pass releases
// from LastMove. A claim already released by this move (a binary that
// closed claim first) only closes.
func (r *recoverRun) finalize(ctx context.Context) (bool, error) {
	cfg := r.cfg()
	host, id, rb := r.rec.Host, r.g.ID, r.rollback()
	if rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil {
		r.w.notice(host+"#rc-final", "moves: %s: rollback %s waits for the target's park before it closes", host, r.rb)
		return false, nil
	}
	obs, err := hostclaim.Read(ctx, r.s, host)
	if err != nil {
		return false, fmt.Errorf("read the host claim: %w", err)
	}
	if obs.Absent {
		return false, fmt.Errorf("move %s has no host claim", id)
	}
	released := hostclaim.RequireReleasedMove(id)(obs.Claim) == nil
	if !released {
		if err := r.ours(obs.Claim); err != nil {
			return false, fmt.Errorf("move %s: %w", id, err)
		}
		if err := r.executing(obs.Claim, hostclaim.PhaseOpened); err != nil {
			return false, err
		}
		// 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+"#rc-final", "moves: %s: rollback %s waits for %d chore(s) of the move to be acknowledged before it closes", host, r.rb, owed)
			return false, nil
		}
	}
	var pre provision.Record
	closed, err := provision.SetRecordMove(ctx, r.s, r.rec.Account, r.rec.App, host, observedMove(&pre, provision.MoveRolledBackClose(r.g, r.now())))
	if err != nil {
		if errors.Is(err, provision.ErrMoveConflict) {
			log.Printf("moves: %s: the close of rollback %s refused (%v); the record moved on", host, r.rb, err)
			return false, nil
		}
		return false, fmt.Errorf("close the rolled-back move: %w", err)
	}
	r.w.logTransition(ctx, r.s, pre, *closed)
	log.Printf("moves: %s: move %s is rolled back and closed", host, id)
	r.w.pageRolledBack(ctx, r.rec)
	nudgeSerial(ctx, r.s)
	if !released {
		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 rollback %s", host, r.rb)
		if err := r.fault("after-release-move"); err != nil {
			return true, err
		}
	}
	if err := r.fault("after-close"); err != nil {
		return true, err
	}
	// The journals go in the pass that closes; a crash before this leaves
	// them to the local journal sweep (sweepRecoveryJournal).
	for _, op := range []string{r.rb, id} {
		if err := parkRecoveryJournal(cfg.DataDir, host, op, true); err != nil {
			return true, fmt.Errorf("park the source's journal of %s: %w", op, err)
		}
	}
	return true, nil
}

// The source's recovery steps in its two op journals, the cut's under the
// move id and the recovery's under the rollback id (§7.8: "journaled ... in
// <DataDir>/.moves/<host>/<rollback id>/"; a dirty return's install keeps
// its build and extract scratch there too). Fix round 1, ruling 2.
const (
	// rcStepSource marks a journal as a source recovery's (written at
	// claimed): sourceJournalMark, so the sweep can tell it from a cut, a
	// landing or a target's, and knows the incarnation to park it under.
	rcStepSource = "rb-source"
	// rcStepClosed is "done" once the record closed.
	rcStepClosed = "rb-source-closed"
	// rcStepJournalPark is parkLocalArtifacts' step: the retirement the
	// journal is parked into (no artifacts of its own).
	rcStepJournalPark = "rb-source-journal-park"
)

// sourceJournalMark is rcStepSource's value.
type sourceJournalMark struct {
	Move        string `json:"move"`
	Rollback    string `json:"rollback"`
	Incarnation string `json:"incarnation"`
}

// markJournals marks the source's journals as this recovery's: the cut's,
// when this box holds it, and the recovery's own, made here for a clean
// return (the installer's for a dirty one). A mark already there is kept.
func (r *recoverRun) markJournals() error {
	dd, host := r.cfg().DataDir, r.rec.Host
	body, err := json.Marshal(sourceJournalMark{Move: r.g.ID, Rollback: r.rb, Incarnation: r.g.Incarnation})
	if err != nil {
		return err
	}
	for _, op := range []string{r.g.ID, r.rb} {
		if op == r.g.ID && !moveJournalPresent(dd, host, op) {
			continue
		}
		j, err := moves.ReadJournal(dd, host, op)
		if err != nil {
			return err
		}
		if j.Done(rcStepSource) {
			continue
		}
		j.Mark(rcStepSource, string(body))
		if err := j.Commit(dd); err != nil {
			return err
		}
	}
	return nil
}

// parkRecoveryJournal parks one source journal of a closed rollback: 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 journal's op) and the journal is renamed into it. A journal that is
// not there was parked already; one with no recovery mark is not this
// recovery's and is left.
func parkRecoveryJournal(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
	}
	var mark sourceJournalMark
	raw, ok := j.Steps[rcStepSource]
	if !ok {
		return nil
	}
	if err := json.Unmarshal([]byte(raw), &mark); err != nil || !journalComponent(mark.Incarnation) {
		return fmt.Errorf("the journal of %s has an unreadable recovery mark", op)
	}
	write := func(step, value string) error {
		j.Mark(step, value)
		return j.Commit(dataDir)
	}
	if closed && j.Steps[rcStepClosed] != "done" {
		if err := write(rcStepClosed, "done"); err != nil {
			return err
		}
	}
	retirement, err := parkLocalArtifacts(context.Background(), j, write, localArtifactPark{DataDir: dataDir, Incarnation: mark.Incarnation,
		Host: host, Op: op, Step: rcStepJournalPark, 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 %s is parked", host, op)
	return nil
}

// sweepRecoveryJournal is the local journal sweep's source-recovery half
// (sweepFinalJournal, fix round 1, ruling 2): a journal the recovery marked
// (rcStepSource) is parked once the close is shown by durable evidence:
//   - the journal records the close (rcStepClosed "done");
//   - no listed record carries the move: a rolling_back or stuck move is
//     cleared only by provision.MoveRolledBackClose, which refuses unless
//     the rollback's recovered_at, opened_at and target_parked_at are all
//     stamped (the aborts clear only an aborting move, MoveComplete only a
//     cutting one), and the mark is written while the record is rolling
//     back, so the close landed;
//   - a listed record carries the move with exactly those three stamps on
//     this rollback: the close's preconditions hold, nothing of the
//     recovery reads its journals again, and the close follows.
//
// A host the listing does not name, or a record that carries the move
// without the three stamps, is left.
func (w *movesWorker) sweepRecoveryJournal(dataDir, host, op string, j *moves.Journal, recs []provision.Record) error {
	var mark sourceJournalMark
	if err := json.Unmarshal([]byte(j.Steps[rcStepSource]), &mark); err != nil || mark.Move == "" || mark.Rollback == "" {
		return fmt.Errorf("the journal of %s has an unreadable recovery mark", op)
	}
	closed := j.Steps[rcStepClosed] == "done"
	if !closed {
		if len(recs) == 0 {
			w.notice(host+"#rc-sweep-"+op, "moves: %s: the source's journal of %s is a recovery's, and no listed record names the host; it is left", host, op)
			return nil
		}
		for _, rec := range recs {
			m := rec.Move
			if m == nil || m.ID != mark.Move {
				continue
			}
			rb := m.Rollback
			if rb == nil || rb.ID != mark.Rollback || rb.RecoveredAt == nil || rb.OpenedAt == nil || rb.TargetParkedAt == nil {
				return nil
			}
			return parkRecoveryJournal(dataDir, host, op, false)
		}
	}
	return parkRecoveryJournal(dataDir, host, op, true)
}