package host

import (
	"context"
	"encoding/json"
	"errors"
	"net/http/httptest"
	"os"
	"path/filepath"
	"sort"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/moves/archive"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
)

// The cut harness: the Phase 1 harness driven to ready (manual cut) and
// cut as the door does, with the cut's hooks faked. fakeEvidence answers a
// configurable report; the final checkpoint writes the stopped database's
// last transaction (TXID 3) into the source's replica; cutFault is the
// crash seam, crashing once at the point named.

var errCrash = errors.New("injected crash")

type fakeEvidence struct {
	calls   int
	stopped bool
	err     error
	onCall  func()
}

func (e *fakeEvidence) report() moves.StoppedReport {
	r := moves.StoppedReport{V: 1, Host: p1Host, Steps: map[string]moves.Check{}, CheckedAt: time.Date(2026, 10, 7, 9, 1, 0, 0, time.UTC)}
	for _, id := range []string{moves.StepTree, moves.StepStray, moves.StepFDs, moves.StepLease} {
		r.Steps[id] = moves.Check{OK: true}
	}
	if !e.stopped {
		r.Steps[moves.StepFDs] = moves.Check{Reason: "1 process(es) hold the database open"}
	}
	return r
}

type cutHarness struct {
	*phase1Harness
	ev          *fakeEvidence
	holds       []string
	fences      int
	drains      int
	checkpoints int
	locations   []string
	bundle      map[string]string
	localPos    uint64
	timers      []registry.Timer
	crashAt     string
	crashed     bool
	hits        []string
	// scoped keeps the Phase 1 harness's scoped holder
	// (instances/<host>/<incarnation>), which the cut refuses (ruling
	// T5-lease). By default ready moves the claim to the root location the
	// activator's lease lives at, instances/<host>, generation cur.
	scoped bool
}

// cutRoot is the activator's lease prefix for the harness's host, the one
// location Phase 1 cuts (ruling T5-lease); cutLineage is the claim
// generation's replica under it, the prefix the lane mirrors.
var (
	cutRoot    = "instances/" + p1Host
	cutLineage = cutRoot + "/g-cur/"
)

// PutFileIfAbsent makes the fake a lease.FilePutIfAbsent.
func (v *fakeVersioned) PutFileIfAbsent(ctx context.Context, bucket, key, path string) (string, error) {
	body, err := os.ReadFile(path)
	if err != nil {
		return "", err
	}
	return v.PutIfAbsent(ctx, bucket, key, body)
}

var _ lease.FilePutIfAbsent = (*fakeVersioned)(nil)

const lastTXKey = "0000/0000000000000003-0000000000000003.ltx"

func newCutHarness(t *testing.T, opts ...p1Option) *cutHarness {
	t.Helper()
	h := &cutHarness{phase1Harness: newPhase1Harness(t, opts...), ev: &fakeEvidence{stopped: true}, localPos: 3,
		bundle: map[string]string{"default": "e1", "environment": ""}}
	h.timers = []registry.Timer{{Host: p1Host, Name: "nightly", Kind: "cron", Spec: "0 3 * * *", NextAt: 1000}}
	last := ltxForTest(t, 4096, 4, 3, 3)
	if err := os.WriteFile(h.route.DBPath, []byte("the stopped database"), 0o600); err != nil {
		t.Fatal(err)
	}
	h.cfg.HoldMoving = func(host, op string) error { h.holds = append(h.holds, host+"="+op); return nil }
	h.cfg.Fence = func(context.Context, string, uint64) error { h.fences++; return nil }
	h.cfg.TimersDrained = func(context.Context, string) error { h.drains++; return nil }
	h.cfg.Evidence = func(_ context.Context, host, location string, _ []string) (moves.StoppedReport, error) {
		h.ev.calls++
		h.locations = append(h.locations, location)
		if h.ev.onCall != nil {
			h.ev.onCall()
		}
		return h.ev.report(), h.ev.err
	}
	h.cfg.FinalCheckpoint = func(ctx context.Context, host, _ string, holder hostclaim.Holder, check func(context.Context) error) error {
		if err := check(ctx); err != nil {
			return err
		}
		h.checkpoints++
		prefix, _ := moves.ReplicaSource(holder)
		if host != p1Host || prefix != cutLineage {
			t.Errorf("checkpoint of %s at %s, want %s at the claim's lineage %s", host, prefix, p1Host, cutLineage)
		}
		h.f.store(h.srcBucket, prefix+lastTXKey, fakeMoveObject{lease.MirrorObject{ETag: "cur-c", Size: int64(len(last))}, string(last)})
		return nil
	}
	h.cfg.LocalPosition = func(string) (uint64, error) { return h.localPos, nil }
	h.cfg.Timers = func(string) ([]registry.Timer, error) { return h.timers, nil }
	h.cfg.InputVersions = func(context.Context, release.Store, provision.Record) (map[string]string, string, error) {
		out := map[string]string{}
		for k, v := range h.bundle {
			out[k] = v
		}
		return out, "", nil
	}
	h.cfg.cutFault = func(point string) error {
		h.hits = append(h.hits, point)
		if point == h.crashAt && !h.crashed {
			h.crashed = true
			return errCrash
		}
		return nil
	}
	// The cut guard: the source can recover a failed cut, as every box
	// will once the wake-source gate opens (armStandIns).
	armStandIns(&h.cfg, h.s, roleSource)
	return h
}

// targetCfg composes with the harness's current bundle versions.
func (h *cutHarness) target() moveCfg {
	c := h.targetCfg()
	c.ComposeHeldEnv = func(ctx context.Context, s release.Store, rec provision.Record, carried []byte) (map[string]string, map[string]bool, []string, provision.MoveInputs, error) {
		env, instance, err := h.heldEnv(ctx, s, rec, carried)
		if err != nil {
			return nil, nil, nil, provision.MoveInputs{}, err
		}
		b := map[string]string{}
		for k, v := range h.bundle {
			b[k] = v
		}
		return env, instance, nil, provision.MoveInputs{BundleETags: b}, nil
	}
	return c
}

// ready drives the move to a persisted ready with its inputs pinned. The
// claim holder is at the root location unless the harness is scoped.
func (h *cutHarness) ready(t *testing.T) {
	t.Helper()
	h.start(t, "manual")
	if !h.scoped {
		rootHolder(t, h.phase1Harness)
	}
	h.pass(t)
	h.targetPass(t, h.target())
	h.pass(t)
	if got := h.read(t, p1Host).Move; got.State != provision.MoveReady || got.Inputs == nil {
		t.Fatalf("the fixture did not reach ready: %+v", got)
	}
}

// rootHolder moves a started Phase 1 harness to the root location: the
// Phase 1 harness keeps the replica under the scoped location, the cut
// accepts only the root one (ruling T5-lease), so the same objects move
// there and the claim follows.
func rootHolder(t *testing.T, h *phase1Harness) {
	t.Helper()
	scoped := "instances/" + p1Host + "/" + h.inc + "/"
	for k, o := range h.f.buckets[h.srcBucket] {
		if rel, ok := strings.CutPrefix(k, scoped); ok {
			delete(h.f.buckets[h.srcBucket], k)
			h.f.store(h.srcBucket, cutRoot+"/"+rel, o)
		}
	}
	rewriteP1Claim(t, h.s, func(c *hostclaim.Claim) { c.Holder.Location = cutRoot })
}

// cut is the console's cut door: ready to cutting/requested.
func (h *cutHarness) cut(t *testing.T) {
	t.Helper()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); err != nil {
		t.Fatal(err)
	}
}

func (h *cutHarness) move(t *testing.T) *provision.MoveState {
	t.Helper()
	return h.read(t, p1Host).Move
}

// crashPass runs one pass that must crash at h.crashAt.
func (h *cutHarness) crashPass(t *testing.T) {
	t.Helper()
	h.pass(t)
	if !h.crashed {
		t.Fatalf("the pass never reached %s (points reached: %v)", h.crashAt, h.hits)
	}
}

// freshWorker drops the source worker: the next pass starts with no
// in-memory state, as a restarted agent does.
func (h *cutHarness) freshWorker() { h.worker1 = nil }

func (h *cutHarness) stoppedKey() string {
	return moves.StoppedKey(moves.ReportPrefix("bab", "app", p1Host, p1ID), "source")
}

func (h *cutHarness) seal(t *testing.T) moves.Seal {
	t.Helper()
	s, err := journalledSeal(h.claim(t))
	if err != nil {
		t.Fatal(err)
	}
	return s
}

func (h *cutHarness) journalArchive(t *testing.T) cutArchiveStep {
	t.Helper()
	j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
	if err != nil {
		t.Fatal(err)
	}
	var step cutArchiveStep
	if err := json.Unmarshal([]byte(j.Steps[cutStepArchive]), &step); err != nil {
		t.Fatalf("no archive identity journalled: %v", err)
	}
	step.dir = archiveDir(h.cfg.DataDir, p1Host, p1ID)
	return step
}

// Admission needs this round's own passing source report, a fresh passing
// target report, and the pin to match (G1, G13). With all of them the
// record is admitted with cut_at, the budget and the deadline; a stale
// target report or a failing source sends it back to ready with the pin
// cleared, and a passing round pins again.
func TestCutAdmitsOnlyFreshPassingReports(t *testing.T) {
	t.Parallel()
	h := newCutHarness(t)
	h.ready(t)
	h.ev.stopped = false // stop at stopping, to look at the admission
	h.cut(t)
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping || m.CutAt == nil || m.CutBudgetS != int(moves.CutWindowDefault) || m.CutDeadline == nil {
		t.Fatalf("after the admission: %+v", m)
	}
	if want := moves.CutDeadline(*m.CutAt, moves.CutWindowDefault); !m.CutDeadline.Equal(want) {
		t.Fatalf("deadline %v, want %v", m.CutDeadline, want)
	}

	for name, spoil := range map[string]func(h *cutHarness){
		"the target's report is stale": func(h *cutHarness) { h.clock.Advance(moves.ReportFreshness + time.Minute) },
		"the source does not answer":   func(h *cutHarness) { h.probe = moves.Check{Reason: s1NoAnswer(p1Host)} },
		"a second box in the target region": func(h *cutHarness) {
			if _, err := h.s.Put(context.Background(), "control/edges/uk-box2.json",
				[]byte(`{"v":1,"box":"uk-box2","eip":"1.2.3.9","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
				panic(err)
			}
		},
	} {
		t.Run(name, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.cut(t)
			spoil(h)
			h.pass(t)
			m := h.move(t)
			if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.CutAt != nil {
				t.Fatalf("not back to ready: %+v", m)
			}
			if h.ev.calls != 0 || len(h.holds) != 0 {
				t.Fatalf("something stopped: %d evidence calls, holds %v", h.ev.calls, h.holds)
			}
		})
	}
}

// The inputs pinned at readiness are compared at requested and again at
// admitted (§7.4). A drift goes back to ready; the target reports the new
// versions, a passing round pins them (MoveGuardedRepin), and the next cut
// admits: no loop.
func TestCutGoesBackToReadyWhenInputsMove(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newCutHarness(t)
	h.ready(t)
	h.ev.stopped = false
	h.bundle["environment"] = "e2"
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil {
		t.Fatalf("a drift at requested did not go back to ready: %+v", m)
	}
	// Ready without a pin is never cut: the door refuses.
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); !errors.Is(err, provision.ErrMoveConflict) {
		t.Fatalf("cut without a pin: %v", err)
	}
	h.targetPass(t, h.target())
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveReady || m.Inputs == nil || m.Inputs.BundleETags["environment"] != "e2" {
		t.Fatalf("not pinned again: %+v", m)
	}
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopping {
		t.Fatalf("the repinned cut did not admit: %+v", m)
	}

	// At admitted: a record admitted before the drift goes back to ready
	// on the replay of admitted.
	h2 := newCutHarness(t)
	h2.ready(t)
	h2.cut(t)
	now := h2.clock.Now()
	if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
		t.Fatal(err)
	}
	h2.bundle["default"] = "e9"
	h2.pass(t)
	if m := h2.move(t); m.State != provision.MoveReady || m.Inputs != nil || h2.ev.calls != 0 {
		t.Fatalf("a drift at admitted: %+v, %d evidence calls", m, h2.ev.calls)
	}
}

// Review Focus 1: a fresh worker on the same store, with the record at
// stopping and nothing in memory, holds, fences, drains, asks for the
// evidence and advances; the cut then runs to the flip in one pass.
func TestCutResumesStoppingFromEvidenceAlone(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	h.ev.stopped = false
	h.cut(t)
	h.pass(t)
	h.pass(t)
	if m := h.move(t); m.CutPhase != provision.CutStopping || h.ev.calls != 2 {
		t.Fatalf("an unproven stop advanced: %+v after %d calls", m, h.ev.calls)
	}
	if h.claim(t).Op.Exec != "ie-box" {
		t.Fatal("execution moved before the evidence")
	}
	h.freshWorker()
	h.ev.stopped = true
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped {
		t.Fatalf("the resumed cut: %+v", m)
	}
	if h.ev.calls != 3 || h.fences != 3 || h.drains != 3 || len(h.holds) != 3 || h.holds[0] != p1Host+"="+p1ID {
		t.Fatalf("evidence %d, fences %d, drains %d, holds %v", h.ev.calls, h.fences, h.drains, h.holds)
	}
	for _, loc := range h.locations {
		if loc != cutRoot {
			t.Fatalf("the lease step read %s, not the activator's lease location %s", loc, cutRoot)
		}
	}
}

// Ruling T5-lease: Phase 1 cuts only a holder whose location is the
// activator's own lease prefix, activator.HibernatePrefix(prefix, host),
// because the final checkpoint and the stop evidence take that very lease
// object. A scoped holder (instances/<host>/<incarnation>) goes back to
// ready at requested and at admitted, with the reason recorded, and
// nothing stops. The refusal stands: a passing round pins again and the
// next cut is refused the same way.
func TestCutRefusesAScopedHolder(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	check := func(t *testing.T, h *cutHarness, when string) {
		t.Helper()
		m := h.move(t)
		if m.State != provision.MoveReady || m.CutPhase != "" || m.Inputs != nil || m.Reason != provision.RefusedReplicaLayout {
			t.Fatalf("%s: not refused back to ready with the reason: %+v", when, m)
		}
		if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 || h.checkpoints != 0 {
			t.Fatalf("%s: something stopped: evidence %d, holds %v, fences %d, checkpoints %d", when, h.ev.calls, h.holds, h.fences, h.checkpoints)
		}
		if c := h.claim(t); c.Op.Exec != "ie-box" || c.Op.Kind != hostclaim.KindMoving {
			t.Fatalf("%s: the claim moved: %+v", when, c.Op)
		}
	}
	h := newCutHarness(t)
	h.scoped = true
	h.ready(t)
	if loc := h.claim(t).Holder.Location; loc != "instances/"+p1Host+"/"+h.inc {
		t.Fatalf("fixture: holder at %s", loc)
	}
	h.cut(t)
	h.pass(t)
	check(t, h, "at requested")
	// Fix round 2: the refusal stands and nothing loops. decideReady never
	// repins a scoped holder, so later passes write nothing, and the door
	// refuses a cut with no pin.
	refused := h.move(t)
	h.pass(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveReady || m.Inputs != nil || m.Reason != provision.RefusedReplicaLayout || m.UpdatedAt != refused.UpdatedAt {
		t.Fatalf("passes after the refusal wrote the record: %+v", m)
	}
	g := provision.MoveGuard{ID: p1ID, Incarnation: h.inc}
	if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveGuardedCut(g)); !errors.Is(err, provision.ErrMoveConflict) {
		t.Fatalf("a cut of the refused move: %v", err)
	}

	h2 := newCutHarness(t)
	h2.scoped = true
	h2.ready(t)
	h2.cut(t)
	now := h2.clock.Now()
	if _, err := provision.SetRecordMove(ctx, h2.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
		t.Fatal(err)
	}
	h2.pass(t)
	check(t, h2, "at admitted")
}

// Ruling T5-lease: the cut's final checkpoint takes the activator's own
// lease object, with the real lease package against a fake S3. While the
// activator's writer holds it the cut cannot checkpoint: execution stays
// with the source, no report is published and nothing but the lock is
// written. The production hook refuses a holder at any other location
// before it asks for a lease at all.
func TestCutCheckpointWaitsForTheActivatorsLease(t *testing.T) {
	ctx := context.Background()
	fake := leasetest.NewVersionedBucket()
	fake.AddBucket("data-ie")
	handler, requests := leasetest.NewS3Server(fake, "data-ie")
	srv := httptest.NewServer(handler) // 127.0.0.1 only
	defer srv.Close()
	t.Setenv("CARLOS_S3_ENDPOINT", srv.URL)
	t.Setenv("AWS_ACCESS_KEY_ID", "test-key")
	t.Setenv("AWS_SECRET_ACCESS_KEY", "test-secret")
	t.Setenv("AWS_SESSION_TOKEN", "")
	t.Setenv("AWS_PROFILE", "")
	t.Setenv("AWS_REGION", "")

	h := newCutHarness(t)
	h.cfg.FinalCheckpoint = finalCheckpoint("data-ie", "instances", "eu-west-1")
	h.ready(t)
	// The activator's writer, as ExecBackend.Acquire takes it.
	writer, err := lease.Acquire(ctx, "data-ie", activator.HibernatePrefix("instances", p1Host), "eu-west-1")
	if err != nil {
		t.Fatal(err)
	}
	defer writer.Release(ctx)
	h.cut(t)
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveCutting || m.CutPhase != provision.CutStopped {
		t.Fatalf("with the writer's lease held: %+v", m)
	}
	if exec := h.claim(t).Op.Exec; exec != "ie-box" {
		t.Fatalf("execution left the source: %q", exec)
	}
	if _, _, err := h.s.Get(ctx, h.stoppedKey()); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a stopped report was published: %v", err)
	}
	vers, err := fake.ListVersions(ctx, "data-ie", "")
	if err != nil {
		t.Fatal(err)
	}
	for _, v := range vers {
		if v.Key != activator.HibernatePrefix("instances", p1Host)+"/lock.json" {
			t.Fatalf("the cut wrote %s while the writer held the lease", v.Key)
		}
	}
	meta := filepath.Join(filepath.Dir(h.route.DBPath), ".activator-"+filepath.Base(h.route.DBPath)+"-litestream")
	if _, err := os.Stat(meta); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("a checkpoint ran: %s exists (%v)", meta, err)
	}
	holder := *h.claim(t).Holder
	var he *lease.HeldError
	if err := h.cfg.FinalCheckpoint(ctx, p1Host, h.route.DBPath, holder, func(context.Context) error { return nil }); !errors.As(err, &he) {
		t.Fatalf("the hook against the held lease: %v", err)
	}
	// A holder away from the activator's prefix: refused with no request.
	before := requests()
	foreign := holder
	foreign.Location = "instances/" + p1Host + "/" + h.inc
	err = h.cfg.FinalCheckpoint(ctx, p1Host, h.route.DBPath, foreign, func(context.Context) error { return nil })
	if !errors.Is(err, errHolderLocation) || requests() != before {
		t.Fatalf("a scoped holder: %v (%d requests)", err, requests()-before)
	}
}

// K3 and K13: the stopped report exists, complete with the replica step
// and N, before the claim's execution is none. A crash after the
// checkpoint leaves execution with the source and no report; a crash after
// the report leaves one report, which the replay reuses as it is.
func TestStoppedReportIsCompleteBeforeExecToNone(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	for _, point := range []string{"after-checkpoint", "after-report", "after-record-source", "after-exec-none"} {
		t.Run(point, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.crashAt = point
			h.cut(t)
			h.crashPass(t)
			if m := h.move(t); m.CutPhase != provision.CutStopped {
				t.Fatalf("after the crash: %+v", m)
			}
			body, _, err := h.s.Get(ctx, h.stoppedKey())
			exec := h.claim(t).Op.Exec
			switch point {
			case "after-checkpoint":
				if !errors.Is(err, os.ErrNotExist) || exec != "ie-box" {
					t.Fatalf("report %v, exec %q", err, exec)
				}
			default:
				var r moves.StoppedReport
				if err != nil || json.Unmarshal(body, &r) != nil || !r.Complete() || r.FinalTXID != 3 || r.DBSHA256 == "" || r.TimersSHA256 == "" || r.FilesPlain == "" {
					t.Fatalf("report %v: %+v", err, r)
				}
				if want := point == "after-exec-none"; (exec == "") != want {
					t.Fatalf("exec %q at %s", exec, point)
				}
			}
			h.freshWorker()
			h.pass(t)
			if m := h.move(t); m.State != provision.MoveLanding {
				t.Fatalf("the replay did not finish: %+v", m)
			}
			after, _, err := h.s.Get(ctx, h.stoppedKey())
			if err != nil {
				t.Fatal(err)
			}
			if point != "after-checkpoint" && string(after) != string(body) {
				t.Fatalf("the replay published a second report:\n%s\n%s", body, after)
			}
			keys, _ := h.s.List(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID))
			n := 0
			for _, k := range keys {
				if strings.Contains(k, "stopped-") {
					n++
				}
			}
			if n != 1 {
				t.Fatalf("%d stopped reports: %v", n, keys)
			}
			src := h.move(t).Source
			if src == nil || src.FinalTXID != 3 || src.Evidence != h.stoppedKey() || src.Files == nil || src.Files.Plain == "" || src.Files.Sealed == "" {
				t.Fatalf("source baselines: %+v", src)
			}
			c := h.claim(t)
			if c.Op.Journal[hostclaim.JournalEvidenceSource] != h.stoppedKey() || c.Op.Journal[hostclaim.JournalCutSession] == "" {
				t.Fatalf("claim journal: %v", c.Op.Journal)
			}
		})
	}
}

// Review Focus 2: three final rounds that defer revert the cut, and the
// claim follows once execution is none: reverting under the source, the
// lane's park queued with the writers' horizon.
func TestCutRevertsAfterThreeDeferredRounds(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	n := 0
	h.f.beforeCopy = func(srcKey string) {
		if strings.HasSuffix(srcKey, lastTXKey) {
			// The source changes under the copy: the conditional copy is
			// refused and the object deferred.
			h.f.bmu.Lock()
			n++
			o := h.f.buckets["data-ie"][srcKey]
			o.ETag = "changed-" + string(rune('a'+n))
			h.f.buckets["data-ie"][srcKey] = o
			h.f.bmu.Unlock()
		}
	}
	h.cut(t)
	for i := 1; i <= 2; i++ {
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveCutting || m.CutPhase != provision.CutStopped {
			t.Fatalf("after deferred round %d: %+v", i, m)
		}
		if got := h.ledger(t).DeferredRounds; got != i {
			t.Fatalf("ledger counts %d deferred rounds, want %d", got, i)
		}
	}
	h.pass(t)
	m := h.move(t)
	if m.State != provision.MoveReverting || m.Reason != provision.RevertDeferredRounds {
		t.Fatalf("after three deferred rounds: %+v", m)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" || c.Op.Journal[hostclaim.JournalRevPhase] != "evidence" {
		t.Fatalf("claim: %+v", c.Op)
	}
	var park *hostclaim.ParkScope
	for _, o := range c.Obligations {
		if o.Op == p1ID && o.Kind == "park-replica" && o.Region == "uk" {
			park = o.Park
		}
	}
	if park == nil || park.Location != moves.ScopeLocation(p1Host, p1ID) || park.Incarnation != h.inc || park.RequestLimit != cutRequestLimit {
		t.Fatalf("staging park: %+v", park)
	}
	if !park.EvidenceAt.After(h.clock.now.Add(moves.SignatureLife)) {
		t.Fatalf("evidence_at %v is not past every signature's life", park.EvidenceAt)
	}
	// A replay changes nothing: the claim is reverting already.
	h.pass(t)
	if c2 := h.claim(t); c2.Gen != c.Gen {
		t.Fatalf("a replay wrote the claim again (%d to %d)", c.Gen, c2.Gen)
	}
}

// §7.3 seal by version: the seal names N and the exact version of every
// current replica object, of the timers and of both archive objects. The
// witness (wSealByVersion): a stale copy that lands after the seal makes a
// version the seal does not name, and the seal still names the earlier one.
func TestSealNamesEveryCurrentVersionAndN(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding {
		t.Fatalf("cut did not finish: %+v", m)
	}
	seal := h.seal(t)
	if seal.N != 3 || !seal.ReachesN() || !seal.FromDB || !seal.SrcCovered || len(seal.Timers) != 1 || len(seal.Files) != 2 {
		t.Fatalf("seal: %+v", seal)
	}
	lane := moves.ReplicaPrefix(p1Host, p1ID)
	want := []string{lane + "0000/0000000000000001-0000000000000002.ltx", lane + "0001/0000000000000001-0000000000000002.ltx", lane + lastTXKey}
	sort.Strings(want)
	var got []string
	for _, o := range seal.Replica {
		got = append(got, o.Key)
		if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID {
			t.Fatalf("%s: sealed %s, current %s", o.Key, o.VersionID, cur)
		}
	}
	sort.Strings(got)
	if strings.Join(got, ",") != strings.Join(want, ",") {
		t.Fatalf("sealed %v, want %v", got, want)
	}
	for _, o := range append(append([]moves.SealedObject{}, seal.Timers...), seal.Files...) {
		if cur := h.f.versionOf("data-uk", o.Key); cur != o.VersionID || !lease.UsableVersionID(o.VersionID) {
			t.Fatalf("%s: sealed %q, current %q", o.Key, o.VersionID, cur)
		}
	}
	if !seal.Bound().After(seal.SealedAt) {
		t.Fatalf("bound %v is not after the seal %v", seal.Bound(), seal.SealedAt)
	}
	// The witness.
	stale := seal.Replica[0]
	if _, _, err := h.f.CopyVersionedAt(context.Background(), h.clock.Now(), "data-ie",
		cutLineage+"0000/0000000000000001-0000000000000002.ltx", "", "data-uk", stale.Key); err != nil {
		t.Fatal(err)
	}
	if cur := h.f.versionOf("data-uk", stale.Key); cur == stale.VersionID {
		t.Fatal("the stale copy did not make a new version")
	}
	if again := h.seal(t); again.Replica[0].VersionID != stale.VersionID {
		t.Fatal("the seal moved with the stale copy")
	}
	// Every staging write of the cut went through the guard.
	for _, op := range h.f.ops {
		if strings.Contains(op, "data-uk/") && !strings.HasPrefix(op, "copy ") {
			t.Fatalf("an unguarded write reached the target bucket: %s", op)
		}
	}
	for _, c := range h.f.stagingCalls() {
		if c.SignAt.IsZero() || c.Bucket != "data-uk" {
			t.Fatalf("unguarded or misdirected request: %+v", c)
		}
	}
}

// writeOnStop makes the instance write a file just before it is stopped:
// the stopped files differ from the archive the target verified, so the
// cut builds its own.
func (h *cutHarness) writeOnStop(t *testing.T) {
	h.ev.onCall = func() {
		if h.ev.calls == 1 {
			if err := os.WriteFile(filepath.Join(h.cfg.DataDir, p1Host, "late.txt"), []byte("written before the stop"), 0o644); err != nil {
				t.Error(err)
			}
		}
	}
}

func snapshotSealed(h *cutHarness) []string {
	var out []string
	for k := range h.f.buckets[h.srcBucket] {
		if strings.HasPrefix(k, moves.SnapshotsPrefix(p1Host, p1ID)) && strings.HasSuffix(k, ".sealed") {
			out = append(out, k)
		}
	}
	sort.Strings(out)
	return out
}

// codex 19, with the real randomized sealer: the archive's identity is
// journalled before any PUT, and every replay publishes that same key,
// whatever it crashed after: the journal entry, the first snapshot PUT,
// the second, the first copy. Once journalled the identity is final
// (fix round 1, ruling 4): a replay with the sealed file gone finishes it
// from what was published (the manifest from the journalled bytes if only
// the sealed object landed), and one that finds nothing published cannot,
// so the cut reverts rather than reseal under a new name.
func TestCutArchiveIsCreateOnlyAndReused(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"archive-journaled", "archive-sealed-snapshot", "archive-own-put", "archive-first-copy"} {
		t.Run("crash "+point, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.writeOnStop(t)
			h.crashAt = point
			h.cut(t)
			h.crashPass(t)
			if m := h.move(t); m.CutPhase != provision.CutStaged {
				t.Fatalf("after the crash: %+v", m)
			}
			id := h.journalArchive(t)
			if id.Reuse {
				t.Fatal("the stopped files changed, yet the preparing archive was reused")
			}
			h.freshWorker()
			h.pass(t)
			m := h.move(t)
			if m.State != provision.MoveLanding || m.Source.Files.Sealed != id.Sealed {
				t.Fatalf("the replay: %+v, files %+v, journalled %s", m, m.Source.Files, id.Sealed)
			}
			if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], id.Name+".sealed") {
				t.Fatalf("snapshots %v, want only %s", got, id.Name)
			}
			if !sealNamesArchive(h.seal(t), id.Sealed) {
				t.Fatalf("the seal names %+v, not %s", h.seal(t).Files, id.Name)
			}
		})
	}
	t.Run("file gone, nothing published", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.writeOnStop(t)
		h.crashAt = "archive-journaled"
		h.cut(t)
		h.crashPass(t)
		first := h.journalArchive(t)
		if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveReverting || m.Reason != provision.RevertArchiveRefused {
			t.Fatalf("an identity nobody can finish did not revert: %+v", m)
		}
		if again := h.journalArchive(t); again.Sealed != first.Sealed {
			t.Fatalf("the journalled identity changed from %s to %s", first.Name, again.Name)
		}
		if got := snapshotSealed(h); len(got) != 0 {
			t.Fatalf("snapshots %v, want none", got)
		}
		if h.seal(t).HasFiles() {
			t.Fatal("the seal names files")
		}
	})
	t.Run("file gone, sealed snapshot only", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.writeOnStop(t)
		h.crashAt = "archive-sealed-snapshot"
		h.cut(t)
		h.crashPass(t)
		first := h.journalArchive(t)
		snap := moves.SnapshotsPrefix(p1Host, p1ID)
		if _, ok := h.f.buckets["data-ie"][snap+first.Name+".sealed"]; !ok {
			t.Fatal("fixture: the sealed object did not land")
		}
		if _, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]; ok {
			t.Fatal("fixture: the manifest landed too")
		}
		if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
			t.Fatalf("the identity was not finished: %+v files %+v", m, m.Source.Files)
		}
		manifest, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]
		if !ok {
			t.Fatal("the manifest was not written from the journal")
		}
		var man archive.Manifest
		if err := json.Unmarshal([]byte(manifest.body), &man); err != nil || man.Sealed != first.Sealed || man.Plain != first.Plain {
			t.Fatalf("the written manifest: %v %+v", err, man)
		}
		copied := 0
		for _, c := range h.f.stagingCalls() {
			if c.Op == "copy" && strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)+first.Name) {
				copied++
			}
		}
		if copied != 2 || len(snapshotSealed(h)) != 1 {
			t.Fatalf("%d copies from the snapshot, %d sealed snapshots", copied, len(snapshotSealed(h)))
		}
	})
	t.Run("file gone, snapshot published", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.writeOnStop(t)
		h.crashAt = "archive-own-put"
		h.cut(t)
		h.crashPass(t)
		first := h.journalArchive(t)
		if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
			t.Fatalf("not reused: %+v files %+v", m, m.Source.Files)
		}
		// Staged by a server-side copy from the snapshot.
		copied := 0
		for _, c := range h.f.stagingCalls() {
			if c.Op == "copy" && strings.HasPrefix(c.Key, moves.FilesPrefix(p1Host, p1ID)+first.Name) {
				copied++
			}
		}
		if copied != 2 {
			t.Fatalf("%d copies from the snapshot, want 2", copied)
		}
	})
	t.Run("unchanged files reuse the verified archive", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		staged := h.ledger(t)
		h.cut(t)
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveLanding || m.Source.Files.Sealed != staged.ArchiveSealed {
			t.Fatalf("the verified archive was not reused: %+v (ledger %s)", m.Source.Files, staged.ArchiveSealed)
		}
		seal := h.seal(t)
		for _, o := range seal.Files {
			if o.VersionID != staged.ArchiveVersions.SealedVersion && o.VersionID != staged.ArchiveVersions.ManifestVersion {
				t.Fatalf("the seal names %s@%s, not the verified versions", o.Key, o.VersionID)
			}
		}
		if got := snapshotSealed(h); len(got) != 1 || !strings.HasSuffix(got[0], archiveName(staged.ArchiveSealed)+".sealed") {
			t.Fatalf("snapshots %v", got)
		}
	})
}

// G8: archived is written only after the seal names the archive. A crash
// after each of the archive's PUTs leaves the record staged; the replay
// completes and the seal names both files/ versions before archived. A
// record hand-edited to archived whose seal has no files does not flip.
func TestArchivedNeverPrecedesTheSealedArchive(t *testing.T) {
	t.Parallel()
	for _, point := range []string{"archive-own-put", "archive-first-copy", "archive-second-copy", "after-archive-seal"} {
		t.Run(point, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.writeOnStop(t)
			h.crashAt = point
			h.cut(t)
			h.crashPass(t)
			if m := h.move(t); m.CutPhase != provision.CutStaged {
				t.Fatalf("archived before the seal named the archive: %+v", m)
			}
			if point != "after-archive-seal" && h.seal(t).HasFiles() {
				t.Fatal("the seal names files before the copies finished")
			}
			h.pass(t)
			m := h.move(t)
			seal := h.seal(t)
			if m.State != provision.MoveLanding || !sealNamesArchive(seal, m.Source.Files.Sealed) {
				t.Fatalf("after the replay: %+v, seal files %+v", m, seal.Files)
			}
			for _, o := range seal.Files {
				if !strings.HasPrefix(o.Key, moves.FilesPrefix(p1Host, p1ID)) || h.f.versionOf("data-uk", o.Key) != o.VersionID {
					t.Fatalf("seal names %s@%s", o.Key, o.VersionID)
				}
			}
		})
	}
	t.Run("hand-edited to archived", func(t *testing.T) {
		ctx := context.Background()
		h := newCutHarness(t)
		h.ready(t)
		h.crashAt = "archive-journaled"
		h.cut(t)
		h.crashPass(t)
		if h.seal(t).HasFiles() {
			t.Fatal("fixture: the seal already names files")
		}
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, func(rec *provision.Record) error {
			rec.Move.CutPhase = provision.CutArchived
			return nil
		}); err != nil {
			t.Fatal(err)
		}
		h.pass(t)
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveCutting || m.CutPhase != provision.CutArchived || h.read(t, p1Host).Region != "ie" {
			t.Fatalf("a hand-edited archived flipped: %+v", m)
		}
		if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
			t.Fatalf("the claim flipped: %+v", c.Op)
		}
	})
}

// §7.5: the record flips first (region, landing, the landing deadline,
// checking), then the claim follows: landing/checking with the target
// executing, the source's stamp and generation journalled, the holder's
// generation untouched. A crash between the two leaves the record ahead
// and the claim moving/sealed with nobody executing, which anyone who sees
// it may finish with the same Flip.
func TestFlipWritesRecordThenClaim(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	h := newCutHarness(t)
	h.ready(t)
	h.cut(t)
	h.pass(t)
	rec := h.read(t, p1Host)
	m := rec.Move
	if rec.Region != "uk" || m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped || m.LandingPhase != provision.LandChecking || m.FlippedAt == nil || m.LandingDeadline == nil {
		t.Fatalf("record: region %s %+v", rec.Region, m)
	}
	if d := m.LandingDeadline.Sub(*m.FlippedAt); d != cutLandingWindow {
		t.Fatalf("landing window %v", d)
	}
	c := h.claim(t)
	if c.Op.Kind != hostclaim.KindLanding || c.Op.Phase != "checking" || c.Op.Exec != "uk-box" ||
		c.Op.Journal[hostclaim.JournalSourceStamp] != "1" || c.Op.Journal[hostclaim.JournalSrcGen] != "cur" || c.Holder.Generation != "cur" {
		t.Fatalf("claim: %+v holder %+v", c.Op, c.Holder)
	}

	h2 := newCutHarness(t)
	h2.ready(t)
	h2.crashAt = "after-record-flip"
	h2.cut(t)
	h2.crashPass(t)
	if r := h2.read(t, p1Host); r.Region != "uk" || r.Move.State != provision.MoveLanding {
		t.Fatalf("record not flipped: %+v", r.Move)
	}
	c2 := h2.claim(t)
	if c2.Op.Kind != hostclaim.KindMoving || c2.Op.Phase != hostclaim.PhaseSealed || c2.Op.Exec != "" {
		t.Fatalf("claim after the crash: %+v", c2.Op)
	}
	obs, err := hostclaim.Read(ctx, h2.s, p1Host)
	if err != nil {
		t.Fatal(err)
	}
	if _, err := hostclaim.Update(ctx, h2.s, obs, hostclaim.Flip(p1ID, "uk-box", obs.Claim.Holder.Stamp, obs.Claim.Holder.Generation)); err != nil {
		t.Fatalf("the recovery's flip: %v", err)
	}
	if c3 := h2.claim(t); c3.Op.Kind != hostclaim.KindLanding || c3.Op.Exec != "uk-box" {
		t.Fatalf("claim after the recovery: %+v", c3.Op)
	}
}

// The deadline (§7.4): at admitted it goes back to ready; at stopping it
// reverts with "stop evidence budget" and the claim stays with the source,
// which still executes; after execution was given up it reverts with the
// deadline's reason and the claim follows to reverting.
func TestCutDeadlinePassedReverts(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	g := provision.MoveGuard{ID: p1ID, Incarnation: h0inc(t)}

	t.Run("admitted", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.cut(t)
		now := h.clock.Now()
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveAdmit(g, now, 120, moves.CutDeadline(now, 120), 3)); err != nil {
			t.Fatal(err)
		}
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveReady || h.ev.calls != 0 {
			t.Fatalf("admitted past the deadline: %+v", m)
		}
	})
	t.Run("stopping", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.ev.stopped = false
		h.cut(t)
		h.pass(t)
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveReverting || m.Reason != provision.RevertStopEvidenceBudget {
			t.Fatalf("stopping past the deadline: %+v", m)
		}
		if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving || c.Op.Exec != "ie-box" {
			t.Fatalf("the claim moved without the source's evidence: %+v", c.Op)
		}
	})
	t.Run("stopped, execution given up", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.crashAt = "after-exec-none"
		h.cut(t)
		h.crashPass(t)
		h.clock.Advance(2 * time.Minute)
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveReverting || m.Reason != provision.RevertCutDeadline {
			t.Fatalf("stopped past the deadline: %+v", m)
		}
		c := h.claim(t)
		if c.Op.Kind != hostclaim.KindReverting || c.Op.Exec != "ie-box" {
			t.Fatalf("claim: %+v", c.Op)
		}
		j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
		if err != nil || j.Steps[cutStepRevertBound] == "" {
			t.Fatalf("the revert's bound is not journalled: %v %+v", err, j)
		}
	})
	t.Run("the claim half after a crash", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.crashAt = "after-exec-none"
		h.cut(t)
		h.crashPass(t)
		// The record decides the revert; the claim half is lost.
		if _, err := provision.SetRecordMove(ctx, h.s, "bab", "app", p1Host, provision.MoveRevert(g, provision.RevertCutDeadline)); err != nil {
			t.Fatal(err)
		}
		if c := h.claim(t); c.Op.Kind != hostclaim.KindMoving {
			t.Fatalf("fixture: %+v", c.Op)
		}
		h.pass(t)
		if c := h.claim(t); c.Op.Kind != hostclaim.KindReverting {
			t.Fatalf("the reverting arm did not write the claim half: %+v", c.Op)
		}
	})
}

// With the switch off, a cutting record is never touched.
func TestCutIsInertWithTheSwitchOff(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	h.cut(t)
	before := h.read(t, p1Host)
	claimBefore := h.claim(t)
	h.cfg.Enabled = false
	h.pass(t)
	h.pass(t)
	if after := h.read(t, p1Host); after.Move.CutPhase != before.Move.CutPhase || after.Move.UpdatedAt != before.Move.UpdatedAt {
		t.Fatalf("the record moved: %+v", after.Move)
	}
	if c := h.claim(t); c.Gen != claimBefore.Gen {
		t.Fatal("the claim was written")
	}
	if h.ev.calls != 0 || len(h.holds) != 0 || h.fences != 0 {
		t.Fatal("the cut ran with the switch off")
	}
}

func h0inc(t *testing.T) string {
	t.Helper()
	inc, err := hostclaim.Incarnation("bab", "app", p1Host, p1Created)
	if err != nil {
		t.Fatal(err)
	}
	return inc
}

// nullVersions is a staging writer whose copies of matching keys answer
// "null", as a versioning-suspended bucket does.
type nullVersions struct {
	lease.StagingWriter
	match func(key string) bool
}

func (n nullVersions) CopyVersionedAt(ctx context.Context, signAt time.Time, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, string, error) {
	etag, ver, err := n.StagingWriter.CopyVersionedAt(ctx, signAt, srcBucket, srcKey, srcETag, dstBucket, dstKey)
	if err == nil && n.match(dstKey) {
		ver = "null"
	}
	return etag, ver, err
}

// G27: the seal needs a usable version for every current replica key. A
// final round that copied one without (a "null") seals nothing and runs
// again; that is not a deferred round.
func TestSealRefusesAnObjectWithoutAUsableVersion(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	suspended := true
	h.cfg.Staging = func(context.Context, string) (lease.StagingWriter, error) {
		return nullVersions{StagingWriter: h.f, match: func(key string) bool { return suspended && strings.HasSuffix(key, lastTXKey) }}, nil
	}
	h.freshWorker() // the worker memoizes its staging writer
	h.cut(t)
	h.pass(t)
	if m := h.move(t); m.CutPhase != provision.CutStopped {
		t.Fatalf("sealed over a null version: %+v", m)
	}
	if _, ok := h.claim(t).Op.Journal[hostclaim.JournalSeal]; ok {
		t.Fatal("a seal was journalled")
	}
	if n := h.ledger(t).DeferredRounds; n != 0 {
		t.Fatalf("counted as %d deferred rounds", n)
	}
	suspended = false
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding {
		t.Fatalf("the next round did not seal: %+v", m)
	}
	for _, o := range h.seal(t).Replica {
		if !lease.UsableVersionID(o.VersionID) {
			t.Fatalf("sealed %s@%q", o.Key, o.VersionID)
		}
	}
}

// holdMoving holds the row moving under the op, once, with the
// stage-requested marker touched for a write that landed; a row already so
// held is left alone, and a row another op holds is refused.
func TestHoldMovingIsConditional(t *testing.T) {
	reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	defer reg.Close()
	if err := reg.Upsert(registry.Route{Host: p1Host, Kind: "instance", App: "app", Account: "bab", Status: "ready", Provisioned: true}); err != nil {
		t.Fatal(err)
	}
	var written []string
	hold := holdMoving(reg, func(host string) { written = append(written, host) })
	if err := hold(p1Host, p1ID); err != nil {
		t.Fatal(err)
	}
	if err := hold(p1Host, p1ID); err != nil {
		t.Fatalf("a second hold under the same op: %v", err)
	}
	rt, err := reg.Get(p1Host)
	if err != nil || rt.Status != "moving" || rt.HoldOp != p1ID || len(written) != 1 {
		t.Fatalf("row %+v, %v; marker touched %d times", rt, err, len(written))
	}
	if err := hold(p1Host, p1ID2); err == nil {
		t.Fatal("another op's hold was taken over")
	}
	if rt, _ := reg.Get(p1Host); rt.HoldOp != p1ID {
		t.Fatalf("the hold moved to %s", rt.HoldOp)
	}
	if err := hold("absent.example.com", p1ID); err == nil {
		t.Fatal("a missing row was held")
	}
}

// localLTXPosition reads the newest level-0 file of the activator's meta
// directory (lease.Held's SetMetaPath); no directory is position 0.
func TestLocalLTXPosition(t *testing.T) {
	dir := t.TempDir()
	db := filepath.Join(dir, p1Host+".db")
	if n, err := localLTXPosition(db); err != nil || n != 0 {
		t.Fatalf("no meta directory: %d, %v", n, err)
	}
	l0 := filepath.Join(dir, ".activator-"+p1Host+".db-litestream", "ltx", "0")
	if err := os.MkdirAll(l0, 0o755); err != nil {
		t.Fatal(err)
	}
	for _, name := range []string{"0000000000000001-0000000000000005.ltx", "0000000000000006-000000000000000a.ltx", "junk", "0000000000000003-0000000000000004.ltx"} {
		if err := os.WriteFile(filepath.Join(l0, name), nil, 0o600); err != nil {
			t.Fatal(err)
		}
	}
	if n, err := localLTXPosition(db); err != nil || n != 10 {
		t.Fatalf("position %d, %v; want 10", n, err)
	}
}

// Ruling 3: coverage is never assumed. A database that is gone before the
// final checkpoint leaves the seal FromDB false and SrcCovered false, since
// nothing proved the replica holds its last write; one that was there
// through the checkpoint, with its position journalled equal to N, keeps
// SrcCovered true if the disk is lost afterwards; one still present but
// behind its replica is not covered.
func TestMissingDatabaseIsNotCovered(t *testing.T) {
	t.Parallel()
	t.Run("lost before the checkpoint", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.crashAt = "after-evidence"
		h.cut(t)
		h.crashPass(t)
		if m := h.move(t); m.CutPhase != provision.CutStopping {
			t.Fatalf("after the crash: %+v", m)
		}
		if err := os.Remove(h.route.DBPath); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveLanding || h.checkpoints != 0 {
			t.Fatalf("the cut without a database: %+v (%d checkpoints)", m, h.checkpoints)
		}
		if m.Source.FinalTXID != 2 || m.Source.DBSHA256 != "" {
			t.Fatalf("source baselines: %+v", m.Source)
		}
		seal := h.seal(t)
		if seal.FromDB || seal.SrcCovered || seal.N != 2 {
			t.Fatalf("seal claims coverage with no database: %+v", seal)
		}
	})
	t.Run("lost after the checkpoint journalled coverage", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.crashAt = "after-exec-none"
		h.cut(t)
		h.crashPass(t)
		if err := os.Remove(h.route.DBPath); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveLanding {
			t.Fatalf("the replay: %+v", m)
		}
		seal := h.seal(t)
		if seal.FromDB || !seal.SrcCovered || seal.N != 3 {
			t.Fatalf("seal: %+v", seal)
		}
	})
	t.Run("present but behind its replica", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.localPos = 2
		h.cut(t)
		h.pass(t)
		if m := h.move(t); m.State != provision.MoveLanding {
			t.Fatalf("the cut: %+v", m)
		}
		if seal := h.seal(t); !seal.FromDB || seal.SrcCovered {
			t.Fatalf("seal: %+v", seal)
		}
	})
}

// regularFile: a regular file is one; an absent path is not and no error;
// anything else (a path under a file, a directory where a file should be)
// is an error, never absence.
func TestRegularFileDistinguishesAbsence(t *testing.T) {
	dir := t.TempDir()
	file := filepath.Join(dir, "a.db")
	if err := os.WriteFile(file, []byte("x"), 0o600); err != nil {
		t.Fatal(err)
	}
	if ok, err := regularFile(file); !ok || err != nil {
		t.Fatalf("a file: %v %v", ok, err)
	}
	if ok, err := regularFile(filepath.Join(dir, "absent.db")); ok || err != nil {
		t.Fatalf("absent: %v %v", ok, err)
	}
	if ok, err := regularFile(filepath.Join(file, "under-a-file.db")); ok || err == nil {
		t.Fatalf("a path under a file: %v %v", ok, err)
	}
	if ok, err := regularFile(dir); ok || err == nil {
		t.Fatalf("a directory: %v %v", ok, err)
	}
}

// Ruling 4: a listing error is an error, never absence. With the sealed
// file gone, a replay whose ListVersions fails changes nothing: the record
// stays staged, the journalled identity stands, nothing is rebuilt and
// nothing reverts. The next replay, listing again, finishes the identity
// when something was published, and reverts when nothing was.
func TestArchiveListingFailureIsNotAbsence(t *testing.T) {
	t.Parallel()
	for _, tc := range []struct {
		name, crash string
		then        string
	}{
		{"the sealed snapshot landed", "archive-sealed-snapshot", provision.MoveLanding},
		{"nothing landed", "archive-journaled", provision.MoveReverting},
	} {
		t.Run(tc.name, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.writeOnStop(t)
			h.crashAt = tc.crash
			h.cut(t)
			h.crashPass(t)
			first := h.journalArchive(t)
			if err := os.RemoveAll(archiveDir(h.cfg.DataDir, p1Host, p1ID)); err != nil {
				t.Fatal(err)
			}
			failed := 0
			h.vs.deny = func(op, _, key string) error {
				if op == "list-versions" && strings.Contains(key, moves.SnapshotsPrefix(p1Host, p1ID)) {
					failed++
					return errors.New("injected listing failure")
				}
				return nil
			}
			h.freshWorker()
			h.pass(t)
			if failed == 0 {
				t.Fatal("the replay never listed the snapshots")
			}
			m := h.move(t)
			if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
				t.Fatalf("after the listing failure: %+v", m)
			}
			if again := h.journalArchive(t); again.Sealed != first.Sealed {
				t.Fatalf("the listing failure changed the identity from %s to %s", first.Name, again.Name)
			}
			if h.seal(t).HasFiles() {
				t.Fatal("the seal names files")
			}
			h.vs.deny = nil
			h.freshWorker()
			h.pass(t)
			m = h.move(t)
			if m.State != tc.then {
				t.Fatalf("after the listing succeeded: %+v", m)
			}
			if tc.then == provision.MoveLanding && (m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed)) {
				t.Fatalf("not the journalled identity: %+v", m.Source.Files)
			}
		})
	}
}

// Ruling 5: the archive seal's bound goes into the ledger as well as the
// seal, and a ledger that cannot be read or written stops the phase. An
// injected ledger-read failure after the archive is sealed leaves the
// record staged and the ledger's bound as the stopped seal wrote it; the
// next pass finishes and the ledger carries the archive seal's bound.
func TestArchiveSealStopsOnALedgerFailure(t *testing.T) {
	h := newCutHarness(t)
	h.ready(t)
	ledgerKey := moves.LedgerKey(p1Host, p1ID)
	armed := false
	h.cfg.cutFault = func(point string) error {
		if point == "after-archive-seal" && !armed {
			armed = true
			h.f.failGet = func(_, key string) error {
				if key == ledgerKey {
					return errors.New("injected ledger read failure")
				}
				return nil
			}
		}
		return nil
	}
	h.cut(t)
	h.pass(t)
	if !armed {
		t.Fatal("the pass never sealed the archive")
	}
	m := h.move(t)
	if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
		t.Fatalf("the phase advanced over a ledger failure: %+v", m)
	}
	sealBound := h.seal(t).LeaseUntil
	h.f.failGet = nil
	// The ledger still carries the stopped seal's bound, written before the
	// archive: never later than the archive seal's.
	before := h.ledger(t)
	if before.LeaseUntil.IsZero() || before.LeaseUntil.After(sealBound) {
		t.Fatalf("ledger bound %v against the archive seal's %v", before.LeaseUntil, sealBound)
	}
	h.freshWorker()
	h.pass(t)
	if m := h.move(t); m.State != provision.MoveLanding {
		t.Fatalf("the replay: %+v", m)
	}
	after := h.ledger(t)
	if seal := h.seal(t); !after.LeaseUntil.Equal(seal.LeaseUntil) || after.LeaseUntil.Before(before.LeaseUntil) {
		t.Fatalf("ledger bound %v, seal bound %v (was %v)", after.LeaseUntil, seal.LeaseUntil, before.LeaseUntil)
	}
}

// Ruling 6: the crash matrix for the side effects the review found
// unexercised. Each point crashes once; the durable state a fresh worker
// would find is asserted; the fresh worker replays to the flip, with one
// report, a seal naming the archive at the record's N, and the ledger
// carrying the seal's bound.
func TestCutCrashMatrix(t *testing.T) {
	t.Parallel()
	ctx := context.Background()
	lane := moves.ReplicaPrefix(p1Host, p1ID)
	for _, point := range []string{"after-evidence", "after-final-round", "after-seal", "archive-sealed-snapshot"} {
		t.Run(point, func(t *testing.T) {
			h := newCutHarness(t)
			h.ready(t)
			h.writeOnStop(t) // so the archive is the cut's own build
			h.crashAt = point
			h.cut(t)
			h.crashPass(t)
			m := h.move(t)
			c := h.claim(t)
			j, err := moves.ReadJournal(h.cfg.DataDir, p1Host, p1ID)
			if err != nil {
				t.Fatal(err)
			}
			_, _, reportErr := h.s.Get(ctx, h.stoppedKey())
			_, sealed := c.Op.Journal[hostclaim.JournalSeal]
			switch point {
			case "after-evidence":
				// stopping: the evidence journalled, execution still the
				// source's, no report, no hold given up.
				if m.CutPhase != provision.CutStopping || !j.Done(cutStepEvidence) || c.Op.Exec != "ie-box" || !errors.Is(reportErr, os.ErrNotExist) {
					t.Fatalf("durable state: %+v, evidence %v, exec %q, report %v", m, j.Done(cutStepEvidence), c.Op.Exec, reportErr)
				}
			case "after-final-round":
				// stopped: execution given up, the report published, the
				// lane holding the final round's copy, and no seal yet.
				if m.CutPhase != provision.CutStopped || c.Op.Exec != "" || c.Op.Phase != hostclaim.PhaseMoving || reportErr != nil || sealed {
					t.Fatalf("durable state: %+v, claim %+v, report %v", m, c.Op, reportErr)
				}
				if _, ok := h.f.buckets["data-uk"][lane+lastTXKey]; !ok {
					t.Fatal("the final round's copy is not in the lane")
				}
			case "after-seal":
				// stopped: the claim sealed with a seal naming the replica and
				// the timers but no files; the ledger without a bound yet.
				if m.CutPhase != provision.CutStopped || c.Op.Phase != hostclaim.PhaseSealed || !sealed {
					t.Fatalf("durable state: %+v, claim %+v", m, c.Op)
				}
				if seal := h.seal(t); seal.HasFiles() || seal.N != 3 || len(seal.Timers) != 1 {
					t.Fatalf("seal: %+v", seal)
				}
				if l := h.ledger(t); !l.LeaseUntil.IsZero() {
					t.Fatalf("the ledger carries a bound (%v) before the seal's pass finished", l.LeaseUntil)
				}
			case "archive-sealed-snapshot":
				// staged: the identity journalled, the sealed object in
				// move-snapshots/ and the manifest not, the seal without files.
				id := h.journalArchive(t)
				snap := moves.SnapshotsPrefix(p1Host, p1ID)
				_, sealedThere := h.f.buckets["data-ie"][snap+id.Name+".sealed"]
				_, manifestThere := h.f.buckets["data-ie"][snap+id.Name+".json"]
				if m.CutPhase != provision.CutStaged || !sealedThere || manifestThere || h.seal(t).HasFiles() {
					t.Fatalf("durable state: %+v, sealed %v, manifest %v", m, sealedThere, manifestThere)
				}
			}
			h.freshWorker()
			h.pass(t)
			m = h.move(t)
			if m.State != provision.MoveLanding || m.CutPhase != provision.CutFlipped {
				t.Fatalf("the replay: %+v", m)
			}
			keys, _ := h.s.List(ctx, moves.ReportPrefix("bab", "app", p1Host, p1ID))
			n := 0
			for _, k := range keys {
				if strings.Contains(k, "stopped-") {
					n++
				}
			}
			seal := h.seal(t)
			if n != 1 || !sealNamesArchive(seal, m.Source.Files.Sealed) || seal.N != m.Source.FinalTXID || !seal.ReachesN() {
				t.Fatalf("%d reports; seal %+v; source %+v", n, seal, m.Source)
			}
			for _, o := range append(append(append([]moves.SealedObject{}, seal.Replica...), seal.Timers...), seal.Files...) {
				if h.f.versionOf("data-uk", o.Key) != o.VersionID {
					t.Fatalf("the seal names %s@%s, current %s", o.Key, o.VersionID, h.f.versionOf("data-uk", o.Key))
				}
			}
			if l := h.ledger(t); !l.LeaseUntil.Equal(seal.LeaseUntil) {
				t.Fatalf("ledger bound %v, seal bound %v", l.LeaseUntil, seal.LeaseUntil)
			}
			if c := h.claim(t); c.Op.Kind != hostclaim.KindLanding || c.Op.Exec != "uk-box" {
				t.Fatalf("claim: %+v", c.Op)
			}
		})
	}
}

// Fix round 2, item 1: a loss is declared only when it is established. The
// sealed file alone, with the journalled manifest, is enough to publish
// the identity: losing only the local manifest before the first PUT does
// not revert. A read of the sealed file that fails is an error, never a
// loss: the record stays staged, the identity stands, and once the file
// reads again the replay finishes it.
func TestArchiveLossIsDeclaredOnlyWhenEstablished(t *testing.T) {
	t.Parallel()
	t.Run("only the manifest is gone", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.writeOnStop(t)
		h.crashAt = "archive-journaled"
		h.cut(t)
		h.crashPass(t)
		first := h.journalArchive(t)
		if err := os.Remove(first.manifestPath()); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
			t.Fatalf("a kept sealed file with its manifest journalled: %+v files %+v", m, m.Source.Files)
		}
		snap := moves.SnapshotsPrefix(p1Host, p1ID)
		if _, ok := h.f.buckets["data-ie"][snap+first.Name+".json"]; !ok {
			t.Fatal("the manifest was not published from the journal")
		}
	})
	t.Run("the sealed file cannot be read", func(t *testing.T) {
		h := newCutHarness(t)
		h.ready(t)
		h.writeOnStop(t)
		h.crashAt = "archive-journaled"
		h.cut(t)
		h.crashPass(t)
		first := h.journalArchive(t)
		// The sealed path answers a read error (EISDIR), not absence.
		kept, err := os.ReadFile(first.sealedPath())
		if err != nil {
			t.Fatal(err)
		}
		if err := os.Remove(first.sealedPath()); err != nil {
			t.Fatal(err)
		}
		if err := os.Mkdir(first.sealedPath(), 0o700); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		m := h.move(t)
		if m.State != provision.MoveCutting || m.CutPhase != provision.CutStaged {
			t.Fatalf("a read error decided the archive: %+v", m)
		}
		if again := h.journalArchive(t); again.Sealed != first.Sealed {
			t.Fatalf("the identity changed from %s to %s", first.Name, again.Name)
		}
		if got := snapshotSealed(h); len(got) != 0 {
			t.Fatalf("snapshots %v, want none", got)
		}
		if err := os.Remove(first.sealedPath()); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(first.sealedPath(), kept, 0o600); err != nil {
			t.Fatal(err)
		}
		h.freshWorker()
		h.pass(t)
		m = h.move(t)
		if m.State != provision.MoveLanding || m.Source.Files.Sealed != first.Sealed || !sealNamesArchive(h.seal(t), first.Sealed) {
			t.Fatalf("once the file read again: %+v files %+v", m, m.Source.Files)
		}
	})
}

// Fix round 2, item 2: an automatic move on a scoped holder does not loop.
// decideReady never repins or requests a cut for a holder whose location is
// not the activator's prefix, so repeated passes write nothing: the record's
// stamp and the claim's generation stand.
func TestScopedAutomaticMoveDoesNotLoop(t *testing.T) {
	h := newCutHarness(t)
	h.scoped = true
	h.start(t, "auto")
	h.pass(t)
	h.targetPass(t, h.target())
	h.pass(t)
	rec := h.read(t, p1Host)
	if rec.Move.State == provision.MoveCutting {
		t.Fatalf("a cut was requested for a scoped holder: %+v", rec.Move)
	}
	claimGen := h.claim(t).Gen
	for i := 0; i < 3; i++ {
		h.pass(t)
		again := h.read(t, p1Host)
		if again.Move.State == provision.MoveCutting || again.Move.UpdatedAt != rec.Move.UpdatedAt || again.Move.State != rec.Move.State {
			t.Fatalf("pass %d wrote the record: %+v (was %+v)", i, again.Move, rec.Move)
		}
		if h.claim(t).Gen != claimGen {
			t.Fatalf("pass %d wrote the claim", i)
		}
	}
	if h.ev.calls != 0 || len(h.holds) != 0 {
		t.Fatal("something stopped")
	}
}