//go:build linux

package layout

import (
	"context"
	"errors"
	"net/http"
	"net/http/httptest"
	"os"
	"path/filepath"
	"reflect"
	"strings"
	"sync"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/activator"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/registry"
)

const (
	host    = "h.example.com"
	replica = "r.example.com"
)

// fakeFence is the activator's fence as the pass sees it. A fence already
// in fences, of any generation, refuses a new one with ErrFenceHeld, as the
// real exclusive install does. Otherwise every stop is proven; unproven,
// awake and uncertain make it answer ErrStopUnproven and shape what
// Quiesced and StopUncertain say after it. events records every fence,
// refusal and lift, and box's Reload adds "reload", so a test can read the
// order the pass did things in.
type fakeFence struct {
	fences    map[string]uint64
	unproven  map[string]bool
	awake     map[string]bool
	uncertain map[string]bool
	onFence   func(host string)
	events    []string
}

func (f *fakeFence) FenceAndStopIfUnfenced(_ context.Context, h string, gen uint64) (*activator.StopEvidence, error) {
	if f.fences == nil {
		f.fences = map[string]uint64{}
	}
	if f.fences[h] != 0 {
		f.events = append(f.events, "held "+h)
		return nil, activator.ErrFenceHeld
	}
	f.fences[h] = gen
	f.events = append(f.events, "fence "+h)
	if f.onFence != nil {
		f.onFence(h)
	}
	if f.unproven[h] {
		return nil, activator.ErrStopUnproven
	}
	return &activator.StopEvidence{}, nil
}

func (f *fakeFence) LiftFence(h string, gen uint64) bool {
	if f.fences[h] != gen {
		return false
	}
	delete(f.fences, h)
	f.events = append(f.events, "lift "+h)
	return true
}

func (f *fakeFence) FenceGeneration(h string) uint64 { return f.fences[h] }
func (f *fakeFence) Quiesced(h string) bool          { return !f.awake[h] }
func (f *fakeFence) StopUncertain(h string) bool     { return f.uncertain[h] }

func openLayoutRegistry(t *testing.T, dir string) *registry.Registry {
	t.Helper()
	reg, err := registry.Open(filepath.Join(dir, "registry.db"))
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { reg.Close() })
	return reg
}

// seedFlat writes host's flat files: the database and its wal, the admin
// env, a litestream shadow directory with a file in it, and (for the
// primary host only) a quarantine. The sidecar env and the rendered bundle
// are absent, so the journal records them absent.
func seedFlat(t *testing.T, dd, h string) {
	t.Helper()
	db := provision.DefaultDBPath(dd, h)
	files := map[string]string{
		db:                          "db " + h,
		db + "-wal":                 "wal",
		filepath.Join(dd, h+".env"): "CARLOS_ADMIN_TOKEN=t\n",
		filepath.Join(dd, shadowName(h), "generation"): "g",
	}
	if h == host {
		files[db+".corrupt"] = "quarantined"
	}
	for path, body := range files {
		if err := os.MkdirAll(filepath.Dir(path), 0o700); err != nil {
			t.Fatal(err)
		}
		if err := os.WriteFile(path, []byte(body), 0o600); err != nil {
			t.Fatal(err)
		}
	}
}

// box is a temp data dir, a temp socket dir and a registry with two flat
// managed rows whose files exist: an ordinary instance and a replica
// (#162), which is its own host and migrates like any other.
func box(t *testing.T) (*Pass, *fakeFence) {
	t.Helper()
	dd := t.TempDir()
	// The socket dir is short on purpose: t.TempDir embeds the test's name,
	// and the nested control socket must fit a unix socket address, which
	// Preflight enforces.
	rd, err := os.MkdirTemp("", "ly")
	if err != nil {
		t.Fatal(err)
	}
	t.Cleanup(func() { os.RemoveAll(rd) })
	reg := openLayoutRegistry(t, dd)
	for _, rt := range []registry.Route{
		{Host: host, Kind: "instance", Hibernate: true, Status: "ready",
			Socket: filepath.Join(rd, host+".sock"), DBPath: provision.DefaultDBPath(dd, host)},
		{Host: replica, Kind: "instance", Hibernate: true, Replica: true,
			Socket: filepath.Join(rd, replica+".sock"), DBPath: provision.DefaultDBPath(dd, replica)},
	} {
		if err := reg.Upsert(rt); err != nil {
			t.Fatal(err)
		}
		seedFlat(t, dd, rt.Host)
	}
	f := &fakeFence{}
	p := &Pass{Registry: reg, Fence: f, DataDir: dd, SocketDir: rd, SocketsGID: -1, Now: time.Now,
		SweptAtStart: true,
		Reload:       func() { f.events = append(f.events, "reload") },
		Logf:         t.Logf}
	return p, f
}

func byHost(t *testing.T, rs []Result) map[string]Result {
	t.Helper()
	out := map[string]Result{}
	for _, r := range rs {
		if _, dup := out[r.Host]; dup {
			t.Errorf("host %q reported twice: %+v", r.Host, rs)
		}
		out[r.Host] = r
	}
	return out
}

func endpoints(t *testing.T, p *Pass, h string, mode Mode) Endpoint {
	t.Helper()
	plan, err := PlanFor(p.DataDir, p.SocketDir, h, mode)
	if err != nil {
		t.Fatal(err)
	}
	return plan.To
}

func rowAt(t *testing.T, p *Pass, h string) Endpoint {
	t.Helper()
	rt, err := p.Registry.Get(h)
	if err != nil || rt == nil {
		t.Fatalf("row %s: %v %v", h, rt, err)
	}
	return Endpoint{DB: rt.DBPath, Socket: rt.Socket}
}

func assertFlatUntouched(t *testing.T, p *Pass, h string) {
	t.Helper()
	if got := rowAt(t, p, h); got != endpoints(t, p, h, ModeFlat) {
		t.Errorf("%s row moved: %+v", h, got)
	}
	if b, err := os.ReadFile(provision.DefaultDBPath(p.DataDir, h)); err != nil || string(b) != "db "+h {
		t.Errorf("%s flat db: %q %v", h, b, err)
	}
	if JournalPresent(p.DataDir, h) {
		t.Errorf("%s has a journal", h)
	}
}

func TestConvergeNestedMovesFilesFlipsTheRowAndRemovesTheJournal(t *testing.T) {
	p, f := box(t)
	// Step 9's order: routing reloads on the committed row with the journal
	// gone, and only then is the host released.
	var reloads []string
	p.Reload = func() {
		for _, h := range []string{host, replica} {
			if f.fences[h] != 0 {
				if JournalPresent(p.DataDir, h) {
					t.Errorf("reload for %s ran with its journal still present", h)
				}
				reloads = append(reloads, h+" "+rowAt(t, p, h).DB)
			}
		}
	}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "migrated" {
			t.Fatalf("%s: %+v", h, res[h])
		}
		want := endpoints(t, p, h, ModeNested)
		if got := rowAt(t, p, h); got != want {
			t.Fatalf("%s row = %+v, want %+v", h, got, want)
		}
		raw := rawInstanceDir(p.DataDir, h)
		for _, name := range []string{h + ".db", h + ".db-wal", h + ".env", filepath.Join(shadowName(h), "generation")} {
			if _, err := os.Stat(filepath.Join(raw, name)); err != nil {
				t.Errorf("%s not at the raw destination: %v", name, err)
			}
		}
		if _, err := os.Lstat(provision.DefaultDBPath(p.DataDir, h)); !errors.Is(err, os.ErrNotExist) {
			t.Errorf("%s flat db still present", h)
		}
		if JournalPresent(p.DataDir, h) {
			t.Errorf("%s journal left behind after a clean move", h)
		}
		if fi, err := os.Lstat(rawRunDir(p.SocketDir, h)); err != nil || !fi.IsDir() {
			t.Errorf("%s run directory: %v %v", h, fi, err)
		}
	}
	if _, err := os.Stat(filepath.Join(rawInstanceDir(p.DataDir, host), host+".db.corrupt")); err != nil {
		t.Errorf("quarantine not moved: %v", err)
	}
	if len(f.fences) != 0 {
		t.Errorf("fences left installed after a clean move: %v", f.fences)
	}
	wantReloads := []string{host + " " + endpoints(t, p, host, ModeNested).DB, replica + " " + endpoints(t, p, replica, ModeNested).DB}
	if !reflect.DeepEqual(reloads, wantReloads) {
		t.Errorf("reloads under the fence = %v, want %v", reloads, wantReloads)
	}
	// Idempotent: a second converge does nothing.
	for h, r := range byHost(t, p.Converge(context.Background(), ModeNested)) {
		if r.Outcome != "unchanged" {
			t.Fatalf("second converge, %s: %+v", h, r)
		}
	}
	// Revert restores the flat layout byte for byte.
	p.Reload = nil
	res = byHost(t, p.Converge(context.Background(), ModeFlat))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "reverted" {
			t.Fatalf("revert %s = %+v", h, res[h])
		}
		assertFlatUntouched(t, p, h)
		if _, err := os.Stat(filepath.Join(p.DataDir, shadowName(h), "generation")); err != nil {
			t.Errorf("%s shadow not back: %v", h, err)
		}
	}
	if len(f.fences) != 0 {
		t.Errorf("fences left installed after a revert: %v", f.fences)
	}
}

// An unproven stop is accepted only for a host with no live entry, no
// uncertain stop on record, and an agent whose startup unit sweep
// succeeded. Anything else defers the host, writes no journal, and lifts
// the fence this pass installed.
func TestConvergeTrustsAnUnprovenStopOnlyWhenNothingCanBeRunning(t *testing.T) {
	for _, tc := range []struct {
		name      string
		swept     bool
		awake     bool
		uncertain bool
		want      string
	}{
		{"swept, quiesced, certain", true, false, false, "migrated"},
		{"no startup sweep", false, false, false, "deferred"},
		{"an entry is live", true, true, false, "deferred"},
		{"the last stop is uncertain", true, false, true, "deferred"},
	} {
		t.Run(tc.name, func(t *testing.T) {
			p, f := box(t)
			p.SweptAtStart = tc.swept
			f.unproven = map[string]bool{host: true}
			f.awake = map[string]bool{host: tc.awake}
			f.uncertain = map[string]bool{host: tc.uncertain}
			r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
			if r.Outcome != tc.want {
				t.Fatalf("result = %+v, want %s", r, tc.want)
			}
			if tc.want == "deferred" {
				if !strings.Contains(r.Reason, "stop unproven") {
					t.Errorf("reason = %q", r.Reason)
				}
				assertFlatUntouched(t, p, host)
			}
			if len(f.fences) != 0 {
				t.Errorf("a layout fence was left installed: %v", f.fences)
			}
		})
	}
}

// wakeBackend is an activator.Backend with no processes, no S3 and no
// replication: just enough for a real Service to wake a host.
type wakeBackend struct {
	mu      sync.Mutex
	started []registry.Route
}

type nopLease struct{}

func (nopLease) Replicate(context.Context, string, time.Duration) (activator.Rep, error) {
	return nopRep{}, nil
}
func (nopLease) Release(context.Context) error { return nil }

type nopRep struct{}

func (nopRep) Err() error                  { return nil }
func (nopRep) Close(context.Context) error { return nil }

type nopProc struct{}

func (nopProc) Stop(context.Context) error { return nil }
func (nopProc) Exited() bool               { return false }

func (b *wakeBackend) Acquire(context.Context, registry.Route) (activator.Lease, error) {
	return nopLease{}, nil
}
func (b *wakeBackend) Restore(context.Context, registry.Route) (bool, error) { return false, nil }
func (b *wakeBackend) Seed(context.Context, registry.Route) error            { return nil }
func (b *wakeBackend) Start(_ context.Context, rt registry.Route) (activator.Proc, string, error) {
	b.mu.Lock()
	defer b.mu.Unlock()
	b.started = append(b.started, rt)
	return nopProc{}, "", nil
}
func (b *wakeBackend) NextDue(context.Context, registry.Route) (int64, error) { return 0, nil }
func (b *wakeBackend) HasSidecar(registry.Route) bool                         { return false }
func (b *wakeBackend) StartSidecar(context.Context, registry.Route) (activator.Proc, error) {
	return nil, errors.New("wakeBackend: no sidecar in this fixture")
}
func (b *wakeBackend) StartWorker(context.Context, registry.Route, map[string]string) (activator.Proc, string, error) {
	return nil, "", errors.New("wakeBackend: no worker in this fixture")
}

func realService(t *testing.T, p *Pass) (*activator.Service, *wakeBackend) {
	t.Helper()
	b := &wakeBackend{}
	svc := activator.New(activator.Config{Bucket: "test-bucket", Idle: time.Hour}, b, p.Registry)
	// The wake re-reads the committed row, so the test sees which paths it
	// actually woke on.
	svc.Routes = p.Registry
	t.Cleanup(func() {
		ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
		defer cancel()
		svc.Shutdown(ctx)
	})
	p.Fence = svc
	return svc, b
}

// After an agent restart no asleep host has a stop on record, so the real
// activator answers ErrStopUnproven for every one of them. The pass must
// still migrate such a host, and must leave it wakeable.
func TestConvergeAgainstARealActivatorMigratesAnEntrylessHostAndItWakes(t *testing.T) {
	p, _ := box(t)
	svc, b := realService(t, p)
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "migrated" {
		t.Fatalf("result = %+v", res[host])
	}
	if svc.Fenced(host) {
		t.Fatal("the host is still fenced after the pass")
	}
	rt, err := p.Registry.Get(host)
	if err != nil || rt == nil {
		t.Fatal(err)
	}
	rec := httptest.NewRecorder()
	req := httptest.NewRequest(http.MethodGet, "http://"+host+"/", nil)
	ok := http.HandlerFunc(func(w http.ResponseWriter, _ *http.Request) { w.WriteHeader(http.StatusOK) })
	if !svc.ServeHTTP(rec, req, *rt, ok) || rec.Code != http.StatusOK {
		t.Fatalf("wake after the pass: code %d", rec.Code)
	}
	b.mu.Lock()
	defer b.mu.Unlock()
	if len(b.started) != 1 || b.started[0].DBPath != endpoints(t, p, host, ModeNested).DB {
		t.Fatalf("started = %+v", b.started)
	}
}

func TestConvergeAgainstARealActivatorDefersWithoutAStartupSweep(t *testing.T) {
	p, _ := box(t)
	svc, _ := realService(t, p)
	p.SweptAtStart = false
	for h, r := range byHost(t, p.Converge(context.Background(), ModeNested)) {
		if r.Outcome != "deferred" || !strings.Contains(r.Reason, "stop unproven") {
			t.Errorf("%s: %+v", h, r)
		}
		if svc.Fenced(h) {
			t.Errorf("%s left fenced after a refused quiesce", h)
		}
		assertFlatUntouched(t, p, h)
	}
}

// Another operation's fence (the residence move's cut) shares the
// activator's one fence slot per host. The pass must neither take it over
// nor lift it.
func TestConvergeDefersAHostSomeoneElseFenced(t *testing.T) {
	p, _ := box(t)
	svc, _ := realService(t, p)
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if _, err := svc.FenceAndStop(ctx, host, 5); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
		t.Fatalf("pre-installing the other fence: %v", err)
	}
	p.Generation = func() uint64 { return 1000 }
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "deferred" || !strings.Contains(res[host].Reason, "fenced by another operation") {
		t.Fatalf("fenced host: %+v", res[host])
	}
	if res[replica].Outcome != "migrated" {
		t.Fatalf("the other host: %+v", res[replica])
	}
	assertFlatUntouched(t, p, host)
	if !svc.LiftFence(host, 5) {
		t.Fatal("the original fence is not in place at its own generation")
	}
}

// On the serial-move leg a row that becomes eligible while its host is
// awake is left for a pass that finds it asleep: DeferAwake defers the
// host before any fence is asked for, so the move never stops a running
// tenant. A host already at the intent is unchanged, awake or not, and the
// same awake host is moved when DeferAwake is off.
func TestConvergeDefersAnAwakeHostBeforeFencingIt(t *testing.T) {
	p, f := box(t)
	p.DeferAwake = true
	f.awake = map[string]bool{host: true}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "deferred" || !strings.HasPrefix(r.Reason, "awake") {
		t.Fatalf("awake host: %+v", r)
	}
	if res[replica].Outcome != "migrated" {
		t.Fatalf("the asleep host: %+v", res[replica])
	}
	for _, e := range f.events {
		if e == "fence "+host || e == "held "+host {
			t.Fatalf("the awake host was offered a fence: %v", f.events)
		}
	}
	assertFlatUntouched(t, p, host)

	// The next pass finds the host asleep and moves it; the replica is awake
	// now, but it is already nested.
	f.awake = map[string]bool{replica: true}
	res = byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "migrated" {
		t.Fatalf("the host once asleep: %+v", res[host])
	}
	if res[replica].Outcome != "unchanged" {
		t.Fatalf("an awake host already at the intent: %+v", res[replica])
	}

	p.DeferAwake = false
	f.awake = map[string]bool{host: true}
	if r := byHost(t, p.Converge(context.Background(), ModeFlat))[host]; r.Outcome != "reverted" {
		t.Fatalf("without DeferAwake the awake host: %+v", r)
	}
}

func TestConvergeDefersARowThatIsNotReady(t *testing.T) {
	p, f := box(t)
	if err := p.Registry.SetStatus(host, "moving"); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
	if r.Outcome != "deferred" || !strings.Contains(r.Reason, "moving") {
		t.Fatalf("result = %+v", r)
	}
	for _, e := range f.events {
		if e == "fence "+host {
			t.Fatal("a moving row was fenced")
		}
	}
	assertFlatUntouched(t, p, host)
}

// The row the pass classified can change before its fence lands: a move
// cut starting, an operator unflagging it. The pass re-reads it under the
// fence and refuses, writing nothing.
func TestConvergeRefusesARowThatChangedUnderTheFence(t *testing.T) {
	for _, tc := range []struct {
		name   string
		change func(reg *registry.Registry) error
	}{
		{"status", func(reg *registry.Registry) error { return reg.SetStatus(host, "moving") }},
		{"unflagged", func(reg *registry.Registry) error { return reg.SetHibernate(host, false) }},
	} {
		t.Run(tc.name, func(t *testing.T) {
			p, f := box(t)
			f.onFence = func(h string) {
				if h == host {
					if err := tc.change(p.Registry); err != nil {
						t.Fatal(err)
					}
				}
			}
			r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
			if r.Outcome != "refused" {
				t.Fatalf("result = %+v", r)
			}
			assertFlatUntouched(t, p, host)
			if len(f.fences) != 0 {
				t.Errorf("fence left installed: %v", f.fences)
			}
		})
	}
}

// A box whose isolation probe is not ready does not start nested tenants,
// so it does not make any.
func TestConvergeNestedIsGatedOnIsolation(t *testing.T) {
	p, f := box(t)
	calls := 0
	p.NestedOK = func() error { calls++; return errors.New("probe: not ready") }
	res := p.Converge(context.Background(), ModeNested)
	if len(res) != 1 || res[0].Host != "" || res[0].Outcome != "refused" || res[0].Reason != "isolation not ready: probe: not ready" {
		t.Fatalf("results = %+v", res)
	}
	if calls != 1 {
		t.Errorf("NestedOK called %d times", calls)
	}
	if len(f.events) != 0 {
		t.Errorf("the pass touched hosts: %v", f.events)
	}
	for _, h := range []string{host, replica} {
		assertFlatUntouched(t, p, h)
	}
	// Reverting and recovering are never gated: they are how a box gets
	// back to a layout it can serve.
	p.NestedOK = nil
	p.Converge(context.Background(), ModeNested)
	p.NestedOK = func() error { return errors.New("probe: not ready") }
	for h, r := range byHost(t, p.Converge(context.Background(), ModeFlat)) {
		if r.Outcome != "reverted" {
			t.Errorf("revert %s under a failing probe: %+v", h, r)
		}
	}
}

func TestConvergeSkipsForeignAndUnmanagedRows(t *testing.T) {
	p, _ := box(t)
	if err := p.Registry.Upsert(registry.Route{Host: "custom.example.com", Kind: "instance", Hibernate: true,
		Socket: filepath.Join(p.SocketDir, "custom.example.com.sock"), DBPath: "/data/sub/store.db"}); err != nil {
		t.Fatal(err)
	}
	if err := p.Registry.Upsert(registry.Route{Host: "ext.example.com", Kind: "instance", Hibernate: false,
		Socket: filepath.Join(p.SocketDir, "ext.example.com.sock"), DBPath: provision.DefaultDBPath(p.DataDir, "ext.example.com")}); err != nil {
		t.Fatal(err)
	}
	out := byHost(t, p.Converge(context.Background(), ModeNested))
	if out["custom.example.com"].Outcome != "skipped" || !strings.Contains(out["custom.example.com"].Reason, "foreign") {
		t.Errorf("foreign row: %+v", out["custom.example.com"])
	}
	if _, listed := out["ext.example.com"]; listed {
		t.Errorf("an unmanaged row was considered: %+v", out["ext.example.com"])
	}
}

// A same-uid tenant can plant <data>/.instances/<iid> as a symlink before
// the pass arrives. The pass moves nothing through it, and since the
// symlink is the tenant's own doing the host is held fenced and down
// (review G1) rather than released to run flat again; once the link is
// gone the next pass moves it from the kept fence.
func TestConvergeRefusesASymlinkedInstanceDirectory(t *testing.T) {
	p, f := box(t)
	elsewhere := t.TempDir()
	raw := rawInstanceDir(p.DataDir, host)
	if err := os.MkdirAll(filepath.Dir(raw), 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.Symlink(elsewhere, raw); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
	if r.Outcome != "held" || !strings.Contains(r.Reason, "symlink") {
		t.Fatalf("result = %+v", r)
	}
	assertFlatUntouched(t, p, host)
	if entries, _ := os.ReadDir(elsewhere); len(entries) != 0 {
		t.Errorf("files landed through the symlink: %v", entries)
	}
	if f.fences[host] == 0 {
		t.Error("the fence was lifted; the host would wake flat again behind its own symlink")
	}
	os.Remove(raw)
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "migrated" || f.fences[host] != 0 {
		t.Fatalf("after the symlink is gone: %+v fence=%d", r, f.fences[host])
	}
}

func TestSerializeWrapsJournalWriteToJournalRemoval(t *testing.T) {
	p, _ := box(t)
	var wrapped []string
	p.Serialize = func(fn func()) {
		for _, h := range []string{host, replica} {
			if JournalPresent(p.DataDir, h) {
				t.Errorf("%s journal present before the serialized section", h)
			}
		}
		fn()
		for _, h := range []string{host, replica} {
			if JournalPresent(p.DataDir, h) {
				t.Errorf("%s journal present after the serialized section", h)
			}
		}
		wrapped = append(wrapped, "section")
	}
	p.Converge(context.Background(), ModeNested)
	if len(wrapped) != 2 {
		t.Fatalf("serialized sections = %d, want one per host", len(wrapped))
	}
}

func TestRecoverCompletesEveryCrashBoundary(t *testing.T) {
	// Drive the pass through a move with a seam that aborts after each
	// step, then Recover, and assert the end state is the nested layout
	// with no journal, every time.
	for _, step := range []string{"after-journal", "mid-files", "after-files", "after-cas", "before-remove"} {
		t.Run(step, func(t *testing.T) {
			p, f := box(t)
			p.abortAt = step
			p.Converge(context.Background(), ModeNested)
			p.abortAt = ""
			for _, h := range []string{host, replica} {
				if !JournalPresent(p.DataDir, h) {
					t.Fatalf("%s: no journal after aborting at %s", h, step)
				}
			}
			if len(f.fences) != 0 {
				t.Errorf("an aborted move left its fence: %v", f.fences)
			}
			res := byHost(t, p.Recover(context.Background(), ""))
			for _, h := range []string{host, replica} {
				if res[h].Outcome != "recovered" {
					t.Fatalf("recover %s = %+v", h, res[h])
				}
				if got := rowAt(t, p, h); got != endpoints(t, p, h, ModeNested) {
					t.Errorf("%s row not nested after recovery from %s: %+v", h, step, got)
				}
				if b, err := os.ReadFile(filepath.Join(rawInstanceDir(p.DataDir, h), h+".db")); err != nil || string(b) != "db "+h {
					t.Errorf("%s db not at destination after recovery from %s: %q %v", h, step, b, err)
				}
				if JournalPresent(p.DataDir, h) {
					t.Errorf("%s journal remains after recovery", h)
				}
			}
			if len(f.fences) != 0 {
				t.Errorf("recovery left a fence: %v", f.fences)
			}
		})
	}
}

func TestRecoverRefusesAForgedJournalAndKeepsTheHostFenced(t *testing.T) {
	p, _ := box(t)
	j := &Journal{Host: host, Mode: ModeNested,
		From:  Endpoint{DB: "/data/console/console.db", Socket: filepath.Join(p.SocketDir, host+".sock")},
		To:    endpoints(t, p, host, ModeNested),
		Phase: PhaseFiles}
	for _, n := range artifactNames {
		j.Artifacts = append(j.Artifacts, JournalArtifact{Name: n, Present: true})
	}
	if err := WriteJournal(p.DataDir, j); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome != "refused" || !strings.Contains(r.Reason, "not acting") {
		t.Fatalf("forged journal: %+v", r)
	}
	if !JournalPresent(p.DataDir, host) {
		t.Error("forged journal removed; the host must stay fenced for an operator")
	}
	if _, err := os.Stat(filepath.Join(p.DataDir, "console")); err == nil {
		t.Error("something touched the forged source")
	}
	// Converge reports the host once, as the recovery refused it.
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "refused" {
		t.Errorf("converge over a refused journal: %+v", res[host])
	}
}

// A file at both ends of a move cannot come from a crash: the destination
// check cleared every destination before the journal was written, and a
// rename leaves a file at exactly one of them. Recovery refuses to choose.
func TestRecoverRefusesAnArtifactAtBothEnds(t *testing.T) {
	p, _ := box(t)
	p.abortAt = "after-journal"
	p.Converge(context.Background(), ModeNested)
	p.abortAt = ""
	dst := filepath.Join(rawInstanceDir(p.DataDir, host), host+".db")
	if err := os.WriteFile(dst, []byte("planted"), 0o600); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Recover(context.Background(), ""))[host]
	if r.Outcome != "refused" || !strings.Contains(r.Reason, "both") {
		t.Fatalf("result = %+v", r)
	}
	if !JournalPresent(p.DataDir, host) {
		t.Error("journal removed over an ambiguous artifact")
	}
	if b, _ := os.ReadFile(provision.DefaultDBPath(p.DataDir, host)); string(b) != "db "+host {
		t.Error("the source database was touched")
	}
}

// A journal at row says every file it records present is at its
// destination. Recovery checks that before it flips the row, because a row
// pointed at a missing database would have the next wake restore a second
// one there.
func TestRecoverAtRowRefusesAMissingDestinationBeforeTheCAS(t *testing.T) {
	p, f := box(t)
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeNested)
	p.abortAt = ""
	j, found, err := ReadJournal(p.DataDir, host)
	if err != nil || !found {
		t.Fatalf("setup: %v %v", found, err)
	}
	if err := UpdatePhase(p.DataDir, j, PhaseRow); err != nil {
		t.Fatal(err)
	}
	wal := filepath.Join(rawInstanceDir(p.DataDir, host), host+".db-wal")
	if err := os.Remove(wal); err != nil {
		t.Fatal(err)
	}
	f.events = nil
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome != "refused" || !strings.Contains(r.Reason, wal) {
		t.Fatalf("result = %+v, want a refusal naming %s", r, wal)
	}
	if got, found, _ := ReadJournal(p.DataDir, host); !found || got.Phase != PhaseRow {
		t.Errorf("journal not kept at row: %+v", got)
	}
	if rowAt(t, p, host) != endpoints(t, p, host, ModeFlat) {
		t.Errorf("the row flipped over a missing file: %+v", rowAt(t, p, host))
	}
}

// Design §3.3: a journal is completed to its own endpoint, and the host is
// then converged to the fleet's intent before it is released. The fence
// is held from the recovery through the convergence: nothing lifts it in
// between, so the host never wakes at the journal's endpoint.
func TestConvergeRecoversAJournalBeforeConvergingToANewIntent(t *testing.T) {
	p, f := box(t)
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeNested)
	p.abortAt = ""
	f.events = nil
	// The intent changed to flat while the move was in flight: the journal
	// completes to ITS endpoint (nested), then the host converges to flat.
	res := byHost(t, p.Converge(context.Background(), ModeFlat))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "recovered" {
			t.Errorf("%s: %+v", h, res[h])
		}
		assertFlatUntouched(t, p, h)
	}
	want := []string{
		"fence " + host, "reload", "reload", "lift " + host,
		"fence " + replica, "reload", "reload", "lift " + replica,
	}
	if !reflect.DeepEqual(f.events, want) {
		t.Fatalf("events = %v\nwant     %v", f.events, want)
	}
}

// With no intent recorded, a journal is completed and the host released
// where the journal ends.
func TestRecoverWithNoIntentReleasesAtTheJournalsEndpoint(t *testing.T) {
	p, f := box(t)
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeNested)
	p.abortAt = ""
	for h, r := range byHost(t, p.Recover(context.Background(), "")) {
		if r.Outcome != "recovered" || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) {
			t.Errorf("%s: %+v at %+v", h, r, rowAt(t, p, h))
		}
	}
	if len(f.fences) != 0 {
		t.Errorf("fences: %v", f.fences)
	}
}

// When the agent starts and cannot read the fleet's intent, a revert in
// flight is completed but kept fenced at phase done: releasing it flat
// could let a forged revert run a nested host unisolated until an intent
// is read. A move to nested is completed and released as usual.
func TestRecoverHoldsACompletedRevertWhileTheIntentIsUnreadable(t *testing.T) {
	p, f := box(t)
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
		t.Fatalf("setup: %+v", res)
	}
	p.abortAt = "after-cas"
	p.Converge(context.Background(), ModeFlat)
	p.abortAt = ""
	res := byHost(t, p.Recover(context.Background(), IntentUnreadable))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "deferred" || !strings.Contains(res[h].Reason, "intent") {
			t.Errorf("%s: %+v", h, res[h])
		}
		j, found, err := ReadJournal(p.DataDir, h)
		if err != nil || !found || j.Phase != PhaseDone {
			t.Fatalf("%s journal: %+v %v %v", h, j, found, err)
		}
		if rowAt(t, p, h) != endpoints(t, p, h, ModeFlat) {
			t.Errorf("%s row not completed to flat", h)
		}
	}
	if len(f.fences) != 0 {
		t.Errorf("the journal is the fence; the activator's must not stay: %v", f.fences)
	}
	// Once the intent is read, the held host converges to it.
	res = byHost(t, p.Recover(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "recovered" || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) || JournalPresent(p.DataDir, h) {
			t.Errorf("%s after the intent: %+v at %+v", h, res[h], rowAt(t, p, h))
		}
	}

	// A move toward nested is released even while the intent is unreadable.
	q, _ := box(t)
	q.abortAt = "after-files"
	q.Converge(context.Background(), ModeNested)
	q.abortAt = ""
	for h, r := range byHost(t, q.Recover(context.Background(), IntentUnreadable)) {
		if r.Outcome != "recovered" || JournalPresent(q.DataDir, h) {
			t.Errorf("%s nested journal under an unreadable intent: %+v", h, r)
		}
	}
}

// abortedNestedMove leaves both hosts with a nested journal at phase files
// and their files already at the nested destination.
func abortedNestedMove(t *testing.T, p *Pass) {
	t.Helper()
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeNested)
	p.abortAt = ""
	for _, h := range []string{host, replica} {
		if j, found, err := ReadJournal(p.DataDir, h); err != nil || !found || j.Phase != PhaseFiles {
			t.Fatalf("setup %s: %+v %v %v", h, j, found, err)
		}
	}
}

func installForeignFence(t *testing.T, svc *activator.Service, h string, gen uint64) {
	t.Helper()
	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
	defer cancel()
	if _, err := svc.FenceAndStop(ctx, h, gen); err != nil && !errors.Is(err, activator.ErrStopUnproven) {
		t.Fatalf("installing the foreign fence: %v", err)
	}
}

// A journal on a host another operation holds fenced is completed under
// that operation's fence: the pass installs and lifts nothing. When the
// completed endpoint is the fleet's intent there is nothing left to do,
// and the journal goes.
func TestRecoverUnderAForeignFenceCompletesAndRemovesAtTheIntent(t *testing.T) {
	p, _ := box(t)
	svc, _ := realService(t, p)
	abortedNestedMove(t, p)
	installForeignFence(t, svc, host, 5)
	res := byHost(t, p.Recover(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "recovered" || JournalPresent(p.DataDir, h) || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) {
			t.Errorf("%s: %+v at %+v", h, res[h], rowAt(t, p, h))
		}
	}
	if svc.Fenced(replica) {
		t.Error("the unfenced host was left fenced")
	}
	if !svc.LiftFence(host, 5) {
		t.Fatal("the foreign fence is not in place at its own generation")
	}
}

// When the completed endpoint is not the intent, the pass cannot converge
// the host without a fence of its own. It keeps the journal at done, which
// keeps the host fenced if the other operation lifts its fence, and the
// next recovery after that lift converges it.
func TestRecoverUnderAForeignFenceKeepsTheJournalWhenNotAtTheIntent(t *testing.T) {
	p, _ := box(t)
	svc, _ := realService(t, p)
	abortedNestedMove(t, p)
	installForeignFence(t, svc, host, 5)
	res := byHost(t, p.Recover(context.Background(), ModeFlat))
	if r := res[host]; r.Outcome != "deferred" || !strings.Contains(r.Reason, "fence") {
		t.Fatalf("held host: %+v", r)
	}
	// The move waits for the journal and the journal waits for the move,
	// so the reason has to tell an operator both ways out.
	for _, want := range []string{"instance move", "abort the move", provision.LayoutJournalPath(p.DataDir, host), "stands at its destination"} {
		if r := res[host]; !strings.Contains(r.Reason, want) {
			t.Errorf("reason %q does not say %q", r.Reason, want)
		}
	}
	if r := res[host]; strings.Contains(r.Reason, "\u2014") {
		t.Errorf("reason has an em dash: %q", r.Reason)
	}
	if j, found, err := ReadJournal(p.DataDir, host); err != nil || !found || j.Phase != PhaseDone {
		t.Fatalf("journal: %+v %v %v", j, found, err)
	}
	if rowAt(t, p, host) != endpoints(t, p, host, ModeNested) {
		t.Errorf("row not completed to the journal's endpoint: %+v", rowAt(t, p, host))
	}
	if res[replica].Outcome != "recovered" {
		t.Errorf("unfenced host: %+v", res[replica])
	}
	assertFlatUntouched(t, p, replica)
	if !svc.Fenced(host) {
		t.Fatal("the foreign fence was lifted")
	}
	if !svc.LiftFence(host, 5) {
		t.Fatal("the foreign fence is not at its own generation")
	}
	// The other operation is gone: the next recovery converges the host.
	if r := byHost(t, p.Recover(context.Background(), ModeFlat))[host]; r.Outcome != "recovered" {
		t.Fatalf("after the foreign fence lifted: %+v", r)
	}
	assertFlatUntouched(t, p, host)
	if svc.Fenced(host) {
		t.Error("left fenced")
	}
}

// Under a foreign fence the pass has no stop proof of its own, so it acts
// only on a host with no live entry. Otherwise nothing changes.
func TestRecoverUnderAForeignFenceDefersAnAwakeHost(t *testing.T) {
	p, f := box(t)
	abortedNestedMove(t, p)
	f.fences = map[string]uint64{host: 5}
	f.awake = map[string]bool{host: true}
	f.events = nil
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome != "deferred" {
		t.Fatalf("result = %+v", r)
	}
	if j, found, _ := ReadJournal(p.DataDir, host); !found || j.Phase != PhaseFiles {
		t.Fatalf("journal changed: %+v", j)
	}
	if rowAt(t, p, host) != endpoints(t, p, host, ModeFlat) {
		t.Errorf("row changed: %+v", rowAt(t, p, host))
	}
	if f.fences[host] != 5 {
		t.Errorf("foreign fence = %d", f.fences[host])
	}
	for _, e := range f.events {
		if e == "lift "+host || e == "fence "+host {
			t.Errorf("the pass touched the host's fence: %v", f.events)
		}
	}
}

// Completing a journal is never gated, but taking the host on toward a
// nested intent is, as Converge(ModeNested) is. With the probe failing, a
// revert completes and the host is held flat, not released: there is no
// degraded mode (design §8.5), and a released flat host is exactly what a
// forged revert wants. The journal at done is the fence, so the pass
// lifts its in-memory one.
func TestRecoverGatesTheLegTowardNestedAndHoldsFlat(t *testing.T) {
	p, f := box(t)
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
		t.Fatalf("setup: %+v", res)
	}
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeFlat)
	p.abortAt = ""
	calls := 0
	p.NestedOK = func() error { calls++; return errors.New("probe: not ready") }
	f.events = nil
	res := byHost(t, p.Recover(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if r := res[h]; r.Outcome != "deferred" || !strings.Contains(r.Reason, "isolation not ready: probe: not ready") {
			t.Errorf("%s: %+v", h, r)
		}
		assertHeldFlat(t, p, h)
	}
	if calls != 1 {
		t.Errorf("NestedOK called %d times, want once per recovery", calls)
	}
	want := []string{"fence " + host, "reload", "lift " + host, "fence " + replica, "reload", "lift " + replica}
	if !reflect.DeepEqual(f.events, want) {
		t.Fatalf("events = %v\nwant     %v", f.events, want)
	}
	// Once the probe passes, the next recovery takes the held host on to
	// nested.
	p.NestedOK = nil
	for h, r := range byHost(t, p.Recover(context.Background(), ModeNested)) {
		if r.Outcome != "recovered" || JournalPresent(p.DataDir, h) || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) {
			t.Errorf("%s after the probe passed: %+v", h, r)
		}
	}
}

// assertHeldFlat is a host whose revert completed and is held: the row
// and files at flat, and the journal still there at done.
func assertHeldFlat(t *testing.T, p *Pass, h string) {
	t.Helper()
	if got := rowAt(t, p, h); got != endpoints(t, p, h, ModeFlat) {
		t.Errorf("%s row not at flat: %+v", h, got)
	}
	if b, err := os.ReadFile(provision.DefaultDBPath(p.DataDir, h)); err != nil || string(b) != "db "+h {
		t.Errorf("%s flat db: %q %v", h, b, err)
	}
	if j, found, err := ReadJournal(p.DataDir, h); err != nil || !found || j.Mode != ModeFlat || j.Phase != PhaseDone {
		t.Errorf("%s journal: %+v %v %v, want a flat journal held at done", h, j, found, err)
	}
}

// addFlatRow adds a fresh flat managed row with its files, after any setup
// that would otherwise have moved it.
func addFlatRow(t *testing.T, p *Pass, h string) {
	t.Helper()
	if err := p.Registry.Upsert(registry.Route{Host: h, Kind: "instance", Hibernate: true,
		Socket: filepath.Join(p.SocketDir, h+".sock"), DBPath: provision.DefaultDBPath(p.DataDir, h)}); err != nil {
		t.Fatal(err)
	}
	seedFlat(t, p.DataDir, h)
}

// forgeFlatJournal writes a well-formed revert journal for a nested host,
// recording each artifact present exactly when it is at its nested source,
// except the names in absent, which it records absent regardless. It is
// what a flat tenant sharing the agent's uid can write into <data>/layout.
func forgeFlatJournal(t *testing.T, p *Pass, h string, absent ...string) {
	t.Helper()
	plan, err := PlanFor(p.DataDir, p.SocketDir, h, ModeFlat)
	if err != nil {
		t.Fatal(err)
	}
	present := make([]bool, len(plan.Artifacts))
	for i, a := range plan.Artifacts {
		ok, _ := lstatPresent(a.Src)
		present[i] = ok && a.Name != "socket" && !slicesContains(absent, a.Name)
	}
	if err := WriteJournal(p.DataDir, plan.Journal(present)); err != nil {
		t.Fatal(err)
	}
}

func slicesContains(xs []string, x string) bool {
	for _, v := range xs {
		if v == x {
			return true
		}
	}
	return false
}

// A journal completed to an endpoint that is not the fleet's intent keeps
// the host fenced until the converge leg succeeds. If the leg refuses, the
// journal stays at done: a forged revert must not leave a nested host
// released flat.
func TestRecoverHoldsTheJournalWhenTheLegTowardTheIntentRefuses(t *testing.T) {
	p, f := box(t)
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
		t.Fatalf("setup: %+v", res)
	}
	forgeFlatJournal(t, p, host)
	// A route that owns <host>.config.env under the flat rules makes the
	// way back to nested refuse at the ownership preflight.
	if err := p.Registry.Upsert(registry.Route{Host: host + ".config", Kind: "instance",
		Socket: filepath.Join(p.SocketDir, host+".config.sock"), DBPath: provision.DefaultDBPath(p.DataDir, host+".config")}); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome != "deferred" || !strings.Contains(r.Reason, "config") {
		t.Fatalf("result = %+v", r)
	}
	j, found, err := ReadJournal(p.DataDir, host)
	if err != nil || !found || j.Phase != PhaseDone || j.Mode != ModeFlat {
		t.Fatalf("journal not held at done: %+v %v %v", j, found, err)
	}
	if len(f.fences) != 0 {
		t.Errorf("fences: %v", f.fences)
	}
	// Every later pass holds it the same way; it is never released flat.
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "deferred" || !JournalPresent(p.DataDir, host) {
		t.Fatalf("second pass: %+v", r)
	}
}

// The attack the hold closes, end to end: a forged revert that records a
// real nested artifact absent. Recovery refuses it outright, and the host
// stays nested and fenced.
func TestRecoverDoesNotReleaseANestedHostFlatOverAForgedRevert(t *testing.T) {
	p, _ := box(t)
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
		t.Fatalf("setup: %+v", res)
	}
	shadow := filepath.Join(rawInstanceDir(p.DataDir, host), shadowName(host))
	if _, err := os.Stat(shadow); err != nil {
		t.Fatalf("setup: %v", err)
	}
	forgeFlatJournal(t, p, host, "litestream-shadow")
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome == "recovered" {
		t.Fatalf("result = %+v", r)
	}
	if rowAt(t, p, host) != endpoints(t, p, host, ModeNested) {
		t.Errorf("row released flat: %+v", rowAt(t, p, host))
	}
	if !JournalPresent(p.DataDir, host) {
		t.Error("journal removed; the host is no longer fenced")
	}
}

// A journal's "absent" is a claim a crash cannot contradict: the host has
// been fenced since the plan was made, so nothing recorded absent can have
// appeared at either end. Recovery refuses one that has, naming it.
func TestRecoverRefusesAnArtifactRecordedAbsentThatExists(t *testing.T) {
	for _, end := range []string{"source", "destination"} {
		t.Run(end, func(t *testing.T) {
			p, _ := box(t)
			abortedNestedMove(t, p)
			plan, _ := PlanFor(p.DataDir, p.SocketDir, host, ModeNested)
			sidecar := artifactNamed(t, plan, "sidecar-env")
			path := sidecar.Src
			if end == "destination" {
				path = sidecar.Dst
			}
			if err := os.WriteFile(path, []byte("X=1\n"), 0o600); err != nil {
				t.Fatal(err)
			}
			r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
			if r.Outcome != "refused" || !strings.Contains(r.Reason, path) || !strings.Contains(r.Reason, "absent") {
				t.Fatalf("result = %+v", r)
			}
			if j, found, _ := ReadJournal(p.DataDir, host); !found || j.Phase != PhaseFiles {
				t.Fatalf("journal changed: %+v", j)
			}
		})
	}
}

// Completing a journal is never gated, so a failing probe does not leave a
// crashed move fenced. Only fresh moves wait for the probe.
func TestConvergeNestedCompletesAJournalWhileTheProbeFails(t *testing.T) {
	p, _ := box(t)
	abortedNestedMove(t, p)
	addFlatRow(t, p, "f.example.com")
	calls := 0
	p.NestedOK = func() error { calls++; return errors.New("probe: not ready") }
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "recovered" || JournalPresent(p.DataDir, h) || rowAt(t, p, h) != endpoints(t, p, h, ModeNested) {
			t.Errorf("%s: %+v", h, res[h])
		}
	}
	if r := res[""]; r.Outcome != "refused" || r.Reason != "isolation not ready: probe: not ready" {
		t.Errorf("pass result: %+v", r)
	}
	if _, listed := res["f.example.com"]; listed {
		t.Errorf("a fresh move ran under a failing probe: %+v", res["f.example.com"])
	}
	assertFlatUntouched(t, p, "f.example.com")
	if calls != 1 {
		t.Errorf("NestedOK called %d times", calls)
	}
}

// One pass asks the probe once, whether the question comes from a
// recovery's leg toward nested or from the fresh moves.
func TestConvergeAsksTheIsolationGateOncePerPass(t *testing.T) {
	for _, tc := range []struct {
		name string
		gate error
	}{{"failing", errors.New("probe: not ready")}, {"ready", nil}} {
		t.Run(tc.name, func(t *testing.T) {
			p, _ := box(t)
			if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
				t.Fatalf("setup: %+v", res)
			}
			p.abortAt = "after-files"
			p.Converge(context.Background(), ModeFlat)
			p.abortAt = ""
			addFlatRow(t, p, "f.example.com")
			calls := 0
			p.NestedOK = func() error { calls++; return tc.gate }
			res := byHost(t, p.Converge(context.Background(), ModeNested))
			if calls != 1 {
				t.Fatalf("NestedOK called %d times in one pass", calls)
			}
			want := "recovered"
			if tc.gate != nil {
				want = "deferred"
			}
			for _, h := range []string{host, replica} {
				if res[h].Outcome != want {
					t.Errorf("%s: %+v, want %s", h, res[h], want)
				}
			}
			if tc.gate != nil {
				assertHeldFlat(t, p, host)
				assertFlatUntouched(t, p, "f.example.com")
			} else if res["f.example.com"].Outcome != "migrated" || rowAt(t, p, host) != endpoints(t, p, host, ModeNested) {
				t.Errorf("ready probe: %+v", res)
			}
		})
	}
}

// A host whose database is not on this box cannot move; finding that out
// must not cost it a stop, or directories and a destination check, on
// every pass.
func TestConvergeRefusesAHostWithNoLocalDatabaseWithoutFencingIt(t *testing.T) {
	p, f := box(t)
	db := provision.DefaultDBPath(p.DataDir, host)
	if err := os.Remove(db); err != nil {
		t.Fatal(err)
	}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "refused" || !strings.Contains(r.Reason, "db is missing at "+db) {
		t.Fatalf("result = %+v", r)
	}
	for _, e := range f.events {
		if strings.HasSuffix(e, " "+host) {
			t.Errorf("the host's fence was touched: %v", f.events)
		}
	}
	if _, err := os.Lstat(rawInstanceDir(p.DataDir, host)); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("directories were made for a host that cannot move: %v", err)
	}
	if res[replica].Outcome != "migrated" {
		t.Errorf("the other host: %+v", res[replica])
	}
}

// crashedRevert leaves both hosts nested-then-mid-revert: each has a flat
// journal at phase files with its files already at the flat destination.
func crashedRevert(t *testing.T, p *Pass) {
	t.Helper()
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "migrated" {
		t.Fatalf("setup: %+v", res)
	}
	p.abortAt = "after-files"
	p.Converge(context.Background(), ModeFlat)
	p.abortAt = ""
}

// The handover from the completed journal to the converge leg's journal
// is one rename, and anything going wrong on it leaves the host down: the
// held journal stays whole, the pass's fence stays installed, and nothing
// is released. A same-uid tenant can make <data>/layout unwritable at the
// right moment, so this is reachable, not only a disk fault.
func TestRecoverKeepsTheFenceWhenTheHandoverFails(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("root ignores the directory mode this test relies on")
	}
	p, f := box(t)
	crashedRevert(t, p)
	dir := journalDir(p.DataDir)
	calls := 0
	p.Serialize = func(fn func()) {
		calls++
		// The second section on the first host is the handover: make the
		// temp file impossible to create there, as a racing tenant could.
		if calls == 2 {
			if err := os.Chmod(dir, 0o500); err != nil {
				t.Fatal(err)
			}
			defer os.Chmod(dir, 0o700)
		}
		fn()
	}
	f.events = nil
	r := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if r.Outcome != "refused" || !strings.Contains(r.Reason, "held by this pass after a failed handover") {
		t.Fatalf("result = %+v", r)
	}
	j, found, err := ReadJournal(p.DataDir, host)
	if err != nil || !found || j.Mode != ModeFlat || j.Phase != PhaseDone {
		t.Fatalf("held journal not intact: %+v %v %v", j, found, err)
	}
	if f.fences[host] == 0 {
		t.Error("the pass's fence was lifted after a failed handover")
	}
	for _, e := range f.events {
		if e == "lift "+host {
			t.Errorf("released after a failed handover: %v", f.events)
		}
	}
	if left := tempFilesIn(t, dir); len(left) != 0 {
		t.Errorf("temp files left: %v", left)
	}
}

// A fence the pass kept after a failed handover is its own, not another
// operation's. The next pass recognises it by its generation, re-runs the
// leg once <data>/layout is writable again, and lifts it when the host
// stands at the intent.
func TestRecoverRetriesAHandoverUnderTheFenceItKept(t *testing.T) {
	if os.Geteuid() == 0 {
		t.Skip("root ignores the directory mode this test relies on")
	}
	p, f := box(t)
	crashedRevert(t, p)
	dir := journalDir(p.DataDir)
	calls := 0
	p.Serialize = func(fn func()) {
		calls++
		if calls == 2 {
			if err := os.Chmod(dir, 0o500); err != nil {
				t.Fatal(err)
			}
			defer os.Chmod(dir, 0o700)
		}
		fn()
	}
	first := byHost(t, p.Recover(context.Background(), ModeNested))[host]
	if first.Outcome != "refused" || !strings.Contains(first.Reason, "held by this pass after a failed handover") || !strings.Contains(first.Reason, "the next pass retries it") {
		t.Fatalf("first pass: %+v", first)
	}
	kept := f.fences[host]
	if kept == 0 {
		t.Fatal("the fence was not kept")
	}
	p.Serialize = nil
	f.events = nil
	res := p.Recover(context.Background(), ModeNested)
	second := byHost(t, res)[host]
	if second.Outcome != "recovered" || !strings.Contains(second.Reason, "converged to nested") {
		t.Fatalf("second pass: %+v", second)
	}
	for _, r := range append(res, first) {
		if strings.Contains(r.Reason, "another operation") {
			t.Errorf("the pass's own fence reported as another operation's: %+v", r)
		}
	}
	if JournalPresent(p.DataDir, host) || rowAt(t, p, host) != endpoints(t, p, host, ModeNested) {
		t.Fatalf("host not converged: journal %v, row %+v", JournalPresent(p.DataDir, host), rowAt(t, p, host))
	}
	if f.fences[host] != 0 {
		t.Fatalf("the kept fence (generation %d) is still installed at %d", kept, f.fences[host])
	}
	if !slicesContains(f.events, "lift "+host) || slicesContains(f.events, "fence "+host) {
		t.Errorf("events = %v, want the kept fence lifted and no new one installed", f.events)
	}
	// The record went with the fence: a later foreign fence at the same
	// generation is not mistaken for the pass's own.
	p.keptMu.Lock()
	_, still := p.kept[host]
	p.keptMu.Unlock()
	if still {
		t.Error("the kept generation is still recorded after the lift")
	}
}

// A successful handover leaves exactly one journal, the converge leg's
// own, and no temp file.
func TestRecoverHandsOverToTheNewJournalInOneStep(t *testing.T) {
	p, _ := box(t)
	crashedRevert(t, p)
	p.abortAt = "after-journal"
	p.Recover(context.Background(), ModeNested)
	p.abortAt = ""
	j, found, err := ReadJournal(p.DataDir, host)
	if err != nil || !found || j.Mode != ModeNested || j.Phase != PhaseFiles {
		t.Fatalf("journal after the handover: %+v %v %v", j, found, err)
	}
	entries, err := os.ReadDir(journalDir(p.DataDir))
	if err != nil {
		t.Fatal(err)
	}
	var names []string
	for _, e := range entries {
		names = append(names, e.Name())
	}
	want := []string{host + ".json", replica + ".json"}
	if !reflect.DeepEqual(names, want) {
		t.Fatalf("journal dir = %v, want %v", names, want)
	}
}

// The presence scan runs inside the section config sync is locked out of,
// so a bundle rendered just before the pass took the lock is recorded and
// moved, never stranded at the flat path behind a journal that calls it
// absent.
func TestThePresenceScanRunsInsideTheSerializedSection(t *testing.T) {
	p, _ := box(t)
	rendered := false
	p.Serialize = func(fn func()) {
		if !rendered {
			// Config sync's last render, landing while it still held the
			// lock the pass is now taking.
			for _, h := range []string{host, replica} {
				if err := os.WriteFile(filepath.Join(p.DataDir, h+".config.env"), []byte("A=1\n"), 0o600); err != nil {
					t.Fatal(err)
				}
			}
			rendered = true
		}
		fn()
	}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "migrated" {
			t.Fatalf("%s: %+v", h, res[h])
		}
		if _, err := os.Lstat(filepath.Join(p.DataDir, h+".config.env")); !errors.Is(err, os.ErrNotExist) {
			t.Errorf("%s bundle stranded at the flat path", h)
		}
		if _, err := os.Stat(filepath.Join(rawInstanceDir(p.DataDir, h), h+".config.env")); err != nil {
			t.Errorf("%s bundle not moved: %v", h, err)
		}
	}
}

// Review F1 end to end, as amended for landing: a planted hard link in
// the host's future directory is quarantined and the host migrates; the
// aside copy is outside the instance tree, status gets the event as a
// note, and the clean sibling is unaffected.
func TestConvergeQuarantinesPlantedEntriesAndMigrates(t *testing.T) {
	p, f := box(t)
	secret := filepath.Join(p.DataDir, "litestream.env")
	if err := os.WriteFile(secret, []byte("AWS_SECRET_ACCESS_KEY=x\n"), 0o600); err != nil {
		t.Fatal(err)
	}
	dir := rawInstanceDir(p.DataDir, host)
	if err := os.MkdirAll(dir, 0o700); err != nil {
		t.Fatal(err)
	}
	if err := os.Link(secret, filepath.Join(dir, "creds")); err != nil {
		t.Skipf("hard links unavailable here: %v", err)
	}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	for _, h := range []string{host, replica} {
		if res[h].Outcome != "migrated" {
			t.Fatalf("%s: %+v", h, res[h])
		}
	}
	if f.fences[host] != 0 {
		t.Fatal("fence still installed after the move")
	}
	if _, err := os.Lstat(filepath.Join(dir, "creds")); !errors.Is(err, os.ErrNotExist) {
		t.Errorf("the planted link is still in the instance directory: %v", err)
	}
	q, err := os.ReadDir(asideDir(p.DataDir, host))
	if err != nil || len(q) != 1 || !strings.HasPrefix(q[0].Name(), "creds.q-") {
		t.Fatalf("quarantine = %v %v", q, err)
	}
	if n := p.Note(host); !strings.Contains(n, "kept aside") || !strings.Contains(n, "creds.q-") {
		t.Errorf("note = %q", n)
	}
	if _, held := p.Held(host); held {
		t.Error("a quarantined entry became a hold")
	}
}

// Review G1: a database with a second name is refused before the journal
// is written, and the refusal does NOT let the host run flat again. One
// `ln /data/B.db /data/.x` would otherwise keep a tenant's own move
// refused for ever while it woke flat on every request. The host stays
// fenced and down, status reads the hold, the refusal names the inode and
// the find that lists the other name, a flat intent lifts the fence, and
// once the second name is gone the move goes through from the kept fence.
func TestConvergeHoldsAFlatHostWhoseMoveWasRefusedAfterItsStop(t *testing.T) {
	p, f := box(t)
	other := filepath.Join(p.DataDir, ".x")
	if err := os.Link(provision.DefaultDBPath(p.DataDir, host), other); err != nil {
		t.Skipf("hard links unavailable here: %v", err)
	}
	res := byHost(t, p.Converge(context.Background(), ModeNested))
	r := res[host]
	if r.Outcome != "held" || !strings.Contains(r.Reason, "2 names") || !strings.Contains(r.Reason, "-samefile") || !strings.Contains(r.Reason, "inode") {
		t.Fatalf("linked database: %+v", r)
	}
	assertFlatUntouched(t, p, host)
	if f.fences[host] == 0 {
		t.Fatal("the fence was lifted; the host would wake flat on its next request")
	}
	if why, held := p.Held(host); !held || !strings.Contains(why, "2 names") {
		t.Fatalf("Held = %q %v", why, held)
	}
	// The tenant keeps the link alive: every pass meets the same refusal,
	// and the host stays down each time, without a second fence or lift.
	res = byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "held" || f.fences[host] == 0 {
		t.Fatalf("second pass: %+v fence=%d", res[host], f.fences[host])
	}
	lifts := 0
	for _, e := range f.events {
		if e == "lift "+host {
			lifts++
		}
	}
	if lifts != 0 {
		t.Fatalf("the held fence was lifted and re-installed: %v", f.events)
	}
	// The fleet decides it wants flat after all: the hold ends and the
	// host runs flat, as asked.
	res = byHost(t, p.Converge(context.Background(), ModeFlat))
	if res[host].Outcome != "unchanged" || f.fences[host] != 0 {
		t.Fatalf("flat intent: %+v fence=%d", res[host], f.fences[host])
	}
	if _, held := p.Held(host); held {
		t.Fatal("still held under a flat intent")
	}
	// Back to nested, still linked: held again. Then the operator removes
	// the second name, and the next pass moves the host and releases it.
	if res := byHost(t, p.Converge(context.Background(), ModeNested)); res[host].Outcome != "held" {
		t.Fatalf("nested again: %+v", res[host])
	}
	os.Remove(other)
	res = byHost(t, p.Converge(context.Background(), ModeNested))
	if res[host].Outcome != "migrated" || f.fences[host] != 0 {
		t.Fatalf("after the link is gone: %+v fence=%d", res[host], f.fences[host])
	}
	if _, held := p.Held(host); held {
		t.Fatal("still held after the move went through")
	}
}

// A nested host refused its revert is not held: it is already isolated,
// and running is the safe side of that refusal.
func TestConvergeDoesNotHoldANestedHostRefusedItsRevert(t *testing.T) {
	p, f := box(t)
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "migrated" {
		t.Fatalf("migrate: %+v", r)
	}
	// A database already at the flat path makes the revert refuse.
	if err := os.WriteFile(provision.DefaultDBPath(p.DataDir, host), []byte("stray"), 0o600); err != nil {
		t.Fatal(err)
	}
	r := byHost(t, p.Converge(context.Background(), ModeFlat))[host]
	if r.Outcome != "refused" || !strings.Contains(r.Reason, "two databases") {
		t.Fatalf("revert: %+v", r)
	}
	if f.fences[host] != 0 {
		t.Fatal("a refused revert left the nested host fenced")
	}
}

// Review F2: an awake host is deferred, but not for ever. Once it has been
// seen awake for the grace, the pass stops it as it stops any host and
// moves it; a host that was deferred and then found at the intent forgets
// its clock.
func TestConvergeStopsAnAwakeHostOnceItsGraceIsSpent(t *testing.T) {
	p, f := box(t)
	p.DeferAwake = true
	p.AwakeGrace = time.Hour
	clock := time.Unix(1700000000, 0)
	p.Now = func() time.Time { return clock }
	f.awake = map[string]bool{host: true}

	res := byHost(t, p.Converge(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "deferred" || !strings.Contains(r.Reason, "1h0m0s") {
		t.Fatalf("first pass: %+v", r)
	}
	clock = clock.Add(30 * time.Minute)
	res = byHost(t, p.Converge(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "deferred" || !strings.Contains(r.Reason, "30m0s") {
		t.Fatalf("half way: %+v", r)
	}
	for _, e := range f.events {
		if e == "fence "+host {
			t.Fatalf("the awake host was fenced inside its grace: %v", f.events)
		}
	}
	clock = clock.Add(30 * time.Minute)
	res = byHost(t, p.Converge(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "migrated" {
		t.Fatalf("grace spent: %+v", r)
	}
	if got := rowAt(t, p, host); got != endpoints(t, p, host, ModeNested) {
		t.Fatalf("row = %+v", got)
	}
	// The move went through the fence, so the running tenant was stopped
	// for it, and released after.
	fenced, lifted := false, false
	for _, e := range f.events {
		fenced = fenced || e == "fence "+host
		lifted = lifted || e == "lift "+host
	}
	if !fenced || !lifted {
		t.Fatalf("events = %v, want a fence and a lift for %s", f.events, host)
	}
	if len(p.awakeSince) != 0 {
		t.Errorf("awake clocks still held: %v", p.awakeSince)
	}
}

// Landing follow-up 2: the hold survives an agent restart. While a host is
// held its marker sits in the layout directory; a fresh pass (a new agent)
// re-derives the hold from the files before anything wakes, re-holds while
// the planted link persists, and drops the marker once the link is gone
// or the intent is not nested. A marker the tenant deleted costs only one
// grace, since the next pass meets the same refusal.
func TestRestoreHoldsReholdsFromTheMarkerWhileTheReasonPersists(t *testing.T) {
	p, f := box(t)
	other := filepath.Join(p.DataDir, ".x")
	if err := os.Link(provision.DefaultDBPath(p.DataDir, host), other); err != nil {
		t.Skipf("hard links unavailable here: %v", err)
	}
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "held" {
		t.Fatalf("first pass: %+v", r)
	}
	marker := filepath.Join(p.DataDir, "layout", host+".hold")
	if b, err := os.ReadFile(marker); err != nil || !strings.Contains(string(b), "2 names") {
		t.Fatalf("hold marker: %q %v", b, err)
	}
	if hosts, _ := JournalHosts(p.DataDir); len(hosts) != 0 {
		t.Fatalf("a hold marker read as a journal: %v", hosts)
	}

	// The agent restarts: a new pass over the same registry and files, a
	// new fence (every tenant is stopped at start).
	restart := func() (*Pass, *fakeFence) {
		f2 := &fakeFence{}
		p2 := &Pass{Registry: p.Registry, Fence: f2, DataDir: p.DataDir, SocketDir: p.SocketDir, SocketsGID: -1,
			Now: time.Now, SweptAtStart: true, Logf: t.Logf}
		return p2, f2
	}
	p2, f2 := restart()
	res := byHost(t, p2.RestoreHolds(context.Background(), ModeNested))
	if r := res[host]; r.Outcome != "held" || !strings.Contains(r.Reason, "restored at start") {
		t.Fatalf("restore: %+v", r)
	}
	if f2.fences[host] == 0 {
		t.Fatal("the restored hold installed no fence; the host would wake flat")
	}
	if why, held := p2.Held(host); !held || !strings.Contains(why, "2 names") {
		t.Fatalf("Held after restore = %q %v", why, held)
	}
	// Unreadable intent is the fail-closed side: held too.
	p3, f3 := restart()
	if r := byHost(t, p3.RestoreHolds(context.Background(), IntentUnreadable))[host]; r.Outcome != "held" || f3.fences[host] == 0 {
		t.Fatalf("unreadable intent: %+v fence=%d", r, f3.fences[host])
	}
	// A flat intent drops the marker and holds nothing.
	p4, f4 := restart()
	if r := byHost(t, p4.RestoreHolds(context.Background(), ModeFlat))[host]; r.Outcome != "released" || f4.fences[host] != 0 {
		t.Fatalf("flat intent: %+v fence=%d", r, f4.fences[host])
	}
	if _, err := os.Lstat(marker); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("marker survived a flat intent: %v", err)
	}
	// Held again, then the operator removes the link: at the next start
	// the marker's reason is gone, so it is dropped and nothing is held;
	// the first pass then moves the host.
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "held" {
		t.Fatalf("re-hold: %+v", r)
	}
	os.Remove(other)
	p5, f5 := restart()
	if r := byHost(t, p5.RestoreHolds(context.Background(), ModeNested))[host]; r.Outcome != "released" || f5.fences[host] != 0 {
		t.Fatalf("reason gone: %+v fence=%d", r, f5.fences[host])
	}
	if _, err := os.Lstat(marker); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("marker survived its reason: %v", err)
	}
	if r := byHost(t, p5.Converge(context.Background(), ModeNested))[host]; r.Outcome != "migrated" {
		t.Fatalf("after restore: %+v", r)
	}
	_ = f
}

// A hold released by the move going through takes its marker with it.
func TestHoldMarkerIsRemovedWhenTheHostMoves(t *testing.T) {
	p, _ := box(t)
	other := filepath.Join(p.DataDir, ".x")
	if err := os.Link(provision.DefaultDBPath(p.DataDir, host), other); err != nil {
		t.Skipf("hard links unavailable here: %v", err)
	}
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "held" {
		t.Fatalf("hold: %+v", r)
	}
	marker := filepath.Join(p.DataDir, "layout", host+".hold")
	if _, err := os.Lstat(marker); err != nil {
		t.Fatal(err)
	}
	os.Remove(other)
	if r := byHost(t, p.Converge(context.Background(), ModeNested))[host]; r.Outcome != "migrated" {
		t.Fatalf("after cleanup: %+v", r)
	}
	if _, err := os.Lstat(marker); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("marker survived the move: %v", err)
	}
}

// The aside directory is carlos 0700 like the rest of the layout
// directory, so a flat tenant can plant a symlink at it, or a file at a
// sibling's directory under it, and a refusal there would hold the
// sibling (or every host) down with one command. The blocker is renamed
// to a random name under /data/layout and the move goes on.
func TestConvergeDisplacesABlockerAtTheAsideDirectory(t *testing.T) {
	for _, tc := range []struct {
		name  string
		plant func(t *testing.T, dd string)
	}{
		{"symlink at aside/", func(t *testing.T, dd string) {
			if err := os.MkdirAll(filepath.Join(dd, "layout"), 0o700); err != nil {
				t.Fatal(err)
			}
			if err := os.Symlink(t.TempDir(), filepath.Join(dd, "layout", "aside")); err != nil {
				t.Fatal(err)
			}
		}},
		{"file at aside/<host>", func(t *testing.T, dd string) {
			if err := os.MkdirAll(filepath.Join(dd, "layout", "aside"), 0o700); err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(filepath.Join(dd, "layout", "aside", host), []byte("planted"), 0o600); err != nil {
				t.Fatal(err)
			}
		}},
	} {
		t.Run(tc.name, func(t *testing.T) {
			p, f := box(t)
			tc.plant(t, p.DataDir)
			// Something for the move to set aside, so the aside directory
			// is needed: a planted entry in the host's future directory.
			dir := rawInstanceDir(p.DataDir, host)
			if err := os.MkdirAll(dir, 0o700); err != nil {
				t.Fatal(err)
			}
			if err := os.WriteFile(filepath.Join(dir, "x"), []byte("x"), 0o600); err != nil {
				t.Fatal(err)
			}
			r := byHost(t, p.Converge(context.Background(), ModeNested))[host]
			if r.Outcome != "migrated" {
				t.Fatalf("result = %+v", r)
			}
			if f.fences[host] != 0 {
				t.Fatal("fence left installed")
			}
			fi, err := os.Lstat(asideDir(p.DataDir, host))
			if err != nil || !fi.IsDir() {
				t.Fatalf("aside/<host> = %v %v, want a real directory", fi, err)
			}
			q, err := os.ReadDir(asideDir(p.DataDir, host))
			if err != nil || len(q) != 1 || !strings.HasPrefix(q[0].Name(), "x.q-") {
				t.Fatalf("quarantine = %v %v", q, err)
			}
			entries, err := os.ReadDir(filepath.Join(p.DataDir, "layout"))
			if err != nil {
				t.Fatal(err)
			}
			blockers := 0
			for _, e := range entries {
				if strings.Contains(e.Name(), ".blocker-") {
					blockers++
				}
			}
			if blockers != 1 {
				t.Fatalf("layout dir = %v, want the one displaced blocker", entries)
			}
			if _, held := p.Held(host); held {
				t.Fatal("the blocker became a hold")
			}
		})
	}
}