//go:build linux

package host

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

	"github.com/carlosframework/platform/internal/hostbox"
	"github.com/carlosframework/platform/internal/hostguard"
	"github.com/carlosframework/platform/internal/instancectl"
	"github.com/carlosframework/platform/internal/provision"
	"github.com/carlosframework/platform/internal/rootstate"
	"golang.org/x/sys/unix"
)

// This fixture exercises the durable driver, with explicit seams for root UID,
// nosymfollow and the outer claim/host-guard authorization. It is not a
// privileged mount or production transport witness.
type servingParkDiskFixture struct {
	t      *testing.T
	ledger *servingParkLedger
	fs     *servingParkFS
	disk   *rollbackDisk
	data   string
	now    time.Time
	base   servingArtifactIdentity
	target hostguard.FenceTarget
}

func newServingParkDiskFixture(t *testing.T) *servingParkDiskFixture {
	t.Helper()
	rf, _, _, in := servingRetirementFixture(t)
	admitted := servingAdmit(t, rf, &in, "G")
	receipt := servingStop(t, rf, admitted)
	base, err := servingWriterIdentity(admitted, "box-1", "ie")
	if err != nil {
		t.Fatal(err)
	}
	disk := &rollbackDisk{dir: t.TempDir()}
	fs := newServingParkFS(rf.guard.DataDir, disk)
	fs.volume = func(uint64) string { return "vol0123456789abcdef0" }
	fs.owner = uint32(os.Geteuid())
	fs.mount = func(string) (bool, error) { return true, nil }
	f := &servingParkDiskFixture{t: t, fs: fs, disk: disk, data: rf.guard.DataDir, now: time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC), base: base, target: receipt.Target}
	f.ledger = &servingParkLedger{state: disk, fs: fs, journal: rf.journal, now: func() time.Time { return f.now }}
	return f
}

func (f *servingParkDiskFixture) evidence(n uint64, legacy bool, state string) (servingParkAuthorization, []hostguard.FenceReceipt) {
	t := f.t
	t.Helper()
	prior, next := f.base, f.base
	prior.Stamp, prior.Session, prior.Generation = n, n, fmt.Sprintf("G%d", n)
	next.Stamp, next.Session, next.Generation = n+1, n+1, fmt.Sprintf("G%d", n+1)
	target := f.target
	target.Stamp, target.Session, target.FenceID = n, n, fmt.Sprintf("stop-%d", n)
	s := StopReceipt{V: 1, ID: target.FenceID, Kind: StopReceiptStopped, Region: target.Region, ClaimETag: target.ClaimETag, ClaimGen: target.ClaimGen, FenceID: target.FenceID, Host: target.Host, Incarnation: target.Incarnation, InstallationStamp: target.Stamp, WriterSession: target.Session, Box: target.Box, Enrollment: target.Enrollment, BootID: target.BootID, ControllerID: target.ControllerID, DirectoryRevision: testReceiptNumeral(target), Binding: target.DirectoryRevision, EvidenceAt: f.now}
	receipts := []hostguard.FenceReceipt{{Target: target, Bytes: mustMarshalStopReceipt(s), EvidenceAt: f.now}}
	if state != "missing" && !legacy {
		if err := f.ledger.journal.Prelaunch(t.Context(), hostbox.LaunchInventory{Target: target, ControllerPID: 77, ProcessStart: "controller-start", NoState: true, Groups: []hostbox.LaunchGroup{{RelativePath: "carlos/parent", Device: 1, Inode: 2}}}); err != nil {
			t.Fatal(err)
		}
		if state != "prelaunch" {
			res := hostbox.LaunchReservation{ID: fmt.Sprintf("%d:main", n), Role: "main", InvocationID: fmt.Sprintf("invocation-%d", n), Inputs: "inputs", Target: target, Group: hostbox.LaunchGroup{RelativePath: instancectl.ClaimUnitControlGroup(target.Host)}, Executable: hostbox.LaunchFile{Device: 1, Inode: 3}}
			if err := f.ledger.journal.IntendUnit(t.Context(), res); err != nil {
				t.Fatal(err)
			}
			if state == "started" {
				group := hostbox.LaunchGroup{RelativePath: res.Group.RelativePath, Device: 1, Inode: 4, InvocationID: fmt.Sprintf("unit-%d", n)}
				if err := f.ledger.journal.StartedUnit(t.Context(), res, group, 99, "tenant-start"); err != nil {
					t.Fatal(err)
				}
			}
			if state != "intended" {
				if err := f.ledger.journal.SettleUnit(t.Context(), target, res.ID); err != nil {
					t.Fatal(err)
				}
			}
		}
	}
	source, err := os.Open(filepath.Dir(provision.NestedDBPath(f.data, next.Host)))
	if err != nil {
		t.Fatal(err)
	}
	id, err := servingParkDirID(source)
	source.Close()
	if err != nil {
		t.Fatal(err)
	}
	a := servingParkAuthorization{TagsFrom: n, Predecessor: prior, OriginalSuccessor: next, Successor: next, Directory: id, Legacy: legacy}
	if legacy {
		a.Predecessor = servingArtifactIdentity{}
		a.TagsFrom = 0
	}
	a.ID = servingRetirementID(a.Predecessor, next, legacy)
	return a, receipts
}
func (f *servingParkDiskFixture) put(suffix, body string) {
	f.t.Helper()
	if err := os.WriteFile(provision.NestedDBPath(f.data, f.base.Host)+suffix, []byte(body), 0600); err != nil {
		f.t.Fatal(err)
	}
}
func (f *servingParkDiskFixture) park(a servingParkAuthorization, receipts []hostguard.FenceReceipt) (servingParkRecord, error) {
	var r servingParkRecord
	err := f.ledger.withNamespace(f.t.Context(), func() error { var err error; r, err = f.ledger.parkLocked(f.t.Context(), a, receipts); return err })
	return r, err
}
func (f *servingParkDiskFixture) complete(a servingParkAuthorization) {
	f.t.Helper()
	if err := f.ledger.withNamespace(f.t.Context(), func() error { return f.ledger.completeLocked(f.t.Context(), a.ID, a.Successor) }); err != nil {
		f.t.Fatal(err)
	}
}
func (f *servingParkDiskFixture) path(a servingParkAuthorization, suffix string) string {
	return filepath.Join(f.data, "instances", ".serving-parked", a.ID, a.Successor.Host+".db"+suffix)
}
func (f *servingParkDiskFixture) retained(a servingParkAuthorization, body string, deadline time.Time) {
	f.t.Helper()
	slot, err := f.ledger.slot(a)
	if err != nil || slot.ID != a.ID {
		f.t.Fatalf("park owner changed: %+v %v", slot, err)
	}
	b, err := os.ReadFile(f.path(a, ""))
	if err != nil || string(b) != body {
		f.t.Fatalf("retained database: %q %v", b, err)
	}
	r, err := f.ledger.read(a.ID)
	if err != nil || !r.ReapAfter.Equal(deadline) {
		f.t.Fatalf("park deadline changed: %v %v", r.ReapAfter, err)
	}
}

func TestServingParkLedgerKeepsLastStartedDatabase(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy writes")
	f.put("-wal", "legacy WAL")
	f.put("-shm", "legacy SHM")
	legacy, stops := f.evidence(10, true, "missing")
	parked, err := f.park(legacy, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(legacy)
	for _, suffix := range []string{"-wal", "-shm"} {
		if b, e := os.ReadFile(f.path(legacy, suffix)); e != nil || len(b) == 0 {
			t.Fatalf("companion not preserved: %s %v", suffix, e)
		}
	}
	for n := uint64(11); n < 179; n++ {
		f.now = f.now.Add(time.Hour)
		// Alternate a failed restore with no database and a successful restore
		// whose tenant never started. Neither owns the retention slot.
		if n%2 == 0 {
			f.put("", "unlaunched baseline")
		}
		f.put(".gen", fmt.Sprintf("G%d", n))
		f.put("-wal", "companion")
		next, receipts := f.evidence(n, false, "prelaunch")
		r, e := f.park(next, receipts)
		if e != nil {
			t.Fatalf("retirement %d: %v", n, e)
		}
		if !r.Closed || r.Disposition == servingParkRetain {
			t.Fatalf("unlaunched source retained: %+v", r)
		}
		f.complete(next)
		f.retained(legacy, "legacy writes", parked.ReapAfter)
	}
	f.put("", "tenant writes")
	next, receipts := f.evidence(179, false, "started")
	r, err := f.park(next, receipts)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(next)
	if _, err := os.Stat(f.path(legacy, "")); !errors.Is(err, os.ErrNotExist) {
		t.Fatalf("legacy not replaced: %v", err)
	}
	f.retained(next, "tenant writes", r.ReapAfter)
	// A tombstoned replay cannot replace the current slot or reset its clock.
	if _, err := f.park(legacy, stops); err != nil {
		t.Fatal(err)
	}
	f.complete(legacy)
	f.retained(next, "tenant writes", r.ReapAfter)
}

func TestServingParkLedgerAmbiguousLaunchPreservesBothCopies(t *testing.T) {
	for _, state := range []string{"missing", "intended", "settled-unobserved"} {
		t.Run(state, func(t *testing.T) {
			f := newServingParkDiskFixture(t)
			f.put("", "legacy")
			legacy, stops := f.evidence(10, true, "missing")
			park, err := f.park(legacy, stops)
			if err != nil {
				t.Fatal(err)
			}
			f.complete(legacy)
			f.put("", "uncertain tenant writes")
			next, receipts := f.evidence(11, false, state)
			if _, err := f.park(next, receipts); err == nil {
				t.Fatal("ambiguous launch accepted")
			}
			f.retained(legacy, "legacy", park.ReapAfter)
			b, err := os.ReadFile(provision.NestedDBPath(f.data, f.base.Host))
			if err != nil || string(b) != "uncertain tenant writes" {
				t.Fatalf("source changed: %q %v", b, err)
			}
		})
	}
}

func TestServingParkLedgerExpiryAndIdentity(t *testing.T) {
	for _, attack := range []string{"none", "inode", "hardlink", "symlink", "unknown child", "event", "root", "parent", "missing", "mount"} {
		t.Run(attack, func(t *testing.T) {
			f := newServingParkDiskFixture(t)
			f.put("", "parked")
			a, stops := f.evidence(10, true, "missing")
			r, err := f.park(a, stops)
			if err != nil {
				t.Fatal(err)
			}
			f.complete(a)
			f.now = r.ReapAfter.Add(-time.Nanosecond)
			reap := func() error {
				return f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) })
			}
			if err := reap(); err == nil {
				t.Fatal("reaped before deadline")
			}
			f.now = r.ParkedAt.Add(-time.Hour)
			if err := reap(); err == nil {
				t.Fatal("clock rollback reaped park")
			}
			f.now = r.ReapAfter
			path := f.path(a, "")
			victim := filepath.Join(t.TempDir(), "victim")
			if err := os.WriteFile(victim, []byte("victim"), 0600); err != nil {
				t.Fatal(err)
			}
			switch attack {
			case "inode":
				if err := os.Rename(path, path+".saved"); err != nil {
					t.Fatal(err)
				}
				if err := os.WriteFile(path, []byte("victim"), 0600); err != nil {
					t.Fatal(err)
				}
				if err := os.Rename(path+".saved", filepath.Join(t.TempDir(), "original")); err != nil {
					t.Fatal(err)
				}
			case "hardlink":
				if err := os.Link(path, path+".link"); err != nil {
					t.Fatal(err)
				}
			case "symlink":
				if err := os.Remove(path); err != nil {
					t.Fatal(err)
				}
				if err := os.Symlink(victim, path); err != nil {
					t.Fatal(err)
				}
			case "unknown child":
				if err := os.WriteFile(path+".unknown", []byte("keep"), 0600); err != nil {
					t.Fatal(err)
				}
			case "event", "root", "parent":
				dir := filepath.Dir(path)
				if attack == "root" {
					dir = filepath.Dir(dir)
				}
				if attack == "parent" {
					dir = filepath.Dir(filepath.Dir(dir))
				}
				if err := os.Rename(dir, dir+".saved"); err != nil {
					t.Fatal(err)
				}
				if err := os.Mkdir(dir, 0700); err != nil {
					t.Fatal(err)
				}
			case "missing":
				if err := os.Remove(path); err != nil {
					t.Fatal(err)
				}
			case "mount":
				f.fs.mount = func(string) (bool, error) { return false, nil }
			}
			err = reap()
			if attack == "none" {
				if err != nil {
					t.Fatal(err)
				}
				if _, e := os.Stat(path); !errors.Is(e, os.ErrNotExist) {
					t.Fatalf("not reaped: %v", e)
				}
			} else if err == nil {
				t.Fatal("unsafe reaping accepted")
			}
			if b, e := os.ReadFile(victim); e != nil || string(b) != "victim" {
				t.Fatalf("victim changed: %q %v", b, e)
			}
		})
	}
}

func TestServingParkLedgerNoReplaceAndCrossMountRefusal(t *testing.T) {
	for _, failure := range []error{unix.EXDEV, unix.EEXIST} {
		t.Run(failure.Error(), func(t *testing.T) {
			f := newServingParkDiskFixture(t)
			f.put("", "source")
			a, stops := f.evidence(10, true, "missing")
			f.fs.rename = func(int, string, int, string, uint) error { return failure }
			if _, err := f.park(a, stops); !errors.Is(err, failure) {
				t.Fatalf("rename result: %v", err)
			}
			b, err := os.ReadFile(provision.NestedDBPath(f.data, f.base.Host))
			if err != nil || string(b) != "source" {
				t.Fatalf("rename fallback touched source: %q %v", b, err)
			}
			f.fs.rename = unix.Renameat2
			if _, err := f.park(a, stops); err != nil {
				t.Fatal(err)
			}
		})
	}
}

func TestServingParkLedgerLostRenameAndDeleteReplies(t *testing.T) {
	for _, effect := range []string{"rename", "sync", "unlink"} {
		for boundary := 1; boundary <= 4; boundary++ {
			t.Run(fmt.Sprintf("%s-%d", effect, boundary), func(t *testing.T) {
				f := newServingParkDiskFixture(t)
				for _, suffix := range []string{"", "-wal", "-shm"} {
					f.put(suffix, "source"+suffix)
				}
				a, stops := f.evidence(10, true, "missing")
				calls := 0
				fail := func(err error) error {
					if err != nil {
						return err
					}
					calls++
					if calls == boundary {
						return os.ErrPermission
					}
					return nil
				}
				if effect == "rename" {
					f.fs.rename = func(a int, b string, c int, d string, e uint) error { return fail(unix.Renameat2(a, b, c, d, e)) }
				}
				if effect == "sync" {
					f.fs.sync = func(fd int) error { return fail(unix.Fsync(fd)) }
				}
				_, _ = f.park(a, stops)
				f.fs.rename, f.fs.sync = unix.Renameat2, unix.Fsync
				// A failure before namespace.json exists is deliberately an obstruction;
				// quarantine recovery (still unbound) must resolve it, never adoption.
				if _, err := f.disk.Read("namespace.json"); errors.Is(err, os.ErrNotExist) {
					if _, err := f.park(a, stops); err == nil {
						t.Fatal("unrecorded namespace adopted")
					}
					return
				}
				r, err := f.park(a, stops)
				if err != nil {
					t.Fatal(err)
				}
				f.complete(a)
				for _, suffix := range []string{"", "-wal", "-shm"} {
					if b, e := os.ReadFile(f.path(a, suffix)); e != nil || string(b) != "source"+suffix {
						t.Fatalf("parked bytes: %q %v", b, e)
					}
				}
				f.now = r.ReapAfter
				if effect == "unlink" {
					f.fs.unlink = func(fd int, n string, flags int) error { return fail(unix.Unlinkat(fd, n, flags)) }
				}
				_ = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) })
				f.fs.unlink = unix.Unlinkat
				if err := f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) }); err != nil {
					t.Fatal(err)
				}
			})
		}
	}
}

func TestServingParkLedgerPendingCannotBeReplaced(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	a, stops := f.evidence(10, true, "missing")
	r, err := f.park(a, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.put("", "new tenant writes")
	next, receipts := f.evidence(11, false, "started")
	if _, err := f.park(next, receipts); err == nil {
		t.Fatal("pending park replaced")
	}
	f.retained(a, "legacy", r.ReapAfter)
	if err := os.Remove(provision.NestedDBPath(f.data, f.base.Host)); err != nil {
		t.Fatal(err)
	}
	f.complete(a)
	f.put("", "new tenant writes")
	if _, err := f.park(next, receipts); err != nil {
		t.Fatal(err)
	}
}

func TestServingParkLedgerCancelIsEffectFree(t *testing.T) {
	f := newServingParkDiskFixture(t)
	ctx, cancel := context.WithCancel(t.Context())
	cancel()
	called := false
	if err := f.ledger.withNamespace(ctx, func() error { called = true; return nil }); !errors.Is(err, context.Canceled) || called {
		t.Fatalf("cancel: %v, called=%v", err, called)
	}
}

func TestServingParkLedgerDiscardResumesAfterUnlink(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	old, stops := f.evidence(10, true, "missing")
	r, err := f.park(old, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(old)
	f.put("", "unlaunched")
	next, receipts := f.evidence(11, false, "prelaunch")
	f.fs.unlink = func(fd int, n string, flags int) error {
		if err := unix.Unlinkat(fd, n, flags); err != nil {
			return err
		}
		return os.ErrPermission
	}
	if _, err := f.park(next, receipts); err == nil {
		t.Fatal("lost unlink reply succeeded")
	}
	f.retained(old, "legacy", r.ReapAfter)
	f.fs.unlink = unix.Unlinkat
	if _, err := f.park(next, receipts); err != nil {
		t.Fatal(err)
	}
	f.complete(next)
	f.retained(old, "legacy", r.ReapAfter)
}

func TestServingParkLedgerSourceReplacementDoesNotDeleteOldSlot(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	old, stops := f.evidence(10, true, "missing")
	r, err := f.park(old, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(old)
	f.put("", "served")
	next, receipts := f.evidence(11, false, "started")
	// Interrupt just after the new source was captured and its intent saved.
	f.disk.afterWrite = func(name string, _ []byte) {
		if name == "retirement-"+next.ID+".json" {
			panic("crash")
		}
	}
	func() {
		defer func() {
			if recover() == nil {
				t.Fatal("crash not reached")
			}
		}()
		_, _ = f.park(next, receipts)
	}()
	f.disk.afterWrite = nil
	source := provision.NestedDBPath(f.data, f.base.Host)
	if err := os.Rename(source, source+".saved"); err != nil {
		t.Fatal(err)
	}
	f.put("", "replacement")
	if _, err := f.park(next, receipts); err == nil {
		t.Fatal("replaced source accepted")
	}
	f.retained(old, "legacy", r.ReapAfter)
}

func TestServingParkLedgerChecksAllArtifactsBeforeDeletion(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "database")
	f.put("-shm", "shared memory")
	a, stops := f.evidence(10, true, "missing")
	r, err := f.park(a, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(a)
	if err := os.Remove(f.path(a, "-shm")); err != nil {
		t.Fatal(err)
	}
	f.now = r.ReapAfter
	if err := f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) }); err == nil {
		t.Fatal("unrecorded absence accepted")
	}
	if b, e := os.ReadFile(f.path(a, "")); e != nil || string(b) != "database" {
		t.Fatalf("partial validation deleted database: %q %v", b, e)
	}
}

func TestServingParkLedgerCompletionRechecksSource(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	a, stops := f.evidence(10, true, "missing")
	_, err := f.park(a, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.put("", "replacement")
	if err := f.ledger.withNamespace(t.Context(), func() error { return f.ledger.completeLocked(t.Context(), a.ID, a.Successor) }); err == nil {
		t.Fatal("completed with replacement source")
	}
	r, err := f.ledger.read(a.ID)
	if err != nil || r.Complete {
		t.Fatalf("completion persisted: %v %v", r.Complete, err)
	}
}

func TestServingParkLedgerLegacyConsumptionSurvivesReaping(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	a, stops := f.evidence(10, true, "missing")
	r, err := f.park(a, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(a)
	f.now = r.ReapAfter
	if err := f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) }); err != nil {
		t.Fatal(err)
	}
	// Even removal of the disposable retirement tombstone does not reauthorize
	// a different legacy retirement for this installation/incarnation.
	if err := f.disk.Remove("retirement-" + a.ID + ".json"); err != nil {
		t.Fatal(err)
	}
	slot, err := f.ledger.slot(a)
	if err == nil {
		t.Fatal("slot accepted missing tombstone")
	}
	// Model future slot/tombstone GC; the durable consumed-source record stays.
	slot = servingParkSlot{Version: 1, Box: a.Successor.Box, Host: a.Successor.Host}
	if err := f.ledger.write("slot-"+servingParkKey(slot.Box, slot.Host)+".json", slot); err != nil {
		t.Fatal(err)
	}
	f.put("", "restored successor")
	next, receipts := f.evidence(11, true, "missing")
	if _, err := f.park(next, receipts); err == nil {
		t.Fatal("successor parked as legacy")
	}
}

type servingParkFaultStore struct {
	rootstate.Store
	count, failAt int
	before        bool
}

func (s *servingParkFaultStore) Write(name string, body []byte) error {
	s.count++
	if s.count == s.failAt && s.before {
		return os.ErrPermission
	}
	if err := s.Store.Write(name, body); err != nil {
		return err
	}
	if s.count == s.failAt {
		return os.ErrPermission
	}
	return nil
}

func TestServingParkLedgerDurableWriteFailures(t *testing.T) {
	for _, before := range []bool{false, true} {
		for boundary := 1; boundary <= 24; boundary++ {
			t.Run(fmt.Sprintf("before-%t-boundary-%d", before, boundary), func(t *testing.T) {
				f := newServingParkDiskFixture(t)
				f.put("", "database")
				f.put("-wal", "wal")
				f.put("-shm", "shm")
				a, stops := f.evidence(10, true, "missing")
				faults := &servingParkFaultStore{Store: f.disk, failAt: boundary, before: before}
				f.ledger.state, f.fs.state = faults, faults
				r, err := f.park(a, stops)
				if err == nil {
					err = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.completeLocked(t.Context(), a.ID, a.Successor) })
				}
				if err == nil {
					f.now = r.ReapAfter
					err = f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) })
				}
				f.ledger.state, f.fs.state = f.disk, f.disk
				if _, readErr := f.disk.Read("namespace.json"); os.IsNotExist(readErr) {
					if _, statErr := os.Stat(filepath.Join(f.data, "instances", ".serving-parked")); statErr == nil {
						if _, err := f.park(a, stops); err == nil {
							t.Fatal("unknown root adopted after write failure")
						}
						return
					}
				}
				r, err = f.park(a, stops)
				if err != nil {
					t.Fatalf("park replay: %v", err)
				}
				f.complete(a)
				f.now = r.ReapAfter
				if err := f.ledger.withNamespace(t.Context(), func() error { return f.ledger.reapLocked(t.Context(), a.ID) }); err != nil {
					t.Fatalf("reap replay: %v", err)
				}
				got, err := f.ledger.read(a.ID)
				if err != nil || !got.Closed || !got.Complete {
					t.Fatalf("unfinished replay: %+v %v", got, err)
				}
			})
		}
	}
}

func TestServingParkLedgerSlotsAcrossIncarnationsAndHosts(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "first incarnation")
	old, stops := f.evidence(10, true, "missing")
	_, err := f.park(old, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(old)
	f.base.Incarnation, f.target.Incarnation = "next-incarnation", "next-incarnation"
	f.put("", "next incarnation")
	next, receipts := f.evidence(11, false, "started")
	r, err := f.park(next, receipts)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(next)
	if _, err := os.Stat(f.path(old, "")); !os.IsNotExist(err) {
		t.Fatalf("old incarnation remains: %v", err)
	}
	f.base.Host, f.target.Host = "other.example", "other.example"
	if err := os.MkdirAll(filepath.Dir(provision.NestedDBPath(f.data, f.base.Host)), 0700); err != nil {
		t.Fatal(err)
	}
	f.put("", "other host")
	other, receipts := f.evidence(12, true, "missing")
	if _, err := f.park(other, receipts); err != nil {
		t.Fatal(err)
	}
	f.complete(other)
	f.retained(next, "next incarnation", r.ReapAfter)
}

func TestServingParkLedgerDiscardPendingBlocksNextRetirement(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	old, stops := f.evidence(10, true, "missing")
	r, err := f.park(old, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(old)
	f.put("", "unlaunched")
	pending, receipts := f.evidence(11, false, "prelaunch")
	if _, err := f.park(pending, receipts); err != nil {
		t.Fatal(err)
	}
	f.put("", "next tenant writes")
	next, receipts := f.evidence(12, false, "started")
	if _, err := f.park(next, receipts); err == nil {
		t.Fatal("unfinished discard overtaken")
	}
	f.retained(old, "legacy", r.ReapAfter)
}

func TestServingParkLedgerRefusesPrematureTombstone(t *testing.T) {
	f := newServingParkDiskFixture(t)
	f.put("", "legacy")
	a, stops := f.evidence(10, true, "missing")
	r, err := f.park(a, stops)
	if err != nil {
		t.Fatal(err)
	}
	f.complete(a)
	r.Complete, r.Deleting, r.Closed, r.Cause = true, true, true, "expiry"
	if err := f.ledger.write("retirement-"+a.ID+".json", r); err != nil {
		t.Fatal(err)
	}
	if _, err := f.ledger.read(a.ID); err == nil {
		t.Fatal("incomplete deletion accepted as a tombstone")
	}
	if b, e := os.ReadFile(f.path(a, "")); e != nil || string(b) != "legacy" {
		t.Fatalf("malformed record changed data: %q %v", b, e)
	}
}