amadan
park helpers, fix round 1: location parks only where the region is retired; scope equals holder.location; Parked lists each key once
QueueLocationPark is legal for the source's region at landing/committed and a rolled-back target's region at restoring/released or claimed; anything else is ErrWrongState with nothing queued. Location and residue scopes must equal holder.location after trimming one trailing slash. Parked groups the deletion set by key.
Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
4 files changed,
+250
−29
internal/hostclaim/move.go+71 −16internal/hostclaim/move_test.go+117 −8internal/park/remote.go+16 −4internal/park/remote_test.go+46 −1
diff --git a/internal/hostclaim/move.go b/internal/hostclaim/move.go| index 98956e59e..0199d6a0b 100644 |
| --- a/internal/hostclaim/move.go |
| +++ b/internal/hostclaim/move.go |
| @@ -321,18 +321,31 @@ func sameParkScope(a, b ParkScope) bool { |
| // 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 { |
| +func parkQueueGuard(c *Claim, opID, region string, scope ParkScope, newID func() string) (ParkScope, error) { |
| if err := RequireOp(opID)(c); err != nil { |
| - return err |
| + return scope, 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) |
| + return scope, fmt.Errorf("%w: a move's park names its region and holder and has a way to mint ids", ErrMalformed) |
| } |
| + scope.Location = strings.TrimSuffix(scope.Location, "/") |
| if err := scope.Validate(c.Host); err != nil { |
| - return err |
| + return scope, err |
| } |
| if scope.Incarnation != c.Holder.Incarnation { |
| - return fmt.Errorf("%w: the park's incarnation is not this holder's", ErrMalformed) |
| + return scope, fmt.Errorf("%w: the park's incarnation is not this holder's", ErrMalformed) |
| + } |
| + return scope, nil |
| +} |
| + |
| +// requireHolderLocation: location and residue parks name holder.location |
| +// (§6.4), in whichever form the holder uses, not the other one. |
| +func requireHolderLocation(c *Claim, scope ParkScope) error { |
| + if isStagingLocation(scope.Location) { |
| + return fmt.Errorf("%w: a location park is not a staging lane", ErrMalformed) |
| + } |
| + if scope.Location != strings.TrimSuffix(c.Holder.Location, "/") { |
| + return fmt.Errorf("%w: the park names %q, the holder's location is %q", ErrMalformed, scope.Location, c.Holder.Location) |
| } |
| return nil |
| } |
| @@ -366,18 +379,58 @@ func queueLocalPark(c *Claim, opID, box string, newID func() string) { |
| c.Obligations = append(c.Obligations, Obligation{ID: newID(), Op: opID, Kind: "park-local", Box: box, Retirement: newID(), Owner: "box:" + box}) |
| } |
| +// Claim states in which a region's whole location is retired (§6.4, §7.8): |
| +// the landing after the commit, and a rollback from the target's release. |
| +const ( |
| + phaseLandingCommitted = "committed" |
| + phaseRestoringReleased = "released" |
| + phaseRestoringClaimed = "claimed" |
| +) |
| + |
| +// requireRetiredRegion is QueueLocationPark's lifecycle check. A whole |
| +// location is parked only where the region has stopped being authoritative: |
| +// - the source's region (any region but the move's target) once the op is |
| +// landing/committed; |
| +// - the target's region once the op is restoring and the target has |
| +// released (released, or claimed by the source). |
| +// |
| +// The source before the commit and the target after it are live, and so is |
| +// every other state, which is ErrWrongState. A box, when named, is the retiring |
| +// side's box. The claim records the target region and both boxes at the start |
| +// of the move (Intent), not the source's region, so a region that is neither |
| +// is not told apart from the source here. |
| +func requireRetiredRegion(c *Claim, region, box string) error { |
| + to := c.Op.Intent["to"] |
| + switch { |
| + case c.Op.Kind == KindLanding && c.Op.Phase == phaseLandingCommitted: |
| + if to != "" && region != to && (box == "" || box == c.Op.Intent["source"]) { |
| + return nil |
| + } |
| + case c.Op.Kind == KindRestoring && (c.Op.Phase == phaseRestoringReleased || c.Op.Phase == phaseRestoringClaimed): |
| + if to != "" && region == to && (box == "" || box == c.Op.Intent["target"]) { |
| + return nil |
| + } |
| + } |
| + return fmt.Errorf("%w: a location park in %s is not legal for %s/%q (the source's after the commit, a rolled-back target's after its release)", ErrWrongState, region, c.Op.Kind, c.Op.Phase) |
| +} |
| + |
| // 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). |
| +// target's at cleaning (requireRetiredRegion; any other state is |
| +// ErrWrongState with nothing queued). The scope is holder.location, exactly. |
| +// 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 { |
| + var err error |
| + if scope, 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 := requireHolderLocation(c, scope); err != nil { |
| + return err |
| + } |
| + if err := requireRetiredRegion(c, region, box); err != nil { |
| + return err |
| } |
| if _, err := queueReplicaPark(c, opID, region, scope, newID); err != nil { |
| return err |
| @@ -397,7 +450,8 @@ func QueueLocationPark(opID, region, box string, scope ParkScope, newID func() s |
| // (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 { |
| + var err error |
| + if scope, err = parkQueueGuard(c, opID, region, scope, newID); err != nil { |
| return err |
| } |
| // Restates moves.ScopeLocation(host, op); moves imports hostclaim, so |
| @@ -405,7 +459,7 @@ func QueueStagingPark(opID, region string, scope ParkScope, newID func() string) |
| 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) |
| + _, err = queueReplicaPark(c, opID, region, scope, newID) |
| return err |
| } |
| } |
| @@ -475,11 +529,12 @@ func CompleteLocalPark(opID, box string, earlierStamp uint64) func(*Claim) error |
| // 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 { |
| + var err error |
| + if location, 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) |
| + if err := requireHolderLocation(c, location); err != nil { |
| + return err |
| } |
| for _, g := range generations { |
| if g == "" { |
diff --git a/internal/hostclaim/move_test.go b/internal/hostclaim/move_test.go| index baa2eb24a..68fc2dfb1 100644 |
| --- a/internal/hostclaim/move_test.go |
| +++ b/internal/hostclaim/move_test.go |
| @@ -695,6 +695,12 @@ func parkFixture(t *testing.T) (*Claim, func(location string) ParkScope) { |
| return c, scope |
| } |
| +// inState puts a claim in the op state a park helper's lifecycle check reads: |
| +// the kind and phase, as the transitions that own them would leave it. |
| +func inState(c *Claim, kind, phase string) { |
| + c.Op.Kind, c.Op.Phase = kind, phase |
| +} |
| + |
| // 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) { |
| @@ -708,6 +714,10 @@ func TestLocationParkIsAParkReplicaTheStoreAccepts(t *testing.T) { |
| } |
| inc := o.Claim.Holder.Incarnation |
| scope := ParkScope{Location: "instances/" + testHost + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: 2 * time.Minute} |
| + // The commit's claim state: the source's location is parked from here. |
| + if o, err = Update(ctx, s, o, func(c *Claim) error { inState(c, KindLanding, "committed"); return nil }); err != nil { |
| + t.Fatal(err) |
| + } |
| o, err = Update(ctx, s, o, QueueLocationPark(moveA, "ie", "adi-knitting", scope, counter())) |
| if err != nil { |
| t.Fatal(err) |
| @@ -819,8 +829,9 @@ 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) |
| + inState(c, KindRestoring, "released") |
| lane := scope("moves/" + c.Host + "/" + moveA) |
| - loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation) |
| + loc := scope(c.Holder.Location) |
| n := counter() |
| for i := 0; i < 2; i++ { |
| if err := QueueStagingPark(moveA, "uk", lane, n)(c); err != nil { |
| @@ -844,15 +855,112 @@ func TestLocationParkAndStagingParkCoexistInOneRegion(t *testing.T) { |
| 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) |
| +} |
| + |
| +// A whole location is retired only where the region has stopped being |
| +// authoritative: the source after the commit (landing/committed), and a |
| +// rolled-back target (restoring, from its release). Every other state, and |
| +// the other region in either state, is refused with nothing queued and no |
| +// ids spent. |
| +func TestLocationParkOnlyWhereTheRegionIsRetired(t *testing.T) { |
| + type tc struct { |
| + kind, phase, region, box string |
| + allowed bool |
| + } |
| + cases := map[string]tc{ |
| + "source after the commit": {KindLanding, "committed", "ie", "ie-box", true}, |
| + "source after the commit, region only": {KindLanding, "committed", "ie", "", true}, |
| + "rolled-back target at cleaning": {KindRestoring, "released", "uk", "uk-box", true}, |
| + "rolled-back target after the source": {KindRestoring, "claimed", "uk", "uk-box", true}, |
| + "source before the cut": {KindMoving, PhaseMoving, "ie", "ie-box", false}, |
| + "source sealed": {KindMoving, PhaseSealed, "ie", "ie-box", false}, |
| + "source in the interim abort": {KindMoving, PhaseAborting, "ie", "ie-box", false}, |
| + "source while the landing checks": {KindLanding, "checking", "ie", "ie-box", false}, |
| + "source while the landing probes": {KindLanding, "probing", "ie", "ie-box", false}, |
| + "source while reverting": {KindReverting, "evidence", "ie", "ie-box", false}, |
| + "source after the rollback began": {KindRestoring, "released", "ie", "ie-box", false}, |
| + "target after the commit": {KindLanding, "committed", "uk", "uk-box", false}, |
| + "target before the commit": {KindLanding, "probing", "uk", "uk-box", false}, |
| + "target before its release": {KindRestoring, "releasing", "uk", "uk-box", false}, |
| + "target still serving": {KindServing, "", "uk", "uk-box", false}, |
| + "the target's box from the source": {KindLanding, "committed", "ie", "uk-box", false}, |
| + "the source's box from a rolled-back uk": {KindRestoring, "released", "uk", "ie-box", false}, |
| + } |
| + for name, tc := range cases { |
| + t.Run(name, func(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + inState(c, tc.kind, tc.phase) |
| + ids := 0 |
| + newID := func() string { ids++; return fmt.Sprintf("p%d", ids) } |
| + before := mustJSON(t, c) |
| + err := QueueLocationPark(moveA, tc.region, tc.box, scope(c.Holder.Location), newID)(c) |
| + if tc.allowed { |
| + if err != nil || len(c.Obligations) == 0 { |
| + t.Fatalf("refused: %v", err) |
| + } |
| + c.Gen++ |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + return |
| + } |
| + if !errors.Is(err, ErrWrongState) || ids != 0 || string(mustJSON(t, c)) != string(before) { |
| + t.Fatalf("err=%v ids=%d, claim changed=%v", err, ids, string(mustJSON(t, c)) != string(before)) |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// Location and residue parks name holder.location, whichever form the |
| +// holder uses (spec §6.4): the legacy instances/<host> or the |
| +// incarnation-scoped instances/<host>/<inc>, never the other one. A trailing |
| +// slash is normalised away on both sides. |
| +func TestParkLocationMustBeTheHoldersLocation(t *testing.T) { |
| + for _, form := range []string{"scoped", "legacy"} { |
| + t.Run(form, func(t *testing.T) { |
| + c, scope := parkFixture(t) |
| + scoped := "instances/" + c.Host + "/" + c.Holder.Incarnation |
| + legacy := "instances/" + c.Host |
| + own, other := scoped, legacy |
| + if form == "legacy" { |
| + own, other = legacy, scoped |
| + } |
| + c.Holder.Location = own |
| + inState(c, KindLanding, "committed") |
| + if err := c.Validate(); err != nil { |
| + t.Fatal(err) |
| + } |
| + ids := 0 |
| + newID := func() string { ids++; return fmt.Sprintf("p%d", ids) } |
| + before := mustJSON(t, c) |
| + for name, fn := range map[string]func(*Claim) error{ |
| + "location": QueueLocationPark(moveA, "ie", "ie-box", scope(other), newID), |
| + "residue": QueueResidueParks(moveA, "ie", scope(other), []string{"gen-1"}, newID), |
| + "slash/other": QueueLocationPark(moveA, "ie", "", scope(other+"/"), newID), |
| + } { |
| + if err := fn(c); !errors.Is(err, ErrMalformed) && !errors.Is(err, ErrWrongState) || ids != 0 || string(mustJSON(t, c)) != string(before) { |
| + t.Errorf("%s: err=%v ids=%d", name, err, ids) |
| + } |
| + } |
| + // The holder's own form is accepted, with or without a trailing slash, |
| + // and what is stored has none. |
| + if err := QueueLocationPark(moveA, "ie", "", scope(own+"/"), newID)(c); err != nil { |
| + t.Fatalf("own location with a slash: %v", err) |
| + } |
| + if err := QueueResidueParks(moveA, "ie", scope(own), []string{"gen-1"}, newID)(c); err != nil { |
| + t.Fatalf("own location residue: %v", err) |
| + } |
| + if len(c.Obligations) != 2 || c.Obligations[0].Park.Location != own || c.Obligations[1].Park.Location != own { |
| + t.Fatalf("stored: %+v", c.Obligations) |
| + } |
| + }) |
| } |
| } |
| func TestLocationParkRefusals(t *testing.T) { |
| c, scope := parkFixture(t) |
| - loc := scope("instances/" + c.Host + "/" + c.Holder.Incarnation) |
| + inState(c, KindLanding, "committed") |
| + loc := scope(c.Holder.Location) |
| lane := scope("moves/" + c.Host + "/" + moveA) |
| bad := loc |
| bad.RequestLimit = 0 |
| @@ -864,10 +972,10 @@ func TestLocationParkRefusals(t *testing.T) { |
| 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()), |
| + "an invalid scope": QueueLocationPark(moveA, "ie", "", 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), |
| + "no id source": QueueLocationPark(moveA, "ie", "", loc, nil), |
| } { |
| before := mustJSON(t, c) |
| if err := fn(c); err == nil { |
| @@ -1001,6 +1109,7 @@ func TestParkQueueHelpersPreserveTheRestOfTheClaim(t *testing.T) { |
| if err := BeginMove(moveA, "ops", "ie-box", "enr-ie", "uk-box", "uk", now)(c); err != nil { |
| t.Fatal(err) |
| } |
| + inState(c, KindLanding, "committed") |
| if err := c.Validate(); err != nil { |
| t.Fatalf("fixture: %v", err) |
| } |
| @@ -1010,7 +1119,7 @@ func TestParkQueueHelpersPreserveTheRestOfTheClaim(t *testing.T) { |
| 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) |
| + loc := scope(base.Holder.Location) |
| lane := scope("moves/" + base.Host + "/" + moveA) |
| cases := map[string]struct { |
| fn func(*Claim) error |
diff --git a/internal/park/remote.go b/internal/park/remote.go| index 5ca78e372..eaac7f211 100644 |
| --- a/internal/park/remote.go |
| +++ b/internal/park/remote.go |
| @@ -607,18 +607,30 @@ func (e Engine) Parked(ctx context.Context, w Work) (bool, error) { |
| if err != nil || m == nil { |
| return false, err |
| } |
| - // Each listed key is listed on its own: a version that is still there |
| + // Each distinct listed key is listed once: a version that is still there |
| // shows up under its key, and a listing that fails is an error. |
| + listed := map[string]map[string]bool{} |
| + var keys []string |
| for _, d := range m.Delete { |
| - vs, err := e.Store.ListVersions(ctx, e.Bucket, d.Key) |
| + if listed[d.Key] == nil { |
| + listed[d.Key] = map[string]bool{} |
| + keys = append(keys, d.Key) |
| + } |
| + listed[d.Key][d.VersionID] = true |
| + } |
| + for _, key := range keys { |
| + vs, err := e.Store.ListVersions(ctx, e.Bucket, key) |
| if err != nil { |
| return false, err |
| } |
| for _, v := range vs { |
| - if v.Key == d.Key && v.VersionID == "" { |
| + if v.Key != key { |
| + continue |
| + } |
| + if v.VersionID == "" { |
| return false, ErrUnsafe |
| } |
| - if v.Key == d.Key && v.VersionID == d.VersionID { |
| + if listed[key][v.VersionID] { |
| return false, nil |
| } |
| } |
diff --git a/internal/park/remote_test.go b/internal/park/remote_test.go| index 0e408cd34..84a7a978c 100644 |
| --- a/internal/park/remote_test.go |
| +++ b/internal/park/remote_test.go |
| @@ -637,6 +637,50 @@ func TestParkedReportsManifestAndDeletions(t *testing.T) { |
| } |
| } |
| +// 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() |
| @@ -707,7 +751,8 @@ func TestEngineArchivesAMoveRetirementUnderParked(t *testing.T) { |
| t.Fatal(err) |
| } |
| _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { |
| - c.Op.ID = move |
| + // 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) |
| }) |