amadan

carlos / platform Public

park: the sweep walks past foreign links and covers the predecessors it parks (Plan 4 pre-merge review)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 23398774b9e71eeff5ceaa5ccfc768d1a50f8c30 parent 381a68a
4 files changed, +402 −19
  • internal/park/remote.go +32 −16
  • internal/park/sweep.go +57 −3
  • internal/park/sweep_coverage_test.go +249 −0
  • internal/park/sweep_test.go +64 −0

CI passed — run details

diff --git a/internal/park/remote.go b/internal/park/remote.go
index 225596dac..983d584c0 100644
--- a/internal/park/remote.go
+++ b/internal/park/remote.go
@@ -326,49 +326,49 @@ func (e Engine) observe(ctx context.Context, w Work, held HeldLease, expected ho
}
return obs, v, nil
}
-func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, held HeldLease, vs []lease.ObjectVersion) (hostclaim.Observed, []lease.ObjectVersion, error) {
+func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, held HeldLease, vs []lease.ObjectVersion) (hostclaim.Observed, View, []lease.ObjectVersion, error) {
obs, v, err := e.observe(ctx, w, held, o)
if err != nil {
- return obs, nil, err
+ return obs, v, nil, err
}
live, err := protected(obs.Claim, v, o.Park.Location)
if err != nil {
- return obs, nil, err
+ return obs, v, nil, err
}
out := make([]lease.ObjectVersion, 0, len(vs))
for _, x := range vs {
inScope, err := scopeContains(w.Host, o.Park.Location, x.Key)
if err != nil || !inScope || x.VersionID == "" {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
if isLock(w.Host, o.Park.Location, x.Key) {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && o.Park.Location == obs.Claim.Holder.Location && v.Region == v.AuthoritativeRegion {
if x.Key == v.Marker.Key {
if o.Kind == "park-generation" {
continue
}
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
g, ok := v.GenerationOf(ref(x))
if !ok {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
g = generation(g)
if o.Kind == "park-generation" && g != o.Generation {
continue
}
if live[g] || v.Pinned[ref(x)] || ref(x) == v.Marker {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
} else if o.Kind == "park-generation" {
if v.GenerationOf == nil {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
g, ok := v.GenerationOf(ref(x))
if !ok {
- return obs, nil, ErrUnsafe
+ return obs, v, nil, ErrUnsafe
}
if generation(g) != o.Generation {
continue
@@ -376,7 +376,7 @@ func (e Engine) selected(ctx context.Context, w Work, o hostclaim.Obligation, he
}
out = append(out, x)
}
- return obs, out, nil
+ return obs, v, out, nil
}
// Run performs one resumable event. ErrPending is a scheduler retry, never an
@@ -419,7 +419,7 @@ func (e Engine) Run(ctx context.Context, w Work) error {
if err != nil {
return err
}
- _, vs, err = e.selected(ctx, w, o, held, vs)
+ _, _, vs, err = e.selected(ctx, w, o, held, vs)
if err != nil {
return err
}
@@ -428,7 +428,7 @@ func (e Engine) Run(ctx context.Context, w Work) error {
if v.DeleteMarker {
continue
}
- if _, _, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v}); err != nil {
+ if _, _, _, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v}); err != nil {
return err
}
dst := destination(w.Host, o, v)
@@ -454,7 +454,7 @@ func (e Engine) Run(ctx context.Context, w Work) error {
return err
}
}
- if _, _, err = e.selected(ctx, w, o, held, m.Delete); err != nil {
+ if _, _, _, err = e.selected(ctx, w, o, held, m.Delete); err != nil {
return err
}
raw, err := json.Marshal(m)
@@ -478,7 +478,7 @@ func (e Engine) Run(ctx context.Context, w Work) error {
// Existing manifests go directly to their recorded deletes: never re-copy
// vanished source versions. The live-set admission check still applies.
for _, v := range m.Delete {
- _, selected, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v})
+ _, _, selected, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v})
if err != nil {
return err
}
@@ -497,7 +497,7 @@ func (e Engine) Run(ctx context.Context, w Work) error {
if err != nil {
return err
}
- obs, vs, err := e.selected(ctx, w, o, held, vs)
+ obs, view, vs, err := e.selected(ctx, w, o, held, vs)
if err != nil {
return err
}
@@ -508,12 +508,25 @@ func (e Engine) Run(ctx context.Context, w Work) error {
if !started.After(deadline) {
return ErrPending
}
+ generations, err := parkedGenerations(obs.Claim, view, []hostclaim.Obligation{o}, m.Delete)
+ if err != nil {
+ return err
+ }
+ cover := func(c *hostclaim.Claim) error {
+ if c.Holder == nil {
+ return nil
+ }
+ return hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, o.Region, generations)(c)
+ }
if len(vs) > 0 {
next := e.NewID()
if !token(next) || next == o.Retirement {
return ErrUnsafe
}
_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ if err := cover(c); err != nil {
+ return err
+ }
for i := range c.Obligations {
if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
c.Obligations[i].Retirement = next
@@ -534,6 +547,9 @@ func (e Engine) Run(ctx context.Context, w Work) error {
return err
}
_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ if err := cover(c); err != nil {
+ return err
+ }
if c.QuietAfter == nil {
c.QuietAfter = map[string]time.Time{}
}
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index da3886601..8acf33fb7 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -80,17 +80,21 @@ func protected(c *hostclaim.Claim, v View, location string) (map[string]bool, er
if generation(a.Generation) != want || a.Region != region {
continue
}
- if a.Covered || a.Mode != "cold" {
+ if a.Mode != "cold" {
return nil, ErrUnsafe
}
pred := generation(a.Predecessor)
if pred == want {
return nil, ErrUnsafe
}
- live[want] = true
- live[pred] = true
want = pred
region = a.PredecessorRegion
+ if region == "" || region == v.AuthoritativeRegion {
+ if a.Covered {
+ return nil, ErrUnsafe
+ }
+ live[want] = true
+ }
if want == marker {
if region != "" && region != v.AuthoritativeRegion {
return nil, ErrUnsafe
@@ -197,6 +201,47 @@ func (e Engine) Discover(ctx context.Context, host string) ([]lease.ObjectVersio
return out, nil
}
+func parkedGenerations(c *hostclaim.Claim, v View, chores []hostclaim.Obligation, versions []lease.ObjectVersion) ([]string, error) {
+ if c.Holder == nil || len(c.Holder.Admissions) == 0 {
+ return nil, nil
+ }
+ seen := map[string]bool{}
+ for _, o := range chores {
+ if o.Park.Location != c.Holder.Location || o.Park.Incarnation != c.Holder.Incarnation {
+ continue
+ }
+ for _, x := range versions {
+ if !strings.HasPrefix(x.Key, c.Holder.Location+"/") || isLock(c.Host, o.Park.Location, x.Key) || x.Key == v.Marker.Key {
+ continue
+ }
+ inScope, err := scopeContains(c.Host, o.Park.Location, x.Key)
+ if err != nil {
+ return nil, err
+ }
+ if !inScope {
+ continue
+ }
+ if v.GenerationOf == nil {
+ return nil, ErrUnsafe
+ }
+ g, ok := v.GenerationOf(ref(x))
+ if !ok {
+ return nil, ErrUnsafe
+ }
+ g = generation(g)
+ if o.Kind != "park-generation" || o.Generation == g {
+ seen[g] = true
+ }
+ }
+ }
+ var generations []string
+ for g := range seen {
+ generations = append(generations, g)
+ }
+ sort.Strings(generations)
+ return generations, nil
+}
+
// Sweep queues every decision in one claim CAS; Run is the common executor.
// EvidenceAt is genuine prior stop/fence evidence for the residue. A sweep
// must not substitute now for evidence, even on a free host.
@@ -246,7 +291,16 @@ func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evide
}
chores = append(chores, o)
}
+ generations, err := parkedGenerations(obs.Claim, v, chores, vs)
+ if err != nil {
+ return d, err
+ }
_, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ if c.Holder != nil {
+ if err := hostclaim.CoverParkedPredecessors(obs.Claim.Op.ID, obs.Claim.Op.Phase, e.Region, generations)(c); err != nil {
+ return err
+ }
+ }
if c.Free() {
c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: op, Intent: map[string]string{"park_sweep": "v1"}}
}
diff --git a/internal/park/sweep_coverage_test.go b/internal/park/sweep_coverage_test.go
new file mode 100644
index 000000000..6fc63a407
--- /dev/null
+++ b/internal/park/sweep_coverage_test.go
@@ -0,0 +1,249 @@
+package park
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func coverageFixture(t *testing.T, foreign bool) *fixture {
+ t.Helper()
+ f := newFixture(t)
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations = nil
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
+ c.Holder.Generation = "live"
+ c.Holder.Session = 3
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Region: "uk", PredecessorRegion: "ie", Predecessor: "dead"},
+ {Region: "ie", PredecessorRegion: "uk", Predecessor: "dead"},
+ {Region: "ie", PredecessorRegion: "ie", Predecessor: "unseen"},
+ {Region: "ie", PredecessorRegion: "ie", Predecessor: "stage-only"},
+ {Region: "ie", PredecessorRegion: "ie", Predecessor: "old-only"},
+ {Region: "uk", PredecessorRegion: "ie", Predecessor: ""},
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ base := f.ob.Park.Location + "/"
+ for _, key := range []string{
+ base + "dead/data", base + "0000/data",
+ "moves/" + f.w.Host + "/stage-only/data",
+ "instances/" + f.w.Host + "/" + strings.Repeat("a", 64) + "/old-only/data",
+ } {
+ f.b.Put("data", key, []byte(key))
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ authority := "ie"
+ if foreign {
+ authority = "uk"
+ }
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: authority,
+ MarkerKnown: true, MarkerGeneration: "live", GenerationOf: func(r Ref) (string, bool) {
+ parts := strings.Split(r.Key, "/")
+ g := parts[len(parts)-2]
+ if g == "0000" {
+ g = ""
+ }
+ return g, true
+ }}, nil
+ }
+ return f
+}
+
+func TestSweepQueueCoversOnlyQueuedPredecessors(t *testing.T) {
+ for _, foreign := range []bool{false, true} {
+ name := "generation"
+ if foreign {
+ name = "replica"
+ }
+ t.Run(name, func(t *testing.T) {
+ f := coverageFixture(t, foreign)
+ before := f.b.Snapshot()
+ f.held.failed = true
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(obs.Claim.Obligations) == 0 || f.b.Snapshot() != before {
+ t.Fatal("queue boundary did not preserve pending work and storage")
+ }
+ for i, a := range obs.Claim.Holder.Admissions {
+ if want := i == 0 || i == 5; a.Covered != want {
+ t.Errorf("queue admission %d covered=%v, want %v", i, a.Covered, want)
+ }
+ }
+ })
+ }
+}
+
+func TestSweepAcknowledgementCoversOnlyRetiredPredecessors(t *testing.T) {
+ for _, scenario := range []string{"generation", "replica", "empty", "old-incarnation", "staging"} {
+ t.Run(scenario, func(t *testing.T) {
+ f := coverageFixture(t, scenario == "replica")
+ f.held.failed = true
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var chore hostclaim.Obligation
+ for _, o := range obs.Claim.Obligations {
+ match := o.Park.Location == f.ob.Park.Location && (o.Generation == "dead" || scenario == "replica")
+ if scenario == "old-incarnation" {
+ match = strings.HasSuffix(o.Park.Location, strings.Repeat("a", 64))
+ } else if scenario == "staging" {
+ match = strings.HasPrefix(o.Park.Location, "moves/")
+ }
+ if match {
+ chore = o
+ break
+ }
+ }
+ if chore.ID == "" {
+ t.Fatal("target chore was not queued")
+ }
+ if scenario == "empty" {
+ versions, err := f.e.list(t.Context(), f.w.Host, chore.Park.Location)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, x := range versions {
+ if strings.Contains(x.Key, "/dead/") {
+ if err := f.b.DeleteVersion(t.Context(), "data", x.Key, x.VersionID); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ }
+ _, err = hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ later := append([]hostclaim.AdmissionRecord(nil), c.Holder.Admissions...)
+ for i := range later {
+ later[i].Covered = false
+ }
+ c.Holder.Admissions = append(c.Holder.Admissions, later...)
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.held.failed = false
+ f.e.After = func(step string) error {
+ if step == "before-completion" {
+ return ErrPending
+ }
+ return nil
+ }
+ w := Work{f.w.Host, chore.ID, chore.Op}
+ if err := f.e.Run(t.Context(), w); !errors.Is(err, ErrPending) {
+ t.Fatal(err)
+ }
+ pending, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, a := range pending.Claim.Holder.Admissions[6:] {
+ if a.Covered {
+ t.Fatal("later admission covered before acknowledgement")
+ }
+ }
+ f.e.After = nil
+ f.b.FailOn = func(op, key string) error {
+ if op == "copy" {
+ t.Error("completion retry copied again")
+ return ErrUnsafe
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), w); err != nil {
+ t.Fatal(err)
+ }
+ completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, present := find(completed.Claim, w); present || f.held.released != 1 || completed.Claim.QuietAfter["ie"].IsZero() {
+ t.Fatal("retirement was not acknowledged")
+ }
+ for i, a := range completed.Claim.Holder.Admissions[6:] {
+ want := (i == 0 && (scenario == "generation" || scenario == "replica")) || (i == 5 && scenario == "replica")
+ if a.Covered != want {
+ t.Errorf("ack admission %d covered=%v, want %v", i, a.Covered, want)
+ }
+ }
+ })
+ }
+}
+
+func TestSweepCoverageSurvivesRetirementRotation(t *testing.T) {
+ f := coverageFixture(t, true)
+ 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.Holder.Admissions = append(c.Holder.Admissions, hostclaim.AdmissionRecord{Region: "uk", PredecessorRegion: "ie", Predecessor: "late"})
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.held.failed = true
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var w Work
+ for _, o := range obs.Claim.Obligations {
+ if o.Park.Location == f.ob.Park.Location {
+ w = Work{f.w.Host, o.ID, o.Op}
+ }
+ }
+ if w.ID == "" || obs.Claim.Holder.Admissions[6].Covered {
+ t.Fatal("late generation was queued or covered without an object")
+ }
+ f.held.failed = false
+ for _, next := range []string{"late", "last"} {
+ f.e.After = func(step string) error {
+ if step == "manifest" {
+ f.b.Put("data", f.ob.Park.Location+"/"+next+"/data", []byte(next))
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), w); !errors.Is(err, ErrPending) {
+ t.Fatalf("retirement rotation: %v", err)
+ }
+ }
+ f.e.After = nil
+ if err := f.e.Run(t.Context(), w); err != nil {
+ t.Fatal(err)
+ }
+ completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if !completed.Claim.Holder.Admissions[6].Covered || completed.Claim.Holder.Admissions[2].Covered {
+ t.Fatalf("coverage lost a retired generation or covered unseen data: %+v", completed.Claim.Holder.Admissions)
+ }
+ if _, present := find(completed.Claim, w); present {
+ t.Fatal("rotated retirement did not complete")
+ }
+}
diff --git a/internal/park/sweep_test.go b/internal/park/sweep_test.go
index 23389ffa1..0bcd3aaeb 100644
--- a/internal/park/sweep_test.go
+++ b/internal/park/sweep_test.go
@@ -188,6 +188,70 @@ func TestSweepProtectsUnfinishedRunAndPins(t *testing.T) {
}
}
}
+
+func TestSweepUnfinishedRunRegionalCoverage(t *testing.T) {
+ for _, scenario := range []string{"covered-foreign", "uncovered-foreign", "covered-local"} {
+ t.Run(scenario, func(t *testing.T) {
+ f, obs, v, _ := sweepFixture(t)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Generation: "T", Predecessor: "g1", Region: "us", PredecessorRegion: "ie", Mode: "cold", Covered: scenario == "covered-local"},
+ {Generation: "U", Predecessor: "T", Region: "uk", PredecessorRegion: "us", Mode: "cold", Covered: scenario == "covered-foreign"},
+ {Generation: "g3", Predecessor: "U", Region: "ie", PredecessorRegion: "uk", Mode: "cold", Covered: scenario == "covered-foreign"},
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, g := range []string{"T", "U"} {
+ f.b.Put("data", f.ob.Park.Location+"/"+g+"/data", []byte(g))
+ }
+ before, err := f.e.Discover(t.Context(), f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ x := v
+ x.ClaimETag = o.ETag
+ return x, nil
+ }
+ snapshot := f.b.Snapshot()
+ d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if scenario == "covered-local" {
+ if len(d.Alerts) != 1 || len(d.Candidates) != 0 || f.b.Snapshot() != snapshot {
+ t.Fatalf("covered local baseline did not refuse unchanged: %+v", d)
+ }
+ return
+ }
+ var got []string
+ for _, c := range d.Candidates {
+ got = append(got, c.Generation)
+ }
+ if len(d.Alerts) != 0 || !reflect.DeepEqual(got, []string{"T", "U", "g0", "g2"}) {
+ t.Fatalf("foreign links blocked sweep or were protected: %+v", d)
+ }
+ for _, x := range before {
+ g, _ := v.GenerationOf(ref(x))
+ _, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID)
+ if g == "T" || g == "U" || g == "g0" || g == "g2" {
+ if !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatalf("dead version retained: %+v %v", x, err)
+ }
+ } else if err != nil {
+ t.Fatalf("live version lost: %+v %v", x, err)
+ }
+ }
+ completed, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || len(completed.Claim.Obligations) != 0 {
+ t.Fatalf("sweep did not complete: %+v %v", completed, err)
+ }
+ })
+ }
+}
func TestSweepDisagreementAndBusyRefuse(t *testing.T) {
f, obs, v, vs := sweepFixture(t)
v.MarkerGeneration = "unrelated"