package host

import (
	"context"
	"os"
	"path/filepath"
	"strings"
	"testing"

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

// The source's recovery after a rollback (moves spec §7.8 source table; Plan
// 4 Task 5). The fixture is Task 4's: the landing harness landed through
// restored and stepped to probing, then rolled back and released by the
// target's own pass (clean, or dirty with the target's rows changed, so its
// return lane holds a snapshot of the stopped database). The source box
// (ie-box) then runs its own passes over the same store and buckets: its
// row is the cut's, held moving by the move id; its warm database's marker
// names the pre-move generation; its hold hand-over, its open and the probe
// (Task 9, BLOCKED-ON-CLAIM C1) are faked; everything else is the real
// recovery, the shared installer (Task 3) and provision's and hostclaim's own
// transitions.

type recoverHarness struct {
	*rollbackHarness
	sdir      string // the source's data directory
	lockDir   string
	srcRoute  registry.Route
	sw        *movesWorker
	srcRender appconfig.HeldRender
	srcRends  int      // the installer's held renders on the source
	srcAsked  []string // the releases the source's gate was asked about
	srcHanded []string // status:to:from of each hand-over the source asked for
	sEvents   []string
	// probe answers each recovery probe; nil records the render's inputs
	// and answers.
	probe  func(req recoveryProbeRequest) (recoveryProbeOutcome, error)
	probes []recoveryProbeRequest
	// recoveryWake, when set, replaces the fake: the production adapter
	// runs with it as the activator's recovery probe entry (home part 2,
	// Task 4).
	recoveryWake ClaimRecoveryProbeWake
	// srcPage, when set, is the source's pager instead of pages.hook().
	srcPage pageFunc
	rcAt    string
	rcDone  bool
	rcHits  []string
	rcHook  func(point string)
}

// newRecoverHarness builds the fixture up to the rollback decision's
// release: dirty makes the target's state differ from what it installed, so
// its release is dirty and its return a snapshot. The caller rolls back and
// runs the target's pass.
func newRecoverHarness(t *testing.T, dirty bool, opts ...p1Option) *recoverHarness {
	t.Helper()
	h := &recoverHarness{rollbackHarness: newRollbackHarness(t, opts...)}
	if dirty {
		h.targetChanged(t)
	}
	h.sdir = h.cfg.DataDir
	h.lockDir = t.TempDir()
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.MkdirAll(filepath.Join(h.sdir, d), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	// The source's warm database (the cut's stopped database) carries the
	// marker of its pre-move generation, src_gen.
	db := provision.DefaultDBPath(h.sdir, p1Host)
	if err := os.WriteFile(db+".gen", []byte("cur\n"), 0o644); err != nil {
		t.Fatal(err)
	}
	h.srcRoute = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "moving", HoldOp: p1ID,
		Socket: provision.DefaultSocket(p1Host), DBPath: db, Provisioned: true}
	// The current bundle is not the pinned one: a recovery renders and
	// records what is current, never compared with move.inputs.
	h.srcRender = appconfig.HeldRender{Path: filepath.Join(h.sdir, p1Host+".json"), SHA256: "y", BundleETags: map[string]string{"default": "e9", "environment": ""}, OverlayETag: "o9"}
	return h
}

// released rolls the landing back and runs the target's pass: released
// (clean) or returned (dirty).
func (h *recoverHarness) released(t *testing.T) *provision.MoveRollback {
	t.Helper()
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if want := map[bool]string{false: provision.RbTargetReleased, true: provision.RbTargetReturned}[rb.Dirty]; rb.TargetPhase != want {
		t.Fatalf("the target did not release: %+v", rb)
	}
	return rb
}

// sourceCfg is the source box's configuration with the recovery armed.
func (h *recoverHarness) sourceCfg(t *testing.T) moveCfg {
	t.Helper()
	c := h.landingCfg(t)
	c.Region, c.Box, c.OwnBucket, c.OwnBucketRegion = h.src, h.srcBox, h.srcBucket, h.srcAWS
	c.ParkHomeRegion = ""
	c.DataDir = h.sdir
	c.OptDir = h.optDir
	c.OwnsRoute = func(host string) bool { return host == p1Host }
	c.Route = func(host string) (*registry.Route, error) {
		if host != p1Host {
			return nil, nil
		}
		rt := h.srcRoute
		return &rt, nil
	}
	c.FilesDir = func(host string) (string, string) {
		return filepath.Join(h.sdir, host), filepath.Join(h.sdir, host+".env")
	}
	c.Guard = hostguard.Guard{DataDir: h.sdir, LockDir: h.lockDir}
	c.Local = park.Local{DataDir: h.sdir, Box: h.srcBox}
	c.ParkView = recordView(h.s, park.PlacementRegion(h.src), roleSource)
	c.RenderHeld = func(_ context.Context, _ release.Store, rt registry.Route, op string, _ *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
		h.srcRends++
		if err := before(h.srcRender.SHA256); err != nil {
			return appconfig.HeldRender{}, err
		}
		if rb := h.read(t, p1Host).Move.Rollback; rb == nil || op != rb.ID || rt.HoldOp != rb.ID {
			t.Errorf("rendered for %q over a row held by %q, want the rollback", op, rt.HoldOp)
		}
		return h.srcRender, nil
	}
	c.ReleaseReady = func(_ context.Context, _ release.Store, _ provision.Record, v string) moves.Check {
		h.srcAsked = append(h.srcAsked, v)
		return moves.Check{OK: true}
	}
	c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) {
		h.srcHanded = append(h.srcHanded, status+":"+to+":"+from)
		if validate != nil {
			if err := validate(); err != nil {
				return false, err
			}
		}
		if h.srcRoute.HoldOp == to && h.srcRoute.Status == status {
			return true, nil
		}
		if h.srcRoute.HoldOp != from {
			return false, nil
		}
		h.srcRoute.Status, h.srcRoute.HoldOp = status, to
		return true, nil
	}
	c.ReleaseHeld = func(host, op string) (bool, error) {
		if op == "" || h.srcRoute.HoldOp != op {
			return false, nil
		}
		h.srcRoute.Status, h.srcRoute.HoldOp = "ready", ""
		return true, nil
	}
	c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
		h.sEvents = append(h.sEvents, ev.Event)
		return nil
	}
	c.Page = h.pages.hook()
	if h.srcPage != nil {
		c.Page = h.srcPage
	}
	c.recoveryProbe = func(ctx context.Context, s release.Store, rec provision.Record, req recoveryProbeRequest) (recoveryProbeOutcome, error) {
		h.probes = append(h.probes, req)
		if h.recoveryWake != nil {
			// The production adapter (armRecoveryProbe's) over this worker:
			// the box's current inputs recorded, then the wake.
			w := h.sw
			a := recoveryProbeAdapter{Box: w.cfg.Box, Claims: true, Route: w.cfg.Route, Current: w.recoveryInputs, Wake: h.recoveryWake}
			return a.probe(ctx, s, rec, req)
		}
		if h.probe != nil {
			return h.probe(req)
		}
		if err := req.Record(ctx, h.renderedInputs()); err != nil {
			return recoveryProbeWaiting, err
		}
		return recoveryProbeAnswered, nil
	}
	c.rcFault = func(point string) error {
		h.rcHits = append(h.rcHits, point)
		if h.rcHook != nil {
			h.rcHook(point)
		}
		if point == h.rcAt && !h.rcDone {
			h.rcDone = true
			return errCrash
		}
		return nil
	}
	return c
}

// renderedInputs are what the source renders now: the render's etags and
// the live release.
func (h *recoverHarness) renderedInputs() provision.MoveInputs {
	b := map[string]string{}
	for k, v := range h.srcRender.BundleETags {
		b[k] = v
	}
	return provision.MoveInputs{BundleETags: b, OverlayETag: h.srcRender.OverlayETag, Release: "v7"}
}

// srcPass runs one pass as the source box, its worker kept across passes.
func (h *recoverHarness) srcPass(t *testing.T) {
	t.Helper()
	if h.sw == nil {
		h.sw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	}
	h.sw.cfg = h.sourceCfg(t)
	if err := movesPass(context.Background(), h.s, h.sw); err != nil {
		t.Fatal(err)
	}
}

func (h *recoverHarness) sourceAt(t *testing.T, phase string) *provision.MoveRollback {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m == nil || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.SourcePhase != phase {
		if m != nil {
			t.Fatalf("the source is not at %s: %+v %+v", phase, m, m.Rollback)
		}
		t.Fatalf("the source is not at %s: no move", phase)
	}
	return m.Rollback
}

// opened asserts the end of a recovery: the record opened with RecoveredAt
// and OpenedAt, the claim restoring/opened with the source executing, the
// row open.
func (h *recoverHarness) opened(t *testing.T) *provision.MoveRollback {
	t.Helper()
	rb := h.sourceAt(t, provision.RbSourceOpened)
	if rb.RecoveredAt == nil || rb.OpenedAt == nil {
		t.Fatalf("opened without its stamps: %+v", rb)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseOpened || c.Op.Exec != h.srcBox || c.Op.Intent["rollback"] != rb.ID {
		t.Fatalf("claim: %+v", c.Op)
	}
	if h.srcRoute.Status != "ready" || h.srcRoute.HoldOp != "" {
		t.Fatalf("the row: %+v", h.srcRoute)
	}
	return rb
}

// TestSourceWaitsForTargetEvidence: the source's recovery starts only on the
// target's evidence (§7.8 waiting): the record released clean or returned
// dirty, and the claim's execution holder none. Until then nothing is
// claimed, held or probed.
func TestSourceWaitsForTargetEvidence(t *testing.T) {
	t.Parallel()
	untouched := func(t *testing.T, h *recoverHarness) {
		t.Helper()
		h.sourceAt(t, provision.RbSourceWaiting)
		if h.srcRoute.Status != "moving" || h.srcRoute.HoldOp != p1ID || len(h.srcHanded) != 0 || len(h.probes) != 0 {
			t.Fatalf("the source acted: row %+v handed %v probes %d", h.srcRoute, h.srcHanded, len(h.probes))
		}
		if c := h.claim(t); c.Op.Exec == "ie-box" || c.Op.Phase == hostclaim.PhaseClaimed {
			t.Fatalf("the source claimed: %+v", c.Op)
		}
	}
	t.Run("the target is releasing", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		h.rollBack(t)
		h.srcPass(t)
		untouched(t, h)
	})
	t.Run("released clean on the record, the target still executing", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		h.rollBack(t)
		h.rbAt = "after-record-released"
		h.rollbackPass(t)
		if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased || rb.Dirty {
			t.Fatalf("fixture: %+v", rb)
		}
		if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Exec != "uk-box" {
			t.Fatalf("fixture claim: %+v", c.Op)
		}
		h.srcPass(t)
		untouched(t, h)
		// The target's next pass writes the claim half; the source then
		// recovers.
		h.rollbackPass(t)
		h.srcPass(t)
		h.opened(t)
	})
	t.Run("dirty, released but not returned", func(t *testing.T) {
		h := newRecoverHarness(t, true)
		h.rollBack(t)
		h.rbAt = "after-record-returning"
		h.rollbackPass(t)
		if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturning || !rb.Dirty {
			t.Fatalf("fixture: %+v", rb)
		}
		if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleased || c.Op.Exec != "" {
			t.Fatalf("fixture claim: %+v", c.Op)
		}
		h.srcPass(t)
		untouched(t, h)
	})
	t.Run("unarmed in production", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		h.released(t)
		// The source box as production wires it: no park engine leases
		// (ruling G17), so the recovery waits for Plan 5. h.cfg was armed
		// for the cut guard (armStandIns), so the leases are taken away.
		h.cfg.ParkLeases = nil
		if recoverySourceArmed(h.cfg) {
			t.Fatal("fixture: the source's recovery is still armed")
		}
		h.pass(t)
		untouched(t, h)
	})
}

// TestCleanReturnVerifiesAndOpens: a clean return claims the host back with
// the source's own generation named again (holder.generation = src_gen), its
// row handed from the move to the rollback as restoring, its files, timers
// and warm database marker verified against what it recorded when it
// stopped, then the probe (faked until Task 9), probed, recovered and the
// open. No install phase runs.
func TestCleanReturnVerifiesAndOpens(t *testing.T) {
	t.Parallel()
	h := newRecoverHarness(t, false)
	pinned := h.read(t, p1Host).Move.Inputs
	rb := h.released(t)
	srcGen := h.claim(t).Op.Journal[hostclaim.JournalSrcGen]
	// The landing's probe admission named the landing's generation; the
	// return must name the source's again.
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = p1ID })
	var atClaimed *hostclaim.Holder
	h.rcHook = func(point string) {
		if point == "after-activate" {
			c := h.claim(t)
			hc := *c.Holder
			atClaimed = &hc
		}
	}
	h.srcPass(t)
	got := h.opened(t)
	if atClaimed == nil || atClaimed.Generation != srcGen || srcGen != "cur" || atClaimed.Stamp != 1 {
		t.Fatalf("the holder after claimed: %+v (src_gen %q)", atClaimed, srcGen)
	}
	if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
		t.Fatalf("the source's installation record: %+v %v", captured, err)
	}
	if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
		t.Fatalf("hand-overs: %v", h.srcHanded)
	}
	// No install phase ran: no render, no install steps in the recovery's
	// journal (which holds only its mark, fix round 1), no generation.
	if h.srcRends != 0 {
		t.Fatalf("a clean return rendered through the installer %d times", h.srcRends)
	}
	if j, err := moves.ReadJournal(h.sdir, p1Host, rb.ID); err != nil || len(j.Steps) != 1 || !j.Done(rcStepSource) {
		t.Fatalf("a clean return's journal: %+v %v", j, err)
	}
	for k := range h.f.buckets["data-ie"] {
		if strings.HasPrefix(k, moves.GenerationPrefix(cutRoot, rb.ID)) {
			t.Fatalf("a clean return built a generation: %s", k)
		}
	}
	for _, p := range h.rcHits {
		if strings.HasPrefix(p, "after-record-") {
			switch strings.TrimPrefix(p, "after-record-") {
			case provision.RbSourceParking, provision.RbSourceBuilding, provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored:
				t.Fatalf("a clean return reached %s", p)
			}
		}
	}
	// The probe (Task 9's): clean, coverage clean-return, nine attempts; its
	// inputs recorded, never compared with move.inputs.
	if len(h.probes) != 1 || !h.probes[0].Clean || h.probes[0].Coverage != "clean-return" || h.probes[0].Attempts != 9 || h.probes[0].RollbackID != rb.ID ||
		h.probes[0].SourceGeneration != "cur" {
		t.Fatalf("probes: %+v", h.probes)
	}
	m := h.read(t, p1Host).Move
	if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e9" || in.OverlayETag != "o9" || in.Release != "v7" {
		t.Fatalf("recovery inputs: %+v", got.RecoveryInputs)
	}
	if m.Inputs == nil || pinned == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
		t.Fatalf("move.inputs changed: %+v was %+v", m.Inputs, pinned)
	}
	// A clean return queues no park the source owes: no return lane, no
	// park-local of its own.
	for _, o := range h.claim(t).Obligations {
		if o.Region == "ie" || o.Box == "ie-box" {
			t.Fatalf("a clean return queued %+v", o)
		}
	}
	if !strings.Contains(strings.Join(h.sEvents, " "), "recovered opened") {
		t.Fatalf("events %v", h.sEvents)
	}
	// The record waits for the target's cleaning: a second pass changes
	// nothing.
	h.srcPass(t)
	h.opened(t)
}

// TestDirtyReturnInstallsFromTheLane: a dirty return installs the target's
// snapshot from the return lane through the shared installer: the return
// generation g-<rb>/ (snapshot and anchor) in the source's own bucket, the
// database restored to what the target wrote, .gen naming the rollback;
// recovery inputs recorded at configured and again by the probe, never
// move.inputs (witness wRecoveryInputs: a bundle published after the
// configured render does not block the probe); then probed, which queues
// the return lane's park in the source's region, recovered and opened.
func TestDirtyReturnInstallsFromTheLane(t *testing.T) {
	t.Parallel()
	for _, registered := range []bool{false, true} {
		name := "unregistered"
		if registered {
			name = "registered"
		}
		t.Run(name, func(t *testing.T) {
			h := newRecoverHarness(t, true)
			pinned := h.read(t, p1Host).Move.Inputs
			rb := h.released(t)
			c := h.claim(t)
			returnSeal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
			if err != nil {
				t.Fatal(err)
			}
			if registered {
				rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
					c.Inventory = map[string]hostclaim.InventoryEntry{"ie-box": {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
				})
			}
			var queuedLocal bool
			var atConfigured *provision.MoveInputs
			h.landFault = func(point string) error {
				if point == "after-record-"+provision.RbSourceRestored {
					atConfigured = h.read(t, p1Host).Move.Rollback.RecoveryInputs
					// A bundle published after the configured render.
					h.srcRender.BundleETags = map[string]string{"default": "e10", "environment": ""}
				}
				return nil
			}
			h.rcHook = func(point string) {
				if point == "after-record-"+provision.RbSourceRestoring {
					for _, o := range h.claim(t).Obligations {
						if o.Kind == "park-local" && o.Box == "ie-box" {
							queuedLocal = true
						}
					}
				}
			}
			h.srcPass(t)
			got := h.opened(t)
			if queuedLocal != registered {
				t.Fatalf("the source's park-local queued at claimed: %v, registered %v", queuedLocal, registered)
			}
			if atConfigured == nil || atConfigured.BundleETags["default"] != "e9" || atConfigured.Release != "v7" {
				t.Fatalf("configured recorded %+v", atConfigured)
			}
			if in := got.RecoveryInputs; in == nil || in.BundleETags["default"] != "e10" || in.Release != "v7" {
				t.Fatalf("the probe's recovery inputs: %+v", got.RecoveryInputs)
			}
			m := h.read(t, p1Host).Move
			if m.Inputs == nil || m.Inputs.BundleETags["default"] != pinned.BundleETags["default"] || m.Inputs.Release != pinned.Release {
				t.Fatalf("move.inputs changed: %+v", m.Inputs)
			}
			// The return generation: a snapshot and an anchor at N'.
			prefix := moves.GenerationPrefix(cutRoot, rb.ID)
			for _, k := range []string{moves.SnapshotKey(prefix, returnSeal.N), moves.AnchorKey(prefix, returnSeal.N)} {
				if _, ok := h.f.buckets["data-ie"][k]; !ok {
					t.Fatalf("the return generation has no %s", k)
				}
			}
			dbPath := provision.DefaultDBPath(h.sdir, p1Host)
			if n := countRowsForTest(t, dbPath); n != h.rows+1 {
				t.Fatalf("the returned database holds %d rows, the target wrote %d", n, h.rows+1)
			}
			if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != rb.ID {
				t.Fatalf(".gen %q %v", gen, err)
			}
			// The probe: dirty, into the return generation, coverage sealed.
			if len(h.probes) != 1 || h.probes[0].Clean || h.probes[0].Coverage != "sealed" || h.probes[0].Generation != rb.ID || h.probes[0].Attempts != 9 {
				t.Fatalf("probes: %+v", h.probes)
			}
			// probed queued the return lane's park, in the source's region,
			// its horizon the return seal's bound.
			c = h.claim(t)
			lane := 0
			for _, o := range c.Obligations {
				if o.Kind == "park-replica" && o.Region == "ie" && o.Park != nil && o.Park.Location == moves.ScopeLocation(p1Host, rb.ID) {
					lane++
					if !o.Park.EvidenceAt.Equal(returnSeal.Bound()) || o.Park.RequestLimit != cutRequestLimit || o.Park.Incarnation != h.inc {
						t.Fatalf("the lane park's scope: %+v", o.Park)
					}
				}
				if o.Kind == "park-local" && o.Box == "ie-box" {
					t.Fatalf("the source's park-local is still owed: %+v", o)
				}
			}
			if lane != 1 {
				t.Fatalf("return lane parks: %d (%+v)", lane, c.Obligations)
			}
			if strings.Join(h.srcHanded, " ") != "restoring:"+rb.ID+":"+p1ID {
				t.Fatalf("hand-overs: %v", h.srcHanded)
			}
			if held := h.leases.holding(); len(held) != 0 {
				t.Fatalf("the source still holds %v", held)
			}
		})
	}
}

// TestVerifyMismatchIsStuck (INV-4): a clean return whose files, timers or
// warm database marker no longer match what the source recorded when it
// stopped stops for an operator with the baseline-mismatch sentence, at
// verified, before any probe; the row stays held. A retry verifies again,
// never skips it: unchanged, it stops again; repaired, it recovers.
func TestVerifyMismatchIsStuck(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		name          string
		breakIt, mend func(h *recoverHarness)
	}{
		{"files", func(h *recoverHarness) {
			os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("changed on the source"), 0o644)
		}, func(h *recoverHarness) {
			os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("hello"), 0o644)
		}},
		{"timers", func(h *recoverHarness) {
			h.timers = append(append([]registry.Timer(nil), h.timers...), registry.Timer{Host: p1Host, Name: "extra", Kind: "cron", Spec: "0 4 * * *", NextAt: 2000})
		}, func(h *recoverHarness) {
			h.timers = h.timers[:len(h.timers)-1]
		}},
		{"generation marker", func(h *recoverHarness) {
			os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("other\n"), 0o644)
		}, func(h *recoverHarness) {
			os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("cur\n"), 0o644)
		}},
	} {
		t.Run(c.name, func(t *testing.T) {
			h := newRecoverHarness(t, false)
			rb := h.released(t)
			c.breakIt(h)
			stuck := func() {
				t.Helper()
				m := h.read(t, p1Host).Move
				if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified {
					t.Fatalf("not stuck at verified: %+v %+v", m, m.Rollback)
				}
				if len(h.probes) != 0 {
					t.Fatalf("probed past a mismatch: %+v", h.probes)
				}
				if h.srcRoute.Status != "restoring" || h.srcRoute.HoldOp != rb.ID {
					t.Fatalf("the row: %+v", h.srcRoute)
				}
				if cl := h.claim(t); cl.Op.Phase != hostclaim.PhaseClaimed || cl.Op.Exec != "ie-box" {
					t.Fatalf("claim: %+v", cl.Op)
				}
			}
			h.srcPass(t)
			stuck()
			h.srcPass(t)
			stuck()
			// A retry verifies again: still broken, stuck again.
			h.retry(t)
			h.srcPass(t)
			stuck()
			c.mend(h)
			h.retry(t)
			h.srcPass(t)
			h.opened(t)
			if !strings.Contains(strings.Join(h.sEvents, " "), "stuck") {
				t.Fatalf("events %v", h.sEvents)
			}
		})
	}
}

// TestSourceRecoveryNeverReverifiesAfterProbing (Review Focus 2; INV-2,
// INV-7): once the record says probing, the recovery's own writes are
// authoritative: a probe that has not answered yet, or a restart, runs only
// the probe again, never the verification or the install, whatever the
// files say now. Nine failed attempts stop it for an operator at probing,
// and a retry probes again.
func TestSourceRecoveryNeverReverifiesAfterProbing(t *testing.T) {
	t.Parallel()
	h := newRecoverHarness(t, false)
	h.released(t)
	waiting := true
	h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
		if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
			return recoveryProbeWaiting, err
		}
		if waiting {
			return recoveryProbeWaiting, nil
		}
		return recoveryProbeAnswered, nil
	}
	h.srcPass(t)
	h.sourceAt(t, provision.RbSourceProbing)
	if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbing || c.Op.Exec != "ie-box" {
		t.Fatalf("the claim is not probing before the attempt: %+v", c.Op)
	}
	// The source's files and marker change under the probe (its app wrote).
	os.WriteFile(filepath.Join(h.sdir, p1Host, "upload.txt"), []byte("written by the recovered app"), 0o644)
	os.WriteFile(provision.DefaultDBPath(h.sdir, p1Host)+".gen", []byte("fresh\n"), 0o644)
	h.sw = nil // a restarted agent
	h.srcPass(t)
	h.sourceAt(t, provision.RbSourceProbing)
	// Nine attempts used: stuck at probing; the retry probes again.
	h.probe = func(recoveryProbeRequest) (recoveryProbeOutcome, error) { return recoveryProbeExhausted, nil }
	h.srcPass(t)
	m := h.read(t, p1Host).Move
	if m.State != provision.MoveStuck || m.StuckReason != provision.StuckRecoveryProbeAttempts || m.Rollback.StuckAt != provision.RbSourceProbing {
		t.Fatalf("not stuck at probing: %+v %+v", m, m.Rollback)
	}
	h.retry(t)
	waiting = false
	h.probe = func(req recoveryProbeRequest) (recoveryProbeOutcome, error) {
		if err := req.Record(context.Background(), h.renderedInputs()); err != nil {
			return recoveryProbeWaiting, err
		}
		return recoveryProbeAnswered, nil
	}
	h.srcPass(t)
	h.opened(t)
	if n := len(h.probes); n != 4 {
		t.Fatalf("probe attempts: %d", n)
	}
	verified := 0
	for _, p := range h.rcHits {
		if p == "verify" {
			verified++
		}
	}
	if verified != 1 {
		t.Fatalf("verified %d times (%v)", verified, h.rcHits)
	}
}

// TestRolledBackFinalizationReleasesTheClaim (§7.8 rolled_back): once the
// source has opened and the target has parked, and no obligation of the op
// remains, the source writes the terminal event, closes the record with the
// rolled_back outcome and releases the claim to serving, in that order
// (ruling M4). Until every chore is acknowledged it waits; a crash between
// the record's close and the release is the console's to finish
// (TestBoxClosesAreRecordFirst).
func TestRolledBackFinalizationReleasesTheClaim(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRecoverHarness(t, false)
	h.released(t)
	h.srcPass(t)
	h.opened(t)
	// The target cleans now that the source has recovered.
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
		t.Fatalf("the target did not clean: %+v", rb)
	}
	// The parks are acknowledged by the regional runner (C4), not here:
	// the source waits for them.
	h.srcPass(t)
	h.opened(t)
	owed := 0
	for _, o := range h.claim(t).Obligations {
		if o.Op == p1ID {
			owed++
		}
	}
	if owed == 0 {
		t.Fatal("the fixture owes no chore; the wait is not exercised")
	}
	if strings.Contains(strings.Join(h.sEvents, " "), "rolled_back") {
		t.Fatalf("finalized with chores owed: %v", h.sEvents)
	}
	for _, o := range h.claim(t).Obligations {
		if o.Op != p1ID {
			continue
		}
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
			t.Fatal(err)
		}
	}
	h.srcPass(t)
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
		t.Fatalf("the claim was not released: %+v", c.Op)
	}
	if err := hostclaim.RequireReleasedMove(p1ID)(c); err != nil {
		t.Fatal(err)
	}
	rec := h.read(t, p1Host)
	if rec.Move != nil || rec.LastMove == nil || rec.LastMove.ID != p1ID || rec.LastMove.Outcome != "rolled_back" || rec.Region != "ie" {
		t.Fatalf("not closed: region %s move %+v last %+v", rec.Region, rec.Move, rec.LastMove)
	}
	if n := strings.Count(strings.Join(h.sEvents, " "), "rolled_back"); n != 1 {
		t.Fatalf("the terminal event %d times: %v", n, h.sEvents)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindServing || c.Op.Exec != "ie-box" {
		t.Fatalf("claim after the close: %+v", c.Op)
	}
	// Nothing routes to the source any more.
	h.srcPass(t)
	if rec := h.read(t, p1Host); rec.Move != nil {
		t.Fatalf("record %+v", rec.Move)
	}
}

// Fix round 1, ruling 1: an evicted warm database. A clean return whose
// warm database and marker are both gone is restored cold from src_gen by
// its probe (J14), so verified passes; a database with no marker is refused
// unless the generation is the origin; a marker naming another generation
// is refused, with or without the database.
func TestCleanReturnWarmDatabaseMarker(t *testing.T) {
	t.Parallel()
	for _, c := range []struct {
		name          string
		db, marker    bool
		markerBody    string
		origin, stuck bool
	}{
		{name: "database and marker both evicted", db: false, marker: false},
		{name: "database without its marker", db: true, marker: false, stuck: true},
		{name: "database without its marker, the origin generation", db: true, marker: false, origin: true},
		{name: "a conflicting marker beside the database", db: true, marker: true, markerBody: "other\n", stuck: true},
		{name: "a conflicting marker, the database evicted", db: false, marker: true, markerBody: "other\n", stuck: true},
	} {
		t.Run(c.name, func(t *testing.T) {
			h := newRecoverHarness(t, false)
			h.released(t)
			db := provision.DefaultDBPath(h.sdir, p1Host)
			if c.origin {
				rewriteP1Claim(t, h.s, func(cl *hostclaim.Claim) { cl.Op.Journal[hostclaim.JournalSrcGen] = "" })
			}
			if !c.db {
				if err := os.Remove(db); err != nil {
					t.Fatal(err)
				}
			}
			if !c.marker {
				if err := os.Remove(db + ".gen"); err != nil {
					t.Fatal(err)
				}
			} else if err := os.WriteFile(db+".gen", []byte(c.markerBody), 0o644); err != nil {
				t.Fatal(err)
			}
			h.srcPass(t)
			if !c.stuck {
				h.opened(t)
				return
			}
			m := h.read(t, p1Host).Move
			if m.State != provision.MoveStuck || m.StuckReason != provision.StuckBaselineMismatch || m.Rollback.StuckAt != provision.RbSourceVerified || len(h.probes) != 0 {
				t.Fatalf("not stuck at verified: %+v %+v probes %d", m, m.Rollback, len(h.probes))
			}
		})
	}
}

// parkedJournal reports whether op's journal is live under the source's
// .moves/, and whether it was parked into a retirement's artifacts.
func (h *recoverHarness) parkedJournal(t *testing.T, op string) (live, parked bool) {
	t.Helper()
	live = moveJournalPresent(h.sdir, p1Host, op)
	suffix := filepath.Join("artifacts", ".moves", p1Host, op)
	filepath.WalkDir(filepath.Join(h.sdir, ".parked"), func(p string, d os.DirEntry, err error) error {
		if err == nil && d.IsDir() && strings.HasSuffix(p, suffix) {
			if _, err := os.Stat(filepath.Join(p, "journal.json")); err == nil {
				parked = true
			}
		}
		return nil
	})
	return live, parked
}

// closeOut drives a recovered source and its target to the close: the
// target cleans, the chores are acknowledged (the regional runner's, C4),
// and the source's next pass finalizes.
func (h *recoverHarness) closeOut(t *testing.T) {
	t.Helper()
	ctx := context.Background()
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetParkedAt == nil {
		t.Fatalf("the target did not park: %+v", rb)
	}
	for _, o := range h.claim(t).Obligations {
		if o.Op != p1ID {
			continue
		}
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, hostclaim.CompleteObligation(p1ID, o.ID)); err != nil {
			t.Fatal(err)
		}
	}
	h.srcPass(t)
}

// Fix round 1, ruling 2: the source keeps two journals through a rollback,
// the cut's under the move id and the recovery's under the rollback id (a
// dirty return's holds the install's build and extract scratch). The pass
// that closes the record parks both; a crash right after the close leaves
// them to the local journal sweep, which parks them after a restart on the
// close's evidence (no listed record carries the move). Until the close's
// preconditions hold, the sweep leaves them.
func TestSourceJournalsParkedAfterTheClose(t *testing.T) {
	t.Parallel()
	for _, dirty := range []bool{false, true} {
		name := "clean"
		if dirty {
			name = "dirty"
		}
		t.Run(name+", in the pass that closes", func(t *testing.T) {
			h := newRecoverHarness(t, dirty)
			rb := h.released(t)
			h.srcPass(t)
			h.opened(t)
			for _, op := range []string{p1ID, rb.ID} {
				if live, _ := h.parkedJournal(t, op); !live {
					t.Fatalf("the journal of %s is gone before the close's preconditions", op)
				}
			}
			h.srcPass(t)
			for _, op := range []string{p1ID, rb.ID} {
				if live, _ := h.parkedJournal(t, op); !live {
					t.Fatalf("the sweep parked the journal of %s while the target still holds its state", op)
				}
			}
			h.closeOut(t)
			if rec := h.read(t, p1Host); rec.Move != nil {
				t.Fatalf("not closed: %+v", rec.Move)
			}
			for _, op := range []string{p1ID, rb.ID} {
				if live, parked := h.parkedJournal(t, op); live || !parked {
					t.Fatalf("the journal of %s: live %v parked %v", op, live, parked)
				}
			}
		})
		t.Run(name+", a crash after the close", func(t *testing.T) {
			h := newRecoverHarness(t, dirty)
			rb := h.released(t)
			h.srcPass(t)
			h.opened(t)
			if dirty {
				if _, err := os.Stat(filepath.Join(moves.JournalDir(h.sdir, p1Host, rb.ID), "build")); err != nil {
					t.Fatalf("the dirty return's build scratch: %v", err)
				}
			}
			// The agent dies right after the close's write: nothing else of
			// that pass runs, the journal sweep included.
			crashed := false
			h.rcHook = func(point string) {
				if point == "after-close" {
					crashed = true
					panic(errCrash)
				}
			}
			func() {
				defer func() {
					if r := recover(); r != nil && r != errCrash {
						panic(r)
					}
				}()
				h.closeOut(t)
			}()
			h.rcHook = nil
			if !crashed {
				t.Fatalf("never reached the close (%v)", h.rcHits)
			}
			if rec := h.read(t, p1Host); rec.Move != nil {
				t.Fatalf("not closed: %+v", rec.Move)
			}
			for _, op := range []string{p1ID, rb.ID} {
				if live, _ := h.parkedJournal(t, op); !live {
					t.Fatalf("the journal of %s went with the crash", op)
				}
			}
			h.sw = nil // a restarted agent
			h.srcPass(t)
			for _, op := range []string{p1ID, rb.ID} {
				if live, parked := h.parkedJournal(t, op); live || !parked {
					t.Fatalf("the journal of %s after the sweep: live %v parked %v", op, live, parked)
				}
			}
			if entries, err := os.ReadDir(filepath.Join(h.sdir, ".moves", p1Host)); err == nil && len(entries) != 0 {
				t.Fatalf("journals left: %v", entries)
			}
		})
	}
}

// Fix round 1, ruling 3: every recovery step is replay-safe. A crash after
// each of the claim's return, the guard's activation, the hold's hand-over,
// the claim's probing and probed, and the opening of the row and of the
// claim (and after each record step), then a restarted worker, ends opened
// with one hold, opened once.
func TestSourceRecoveryReplays(t *testing.T) {
	slots := make(chan struct{}, 4)
	for _, point := range []string{
		"after-record-claimed", "after-return-claimed", "after-activate", "after-record-restoring", "after-hand",
		"verify", "after-record-verified", "after-claim-probing", "after-record-probing", "after-claim-probed",
		"after-record-recovered", "after-row-opened", "after-claim-opened", "after-record-opened",
	} {
		t.Run(point, func(t *testing.T) {
			t.Parallel()
			slots <- struct{}{}
			defer func() { <-slots }()
			h := newRecoverHarness(t, false)
			rb := h.released(t)
			h.rcAt = point
			h.srcPass(t)
			if !h.rcDone {
				t.Fatalf("never reached %s (%v)", point, h.rcHits)
			}
			h.sw = nil
			h.srcPass(t)
			h.opened(t)
			for _, hd := range h.srcHanded {
				if hd != "restoring:"+rb.ID+":"+p1ID {
					t.Fatalf("hand-overs %v", h.srcHanded)
				}
			}
			if captured, err := hostguard.Capture(h.sdir, p1Host); err != nil || captured.Stamp != 1 {
				t.Fatalf("the installation record: %+v %v", captured, err)
			}
		})
	}
}

// Fix round 1, ruling 3: a row held by another op is never taken: the hand-
// over at restoring and the open at recovered each refuse, and the record
// stays where it is.
func TestSourceRecoveryRefusesARowHeldByAnotherOp(t *testing.T) {
	t.Parallel()
	t.Run("at restoring", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		h.released(t)
		h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
		h.srcPass(t)
		h.sourceAt(t, provision.RbSourceRestoring)
		if h.srcRoute.HoldOp != "op_someoneelse" || len(h.probes) != 0 {
			t.Fatalf("the row %+v, probes %d", h.srcRoute, len(h.probes))
		}
	})
	t.Run("at recovered", func(t *testing.T) {
		h := newRecoverHarness(t, false)
		rb := h.released(t)
		h.rcHook = func(point string) {
			if point == "after-record-recovered" {
				h.srcRoute.Status, h.srcRoute.HoldOp = "restoring", "op_someoneelse"
			}
		}
		h.srcPass(t)
		h.sourceAt(t, provision.RbSourceRecovered)
		if h.srcRoute.HoldOp != "op_someoneelse" {
			t.Fatalf("the row %+v", h.srcRoute)
		}
		if c := h.claim(t); c.Op.Phase != hostclaim.PhaseProbed || c.Op.Intent["rollback"] != rb.ID {
			t.Fatalf("the claim opened over another op's row: %+v", c.Op)
		}
	})
}