amadan

carlos / platform Public

host: the moves retirement evidence provider (Plan 5 Task 5; K10, M7, M8)

movesRetirementEvidence answers the claim's regional worker before it
sweeps moves/<host>: VerifyMovesSweep returns the K10 precondition proof
from the record's LastMove, and VerifyRetirement labels the sweep with
the same proof's incarnation and the close's time.

The proof holds only for a serving claim released by LastMove's move,
the holder's own record with no live move, a close with a real time
before now, and moves lanes that are that move's or older than its
close. VerifyRetirement also refuses, with a typed dependency error,
while instances/<host> residue lies outside the live location (the
legacy census is claim Plan 12's). A free claim names no record, so it
refuses the same way. Not registered: Task 9a does that.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude c8a04d7c5bd1121e89336d9d5af2af5b09ced087 parent 44c1b6d
2 files changed, +777 −0
  • internal/host/movesevidence.go +194 −0
  • internal/host/movesevidence_test.go +583 −0
diff --git a/internal/host/movesevidence.go b/internal/host/movesevidence.go
new file mode 100644
index 000000000..5393648f3
--- /dev/null
+++ b/internal/host/movesevidence.go
@@ -0,0 +1,194 @@
+package host
+
+// The moves retirement evidence provider (moves Plan 5 Task 5; rulings K10,
+// M7 and M8 of the Plan 5 ledger). The claim's regional worker
+// (hostbox.RegionalWorker) sweeps instances/<host>/ and moves/<host>/ at rest.
+// Before it sweeps it asks this provider twice:
+//
+// - VerifyRetirement, once per sweep, for the incarnation label of every
+// candidate outside an incarnation directory and for the evidence time of
+// every instances/<host> candidate. Decide labels the moves/<host>
+// candidate with the same incarnation (park/sweep.go), and the engine's
+// proof Check requires the proof to name it, so both answers come from one
+// proof.
+// - VerifyMovesSweep, per moves/<host> candidate (park/move_sweep.go). The
+// engine then stamps the listing's end as that chore's evidence, which is
+// valid only with the precondition proof (K10): the claim shows no move
+// op, and the newest retired move's close is earlier than the listing.
+//
+// The proof comes from the record's LastMove (M8). It holds only when:
+//
+// - the claim is serving and was released by LastMove's move;
+// - the record is the holder's own and carries no live move;
+// - LastMove names a move closed at a real time before this clock;
+// - every moves/<host>/<op>/ object is that move's own, or was written
+// before its close.
+//
+// With M4's record-first close, LastMove.At follows every park of the move's
+// lanes, and so every staging horizon. A zero time is never evidence.
+//
+// VerifyRetirement also refuses while any instances/<host> residue lies
+// outside the holder's live location (M7). Moves keeps no archived history
+// for legacy or older incarnations; that census is claim Plan 12's.
+//
+// Nothing registers this provider yet. Task 9a passes it to
+// RegisterProductionClaimMoveParks, which reaches VerifyMovesSweep by type
+// assertion (claimassembly.go).
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+)
+
+// ErrMovesSweepRefused is what every MovesSweepRefusal unwraps to.
+var ErrMovesSweepRefused = errors.New("host: the moves sweep is refused")
+
+// MovesSweepRefusal is the provider's refusal of one observation. Each one
+// skips that pass's sweep, and the next pass asks again. Anomaly marks a
+// refusal that no ordinary move produces: the history contradicts itself, or
+// something wrote where no known move could. M8 asks for one page per
+// anomaly; the provider sends none itself, and its wiring (Task 9a) pages.
+// Reason is for the agent's log, never shown to a user.
+type MovesSweepRefusal struct {
+ Host, Reason string
+ Anomaly bool
+}
+
+func (e *MovesSweepRefusal) Error() string {
+ return "host: the moves sweep of " + e.Host + " is refused: " + e.Reason
+}
+
+func (e *MovesSweepRefusal) Unwrap() error { return ErrMovesSweepRefused }
+
+// movesLegacyCensus is the contract VerifyRetirement waits on for residue it
+// has no history for (M7: the legacy census is claim Plan 12's).
+const movesLegacyCensus = "claim Plan 12 archived-incarnation census for instances/<host> residue outside the live location"
+
+// movesRetirementEvidence implements hostbox.RegionalEvidence and
+// park.MoveSweepVerifier over the host's provisioning record.
+type movesRetirementEvidence struct {
+ Records MovesRecordReader
+ Now func() time.Time
+}
+
+// movesRetired is the proof both methods answer from.
+type movesRetired struct {
+ host, incarnation, location, moveID string
+ at time.Time
+}
+
+func (e movesRetirementEvidence) VerifyMovesSweep(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (park.MoveSweepProof, error) {
+ r, err := e.retired(ctx, o, versions)
+ if err != nil {
+ return park.MoveSweepProof{}, err
+ }
+ return park.MoveSweepProof{Host: r.host, ClaimETag: o.ETag, MoveID: r.moveID, Incarnation: r.incarnation, RetiredAt: r.at}, nil
+}
+
+func (e movesRetirementEvidence) VerifyRetirement(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
+ r, err := e.retired(ctx, o, versions)
+ if err != nil {
+ return "", time.Time{}, err
+ }
+ live := r.location + "/"
+ for _, v := range versions {
+ if strings.HasPrefix(v.Key, "instances/"+r.host+"/") && !strings.HasPrefix(v.Key, live) {
+ return "", time.Time{}, &ClaimDependencyError{Contracts: []string{movesLegacyCensus}}
+ }
+ }
+ return r.incarnation, r.at, nil
+}
+
+// retired proves the newest retired move from the claim observation, the
+// holder's record and the complete census, or refuses.
+func (e movesRetirementEvidence) retired(ctx context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (movesRetired, error) {
+ if ctx == nil || e.Records == nil || e.Now == nil {
+ return movesRetired{}, hostwrite.ErrUnavailable
+ }
+ if o.Absent || o.Claim == nil || o.ETag == "" || o.Host == "" || o.Claim.Host != o.Host {
+ return movesRetired{}, hostwrite.ErrUnavailable
+ }
+ c, host := o.Claim, o.Host
+ refuse := func(anomaly bool, format string, args ...any) (movesRetired, error) {
+ return movesRetired{}, &MovesSweepRefusal{Host: host, Reason: fmt.Sprintf(format, args...), Anomaly: anomaly}
+ }
+ if c.Holder == nil {
+ // A free or joining claim names no record, so no move history.
+ return movesRetired{}, &ClaimDependencyError{Contracts: []string{movesLegacyCensus}}
+ }
+ if c.Op.Kind != hostclaim.KindServing {
+ return refuse(false, "the claim is under a %s op", c.Op.Kind)
+ }
+ h := c.Holder
+ if h.Family != "instance" || h.Key != provision.RecordKey(h.Account, h.App, host) {
+ return refuse(false, "the holder is not an instance record")
+ }
+ location := strings.TrimSuffix(h.Location, "/")
+ if location != "instances/"+host+"/"+h.Incarnation {
+ return movesRetired{}, &ClaimDependencyError{Contracts: []string{movesLegacyCensus}}
+ }
+ rec, err := e.Records(ctx, h.Account, h.App, host)
+ if err != nil {
+ return movesRetired{}, err
+ }
+ if rec == nil {
+ return refuse(true, "the holder's record is missing")
+ }
+ inc, err := hostclaim.Incarnation(rec.Account, rec.App, rec.Host, rec.CreatedAt)
+ if err != nil || rec.Account != h.Account || rec.App != h.App || rec.Host != host || inc != h.Incarnation {
+ return refuse(true, "the record is not the holder's incarnation")
+ }
+ if rec.Move != nil {
+ return refuse(false, "the record carries the live move %s", rec.Move.ID)
+ }
+ last := rec.LastMove
+ if last == nil || last.ID == "" {
+ return refuse(false, "the record has closed no move")
+ }
+ if err := hostclaim.RequireReleasedMove(last.ID)(c); err != nil {
+ // A serving op names the move that released it until another op
+ // replaces it. Any other name is a release the record never closed.
+ // The key is hostclaim's journalReleasedMove; it only picks the
+ // reason, RequireReleasedMove above is the check.
+ if c.Op.Journal["released_move"] == "" {
+ return refuse(false, "the serving claim names no released move")
+ }
+ return refuse(true, "the claim was released by %s, the record closed %s", c.Op.Journal["released_move"], last.ID)
+ }
+ if last.At.IsZero() || !last.At.Before(e.Now()) {
+ return refuse(true, "the close of %s has no time before now", last.ID)
+ }
+ movesRoot := "moves/" + host + "/"
+ for _, v := range versions {
+ if strings.HasPrefix(v.Key, "instances/"+host+"/") {
+ continue
+ }
+ tail, ok := strings.CutPrefix(v.Key, movesRoot)
+ if !ok {
+ return refuse(true, "the census holds %s, outside this host", v.Key)
+ }
+ op, _, inLane := strings.Cut(tail, "/")
+ if !inLane {
+ if tail == "lock.json" {
+ continue // the moves/<host> park's own lease
+ }
+ return refuse(true, "%s is outside every lane", v.Key)
+ }
+ if op == last.ID {
+ continue
+ }
+ if v.LastModified.IsZero() || !v.LastModified.Before(last.At) {
+ return refuse(true, "lane %s was written after %s closed", op, last.ID)
+ }
+ }
+ return movesRetired{host: host, incarnation: h.Incarnation, location: location, moveID: last.ID, at: last.At.UTC()}, nil
+}
diff --git a/internal/host/movesevidence_test.go b/internal/host/movesevidence_test.go
new file mode 100644
index 000000000..a3d7afc9d
--- /dev/null
+++ b/internal/host/movesevidence_test.go
@@ -0,0 +1,583 @@
+package host
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostbox"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// The moves retirement evidence provider (Plan 5 Task 5, rulings K10, M7,
+// M8). The fake bucket's clock starts at 2026-10-05T00:00:00Z and ticks a
+// second per write, so every object a fixture writes is older than evAt.
+var (
+ evCreated = time.Date(2026, 9, 1, 0, 0, 0, 0, time.UTC)
+ evAt = time.Date(2026, 10, 5, 1, 0, 0, 0, time.UTC) // LastMove.At, the close
+ evNow = time.Date(2026, 10, 5, 2, 0, 0, 0, time.UTC) // the sweep's clock
+)
+
+const (
+ evHost = "ev.example"
+ evAcct = "acct"
+ evApp = "app"
+ evMoveID = "mv_retired"
+)
+
+type evFixture struct {
+ inc, location string
+ claims release.Store
+ bucket *leasetest.VersionedBucket
+ rec *provision.Record
+ reads int
+}
+
+// newEvFixture is a serving claim released by evMoveID, whose record closed
+// that move at evAt and carries no live move.
+func newEvFixture(t *testing.T) *evFixture {
+ t.Helper()
+ inc, err := hostclaim.Incarnation(evAcct, evApp, evHost, evCreated)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f := &evFixture{inc: inc, location: "instances/" + evHost + "/" + inc, claims: release.NewDirStore(t.TempDir()), bucket: leasetest.NewVersionedBucket()}
+ f.bucket.AddBucket("data")
+ f.rec = &provision.Record{Schema: 1, Host: evHost, App: evApp, Account: evAcct, CreatedAt: evCreated,
+ LastMove: &provision.MoveOutcome{ID: evMoveID, Outcome: "done", At: evAt}}
+ f.setClaim(t, func(c *hostclaim.Claim) {
+ c.Holder = &hostclaim.Holder{Family: "instance", Key: provision.RecordKey(evAcct, evApp, evHost), Account: evAcct, App: evApp, Incarnation: inc, CreatedAt: evCreated, Location: f.location, Stamp: 1, Session: 1}
+ c.Op = servingReleasedBy(evMoveID)
+ c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
+ })
+ return f
+}
+
+// servingReleasedBy is the serving op hostclaim.ReleaseMove leaves.
+func servingReleasedBy(id string) hostclaim.Op {
+ return hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box", Journal: map[string]string{"released_move": id}}
+}
+
+func (f *evFixture) setClaim(t *testing.T, mut func(*hostclaim.Claim)) {
+ t.Helper()
+ obs, err := hostclaim.Read(t.Context(), f.claims, evHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := hostclaim.Update(t.Context(), f.claims, obs, func(c *hostclaim.Claim) error { mut(c); return nil }); err != nil {
+ t.Fatal(err)
+ }
+}
+
+func (f *evFixture) observe(t *testing.T) hostclaim.Observed {
+ t.Helper()
+ obs, err := hostclaim.Read(t.Context(), f.claims, evHost)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return obs
+}
+
+func (f *evFixture) census(t *testing.T) []lease.ObjectVersion {
+ t.Helper()
+ var out []lease.ObjectVersion
+ for _, p := range []string{"instances/" + evHost + "/", "moves/" + evHost + "/"} {
+ vs, err := f.bucket.ListVersions(t.Context(), "data", p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ out = append(out, vs...)
+ }
+ return out
+}
+
+func (f *evFixture) provider() movesRetirementEvidence {
+ return movesRetirementEvidence{
+ Records: func(_ context.Context, acct, app, host string) (*provision.Record, error) {
+ f.reads++
+ if acct != evAcct || app != evApp || host != evHost {
+ return nil, fmt.Errorf("read %s/%s/%s", acct, app, host)
+ }
+ if f.rec == nil {
+ return nil, errors.New("no record")
+ }
+ cp := *f.rec
+ return &cp, nil
+ },
+ Now: func() time.Time { return evNow },
+ }
+}
+
+// lane writes one object of a move's staging lane.
+func (f *evFixture) lane(op, name string) {
+ f.bucket.Put("data", "moves/"+evHost+"/"+op+"/"+name, []byte(op))
+}
+
+func wantRefusal(t *testing.T, err error, anomaly bool) {
+ t.Helper()
+ var r *MovesSweepRefusal
+ if !errors.As(err, &r) {
+ t.Fatalf("err = %v, want a *MovesSweepRefusal", err)
+ }
+ if r.Host != evHost || r.Reason == "" || r.Anomaly != anomaly {
+ t.Fatalf("refusal = %+v, want host %s, a reason and anomaly %v", r, evHost, anomaly)
+ }
+}
+
+// bothRefuse asserts the two methods refuse alike: VerifyRetirement's
+// label is the proof's, so neither may answer where the other refuses.
+func bothRefuse(t *testing.T, f *evFixture, anomaly bool) {
+ t.Helper()
+ obs, census := f.observe(t), f.census(t)
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), obs, census); err == nil {
+ t.Fatal("VerifyMovesSweep answered")
+ } else {
+ wantRefusal(t, err, anomaly)
+ }
+ inc, at, err := f.provider().VerifyRetirement(t.Context(), obs, census)
+ if err == nil || inc != "" || !at.IsZero() {
+ t.Fatalf("VerifyRetirement = %q, %v, %v; want a refusal and no evidence", inc, at, err)
+ }
+ wantRefusal(t, err, anomaly)
+}
+
+// TestMovesSweepProofRetiredAtIsTheClose (M8): the proof binds the exact
+// claim observation, names LastMove's id, the holder's incarnation and the
+// close's time, and the engine's Check accepts it only for a listing that
+// started after that close.
+func TestMovesSweepProofRetiredAtIsTheClose(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.lane(evMoveID, "replica/0000/db")
+ f.bucket.Put("data", "moves/"+evHost+"/lock.json", []byte("lease"))
+ obs := f.observe(t)
+ proof, err := f.provider().VerifyMovesSweep(t.Context(), obs, f.census(t))
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := park.MoveSweepProof{Host: evHost, ClaimETag: obs.ETag, MoveID: evMoveID, Incarnation: f.inc, RetiredAt: evAt}
+ if proof != want {
+ t.Fatalf("proof = %+v\nwant %+v", proof, want)
+ }
+ if err := proof.Check(obs, f.inc, evNow); err != nil {
+ t.Fatalf("a listing after the close: %v", err)
+ }
+ for _, listing := range []time.Time{evAt, evAt.Add(-time.Second)} {
+ if err := proof.Check(obs, f.inc, listing); !errors.Is(err, park.ErrUnsafe) {
+ t.Fatalf("a listing at %v, not after the close %v: %v", listing, evAt, err)
+ }
+ }
+}
+
+// TestVerifyRetirementLabelsMovesResidueWithTheProofsIncarnation (M7):
+// Decide labels every moves/<host> candidate with VerifyRetirement's
+// answer, and Check requires the proof's incarnation to equal it, so the
+// two answers come from one source. The evidence time is the close's, never
+// zero and never the clock's.
+func TestVerifyRetirementLabelsMovesResidueWithTheProofsIncarnation(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.bucket.Put("data", f.location+"/g-live/0000/db", []byte("live"))
+ obs, census := f.observe(t), f.census(t)
+ inc, at, err := f.provider().VerifyRetirement(t.Context(), obs, census)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if inc != f.inc || !at.Equal(evAt) {
+ t.Fatalf("VerifyRetirement = %q, %v; want %q, %v", inc, at, f.inc, evAt)
+ }
+ proof, err := f.provider().VerifyMovesSweep(t.Context(), obs, census)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if proof.Incarnation != inc {
+ t.Fatalf("label %q, proof %q", inc, proof.Incarnation)
+ }
+ if err := proof.Check(obs, inc, evNow); err != nil {
+ t.Fatal(err)
+ }
+}
+
+// TestVerifyRetirementRefusesLegacyResidue (M7): residue under
+// instances/<host> outside the holder's live location has no moves-owned
+// archived-incarnation history. VerifyRetirement refuses it with a typed
+// dependency error (the legacy census is claim Plan 12's), which skips the
+// whole at-rest sweep. The moves proof itself is unaffected.
+func TestVerifyRetirementRefusesLegacyResidue(t *testing.T) {
+ for name, key := range map[string]string{
+ "a legacy level object": "instances/" + evHost + "/0000/db",
+ "a legacy generation object": "instances/" + evHost + "/g-old/0000/db",
+ "an older incarnation": "instances/" + evHost + "/" + strings.Repeat("ab", 32) + "/g-old/0000/db",
+ } {
+ t.Run(name, func(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.bucket.Put("data", f.location+"/g-live/0000/db", []byte("live"))
+ f.bucket.Put("data", key, []byte("legacy"))
+ obs, census := f.observe(t), f.census(t)
+ inc, at, err := f.provider().VerifyRetirement(t.Context(), obs, census)
+ var dep *ClaimDependencyError
+ if !errors.As(err, &dep) || inc != "" || !at.IsZero() {
+ t.Fatalf("VerifyRetirement = %q, %v, %v; want a *ClaimDependencyError and no evidence", inc, at, err)
+ }
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), obs, census); err != nil {
+ t.Fatalf("the moves proof does not depend on instances residue: %v", err)
+ }
+ })
+ }
+ t.Run("a holder in the legacy location form", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Holder.Location = "instances/" + evHost })
+ f.bucket.Put("data", "instances/"+evHost+"/0000/db", []byte("live"))
+ _, _, err := f.provider().VerifyRetirement(t.Context(), f.observe(t), f.census(t))
+ var dep *ClaimDependencyError
+ if !errors.As(err, &dep) {
+ t.Fatalf("err = %v, want a *ClaimDependencyError", err)
+ }
+ })
+}
+
+// TestMovesSweepProofRefusesALiveMove (K10, M8): no proof while a move op
+// is on the claim, while the record carries a live move, or while the
+// claim was released by a move the record has not closed.
+func TestMovesSweepProofRefusesALiveMove(t *testing.T) {
+ t.Run("the record carries a live move", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.Move = &provision.MoveState{ID: "mv_next", Incarnation: f.inc, To: "uk", State: provision.MovePreparing}
+ f.lane("mv_next", "move.json")
+ bothRefuse(t, f, false)
+ })
+ for _, op := range []hostclaim.Op{
+ {Kind: hostclaim.KindMoving, ID: "mv_next", Phase: hostclaim.PhaseMoving, Exec: "box"},
+ {Kind: hostclaim.KindLanding, ID: evMoveID, Phase: hostclaim.PhaseCommitted, Exec: "box"},
+ {Kind: hostclaim.KindRestoring, ID: evMoveID, Phase: hostclaim.PhaseOpened, Exec: "box"},
+ {Kind: hostclaim.KindReverting, ID: evMoveID, Phase: hostclaim.PhaseOpened, Exec: "box"},
+ } {
+ // The last three are the record-first close's window (M4): the
+ // record closed the move, the claim has not released it yet.
+ t.Run("the claim holds "+op.Kind, func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Op = op })
+ bothRefuse(t, f, false)
+ })
+ }
+ t.Run("the claim was released by another move", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Op = servingReleasedBy("mv_orphan") })
+ bothRefuse(t, f, true)
+ })
+ t.Run("the serving claim names no released move", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"} })
+ bothRefuse(t, f, false)
+ })
+ t.Run("the record names no closed move", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.LastMove = nil
+ bothRefuse(t, f, false)
+ })
+}
+
+// TestMovesEvidenceRefusesAZeroTime (M7, M8): a zero time is never
+// evidence: not the close's, and not a lane object's.
+func TestMovesEvidenceRefusesAZeroTime(t *testing.T) {
+ t.Run("a zero close", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.LastMove.At = time.Time{}
+ bothRefuse(t, f, true)
+ })
+ t.Run("a close not before the provider's clock", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.LastMove.At = evNow
+ bothRefuse(t, f, true)
+ })
+ t.Run("an older lane object with no time", func(t *testing.T) {
+ f := newEvFixture(t)
+ obs := f.observe(t)
+ census := []lease.ObjectVersion{{Key: "moves/" + evHost + "/mv_older/move.json", VersionID: "v1", IsLatest: true}}
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), obs, census); err == nil {
+ t.Fatal("a zero LastModified counted as before the close")
+ } else {
+ wantRefusal(t, err, true)
+ }
+ if _, _, err := f.provider().VerifyRetirement(t.Context(), obs, census); err == nil {
+ t.Fatal("VerifyRetirement answered over a lane with no time")
+ }
+ })
+}
+
+// TestMovesSweepProofRefusesAnUnknownLane (M8): every moves/<host>/<op>/
+// object is the closed move's own, or older than its close. An object of
+// another op written after the close has an unknown writer.
+func TestMovesSweepProofRefusesAnUnknownLane(t *testing.T) {
+ t.Run("an older lane is covered by the close", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane("mv_older", "move.json")
+ f.lane("rb_return", "replica/0000/db")
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), f.observe(t), f.census(t)); err != nil {
+ t.Fatal(err)
+ }
+ })
+ t.Run("a lane written after the close", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.LastMove.At = time.Date(2026, 10, 4, 0, 0, 0, 0, time.UTC) // before every write
+ f.lane(evMoveID, "move.json")
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), f.observe(t), f.census(t)); err != nil {
+ t.Fatalf("the closed move's own lane is its own: %v", err)
+ }
+ f.lane("mv_unknown", "move.json")
+ bothRefuse(t, f, true)
+ })
+ t.Run("an object outside any lane", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.bucket.Put("data", "moves/"+evHost+"/stray.json", []byte("?"))
+ bothRefuse(t, f, true)
+ })
+ t.Run("an object outside the host's namespaces", func(t *testing.T) {
+ f := newEvFixture(t)
+ census := []lease.ObjectVersion{{Key: "moves/other.example/" + evMoveID + "/move.json", VersionID: "v1", LastModified: evAt.Add(-time.Hour)}}
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), f.observe(t), census); err == nil {
+ t.Fatal("another host's object counted")
+ }
+ })
+}
+
+// TestMovesEvidenceRefusesAnotherIdentity: the record must be the holder's
+// exact record, and a holderless claim has no record to prove from.
+func TestMovesEvidenceRefusesAnotherIdentity(t *testing.T) {
+ t.Run("a record of another incarnation", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.CreatedAt = evCreated.Add(time.Hour)
+ bothRefuse(t, f, true)
+ })
+ t.Run("a stale observation", func(t *testing.T) {
+ f := newEvFixture(t)
+ obs := f.observe(t)
+ obs.ETag = ""
+ if _, err := f.provider().VerifyMovesSweep(t.Context(), obs, f.census(t)); err == nil {
+ t.Fatal("answered without an etag")
+ }
+ })
+ t.Run("a free claim", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Holder, c.Op = nil, hostclaim.Op{} })
+ obs, census := f.observe(t), f.census(t)
+ _, err := f.provider().VerifyMovesSweep(t.Context(), obs, census)
+ var dep *ClaimDependencyError
+ if !errors.As(err, &dep) {
+ t.Fatalf("err = %v, want a *ClaimDependencyError", err)
+ }
+ if _, _, err := f.provider().VerifyRetirement(t.Context(), obs, census); !errors.As(err, &dep) {
+ t.Fatalf("err = %v, want a *ClaimDependencyError", err)
+ }
+ if f.reads != 0 {
+ t.Fatalf("read %d records with no holder to name one", f.reads)
+ }
+ })
+ t.Run("no providers", func(t *testing.T) {
+ f := newEvFixture(t)
+ if _, err := (movesRetirementEvidence{}).VerifyMovesSweep(t.Context(), f.observe(t), nil); err == nil {
+ t.Fatal("answered with no record reader")
+ }
+ if _, _, err := (movesRetirementEvidence{Records: f.provider().Records}).VerifyRetirement(t.Context(), f.observe(t), nil); err == nil {
+ t.Fatal("answered with no clock")
+ }
+ })
+}
+
+// The provider satisfies both slots RegisterProductionClaimMoveParks reads:
+// the regional evidence and, by type assertion, the sweep verifier.
+var (
+ _ hostbox.RegionalEvidence = movesRetirementEvidence{}
+ _ park.MoveSweepVerifier = movesRetirementEvidence{}
+)
+
+// --- against the claim's regional worker, with a fake store ---
+
+type evLease struct{}
+
+func (evLease) Check(context.Context) error { return nil }
+func (evLease) Release(context.Context) error { return nil }
+func (evLease) Close() {}
+
+type evLeases struct{}
+
+func (evLeases) Acquire(context.Context, string, string) (park.HeldLease, error) {
+ return evLease{}, nil
+}
+
+type evEnrollment struct{}
+
+func (evEnrollment) VerifyRegional(context.Context, string, string, time.Duration) error { return nil }
+
+// worker is hostbox.RegionalWorker over park.Engine, wired as the claim's
+// assembly wires them (host/claimassembly.go): the provider is the worker's
+// Evidence and the engine's VerifyMovesSweep. This region (ie) is not the
+// authoritative one, so the holder's location here is residue too.
+func (f *evFixture) worker(t *testing.T, listing time.Time) hostbox.RegionalWorker {
+ t.Helper()
+ seq := 0
+ p := f.provider()
+ engine := park.Engine{
+ Store: f.bucket, Claims: f.claims, Bucket: "data", Region: "ie", Leases: evLeases{},
+ Now: func() time.Time { return listing },
+ NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
+ View: func(_ context.Context, o hostclaim.Observed) (park.View, error) {
+ return park.View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
+ },
+ Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil },
+ VerifyMovesSweep: p.VerifyMovesSweep,
+ }
+ return hostbox.RegionalWorker{
+ Regional: hostbox.Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }},
+ Enrollment: evEnrollment{},
+ Evidence: p,
+ }
+}
+
+func (f *evFixture) chores(t *testing.T) map[string]hostclaim.ParkScope {
+ t.Helper()
+ out := map[string]hostclaim.ParkScope{}
+ for _, o := range f.observe(t).Claim.Obligations {
+ if o.Park == nil {
+ t.Fatalf("an unscoped chore: %+v", o)
+ }
+ out[o.Park.Location] = *o.Park
+ }
+ return out
+}
+
+// TestAllowedSweepQueuesTheMovesChore: with the proof, the worker queues the
+// moves/<host> chore under the serving session, labelled with the holder's
+// incarnation, its evidence the listing's end, so it waits for its horizon.
+// The holder location's residue in this non-authoritative region takes the
+// close as its evidence: that horizon has passed, so the same pass parks it.
+func TestAllowedSweepQueuesTheMovesChore(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.lane("mv_older", "replica/0000/db")
+ left := f.location + "/g-left/0000/db"
+ f.bucket.Put("data", left, []byte("left behind"))
+ err := f.worker(t, evNow).Run(t.Context())
+ if !errors.Is(err, park.ErrPending) {
+ t.Fatalf("a new sweep waits for its horizon: %v", err)
+ }
+ chores := f.chores(t)
+ moves, ok := chores["moves/"+evHost]
+ if !ok || len(chores) != 1 {
+ t.Fatalf("chores = %+v; want the moves chore alone", chores)
+ }
+ if moves.Incarnation != f.inc || !moves.EvidenceAt.Equal(evNow) || moves.RequestLimit != hostbox.RequestLimit {
+ t.Fatalf("moves chore = %+v; want incarnation %s, evidence %v", moves, f.inc, evNow)
+ }
+ if vs, err := f.bucket.ListVersions(t.Context(), "data", left); err != nil || len(vs) != 0 {
+ t.Fatalf("the location's residue was not parked: %v, %v", vs, err)
+ }
+ if vs, err := f.bucket.ListVersions(t.Context(), "data", "parked/"+f.inc+"/"+evHost+"/"); err != nil || len(vs) == 0 {
+ t.Fatalf("no archive under the holder's incarnation: %v, %v", vs, err)
+ }
+ for _, o := range f.observe(t).Claim.Obligations {
+ if o.Op != "session:1" || o.Owner != "region:ie" {
+ t.Fatalf("chore %+v is not the serving session's regional chore", o)
+ }
+ }
+ if c := f.observe(t).Claim; c.Op.Kind != hostclaim.KindServing || c.Op.Journal["released_move"] != evMoveID {
+ t.Fatalf("the sweep changed the serving op: %+v", c.Op)
+ }
+}
+
+// TestSweepSkipsWhileAMoveStages (Review Focus 3, K10): while a move op is
+// on the claim the worker never asks for evidence; while the record carries
+// a live move under a serving claim the provider refuses and nothing is
+// queued.
+func TestSweepSkipsWhileAMoveStages(t *testing.T) {
+ t.Run("a moving claim", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindMoving, ID: "mv_next", Phase: hostclaim.PhaseMoving, Exec: "box"}
+ })
+ f.rec.Move = &provision.MoveState{ID: "mv_next", Incarnation: f.inc, To: "uk", State: provision.MovePreparing}
+ f.lane("mv_next", "replica/0000/db")
+ if err := f.worker(t, evNow).Run(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ if len(f.chores(t)) != 0 || f.reads != 0 {
+ t.Fatalf("chores %+v, record reads %d", f.chores(t), f.reads)
+ }
+ })
+ t.Run("a live move on the record", func(t *testing.T) {
+ f := newEvFixture(t)
+ f.rec.Move = &provision.MoveState{ID: "mv_next", Incarnation: f.inc, To: "uk", State: provision.MovePreparing}
+ f.lane("mv_next", "replica/0000/db")
+ err := f.worker(t, evNow).Run(t.Context())
+ var r *MovesSweepRefusal
+ if !errors.As(err, &r) {
+ t.Fatalf("err = %v, want the provider's refusal", err)
+ }
+ if len(f.chores(t)) != 0 {
+ t.Fatalf("chores %+v", f.chores(t))
+ }
+ })
+}
+
+// TestSweepRefusedUnlessTheCloseIsBeforeTheListing (K10): the worker's own
+// Check drops the moves/<host> candidate when the newest retired move's
+// close is not earlier than the listing; the rest of the decision stands.
+func TestSweepRefusedUnlessTheCloseIsBeforeTheListing(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.bucket.Put("data", f.location+"/g-left/0000/db", []byte("left behind"))
+ err := f.worker(t, evAt).Run(t.Context())
+ if !errors.Is(err, park.ErrPending) {
+ t.Fatalf("the location chore still queues: %v", err)
+ }
+ chores := f.chores(t)
+ if _, ok := chores["moves/"+evHost]; ok || len(chores) != 1 {
+ t.Fatalf("chores = %+v; want only the location's", chores)
+ }
+ if loc := chores[f.location]; loc.Incarnation != f.inc || !loc.EvidenceAt.Equal(evAt) {
+ t.Fatalf("location chore = %+v; want the close %v as its evidence", loc, evAt)
+ }
+}
+
+// TestSweepScopeMembership (witness wSweepScopeMembership, M7): residue of
+// an older incarnation is never the moves/<host> chore's. Without its own
+// evidence the provider refuses the whole sweep, so nothing is queued.
+func TestSweepScopeMembership(t *testing.T) {
+ f := newEvFixture(t)
+ f.lane(evMoveID, "move.json")
+ f.bucket.Put("data", "instances/"+evHost+"/"+strings.Repeat("cd", 32)+"/g-old/0000/db", []byte("late"))
+ err := f.worker(t, evNow).Run(t.Context())
+ var dep *ClaimDependencyError
+ if !errors.As(err, &dep) {
+ t.Fatalf("err = %v, want the legacy census dependency", err)
+ }
+ if len(f.chores(t)) != 0 {
+ t.Fatalf("chores %+v", f.chores(t))
+ }
+}
+
+// TestFreeHostSweepIsRefusedWithoutAHolder: a free claim names no record,
+// so no closed move can be proven and the claim stays free. Its residue is
+// the legacy census's (M7, claim Plan 12).
+func TestFreeHostSweepIsRefusedWithoutAHolder(t *testing.T) {
+ f := newEvFixture(t)
+ f.setClaim(t, func(c *hostclaim.Claim) { c.Holder, c.Op, c.Touched = nil, hostclaim.Op{}, hostclaim.Touched{} })
+ f.lane(evMoveID, "move.json")
+ err := f.worker(t, evNow).Run(t.Context())
+ var dep *ClaimDependencyError
+ if !errors.As(err, &dep) {
+ t.Fatalf("err = %v, want a *ClaimDependencyError", err)
+ }
+ if c := f.observe(t).Claim; !c.Free() || len(c.Obligations) != 0 {
+ t.Fatalf("claim = %+v", c)
+ }
+}