amadan
park: Parked; hostclaim: park queue helpers for move retirements on the existing kinds (§6.4, D6)
Engine.Parked reports the manifest present and every listed version gone. QueueLocationPark, QueueStagingPark, QueueLocalPark and QueueResidueParks queue park-replica, park-local and park-generation chores for a move; no new kinds. InventoryHasStamp and CompleteLocalPark carry G5's first-landing rule.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
4 files changed,
+793
−0
internal/hostclaim/move.go+208 −0internal/hostclaim/move_test.go+366 −0internal/park/remote.go+54 −0internal/park/remote_test.go+165 −0
diff --git a/internal/hostclaim/move.go b/internal/hostclaim/move.go| index fdcb41036..98956e59e 100644 |
| --- a/internal/hostclaim/move.go |
| +++ b/internal/hostclaim/move.go |
| @@ -3,6 +3,7 @@ package hostclaim |
| import ( |
| "fmt" |
| "strconv" |
| + "strings" |
| "time" |
| ) |
| @@ -302,3 +303,210 @@ func Revert(opID, sourceBox, targetRegion string, stagingScope *ParkScope, newID |
| func sameParkScope(a, b ParkScope) bool { |
| return a.Location == b.Location && a.Incarnation == b.Incarnation && a.EvidenceAt.Equal(b.EvidenceAt) && a.RequestLimit == b.RequestLimit |
| } |
| + |
| +// Park queue helpers for move retirements (moves spec §6.4, index D6). A |
| +// move adds no obligation kind: a staging lane is a park-replica scoped to |
| +// moves/<host>/<op>; a whole location retired where it is no longer |
| +// authoritative is a park-replica scoped to instances/<host>[/<inc>]; a |
| +// landing's earlier-stay residue is one park-generation per generation |
| +// (RootGeneration for the origin); the box's own artifacts are a park-local. |
| +// Claim.Validate accepts all of them unchanged, and the park engine archives |
| +// them under parked/<incarnation>/<host>/<retirement>/ with the op in the |
| +// manifest. |
| +// |
| +// Each helper changes obligations only, is idempotent per its key (a replay |
| +// allocates no ids), and refuses a replay that names the same chore with a |
| +// different scope rather than overwriting it. |
| + |
| +// parkQueueGuard is the check every helper shares: the op is current, the |
| +// claim has its holder, the scope is valid for this host and names this |
| +// holder's incarnation. |
| +func parkQueueGuard(c *Claim, opID, region string, scope ParkScope, newID func() string) error { |
| + if err := RequireOp(opID)(c); err != nil { |
| + return err |
| + } |
| + if c.Holder == nil || newID == nil || region == "" { |
| + return fmt.Errorf("%w: a move's park names its region and holder and has a way to mint ids", ErrMalformed) |
| + } |
| + if err := scope.Validate(c.Host); err != nil { |
| + return err |
| + } |
| + if scope.Incarnation != c.Holder.Incarnation { |
| + return fmt.Errorf("%w: the park's incarnation is not this holder's", ErrMalformed) |
| + } |
| + return nil |
| +} |
| + |
| +func isStagingLocation(location string) bool { return strings.HasPrefix(location, "moves/") } |
| + |
| +// queueReplicaPark appends a park-replica for the region unless this op |
| +// already owes one of the same class (a lane or a location) there. It |
| +// reports whether it appended. |
| +func queueReplicaPark(c *Claim, opID, region string, scope ParkScope, newID func() string) (bool, error) { |
| + for _, o := range c.Obligations { |
| + if o.Op == opID && o.Kind == "park-replica" && o.Region == region && o.Park != nil && isStagingLocation(o.Park.Location) == isStagingLocation(scope.Location) { |
| + if !sameParkScope(*o.Park, scope) { |
| + return false, fmt.Errorf("%w: %s already owes a park in %s with another scope", ErrWrongState, opID, region) |
| + } |
| + return false, nil |
| + } |
| + } |
| + s := scope |
| + c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-replica", Region: region, Retirement: newID(), Owner: "region:" + region, Park: &s}) |
| + return true, nil |
| +} |
| + |
| +// queueLocalPark appends a park-local for the box unless this op has one. |
| +func queueLocalPark(c *Claim, opID, box string, newID func() string) { |
| + for _, o := range c.Obligations { |
| + if o.Op == opID && o.Kind == "park-local" && o.Box == box { |
| + return |
| + } |
| + } |
| + c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-local", Box: box, Retirement: newID(), Owner: "box:" + box}) |
| +} |
| + |
| +// QueueLocationPark queues the park of a whole location in a region where it |
| +// is no longer authoritative: the source's after the commit, a rolled-back |
| +// target's at cleaning. The scope is instances/<host> or |
| +// instances/<host>/<incarnation>. With a box, the box's park-local is queued |
| +// in the same write. Idempotent per (op, region) and (op, box). |
| +func QueueLocationPark(opID, region, box string, scope ParkScope, newID func() string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := parkQueueGuard(c, opID, region, scope, newID); err != nil { |
| + return err |
| + } |
| + if isStagingLocation(scope.Location) { |
| + return fmt.Errorf("%w: a location park is not a staging lane", ErrMalformed) |
| + } |
| + if _, err := queueReplicaPark(c, opID, region, scope, newID); err != nil { |
| + return err |
| + } |
| + if box != "" { |
| + queueLocalPark(c, opID, box, newID) |
| + } |
| + return nil |
| + } |
| +} |
| + |
| +// QueueStagingPark queues the park of this move's own staging lane in a |
| +// region. The scope's location must be exactly moves/<host>/<op>: not the |
| +// live location, not the host-wide staging prefix, not another move's lane. |
| +// A lane park a revert already queued under the op is the same chore. |
| +// Idempotent per (op, region). EvidenceAt is the recorded writers' bound |
| +// (G7); the caller computes it. |
| +func QueueStagingPark(opID, region string, scope ParkScope, newID func() string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := parkQueueGuard(c, opID, region, scope, newID); err != nil { |
| + return err |
| + } |
| + // Restates moves.ScopeLocation(host, op); moves imports hostclaim, so |
| + // it cannot be called from here. Keep the two together. |
| + if scope.Location != "moves/"+c.Host+"/"+opID { |
| + return fmt.Errorf("%w: a staging park is this move's own lane, moves/%s/%s", ErrMalformed, c.Host, opID) |
| + } |
| + _, err := queueReplicaPark(c, opID, region, scope, newID) |
| + return err |
| + } |
| +} |
| + |
| +// QueueLocalPark queues a park-local for the box alone, for the landing's |
| +// local phase: park.Local runs only against a persisted park-local chore of |
| +// the claim and treats an unknown id as already completed, so the chore must |
| +// be in the claim before the call. Idempotent per (op, box). Queue it only |
| +// when InventoryHasStamp says the box has paths of the earlier stay (G5). |
| +func QueueLocalPark(opID, box string, newID func() string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := RequireOp(opID)(c); err != nil { |
| + return err |
| + } |
| + if box == "" || newID == nil { |
| + return fmt.Errorf("%w: a local park names its box and has a way to mint ids", ErrMalformed) |
| + } |
| + queueLocalPark(c, opID, box, newID) |
| + return nil |
| + } |
| +} |
| + |
| +// InventoryHasStamp reports whether the box's inventory lists any path |
| +// tagged with the stamp: the landing's earlier stay S' (G5). A first landing |
| +// has none, and then no park-local is queued and none is retired. |
| +func InventoryHasStamp(c *Claim, box string, stamp uint64) bool { |
| + if c == nil || stamp == 0 { |
| + return false |
| + } |
| + for _, s := range c.Inventory[box].Paths { |
| + if s == stamp { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +// CompleteLocalPark is the replay of a local phase that finds the box's |
| +// park-local chore and no paths of the earlier stamp left: there is nothing |
| +// to retire, so the chore is completed under the op (CompleteObligation). |
| +// With paths of that stamp still listed it refuses (ErrWrongState): the |
| +// retirement has to run. No chore is a no-op. |
| +func CompleteLocalPark(opID, box string, earlierStamp uint64) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := RequireOp(opID)(c); err != nil { |
| + return err |
| + } |
| + if box == "" { |
| + return fmt.Errorf("%w: a local park names its box", ErrMalformed) |
| + } |
| + for _, o := range c.Obligations { |
| + if o.Op == opID && o.Kind == "park-local" && o.Box == box { |
| + if InventoryHasStamp(c, box, earlierStamp) { |
| + return fmt.Errorf("%w: %s still lists paths of stamp %d", ErrWrongState, box, earlierStamp) |
| + } |
| + return CompleteObligation(opID, o.ID)(c) |
| + } |
| + } |
| + return nil |
| + } |
| +} |
| + |
| +// QueueResidueParks queues one park-generation per listed generation for a |
| +// landing's earlier-stay residue in the target region: RootGeneration for |
| +// the origin, the generation name for the rest, all scoped to the same |
| +// location. Deduplicated by (op, region, generation); a generation already |
| +// queued with another scope is refused, and nothing is queued then. |
| +func QueueResidueParks(opID, region string, location ParkScope, generations []string, newID func() string) func(*Claim) error { |
| + return func(c *Claim) error { |
| + if err := parkQueueGuard(c, opID, region, location, newID); err != nil { |
| + return err |
| + } |
| + if isStagingLocation(location.Location) { |
| + return fmt.Errorf("%w: residue is parked from a location, not a staging lane", ErrMalformed) |
| + } |
| + for _, g := range generations { |
| + if g == "" { |
| + return fmt.Errorf("%w: a residue generation has a name", ErrMalformed) |
| + } |
| + } |
| + queued := map[string]bool{} |
| + for _, o := range c.Obligations { |
| + if o.Op == opID && o.Kind == "park-generation" && o.Region == region { |
| + if o.Park == nil || !sameParkScope(*o.Park, location) { |
| + for _, g := range generations { |
| + if g == o.Generation { |
| + return fmt.Errorf("%w: %s already owes generation %q in %s with another scope", ErrWrongState, opID, g, region) |
| + } |
| + } |
| + } |
| + queued[o.Generation] = true |
| + } |
| + } |
| + for _, g := range generations { |
| + if queued[g] { |
| + continue |
| + } |
| + queued[g] = true |
| + s := location |
| + c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-generation", Region: region, Generation: g, Retirement: newID(), Owner: "region:" + region, Park: &s}) |
| + } |
| + return nil |
| + } |
| +} |
diff --git a/internal/hostclaim/move_test.go b/internal/hostclaim/move_test.go| index e42c65c47..baa2eb24a 100644 |
| --- a/internal/hostclaim/move_test.go |
| +++ b/internal/hostclaim/move_test.go |
| @@ -680,3 +680,369 @@ func TestRevertReplayIsIdempotent(t *testing.T) { |
| t.Fatalf("replay after the revert moved on: %v", err) |
| } |
| } |
| + |
| +// parkFixture is a valid landing-shaped claim for the park queue helpers: |
| +// op moveA, a holder with an incarnation, and the scope helpers an op needs. |
| +func parkFixture(t *testing.T) (*Claim, func(location string) ParkScope) { |
| + t.Helper() |
| + c := movingTestClaim(t) |
| + if err := BeginMove(moveA, "ops", "ie-box", "enr-ie", "uk-box", "uk", time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC))(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + scope := func(location string) ParkScope { |
| + return ParkScope{Location: location, Incarnation: c.Holder.Incarnation, EvidenceAt: time.Date(2026, 10, 6, 9, 5, 0, 0, time.UTC), RequestLimit: 2 * time.Minute} |
| + } |
| + return c, scope |
| +} |
| + |
| +// The move's retirements ride the engine's existing kinds. A real store |
| +// round trip: Validate accepts what the helper wrote and it reads back. |
| +func TestLocationParkIsAParkReplicaTheStoreAccepts(t *testing.T) { |
| + ctx := context.Background() |
| + s := release.NewDirStore(t.TempDir()) |
| + o := servingClaim(t, ctx, s) |
| + now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + o, err := Update(ctx, s, o, BeginMove(moveA, "ops", "adi-knitting", "enr-ie", "uk-box", "uk", now)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + inc := o.Claim.Holder.Incarnation |
| + scope := ParkScope{Location: "instances/" + testHost + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: 2 * time.Minute} |
| + o, err = Update(ctx, s, o, QueueLocationPark(moveA, "ie", "adi-knitting", scope, counter())) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + o, err = Read(ctx, s, testHost) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := o.Claim.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(o.Claim.Obligations) != 2 { |
| + t.Fatalf("obligations: %+v", o.Claim.Obligations) |
| + } |
| + r, l := o.Claim.Obligations[0], o.Claim.Obligations[1] |
| + if r.Kind != "park-replica" || r.Op != moveA || r.Region != "ie" || r.Owner != "region:ie" || r.Park == nil || !sameParkScope(*r.Park, scope) || r.Retirement == "" || r.ID == "" || r.ID == r.Retirement { |
| + t.Fatalf("replica park: %+v", r) |
| + } |
| + if l.Kind != "park-local" || l.Op != moveA || l.Box != "adi-knitting" || l.Owner != "box:adi-knitting" || l.Park != nil || l.Region != "" || l.Retirement == "" { |
| + t.Fatalf("local park: %+v", l) |
| + } |
| + // The same call again, across the store: nothing new, no ids spent. |
| + ids := 0 |
| + again := func() string { ids++; return fmt.Sprintf("late_%d", ids) } |
| + if o, err = Update(ctx, s, o, QueueLocationPark(moveA, "ie", "adi-knitting", scope, again)); err != nil || ids != 0 || len(o.Claim.Obligations) != 2 { |
| + t.Fatalf("replay: ids=%d obligations=%d err=%v", ids, len(o.Claim.Obligations), err) |
| + } |
| + // Completing both by the existing chore completion ends them. |
| + for _, ob := range o.Claim.Obligations { |
| + if o, err = Update(ctx, s, o, CompleteObligation(ob.Op, ob.ID)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if len(o.Claim.Obligations) != 0 { |
| + t.Fatal("chores left") |
| + } |
| +} |
| + |
| +func TestStagingParkScopeIsTheLane(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + if err := QueueStagingPark(moveA, "uk", lane, counter())(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(c.Obligations) != 1 { |
| + t.Fatalf("obligations: %+v", c.Obligations) |
| + } |
| + o := c.Obligations[0] |
| + if o.Kind != "park-replica" || o.Region != "uk" || o.Owner != "region:uk" || o.Op != moveA || o.Park.Location != "moves/"+c.Host+"/"+moveA { |
| + t.Fatalf("staging park: %+v", o) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Anything but this move's own lane is refused, and nothing is queued. |
| + for name, location := range map[string]string{ |
| + "the live location": "instances/" + c.Host + "/" + c.Holder.Incarnation, |
| + "the legacy prefix": "instances/" + c.Host, |
| + "the staging prefix": "moves/" + c.Host, |
| + "another move": "moves/" + c.Host + "/" + moveB, |
| + "another host": "moves/other.example.com/" + moveA, |
| + "below the lane": "moves/" + c.Host + "/" + moveA + "/x", |
| + } { |
| + d, _ := parkFixture(t) |
| + bad := scope(location) |
| + ids := 0 |
| + if err := QueueStagingPark(moveA, "uk", bad, func() string { ids++; return "x" })(d); err == nil || len(d.Obligations) != 0 || ids != 0 { |
| + t.Errorf("%s: accepted (%v), obligations %d", name, err, len(d.Obligations)) |
| + } |
| + } |
| + // Another holder's incarnation is refused. |
| + d, _ := parkFixture(t) |
| + foreign := scope("moves/" + d.Host + "/" + moveA) |
| + foreign.Incarnation = strings.Repeat("a", 64) |
| + if err := QueueStagingPark(moveA, "uk", foreign, counter())(d); !errors.Is(err, ErrMalformed) { |
| + t.Errorf("foreign incarnation: %v", err) |
| + } |
| + // A stale op is refused. |
| + if err := QueueStagingPark(moveB, "uk", lane, counter())(d); !errors.Is(err, ErrStaleOp) { |
| + t.Errorf("stale op: %v", err) |
| + } |
| +} |
| + |
| +// A revert already queued the lane's park: queueing it again is a no-op, |
| +// and a different scope for the same lane is refused, not overwritten. |
| +func TestStagingParkAfterRevertIsTheSameChore(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + if err := ExecToNone(moveA, "ie-box", stoppedSource, JournalEvidenceSource)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + if err := Revert(moveA, "ie-box", "uk", &lane, counter())(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + ids := 0 |
| + newID := func() string { ids++; return "z" } |
| + if err := QueueStagingPark(moveA, "uk", lane, newID)(c); err != nil || ids != 0 || len(c.Obligations) != 1 { |
| + t.Fatalf("same lane: %v ids=%d obligations=%d", err, ids, len(c.Obligations)) |
| + } |
| + later := lane |
| + later.EvidenceAt = lane.EvidenceAt.Add(time.Second) |
| + if err := QueueStagingPark(moveA, "uk", later, newID)(c); !errors.Is(err, ErrWrongState) || ids != 0 || len(c.Obligations) != 1 { |
| + t.Fatalf("changed scope: %v ids=%d", err, ids) |
| + } |
| +} |
| + |
| +func TestLocationParkAndStagingParkCoexistInOneRegion(t *testing.T) { |
| + // A rolled-back target at cleaning owes the lane and the location in the |
| + // same region under one op. |
| + c, scope := parkFixture(t) |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation) |
| + n := counter() |
| + for i := 0; i < 2; i++ { |
| + if err := QueueStagingPark(moveA, "uk", lane, n)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := QueueLocationPark(moveA, "uk", "", loc, n)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if len(c.Obligations) != 2 { |
| + t.Fatalf("obligations: %+v", c.Obligations) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The location's local chore is per box, not per region. |
| + if err := QueueLocationPark(moveA, "uk", "uk-box", loc, n)(c); err != nil || len(c.Obligations) != 3 { |
| + t.Fatalf("add the box: %v %+v", err, c.Obligations) |
| + } |
| + if err := QueueLocationPark(moveA, "uk", "uk-box", loc, n)(c); err != nil || len(c.Obligations) != 3 { |
| + t.Fatalf("box twice: %v %+v", err, c.Obligations) |
| + } |
| + // Another region is another chore. |
| + if err := QueueLocationPark(moveA, "ie", "", loc, n)(c); err != nil || len(c.Obligations) != 4 { |
| + t.Fatalf("another region: %v %+v", err, c.Obligations) |
| + } |
| +} |
| + |
| +func TestLocationParkRefusals(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation) |
| + lane := scope("moves/" + c.Host + "/" + moveA) |
| + bad := loc |
| + bad.RequestLimit = 0 |
| + changed := loc |
| + changed.EvidenceAt = loc.EvidenceAt.Add(time.Hour) |
| + if err := QueueLocationPark(moveA, "ie", "", loc, counter())(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + for name, fn := range map[string]func(*Claim) error{ |
| + "a lane through the location door": QueueLocationPark(moveA, "ie", "", lane, counter()), |
| + "no region": QueueLocationPark(moveA, "", "", loc, counter()), |
| + "an invalid scope": QueueLocationPark(moveA, "uk", "", bad, counter()), |
| + "another scope, same chore": QueueLocationPark(moveA, "ie", "", changed, counter()), |
| + "another op": QueueLocationPark(moveB, "ie", "", loc, counter()), |
| + "no id source": QueueLocationPark(moveA, "uk", "", loc, nil), |
| + } { |
| + before := mustJSON(t, c) |
| + if err := fn(c); err == nil { |
| + t.Errorf("%s: accepted", name) |
| + } |
| + if string(mustJSON(t, c)) != string(before) { |
| + t.Errorf("%s: changed the claim", name) |
| + } |
| + } |
| +} |
| + |
| +func TestResidueParksArePerGeneration(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation) |
| + n := counter() |
| + gens := []string{"gen-4", RootGeneration, "gen-9"} |
| + if err := QueueResidueParks(moveA, "uk", loc, gens, n)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(c.Obligations) != 3 { |
| + t.Fatalf("obligations: %+v", c.Obligations) |
| + } |
| + seen := map[string]bool{} |
| + for _, o := range c.Obligations { |
| + if o.Kind != "park-generation" || o.Op != moveA || o.Region != "uk" || o.Owner != "region:uk" || o.Park == nil || !sameParkScope(*o.Park, loc) || o.Retirement == "" || o.Box != "" { |
| + t.Fatalf("residue park: %+v", o) |
| + } |
| + seen[o.Generation] = true |
| + } |
| + if !seen["gen-4"] || !seen["gen-9"] || !seen["@root"] { |
| + t.Fatalf("generations: %v", seen) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + // Each chore owns its scope: editing one does not touch another. |
| + c.Obligations[0].Park.RequestLimit = time.Hour |
| + if c.Obligations[1].Park.RequestLimit == time.Hour { |
| + t.Fatal("chores share one scope") |
| + } |
| + // A double call, a repeated generation in the list and a superset add |
| + // only what is missing, and spend ids only for that. |
| + ids := 0 |
| + next := func() string { ids++; return fmt.Sprintf("n%d", ids) } |
| + loc.RequestLimit = 2 * time.Minute |
| + c.Obligations[0].Park.RequestLimit = 2 * time.Minute |
| + if err := QueueResidueParks(moveA, "uk", loc, append(gens, "gen-4", "gen-11", "gen-11"), next)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(c.Obligations) != 4 || ids != 2 || c.Obligations[3].Generation != "gen-11" { |
| + t.Fatalf("superset: %d obligations, %d ids", len(c.Obligations), ids) |
| + } |
| + // No generations is nothing to do; an empty name is malformed. |
| + if err := QueueResidueParks(moveA, "uk", loc, nil, next)(c); err != nil || len(c.Obligations) != 4 { |
| + t.Fatalf("none: %v", err) |
| + } |
| + if err := QueueResidueParks(moveA, "uk", loc, []string{"gen-12", ""}, next)(c); !errors.Is(err, ErrMalformed) { |
| + t.Fatalf("empty generation: %v", err) |
| + } |
| + // The staging lane is not a residue location. |
| + if err := QueueResidueParks(moveA, "uk", scope("moves/"+c.Host+"/"+moveA), []string{"gen-1"}, next)(c); err == nil { |
| + t.Fatal("residue parked a lane") |
| + } |
| +} |
| + |
| +func TestLocalParkAndTheEarlierStampsPaths(t *testing.T) { |
| + c, _ := parkFixture(t) |
| + c.Inventory = map[string]InventoryEntry{"uk-box": {Layout: "nested", Paths: map[string]uint64{"/data/a": 4, "/data/b": 7}}} |
| + if !InventoryHasStamp(c, "uk-box", 4) || InventoryHasStamp(c, "uk-box", 5) || InventoryHasStamp(c, "ie-box", 4) || InventoryHasStamp(c, "uk-box", 0) { |
| + t.Fatal("InventoryHasStamp") |
| + } |
| + ids := 0 |
| + newID := func() string { ids++; return fmt.Sprintf("l%d", ids) } |
| + if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil || ids != 2 || len(c.Obligations) != 1 { |
| + t.Fatalf("double call: %v ids=%d obligations=%d", err, ids, len(c.Obligations)) |
| + } |
| + o := c.Obligations[0] |
| + if o.Kind != "park-local" || o.Box != "uk-box" || o.Owner != "box:uk-box" || o.Op != moveA || o.Region != "" || o.Park != nil { |
| + t.Fatalf("local park: %+v", o) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + // The replay with S' paths still there must not complete the chore. |
| + if err := CompleteLocalPark(moveA, "uk-box", 4)(c); !errors.Is(err, ErrWrongState) || len(c.Obligations) != 1 { |
| + t.Fatalf("complete with paths left: %v", err) |
| + } |
| + // With no S' paths the chore is completed under the op; a second time is a no-op. |
| + c.Inventory["uk-box"] = InventoryEntry{Layout: "nested", Paths: map[string]uint64{"/data/b": 7}} |
| + if err := CompleteLocalPark(moveA, "uk-box", 4)(c); err != nil || len(c.Obligations) != 0 { |
| + t.Fatalf("complete: %v %+v", err, c.Obligations) |
| + } |
| + if err := CompleteLocalPark(moveA, "uk-box", 4)(c); err != nil { |
| + t.Fatalf("replay: %v", err) |
| + } |
| + // It is the op's own chore only. |
| + if err := QueueLocalPark(moveA, "uk-box", newID)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := CompleteLocalPark(moveB, "uk-box", 4)(c); !errors.Is(err, ErrStaleOp) || len(c.Obligations) != 1 { |
| + t.Fatalf("another op: %v", err) |
| + } |
| + if err := QueueLocalPark(moveA, "", newID)(c); err == nil { |
| + t.Fatal("a local park with no box") |
| + } |
| +} |
| + |
| +// The park queue helpers own only the obligations they append. Every |
| +// other claim field, and every other op's chore, is carried through. |
| +func TestParkQueueHelpersPreserveTheRestOfTheClaim(t *testing.T) { |
| + now := time.Date(2026, 10, 6, 9, 0, 0, 0, time.UTC) |
| + build := func() *Claim { |
| + c := movingTestClaim(t) |
| + c.Holder.Generation = "gen-7" |
| + c.Holder.Session = 9 |
| + c.Holder.WriterBox = "ie-box" |
| + c.Holder.Admissions = []AdmissionRecord{{Region: "ie", Session: 9, Box: "ie-box", Generation: "gen-7", Mode: "admitted", Coverage: "covered", Covered: true, |
| + WritesLost: &WritesLost{Box: "ie-box", Generation: "gen-6", Session: 8, RemoteMaxTXID: 41}, |
| + AdditionalWritesLost: []WritesLost{{Box: "ie-box", Generation: "gen-5", Session: 7, RemoteMaxTXID: 12}}}} |
| + c.Obligations = []Obligation{{ID: "o1", Op: "old", Kind: "park-replica", Region: "se", Retirement: "r", Owner: "region:se"}} |
| + c.Lost = []string{"se"} |
| + c.RetiredIncarnations = []string{"abc"} |
| + c.QuietAfter = map[string]time.Time{"se": now} |
| + c.Touched = Touched{Complete: true, Regions: []string{"ie", "uk"}, Boxes: []string{"ie-box", "uk-box"}} |
| + c.Inventory = map[string]InventoryEntry{"ie-box": {Layout: "nested", Paths: map[string]uint64{"/data/x": 3}}} |
| + if err := BeginMove(moveA, "ops", "ie-box", "enr-ie", "uk-box", "uk", now)(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := c.Validate(); err != nil { |
| + t.Fatalf("fixture: %v", err) |
| + } |
| + return c |
| + } |
| + base := build() |
| + scope := func(loc string) ParkScope { |
| + return ParkScope{Location: loc, Incarnation: base.Holder.Incarnation, EvidenceAt: now, RequestLimit: time.Minute} |
| + } |
| + loc := scope("instances/" + base.Host + "/" + base.Holder.Incarnation) |
| + lane := scope("moves/" + base.Host + "/" + moveA) |
| + cases := map[string]struct { |
| + fn func(*Claim) error |
| + want int |
| + }{ |
| + "location+box": {QueueLocationPark(moveA, "ie", "ie-box", loc, counter()), 2}, |
| + "staging": {QueueStagingPark(moveA, "uk", lane, counter()), 1}, |
| + "local": {QueueLocalPark(moveA, "uk-box", counter()), 1}, |
| + "residue": {QueueResidueParks(moveA, "uk", loc, []string{"gen-1", RootGeneration}, counter()), 2}, |
| + } |
| + for name, tc := range cases { |
| + t.Run(name, func(t *testing.T) { |
| + c := build() |
| + before := mustJSON(t, c) |
| + if err := tc.fn(c); err != nil { |
| + t.Fatal(err) |
| + } |
| + if len(c.Obligations) != 1+tc.want || !reflect.DeepEqual(c.Obligations[0], build().Obligations[0]) { |
| + t.Fatalf("obligations: %+v", c.Obligations) |
| + } |
| + var round Claim |
| + if err := json.Unmarshal(before, &round); err != nil { |
| + t.Fatal(err) |
| + } |
| + round.Obligations = c.Obligations |
| + if !reflect.DeepEqual(round, *c) { |
| + t.Fatalf("changed more than the obligations:\nbefore %s\nafter %s", before, mustJSON(t, c)) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + }) |
| + } |
| +} |
diff --git a/internal/park/remote.go b/internal/park/remote.go| index d30560c92..5ca78e372 100644 |
| --- a/internal/park/remote.go |
| +++ b/internal/park/remote.go |
| @@ -571,3 +571,57 @@ func (e Engine) Run(ctx context.Context, w Work) error { |
| } |
| return held.Release(ctx) |
| } |
| + |
| +// Parked reports whether a retirement's archive is complete: its manifest |
| +// exists and every version in the manifest's deletion set is gone. This is |
| +// what the landing's parking phase waits for (moves spec §7.6); it is not the |
| +// acknowledgement, which still waits for the horizon and a clean listing in |
| +// Run. It takes no lease and writes nothing. |
| +// |
| +// Anything it cannot read, or a manifest that does not match the chore, is an |
| +// error and never true. A chore that is no longer in the claim returns |
| +// ErrChanged: it has been acknowledged (or never existed), and the caller |
| +// decides which from the claim it already holds. |
| +func (e Engine) Parked(ctx context.Context, w Work) (bool, error) { |
| + if e.Store == nil || e.Claims == nil || e.Bucket == "" { |
| + return false, ErrUnsafe |
| + } |
| + obs, err := hostclaim.Read(ctx, e.Claims, w.Host) |
| + if err != nil { |
| + return false, err |
| + } |
| + if obs.Absent { |
| + return false, ErrChanged |
| + } |
| + o, ok := find(obs.Claim, w) |
| + if !ok { |
| + return false, ErrChanged |
| + } |
| + if o.Park == nil { |
| + return false, ErrUnsafe |
| + } |
| + if err := validateScope(w.Host, o); err != nil { |
| + return false, err |
| + } |
| + m, err := e.manifest(ctx, w, o) |
| + if err != nil || m == nil { |
| + return false, err |
| + } |
| + // Each listed key is listed on its own: a version that is still there |
| + // shows up under its key, and a listing that fails is an error. |
| + for _, d := range m.Delete { |
| + vs, err := e.Store.ListVersions(ctx, e.Bucket, d.Key) |
| + if err != nil { |
| + return false, err |
| + } |
| + for _, v := range vs { |
| + if v.Key == d.Key && v.VersionID == "" { |
| + return false, ErrUnsafe |
| + } |
| + if v.Key == d.Key && v.VersionID == d.VersionID { |
| + return false, nil |
| + } |
| + } |
| + } |
| + return true, nil |
| +} |
diff --git a/internal/park/remote_test.go b/internal/park/remote_test.go| index 9d4f4c4a4..0e408cd34 100644 |
| --- a/internal/park/remote_test.go |
| +++ b/internal/park/remote_test.go |
| @@ -4,6 +4,7 @@ import ( |
| "context" |
| "errors" |
| "fmt" |
| + "net/url" |
| "reflect" |
| "strings" |
| "testing" |
| @@ -572,3 +573,167 @@ func TestMalformedManifestNeverDeletes(t *testing.T) { |
| 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") |
| + } |
| +} |
| + |
| +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 { |
| + c.Op.ID = move |
| + 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) |
| +} |