package host

import (
	"context"
	"errors"
	"fmt"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/hostwrite"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/moves"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
	"github.com/carlosframework/platform/internal/release"
	"github.com/carlosframework/platform/internal/seeds"
)

// fakeMoveMirror keeps S3's ETag rules (lease's fakeMirror explains them):
// a single-part object's ETag is its body, a multipart one's is
// body+"-27", and a copy lands single-part with the body as its ETag. An
// op log records ordering.
//
// A mirror round runs its copies and deletes side by side
// (lease.DefaultMirrorWorkers), so buckets and ops are guarded by bmu
// whenever a round can be running; a test reads them directly between
// rounds. beforeCopy runs without the lock, so it may call the fake. Lock
// order: mu, then bmu.
type fakeMoveMirror struct {
	bmu        sync.Mutex
	buckets    map[string]map[string]fakeMoveObject
	ops        []string
	beforeCopy func(srcKey string)

	// The Phase 1 staging writer's state (movesphase1_harness_test.go):
	// the guarded call log, a failure hook, and object versions.
	mu       sync.Mutex
	at       []stagingCall
	failAt   func(op, key string) error
	failGet  func(bucket, key string) error // a plain read that fails (the ledger, movescut_test.go)
	versions int
	vers     map[string]string
	// keepHistory makes the fake keep every version a later write
	// superseded, readable and copyable by its id, as a versioned bucket
	// does (off by default: the fake's older tests read the current
	// version only).
	keepHistory bool
	history     map[string]fakeMoveObject
}

type fakeMoveObject struct {
	lease.MirrorObject
	body string
}

func newFakeMoveMirror() *fakeMoveMirror {
	return &fakeMoveMirror{buckets: map[string]map[string]fakeMoveObject{}}
}
func (f *fakeMoveMirror) put(bucket, key, etag string) { f.putSized(bucket, key, etag, 0) }
func (f *fakeMoveMirror) putSized(bucket, key, etag string, size int64) {
	f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: etag, Size: size}, etag})
}
func (f *fakeMoveMirror) putMultipart(bucket, key, body string, size int64) {
	f.store(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: body + "-27", Size: size}, body})
}
func (f *fakeMoveMirror) store(bucket, key string, o fakeMoveObject) {
	f.bmu.Lock()
	defer f.bmu.Unlock()
	f.storeLocked(bucket, key, o)
}

func (f *fakeMoveMirror) storeLocked(bucket, key string, o fakeMoveObject) {
	if f.buckets[bucket] == nil {
		f.buckets[bucket] = map[string]fakeMoveObject{}
	}
	f.buckets[bucket][key] = o
}
func (f *fakeMoveMirror) List(_ context.Context, bucket, prefix string) (map[string]lease.MirrorObject, error) {
	f.bmu.Lock()
	defer f.bmu.Unlock()
	out := map[string]lease.MirrorObject{}
	for k, v := range f.buckets[bucket] {
		if strings.HasPrefix(k, prefix) {
			out[k] = v.MirrorObject
		}
	}
	return out, nil
}
func (f *fakeMoveMirror) Copy(_ context.Context, srcBucket, srcKey, srcETag, dstBucket, dstKey string) (string, error) {
	f.bmu.Lock()
	before := f.beforeCopy
	f.bmu.Unlock()
	if before != nil {
		before(srcKey)
	}
	f.bmu.Lock()
	defer f.bmu.Unlock()
	o, ok := f.buckets[srcBucket][srcKey]
	if !ok || (srcETag != "" && o.ETag != srcETag) {
		return "", lease.ErrCopySourceChanged
	}
	f.ops = append(f.ops, "copy "+dstBucket+"/"+dstKey)
	f.storeLocked(dstBucket, dstKey, fakeMoveObject{lease.MirrorObject{ETag: o.body, Size: o.Size}, o.body})
	return o.body, nil
}
func (f *fakeMoveMirror) Delete(_ context.Context, bucket, key string) error {
	f.bmu.Lock()
	defer f.bmu.Unlock()
	f.ops = append(f.ops, "delete "+bucket+"/"+key)
	delete(f.buckets[bucket], key)
	return nil
}
func (f *fakeMoveMirror) Get(_ context.Context, bucket, key string) ([]byte, error) {
	if f.failGet != nil {
		if err := f.failGet(bucket, key); err != nil {
			return nil, err
		}
	}
	f.bmu.Lock()
	defer f.bmu.Unlock()
	o, ok := f.buckets[bucket][key]
	if !ok {
		return nil, lease.ErrMirrorNotFound
	}
	return []byte(o.body), nil
}
func (f *fakeMoveMirror) Put(_ context.Context, bucket, key string, body []byte) error {
	f.bmu.Lock()
	defer f.bmu.Unlock()
	f.ops = append(f.ops, "put "+bucket+"/"+key)
	f.storeLocked(bucket, key, fakeMoveObject{lease.MirrorObject{ETag: "receipts", Size: int64(len(body))}, string(body)})
	return nil
}

// bodies is a prefix's content by relative key: what a byte-exact
// comparison compares (ETags differ across a multipart copy by design).
func (f *fakeMoveMirror) bodies(bucket, prefix string) map[string]string {
	f.bmu.Lock()
	defer f.bmu.Unlock()
	out := map[string]string{}
	for k, v := range f.buckets[bucket] {
		if strings.HasPrefix(k, prefix) {
			out[strings.TrimPrefix(k, prefix)] = v.body
		}
	}
	return out
}

type moveHarness struct {
	s        release.Store
	f        *fakeMoveMirror
	cfg      moveCfg
	statuses []string
	quiet    bool
	owns     bool
}

func newMoveHarness(t *testing.T) *moveHarness {
	t.Helper()
	h := &moveHarness{s: release.NewDirStore(t.TempDir()), f: newFakeMoveMirror(), quiet: true, owns: true}
	h.cfg = moveCfg{
		Region:          "ie",
		OwnBucket:       "data-ie",
		OwnBucketRegion: "eu-west-1",
		OwnPrefix:       "instances",
		DataDir:         t.TempDir(),
		Mirror: func(_ context.Context, dstRegion string) (lease.MirrorStore, error) {
			return h.f, nil
		},
		Quiesced:  func(string) bool { return h.quiet },
		SetStatus: func(host, status string) error { h.statuses = append(h.statuses, host+"="+status); return nil },
		OwnsRoute: func(host string) bool { return h.owns },
		Checkpoint: func(_ context.Context, dbPath string) error {
			h.f.ops = append(h.f.ops, "checkpoint "+filepath.Base(dbPath))
			return nil
		},
		Now:        time.Now,
		ReadyDelta: defaultMoveReadyDelta, // what armMovesPass sets with the env unset
	}
	// The target's edges file: uk publishes a data bucket.
	if _, err := h.s.Put(context.Background(), "control/edges/uk-box.json",
		[]byte(`{"v":1,"box":"uk-box","eip":"1.2.3.4","placement":"uk","data_bucket":"data-uk","data_bucket_region":"eu-west-2","hosts":{}}`), ""); err != nil {
		t.Fatal(err)
	}
	return h
}

// worker wraps the harness cfg the way armMovesPass does in production —
// a fresh worker per call, so memoized mirrors/nudges never leak between
// the table's cases.
func (h *moveHarness) worker() *movesWorker {
	return &movesWorker{cfg: h.cfg, mirrors: map[string]lease.MirrorStore{}}
}

func (h *moveHarness) record(t *testing.T, host string, mut ...func(*provision.Record) error) provision.Record {
	t.Helper()
	rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab",
		Region: "ie", Channel: "stable", CreatedAt: time.Now().UTC()}
	if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
		t.Fatal(err)
	}
	for _, m := range mut {
		if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, m); err != nil {
			t.Fatal(err)
		}
	}
	got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
	if err != nil {
		t.Fatal(err)
	}
	return *got
}

func (h *moveHarness) read(t *testing.T, host string) provision.Record {
	t.Helper()
	got, err := provision.GetRecord(context.Background(), h.s, "bab", "app", host)
	if err != nil {
		t.Fatal(err)
	}
	return *got
}

func TestMovesPassPrepareAndPromote(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", "instances/"+host+"/generations/g/snapshot", "e1", 64<<20)
	h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e2")

	// Round 1: copies happen; the SAME pass does not promote, because the
	// round copied more than the readiness bar (the 64 MiB snapshot): the
	// cut's own round could have that much again to do.
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Move.State != provision.MovePreparing {
		t.Fatalf("after the copying round: %s, want still preparing", got.Move.State)
	}
	staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
	if len(staged) != 2 {
		t.Fatalf("staging = %v", staged)
	}
	// Round 2: nothing to do → ready (zero bytes is under any bar).
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	got := h.read(t, host)
	if got.Move.State != provision.MoveReady || got.Move.ReadyAt == nil {
		t.Fatalf("after the quiet round: %+v, want ready", got.Move)
	}
	// Ready keeps mirroring: a late write flows on the next round.
	h.f.put("data-ie", "instances/"+host+"/generations/g/wal/001", "e3")
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	staged, _ = h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
	if len(staged) != 3 || h.read(t, host).Move.State != provision.MoveReady {
		t.Fatalf("late write did not flow: %v", staged)
	}
}

func TestMovesPassCutSequence(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	// A local warm cache that must be purged.
	db := filepath.Join(h.cfg.DataDir, host+".db")
	if err := writeFileForTest(db, "stale"); err != nil {
		t.Fatal(err)
	}

	// Not quiesced yet: the pass sets the status and stops there.
	h.quiet = false
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.statuses) != 1 || h.statuses[0] != host+"=moving" {
		t.Fatalf("statuses = %v", h.statuses)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
		t.Fatalf("cut advanced past an un-quiesced child: %+v", got)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
	}

	// Quiesced: the whole tail runs, in order — the final own-bucket
	// checkpoint strictly BEFORE the mirror round (the mirror of a stale
	// replica is faithful and useless).
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	ckIdx, cpIdx := -1, -1
	for i, op := range h.f.ops {
		if strings.HasPrefix(op, "checkpoint ") && ckIdx == -1 {
			ckIdx = i
		}
		if strings.HasPrefix(op, "copy ") && cpIdx == -1 {
			cpIdx = i
		}
	}
	if ckIdx == -1 || cpIdx == -1 || ckIdx > cpIdx {
		t.Fatalf("cut op order = %v, want checkpoint before the mirror copy", h.f.ops)
	}
	got := h.read(t, host)
	if got.Region != "uk" || got.Move != nil {
		t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
	}
	if fileExistsForTest(db) {
		t.Fatal("local DB survived the cut — the warm-cache trap")
	}
	staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
	if len(staged) != 1 {
		t.Fatalf("staging after final round = %v", staged)
	}
	// The pass is now inert for this record (region uk ≠ cfg ie).
	before := len(h.f.ops)
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != before {
		t.Fatal("a completed move kept mirroring")
	}
}

func TestMovesPassSkipsWithoutTargetCoords(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("se", time.Now())) // se publishes nothing
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Move.State != provision.MovePreparing {
		t.Fatalf("state advanced with no target coords: %+v", got.Move)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("mirrored toward an unknown target: %v", h.f.ops)
	}
}

// A cutting move for a host whose layout migration is in flight (the
// journal at provision.LayoutJournalPath) does nothing this round: its DB
// may be mid-copy between the flat and nested paths, so neither checkpoint
// nor the warm-cache purge is safe until the journal clears. It is the same
// fence Classify applies to a wake, applied here to a cut.
func TestMovesPassCutFencedByLayoutJournal(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	journal := provision.LayoutJournalPath(h.cfg.DataDir, host)
	if err := os.MkdirAll(filepath.Dir(journal), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := writeFileForTest(journal, "{}"); err != nil {
		t.Fatal(err)
	}
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("cut proceeded past the layout fence: %v", h.f.ops)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
		t.Fatalf("cut state advanced past the layout fence: %+v", got)
	}
}

// TestMovesPassCutFencedWhenTheJournalCannotBeRead pins the fail-closed
// fix: a journal directory os.Stat cannot even resolve (ENOTDIR here, a
// plain file sitting where "layout" should be a directory) is not proof
// there is no journal, and must fence the cut exactly as a readable
// journal does, not fail open and let it proceed.
func TestMovesPassCutFencedWhenTheJournalCannotBeRead(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	// A file where the journal's own parent directory should be: any stat
	// under it fails with ENOTDIR, never ENOENT.
	if err := writeFileForTest(filepath.Join(h.cfg.DataDir, "layout"), "not a directory"); err != nil {
		t.Fatal(err)
	}
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("cut proceeded past an unreadable layout fence: %v", h.f.ops)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
		t.Fatalf("cut state advanced past an unreadable layout fence: %+v", got)
	}
}

// TestMovesPassCutFencedByDanglingJournalSymlink pins the same fence for a
// dangling symlink at the journal path: os.Stat follows the link and reports
// ErrNotExist for a target that is gone, which would read as "no journal"
// even though something still sits at the path. Lstat sees the link itself
// and does not resolve it, so the cut must fence here exactly as it does for
// a regular journal file.
func TestMovesPassCutFencedByDanglingJournalSymlink(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	journal := provision.LayoutJournalPath(h.cfg.DataDir, host)
	if err := os.MkdirAll(filepath.Dir(journal), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(filepath.Join(filepath.Dir(journal), "gone.json"), journal); err != nil {
		t.Fatal(err)
	}
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("cut proceeded past a dangling journal symlink: %v", h.f.ops)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move.State != provision.MoveCutting {
		t.Fatalf("cut state advanced past a dangling journal symlink: %+v", got)
	}
}

// A cutting move for a nested row checkpoints the row's OWN DBPath, not the
// legacy <DataDir>/<host>.db guess: a nested row's DB lives under
// provision.InstanceDir and the flat guess names a file that was never
// written for it.
func TestMovesPassCutUsesRouteDBPath(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	nested := provision.NestedDBPath(h.cfg.DataDir, host)
	if err := os.MkdirAll(filepath.Dir(nested), 0o755); err != nil {
		t.Fatal(err)
	}
	if err := writeFileForTest(nested, "stale"); err != nil {
		t.Fatal(err)
	}
	h.cfg.Route = func(string) (*registry.Route, error) {
		return &registry.Route{Host: host, DBPath: nested}, nil
	}
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	found := false
	for _, op := range h.f.ops {
		if op == "checkpoint "+filepath.Base(nested) {
			found = true
		}
	}
	if !found {
		t.Fatalf("checkpoint did not use the route's own DBPath: %v", h.f.ops)
	}
	if fileExistsForTest(nested) {
		t.Fatal("the route's own DB survived the cut. The warm-cache trap did not follow DBPath")
	}
	if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
		t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
	}
}

// The cut fences the host against wakes before it trusts Quiesced (G12),
// holds the fence through the flip while this box still has the row, and
// lifts it once the row is retired.
func TestMovesPassCutFencesWakesUntilTheRouteRetires(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	rec := h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	fences := map[string]uint64{}
	h.cfg.Fence = func(_ context.Context, host string, gen uint64) error {
		fences[host] = gen
		return nil
	}
	h.cfg.LiftFence = func(host string, gen uint64) bool {
		if fences[host] != gen {
			return false
		}
		delete(fences, host)
		return true
	}
	routeHeld := true
	h.cfg.Route = func(string) (*registry.Route, error) {
		if !routeHeld {
			return nil, nil
		}
		return &registry.Route{Host: host, Status: "moving"}, nil
	}
	h.cfg.Quiesced = func(string) bool {
		if fences[host] == 0 {
			t.Fatal("Quiesced was asked before the wake fence was installed")
		}
		return h.quiet
	}
	w := h.worker()

	// Not quiesced: fenced, nothing else.
	h.quiet = false
	if err := movesPass(context.Background(), h.s, w); err != nil {
		t.Fatal(err)
	}
	want := uint64(rec.Move.RequestedAt.UnixNano())
	if fences[host] != want {
		t.Fatalf("fence generation = %d, want the move's request time %d", fences[host], want)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("mirror ran before quiesce: %v", h.f.ops)
	}

	// Quiesced: the cut completes and the fence stays, since the row is
	// still here waiting for reconcile to retire it.
	h.quiet = true
	if err := movesPass(context.Background(), h.s, w); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
		t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
	}
	if fences[host] != want {
		t.Fatal("the fence was lifted while this box still held the moving row")
	}

	// Reconcile retires the row: the next pass lifts the fence.
	routeHeld = false
	if err := movesPass(context.Background(), h.s, w); err != nil {
		t.Fatal(err)
	}
	if _, ok := fences[host]; ok {
		t.Fatal("the fence outlived the route")
	}
}

// A fence on a host whose record is at rest here again (a move back landed)
// is lifted even though this box holds the row: the row serves now.
func TestMovesPassLiftsTheFenceOfARecordAtRestHere(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host) // region ie, no move: at rest on this box
	lifted := uint64(0)
	h.cfg.LiftFence = func(_ string, gen uint64) bool { lifted = gen; return true }
	h.cfg.Route = func(string) (*registry.Route, error) { return &registry.Route{Host: host, Status: "ready"}, nil }
	w := h.worker()
	w.fenced.Store(host, uint64(42))
	if err := movesPass(context.Background(), h.s, w); err != nil {
		t.Fatal(err)
	}
	if lifted != 42 {
		t.Fatalf("lifted generation %d, want 42", lifted)
	}
	if _, ok := w.fenced.Load(host); ok {
		t.Fatal("the worker still remembers a lifted fence")
	}
}

// Passes run one at a time: the fast converge and the instances loop both
// call runMovesPass, and a pass that lifted fences from a record list read
// before another pass's cut would reopen wakes mid-cut.
func TestMovesPassRunsOneAtATime(t *testing.T) {
	t.Parallel()
	h := newMoveHarness(t)
	w := h.worker()
	w.pass.Lock()
	done := make(chan error, 1)
	go func() { done <- movesPass(context.Background(), h.s, w) }()
	select {
	case <-done:
		t.Fatal("a second pass ran while the first held the worker")
	case <-time.After(50 * time.Millisecond):
	}
	w.pass.Unlock()
	if err := <-done; err != nil {
		t.Fatal(err)
	}
}

// A fence that cannot be installed stops the cut before it looks at the
// child: nothing is mirrored, purged or flipped.
func TestMovesPassCutStopsWhenTheFenceFails(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")
	h.cfg.Fence = func(context.Context, string, uint64) error { return errors.New("fence not installed") }
	h.cfg.Quiesced = func(string) bool {
		t.Fatal("Quiesced was asked with no fence in place")
		return true
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("the cut ran without a fence: %v", h.f.ops)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move == nil || got.Move.State != provision.MoveCutting {
		t.Fatalf("after a failed fence: region=%q move=%+v, want still cutting in ie", got.Region, got.Move)
	}
}

// moveRound runs one pass and returns the record's move state after it.
func (h *moveHarness) moveRound(t *testing.T, host string) string {
	t.Helper()
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	got := h.read(t, host)
	if got.Move == nil {
		t.Fatalf("move cleared unexpectedly: %+v", got)
	}
	return got.Move.State
}

// An AWAKE instance never stops writing: litestream lands a new L0 file
// between every two listings, so no round ever copies nothing. Production
// (#175, kpmg.bdn.oncarlos.com) sat in preparing for 80 minutes that way.
// The bar is bytes, not zero: once the initial bulk has moved, a round
// that copied only the small new files is ready.
func TestMovesPassBusySourceReachesReady(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	src := "instances/" + host + "/generations/g/"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)
	h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200)

	// Round 1 moves the bulk: well over the bar, still preparing.
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("after the bulk round: %s, want preparing", st)
	}
	// Every later round finds a fresh L0 file and a lease renewal: never a
	// zero-change round, always a small one.
	st := ""
	for i := 1; i <= 3 && st != provision.MoveReady; i++ {
		h.f.putSized("data-ie", src+fmt.Sprint("ltx/0/", i), fmt.Sprint("l0-", i), 8<<10)
		h.f.putSized("data-ie", "instances/"+host+"/lock.json", fmt.Sprint("renewal-", i), 200)
		st = h.moveRound(t, host)
	}
	if st != provision.MoveReady {
		t.Fatalf("a busy source with small deltas never reached ready (state %s)", st)
	}
}

// A round that copied more than the bar does not promote, however many
// rounds it takes: compaction rewriting a 138 MiB level file every round
// (the #175 shape) keeps the move preparing. The bar is inclusive.
func TestMovesPassLargeDeltaDoesNotPromote(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	src := "instances/" + host + "/generations/g/"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", src+"snapshots/0", "s0", 293<<20)
	for i := 0; i < 4; i++ {
		// Compaction rewrites the level file under a new ETag each round.
		h.f.putSized("data-ie", src+"ltx/1/compacted", fmt.Sprint("c", i), 138<<20)
		if st := h.moveRound(t, host); st != provision.MovePreparing {
			t.Fatalf("round %d copied 138 MiB and promoted to %s", i, st)
		}
	}

	// The boundary, with a small bar: one byte over stays preparing, the
	// bar exactly promotes.
	h.cfg.ReadyDelta = 1000
	h.f.putSized("data-ie", src+"ltx/0/a", "a", 1001)
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("a 1001-byte round under a 1000-byte bar promoted to %s", st)
	}
	h.f.putSized("data-ie", src+"ltx/0/b", "b", 1000)
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("a 1000-byte round under a 1000-byte bar left the move %s", st)
	}
}

// A zero bar means zero bytes moved: a round that copied (or deferred)
// even one byte does not promote. A deletion-only round moves no bytes, so
// it does, which the old Changed == 0 rule did not allow.
func TestMovesPassZeroBarNeedsZeroCopiedBytes(t *testing.T) {
	h := newMoveHarness(t)
	h.cfg.ReadyDelta = 0
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", "instances/"+host+"/generations/g/wal/000", "e1", 1)
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("a 1-byte round under a zero bar promoted to %s", st)
	}
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("a quiet round under a zero bar left the move %s", st)
	}

	// Deletion-only: retention pruned a staged object, nothing to copy.
	other := "b.go.example.com"
	h.record(t, other, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/000", "e1", 1)
	h.moveRound(t, other)
	delete(h.f.buckets["data-ie"], "instances/"+other+"/generations/g/wal/000")
	h.f.putSized("data-ie", "instances/"+other+"/generations/g/wal/001", "e2", 0)
	if st := h.moveRound(t, other); st != provision.MoveReady {
		t.Fatalf("a zero-byte round (one delete, one empty copy) under a zero bar left the move %s", st)
	}
}

// Ready no longer means "staging matched the source". It means the gap was
// small. The cut's final round still has to make staging exact: what was
// written after the last ready round, what compaction rewrote, and what
// retention pruned all reach the target, and the lease never does.
func TestMovesPassCutAfterBusyReadyLandsByteExact(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	own := "instances/" + host + "/"
	gen := own + "generations/g/"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.putSized("data-ie", gen+"snapshots/0", "s0", 293<<20)
	h.f.putSized("data-ie", gen+"ltx/1/compacted", "c0", 138<<20)
	h.f.putSized("data-ie", gen+"ltx/0/0", "l0-0", 4<<10)
	h.f.putSized("data-ie", own+"lock.json", "renewal-0", 200)
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("after the bulk round: %s", st)
	}
	h.f.putSized("data-ie", gen+"ltx/0/1", "l0-1", 4<<10)
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("a 4 KiB round left the move %s, want ready", st)
	}
	staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
	if _, ok := staged["moves/"+host+"/generations/g/ltx/0/1"]; !ok {
		t.Fatalf("the ready round's own delta was not staged: %v", staged)
	}

	// After ready and before the cut, the instance keeps writing: a new L0
	// file, a compaction rewrite, retention pruning an old L0 file, and a
	// lease renewal.
	h.f.putSized("data-ie", gen+"ltx/0/2", "l0-2", 4<<10)
	h.f.putSized("data-ie", gen+"ltx/1/compacted", "c1", 139<<20)
	delete(h.f.buckets["data-ie"], gen+"ltx/0/0")
	h.f.putSized("data-ie", own+"lock.json", "renewal-1", 200)
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil {
		t.Fatal(err)
	}
	h.quiet = true
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
		t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
	}

	assertStagedExact(t, h, host)
}

// assertStagedExact compares CONTENT, source against staging, lease
// excluded: a multipart source and its copy carry different ETags by
// design, so metadata equality would prove nothing.
func assertStagedExact(t *testing.T, h *moveHarness, host string) {
	t.Helper()
	want := h.f.bodies("data-ie", "instances/"+host+"/")
	delete(want, "lock.json")
	got := h.f.bodies("data-uk", "moves/"+host+"/")
	if len(got) != len(want) {
		t.Fatalf("staged %d objects, want %d:\nstaged %v\nwant   %v", len(got), len(want), got, want)
	}
	for k, v := range want {
		if got[k] != v {
			t.Fatalf("staged %s = %q, want %q (the cut's final round must make staging exact)", k, got[k], v)
		}
	}
}

// kpmgShape stocks a source like kpmg.bdn.oncarlos.com's (#175): a
// multipart snapshot and six multipart 138 MB level files, all of which
// land in staging with a DIFFERENT ETag than the source lists.
func kpmgShape(h *moveHarness, host string) {
	gen := "instances/" + host + "/ltx/"
	h.f.putMultipart("data-ie", gen+"9/snapshot.ltx", "snap", 293<<20)
	for i := 0; i < 6; i++ {
		h.f.putMultipart("data-ie", fmt.Sprint(gen, "1/", i, ".ltx"), fmt.Sprint("level-", i), 138<<20)
	}
	h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-0", 200)
}

// The production failure (#175): multipart sources never compare equal to
// their copies, so before receipts every round recopied ~1 GB and the
// move could never reach ready, busy or not. With receipts, a busy
// instance whose rounds add only small L0 files reaches ready on the
// first round after the bulk.
func TestMovesPassBusyMultipartInstanceReachesReady(t *testing.T) {
	h := newMoveHarness(t)
	host := "kpmg.bdn.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	kpmgShape(h, host)
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("after the bulk round: %s, want preparing", st)
	}
	h.f.putSized("data-ie", "instances/"+host+"/ltx/0/1.ltx", "l0-1", 8<<10)
	h.f.putSized("data-ie", "instances/"+host+"/lock.json", "renewal-1", 200)
	before := len(h.f.ops)
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("a busy multipart instance with an 8 KiB delta left the move %s", st)
	}
	for _, op := range h.f.ops[before:] {
		if strings.Contains(op, "/ltx/1/") || strings.Contains(op, "/ltx/9/") {
			t.Fatalf("an unchanged multipart object was recopied: %s (ops %v)", op, h.f.ops[before:])
		}
	}
	// The receipts are in the SOURCE's bucket, never in staging.
	if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok {
		t.Fatal("no receipts in the source's bucket")
	}
	for k := range h.f.buckets["data-uk"] {
		if strings.Contains(k, "receipt") {
			t.Fatalf("receipts staged as data: %s", k)
		}
	}
}

// A source object overwritten between the listing and the copy defers
// that object. Its listed bytes still count against the bar, so a round
// that deferred a large object does not promote.
func TestMovesPassDeferredBytesCountAgainstTheBar(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	kpmgShape(h, host)
	h.moveRound(t, host)
	big := "instances/" + host + "/ltx/1/0.ltx"
	h.f.putMultipart("data-ie", big, "level-0b", 138<<20)
	var once sync.Once
	h.f.beforeCopy = func(k string) {
		if k == big {
			once.Do(func() { h.f.putMultipart("data-ie", big, "level-0c", 138<<20) })
		}
	}
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("a round that deferred 138 MiB promoted to %s", st)
	}
	if st := h.moveRound(t, host); st != provision.MovePreparing {
		t.Fatalf("the round that copied the deferred 138 MiB promoted to %s", st)
	}
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("the quiet round after left the move %s", st)
	}
}

// The cut's final round with multipart sources lands byte-exact, and the
// move's receipts are cleared once the record has flipped.
func TestMovesPassCutWithMultipartSourcesLandsExact(t *testing.T) {
	h := newMoveHarness(t)
	host := "kpmg.bdn.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	kpmgShape(h, host)
	h.moveRound(t, host)
	if st := h.moveRound(t, host); st != provision.MoveReady {
		t.Fatalf("state %s, want ready", st)
	}
	// Compaction after ready: one level file rewritten, one pruned.
	h.f.putMultipart("data-ie", "instances/"+host+"/ltx/1/2.ltx", "level-2b", 140<<20)
	delete(h.f.buckets["data-ie"], "instances/"+host+"/ltx/1/5.ltx")
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveMarkCutting()); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
		t.Fatalf("after cut: region=%q move=%+v", got.Region, got.Move)
	}
	assertStagedExact(t, h, host)
	if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok {
		t.Fatal("the move's receipts outlived the cut")
	}
}

// A source that changes under the cut's final round (it must not, the
// child is down, but nothing else proves it) leaves staging short of
// exact. The cut stops before the purge and the flip, and the next pass
// finishes it exactly.
func TestMovesPassCutStopsOnADeferredCopy(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MovePromoteReady(time.Now()), provision.MoveMarkCutting())
	kpmgShape(h, host)
	db := filepath.Join(h.cfg.DataDir, host+".db")
	if err := writeFileForTest(db, "warm"); err != nil {
		t.Fatal(err)
	}
	big := "instances/" + host + "/ltx/1/3.ltx"
	var once sync.Once
	h.f.beforeCopy = func(k string) {
		if k == big {
			once.Do(func() { h.f.putMultipart("data-ie", big, "level-3b", 138<<20) })
		}
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Region != "ie" || got.Move == nil || got.Move.State != provision.MoveCutting {
		t.Fatalf("the cut completed over a deferred copy: region=%q move=%+v", got.Region, got.Move)
	}
	if !fileExistsForTest(db) {
		t.Fatal("the local DB was purged before staging was exact")
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Region != "uk" || got.Move != nil {
		t.Fatalf("the retried cut: region=%q move=%+v", got.Region, got.Move)
	}
	assertStagedExact(t, h, host)
}

func TestMoveReadyDeltaFromEnv(t *testing.T) {
	for _, tc := range []struct {
		raw  string
		want int64
		ok   bool
	}{
		{"", defaultMoveReadyDelta, true}, // unset: the default, nothing to report
		{"0", 0, true},                    // zero copied bytes: only a round that moved no data promotes
		{"1048576", 1 << 20, true},
		{" 67108864 ", 64 << 20, true},
		{"-1", defaultMoveReadyDelta, false},
		{"16MiB", defaultMoveReadyDelta, false},
		{"1.5", defaultMoveReadyDelta, false},
		{"99999999999999999999", defaultMoveReadyDelta, false},
	} {
		got, ok := moveReadyDeltaFromEnv(func(k string) string {
			if k != "CARLOS_MOVE_READY_DELTA_BYTES" {
				t.Fatalf("read %s, want CARLOS_MOVE_READY_DELTA_BYTES", k)
			}
			return tc.raw
		})
		if got != tc.want || ok != tc.ok {
			t.Errorf("%q: got %d, %v; want %d, %v", tc.raw, got, ok, tc.want, tc.ok)
		}
	}
	if defaultMoveReadyDelta != 16<<20 {
		t.Fatalf("default bar = %d, want 16 MiB", defaultMoveReadyDelta)
	}
}

// armMovesPass is where the bar reaches the worker: the env override when
// it parses, the default when it does not.
func TestArmMovesPassReadsTheReadyDelta(t *testing.T) {
	prev := armedMovesWorker.Load()
	t.Cleanup(func() { armedMovesWorker.Store(prev) })
	for _, tc := range []struct {
		raw  string
		want int64
	}{
		{"", defaultMoveReadyDelta},
		{"4096", 4096},
		{"-5", defaultMoveReadyDelta},
		{"lots", defaultMoveReadyDelta},
	} {
		t.Setenv("CARLOS_MOVE_READY_DELTA_BYTES", tc.raw)
		armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)
		if got := armedMovesWorker.Load().cfg.ReadyDelta; got != tc.want {
			t.Errorf("CARLOS_MOVE_READY_DELTA_BYTES=%q armed a bar of %d, want %d", tc.raw, got, tc.want)
		}
	}
}

// armMovesPass reads the Phase 1 switch into the worker's config: off when
// the variable is unset or not an on-value, on for "on".
func TestArmMovesPassReadsTheSwitch(t *testing.T) {
	prev := armedMovesWorker.Load()
	t.Cleanup(func() { armedMovesWorker.Store(prev) })
	for _, tc := range []struct {
		raw  string
		want bool
	}{
		{"", false}, {"on", true}, {"1", true}, {"true", true}, {"off", false}, {"yes", false},
	} {
		t.Setenv(moves.SwitchEnv, tc.raw)
		armMovesPass(activator.Config{}, t.TempDir(), nil, nil, nil, hostwrite.Config{}, ClaimProviders{}, nil, nil)
		if got := armedMovesWorker.Load().cfg.Enabled; got != tc.want {
			t.Errorf("%s=%q armed Enabled=%v, want %v", moves.SwitchEnv, tc.raw, got, tc.want)
		}
	}
}

func writeFileForTest(path, content string) error { return os.WriteFile(path, []byte(content), 0o600) }
func fileExistsForTest(path string) bool          { _, err := os.Stat(path); return err == nil }

// The rails and the ownership gate (review round 1): pooled, non-hibernate,
// non-instance, and not-my-route records are skipped untouched.
func TestMovesPassRailsAndOwnership(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	h.f.put("data-ie", "instances/"+host+"/generations/g/snapshot", "e1")

	// Not this box's route (a standby sharing the region): untouched.
	h.owns = false
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if len(h.f.ops) != 0 || h.read(t, host).Move.State != provision.MovePreparing {
		t.Fatalf("a non-owning box acted: ops=%v", h.f.ops)
	}
	h.owns = true

	// A pooled sibling with a move field: skipped (the platform box holds
	// none of its data — a round here would promote ready over nothing).
	pooled := "p.go.example.com"
	rec := provision.Record{Schema: 1, Host: pooled, Slug: "p", App: "app", Account: "bab",
		Region: "ie", Pool: "dedicated-x", Channel: "stable", CreatedAt: time.Now().UTC()}
	if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
		t.Fatal(err)
	}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", pooled, provision.MoveRequest("uk", time.Now())); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if h.read(t, pooled).Move.State != provision.MovePreparing {
		t.Fatal("a pooled record's move advanced on the platform box")
	}
}

// Abort's box half: staging deleted, then the field cleared.
func TestMovesPassAbortPhase(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()), provision.MoveAbort())
	h.f.put("data-uk", "moves/"+host+"/generations/g/snapshot", "e1")
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/")
	if len(staged) != 0 {
		t.Fatalf("staging survived the abort: %v", staged)
	}
	if got := h.read(t, host); got.Move != nil || got.Region != "ie" {
		t.Fatalf("after abort phase: %+v region=%q", got.Move, got.Region)
	}
}

// Abort clears the move's receipts along with the staged copy.
func TestMovesPassAbortClearsReceipts(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	h.record(t, host, provision.MoveRequest("uk", time.Now()))
	kpmgShape(h, host)
	h.moveRound(t, host)
	if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; !ok {
		t.Fatal("no receipts after the bulk round")
	}
	if _, err := provision.SetRecordMove(context.Background(), h.s, "bab", "app", host, provision.MoveAbort()); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if got := h.read(t, host); got.Move != nil {
		t.Fatalf("abort did not complete: %+v", got.Move)
	}
	if staged := h.f.bodies("data-uk", "moves/"+host+"/"); len(staged) != 0 {
		t.Fatalf("staging survived the abort: %v", staged)
	}
	if _, ok := h.f.buckets["data-ie"][activator.MoveReceiptsKey(host)]; ok {
		t.Fatal("the receipts survived the abort")
	}
}

func TestMovesPassSkipsAnUnplacedSeed(t *testing.T) {
	logs := captureLog(t)
	h := newMoveHarness(t)
	host := "a.go.example.com"
	id, _ := seeds.NewID()
	rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",
		CreatedAt: time.Now().UTC(),
		Move:      &provision.MoveState{To: "uk", State: provision.MovePreparing, RequestedAt: time.Now().UTC()},
		Seed:      &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: time.Now().UTC()}}
	if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
		t.Fatal(err)
	}
	h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e1")
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 0 {
		t.Fatalf("the worker mirrored an unplaced seed: %v", staged)
	}
	if got := h.read(t, host); got.Move.State != provision.MovePreparing {
		t.Fatalf("state = %s", got.Move.State)
	}
	if len(h.f.ops) != 0 {
		t.Fatalf("a skipped round wrote: %v", h.f.ops)
	}
	if want := "moves: " + host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {
		t.Fatalf("log %q lacks %q", logs(), want)
	}
	// Placed, the same record moves on the next pass.
	if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", host, func(r *provision.Record) error {
		r.Seed.State = seeds.StatePlaced
		return nil
	}); err != nil {
		t.Fatal(err)
	}
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {
		t.Fatalf("a placed seed did not mirror: %v", staged)
	}
}

// The skip is for preparing only: a legacy move an old console already
// took past preparing has had its effects, and the worker carries it on.
func TestMovesPassCarriesOnAReadyMoveWithAnUnplacedSeed(t *testing.T) {
	h := newMoveHarness(t)
	host := "a.go.example.com"
	id, _ := seeds.NewID()
	now := time.Now().UTC()
	rec := provision.Record{Schema: 1, Host: host, Slug: "s", App: "app", Account: "bab", Region: "ie", Channel: "stable",
		CreatedAt: now,
		Move:      &provision.MoveState{To: "uk", State: provision.MoveReady, RequestedAt: now, ReadyAt: &now},
		Seed:      &provision.SeedSource{ID: id, Kind: seeds.KindReplica, State: seeds.StateReady, By: "t", At: now}}
	if err := provision.CreateRecord(context.Background(), h.s, rec); err != nil {
		t.Fatal(err)
	}
	h.f.put("data-ie", "instances/"+host+"/generations/g/wal/000", "e1")
	if err := movesPass(context.Background(), h.s, h.worker()); err != nil {
		t.Fatal(err)
	}
	if staged, _ := h.f.List(context.Background(), "data-uk", "moves/"+host+"/"); len(staged) != 1 {
		t.Fatalf("a ready move stopped mirroring: %v", staged)
	}
}

// seedTheMove gives the Phase 1 harness's record a seed after its move
// started: a record that slipped past an old console, whose doors did not
// refuse it (spec §5.8). The seed writer is the only writer of Seed.
func seedTheMove(t *testing.T, h *phase1Harness, kind, state string) {
	t.Helper()
	id, err := seeds.NewID()
	if err != nil {
		t.Fatal(err)
	}
	if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {
		r.Seed = &provision.SeedSource{ID: id, Kind: kind, State: state, By: "t", At: h.clock.Now()}
		return nil
	}); err != nil {
		t.Fatal(err)
	}
}

func setSeedState(t *testing.T, h *phase1Harness, state string) {
	t.Helper()
	if _, err := provision.SetRecordSeed(context.Background(), h.s, "bab", "app", p1Host, func(r *provision.Record) error {
		r.Seed.State = state
		return nil
	}); err != nil {
		t.Fatal(err)
	}
}

// The Phase 1 source round refuses an uncommitted import before its first
// effect: no staging lease, no claim touch, no staging write, no source
// report. Committed, the same move prepares.
func TestPhase1SourceSkipsAnUncommittedImport(t *testing.T) {
	logs := captureLog(t)
	h := newPhase1Harness(t)
	h.start(t, "manual")
	seedTheMove(t, h, seeds.KindImport, seeds.StateSwapped)
	before := h.claim(t)
	h.pass(t)
	if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
		t.Fatalf("state = %s, want preparing", got)
	}
	if h.lease.acquired != 0 {
		t.Fatalf("the skipped round took the staging lease %d times", h.lease.acquired)
	}
	if calls := h.f.stagingCalls(); len(calls) != 0 {
		t.Fatalf("the skipped round staged: %+v", calls)
	}
	if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); len(keys) != 0 {
		t.Fatalf("the lane holds %v", keys)
	}
	if sr := h.sourceReport(t); sr != nil {
		t.Fatalf("the skipped round published a source report: %+v", sr)
	}
	if after := h.claim(t); fmt.Sprint(after.Touched) != fmt.Sprint(before.Touched) {
		t.Fatalf("the skipped round touched the claim: %+v, was %+v", after.Touched, before.Touched)
	}
	if len(h.probePaths) != 0 {
		t.Fatalf("the skipped round probed the instance: %v", h.probePaths)
	}
	if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUncommitted); !strings.Contains(logs(), want) {
		t.Fatalf("log %q lacks %q", logs(), want)
	}

	setSeedState(t, h, seeds.StateCommitted)
	h.pass(t)
	if keys := stagedKeys(h, moves.StagingPrefix(p1Host, p1ID)); !containsString(keys, "move.json") {
		t.Fatalf("a committed import did not prepare: %v", keys)
	}
	if h.sourceReport(t) == nil {
		t.Fatal("a committed import published no source report")
	}
}

// The target box writes no report for a move whose seed is unplaced.
// Placed, it does.
func TestPhase1TargetSkipsAnUnplacedSeed(t *testing.T) {
	logs := captureLog(t)
	h := newPhase1Harness(t)
	h.start(t, "manual")
	seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)
	h.targetPass(t, h.targetCfg())
	if tr := h.targetReport(t); tr != nil {
		t.Fatalf("the target reported on an unplaced seed: %+v", tr)
	}
	if calls := h.f.stagingCalls(); len(calls) != 0 {
		t.Fatalf("the target's skipped round staged: %+v", calls)
	}
	if got := h.read(t, p1Host).Move.State; got != provision.MovePreparing {
		t.Fatalf("state = %s, want preparing", got)
	}
	if want := "moves: " + p1Host + ": " + provision.SeedMoveSentence(provision.SeedMoveUnplaced); !strings.Contains(logs(), want) {
		t.Fatalf("log %q lacks %q", logs(), want)
	}

	setSeedState(t, h, seeds.StatePlaced)
	h.targetPass(t, h.targetCfg())
	if h.targetReport(t) == nil {
		t.Fatal("the target wrote no report for a placed seed")
	}
}

// A box that is neither the move's source nor its target says nothing
// about the seed: the log line belongs to the boxes the move waits on.
func TestMovesPassSeedNoticeOnlyOnParticipants(t *testing.T) {
	logs := captureLog(t)
	h := newPhase1Harness(t)
	h.start(t, "manual")
	seedTheMove(t, h, seeds.KindReplica, seeds.StateReady)
	c := h.targetCfg()
	c.Region, c.Box = "us", "us-box"
	h.targetPass(t, c)
	if strings.Contains(logs(), provision.SeedMoveSentence(provision.SeedMoveUnplaced)) {
		t.Fatalf("a bystander logged the seed: %q", logs())
	}
}