package host

import (
	"bytes"
	"context"
	"database/sql"
	"encoding/json"
	"errors"
	"io"
	"os"
	"path/filepath"
	"strconv"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/park"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/superfly/ltx"
)

// The rollback on the target (moves spec §7.8; Plan 4 Task 4): the decision
// that ends a landing, then the target's release, its return lane and its
// cleaning. The fixture is the landing harness landed through restored and
// stepped to landing/probing on both the record and the claim (the probe is
// Plan 3's and not on this branch), with the target box's rollback hooks
// faked where main has no production provider in a test: the hold hand-over,
// the final checkpoint, the local position and the row delete.

type rollbackHarness struct {
	*landingHarness
	rw      *movesWorker
	handed  []string // status:to:from of each hand-over asked for
	rowGone bool
	deleted []string
	tchecks int    // the target's final checkpoints
	tpos    uint64 // the target's local position
	events  []string
	pages   pagesSeen // the operator pages (moveCfg.Page), the target's and the source's
	rbAt    string
	rbDone  bool
	rbHits  []string
	// rbHook runs at every fault point, before the crash check; beforeHand
	// runs inside the hand-over, before its validation.
	rbHook     func(point string)
	beforeHand func()
}

func newRollbackHarness(t *testing.T, opts ...p1Option) *rollbackHarness {
	t.Helper()
	h := &rollbackHarness{landingHarness: newLandingHarness(t, opts...)}
	h.flip(t)
	h.land(t)
	h.installed(t)
	h.step(t, provision.LandRestored, provision.LandProbing)
	h.tpos = h.n
	return h
}

// step moves the landing one phase on the record and the claim, as the
// probe (Plan 3) would.
func (h *rollbackHarness) step(t *testing.T, from, to string) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveLandingPhase(g, from, to, nil)); err != nil {
		t.Fatal(err)
	}
	obs, err := hostclaim.Read(context.Background(), h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(context.Background(), h.s, obs, hostclaim.SetMovePhase(p1ID, hostclaim.KindLanding, from, to)); err != nil {
		t.Fatal(err)
	}
}

// rollbackCfg is the target box's configuration with the rollback armed.
func (h *rollbackHarness) rollbackCfg(t *testing.T) moveCfg {
	t.Helper()
	c := h.landingCfg(t)
	c.OwnsRoute = func(host string) bool { return host == p1Host && !h.rowGone }
	c.Route = func(host string) (*registry.Route, error) {
		if host != p1Host || h.rowGone {
			return nil, nil
		}
		rt := h.route
		return &rt, nil
	}
	// The hand-over keeps handHold's contract: under the lock the row is
	// read, then the caller's validation runs, then the hold moves.
	c.HandHold = func(host, status, to, from string, validate func() error) (bool, error) {
		h.handed = append(h.handed, status+":"+to+":"+from)
		if h.rowGone {
			return false, nil
		}
		if h.beforeHand != nil {
			h.beforeHand()
		}
		if validate != nil {
			if err := validate(); err != nil {
				return false, err
			}
		}
		if h.route.HoldOp == to && h.route.Status == status {
			return true, nil
		}
		if h.route.HoldOp != from {
			return false, nil
		}
		h.route.Status, h.route.HoldOp = status, to
		return true, nil
	}
	c.FinalCheckpoint = func(ctx context.Context, host, _ string, holder hostclaim.Holder, check func(context.Context) error) error {
		if err := check(ctx); err != nil {
			return err
		}
		if host != p1Host || strings.TrimSuffix(holder.Location, "/") != cutRoot {
			t.Errorf("checkpoint of %s at %s", host, holder.Location)
		}
		h.tchecks++
		return nil
	}
	c.LocalPosition = func(string) (uint64, error) { return h.tpos, nil }
	c.DeleteHeld = func(host, op string) (bool, error) {
		h.deleted = append(h.deleted, host+"="+op)
		if h.rowGone || h.route.HoldOp != op {
			return false, nil
		}
		h.rowGone = true
		return true, nil
	}
	c.LogEvent = func(_ context.Context, _ release.Store, _ provision.Record, ev moves.LogEvent) error {
		h.events = append(h.events, ev.Event)
		return nil
	}
	c.Page = h.pages.hook()
	c.rbFault = func(point string) error {
		h.rbHits = append(h.rbHits, point)
		if h.rbHook != nil {
			h.rbHook(point)
		}
		if point == h.rbAt && !h.rbDone {
			h.rbDone = true
			return errCrash
		}
		return nil
	}
	return c
}

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

// rollBack passes the landing's deadline and runs the target's pass that
// decides the rollback.
func (h *rollbackHarness) rollBack(t *testing.T) *provision.MoveRollback {
	t.Helper()
	h.clock.Advance(cutLandingWindow + time.Minute)
	h.rollbackPass(t)
	m := h.read(t, p1Host).Move
	if m.State != provision.MoveRollingBack || m.Rollback == nil {
		t.Fatalf("no rollback: %+v", m)
	}
	return m.Rollback
}

func (h *rollbackHarness) rb(t *testing.T) *provision.MoveRollback {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m == nil || m.Rollback == nil {
		t.Fatalf("record: %+v", m)
	}
	return m.Rollback
}

// uncover marks the staged seal as not covering the source (§7.4 rev 7.11):
// the source's own replica did not reach its stopped position.
func (h *rollbackHarness) uncover(t *testing.T) {
	t.Helper()
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		s, err := journalledSeal(c)
		if err != nil {
			t.Fatal(err)
		}
		s.SrcCovered = false
		c.Op.Journal[hostclaim.JournalSeal] = s.Encode()
	})
}

// recovered drives the source's half of the record to recovered, stamping
// RecoveredAt, as Task 5's source would.
func (h *rollbackHarness) recovered(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	path := []string{provision.RbSourceWaiting, provision.RbSourceClaimed, provision.RbSourceRestoring, provision.RbSourceVerified, provision.RbSourceProbing}
	if h.rb(t).Dirty {
		path = []string{provision.RbSourceWaiting, provision.RbSourceClaimed, provision.RbSourceRestoring, provision.RbSourceParking, provision.RbSourceBuilding,
			provision.RbSourcePublished, provision.RbSourceLocal, provision.RbSourceConfigured, provision.RbSourceRestored, provision.RbSourceProbing}
	}
	path = append(path, provision.RbSourceProbed, provision.RbSourceRecovered)
	for i := 0; i+1 < len(path); i++ {
		var stamp func(*provision.MoveRollback)
		if path[i+1] == provision.RbSourceRecovered {
			now := h.clock.Now()
			stamp = func(rb *provision.MoveRollback) { rb.RecoveredAt = &now }
		}
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRollbackSourcePhase(g, path[i], path[i+1], stamp)); err != nil {
			t.Fatalf("%s to %s: %v", path[i], path[i+1], err)
		}
	}
}

// rolledBack asserts the rollback decision took the landing: the record
// rolling_back in the source's region with reason, the landing phase kept,
// the target releasing; the claim restoring/releasing under the rollback.
func rolledBack(t *testing.T, h *landingHarness, reason, phase string) {
	t.Helper()
	rec := h.read(t, p1Host)
	m := rec.Move
	if rec.Region != "ie" || m.State != provision.MoveRollingBack || m.Rollback == nil || m.Rollback.Reason != reason ||
		m.LandingPhase != phase || m.Rollback.TargetPhase != provision.RbTargetReleasing {
		t.Fatalf("not rolled back for %q at %s: %s %+v %+v", reason, phase, rec.Region, m, m.Rollback)
	}
	if c := h.claim(t); c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Intent["rollback"] != m.Rollback.ID {
		t.Fatalf("the claim did not follow: %+v", c.Op)
	}
}

func targetStoppedKey() string {
	return moves.StoppedKey(moves.ReportPrefix("bab", "app", p1Host, p1ID), "target")
}

// laneCalls are the guarded staging requests into the source's bucket.
func (h *rollbackHarness) laneCalls(rb string) []stagingCall {
	var out []stagingCall
	for _, c := range h.f.stagingCalls() {
		if c.Bucket == "data-ie" && strings.HasPrefix(c.Key, "moves/"+p1Host+"/rb_") {
			if !strings.HasPrefix(c.Key, moves.ReturnLane(p1Host, rb)) {
				panic("a lane write outside the rollback's lane: " + c.Key)
			}
			out = append(out, c)
		}
	}
	return out
}

// TestRollbackDecisionConvertsTheMark: a marked landing is rolled back with
// the mark's reason, in the one place the mark is converted, before the
// landing's arming gate (J10); the claim follows (RollBack, or
// RollBackFlipped over a claim the flip's half never reached, J16), with the
// target installation's stamp F journalled (J15).
func TestRollbackDecisionConvertsTheMark(t *testing.T) {
	t.Parallel()
	t.Run("the claim at landing", func(t *testing.T) {
		h := newRollbackHarness(t)
		g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveMarkRollback(g, provision.RollbackProbeAttempts)); err != nil {
			t.Fatal(err)
		}
		f := h.claim(t).Holder.Stamp
		// Unarmed: the decision runs all the same.
		h.rw = &movesWorker{mirrors: map[string]lease.MirrorStore{}}
		c := h.rollbackCfg(t)
		c.ParkLeases = nil
		h.rw.cfg = c
		if err := movesPass(context.Background(), h.s, h.rw); err != nil {
			t.Fatal(err)
		}
		rec := h.read(t, p1Host)
		m := rec.Move
		if m.State != provision.MoveRollingBack || rec.Region != "ie" || m.Rollback.Reason != provision.RollbackProbeAttempts ||
			m.Rollback.TargetPhase != provision.RbTargetReleasing || !provision.ValidRollbackID(m.Rollback.ID) || m.LandingPhase != provision.LandProbing {
			t.Fatalf("record: %s %+v %+v", rec.Region, m, m.Rollback)
		}
		cl := h.claim(t)
		if cl.Op.Kind != hostclaim.KindRestoring || cl.Op.Phase != hostclaim.PhaseReleasing || cl.Op.Intent["rollback"] != m.Rollback.ID ||
			cl.Op.Exec != "uk-box" || cl.Op.Journal[hostclaim.JournalTargetStamp] != strconv.FormatUint(f, 10) {
			t.Fatalf("claim: %+v", cl.Op)
		}
		if len(h.events) != 1 || h.events[0] != "rolling_back" {
			t.Fatalf("events %v", h.events)
		}
	})
	t.Run("the claim still moving", func(t *testing.T) {
		h := &rollbackHarness{landingHarness: 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 claim: %+v", c.Op)
		}
		h.rollBack(t)
		m := h.read(t, p1Host).Move
		if m.Rollback.Reason != provision.RollbackLandingDeadline || m.LandingPhase != provision.LandChecking {
			t.Fatalf("record: %+v", m.Rollback)
		}
		if c := h.claim(t); c.Op.Kind != hostclaim.KindRestoring || c.Op.Phase != hostclaim.PhaseReleasing || c.Op.Exec != "uk-box" ||
			c.Op.Journal[hostclaim.JournalSrcGen] != "cur" || c.Op.Intent["rollback"] != m.Rollback.ID {
			t.Fatalf("claim: %+v", c.Op)
		}
	})
}

// TestRollbackCASLosesToCommit: the rollback and the commit race on the
// record; once the commit has won, the rollback's CAS is refused and leaves
// no trace: the record committed with no rollback, the claim and the row as
// they were. The open is the commit's path (Plan 3 Task 7).
func TestRollbackCASLosesToCommit(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	h.step(t, provision.LandProbing, provision.LandProbed)
	stale := h.read(t, p1Host)
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveCommit(g, h.clock.Now())); err != nil {
		t.Fatal(err)
	}
	before, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	h.clock.Advance(cutLandingWindow + time.Minute)
	w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}}
	done, err := w.rollBackLanding(ctx, h.s, stale)
	if err != nil || !done {
		t.Fatalf("the decision over a stale landing: %v %v", done, err)
	}
	rec := h.read(t, p1Host)
	if rec.Move.State != provision.MoveCommitted || rec.Move.Rollback != nil || rec.Region != "uk" {
		t.Fatalf("record: %s %+v", rec.Region, rec.Move)
	}
	after, err := hostclaim.Read(ctx, h.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if after.ETag != before.ETag || after.Claim.Op.Kind != hostclaim.KindLanding {
		t.Fatalf("the claim was written: %+v", after.Claim.Op)
	}
	if len(h.handed) != 0 || h.route.HoldOp != p1ID || len(h.events) != 0 {
		t.Fatalf("the rollback left a trace: %v %+v %v", h.handed, h.route, h.events)
	}
	// A pass over the committed record leaves it too.
	h.rollbackPass(t)
	if rec := h.read(t, p1Host); rec.Move.State != provision.MoveCommitted || rec.Move.Rollback != nil {
		t.Fatalf("after a pass: %+v", rec.Move)
	}
}

// TestReleaseIsCleanWhenNothingChanged: the landing reached probing, but
// nothing the target holds differs from what it installed (the database,
// the files, the timers) and the staged seal covers the source: the release
// is clean. Execution goes to none citing stopped-target.json, the target
// region's whole location is queued for its park with the target's
// park-local, and the target region's predecessors are covered.
func TestReleaseIsCleanWhenNothingChanged(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) {
		c.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "uk", Session: 3,
			Box: "uk-box", Generation: "probe", Predecessor: "cur", Mode: "cold"})
	})
	f := h.claim(t).Holder.Stamp
	rb := h.rollBack(t)
	fences := h.fences
	if h.route.HoldOp != p1ID || len(h.handed) != 0 {
		t.Fatalf("the decision touched the row: %+v %v", h.route, h.handed)
	}
	h.rollbackPass(t)
	rb = h.rb(t)
	if rb.TargetPhase != provision.RbTargetReleased || rb.Dirty || rb.Evidence != targetStoppedKey() {
		t.Fatalf("rollback: %+v", rb)
	}
	if h.route.Status != "releasing" || h.route.HoldOp != rb.ID || h.fences != fences+1 || h.tchecks != 1 {
		t.Fatalf("row %+v fences %d checkpoints %d", h.route, h.fences, h.tchecks)
	}
	report, err := readStoppedReport(ctx, h.s, rb.Evidence)
	if err != nil {
		t.Fatal(err)
	}
	m := h.read(t, p1Host).Move
	if report.Role != "target" || report.Box != "uk-box" || report.OpID != p1ID || report.FinalTXID != h.n ||
		report.DBSHA256 != m.Landing.InstalledDBSHA256 || report.FilesPlain != m.Source.Files.Plain || report.TimersSHA256 != m.Source.TimersSHA256 {
		t.Fatalf("report %+v", report)
	}
	c := h.claim(t)
	if c.Op.Phase != hostclaim.PhaseReleased || c.Op.Exec != "" || c.Op.Journal[hostclaim.JournalEvidenceTarget] != rb.Evidence {
		t.Fatalf("claim: %+v", c.Op)
	}
	var location, local int
	for _, o := range c.Obligations {
		switch {
		case o.Kind == "park-replica" && o.Region == "uk" && o.Park.Location == cutRoot:
			location++
			if !o.Park.EvidenceAt.Equal(report.CheckedAt) || o.Park.RequestLimit != cutRequestLimit || o.Park.Incarnation != h.inc {
				t.Fatalf("the location park's scope: %+v", o.Park)
			}
		case o.Kind == "park-local" && o.Box == "uk-box":
			local++
		default:
			t.Fatalf("an obligation the release did not owe: %+v", o)
		}
	}
	if location != 1 || local != 1 {
		t.Fatalf("obligations: %+v", c.Obligations)
	}
	if !hostclaim.InventoryHasStamp(c, "uk-box", f) {
		t.Fatal("the fixture's inventory does not list F")
	}
	for _, a := range c.Holder.Admissions {
		if a.PredecessorRegion == "uk" && !a.Covered {
			t.Fatalf("the target region's predecessor is not covered: %+v", a)
		}
	}
	// Nothing goes to the lane, and the target waits for the source.
	if calls := h.laneCalls(rb.ID); len(calls) != 0 {
		t.Fatalf("a clean release wrote the lane: %+v", calls)
	}
	h.rollbackPass(t)
	if got := h.rb(t); got.TargetPhase != provision.RbTargetReleased {
		t.Fatalf("a clean release moved on before the source recovered: %+v", got)
	}
	if strings.Join(h.events, " ") != "rolling_back released" {
		t.Fatalf("events %v", h.events)
	}
}

// A landing rolled back before probing is clean when the seal covers the
// source, whatever its digests: no writer ever ran on it. Nothing is
// checkpointed.
func TestReleaseBeforeProbingNeedsNoDigests(t *testing.T) {
	t.Parallel()
	h := &rollbackHarness{landingHarness: newLandingHarness(t)}
	h.flip(t)
	h.landAt = "after-build"
	h.crashLand(t)
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if rb.TargetPhase != provision.RbTargetReleased || rb.Dirty || h.tchecks != 0 {
		t.Fatalf("rollback %+v checkpoints %d", rb, h.tchecks)
	}
	if m := h.read(t, p1Host).Move; m.LandingPhase != provision.LandBuilding {
		t.Fatalf("landing phase %s", m.LandingPhase)
	}
}

// TestReleaseIsDirtyWithoutSrcCovered (Review Focus 1, witness wSealReturn):
// the digests are equal but the staged seal does not cover the source, so
// the source cannot trust its own copy: the release is dirty, and the
// return lane gets the staged seal's exact versions, as given.
func TestReleaseIsDirtyWithoutSrcCovered(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	h.uncover(t)
	staged := h.seal(t)
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	c := h.claim(t)
	seal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
	if err != nil {
		t.Fatal(err)
	}
	if !seal.SrcCovered || !seal.FromDB || seal.N != staged.N || !seal.Bound().Equal(h.lease.until) {
		t.Fatalf("return seal %+v (lease until %s)", seal, h.lease.until)
	}
	lane, from := moves.ReturnLane(p1Host, rb.ID), moves.StagingPrefix(p1Host, p1ID)
	got := map[string]moves.SealedObject{}
	for _, o := range append(append(append([]moves.SealedObject(nil), seal.Replica...), seal.Files...), seal.Timers...) {
		got[o.Key] = o
	}
	for _, o := range append(append(append([]moves.SealedObject(nil), staged.Replica...), staged.Files...), staged.Timers...) {
		key := lane + strings.TrimPrefix(o.Key, from)
		r, ok := got[key]
		if !ok {
			t.Fatalf("the return seal does not name %s", key)
		}
		want, err := h.vs.GetVersion(ctx, "data-uk", o.Key, o.VersionID)
		if err != nil {
			t.Fatal(err)
		}
		body, err := h.vs.GetVersion(ctx, "data-ie", key, r.VersionID)
		if err != nil {
			t.Fatalf("%s at %s: %v", key, r.VersionID, err)
		}
		if !bytes.Equal(body, want) || r.Size != o.Size {
			t.Fatalf("%s is not the staged version of %s", key, o.Key)
		}
		delete(got, key)
	}
	if len(got) != 0 {
		t.Fatalf("the return seal names more than the staged seal: %v", got)
	}
	// The baseline is the source's own, under the target's evidence.
	base, err := moves.DecodeReturnBaseline(c.Op.Journal[moves.JournalReturnBaseline])
	if err != nil {
		t.Fatal(err)
	}
	src := h.read(t, p1Host).Move.Source
	if base.FinalTXID != staged.N || *base.Files != *src.Files || base.TimersSHA256 != src.TimersSHA256 || base.Evidence != rb.Evidence {
		t.Fatalf("baseline %+v, source %+v", base, src)
	}
	var mj moves.MoveJSON
	if body := h.f.buckets["data-ie"][moves.MoveJSONKey(p1Host, rb.ID)].body; json.Unmarshal([]byte(body), &mj) != nil || mj.Incarnation != h.inc || mj.MoveID != rb.ID {
		t.Fatalf("move.json %q", body)
	}
	// Every lane object went through the guard, under the source's lease.
	if calls := h.laneCalls(rb.ID); len(calls) != 1+len(staged.Replica)+len(staged.Files)+len(staged.Timers) {
		t.Fatalf("lane calls %+v", calls)
	}
	if h.lease.acquired == 0 || h.lease.acquired != h.lease.released {
		t.Fatalf("the staging lease: acquired %d released %d", h.lease.acquired, h.lease.released)
	}
	if strings.Join(h.events, " ") != "rolling_back released returning returned" {
		t.Fatalf("events %v", h.events)
	}
}

// decodeLTXDatabase writes the pages of an LTX snapshot into a database file.
func decodeLTXDatabase(t *testing.T, body []byte, path string) ltx.Header {
	t.Helper()
	dec := ltx.NewDecoder(bytes.NewReader(body))
	if err := dec.DecodeHeader(); err != nil {
		t.Fatal(err)
	}
	hdr := dec.Header()
	f, err := os.Create(path)
	if err != nil {
		t.Fatal(err)
	}
	defer f.Close()
	page := make([]byte, hdr.PageSize)
	for {
		var ph ltx.PageHeader
		if err := dec.DecodePage(&ph, page); errors.Is(err, io.EOF) {
			break
		} else if err != nil {
			t.Fatal(err)
		}
		if _, err := f.WriteAt(page, int64(ph.Pgno-1)*int64(hdr.PageSize)); err != nil {
			t.Fatal(err)
		}
	}
	if err := dec.Close(); err != nil {
		t.Fatal(err)
	}
	if err := f.Truncate(int64(hdr.Commit) * int64(hdr.PageSize)); err != nil {
		t.Fatal(err)
	}
	return hdr
}

// TestDirtyReturnSnapshotsTheStoppedDatabase (witness wDirtyReturn): the
// target served and wrote, so its state is not the staged seal's. The lane
// carries one snapshot [1, N'] of the stopped database, its files archive
// and its timers; the snapshot verifies and restores the target's rows, and
// the return seal names exactly it.
func TestDirtyReturnSnapshotsTheStoppedDatabase(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	dbPath := provision.DefaultDBPath(h.tdir, p1Host)
	d, err := sql.Open("sqlite", dbPath)
	if err != nil {
		t.Fatal(err)
	}
	for i := 0; i < 3; i++ {
		if _, err := d.Exec(`INSERT INTO t (body) VALUES ('written on the target')`); err != nil {
			t.Fatal(err)
		}
	}
	d.Close()
	if err := os.WriteFile(filepath.Join(h.tdir, p1Host, "new.txt"), []byte("uploaded on the target"), 0o644); err != nil {
		t.Fatal(err)
	}
	h.tpos = h.n + 3
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	if !rb.Dirty || rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	c := h.claim(t)
	seal, err := moves.DecodeSeal(c.Op.Journal[moves.JournalReturnSeal])
	if err != nil {
		t.Fatal(err)
	}
	lane := moves.ReturnLane(p1Host, rb.ID)
	want := moves.SnapshotKey(lane+"replica/", h.tpos)
	if seal.N != h.tpos || len(seal.Replica) != 1 || seal.Replica[0].Key != want || !seal.ReachesN() || !seal.FromDB || !seal.SrcCovered {
		t.Fatalf("return seal %+v, want the snapshot %s", seal, want)
	}
	body, err := h.vs.GetVersion(ctx, "data-ie", want, seal.Replica[0].VersionID)
	if err != nil {
		t.Fatal(err)
	}
	if err := ltx.NewDecoder(bytes.NewReader(body)).Verify(); err != nil {
		t.Fatalf("the snapshot does not verify: %v", err)
	}
	restored := filepath.Join(t.TempDir(), "restored.db")
	hdr := decodeLTXDatabase(t, body, restored)
	if hdr.MinTXID != 1 || uint64(hdr.MaxTXID) != h.tpos {
		t.Fatalf("snapshot header %+v", hdr)
	}
	if got := countRowsForTest(t, restored); got != h.rows+3 {
		t.Fatalf("the snapshot holds %d rows, the target %d", got, h.rows+3)
	}
	// The files archive is the target's, bound to the move id (the
	// installer checks it), and the baseline is the target's state.
	base, err := moves.DecodeReturnBaseline(c.Op.Journal[moves.JournalReturnBaseline])
	if err != nil {
		t.Fatal(err)
	}
	plain, _, _, err := archive.PlainDigest(filepath.Join(h.tdir, p1Host), filepath.Join(h.tdir, p1Host+".env"))
	if err != nil {
		t.Fatal(err)
	}
	dbSHA, err := fileSHA256(dbPath)
	if err != nil {
		t.Fatal(err)
	}
	if base.FinalTXID != h.tpos || base.DBSHA256 != dbSHA || base.Files.Plain != plain || base.Files.Sealed == "" || base.Evidence != rb.Evidence {
		t.Fatalf("baseline %+v", base)
	}
	if !sealNamesArchive(seal, base.Files.Sealed) {
		t.Fatalf("the seal's files are not the baseline's archive: %+v", seal.Files)
	}
	for _, o := range seal.Files {
		if !strings.HasSuffix(o.Key, ".json") {
			continue
		}
		raw, err := h.vs.GetVersion(ctx, "data-ie", o.Key, o.VersionID)
		if err != nil {
			t.Fatal(err)
		}
		var am archive.Manifest
		if err := json.Unmarshal(raw, &am); err != nil || am.Op != p1ID || am.Incarnation != h.inc || am.Plain != plain {
			t.Fatalf("archive manifest %+v %v", am, err)
		}
	}
	if len(seal.Timers) != 1 || !strings.HasPrefix(seal.Timers[0].Key, lane+"timers/") {
		t.Fatalf("timers %+v", seal.Timers)
	}
}

// TestReturnLaneWriterIsGuarded: the return lane is a staging lane. A writer
// whose lease is no longer safe for a signature's life sends nothing more; a
// writer that finishes records its lease's bound as the seal's, after every
// request it signed can still land, and the lane's park acknowledges at
// LeaseUntil + H, not before.
func TestReturnLaneWriterIsGuarded(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newRollbackHarness(t)
	h.uncover(t)
	h.rollBack(t)
	// The lease goes unsafe after the first request is signed (a writer
	// paused past its bound): nothing after it is sent, and no seal.
	h.lease.mu.Lock()
	h.lease.reads, h.lease.unsafeAfter = 0, 1
	h.lease.mu.Unlock()
	h.rollbackPass(t)
	rb := h.rb(t)
	if rb.TargetPhase != provision.RbTargetReturning {
		t.Fatalf("rollback %+v", rb)
	}
	if calls := h.laneCalls(rb.ID); len(calls) != 1 {
		t.Fatalf("an unsafe writer sent %d requests: %+v", len(calls), calls)
	}
	if _, ok := h.claim(t).Op.Journal[moves.JournalReturnSeal]; ok {
		t.Fatal("a return seal without the lane")
	}
	if h.lease.acquired != h.lease.released {
		t.Fatalf("the lease is still held: %d %d", h.lease.acquired, h.lease.released)
	}
	// A safe lease: the lane is written and sealed with the lease's bound.
	h.lease.mu.Lock()
	h.lease.unsafeAfter = 0
	h.lease.mu.Unlock()
	h.rollbackPass(t)
	if rb = h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	seal, err := moves.DecodeSeal(h.claim(t).Op.Journal[moves.JournalReturnSeal])
	if err != nil {
		t.Fatal(err)
	}
	if !seal.Bound().Equal(h.lease.until) {
		t.Fatalf("bound %s, lease until %s", seal.Bound(), h.lease.until)
	}
	for _, c := range h.laneCalls(rb.ID) {
		if c.SignAt.Add(moves.SignatureLife + moves.DefaultSkew).After(seal.Bound()) {
			t.Fatalf("%s signed at %s can land after the bound %s", c.Key, c.SignAt, seal.Bound())
		}
	}
	// The lane's park (Plan 5): queued at the source's probed with the
	// seal's bound as its horizon, acknowledged only after it.
	for _, step := range []func(*hostclaim.Claim) error{
		hostclaim.ReturnToSource(p1ID, "ie-box", false, hostclaim.ReturnFence{}),
		hostclaim.SetRestoringPhase(p1ID, hostclaim.PhaseClaimed, hostclaim.PhaseProbing),
		hostclaim.SetRestoringPhase(p1ID, hostclaim.PhaseProbing, hostclaim.PhaseProbed),
		hostclaim.QueueReturnLanePark(p1ID, "ie", hostclaim.ParkScope{Location: moves.ScopeLocation(p1Host, rb.ID), Incarnation: h.inc,
			EvidenceAt: seal.Bound(), RequestLimit: cutRequestLimit}, obligationID),
	} {
		obs, err := hostclaim.Read(ctx, h.s, p1Host)
		if err != nil {
			t.Fatal(err)
		}
		if _, err := hostclaim.Update(ctx, h.s, obs, step); err != nil {
			t.Fatal(err)
		}
	}
	var chore hostclaim.Obligation
	for _, o := range h.claim(t).Obligations {
		if o.Kind == "park-replica" && o.Region == "ie" {
			chore = o
		}
	}
	e := park.Engine{Store: h.vs, Claims: h.s, Bucket: "data-ie", Region: "ie", Leases: h.leases, Now: h.clock.Now, NewID: obligationID,
		// Plan 5's view stands in here; a lane is never the holder's
		// location, so nothing in it is protected either way.
		View: func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
			return park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true,
				GenerationOf: func(park.Ref) (string, bool) { return "", true }}, nil
		},
		Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }}
	work := park.Work{Host: p1Host, ID: chore.ID, Op: p1ID}
	horizon := seal.Bound().Add(park.SignatureAge + cutRequestLimit)
	h.clock.mu.Lock()
	h.clock.now = horizon.Add(-time.Second)
	h.clock.mu.Unlock()
	if err := e.Run(ctx, work); !errors.Is(err, park.ErrPending) {
		t.Fatalf("before LeaseUntil + H: %v", err)
	}
	if parked, err := e.Parked(ctx, work); err != nil || !parked {
		t.Fatalf("the lane's archive: %v %v", parked, err)
	}
	h.clock.Advance(2 * time.Second)
	if err := e.Run(ctx, work); err != nil {
		t.Fatalf("after LeaseUntil + H: %v", err)
	}
	for _, o := range h.claim(t).Obligations {
		if o.ID == chore.ID {
			t.Fatalf("the lane's park was not acknowledged: %+v", o)
		}
	}
}

// TestTargetCleaningWaitsForRecoveredAt (Review Focus 4): the target cleans
// only once the source has recovered; until then it keeps everything it has,
// at released (clean) or returned (dirty).
func TestTargetCleaningWaitsForRecoveredAt(t *testing.T) {
	t.Parallel()
	for _, dirty := range []bool{false, true} {
		t.Run("dirty "+strconv.FormatBool(dirty), func(t *testing.T) {
			h := newRollbackHarness(t)
			if dirty {
				h.uncover(t)
			}
			h.rollBack(t)
			want := provision.RbTargetReleased
			if dirty {
				want = provision.RbTargetReturned
			}
			for i := 0; i < 3; i++ {
				h.rollbackPass(t)
				if rb := h.rb(t); rb.TargetPhase != want || rb.TargetParkedAt != nil {
					t.Fatalf("pass %d: %+v", i, rb)
				}
			}
			if h.rowGone || len(h.deleted) != 0 {
				t.Fatal("the row went before the source recovered")
			}
			if _, err := os.Stat(provision.DefaultDBPath(h.tdir, p1Host)); err != nil {
				t.Fatalf("the database went before the source recovered: %v", err)
			}
			h.recovered(t)
			h.rollbackPass(t)
			if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
				t.Fatalf("after the recovery: %+v", rb)
			}
		})
	}
}

// TestTargetCleaningLeavesNoLiveCopy (INV-10): after cleaned nothing of the
// host is live on the target, in its bucket under instances/<host>/ or
// moves/<host>/<mv>/, or on its disk; everything is archived under
// parked/<inc>/<host>/<retirement>/ (and .parked/ locally), the row and its
// timers are gone, and the park-local chore is complete.
func TestTargetCleaningLeavesNoLiveCopy(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	db := provision.DefaultDBPath(h.tdir, p1Host)
	// A WAL the inventory does not list: the moves-owned rename parks it.
	if err := os.WriteFile(db+"-wal", []byte("a wal"), 0o600); err != nil {
		t.Fatal(err)
	}
	h.rollBack(t)
	h.rollbackPass(t)
	rb := h.rb(t)
	h.recovered(t)
	h.rollbackPass(t)
	if rb = h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
		t.Fatalf("rollback %+v", rb)
	}
	var live, parked []string
	for k := range h.f.buckets["data-uk"] {
		switch {
		case strings.HasPrefix(k, cutRoot+"/") && !strings.HasSuffix(k, "/lock.json"),
			strings.HasPrefix(k, moves.StagingPrefix(p1Host, p1ID)):
			live = append(live, k)
		case strings.HasPrefix(k, "parked/"+h.inc+"/"+p1Host+"/"):
			parked = append(parked, k)
		}
	}
	if len(live) != 0 || len(parked) == 0 {
		t.Fatalf("live %v parked %d", live, len(parked))
	}
	for _, p := range sourceArtifactPaths(p1Host) {
		if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
			t.Fatalf("%s is still live: %v", p, err)
		}
	}
	if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the op journal is still live: %v", err)
	}
	var wal, dbs int
	filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d os.DirEntry, err error) error {
		if err == nil && !d.IsDir() {
			switch filepath.Base(p) {
			case p1Host + ".db-wal":
				wal++
			case p1Host + ".db":
				dbs++
			}
		}
		return nil
	})
	if wal != 1 || dbs != 1 {
		t.Fatalf("parked locally: wal %d db %d", wal, dbs)
	}
	if !h.rowGone || len(h.deleted) != 1 || h.deleted[0] != p1Host+"="+rb.ID {
		t.Fatalf("the row: %v", h.deleted)
	}
	c := h.claim(t)
	for _, o := range c.Obligations {
		if o.Kind == "park-local" {
			t.Fatalf("the park-local is still owed: %+v", o)
		}
	}
	if len(c.Inventory["uk-box"].Paths) != 0 {
		t.Fatalf("inventory %+v", c.Inventory["uk-box"])
	}
	if captured, err := hostguard.Capture(h.tdir, p1Host); err != nil || captured.Stamp != 0 {
		t.Fatalf("the installation record: %+v %v", captured, err)
	}
	// A later pass has nothing to route to.
	if role := movesParticipantRecord(h.rollbackCfg(t), h.read(t, p1Host)); role != roleNone {
		t.Fatalf("a cleaned target still routes as %s", role)
	}
	if !strings.HasSuffix(strings.Join(h.events, " "), "cleaning cleaned") {
		t.Fatalf("events %v", h.events)
	}
}

// Every release and return step is replay-safe: a crash after the record's
// released leaves the claim at releasing, and the next pass writes it; a
// crash after the return seal's claim write leaves the record at returning,
// and the next pass finishes without writing the lane again.
func TestTargetRollbackReplays(t *testing.T) {
	t.Parallel()
	t.Run("after-record-released", func(t *testing.T) {
		h := newRollbackHarness(t)
		h.rollBack(t)
		h.rbAt = "after-record-released"
		h.rollbackPass(t)
		if !h.rbDone || h.claim(t).Op.Phase != hostclaim.PhaseReleasing {
			t.Fatalf("the crash: %v %+v", h.rbHits, h.claim(t).Op)
		}
		h.rollbackPass(t)
		if c := h.claim(t); c.Op.Phase != hostclaim.PhaseReleased || c.Op.Journal[hostclaim.JournalEvidenceTarget] != targetStoppedKey() {
			t.Fatalf("claim %+v", c.Op)
		}
	})
	t.Run("after-return-seal", func(t *testing.T) {
		h := newRollbackHarness(t)
		h.uncover(t)
		h.rollBack(t)
		h.rbAt = "after-return-seal"
		h.rollbackPass(t)
		rb := h.rb(t)
		if !h.rbDone || rb.TargetPhase != provision.RbTargetReturning {
			t.Fatalf("the crash: %v %+v", h.rbHits, rb)
		}
		calls := len(h.laneCalls(rb.ID))
		h.rollbackPass(t)
		if rb = h.rb(t); rb.TargetPhase != provision.RbTargetReturned || len(h.laneCalls(rb.ID)) != calls {
			t.Fatalf("the replay: %+v, %d lane calls after %d", rb, len(h.laneCalls(rb.ID)), calls)
		}
	})
}

// The release waits while the target cannot prove its instance stopped,
// and touches nothing the evidence would vouch for.
func TestTargetReleaseWaitsForStopEvidence(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	h.rollBack(t)
	h.ev.stopped = false
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleasing || h.tchecks != 0 {
		t.Fatalf("rollback %+v checkpoints %d", rb, h.tchecks)
	}
	if c := h.claim(t); c.Op.Exec != "uk-box" || c.Op.Phase != hostclaim.PhaseReleasing {
		t.Fatalf("claim %+v", c.Op)
	}
	if _, _, err := h.s.Get(context.Background(), targetStoppedKey()); err == nil {
		t.Fatal("a stopped report without stop evidence")
	}
	h.ev.stopped = true
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleased {
		t.Fatalf("rollback %+v", rb)
	}
}

// Production wires the hold hand-over and leaves the rollback unarmed with
// the park engine's hooks (ruling J18), as the landing is.
func TestTargetRollbackUnarmedInProduction(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	c := h.rollbackCfg(t)
	if !rollbackTargetArmed(c) {
		t.Fatal("the test configuration is not armed")
	}
	c.ParkLeases = nil
	if rollbackTargetArmed(c) {
		t.Fatal("armed without park leases")
	}
	h.rollBack(t)
	h.rw.cfg = c
	if err := movesPass(context.Background(), h.s, h.rw); err != nil {
		t.Fatal(err)
	}
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReleasing || len(h.handed) != 0 {
		t.Fatalf("an unarmed rollback moved: %+v %v", rb, h.handed)
	}
}

// Fix round 1, ruling 1 (J3): the release re-reads its record under the hold
// lock at the hand-over. A record that changed between the pass's
// observation and the hand-over (here a declared loss took the target) keeps
// the row held by the move, and nothing after the hand-over runs.
func TestTargetReleaseRechecksTheRecordAtTheHandOver(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	h.rollBack(t)
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	h.beforeHand = func() {
		h.beforeHand = nil
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRollbackTargetLost(g, "control/losses/x.json", h.clock.Now())); err != nil {
			t.Fatal(err)
		}
	}
	calls := h.ev.calls
	h.rollbackPass(t)
	if h.route.Status != "landing" || h.route.HoldOp != p1ID || len(h.handed) != 1 {
		t.Fatalf("the hold moved under a changed record: %+v %v", h.route, h.handed)
	}
	if h.ev.calls != calls || h.tchecks != 0 {
		t.Fatalf("the release went on: evidence %d checkpoints %d", h.ev.calls-calls, h.tchecks)
	}
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetLost {
		t.Fatalf("rollback %+v", rb)
	}
	if _, _, err := h.s.Get(context.Background(), targetStoppedKey()); err == nil {
		t.Fatal("a stopped report after a refused hand-over")
	}
}

// targetChanged makes the target's state differ from what it installed: rows
// written by the app on the target, so the return snapshots it.
func (h *rollbackHarness) targetChanged(t *testing.T) {
	t.Helper()
	d, err := sql.Open("sqlite", provision.DefaultDBPath(h.tdir, p1Host))
	if err != nil {
		t.Fatal(err)
	}
	if _, err := d.Exec(`INSERT INTO t (body) VALUES ('written on the target')`); err != nil {
		t.Fatal(err)
	}
	d.Close()
	h.tpos = h.n + 1
}

// stuckAtReturning asserts the record stopped for an operator at the
// target's returning with reason, the target keeping its phase, its row and
// what it holds, and no return seal.
func (h *rollbackHarness) stuckAtReturning(t *testing.T, reason string) {
	t.Helper()
	m := h.read(t, p1Host).Move
	if m.State != provision.MoveStuck || m.StuckReason != reason || m.Rollback.TargetPhase != provision.RbTargetReturning ||
		!m.Rollback.Dirty || m.Rollback.StuckAt != provision.RbSourceWaiting {
		t.Fatalf("not stuck at returning for %q: %+v %+v", reason, m, m.Rollback)
	}
	if h.route.Status != "releasing" || h.route.HoldOp != m.Rollback.ID || h.rowGone {
		t.Fatalf("the row: %+v gone %v", h.route, h.rowGone)
	}
	if _, err := os.Stat(filepath.Join(h.tdir, p1Host, "upload.txt")); err != nil {
		t.Fatalf("the target's files: %v", err)
	}
	if _, ok := h.claim(t).Op.Journal[moves.JournalReturnSeal]; ok {
		t.Fatal("a return seal on a stuck return")
	}
	if h.lease.acquired != h.lease.released {
		t.Fatalf("the staging lease: %d %d", h.lease.acquired, h.lease.released)
	}
}

func (h *rollbackHarness) retry(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	// The claim-backed retry (claim Plan 10): it advances the recovery retry
	// epoch, so the move log keys the next stuck as a new entry (Plan 6
	// Task 3, preflight L11).
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveRetryRecovery(g)); err != nil {
		t.Fatal(err)
	}
}

// Fix round 1, ruling 2: a dirty return that cannot be built from what the
// target holds stops for an operator with a recorded reason, the target at
// returning with everything kept; a retry runs the return again.
func TestDirtyReturnGoesStuck(t *testing.T) {
	t.Parallel()
	t.Run("the database is missing", func(t *testing.T) {
		h := newRollbackHarness(t)
		db := provision.DefaultDBPath(h.tdir, p1Host)
		if err := os.Rename(db, db+".aside"); err != nil {
			t.Fatal(err)
		}
		h.rollBack(t)
		h.rollbackPass(t)
		h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
		// A stuck record is not the target's to work on.
		h.rollbackPass(t)
		h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
		// A retry runs the return again; still missing, it stops again.
		before := len(h.events)
		h.retry(t)
		h.rollbackPass(t)
		h.stuckAtReturning(t, provision.StuckReturnTargetDatabase)
		if got := strings.Join(h.events[before:], " "); got != "stuck" {
			t.Fatalf("the retry did not run the return: events %q", got)
		}
	})
	t.Run("the files cannot be archived", func(t *testing.T) {
		h := newRollbackHarness(t)
		h.targetChanged(t)
		link := filepath.Join(h.tdir, p1Host, "link")
		h.rbHook = func(point string) {
			if point == "after-record-returning" {
				if err := os.Symlink("/etc/passwd", link); err != nil {
					t.Fatal(err)
				}
			}
		}
		h.rollBack(t)
		h.rollbackPass(t)
		h.stuckAtReturning(t, provision.StuckReturnTargetFiles)
		// The operator removes the link and retries: the return completes.
		h.rbHook = nil
		if err := os.Remove(link); err != nil {
			t.Fatal(err)
		}
		h.retry(t)
		h.rollbackPass(t)
		if rb := h.rb(t); h.read(t, p1Host).Move.State != provision.MoveRollingBack || rb.TargetPhase != provision.RbTargetReturned {
			t.Fatalf("after the retry: %+v", rb)
		}
	})
	t.Run("a storage error is retried, not stuck", func(t *testing.T) {
		h := newRollbackHarness(t)
		h.targetChanged(t)
		fail := true
		h.f.failAt = func(op, key string) error {
			if fail && strings.Contains(key, "/replica/") {
				return errors.New("503 SlowDown")
			}
			return nil
		}
		h.rollBack(t)
		h.rollbackPass(t)
		if m := h.read(t, p1Host).Move; m.State != provision.MoveRollingBack || m.Rollback.TargetPhase != provision.RbTargetReturning {
			t.Fatalf("a storage error: %+v %+v", m, m.Rollback)
		}
		fail = false
		h.rollbackPass(t)
		if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
			t.Fatalf("after the storage recovered: %+v", rb)
		}
	})
}

// closeRolledBack is the source's close of the rollback, as Task 5 will
// write it: opened, then MoveRolledBackClose.
func (h *rollbackHarness) closeRolledBack(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	now := h.clock.Now()
	for _, step := range []func(*provision.Record) error{
		provision.MoveRollbackSourcePhase(g, provision.RbSourceRecovered, provision.RbSourceOpened, func(rb *provision.MoveRollback) { rb.OpenedAt = &now }),
		provision.MoveRolledBackClose(g, now),
	} {
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, step); err != nil {
			t.Fatal(err)
		}
	}
}

// Fix round 1, ruling 3: the source closes the record right after the
// target's cleaned CAS, before the target's worker re-reads it, and the
// worker restarts. No record routes the target back, so the local journal
// sweep parks the op's journal (which holds the built files archive) on the
// close's evidence.
func TestTargetJournalSweptAfterTheClose(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	h.targetChanged(t)
	h.rollBack(t)
	h.rollbackPass(t)
	if rb := h.rb(t); rb.TargetPhase != provision.RbTargetReturned {
		t.Fatalf("rollback %+v", rb)
	}
	adir := archiveDir(h.tdir, p1Host, p1ID)
	built, _ := filepath.Glob(filepath.Join(adir, "files-*"))
	if len(built) == 0 {
		t.Fatal("the dirty return built no archive in the journal")
	}
	h.recovered(t)
	h.rbHook = func(point string) {
		if point == "after-record-cleaned" {
			h.closeRolledBack(t)
		}
	}
	h.rbAt = "after-record-cleaned"
	h.rollbackPass(t)
	if !h.rbDone || h.read(t, p1Host).Move != nil {
		t.Fatalf("the fixture: crashed %v, record %+v", h.rbDone, h.read(t, p1Host).Move)
	}
	jdir := moves.JournalDir(h.tdir, p1Host, p1ID)
	if _, err := os.Stat(jdir); err != nil {
		t.Fatalf("the fixture left no journal to sweep: %v", err)
	}
	// A restart: a fresh worker, and the next pass's sweep.
	h.rw, h.rbHook = nil, nil
	h.leases.expire()
	h.rollbackPass(t)
	if _, err := os.Lstat(jdir); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("the op journal is still live: %v", err)
	}
	var parkedArchive int
	filepath.WalkDir(filepath.Join(h.tdir, ".parked"), func(p string, d os.DirEntry, err error) error {
		if err == nil && !d.IsDir() && strings.Contains(p, filepath.Join(".moves", p1Host, p1ID, "archive", "files-")) {
			parkedArchive++
		}
		return nil
	})
	if parkedArchive == 0 {
		t.Fatal("the built archive was not parked with the journal")
	}
}

// The journal sweep leaves a rollback journal alone while its record still
// carries the op without target_parked_at: the cleaning is the pass's.
func TestJournalSweepWaitsForTheTargetPark(t *testing.T) {
	t.Parallel()
	h := newRollbackHarness(t)
	h.rollBack(t)
	h.rollbackPass(t)
	h.recovered(t)
	// Crash right after the moves-owned rename, before the row and the CAS.
	h.rbAt = "after-" + rbStepLocal
	h.rollbackPass(t)
	if !h.rbDone {
		t.Fatalf("never reached %s: %v", h.rbAt, h.rbHits)
	}
	w := &movesWorker{cfg: h.rollbackCfg(t), mirrors: map[string]lease.MirrorStore{}}
	w.sweepFinalJournals(context.Background(), h.s, []provision.Record{h.read(t, p1Host)})
	if _, err := os.Stat(moves.JournalDir(h.tdir, p1Host, p1ID)); err != nil {
		t.Fatalf("the sweep parked a journal whose cleaning is under way: %v", err)
	}
}

// Fix round 1, ruling 4: a crash at each durable step of the cleaning, then a
// fresh worker, finishes it with nothing live left.
func TestTargetCleaningReplays(t *testing.T) {
	slots := make(chan struct{}, 4)
	for _, point := range []string{
		"before-rb-local-retire", "after-rb-local-retire", "after-" + rbStepInstallation,
		"after-" + rbStepLocal + "-plan", "after-rename:" + p1Host + ".db-wal", "after-" + rbStepLocal,
		"after-rb-row-pending", "after-rb-row-delete", "after-" + rbStepRowDeleted,
	} {
		t.Run(point, func(t *testing.T) {
			t.Parallel()
			slots <- struct{}{}
			defer func() { <-slots }()
			h := newRollbackHarness(t)
			if err := os.WriteFile(provision.DefaultDBPath(h.tdir, p1Host)+"-wal", []byte("a wal"), 0o600); err != nil {
				t.Fatal(err)
			}
			h.rollBack(t)
			h.rollbackPass(t)
			h.recovered(t)
			h.rbAt = point
			h.rollbackPass(t)
			if !h.rbDone {
				t.Fatalf("never reached %s: %v", point, h.rbHits)
			}
			if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaning {
				t.Fatalf("after the crash: %+v", rb)
			}
			h.rw = nil
			h.leases.expire()
			h.rollbackPass(t)
			if rb := h.rb(t); rb.TargetPhase != provision.RbTargetCleaned || rb.TargetParkedAt == nil {
				t.Fatalf("after the replay: %+v", rb)
			}
			for _, p := range sourceArtifactPaths(p1Host) {
				if _, err := os.Lstat(filepath.Join(h.tdir, p)); !errors.Is(err, os.ErrNotExist) {
					t.Fatalf("%s is still live: %v", p, err)
				}
			}
			if _, err := os.Lstat(moves.JournalDir(h.tdir, p1Host, p1ID)); !errors.Is(err, os.ErrNotExist) {
				t.Fatalf("the op journal is still live: %v", err)
			}
			if !h.rowGone {
				t.Fatal("the row is still there")
			}
			if c := h.claim(t); len(c.Inventory["uk-box"].Paths) != 0 {
				t.Fatalf("inventory %+v", c.Inventory["uk-box"])
			}
		})
	}
}