amadan

carlos / platform Public

park: a live-location marker disagreement no longer blocks other locations' cleanup (Plan 4 review round 1)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 4b9b1d7793de976f6ac953ce609eb45a134a791c parent 0a4f963
3 files changed, +109 −3
  • internal/park/remote.go +1 −1
  • internal/park/remote_test.go +103 −0
  • internal/park/sweep.go +5 −2

CI passed — run details

diff --git a/internal/park/remote.go b/internal/park/remote.go
index 04a715b25..edf4b88bd 100644
--- a/internal/park/remote.go
+++ b/internal/park/remote.go
@@ -321,7 +321,7 @@ func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, he
if err != nil {
return obs, nil, err
}
- live, err := protected(obs.Claim, v)
+ live, err := protected(obs.Claim, v, o.Park.Location)
if err != nil {
return obs, nil, err
}
diff --git a/internal/park/remote_test.go b/internal/park/remote_test.go
index 0fd8bc541..9d4f4c4a4 100644
--- a/internal/park/remote_test.go
+++ b/internal/park/remote_test.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"fmt"
+ "reflect"
"strings"
"testing"
"time"
@@ -299,6 +300,108 @@ func TestLiveGenerationCannotBeParked(t *testing.T) {
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)
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index bc19207b1..2f468b2ae 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -38,7 +38,7 @@ func generation(s string) string {
return s
}
-func protected(c *hostclaim.Claim, v View) (map[string]bool, error) {
+func protected(c *hostclaim.Claim, v View, location string) (map[string]bool, error) {
live := map[string]bool{}
if c.Holder == nil || c.Op.Kind == hostclaim.KindParking {
return live, nil
@@ -62,6 +62,9 @@ func protected(c *hostclaim.Claim, v View) (map[string]bool, error) {
}
}
h := c.Holder
+ if location != "" && location != h.Location && !strings.HasPrefix(location, h.Location+"/") && !strings.HasPrefix(h.Location, location+"/") {
+ return live, nil
+ }
current := generation(h.Generation)
marker := generation(v.MarkerGeneration)
live[current] = true
@@ -118,7 +121,7 @@ func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, lega
return d, nil
}
}
- live, liveErr := protected(c, v)
+ live, liveErr := protected(c, v, "")
if liveErr != nil && c.Holder != nil {
d.Alerts = append(d.Alerts, Alert{c.Holder.Location, "marker does not match the claim or an unfinished admission run"})
}