package park

import (
	"context"
	"errors"
	"fmt"
	"net/url"
	"reflect"
	"strings"
	"testing"
	"time"

	"github.com/carlosframework/platform/internal/hostclaim"
	"github.com/carlosframework/platform/internal/lease"
	"github.com/carlosframework/platform/internal/lease/leasetest"
	"github.com/carlosframework/platform/internal/release"
)

type testLease struct {
	released int
	failed   bool
}

func (l *testLease) Check(context.Context) error {
	if l.failed {
		return ErrChanged
	}
	return nil
}
func (l *testLease) Release(context.Context) error { l.released++; return nil }
func (*testLease) Close()                          {}

type testLeases struct{ held *testLease }

func (l testLeases) Acquire(context.Context, string, string) (HeldLease, error) { return l.held, nil }

type fixture struct {
	e    Engine
	b    *leasetest.VersionedBucket
	w    Work
	now  time.Time
	ob   hostclaim.Obligation
	held *testLease
}

func newFixture(t *testing.T) *fixture {
	t.Helper()
	ctx := t.Context()
	now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
	host := "shop.bab.example.com"
	inc, err := hostclaim.Incarnation("bab", "jam", host, now)
	if err != nil {
		t.Fatal(err)
	}
	b := leasetest.NewVersionedBucket()
	b.AddBucket("data")
	b.PageSize = 1
	f := &fixture{b: b, w: Work{host, "ob", "op"}, now: now.Add(16 * time.Minute), held: &testLease{}}
	f.ob = hostclaim.Obligation{ID: "ob", Op: "op", Kind: "park-replica", Region: "ie", Retirement: "rt1", Owner: "region:ie", Park: &hostclaim.ParkScope{Location: "instances/" + host + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: time.Second}}
	st := release.NewDirStore(t.TempDir())
	_, err = hostclaim.Update(ctx, st, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
		c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Incarnation: inc, CreatedAt: now, Location: f.ob.Park.Location, Stamp: 1}
		c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}
		c.Obligations = []hostclaim.Obligation{f.ob}
		c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	serial := 1
	f.e = Engine{Store: b, Claims: st, Bucket: "data", Region: "ie", Leases: testLeases{f.held}, Now: func() time.Time { return f.now }, NewID: func() string { serial++; return fmt.Sprintf("rt%d", serial) }, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }, View: func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
	}}
	return f
}
func (f *fixture) seed() []lease.ObjectVersion {
	p := f.ob.Park.Location + "/"
	f.b.Put("data", p+"0000/a", []byte("old"))
	f.b.Put("data", p+"0000/a", []byte("new"))
	f.b.BareDelete("data", p+"0000/a")
	f.b.Put("data", p+"lock.json", []byte("lease"))
	vs, _ := f.e.list(context.Background(), f.w.Host, f.ob.Park.Location)
	return vs
}
func (f *fixture) done(t *testing.T) {
	t.Helper()
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	if len(obs.Claim.Obligations) != 0 {
		t.Fatal("obligation still open")
	}
	if got := obs.Claim.QuietAfter["ie"]; !got.Equal(f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)) {
		t.Fatal(got)
	}
	vs, err := f.e.list(t.Context(), f.w.Host, f.ob.Park.Location)
	if err != nil || len(vs) != 0 {
		t.Fatalf("residue: %v %v", vs, err)
	}
}
func TestRemoteCrashResume(t *testing.T) {
	for _, tc := range []struct {
		step string
		n    int
	}{{"copy", 1}, {"copy", 2}, {"manifest", 1}, {"delete", 1}, {"delete", 2}, {"delete", 3}, {"verify", 1}, {"before-completion", 1}, {"completion", 1}} {
		t.Run(fmt.Sprintf("%s-%d", tc.step, tc.n), func(t *testing.T) {
			f := newFixture(t)
			source := f.seed()
			stop := errors.New("crash")
			n := 0
			f.e.After = func(s string) error {
				if s == tc.step {
					n++
					if n == tc.n {
						return stop
					}
				}
				return nil
			}
			if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
				t.Fatalf("crash: %v", err)
			}
			if tc.step != "completion" && f.held.released != 0 {
				t.Fatal("released before ack")
			}
			f.e.After = nil
			if err := f.e.Run(t.Context(), f.w); err != nil {
				t.Fatal(err)
			}
			f.done(t)
			m, err := f.e.manifest(t.Context(), f.w, f.ob)
			if err != nil || m == nil || len(m.Delete) != len(source) || len(m.Archive) != 2 {
				t.Fatalf("manifest: %#v %v", m, err)
			}
			for _, a := range m.Archive {
				body, err := f.b.GetVersion(t.Context(), "data", a.Destination.Key, a.Destination.VersionID)
				if err != nil || int64(len(body)) != a.Source.Size {
					t.Fatalf("copy: %q %v", body, err)
				}
			}
			if f.b.LatestID("data", f.ob.Park.Location+"/lock.json") == "" {
				t.Fatal("lock deleted")
			}
		})
	}
}
func TestStorageFailuresKeepObligation(t *testing.T) {
	for _, op := range []string{"list", "get", "copy", "put", "delete"} {
		t.Run(op, func(t *testing.T) {
			f := newFixture(t)
			f.seed()
			if op == "get" {
				f.e.After = func(s string) error {
					if s == "manifest" {
						return ErrPending
					}
					return nil
				}
				_ = f.e.Run(t.Context(), f.w)
				f.e.After = nil
			}
			failure := errors.New("injected")
			f.b.FailOn = func(got, key string) error {
				if got == op {
					return failure
				}
				return nil
			}
			if err := f.e.Run(t.Context(), f.w); err == nil {
				t.Fatal("failure acknowledged")
			}
			obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
			if err != nil || len(obs.Claim.Obligations) != 1 || len(obs.Claim.QuietAfter) != 0 {
				t.Fatalf("claim: %v %v", obs, err)
			}
			f.b.FailOn = nil
			if err := f.e.Run(t.Context(), f.w); err != nil {
				t.Fatal(err)
			}
			f.done(t)
		})
	}
}
func TestPageFailureDoesNotAct(t *testing.T) {
	f := newFixture(t)
	f.seed()
	before := f.b.Snapshot()
	f.b.FailListAfterPages = 1
	if err := f.e.Run(t.Context(), f.w); err == nil {
		t.Fatal("short listing accepted")
	}
	if f.b.Snapshot() != before {
		t.Fatal("mutated on listing failure")
	}
}
func TestHorizonLatePut(t *testing.T) {
	f := newFixture(t)
	f.seed()
	deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
	f.now = deadline
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
		t.Fatal(err)
	}
	late := f.b.Put("data", f.ob.Park.Location+"/0000/late", []byte("late"))
	f.now = deadline.Add(time.Nanosecond)
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
		t.Fatal(err)
	}
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if obs.Claim.Obligations[0].Retirement == f.ob.Retirement {
		t.Fatal("manifest reused")
	}
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
	f.done(t)
	if _, err := f.b.GetVersion(t.Context(), "data", f.ob.Park.Location+"/0000/late", late); !errors.Is(err, lease.ErrMirrorNotFound) {
		t.Fatal(err)
	}
}
func TestVerificationListingStartsAfterHorizon(t *testing.T) {
	f := newFixture(t)
	f.seed()
	deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
	f.now = deadline
	f.e.After = func(s string) error {
		if s == "verify" {
			f.now = deadline.Add(time.Hour)
		}
		return nil
	}
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
		t.Fatal(err)
	}
}
func TestManifestRetryDoesNotCopy(t *testing.T) {
	f := newFixture(t)
	f.seed()
	f.e.After = func(s string) error {
		if s == "delete" {
			return ErrPending
		}
		return nil
	}
	_ = f.e.Run(t.Context(), f.w)
	f.e.After = nil
	f.b.FailOn = func(op, key string) error {
		if op == "copy" {
			t.Error("retry copied")
			return ErrUnsafe
		}
		return nil
	}
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
}
func TestCASConflictDoesNotAcknowledge(t *testing.T) {
	f := newFixture(t)
	f.seed()
	f.e.After = func(s string) error {
		if s == "before-completion" {
			obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
			_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Lost = append(c.Lost, "other-box"); return nil })
			return err
		}
		return nil
	}
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, hostclaim.ErrConflict) {
		t.Fatal(err)
	}
	if f.held.released != 0 {
		t.Fatal("released unacknowledged lease")
	}
	f.e.After = nil
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
}
func TestLiveGenerationCannotBeParked(t *testing.T) {
	f := newFixture(t)
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	f.seed()
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
	}
	before := f.b.Snapshot()
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrUnsafe) {
		t.Fatal(err)
	}
	if before != f.b.Snapshot() {
		t.Fatal("live data changed")
	}
}

func TestRemoteMarkerDisagreementIsLocationScoped(t *testing.T) {
	for _, scenario := range []string{"old-incarnation", "live-location", "ancestor-scope", "ancestor-holder", "unknown-region", "unknown-marker", "unknown-generations", "unknown-pinned-generation", "view-error", "authorization-error"} {
		t.Run(scenario, func(t *testing.T) {
			f, obs, view, _ := sweepFixture(t)
			view.MarkerGeneration = "unrelated"
			f.ob.Op = "session:3"
			f.w.Op = f.ob.Op
			if scenario != "live-location" {
				f.ob.Park.Incarnation = strings.Repeat("a", 64)
				f.ob.Park.Location = "instances/" + f.w.Host + "/" + f.ob.Park.Incarnation
			}
			if scenario == "ancestor-scope" {
				f.ob.Park.Location = "instances/" + f.w.Host
			}
			_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
				c.Obligations = []hostclaim.Obligation{f.ob}
				if scenario == "ancestor-holder" {
					c.Holder.Location = "instances/" + f.w.Host
				}
				return nil
			})
			if err != nil {
				t.Fatal(err)
			}
			source := f.seed()
			before := f.b.Snapshot()
			liveBefore, err := f.e.list(t.Context(), f.w.Host, obs.Claim.Holder.Location)
			if err != nil {
				t.Fatal(err)
			}
			switch scenario {
			case "unknown-region":
				view.AuthoritativeRegion = ""
			case "unknown-marker":
				view.MarkerKnown = false
			case "unknown-generations":
				view.GenerationOf = nil
			case "unknown-pinned-generation":
				view.GenerationOf = func(Ref) (string, bool) { return "", false }
			}
			f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
				v := view
				v.ClaimETag = o.ETag
				if scenario == "view-error" {
					return v, ErrUnsafe
				}
				return v, nil
			}
			authorized := false
			f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
				authorized = true
				if !reflect.DeepEqual(o, f.ob) || scenario == "authorization-error" {
					return ErrUnsafe
				}
				return nil
			}
			err = f.e.Run(t.Context(), f.w)
			if !authorized {
				t.Fatal("scope was not authorized")
			}
			if scenario != "old-incarnation" {
				if !errors.Is(err, ErrUnsafe) {
					t.Fatalf("Run = %v, want refusal", err)
				}
				if f.b.Snapshot() != before {
					t.Fatal("refused work changed storage")
				}
				pending, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
				if err != nil {
					t.Fatal(err)
				}
				if len(pending.Claim.Obligations) != 1 || len(pending.Claim.QuietAfter) != 0 || f.held.released != 0 {
					t.Fatal("refused work acknowledged")
				}
				return
			}
			if err != nil {
				t.Fatalf("old incarnation refused by live marker disagreement: %v", err)
			}
			f.done(t)
			m, err := f.e.manifest(t.Context(), f.w, f.ob)
			if err != nil || m == nil || !reflect.DeepEqual(m.Delete, source) || len(m.Archive) != 2 {
				t.Fatalf("old incarnation manifest: %+v %v", m, err)
			}
			for _, copy := range m.Archive {
				body, err := f.b.GetVersion(t.Context(), "data", copy.Destination.Key, copy.Destination.VersionID)
				if err != nil || (string(body) != "old" && string(body) != "new") {
					t.Fatalf("archive: %q %v", body, err)
				}
			}
			liveAfter, err := f.e.list(t.Context(), f.w.Host, obs.Claim.Holder.Location)
			if err != nil || !reflect.DeepEqual(liveBefore, liveAfter) {
				t.Fatalf("live location changed: %v", err)
			}
			if f.held.released != 1 {
				t.Fatal("completed work did not release its lease")
			}
		})
	}
}

func TestBoundScopeAndQuietAfter(t *testing.T) {
	f := newFixture(t)
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	later := f.now.Add(time.Hour)
	_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations[0].Park = nil
		c.QuietAfter = map[string]time.Time{"ie": later}
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if err := f.e.Bind(t.Context(), f.w, *f.ob.Park); err != nil {
		t.Fatal(err)
	}
	bad := *f.ob.Park
	bad.EvidenceAt = bad.EvidenceAt.Add(time.Second)
	if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
		t.Fatal(err)
	}
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
	obs, _ = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if !obs.Claim.QuietAfter["ie"].Equal(later) {
		t.Fatal("quiet_after regressed")
	}
}
func TestBindSurvivesSerialization(t *testing.T) {
	f := newFixture(t)
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations[0].Park = nil
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	scope := *f.ob.Park
	scope.EvidenceAt = time.Now() // Carries a monotonic reading that JSON drops.
	if err := f.e.Bind(t.Context(), f.w, scope); err != nil {
		t.Fatal(err)
	}
	before, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	for _, at := range []time.Time{scope.EvidenceAt, scope.EvidenceAt.UTC()} {
		same := scope
		same.EvidenceAt = at
		if err := f.e.Bind(t.Context(), f.w, same); err != nil {
			t.Fatal(err)
		}
	}
	for _, change := range []func(*hostclaim.ParkScope){
		func(p *hostclaim.ParkScope) { p.Location += "/other" },
		func(p *hostclaim.ParkScope) { p.Incarnation = strings.Repeat("f", 64) },
		func(p *hostclaim.ParkScope) { p.RequestLimit += time.Second },
		func(p *hostclaim.ParkScope) { p.EvidenceAt = p.EvidenceAt.Add(time.Nanosecond) },
	} {
		bad := scope
		change(&bad)
		if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
			t.Fatal(err)
		}
	}
	after, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || after.ETag != before.ETag {
		t.Fatalf("bind changed persisted scope: %v", err)
	}
}

func TestLegacyLocationDoesNotIncludeIncarnationChildren(t *testing.T) {
	f := newFixture(t)
	key := f.ob.Park.Location + "/0000/data"
	f.b.Put("data", key, []byte("new incarnation"))
	vs, err := f.e.list(t.Context(), f.w.Host, "instances/"+f.w.Host)
	if err != nil || len(vs) != 0 {
		t.Fatalf("%v %v", vs, err)
	}
	for _, v := range []string{"a/b", "a@b", "a+b", "a b", "a?b"} {
		key := destination(f.w.Host, f.ob, lease.ObjectVersion{Key: key, VersionID: v})
		if strings.HasSuffix(key, "@"+v) {
			t.Errorf("version not escaped: %s", v)
		}
	}
}

func TestDeleteParksStoppedAuthoritativeLocation(t *testing.T) {
	f := newFixture(t)
	f.seed()
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie"}, nil
	}
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
	f.done(t)
}
func TestForeignRootGenerationRetirement(t *testing.T) {
	f := newFixture(t)
	f.seed()
	obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	_, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations[0].Kind = "park-generation"
		c.Obligations[0].Generation = hostclaim.RootGeneration
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
		return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk", GenerationOf: func(Ref) (string, bool) { return "", true }}, nil
	}
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
	f.done(t)
}

func TestLeaseLossAndAuthorizationRefuse(t *testing.T) {
	for _, phase := range []string{"before-first-effect", "before-completion", "authorize"} {
		t.Run(phase, func(t *testing.T) {
			f := newFixture(t)
			f.seed()
			before := f.b.Snapshot()
			switch phase {
			case "before-first-effect":
				f.held.failed = true
			case "before-completion":
				f.e.After = func(s string) error {
					if s == phase {
						f.held.failed = true
					}
					return nil
				}
			case "authorize":
				f.e.Authorize = func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return ErrChanged }
			}
			if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrChanged) {
				t.Fatal(err)
			}
			obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
			if len(obs.Claim.Obligations) != 1 || f.held.released != 0 {
				t.Fatal("acknowledged without authority")
			}
			if phase != "before-completion" && f.b.Snapshot() != before {
				t.Fatal("acted without authority")
			}
		})
	}
}
func TestMalformedManifestNeverDeletes(t *testing.T) {
	f := newFixture(t)
	f.seed()
	_, err := f.b.PutIfAbsent(t.Context(), "data", archiveBase(f.w.Host, f.ob)+"manifest.json", []byte(`{"V":1,"Host":"other.example"}`))
	if err != nil {
		t.Fatal(err)
	}
	before := f.b.Snapshot()
	if err := f.e.Run(t.Context(), f.w); err == nil {
		t.Fatal("accepted malformed manifest")
	}
	if before != f.b.Snapshot() {
		t.Fatal("deleted from bad manifest")
	}
}

// Parked is what the landing's parking phase waits for: the manifest is
// there and every version it lists is gone. It is not the acknowledgement,
// which still waits for the horizon in Run.
func TestParkedReportsManifestAndDeletions(t *testing.T) {
	f := newFixture(t)
	f.seed()
	parked := func() bool {
		t.Helper()
		ok, err := f.e.Parked(t.Context(), f.w)
		if err != nil {
			t.Fatalf("Parked: %v", err)
		}
		return ok
	}
	if parked() {
		t.Fatal("parked before any work")
	}
	stop := errors.New("crash")
	crashAt := func(step string, n int) {
		t.Helper()
		seen := 0
		f.e.After = func(s string) error {
			if s == step {
				seen++
				if seen == n {
					return stop
				}
			}
			return nil
		}
		if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
			t.Fatalf("crash at %s: %v", step, err)
		}
		f.e.After = nil
	}
	crashAt("manifest", 1)
	if parked() {
		t.Fatal("parked with a manifest and nothing deleted")
	}
	crashAt("delete", 1)
	if parked() {
		t.Fatal("parked with versions left to delete")
	}
	// Resume through every delete, stopping at the verification: the
	// manifest is there and everything in it is gone, but the obligation
	// is not acknowledged.
	crashAt("verify", 1)
	if !parked() {
		t.Fatal("not parked after every listed version is gone")
	}
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || len(obs.Claim.Obligations) != 1 {
		t.Fatalf("Parked must not acknowledge: %v %v", obs.Claim.Obligations, err)
	}
	// A version the manifest does not list is not in the deletion set: the
	// retirement's own verification pass handles it, Parked only reports on
	// what the manifest promised.
	f.b.Put("data", f.ob.Park.Location+"/0000/late", []byte("late"))
	if !parked() {
		t.Fatal("an unlisted late version changed the manifest's answer")
	}
}

// A key with many versions is listed once, however many of its versions are
// in the deletion set.
func TestParkedListsEachKeyOnce(t *testing.T) {
	f := newFixture(t)
	f.seed()
	stop := errors.New("crash")
	f.e.After = func(s string) error {
		if s == "verify" {
			return stop
		}
		return nil
	}
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
		t.Fatal(err)
	}
	f.e.After = nil
	m, err := f.e.manifest(t.Context(), f.w, f.ob)
	if err != nil || m == nil {
		t.Fatal(m, err)
	}
	perKey := map[string]int{}
	for _, d := range m.Delete {
		perKey[d.Key]++
	}
	if len(m.Delete) < 3 || len(perKey) != 1 {
		t.Fatalf("the fixture must put several versions under one key: %v", perKey)
	}
	// The manifest and chore reads list too; count only the deletion keys.
	lists := 0
	f.b.FailOn = func(op, key string) error {
		if op == "list" && perKey[key] > 0 {
			lists++
		}
		return nil
	}
	if ok, err := f.e.Parked(t.Context(), f.w); !ok || err != nil {
		t.Fatalf("parked: %v %v", ok, err)
	}
	f.b.FailOn = nil
	if lists != 1 {
		t.Fatalf("a key with %d deleted versions was listed %d times", len(m.Delete), lists)
	}
}

func TestParkedNeverGuesses(t *testing.T) {
	f := newFixture(t)
	f.seed()
	stop := errors.New("crash")
	f.e.After = func(s string) error {
		if s == "verify" {
			return stop
		}
		return nil
	}
	if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
		t.Fatal(err)
	}
	f.e.After = nil
	boom := errors.New("offline")
	for _, fail := range []string{"list", "get"} {
		f.b.FailOn = func(op, key string) error {
			if op == fail {
				return boom
			}
			return nil
		}
		ok, err := f.e.Parked(t.Context(), f.w)
		if ok || err == nil {
			t.Fatalf("%s failure answered %v, %v", fail, ok, err)
		}
	}
	f.b.FailOn = nil
	if ok, err := f.e.Parked(t.Context(), f.w); !ok || err != nil {
		t.Fatalf("recovered read: %v %v", ok, err)
	}
	// A chore that is not in the claim cannot be verified from its manifest.
	if ok, err := f.e.Parked(t.Context(), Work{f.w.Host, "nope", f.w.Op}); ok || !errors.Is(err, ErrChanged) {
		t.Fatalf("unknown chore: %v %v", ok, err)
	}
	// A rotated retirement has no manifest yet: not parked, and not an error.
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		c.Obligations[0].Retirement = "rt-other"
		return nil
	})
	if err != nil {
		t.Fatal(err)
	}
	if ok, err := f.e.Parked(t.Context(), f.w); ok || err != nil {
		t.Fatalf("a retirement with no manifest answered %v, %v", ok, err)
	}
	if ok, err := (Engine{}).Parked(t.Context(), f.w); ok || !errors.Is(err, ErrUnsafe) {
		t.Fatalf("unconfigured engine: %v %v", ok, err)
	}
}

// A move's retirement goes through the same engine as any other: queued by
// hostclaim.QueueLocationPark on the existing park-replica kind, archived
// under parked/<incarnation>/<host>/<retirement>/, and the manifest's Op is
// the move id (F3 reads it).
func TestEngineArchivesAMoveRetirementUnderParked(t *testing.T) {
	f := newFixture(t)
	const move = "mv_abcdefghijklmnopqrstuvwxyz"
	scope := *f.ob.Park
	n := 0
	newID := func() string { n++; return fmt.Sprintf("q%d", n) }
	obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil {
		t.Fatal(err)
	}
	_, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
		// The source's region after the commit: the claim's landing/committed.
		c.Op = hostclaim.Op{Kind: hostclaim.KindLanding, ID: move, Phase: "committed", Intent: map[string]string{"to": "uk"}}
		c.Obligations = nil
		return hostclaim.QueueLocationPark(move, "ie", "", scope, newID)(c)
	})
	if err != nil {
		t.Fatal(err)
	}
	obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
	if err != nil || len(obs.Claim.Obligations) != 1 {
		t.Fatal(obs.Claim.Obligations, err)
	}
	ob := obs.Claim.Obligations[0]
	f.ob, f.w = ob, Work{f.w.Host, ob.ID, move}
	id := f.b.Put("data", scope.Location+"/0000/x.ltx", []byte("ltx"))
	if err := f.e.Run(t.Context(), f.w); err != nil {
		t.Fatal(err)
	}
	m, err := f.e.manifest(t.Context(), f.w, ob)
	if err != nil || m == nil {
		t.Fatalf("manifest: %v %v", m, err)
	}
	if m.Op != move || m.Retirement != ob.Retirement || m.Location != scope.Location {
		t.Fatalf("manifest: %+v", m)
	}
	want := "parked/" + scope.Incarnation + "/" + f.w.Host + "/" + ob.Retirement + "/instances/0000/x.ltx@" + url.QueryEscape(id)
	if len(m.Archive) != 1 || m.Archive[0].Destination.Key != want {
		t.Fatalf("archive %+v, want %s", m.Archive, want)
	}
	f.done(t)
}