package host

import (
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"io/fs"
	"log/slog"
	"os"
	"path/filepath"
	"slices"
	"sort"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/benbjohnson/litestream"
	lsfile "github.com/benbjohnson/litestream/file"
	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/appconfig"
	"github.com/carlosframework/platform/internal/dbverify"
	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostfiles"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// The landing harness: the cut harness with a REAL litestream replica in
// the source's lineage (so the target's build restores a real SQLite file
// that passes its check), driven to the flip, and a target box (uk-box)
// with its own data directory, the held row reconcile would have minted,
// the landing's hooks faked where main has no production provider yet
// (ruling G17) and real where it has one: park.Local and hostguard.Guard
// over the target's data directory, dbverify over the store, the hostbox
// marker writer, moves.BuildGeneration over the fake versioned bucket.
//
// fakeParkLeases is one lease table for the landing's own-location lease
// and the park engine's: a second acquisition of a held location is
// refused. The landing holds its lease from checking through published
// (spec §7.6), so its residue parks run under it through the claim's
// AggregateLeases and park.Engine.RunHeld (M5); an Engine.Run that took the
// lock again would fail the test.

type landingHarness struct {
	*cutHarness
	// carry is the target's moveCfg.CarryRequest (nil: none), for the return
	// lane's request carry (buildstate_carry_test.go).
	carry    func(ctx context.Context, host, bucket, prefix string) error
	tdir     string // the target's data directory
	leases   *fakeParkLeases
	lw       *movesWorker
	lcfg     moveCfg
	route    registry.Route
	rows     int // rows the source database holds
	n        uint64
	replaced [][]registry.Timer
	repl     []string
	nextDue  []int64
	renders  int
	render   func() (appconfig.HeldRender, error)
	// renderHook, when set, is the whole held render: it gets the
	// before-write callback the installer passes (fix round 1, I1).
	renderHook func(before func(string) error) (appconfig.HeldRender, error)
	release    moves.Check
	landAt     string
	landHits   []string
	landDone   bool
	parkAt     string
	parkHits   []string
	parkDone   bool
	parkView   func(context.Context, hostclaim.Observed) (park.View, error)
	landFault  func(point string) error
	store      lease.VersionedStore // when set, the target's versioned store

	// renderPin, when set, sees the pins each render is asked to hold to
	// (moves Task 7).
	renderPin func(pin *provision.MoveInputs)
	// releaseLive, when set, is the configured step's release
	// (ReleaseLive, moves Task 7).
	releaseLive func(ctx context.Context, s release.Store, rec provision.Record, version string) moves.Check

	// probe, when set, is the target's probe wake (Plan 3 Task 6).
	probe func(context.Context, activator.ProbeConfig) (activator.ProbeResult, error)
	// probeLive, when set, is the activator's live probe entry.
	probeLive func(string) (string, activator.ProbeResult, bool)
	// openLog records the target's open (Plan 3 Task 7): the probe entry's
	// mark and the hold's release, in order, with the record and claim
	// state each saw.
	openLog []string
	// noRelease leaves ReleaseHeld unwired (the open unarmed).
	noRelease bool
	// holdProbed stops every pass at probed, before the commit (Task 6's
	// tests, which end there).
	holdProbed bool
	// releaseNoop makes ReleaseHeld match nothing and change nothing.
	releaseNoop bool
	// releaseErr makes ReleaseHeld fail.
	releaseErr error
	// ownAuthorize drops the park authorizer the target would inherit from
	// the source's configuration (the finalization harness sets one), so
	// the target's parks use the allowing fake behind their own checks
	// (Plan 5 Task 3).
	ownAuthorize bool
	// parkLeases, runHeld and ownLease, when set, replace the park fakes'
	// leases, the claim's RunHeld (park.Engine.RunHeld, M5) and the
	// landing's own-location lease.
	parkLeases park.Leases
	runHeld    func(ctx context.Context, e park.Engine, w park.Work, held park.HeldLease) error
	ownLease   func(ctx context.Context, bucket, location string) (park.HeldLease, error)
}

// fakeParkLeases refuses a second acquisition of a held location.
type fakeParkLeases struct {
	mu       sync.Mutex
	held     map[string]bool
	leases   map[string]*fakeHeldLease
	acquired []string
}

// fakeHeldLease is a park.HeldLease with the Context and LeaseScope methods
// hostbox's renewingHeld has: a context derived from it is cancelled when
// the lease is lost or released, and the claim's AggregateLeases lends it to
// RunHeld for its own bucket and scope.
type fakeHeldLease struct {
	l             *fakeParkLeases
	key           string
	bucket, scope string
	ctx           context.Context
	cancel        context.CancelFunc
}

// LeaseScope is renewingHeld.LeaseScope: the bucket and scope it holds.
func (h *fakeHeldLease) LeaseScope() (string, string) { return h.bucket, h.scope }

func (l *fakeParkLeases) Acquire(_ context.Context, bucket, scope string) (park.HeldLease, error) {
	l.mu.Lock()
	defer l.mu.Unlock()
	key := bucket + "/" + scope
	if l.held[key] {
		return nil, &lease.HeldError{Owner: "the landing"}
	}
	if l.held == nil {
		l.held = map[string]bool{}
	}
	l.held[key] = true
	l.acquired = append(l.acquired, key)
	ctx, cancel := context.WithCancel(context.Background())
	h := &fakeHeldLease{l: l, key: key, bucket: bucket, scope: scope, ctx: ctx, cancel: cancel}
	if l.leases == nil {
		l.leases = map[string]*fakeHeldLease{}
	}
	l.leases[key] = h
	return h, nil
}

// Context is renewingHeld.Context: ctx, cancelled too when the lease is.
func (h *fakeHeldLease) Context(ctx context.Context) context.Context {
	c, cancel := context.WithCancel(ctx)
	stop := context.AfterFunc(h.ctx, cancel)
	context.AfterFunc(c, func() { stop() })
	return c
}

func (l *fakeParkLeases) holding() []string {
	l.mu.Lock()
	defer l.mu.Unlock()
	var out []string
	for k, on := range l.held {
		if on {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}

func (h *fakeHeldLease) Check(context.Context) error {
	h.l.mu.Lock()
	defer h.l.mu.Unlock()
	if !h.l.held[h.key] {
		return errors.New("fake lease: released")
	}
	return nil
}
func (h *fakeHeldLease) Release(context.Context) error {
	h.l.mu.Lock()
	defer h.l.mu.Unlock()
	delete(h.l.held, h.key)
	h.cancel()
	return nil
}
func (h *fakeHeldLease) Close() { _ = h.Release(context.Background()) }

// realReplica writes a small SQLite database with rows rows over syncs
// litestream syncs (a level-9 snapshot too) and returns the file replica
// client and the newest level-0 position, as Task 8's test does.
func realReplica(t *testing.T, syncs int) (litestream.ReplicaClient, uint64, int) {
	t.Helper()
	dir := t.TempDir()
	path := filepath.Join(dir, "src.db")
	rows := 0
	write := func(n int) {
		d, err := sql.Open("sqlite", path)
		if err != nil {
			t.Fatal(err)
		}
		defer d.Close()
		if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil {
			t.Fatal(err)
		}
		for i := 0; i < n; i++ {
			if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil {
				t.Fatal(err)
			}
		}
		rows += n
	}
	write(1)
	client := lsfile.NewReplicaClient(filepath.Join(dir, "replica"))
	db := litestream.NewDB(path)
	db.Logger = slog.New(slog.NewTextHandler(io.Discard, nil))
	db.SetMetaPath(filepath.Join(dir, "meta"))
	db.MonitorInterval = time.Hour
	db.Replica = litestream.NewReplicaWithClient(db, client)
	db.Replica.SyncInterval = time.Hour
	if err := db.Open(); err != nil {
		t.Fatal(err)
	}
	ctx := context.Background()
	for i := 0; i < syncs; i++ {
		if err := db.SyncAndWait(ctx); err != nil {
			t.Fatal(err)
		}
		write(2)
	}
	if err := db.SyncAndWait(ctx); err != nil {
		t.Fatal(err)
	}
	if _, err := db.Snapshot(ctx); err != nil {
		t.Fatal(err)
	}
	if err := db.Close(ctx); err != nil {
		t.Fatal(err)
	}
	var n uint64
	for _, f := range replicaLevelFiles(t, client, 0) {
		if uint64(f.MaxTXID) > n {
			n = uint64(f.MaxTXID)
		}
	}
	if n == 0 {
		t.Fatal("the replica has no level-0 files")
	}
	return client, n, rows
}

func replicaLevelFiles(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo {
	t.Helper()
	itr, err := client.LTXFiles(context.Background(), level, 0, false)
	if err != nil {
		t.Fatal(err)
	}
	infos, err := ltx.SliceFileIterator(itr)
	if err != nil {
		t.Fatal(err)
	}
	return infos
}

func countRowsForTest(t *testing.T, path string) int {
	t.Helper()
	d, err := sql.Open("sqlite", "file:"+path+"?mode=ro")
	if err != nil {
		t.Fatal(err)
	}
	defer d.Close()
	var n int
	if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil {
		t.Fatal(err)
	}
	return n
}

func newLandingHarness(t *testing.T, opts ...p1Option) *landingHarness {
	t.Helper()
	h := &landingHarness{cutHarness: newCutHarness(t, opts...), leases: &fakeParkLeases{}, release: moves.Check{OK: true}}
	// The real replica replaces the harness's fake level files under the
	// source's lineage; the final checkpoint adds nothing (the database is
	// quiet at N).
	client, n, rows := realReplica(t, 2)
	h.n, h.rows = n, rows
	loc := "instances/" + p1Host + "/" + h.inc + "/"
	for k := range h.f.buckets[h.srcBucket] {
		if strings.HasPrefix(k, loc+"g-cur/") && strings.HasSuffix(k, ".ltx") {
			delete(h.f.buckets[h.srcBucket], k)
		}
	}
	for level := 0; level <= litestream.SnapshotLevel; level++ {
		for _, info := range replicaLevelFiles(t, client, level) {
			rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0)
			if err != nil {
				t.Fatal(err)
			}
			body, err := io.ReadAll(rc)
			rc.Close()
			if err != nil {
				t.Fatal(err)
			}
			key := loc + "g-cur/" + fmt.Sprintf("%04x/", level) + ltx.FormatFilename(info.MinTXID, info.MaxTXID)
			h.f.store(h.srcBucket, key, fakeMoveObject{lease.MirrorObject{ETag: "r-" + filepath.Base(key), Size: int64(len(body))}, string(body)})
		}
	}
	h.localPos = n
	h.cfg.FinalCheckpoint = func(ctx context.Context, _, _ string, _ hostclaim.Holder, check func(context.Context) error) error {
		return check(ctx)
	}
	// The target's data directory, with the two directories the guard and
	// park.Local need, and the held row reconcile minted.
	h.tdir = t.TempDir()
	for _, d := range []string{"hostguard", ".parked"} {
		if err := os.Mkdir(filepath.Join(h.tdir, d), 0o770); err != nil {
			t.Fatal(err)
		}
	}
	h.route = registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Hibernate: true, Status: "landing", HoldOp: p1ID,
		Socket: provision.DefaultSocket(p1Host), DBPath: provision.DefaultDBPath(h.tdir, p1Host), Provisioned: true}
	h.render = func() (appconfig.HeldRender, error) {
		in := h.read(t, p1Host).Move.Inputs
		if in == nil {
			return appconfig.HeldRender{}, errors.New("no pinned inputs")
		}
		return appconfig.HeldRender{Path: filepath.Join(h.tdir, p1Host+".json"), SHA256: "x", BundleETags: in.BundleETags, OverlayETag: in.OverlayETag}, nil
	}
	// The restore-verification record names the host: local removes it.
	if err := dbverify.Publish(context.Background(), h.s, &dbverify.Record{V: 1, Box: h.dstBox, CheckedAt: h.clock.Now(),
		Hosts: map[string]dbverify.Verdict{p1Host: {OK: true, VerifiedAt: h.clock.Now()}, "other.example": {OK: true, VerifiedAt: h.clock.Now()}}}); err != nil {
		t.Fatal(err)
	}
	h.lcfg = h.landingCfg(t)
	return h
}

// landingCfg is the target's configuration with the landing armed.
func (h *landingHarness) landingCfg(t *testing.T) moveCfg {
	t.Helper()
	c := h.target()
	c.DataDir = h.tdir
	c.CarryRequest = h.carry
	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.route
		return &rt, nil
	}
	c.FilesDir = func(host string) (string, string) {
		return filepath.Join(h.tdir, host), filepath.Join(h.tdir, host+".env")
	}
	c.Guard = hostguard.Guard{DataDir: h.tdir, LockDir: t.TempDir()}
	c.Local = park.Local{DataDir: h.tdir, Box: h.dstBox}
	inherited := c.ParkAuthorize
	armParkFakes(&c, h.s, h.leases, park.PlacementRegion(c.Region), roleTarget)
	if h.parkView != nil {
		c.ParkView = func(ctx context.Context, obs hostclaim.Observed) (park.View, error) { return h.parkView(ctx, obs) }
	}
	if inherited != nil && !h.ownAuthorize {
		c.ParkAuthorize = inherited
	}
	if h.parkLeases != nil {
		c.ParkLeases = h.parkLeases
	}
	if h.runHeld != nil {
		c.ParkRunHeld = h.runHeld
	}
	c.OwnLease = h.leases.Acquire
	if h.ownLease != nil {
		c.OwnLease = h.ownLease
	}
	if h.store != nil {
		c.Versioned = func(context.Context, string, string) (lease.VersionedStore, error) { return h.store, nil }
	}
	c.RenderHeld = func(_ context.Context, _ release.Store, _ registry.Route, _ string, pin *provision.MoveInputs, before func(string) error) (appconfig.HeldRender, error) {
		h.renders++
		if h.renderPin != nil {
			h.renderPin(pin)
		}
		if h.renderHook != nil {
			return h.renderHook(before)
		}
		hr, err := h.render()
		if hr.SHA256 != "" {
			if berr := before(hr.SHA256); berr != nil {
				return appconfig.HeldRender{}, berr
			}
		}
		return hr, err
	}
	c.ReleaseReady = func(context.Context, release.Store, provision.Record, string) moves.Check { return h.release }
	c.ReleaseLive = h.releaseLive
	c.ReplaceTimers = func(_ context.Context, host string, rows []registry.Timer) error {
		h.replaced = append(h.replaced, append([]registry.Timer(nil), rows...))
		return nil
	}
	c.SetReplication = func(host, class string) error { h.repl = append(h.repl, class); return nil }
	c.SetNextDue = func(host string, due int64) error { h.nextDue = append(h.nextDue, due); return nil }
	c.RemoveVerifyEntry = dbverify.RemoveHost
	c.WriteGen = hostbox.WriteGenerationMarker
	if h.probe != nil {
		c.Probe = h.probe
	}
	if h.probeLive != nil {
		c.ProbeLive = h.probeLive
	}
	c.ReleaseHeld = func(host, op string) (bool, error) {
		h.openLog = append(h.openLog, "release:"+h.openState(t))
		if h.releaseNoop {
			return false, nil
		}
		if h.releaseErr != nil {
			return false, h.releaseErr
		}
		if host != p1Host || !registry.IsHold(h.route.Status) || h.route.HoldOp != op {
			return false, nil
		}
		h.route.Status, h.route.HoldOp = "ready", ""
		return true, nil
	}
	if h.noRelease {
		c.ReleaseHeld = nil
	}
	c.ConfirmProbeOpen = func(host, op string) bool {
		h.openLog = append(h.openLog, "confirm:"+h.openState(t))
		return host == p1Host && op == p1ID
	}
	c.MarkProbeOpen = func(host, op string) bool {
		h.openLog = append(h.openLog, "mark:"+h.openState(t))
		return host == p1Host && op == p1ID
	}
	c.landFault = func(point string) error {
		h.landHits = append(h.landHits, point)
		if h.holdProbed && point == "before-commit" {
			return errHoldProbed
		}
		if h.landFault != nil {
			if err := h.landFault(point); err != nil {
				return err
			}
		}
		if point == h.landAt && !h.landDone {
			h.landDone = true
			return errCrash
		}
		return nil
	}
	c.parkFault = func(step string) error {
		h.parkHits = append(h.parkHits, step)
		if step == h.parkAt && !h.parkDone {
			h.parkDone = true
			return errCrash
		}
		return nil
	}
	return c
}

// flip drives the cut to the flip: the record landing/checking in uk and
// the claim landing/checking under uk-box.
func (h *landingHarness) flip(t *testing.T) {
	t.Helper()
	h.ready(t)
	h.cut(t)
	h.pass(t)
	if rec := h.read(t, p1Host); rec.Region != h.dst || rec.Move.State != provision.MoveLanding || rec.Move.LandingPhase != provision.LandChecking {
		t.Fatalf("the fixture did not flip: %s %+v", rec.Region, rec.Move)
	}
}

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

// crashLand runs one target pass that must reach the landing crash point.
func (h *landingHarness) crashLand(t *testing.T) {
	t.Helper()
	h.land(t)
	if !h.landDone {
		t.Fatalf("the pass never reached %s (points reached: %v)", h.landAt, h.landHits)
	}
}

// freshTarget drops the target worker, as a restarted agent would, and
// with it every lease the dead process held: a real lease lapses at its
// TTL once nobody renews it, and the restarted worker acquires afresh.
func (h *landingHarness) freshTarget() {
	h.lw = nil
	h.leases.expire()
}

// expire lapses every held lease, as the TTL does for a dead holder, and
// cancels every context derived from them.
func (l *fakeParkLeases) expire() {
	l.mu.Lock()
	defer l.mu.Unlock()
	l.held = map[string]bool{}
	for _, h := range l.leases {
		h.cancel()
	}
}

func (h *landingHarness) landJournal(t *testing.T) *moves.Journal {
	t.Helper()
	j, err := moves.ReadJournal(h.tdir, p1Host, p1ID)
	if err != nil {
		t.Fatal(err)
	}
	return j
}

func (h *landingHarness) filesGen(t *testing.T) (filesGenMarker, bool) {
	t.Helper()
	body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, archive.MarkerName))
	if errors.Is(err, os.ErrNotExist) {
		return filesGenMarker{}, false
	}
	if err != nil {
		t.Fatal(err)
	}
	var m filesGenMarker
	if err := json.Unmarshal(body, &m); err != nil {
		t.Fatal(err)
	}
	return m, true
}

// installed asserts the end state of a landing through restored.
func (h *landingHarness) installed(t *testing.T) moves.InstallManifest {
	t.Helper()
	rec := h.read(t, p1Host)
	m := rec.Move
	if m.State != provision.MoveLanding || m.LandingPhase != provision.LandRestored || m.Reason != "" {
		t.Fatalf("record: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != provision.LandRestored || c.Op.Exec != h.dstBox || !containsString(c.Touched.Boxes, h.dstBox) {
		t.Fatalf("claim: %+v touched %+v", c.Op, c.Touched)
	}
	j := h.landJournal(t)
	var im moves.InstallManifest
	if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &im); err != nil || im.Op != p1ID || im.N != h.n {
		t.Fatalf("install manifest in the journal: %v %+v", err, im)
	}
	if m.Landing == nil || m.Landing.InstalledDBSHA256 != "sha256:"+im.DBSHA256 {
		t.Fatalf("installed_db_sha256: %+v want %s", m.Landing, im.DBSHA256)
	}
	dbPath := provision.DefaultDBPath(h.tdir, p1Host)
	if got, err := fileSHA256(dbPath); err != nil || got != "sha256:"+im.DBSHA256 {
		t.Fatalf("the installed database: %v %s want %s", err, got, im.DBSHA256)
	}
	if got := countRowsForTest(t, dbPath); got != h.rows {
		t.Fatalf("the installed database holds %d rows, the source wrote %d", got, h.rows)
	}
	if gen, err := hostbox.ReadGenerationMarker(dbPath + ".gen"); err != nil || gen != p1ID {
		t.Fatalf("%s.gen: %q %v", dbPath, gen, err)
	}
	if fg, ok := h.filesGen(t); !ok || fg.Op != p1ID || fg.Incarnation != h.inc {
		t.Fatalf(".files-gen: %+v present=%v", fg, ok)
	}
	if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil || string(body) != "hello" {
		t.Fatalf("the extracted files: %q %v", body, err)
	}
	if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host+".env")); err != nil || string(body) != h.srcEnvFile {
		t.Fatalf("the extracted env: %q %v", body, err)
	}
	if held := h.leases.holding(); len(held) != 0 {
		t.Fatalf("the landing still holds %v", held)
	}
	if _, err := os.Stat(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build", p1Host+".db")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the build scratch was not renamed away: %v", err)
	}
	return im
}

// The end to end landing through restored: the generation built from the
// seal, the archive extracted, timers replaced with their registrations,
// the carried columns set, the verify entry gone, .files-gen and
// <dbPath>.gen written, installed_db_sha256 recorded, the lease released.
func TestLandingBuildsInstallsAndRestores(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newLandingHarness(t)
	h.timers = []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000, Attempts: 2, Registration: "reg-1"},
		{Host: p1Host, Name: "once", Kind: "once", NextAt: 2000, DueAt: 2000, Registration: "reg-2"}}
	h.flip(t)
	h.route.Replication = ""
	h.land(t)
	im := h.installed(t)
	// The generation holds exactly the build's four objects, at usable
	// versions, in this box's own bucket.
	prefix := moves.GenerationPrefix(cutRoot, p1ID)
	var keys []string
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			keys = append(keys, strings.TrimPrefix(k, prefix))
		}
	}
	sort.Strings(keys)
	want := []string{"0000/" + ltx.FormatFilename(ltx.TXID(h.n), ltx.TXID(h.n)), "0009/" + ltx.FormatFilename(1, ltx.TXID(h.n)), "build-intent.json", "install.json"}
	if strings.Join(keys, " ") != strings.Join(want, " ") {
		t.Fatalf("generation objects %v, want %v", keys, want)
	}
	if !lease.UsableVersionID(im.SnapshotVersion) || !lease.UsableVersionID(im.AnchorVersion) {
		t.Fatalf("manifest versions %q %q", im.SnapshotVersion, im.AnchorVersion)
	}
	// Timers: the sealed snapshot's rows, registrations included, once.
	if len(h.replaced) != 1 || len(h.replaced[0]) != 2 {
		t.Fatalf("ReplaceTimers calls: %+v", h.replaced)
	}
	if moves.TimersDigest(h.replaced[0]) != h.read(t, p1Host).Move.Source.TimersSHA256 {
		t.Fatalf("the replaced rows do not digest to the record's: %+v", h.replaced[0])
	}
	byName := map[string]registry.Timer{}
	for _, r := range h.replaced[0] {
		byName[r.Name] = r
	}
	if byName["nightly"].Registration != "reg-1" || byName["nightly"].Attempts != 2 || byName["once"].Registration != "reg-2" || byName["once"].DueAt != 2000 {
		t.Fatalf("registrations and bookkeeping not preserved: %+v", h.replaced[0])
	}
	src := h.read(t, p1Host).Move.Source
	if len(h.repl) != 1 || h.repl[0] != src.Replication || len(h.nextDue) != 1 || h.nextDue[0] != src.NextDue {
		t.Fatalf("columns: replication %v next_due %v, source %+v", h.repl, h.nextDue, src)
	}
	dv, err := dbverify.Load(ctx, h.s)
	if err != nil || dv == nil {
		t.Fatal(err)
	}
	if _, ok := dv.Hosts[p1Host]; ok || len(dv.Hosts) != 1 {
		t.Fatalf("the verify entry: %+v", dv.Hosts)
	}
	// The inventory names the four flat paths under F, the flip's stamp.
	c := h.claim(t)
	inv := c.Inventory["uk-box"]
	if inv.Layout != "flat" || len(inv.Paths) != 4 {
		t.Fatalf("inventory: %+v", inv)
	}
	for _, p := range []string{p1Host + ".db", p1Host + ".db.gen", p1Host, p1Host + ".env"} {
		if inv.Paths[p] != c.Holder.Stamp {
			t.Fatalf("inventory path %s at %d, want F %d", p, inv.Paths[p], c.Holder.Stamp)
		}
	}
	captured, err := hostguard.Capture(h.tdir, p1Host)
	if err != nil || captured.Stamp != c.Holder.Stamp || captured.Incarnation != h.inc {
		t.Fatalf("the installation record: %+v %v, want stamp %d", captured, err, c.Holder.Stamp)
	}
	// No residue, so no park chores; and no park-local (no earlier stay).
	if len(c.Obligations) != 0 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	// The journal records every sub-step, the prepared manifest as skipped.
	j := h.landJournal(t)
	for _, step := range []string{landStepResidue, landStepInstall, landStepPublished, landStepRetired, landStepExtracted, landStepInventory, landStepTimers, landStepColumns, landStepVerify, landStepFilesManifest, landStepFilesGen} {
		if !j.Done(step) {
			t.Fatalf("journal lacks %s: %v", step, j.Steps)
		}
	}
	if h.renders != 1 {
		t.Fatalf("renders %d", h.renders)
	}
	// Restored is Plan 3's: another pass changes nothing.
	before := h.read(t, p1Host).Move.UpdatedAt
	h.land(t)
	if after := h.read(t, p1Host).Move; after.LandingPhase != provision.LandRestored || !after.UpdatedAt.Equal(before) {
		t.Fatalf("a pass at restored wrote the record: %+v", after)
	}
}

// checking (§7.6): the claim must be this move's landing op with this box
// executing, citing complete source evidence, and this box is added to
// touched.boxes before any local write; the installation record is
// activated at F. A claim the source left at moving/sealed (a crash after
// the record's flip) is flipped by the target itself (ruling G12).
func TestLandingCheckingRequiresClaimEvidenceAndTouch(t *testing.T) {
	ctx := context.Background()
	t.Run("the target writes the flip's claim half", func(t *testing.T) {
		h := newLandingHarness(t)
		h.ready(t)
		h.crashAt = "after-record-flip"
		h.cut(t)
		h.crashPass(t)
		if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Phase != hostclaim.PhaseSealed {
			t.Fatalf("fixture: %+v", c.Op)
		}
		h.landAt = "after-target-flip"
		h.crashLand(t)
		c := h.claim(t)
		if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != provision.LandChecking || c.Op.Exec != "uk-box" {
			t.Fatalf("after the target's flip: %+v", c.Op)
		}
		if c.Op.Journal[hostclaim.JournalSourceStamp] != "1" || c.Op.Journal[hostclaim.JournalSrcGen] != "cur" {
			t.Fatalf("the target's flip did not journal the source's stamp and generation: %+v", c.Op.Journal)
		}
		h.land(t)
		h.installed(t)
	})
	t.Run("another executor is refused", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		// Refused before any claim write: with uk-box taken out of touched,
		// the K12 Touch that follows the executor check must not run (the
		// guard would refuse the activation too, but later and locally).
		rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Op.Exec = "other-box"; c.Touched.Boxes = []string{"ie-box"} })
		h.land(t)
		if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
			t.Fatalf("advanced under another executor: %+v", m)
		}
		if _, err := hostguard.Capture(h.tdir, p1Host); !errors.Is(err, hostguard.ErrNoRecord) {
			t.Fatalf("the installation record was written: %v", err)
		}
		if c := h.claim(t); containsString(c.Touched.Boxes, "uk-box") {
			t.Fatal("a claim write under another executor")
		}
	})
	t.Run("missing evidence is refused", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		if err := h.s.Delete(ctx, h.stoppedKey()); err != nil {
			t.Fatal(err)
		}
		h.land(t)
		if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
			t.Fatalf("advanced without the source's evidence: %+v", m)
		}
	})
	t.Run("touch and the installation record", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		// Plan 1's target report touched uk-box at its first staging write;
		// checking's Touch is the idempotent K12 re-add, so take it away
		// first to see it written.
		rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Touched.Boxes = []string{"ie-box"} })
		h.landAt = "after-record-parking"
		h.crashLand(t)
		c := h.claim(t)
		if !containsString(c.Touched.Boxes, "uk-box") || !containsString(c.Touched.Regions, "ie") {
			t.Fatalf("touched: %+v", c.Touched)
		}
		captured, err := hostguard.Capture(h.tdir, p1Host)
		if err != nil || captured.Stamp != c.Holder.Stamp || captured.Incarnation != h.inc {
			t.Fatalf("active_stamp %+v %v, want F %d", captured, err, c.Holder.Stamp)
		}
		// Spec §7.6 checking (Plan 5 Task 9a, M5): the location's lease in
		// the target's own bucket, taken once and held into parking.
		want := "data-uk/" + strings.TrimSuffix(c.Holder.Location, "/")
		if len(h.leases.acquired) != 1 || h.leases.acquired[0] != want {
			t.Fatalf("checking took %v, want the location's lease %s", h.leases.acquired, want)
		}
		if held := h.leases.holding(); len(held) != 1 || held[0] != want {
			t.Fatalf("after checking the landing holds %v", held)
		}
	})
	t.Run("a nested row is refused for a rollback", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		h.route.DBPath = filepath.Join(h.tdir, "instances", p1Host, "db", "instance.db")
		h.land(t)
		if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackLandingLayout || m.LandingPhase != provision.LandChecking || m.State != provision.MoveLanding {
			t.Fatalf("a nested row: %+v", m)
		}
	})
	t.Run("host claims on refuse at checking", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		t.Setenv("CARLOS_HOST_CLAIM", "1")
		h.land(t)
		if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackClaimWired || m.LandingPhase != provision.LandChecking {
			t.Fatalf("under CARLOS_HOST_CLAIM: %+v", m)
		}
	})
}

// residue plants an earlier stay of this incarnation at the holder's
// location in the target's bucket: two generations (old and the origin)
// and a generation.json, each at a usable version.
func (h *landingHarness) residue(t *testing.T) []string {
	t.Helper()
	ctx := context.Background()
	keys := []string{
		cutRoot + "/g-old/0000/0000000000000001-0000000000000001.ltx",
		cutRoot + "/g-old/0009/0000000000000001-0000000000000001.ltx",
		cutRoot + "/0000/0000000000000001-0000000000000001.ltx",
		cutRoot + "/generation.json",
	}
	for _, k := range keys {
		if _, err := h.vs.PutIfAbsent(ctx, h.dstBucket, k, []byte("residue "+k)); err != nil {
			t.Fatal(err)
		}
	}
	return keys
}

// parking (§7.6, D6): an earlier stay's objects under holder.location are
// archived under parked/<inc>/<host>/<ret>/instances/... by one
// park-generation chore per generation before building; a crash inside the
// engine leaves the phase at parking and the replay finishes; the chores
// stay pending until the horizon; after the probe makes the new generation
// live here, Engine.Run on them acknowledges rather than refusing.
func TestLandingParkingWaitsForTheManifest(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newLandingHarness(t)
	h.flip(t)
	keys := h.residue(t)
	h.parkAt = "copy"
	h.land(t)
	if !h.parkDone {
		t.Fatalf("the engine never copied (%v)", h.parkHits)
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
		t.Fatalf("after the crash: %+v", m)
	}
	c := h.claim(t)
	var gens []string
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-generation" && o.Region == "uk" && o.Park != nil && o.Park.Location == cutRoot && o.Park.EvidenceAt.Equal(h.read(t, p1Host).Move.FlippedAt.UTC()) {
			gens = append(gens, o.Generation)
		}
	}
	sort.Strings(gens)
	if strings.Join(gens, " ") != hostclaim.RootGeneration+" old" {
		t.Fatalf("residue chores: %v (%+v)", gens, c.Obligations)
	}
	if len(h.leases.acquired) == 0 {
		t.Fatal("the engine took no lease")
	}
	// The replay archives and writes the manifest, but one delete the store
	// accepts does not land (its listing still shows the version): the
	// manifest exists and its deletion set is not gone, so Parked is false
	// and the phase waits, chores queued, nothing built.
	stuck := keys[0]
	h.vs.skipDelete = func(key string) bool { return key == stuck }
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
		t.Fatalf("advanced with a deletion outstanding: %+v", m)
	}
	if _, ok := h.f.buckets["data-uk"][stuck]; !ok {
		t.Fatalf("fixture: %s was deleted", stuck)
	}
	manifests := 0
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") && strings.HasSuffix(k, "/manifest.json") {
			manifests++
		}
	}
	if manifests != 2 {
		t.Fatalf("%d park manifests after the second pass", manifests)
	}
	if n := len(h.vs.callsOn(moves.GenerationPrefix(cutRoot, p1ID))); n != 0 {
		t.Fatalf("the build ran while parking waited: %d calls", n)
	}
	// The deletes land: the deletion set is gone, the landing goes on to
	// restored. The chores are still owed.
	h.vs.skipDelete = nil
	h.land(t)
	h.installed(t)
	c = h.claim(t)
	var retirements []string
	for _, o := range c.Obligations {
		if o.Kind == "park-generation" {
			retirements = append(retirements, o.Retirement)
		}
	}
	if len(retirements) != 2 {
		t.Fatalf("chores after the landing: %+v", c.Obligations)
	}
	// The marker is the location's one reused key: it is not parked (the
	// probe supersedes it), every other object is.
	marker := cutRoot + "/generation.json"
	if _, ok := h.f.buckets["data-uk"][marker]; !ok {
		t.Fatal("the earlier stay's generation.json was parked")
	}
	for _, k := range keys {
		if _, ok := h.f.buckets["data-uk"][k]; ok && k != marker {
			t.Fatalf("%s is still at the location", k)
		}
	}
	var archived []string
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/") {
			archived = append(archived, k)
		}
	}
	sort.Strings(archived)
	if len(archived) != len(keys)-1+2 { // every object but the marker, plus two manifests
		t.Fatalf("archived: %v", archived)
	}
	for _, k := range keys {
		if k == marker {
			continue
		}
		rel := strings.TrimPrefix(k, "instances/")
		found := false
		for _, a := range archived {
			if strings.Contains(a, "/instances/"+strings.TrimPrefix(rel, p1Host+"/")+"@") {
				found = true
			}
		}
		if !found {
			t.Fatalf("%s was not archived under parked/<inc>/<host>/<ret>/instances/ (%v)", k, archived)
		}
	}
	// The probe's admission (Plan 3) makes g-<move id> live here: the
	// holder's generation is the move id, the marker names it and pins the
	// snapshot, and uk is authoritative. Past the horizon the chores
	// acknowledge: their generations are not the live one.
	j := h.landJournal(t)
	var im moves.InstallManifest
	if err := json.Unmarshal([]byte(j.Steps[landStepInstall]), &im); err != nil {
		t.Fatal(err)
	}
	// The probe writes the marker If-Match over the earlier stay's: a new
	// current version at the same key.
	markerVer := h.f.nextVersion()
	h.f.store("data-uk", marker, fakeMoveObject{lease.MirrorObject{ETag: "etag-" + markerVer, Size: 20}, `{"op":"` + p1ID + `"}`})
	h.f.setVersion("data-uk", marker, markerVer)
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Generation = p1ID })
	h.clock.Advance(park.SignatureAge + landingRequestLimit + time.Minute)
	view := func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
		return park.View{ClaimETag: obs.ETag, Region: "uk", AuthoritativeRegion: "uk", MarkerKnown: true, MarkerGeneration: p1ID,
			Marker:       park.Ref{Key: cutRoot + "/generation.json", VersionID: markerVer},
			Pinned:       map[park.Ref]bool{{Key: im.SnapshotKey, VersionID: im.SnapshotVersion}: true},
			GenerationOf: func(r park.Ref) (string, bool) { return landingGenerationOf(cutRoot, r.Key) }}, nil
	}
	l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
	eng := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-uk", Region: "uk", Leases: h.leases, Now: h.clock.Now, NewID: obligationID, View: view, Authorize: l.authorizeObligation}
	for _, o := range c.Obligations {
		if o.Kind != "park-generation" {
			continue
		}
		if err := eng.Run(ctx, park.Work{Host: p1Host, ID: o.ID, Op: p1ID}); err != nil {
			t.Fatalf("acknowledge generation %s: %v", o.Generation, err)
		}
	}
	if c := h.claim(t); len(c.Obligations) != 0 {
		t.Fatalf("chores after the acknowledgement: %+v", c.Obligations)
	}
	prefix := moves.GenerationPrefix(cutRoot, p1ID)
	n := 0
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			n++
		}
	}
	if n != 4 {
		t.Fatalf("the acknowledgement touched the live generation: %d objects left", n)
	}
	if held := h.leases.holding(); len(held) != 0 {
		t.Fatalf("leases held after the acknowledgement: %v", held)
	}
}

// Phase catch-up (§7.5, round-2 finding 11): a crash after the record's CAS
// and before the claim's leaves the claim one step behind; the next pass
// walks the claim forward and continues. A claim several steps behind is
// walked step by step, with each transition written separately.
func TestLandingPhaseCatchUp(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "after-record-parking"
	h.crashLand(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandParking {
		t.Fatalf("record: %+v", m)
	}
	if c := h.claim(t); c.Op.Phase != provision.LandChecking {
		t.Fatalf("the claim moved with the record: %+v", c.Op)
	}
	h.freshTarget()
	h.landAt = "after-residue-queued" // never reached: no residue
	h.landDone = false
	h.land(t)
	h.installed(t)

	// A claim left at checking with the record at restored.
	rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandChecking })
	l := &landingRun{w: h.lw, s: h.s, rec: h.read(t, p1Host), g: provision.MoveGuard{ID: p1ID, Incarnation: h.inc}}
	obs, err := l.syncClaimPhase(ctx)
	if err != nil {
		t.Fatal(err)
	}
	if obs.Claim.Op.Phase != provision.LandRestored || h.claim(t).Op.Phase != provision.LandRestored {
		t.Fatalf("after the catch-up: %+v", obs.Claim.Op)
	}
	// A claim ahead of the record is left alone.
	rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandProbing })
	if obs, err := l.syncClaimPhase(ctx); err != nil || obs.Claim.Op.Phase != provision.LandProbing {
		t.Fatalf("a claim ahead: %v %+v", err, obs.Claim.Op)
	}
	// A claim at another kind is an error.
	rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Kind = hostclaim.KindRestoring; op.Phase = "released" })
	if _, err := l.syncClaimPhase(ctx); err == nil {
		t.Fatal("a restoring claim was accepted")
	}
}

// earlierStay plants an earlier stay of this incarnation on the target box:
// the installation record at stamp 1, the inventory listing the four flat
// paths at that stamp, and the files at the live paths.
func (h *landingHarness) earlierStay(t *testing.T) {
	t.Helper()
	rec := fmt.Sprintf(`{"v":1,"host":%q,"incarnation":%q,"active_stamp":1,"retired_stamps":[]}`, p1Host, h.inc)
	if err := os.WriteFile(filepath.Join(h.tdir, "hostguard", p1Host+".json"), []byte(rec), 0o660); err != nil {
		t.Fatal(err)
	}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		c.Inventory = map[string]hostclaim.InventoryEntry{h.dstBox: {Layout: "flat", Paths: map[string]uint64{p1Host + ".db": 1, p1Host + ".db.gen": 1, p1Host: 1, p1Host + ".env": 1}}}
	})
	if err := os.MkdirAll(filepath.Join(h.tdir, p1Host), 0o755); err != nil {
		t.Fatal(err)
	}
	for name, body := range map[string]string{p1Host + ".db": "old database", p1Host + ".db.gen": "old\n", p1Host + ".env": "OLD=1\n", filepath.Join(p1Host, "old.txt"): "old files"} {
		if err := os.WriteFile(filepath.Join(h.tdir, name), []byte(body), 0o644); err != nil {
			t.Fatal(err)
		}
	}
}

// local's order (round-2 finding 8, rulings G5 and G15): an earlier stay's
// artifacts tagged S' are parked into .parked/ and gone from the live paths
// before the new installation is registered under F; registering first is
// what the real RecordInventory refuses. A park-local is queued at checking
// only because the inventory has paths at S'.
func TestLandingInventoryOrder(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newLandingHarness(t)
	h.flip(t)
	h.earlierStay(t)
	// Registering first is refused: with the record activated at F and the
	// old paths still tagged 1, RecordInventory says ErrLocalInventory.
	h.landAt = "after-activate"
	h.crashLand(t)
	c := h.claim(t)
	if !hostclaim.InventoryHasStamp(c, "uk-box", 1) {
		t.Fatal("fixture: no S' paths")
	}
	if j := h.landJournal(t); j.Steps[landStepPrevious] != "1" {
		t.Fatalf("previous: %v", j.Steps)
	}
	captured, err := hostguard.Capture(h.tdir, p1Host)
	if err != nil || captured.Stamp != c.Holder.Stamp {
		t.Fatalf("captured %+v %v", captured, err)
	}
	local := park.Local{Guard: hostguard.Guard{DataDir: h.tdir, LockDir: t.TempDir(), Claims: hostguard.StoreReader{Store: h.s}}, Claims: h.s, DataDir: h.tdir, Box: "uk-box"}
	check := func(c *hostclaim.Claim) error { return nil }
	if err := local.RecordInventory(ctx, p1Host, captured, "flat", []string{p1Host + ".db"}, check); !errors.Is(err, park.ErrLocalInventory) {
		t.Fatalf("registering over S' paths: %v", err)
	}
	// The landing: the park-local chore is queued, local retires then registers.
	h.land(t)
	h.installed(t)
	c = h.claim(t)
	if len(c.Obligations) != 0 {
		t.Fatalf("the park-local was not completed: %+v", c.Obligations)
	}
	inv := c.Inventory["uk-box"]
	for p, s := range inv.Paths {
		if s != c.Holder.Stamp {
			t.Fatalf("inventory %s at %d, want F %d", p, s, c.Holder.Stamp)
		}
	}
	// The old artifacts are under .parked/<inc>/<host>/<ret>/artifacts/, and
	// the live paths hold the new installation.
	var parked []string
	filepath.WalkDir(filepath.Join(h.tdir, ".parked", h.inc, p1Host), func(p string, d os.DirEntry, err error) error {
		if err == nil && !d.IsDir() && !strings.HasSuffix(p, ".journal.json") {
			rel, _ := filepath.Rel(filepath.Join(h.tdir, ".parked", h.inc, p1Host), p)
			parked = append(parked, strings.Join(strings.Split(rel, string(filepath.Separator))[1:], "/"))
		}
		return nil
	})
	sort.Strings(parked)
	wantParked := []string{"artifacts/" + p1Host + ".db", "artifacts/" + p1Host + ".db.gen", "artifacts/" + p1Host + ".env", "artifacts/" + p1Host + "/old.txt"}
	if strings.Join(parked, " ") != strings.Join(wantParked, " ") {
		t.Fatalf("parked: %v, want %v", parked, wantParked)
	}
	if _, err := os.Stat(filepath.Join(h.tdir, p1Host, "old.txt")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the old files are still live: %v", err)
	}
	if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host+".env")); err != nil || string(body) != "APP_MODE=prod\n" {
		t.Fatalf("env: %q %v", body, err)
	}
}

// No earlier stay queues no park-local and runs no retirement (ruling G5),
// and the first landing on a box completes end to end with the real
// park.Local and hostguard.Guard.
func TestLandingFirstStayQueuesNoLocalPark(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "after-record-parking"
	h.crashLand(t)
	if c := h.claim(t); len(c.Obligations) != 0 {
		t.Fatalf("a first landing queued %+v", c.Obligations)
	}
	if j := h.landJournal(t); j.Done(landStepPrevious) {
		t.Fatalf("a first landing noted a previous stay: %v", j.Steps)
	}
	h.land(t)
	h.installed(t)
}

// The markers are written last (witness wMarkersLast, codex R2-6): a crash
// before local's final sub-step leaves no .files-gen with everything else
// durable; the replay finishes local and the landing.
func TestLandingMarkersAreWrittenLast(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "before-files-gen"
	h.crashLand(t)
	if _, ok := h.filesGen(t); ok {
		t.Fatal(".files-gen exists before the last sub-step")
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandLocal {
		t.Fatalf("record: %+v", m)
	}
	if body, err := os.ReadFile(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil || string(body) != "hello" {
		t.Fatalf("the files were not extracted before the marker: %q %v", body, err)
	}
	if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host) + ".gen"); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("<dbPath>.gen before restored: %v", err)
	}
	h.freshTarget()
	h.land(t)
	h.installed(t)
	if len(h.replaced) != 1 {
		t.Fatalf("the replay replaced the timers again: %d", len(h.replaced))
	}
}

// configured (§7.6, Review Focus 4): the held render's bundle and overlay
// etags must be the pinned inputs; a difference marks the rollback with
// "configuration changed during the move" and the landing stops there. The
// release gate waits.
func TestLandingConfiguredPinsInputs(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.render = func() (appconfig.HeldRender, error) {
		return appconfig.HeldRender{BundleETags: map[string]string{"default": "e2", "environment": ""}}, nil
	}
	h.land(t)
	m := h.read(t, p1Host).Move
	if m.Reason != provision.RollbackConfigChanged || m.LandingPhase != provision.LandConfigured || m.State != provision.MoveLanding {
		t.Fatalf("after the render: %+v", m)
	}
	if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the database was installed under a mismatch: %v", err)
	}
	// The next pass rolls the marked landing back (Plan 4's decision, J10),
	// installing nothing.
	h.land(t)
	rolledBack(t, h, provision.RollbackConfigChanged, provision.LandConfigured)
	if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the rollback installed the database: %v", err)
	}

	h2 := newLandingHarness(t)
	h2.flip(t)
	h2.release = moves.Check{Reason: t7Waiting}
	h2.land(t)
	if m := h2.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
		t.Fatalf("waiting on the release: %+v", m)
	}
	h2.release = moves.Check{OK: true}
	h2.land(t)
	h2.installed(t)
}

// landing_deadline passed rolls the landing back (Plan 4's decision, J11),
// at whatever phase the landing is, restored included.
func TestLandingDeadlineRollsBack(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.clock.Advance(cutLandingWindow + time.Minute)
	h.land(t)
	rolledBack(t, h, provision.RollbackLandingDeadline, provision.LandChecking)

	h2 := newLandingHarness(t)
	h2.flip(t)
	h2.release = moves.Check{Reason: t7NotProven}
	h2.land(t)
	if m := h2.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
		t.Fatalf("waiting on the proof: %+v", m)
	}
	h2.clock.Advance(cutLandingWindow + time.Minute)
	h2.land(t)
	rolledBack(t, h2, provision.RollbackLandingDeadline, provision.LandConfigured)

	h3 := newLandingHarness(t)
	h3.flip(t)
	h3.land(t)
	h3.installed(t)
	h3.clock.Advance(cutLandingWindow + time.Minute)
	h3.land(t)
	rolledBack(t, h3, provision.RollbackLandingDeadline, provision.LandRestored)
}

// Every phase is replay-safe: a crash after each durable side effect, then
// a fresh worker to restored, yields one object per generation key, one
// timers replacement, one install manifest, and the same end state.
func TestLandingReplaysArePhaseIdempotent(t *testing.T) {
	slots := make(chan struct{}, 4)
	points := []string{"after-previous", "after-activate", "after-record-parking", "after-residue-queued", "after-record-building", "after-build",
		"after-record-published", "after-record-local", "after-retired", "after-staged", "after-files-rename", "after-files_placed", "after-extracted", "after-inventory", "after-timers",
		"after-columns", "after-verify", "after-files_manifest", "before-files-gen", "after-files_gen", "after-record-configured", "after-release-ready", "after-db-rename"}
	for _, point := range points {
		t.Run(point, func(t *testing.T) {
			t.Parallel()
			slots <- struct{}{}
			defer func() { <-slots }()
			h := newLandingHarness(t)
			h.flip(t)
			h.earlierStay(t)
			h.residue(t)
			h.landAt = point
			h.crashLand(t)
			h.freshTarget()
			h.land(t)
			h.installed(t)
			if len(h.replaced) != 1 {
				t.Fatalf("timers replaced %d times", len(h.replaced))
			}
			prefix := moves.GenerationPrefix(cutRoot, p1ID)
			n := 0
			for k := range h.f.buckets["data-uk"] {
				if strings.HasPrefix(k, prefix) {
					n++
				}
			}
			if n != 4 {
				t.Fatalf("%d generation objects", n)
			}
			c := h.claim(t)
			gens := 0
			for _, o := range c.Obligations {
				switch o.Kind {
				case "park-generation":
					gens++
				default:
					t.Fatalf("obligation left: %+v", o)
				}
			}
			if gens != 2 {
				t.Fatalf("%d residue chores", gens)
			}
			for p, s := range c.Inventory["uk-box"].Paths {
				if s != c.Holder.Stamp {
					t.Fatalf("inventory %s at %d", p, s)
				}
			}
		})
	}
}

// landedPreparedManifest reads the prepared manifest the landing's local
// phase journalled (Task 6, C8): the binding from the op journal, the
// manifest at its key in the target's bucket, decoded.
func landedPreparedManifest(t *testing.T, h *landingHarness) (preparedManifest, hostfiles.Manifest, string) {
	t.Helper()
	raw, ok := h.landJournal(t).Steps[landStepFilesManifest]
	if !ok || raw == landingFilesManifestSkipped {
		t.Fatalf("files_manifest: %q", raw)
	}
	var pm preparedManifest
	if err := json.Unmarshal([]byte(raw), &pm); err != nil {
		t.Fatalf("files_manifest %q: %v", raw, err)
	}
	obj, ok := h.f.buckets[h.dstBucket][pm.Key]
	if !ok {
		t.Fatalf("no manifest at %s in the target's bucket", pm.Key)
	}
	if got := sha([]byte(obj.body)); got != pm.SHA256 {
		t.Fatalf("the manifest at %s is not the journalled bytes: %s", pm.Key, got)
	}
	var m hostfiles.Manifest
	if err := json.Unmarshal([]byte(obj.body), &m); err != nil {
		t.Fatal(err)
	}
	return pm, m, obj.body
}

// The prepared files manifest (spec rev 7.22 §7.6, R19-1; C8): local
// writes it under files/g-<op>/ in the location, bound to {location,
// g-<op>, op, stamp} in the op journal, at sequence 0, while
// holder.generation is still the source's. It carries the placed files,
// <host>.env and the wake state the landing installed.
func TestLocalBindsThePreparedManifestToTheNewGeneration(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.land(t)
	h.installed(t)
	c := h.claim(t)
	if c.Holder.Generation != "cur" {
		t.Fatalf("holder.generation moved before the probe: %q", c.Holder.Generation)
	}
	pm, m, _ := landedPreparedManifest(t, h)
	prefix := movesFilesPrefix(c.Holder.Location, p1ID)
	if pm.Key != hostfiles.ManifestKey(prefix) || pm.Op != p1ID || pm.Stamp != c.Holder.Stamp {
		t.Fatalf("binding: %+v, want key %s op %s stamp %d", pm, hostfiles.ManifestKey(prefix), p1ID, c.Holder.Stamp)
	}
	vs, err := h.vs.ListVersions(context.Background(), "data-uk", pm.Key)
	if err != nil {
		t.Fatal(err)
	}
	exact := ""
	for _, v := range vs {
		if v.Key == pm.Key && v.IsLatest {
			exact = v.VersionID
		}
	}
	if pm.Version == "" || pm.Version != exact {
		t.Fatalf("the binding's version %q is not the manifest's exact version %q", pm.Version, exact)
	}
	if m.Seq != 0 {
		t.Fatalf("prepared manifest at seq %d", m.Seq)
	}
	if err := hostfiles.ValidateManifest(m, []string{prefix}); err != nil {
		t.Fatal(err)
	}
	dir := filepath.Join(h.tdir, p1Host)
	placed := 0
	if err := filepath.WalkDir(dir, func(path string, d fs.DirEntry, err error) error {
		if err != nil || d.IsDir() {
			return err
		}
		rel, _ := filepath.Rel(dir, path)
		if rel == archive.MarkerName {
			return nil // .files-gen is written after the manifest, so it is not in it
		}
		body, err := os.ReadFile(path)
		if err != nil {
			return err
		}
		placed++
		if e := m.Entries[filepath.ToSlash(rel)]; e.SHA256 != sha(body) || e.Seq != 0 {
			t.Errorf("%s: %+v", rel, e)
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if placed == 0 || len(m.Entries) != placed {
		t.Fatalf("%d entries for %d placed files: %+v", len(m.Entries), placed, m.Entries)
	}
	env, err := os.ReadFile(filepath.Join(h.tdir, p1Host+".env"))
	if err != nil {
		t.Fatal(err)
	}
	if e := m.Platform[hostfiles.EnvItem]; e.SHA256 != sha(env) {
		t.Fatalf("env item %+v is not <host>.env", e)
	}
	wobj, ok := h.f.buckets["data-uk"][m.Platform[hostfiles.WakeItem].Object]
	if !ok {
		t.Fatal("no wake state object")
	}
	var ws hostfiles.WakeState
	if err := json.Unmarshal([]byte(wobj.body), &ws); err != nil {
		t.Fatal(err)
	}
	if ws.V != 1 || ws.Host != p1Host || len(h.replaced) != 1 || len(ws.Timers) != len(h.replaced[0]) || ws.NextDue != h.nextDue[0] || ws.Replication != h.repl[0] {
		t.Fatalf("wake state %+v, timers replaced %v, next_due %v, replication %v", ws, h.replaced, h.nextDue, h.repl)
	}
}

// A journal that records the step as skipped (written before C8 existed)
// counts as not done (ruling Q3 item 2): a replay builds the manifest and
// the landing goes on with the binding in its journal.
func TestSkippedManifestStepIsBuiltOnReplay(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "after-verify"
	h.crashLand(t)
	j := h.landJournal(t)
	j.Mark(landStepFilesManifest, landingFilesManifestSkipped)
	if err := j.Commit(h.tdir); err != nil {
		t.Fatal(err)
	}
	if got := h.landJournal(t).Steps[landStepFilesManifest]; got != landingFilesManifestSkipped {
		t.Fatalf("the skip was not planted: %q", got)
	}
	h.freshTarget()
	h.land(t)
	h.installed(t)
	pm, m, _ := landedPreparedManifest(t, h)
	if pm.Op != p1ID || m.Seq != 0 {
		t.Fatalf("replay did not build the manifest: %+v seq %d", pm, m.Seq)
	}
}

// A journalled files_manifest value that is not this install's binding
// (fix round 1, finding 3): empty, malformed, or a binding for another op,
// stamp or key. The install neither skips the build over it nor rebuilds
// over it: the replay refuses, the record stays at local, nothing is
// written under the files prefix and the journal keeps the value for an
// operator to see.
func TestLocalRefusesADamagedFilesManifestJournal(t *testing.T) {
	t.Parallel()
	good := func(h *landingHarness) preparedManifest {
		c := h.claim(t)
		return preparedManifest{Key: hostfiles.ManifestKey(movesFilesPrefix(c.Holder.Location, p1ID)), SHA256: sha([]byte("x")), Version: "v1", Op: p1ID, Stamp: c.Holder.Stamp}
	}
	for name, plant := range map[string]func(h *landingHarness) string{
		"empty":         func(*landingHarness) string { return "" },
		"malformed":     func(*landingHarness) string { return `{"key":` },
		"another op":    func(h *landingHarness) string { pm := good(h); pm.Op = "mv_other"; return string(mustJSON(t, pm)) },
		"another stamp": func(h *landingHarness) string { pm := good(h); pm.Stamp++; return string(mustJSON(t, pm)) },
		"another key": func(h *landingHarness) string {
			pm := good(h)
			pm.Key = hostfiles.ManifestKey(movesFilesPrefix(h.claim(t).Holder.Location, "mv_other"))
			return string(mustJSON(t, pm))
		},
		"no version": func(h *landingHarness) string { pm := good(h); pm.Version = ""; return string(mustJSON(t, pm)) },
	} {
		t.Run(name, func(t *testing.T) {
			h := newLandingHarness(t)
			h.flip(t)
			h.landAt = "after-verify"
			h.crashLand(t)
			value := plant(h)
			j := h.landJournal(t)
			j.Mark(landStepFilesManifest, value)
			if err := j.Commit(h.tdir); err != nil {
				t.Fatal(err)
			}
			h.freshTarget()
			h.land(t)
			if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandLocal || m.Reason != "" {
				t.Fatalf("the record moved over a damaged journal: %+v", m)
			}
			j = h.landJournal(t)
			if got := j.Steps[landStepFilesManifest]; got != value {
				t.Fatalf("the journal value changed: %q", got)
			}
			if _, done := j.Steps[landStepFilesGen]; done {
				t.Fatal("the install went on past the damaged step")
			}
			for k := range h.f.buckets["data-uk"] {
				if strings.HasPrefix(k, cutRoot+"/files/") {
					t.Fatalf("written over a damaged journal: %s", k)
				}
			}
		})
	}
}

// The placed tree or the placed env file vanishes after placement and
// before the manifest is built (fix round 1, finding 4): the install
// refuses rather than publishing an empty or partial image. The record
// stays at local, nothing is written under the files prefix and the step
// is not journalled.
func TestLocalRefusesAVanishedTreeBeforeTheManifest(t *testing.T) {
	t.Parallel()
	for name, vanish := range map[string]func(h *landingHarness) error{
		"the files directory": func(h *landingHarness) error { return os.RemoveAll(filepath.Join(h.tdir, p1Host)) },
		"the env file":        func(h *landingHarness) error { return os.Remove(filepath.Join(h.tdir, p1Host+".env")) },
	} {
		t.Run(name, func(t *testing.T) {
			h := newLandingHarness(t)
			h.flip(t)
			h.landAt = "after-verify"
			h.crashLand(t)
			if err := vanish(h); err != nil {
				t.Fatal(err)
			}
			h.freshTarget()
			h.land(t)
			if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandLocal || m.Reason != "" {
				t.Fatalf("the record moved: %+v", m)
			}
			if _, ok := h.landJournal(t).Steps[landStepFilesManifest]; ok {
				t.Fatal("the step was journalled over a vanished input")
			}
			for k := range h.f.buckets["data-uk"] {
				if strings.HasPrefix(k, cutRoot+"/files/") {
					t.Fatalf("written over a vanished input: %s", k)
				}
			}
		})
	}
}

// A cross-region landing's prepared manifest names destination-local
// objects it materialized from the sealed archive (claim P6-D3): every
// object lies under the manifest's own prefix in the target's bucket, and
// nothing is written under the source's files prefix.
func TestCrossRegionLandingManifestNamesDestinationObjects(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.land(t)
	h.installed(t)
	c := h.claim(t)
	pm, m, _ := landedPreparedManifest(t, h)
	prefix := movesFilesPrefix(c.Holder.Location, p1ID)
	if !strings.HasPrefix(pm.Key, prefix) {
		t.Fatalf("manifest key %s is not under %s", pm.Key, prefix)
	}
	for _, entries := range []map[string]hostfiles.Entry{m.Entries, m.Platform} {
		for name, e := range entries {
			if !strings.HasPrefix(e.Object, prefix+"objects/") {
				t.Fatalf("%s names %s, outside the destination prefix %s", name, e.Object, prefix)
			}
			obj, ok := h.f.buckets["data-uk"][e.Object]
			if !ok || sha([]byte(obj.body)) != e.SHA256 {
				t.Fatalf("%s names %s, which the target's bucket does not hold at that digest", name, e.Object)
			}
		}
	}
	for k := range h.f.buckets["data-ie"] {
		if strings.Contains(k, "/files/") {
			t.Fatalf("the landing wrote into the source region's files: %s", k)
		}
	}
}

// A files manifest already at P that is not this install's (another
// writer's, or a damaged one) is not the landing's to replace: local
// refuses and the move rolls back (RollbackManifestOccupied), as it does
// for an occupied files directory.
func TestLocalRefusesAForeignPreparedManifest(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	c := h.claim(t)
	key := hostfiles.ManifestKey(movesFilesPrefix(c.Holder.Location, p1ID))
	body := `{"v":2,"seq":0,"entries":{"x.txt":{"object":"` + movesFilesPrefix(c.Holder.Location, p1ID) + `objects/0000000000000000-` + sha([]byte("x")) + `","sha256":"` + sha([]byte("x")) + `","size":1,"seq":0}},"platform":{}}`
	h.f.store("data-uk", key, fakeMoveObject{lease.MirrorObject{ETag: "foreign", Size: int64(len(body))}, body})
	h.f.setVersion("data-uk", key, h.f.nextVersion())
	h.land(t)
	if m := h.read(t, p1Host).Move; m.Reason != provision.RollbackManifestOccupied || m.LandingPhase != provision.LandLocal {
		t.Fatalf("not refused for %q at local: %+v", provision.RollbackManifestOccupied, m)
	}
	if got := h.f.buckets["data-uk"][key].body; got != body {
		t.Fatal("the foreign manifest was replaced")
	}
	if _, ok := h.landJournal(t).Steps[landStepFilesManifest]; ok {
		t.Fatal("the refused step was journalled as done")
	}
}

// moveAuthority hands the landing's op to exec in the claim, as a
// replacement or a return of the executor would.
func moveAuthority(t *testing.T, h *landingHarness, exec string) {
	t.Helper()
	obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(context.Background(), h.s, obs, func(c *hostclaim.Claim) error {
		c.Op.Exec = exec
		if !slices.Contains(c.Op.Participants, exec) {
			c.Op.Participants = append(c.Op.Participants, exec)
		}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
}

// The prepared build re-checks the installation's authority (the op at
// phase local with this box executing, the holder at the captured stamp)
// immediately before every bucket write, retries included (fix round 1,
// finding 1; region HA's authorityStore). Authority moves to another
// executor between the first object upload and the next write: the stale
// writer writes nothing more, no manifest exists, and the step is not
// journalled. Once authority returns, the installer completes, reusing
// the one object the stale writer left.
func TestPreparedBuildRechecksAuthorityBeforeEveryWrite(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	c := h.claim(t)
	prefix := movesFilesPrefix(c.Holder.Location, p1ID)
	moved := false
	h.vs.deny = func(op, bucket, key string) error {
		if op == "put-if-absent" && strings.HasPrefix(key, prefix+"objects/") && !moved {
			moved = true
			moveAuthority(t, h, "other-box")
		}
		return nil
	}
	h.land(t)
	if !moved {
		t.Fatal("the build never uploaded an object")
	}
	objects := 0
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix+"objects/") {
			objects++
		}
		if k == hostfiles.ManifestKey(prefix) {
			t.Fatal("the stale writer published the manifest")
		}
	}
	if objects != 1 {
		t.Fatalf("%d objects after authority moved, want the one uploaded before it", objects)
	}
	if _, ok := h.landJournal(t).Steps[landStepFilesManifest]; ok {
		t.Fatal("the step was journalled without the manifest")
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandLocal || m.Reason != "" {
		t.Fatalf("the record moved: %+v", m)
	}
	moveAuthority(t, h, "uk-box")
	h.vs.deny = nil
	h.freshTarget()
	h.land(t)
	h.installed(t)
	pm, m, _ := landedPreparedManifest(t, h)
	if pm.Op != p1ID || m.Seq != 0 {
		t.Fatalf("the authorized installer did not complete: %+v", pm)
	}
	vs, err := h.vs.ListVersions(context.Background(), "data-uk", prefix+"objects/")
	if err != nil {
		t.Fatal(err)
	}
	seen := map[string]int{}
	for _, v := range vs {
		seen[v.Key]++
	}
	for k, n := range seen {
		if n != 1 {
			t.Fatalf("%s has %d versions: the replay did not reuse it", k, n)
		}
	}
}

// Nothing is written under the location's files prefix before the
// files_manifest step: the manifest follows the placed files, the env
// file, the timers and the columns (the step order of local, §7.6).
func TestLocalWritesNoFilesManifestBeforeItsStep(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "after-verify"
	h.crashLand(t)
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, cutRoot+"/files/") {
			t.Fatalf("a files manifest was written before the files_manifest step: %s", k)
		}
	}
}

// Ruling G17: with the park engine's hooks unwired, as armPhase1 leaves
// them, the landing is not armed: the pass logs and writes nothing.
func TestLandingWaitsUnarmedInProduction(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.lw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
	c := h.landingCfg(t)
	c.ParkLeases = nil
	h.lw.cfg = c
	if landingArmed(c) {
		t.Fatal("armed without park leases")
	}
	if err := movesPass(context.Background(), h.s, h.lw); err != nil {
		t.Fatal(err)
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandChecking {
		t.Fatalf("an unarmed landing advanced: %+v", m)
	}
	if _, err := hostguard.Capture(h.tdir, p1Host); !errors.Is(err, hostguard.ErrNoRecord) {
		t.Fatalf("an unarmed landing wrote the installation record: %v", err)
	}
	// Production's own wiring: everything but the park engine's hooks.
	var prod moveCfg
	hib := activator.Config{Bucket: "data-uk", Region: "eu-west-2", Prefix: "instances"}
	armPhase1(&prod, &movesWorker{mirrors: map[string]lease.MirrorStore{}}, hib, h.tdir, nil, nil, nil, hostwrite.Config{}, func(string) string { return "" })
	if prod.Guard.DataDir != h.tdir || prod.Local.DataDir != h.tdir || prod.OwnLease == nil || prod.RenderHeld == nil || prod.RemoveVerifyEntry == nil || prod.WriteGen == nil {
		t.Fatalf("production wiring: %+v", prod)
	}
	if prod.ParkLeases != nil || prod.ParkView != nil || prod.ParkAuthorize != nil || landingArmed(prod) {
		t.Fatal("production armed the landing")
	}
}

// The generation classifier the stand-in view and parking share.
func TestLandingGenerationOf(t *testing.T) {
	loc := "instances/h"
	cases := map[string]struct {
		gen string
		ok  bool
	}{
		loc + "/g-old/0000/x.ltx":                           {"old", true},
		loc + "/g-mv_1/install.json":                        {"mv_1", true},
		loc + "/files/g-mv_1/manifest":                      {"mv_1", true},
		loc + "/0000/x.ltx":                                 {"", true},
		loc + "/0009/x.ltx":                                 {"", true},
		loc + "/generation.json":                            {landingMarkerGeneration, true},
		loc + "/g-/x":                                       {"", false},
		loc + "/stray.json":                                 {"", false},
		loc + "/zzzz/x.ltx":                                 {"", false},
		"instances/other/0000/x.ltx":                        {"", false},
		loc + "/" + strings.Repeat("a", 64) + "/0000/x.ltx": {"", false},
	}
	for key, want := range cases {
		gen, ok := landingGenerationOf(loc, key)
		if gen != want.gen || ok != want.ok {
			t.Errorf("%s: %q %v, want %q %v", key, gen, ok, want.gen, want.ok)
		}
	}
}

// The target report pass keeps a landing's refusal and verdict live (G26).
func TestTargetLiveSetCoversTheCutAndTheLanding(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	w := &movesWorker{cfg: h.landingCfg(t), mirrors: map[string]lease.MirrorStore{}, archives: map[string]verifiedArchive{p1ID: {}, "mv_other": {}}}
	recs, err := provision.ListAllRecords(context.Background(), h.s)
	if err != nil {
		t.Fatal(err)
	}
	fleet, err := loadMoveFleet(context.Background(), h.s)
	if err != nil {
		t.Fatal(err)
	}
	w.targetReportPass(context.Background(), h.s, recs, fleet)
	if _, ok := w.archives[p1ID]; !ok {
		t.Fatal("the landing's verdict was evicted")
	}
	if _, ok := w.archives["mv_other"]; ok {
		t.Fatal("a finished move's verdict was kept")
	}
}

// Fix round 1, item 1: the pass catches the claim up at restored before it
// returns. A crash after the record's restored CAS, or a following claim
// CAS that fails, leaves the claim behind; the next pass (a fresh worker)
// converges through movesPass alone.
func TestLandingRestoredCatchesTheClaimUp(t *testing.T) {
	t.Parallel()
	t.Run("crash after the record's restored", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		h.landAt = "after-record-restored"
		h.crashLand(t)
		if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
			t.Fatalf("fixture: claim %+v", c.Op)
		}
		h.freshTarget()
		h.land(t)
		h.installed(t)
	})
	t.Run("the following claim CAS fails", func(t *testing.T) {
		h := newLandingHarness(t)
		h.flip(t)
		// At the boundary the claim is moved two steps back, so the mirror's
		// SetMovePhase(configured, restored) is refused and the claim stays
		// behind; the pass must not fail, and the next pass walks it up.
		h.landFault = func(point string) error {
			if point == "after-record-restored" {
				rewriteClaimOp(t, h.s, func(op *hostclaim.Op) { op.Phase = provision.LandBuilding })
			}
			return nil
		}
		h.land(t)
		h.landFault = nil
		if !containsString(h.landHits, "after-record-restored") {
			t.Fatalf("the boundary was never reached: %v", h.landHits)
		}
		// The refused mirror is logged, and the same pass's restored step
		// walks the claim from building to restored before it returns.
		h.installed(t)
		h.freshTarget()
		h.land(t)
		h.installed(t)
	})
}

// Fix round 1, item 2: a build that does not verify marks the rollback and
// leaves the phase at configured: no transition to restored, no digest, no
// database installed, no marker.
func TestLandingCorruptBuildMarksRollback(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	h.landFault = func(point string) error {
		if point == "before-db-verify" {
			f, err := os.OpenFile(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build", p1Host+".db"), os.O_APPEND|os.O_WRONLY, 0o600)
			if err != nil {
				t.Fatal(err)
			}
			if _, err := f.Write([]byte("corrupt")); err != nil {
				t.Fatal(err)
			}
			f.Close()
		}
		return nil
	}
	h.land(t)
	if !containsString(h.landHits, "before-db-verify") {
		t.Fatalf("the verify point was never reached: %v", h.landHits)
	}
	m := h.read(t, p1Host).Move
	if m.Reason != provision.RollbackInstallMismatch || m.LandingPhase != provision.LandConfigured || m.State != provision.MoveLanding {
		t.Fatalf("after the corrupt build: %+v", m)
	}
	if m.Landing != nil && m.Landing.InstalledDBSHA256 != "" {
		t.Fatalf("a digest was recorded: %+v", m.Landing)
	}
	dbPath := provision.DefaultDBPath(h.tdir, p1Host)
	for _, p := range []string{dbPath, dbPath + ".gen"} {
		if _, err := os.Stat(p); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s exists after a refused install: %v", p, err)
		}
	}
	if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
		t.Fatalf("claim: %+v", c.Op)
	}
}

// Fix round 1, item 3: the generation is written only under live lease
// authority. A rebuild (the scratch lost after published released the
// lease) takes the location's lease first, and waits while another holder
// has it, writing nothing; a lease lost during the build stops the build
// at the next generation write, and the replay completes.
func TestLandingRebuildNeedsTheLease(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newLandingHarness(t)
	h.flip(t)
	h.landAt = "after-record-configured"
	h.crashLand(t)
	if err := os.RemoveAll(filepath.Join(h.tdir, ".moves", p1Host, p1ID, "build")); err != nil {
		t.Fatal(err)
	}
	h.freshTarget()
	other, err := h.leases.Acquire(ctx, "data-uk", cutRoot)
	if err != nil {
		t.Fatal(err)
	}
	prefix := moves.GenerationPrefix(cutRoot, p1ID)
	before := len(h.vs.callsOn(prefix))
	h.land(t)
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandConfigured || m.Reason != "" {
		t.Fatalf("installed without the lease: %+v", m)
	}
	if got := h.vs.callsOn(prefix); len(got) != before {
		t.Fatalf("generation requests while another holder had the lease: %v", got[before:])
	}
	if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a database was installed without the lease: %v", err)
	}
	if err := other.Release(ctx); err != nil {
		t.Fatal(err)
	}
	h.land(t)
	h.installed(t)
}

func TestLandingBuildStopsWhenTheLeaseIsLost(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	prefix := moves.GenerationPrefix(cutRoot, p1ID)
	lost := false
	h.vs.deny = func(op, bucket, key string) error {
		if op == "put-if-absent" && strings.HasPrefix(key, prefix) && !lost {
			lost = true
			h.leases.expire() // the holder went dark between two generation writes
		}
		return nil
	}
	h.land(t)
	h.vs.deny = nil
	if !lost {
		t.Fatal("the build never wrote")
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
		t.Fatalf("the build completed on a lost lease: %+v", m)
	}
	if _, ok := h.f.buckets["data-uk"][moves.InstallManifestKey(prefix)]; ok {
		t.Fatal("install.json was written on a lost lease")
	}
	var written []string
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			written = append(written, strings.TrimPrefix(k, prefix))
		}
	}
	if len(written) > 1 {
		t.Fatalf("writes after the lease was lost: %v", written)
	}
	h.freshTarget()
	h.land(t)
	h.installed(t)
}

// Fix round 1, item 4: landing_deadline is checked again right before the
// install transition. Past it after the release gate, the rollback is marked
// and nothing is installed; past it after the install itself, the rollback
// is marked and restored is never written.
func TestLandingDeadlineBeforeTheInstallTransition(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"after-release-ready", "after-db-rename"} {
		t.Run(point, func(t *testing.T) {
			h := newLandingHarness(t)
			h.flip(t)
			h.landFault = func(p string) error {
				if p == point {
					h.clock.Advance(cutLandingWindow + time.Minute)
				}
				return nil
			}
			h.land(t)
			m := h.read(t, p1Host).Move
			if m.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandConfigured {
				t.Fatalf("past the deadline at %s: %+v", point, m)
			}
			if m.Landing != nil && m.Landing.InstalledDBSHA256 != "" {
				t.Fatalf("a digest was recorded: %+v", m.Landing)
			}
			_, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host))
			if point == "after-release-ready" && !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("installed past the deadline: %v", err)
			}
			if c := h.claim(t); c.Op.Phase != provision.LandConfigured {
				t.Fatalf("claim: %+v", c.Op)
			}
		})
	}
}

// retryingStore is a versioned store whose first generation write blocks
// like a client retrying a request: the lease lapses while it is in flight,
// and if nothing cancels the request its retry lands after the loss.
type retryingStore struct {
	lease.VersionedStore
	h       *landingHarness
	prefix  string
	once    sync.Once
	landed  bool
	blocked bool
}

func (r *retryingStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
	block := false
	if strings.HasPrefix(key, r.prefix) {
		r.once.Do(func() { block = true })
	}
	if !block {
		return r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
	}
	r.blocked = true
	r.h.leases.expire() // the holder goes dark while the request is in flight
	select {
	case <-ctx.Done():
		return "", ctx.Err()
	case <-time.After(1500 * time.Millisecond):
		r.landed = true // the client's retry succeeds after the lease was lost
		return r.VersionedStore.PutIfAbsent(ctx, bucket, key, body)
	}
}

// Fix round 2: every generation request runs under a context derived from
// the lease's, so a lease lost while a request is in flight cancels it and
// its retries; nothing is recorded. The replay builds afresh.
func TestLandingBuildIsCancelledWhenTheLeaseIsLostMidRequest(t *testing.T) {
	t.Parallel()
	h := newLandingHarness(t)
	h.flip(t)
	prefix := moves.GenerationPrefix(cutRoot, p1ID)
	rs := &retryingStore{VersionedStore: h.vs, h: h, prefix: prefix}
	h.store = rs
	h.land(t)
	h.store = nil
	if !rs.blocked {
		t.Fatal("the build never wrote under the generation")
	}
	if rs.landed {
		t.Fatal("the request landed after the lease was lost: it was not cancelled")
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
		t.Fatalf("record: %+v", m)
	}
	for k := range h.f.buckets["data-uk"] {
		if strings.HasPrefix(k, prefix) {
			t.Fatalf("written on a lost lease: %s", k)
		}
	}
	if j := h.landJournal(t); j.Done(landStepInstall) {
		t.Fatal("the build was recorded on a lost lease")
	}
	h.freshTarget()
	h.land(t)
	h.installed(t)
}

// openState is the record's state, the claim's phase and the row's status
// at a step of the open.
func (h *landingHarness) openState(t *testing.T) string {
	return h.read(t, p1Host).Move.State + "/" + h.claim(t).Op.Phase + "/" + h.route.Status
}

// errHoldProbed is holdProbed's stop before the commit.
var errHoldProbed = errors.New("held at probed by the test")