amadan

carlos / platform Public

park: exact-version remote parking, the horizon, and the at-rest sweep (claim Plan 4 Task 2)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 3f425dd1c1a99c37dbca8471c49a3353c8a2fdab parent 2bc302f
4 files changed, +1487 −0
  • internal/park/remote.go +548 −0
  • internal/park/remote_test.go +471 −0
  • internal/park/sweep.go +256 −0
  • internal/park/sweep_test.go +212 −0
diff --git a/internal/park/remote.go b/internal/park/remote.go
new file mode 100644
index 000000000..04a715b25
--- /dev/null
+++ b/internal/park/remote.go
@@ -0,0 +1,548 @@
+package park
+
+import (
+ "bytes"
+ "context"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "io"
+ "net/url"
+ "reflect"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+const SignatureAge = 15 * time.Minute
+
+var (
+ ErrPending = errors.New("park: waiting for verification after horizon")
+ ErrUnsafe = errors.New("park: unsafe or incomplete input")
+ ErrChanged = errors.New("park: authority changed")
+)
+
+type Ref struct{ Key, VersionID string }
+
+func ref(v lease.ObjectVersion) Ref { return Ref{v.Key, v.VersionID} }
+
+type Copy struct {
+ Source lease.ObjectVersion
+ Destination Ref
+}
+type Manifest struct {
+ V int
+ Host, Obligation, Op, Retirement, Region, Location, Incarnation, Generation string
+ Archive []Copy
+ Delete []lease.ObjectVersion
+}
+type Work struct{ Host, ID, Op string }
+
+// Close stops renewal; Release conditionally removes only the held lock version.
+// Acquire must renew until Close, and Check must refuse expired/lost leases.
+type HeldLease interface {
+ Check(context.Context) error
+ Release(context.Context) error
+ Close()
+}
+type Leases interface {
+ Acquire(context.Context, string, string) (HeldLease, error)
+}
+
+type Engine struct {
+ Store lease.VersionedStore
+ Claims release.Store
+ Bucket, Region string
+ Leases Leases
+ Now func() time.Time
+ NewID func() string
+ // View is supplied by moves/Plan 7b, using precisely this fresh observation.
+ View func(context.Context, hostclaim.Observed) (View, error)
+ // Authorize checks phase-specific stop evidence/fences and chore ownership.
+ // It must not manufacture stop evidence from a stopped parent alone.
+ Authorize func(context.Context, hostclaim.Observed, hostclaim.Obligation) error
+ // After is a test crash point, called after durable effects. Nil in callers.
+ After func(string) error
+}
+
+func (e Engine) after(step string) error {
+ if e.After != nil {
+ return e.After(step)
+ }
+ return nil
+}
+func token(s string) bool {
+ return s != "" && s != "." && s != ".." && !strings.ContainsAny(s, "/\\\x00")
+}
+func hex64(s string) bool { _, err := hex.DecodeString(s); return len(s) == 64 && err == nil }
+func find(c *hostclaim.Claim, w Work) (hostclaim.Obligation, bool) {
+ if c == nil {
+ return hostclaim.Obligation{}, false
+ }
+ for _, o := range c.Obligations {
+ if o.ID == w.ID && o.Op == w.Op {
+ return o, true
+ }
+ }
+ return hostclaim.Obligation{}, false
+}
+func validateScope(host string, o hostclaim.Obligation) error {
+ p := o.Park
+ if p == nil || !hex64(p.Incarnation) || p.EvidenceAt.IsZero() || p.RequestLimit <= 0 || p.RequestLimit > time.Duration(1<<63-1)-SignatureAge || !token(o.Retirement) || o.Region == "" || o.Owner != "region:"+o.Region {
+ return ErrUnsafe
+ }
+ base := "instances/" + host
+ if p.Location != base && p.Location != base+"/"+p.Incarnation {
+ return ErrUnsafe
+ }
+ if o.Kind != "park-replica" && o.Kind != "park-generation" {
+ return ErrUnsafe
+ }
+ if o.Kind == "park-generation" && o.Generation == "" {
+ return ErrUnsafe
+ }
+ return nil
+}
+
+// Bind persists recovery inputs before the first data-bucket effect. Existing
+// Plan 1 chores have no Park field; callers supply real stop-evidence time and
+// a deployment-wide upper bound enforced by every relevant request sender.
+func (e Engine) Bind(ctx context.Context, w Work, p hostclaim.ParkScope) error {
+ obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
+ if err != nil {
+ return err
+ }
+ o, ok := find(obs.Claim, w)
+ if !ok {
+ return ErrChanged
+ }
+ if o.Park != nil {
+ if o.Park.Location == p.Location && o.Park.Incarnation == p.Incarnation && o.Park.RequestLimit == p.RequestLimit && o.Park.EvidenceAt.Equal(p.EvidenceAt) {
+ return nil
+ }
+ return ErrChanged
+ }
+ if obs.Claim.Holder == nil || p.Location != obs.Claim.Holder.Location || p.Incarnation != obs.Claim.Holder.Incarnation {
+ return ErrUnsafe
+ }
+ o.Park = &p
+ if err := validateScope(w.Host, o); err != nil {
+ return err
+ }
+ if e.Authorize == nil {
+ return ErrUnsafe
+ }
+ if err := e.Authorize(ctx, obs, o); err != nil {
+ return err
+ }
+ _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ for i := range c.Obligations {
+ if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
+ c.Obligations[i].Park = &p
+ return nil
+ }
+ }
+ return ErrChanged
+ })
+ return err
+}
+func prefixes(host, location string) []string {
+ suffix := strings.TrimPrefix(location, "instances/"+host)
+ return []string{location + "/", "moves/" + host + suffix + "/"}
+}
+
+// locationOf partitions legacy level/generation children from hex incarnation
+// children, including objects in moves/. No retired-incarnations lookup.
+func locationOf(host, key string) (string, error) {
+ for _, root := range []string{"instances/", "moves/"} {
+ base := root + host + "/"
+ if !strings.HasPrefix(key, base) {
+ continue
+ }
+ tail := strings.TrimPrefix(key, base)
+ if tail == "" {
+ return "", ErrUnsafe
+ }
+ first, _, child := strings.Cut(tail, "/")
+ if child && hex64(first) {
+ return "instances/" + host + "/" + first, nil
+ }
+ return "instances/" + host, nil
+ }
+ return "", ErrUnsafe
+}
+func isLock(host, location, key string) bool {
+ for _, p := range prefixes(host, location) {
+ if key == p+"lock.json" {
+ return true
+ }
+ }
+ return false
+}
+func (e Engine) list(ctx context.Context, host, location string) ([]lease.ObjectVersion, error) {
+ var out []lease.ObjectVersion
+ for _, p := range prefixes(host, location) {
+ vs, err := e.Store.ListVersions(ctx, e.Bucket, p)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, vs...)
+ }
+ seen := map[Ref]bool{}
+ filtered := out[:0]
+ for _, v := range out {
+ l, err := locationOf(host, v.Key)
+ if err != nil || v.VersionID == "" || seen[ref(v)] {
+ return nil, ErrUnsafe
+ }
+ seen[ref(v)] = true
+ if l == location && !isLock(host, location, v.Key) {
+ filtered = append(filtered, v)
+ }
+ }
+ sort.Slice(filtered, func(i, j int) bool {
+ a, b := filtered[i], filtered[j]
+ if a.Key == b.Key {
+ return a.VersionID < b.VersionID
+ }
+ return a.Key < b.Key
+ })
+ return filtered, nil
+}
+func archiveBase(host string, o hostclaim.Obligation) string {
+ return "parked/" + o.Park.Incarnation + "/" + host + "/" + o.Retirement + "/"
+}
+func destination(host string, o hostclaim.Obligation, v lease.ObjectVersion) string {
+ for _, p := range prefixes(host, o.Park.Location) {
+ if strings.HasPrefix(v.Key, p) {
+ kind := strings.SplitN(p, "/", 2)[0]
+ return archiveBase(host, o) + kind + "/" + strings.TrimPrefix(v.Key, p) + "@" + url.QueryEscape(v.VersionID)
+ }
+ }
+ return ""
+}
+func (e Engine) current(ctx context.Context, key string) (lease.ObjectVersion, bool, error) {
+ vs, err := e.Store.ListVersions(ctx, e.Bucket, key)
+ if err != nil {
+ return lease.ObjectVersion{}, false, err
+ }
+ var got lease.ObjectVersion
+ ok := false
+ for _, v := range vs {
+ if v.Key == key && v.IsLatest {
+ if ok || v.VersionID == "" {
+ return got, false, ErrUnsafe
+ }
+ got, ok = v, true
+ }
+ }
+ if ok && got.DeleteMarker {
+ return got, false, ErrUnsafe
+ }
+ return got, ok, nil
+}
+func (e Engine) manifest(ctx context.Context, w Work, o hostclaim.Obligation) (*Manifest, error) {
+ v, ok, err := e.current(ctx, archiveBase(w.Host, o)+"manifest.json")
+ if err != nil || !ok {
+ return nil, err
+ }
+ b, err := e.Store.GetVersion(ctx, e.Bucket, v.Key, v.VersionID)
+ if err != nil {
+ return nil, err
+ }
+ var m Manifest
+ d := json.NewDecoder(bytes.NewReader(b))
+ d.DisallowUnknownFields()
+ if err := d.Decode(&m); err != nil {
+ return nil, err
+ }
+ if _, err := d.Token(); err != io.EOF {
+ return nil, ErrUnsafe
+ }
+ if m.V != 1 || m.Host != w.Host || m.Obligation != w.ID || m.Op != w.Op || m.Retirement != o.Retirement || m.Region != o.Region || m.Location != o.Park.Location || m.Incarnation != o.Park.Incarnation || m.Generation != o.Generation {
+ return nil, ErrUnsafe
+ }
+ deletes := map[Ref]lease.ObjectVersion{}
+ for _, v := range m.Delete {
+ l, err := locationOf(w.Host, v.Key)
+ if err != nil || l != o.Park.Location || v.VersionID == "" || isLock(w.Host, l, v.Key) {
+ return nil, ErrUnsafe
+ }
+ if _, ok := deletes[ref(v)]; ok {
+ return nil, ErrUnsafe
+ }
+ deletes[ref(v)] = v
+ }
+ copies := map[Ref]bool{}
+ for _, a := range m.Archive {
+ v, ok := deletes[ref(a.Source)]
+ if !ok || v.DeleteMarker || !reflect.DeepEqual(v, a.Source) || copies[ref(v)] || a.Destination.Key != destination(w.Host, o, v) || a.Destination.VersionID == "" {
+ return nil, ErrUnsafe
+ }
+ copies[ref(v)] = true
+ }
+ for r, v := range deletes {
+ if !v.DeleteMarker && !copies[r] {
+ return nil, ErrUnsafe
+ }
+ }
+ return &m, nil
+}
+func (e Engine) observe(ctx context.Context, w Work, held HeldLease, expected hostclaim.Obligation) (hostclaim.Observed, View, error) {
+ if err := held.Check(ctx); err != nil {
+ return hostclaim.Observed{}, View{}, err
+ }
+ obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
+ if err != nil {
+ return obs, View{}, err
+ }
+ o, ok := find(obs.Claim, w)
+ if !ok || !reflect.DeepEqual(o, expected) {
+ return obs, View{}, ErrChanged
+ }
+ if err := e.Authorize(ctx, obs, o); err != nil {
+ return obs, View{}, err
+ }
+ v, err := e.View(ctx, obs)
+ if err != nil {
+ return obs, v, err
+ }
+ if v.ClaimETag != obs.ETag || v.Region != e.Region {
+ return obs, v, ErrChanged
+ }
+ 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) {
+ obs, v, err := e.observe(ctx, w, held, o)
+ if err != nil {
+ return obs, nil, err
+ }
+ live, err := protected(obs.Claim, v)
+ if err != nil {
+ return obs, nil, err
+ }
+ out := make([]lease.ObjectVersion, 0, len(vs))
+ for _, x := range vs {
+ loc, err := locationOf(w.Host, x.Key)
+ if err != nil || loc != o.Park.Location || x.VersionID == "" {
+ return obs, nil, ErrUnsafe
+ }
+ if isLock(w.Host, loc, x.Key) {
+ return obs, nil, ErrUnsafe
+ }
+ if obs.Claim.Holder != nil && obs.Claim.Op.Kind != hostclaim.KindParking && loc == obs.Claim.Holder.Location && v.Region == v.AuthoritativeRegion {
+ if x.Key == v.Marker.Key {
+ if o.Kind == "park-generation" {
+ continue
+ }
+ return obs, nil, ErrUnsafe
+ }
+ g, ok := v.GenerationOf(ref(x))
+ if !ok {
+ return obs, 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
+ }
+ } else if o.Kind == "park-generation" {
+ if v.GenerationOf == nil {
+ return obs, nil, ErrUnsafe
+ }
+ g, ok := v.GenerationOf(ref(x))
+ if !ok {
+ return obs, nil, ErrUnsafe
+ }
+ if generation(g) != o.Generation {
+ continue
+ }
+ }
+ out = append(out, x)
+ }
+ return obs, out, nil
+}
+
+// Run performs one resumable event. ErrPending is a scheduler retry, never an
+// acknowledgement. A newly rotated event is persisted before returning it.
+func (e Engine) Run(ctx context.Context, w Work) error {
+ if e.Store == nil || e.Claims == nil || e.Leases == nil || e.Now == nil || e.NewID == nil || e.View == nil || e.Authorize == nil || e.Bucket == "" || e.Region == "" {
+ return ErrUnsafe
+ }
+ initial, err := hostclaim.Read(ctx, e.Claims, w.Host)
+ if err != nil {
+ return err
+ }
+ if initial.Absent {
+ return ErrChanged
+ }
+ o, ok := find(initial.Claim, w)
+ if !ok {
+ return nil
+ }
+ if err := validateScope(w.Host, o); err != nil {
+ return err
+ }
+ if o.Region != e.Region {
+ return ErrUnsafe
+ }
+ held, err := e.Leases.Acquire(ctx, e.Bucket, o.Park.Location)
+ if err != nil {
+ return err
+ }
+ defer held.Close()
+ if _, _, err = e.observe(ctx, w, held, o); err != nil {
+ return err
+ }
+ m, err := e.manifest(ctx, w, o)
+ if err != nil {
+ return err
+ }
+ if m == nil {
+ vs, err := e.list(ctx, w.Host, o.Park.Location)
+ if err != nil {
+ return err
+ }
+ _, vs, err = e.selected(ctx, w, o, held, vs)
+ if err != nil {
+ return err
+ }
+ m = &Manifest{V: 1, Host: w.Host, Obligation: w.ID, Op: w.Op, Retirement: o.Retirement, Region: o.Region, Location: o.Park.Location, Incarnation: o.Park.Incarnation, Generation: o.Generation, Delete: vs}
+ for _, v := range vs {
+ if v.DeleteMarker {
+ continue
+ }
+ if _, _, err := e.selected(ctx, w, o, held, []lease.ObjectVersion{v}); err != nil {
+ return err
+ }
+ dst := destination(w.Host, o, v)
+ id, _, err := e.Store.CopyVersion(ctx, e.Bucket, v.Key, v.VersionID, e.Bucket, dst, true)
+ if errors.Is(err, lease.ErrDestinationExists) {
+ got, ok, getErr := e.current(ctx, dst)
+ if getErr != nil {
+ return getErr
+ }
+ if !ok || got.Size != v.Size {
+ return ErrUnsafe
+ }
+ id, err = got.VersionID, nil
+ }
+ if err != nil {
+ return err
+ }
+ if id == "" {
+ return ErrUnsafe
+ }
+ m.Archive = append(m.Archive, Copy{v, Ref{dst, id}})
+ if err := e.after("copy"); err != nil {
+ return err
+ }
+ }
+ if _, _, err = e.selected(ctx, w, o, held, m.Delete); err != nil {
+ return err
+ }
+ raw, err := json.Marshal(m)
+ if err != nil {
+ return err
+ }
+ _, err = e.Store.PutIfAbsent(ctx, e.Bucket, archiveBase(w.Host, o)+"manifest.json", raw)
+ if errors.Is(err, lease.ErrExists) {
+ m, err = e.manifest(ctx, w, o)
+ }
+ if err != nil {
+ return err
+ }
+ if m == nil {
+ return ErrUnsafe
+ }
+ if err := e.after("manifest"); err != nil {
+ return err
+ }
+ }
+ // 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})
+ if err != nil {
+ return err
+ }
+ if len(selected) != 1 {
+ return ErrUnsafe
+ }
+ if err := e.Store.DeleteVersion(ctx, e.Bucket, v.Key, v.VersionID); err != nil {
+ return err
+ }
+ if err := e.after("delete"); err != nil {
+ return err
+ }
+ }
+ started := e.Now()
+ vs, err := e.list(ctx, w.Host, o.Park.Location)
+ if err != nil {
+ return err
+ }
+ obs, vs, err := e.selected(ctx, w, o, held, vs)
+ if err != nil {
+ return err
+ }
+ if err := e.after("verify"); err != nil {
+ return err
+ }
+ deadline := o.Park.EvidenceAt.Add(SignatureAge + o.Park.RequestLimit)
+ if !started.After(deadline) {
+ return ErrPending
+ }
+ 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 {
+ for i := range c.Obligations {
+ if c.Obligations[i].ID == w.ID && c.Obligations[i].Op == w.Op {
+ c.Obligations[i].Retirement = next
+ return nil
+ }
+ }
+ return ErrChanged
+ })
+ if err != nil {
+ return err
+ }
+ return ErrPending
+ }
+ if err := e.after("before-completion"); err != nil {
+ return err
+ }
+ if err := held.Check(ctx); err != nil {
+ return err
+ }
+ _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ if c.QuietAfter == nil {
+ c.QuietAfter = map[string]time.Time{}
+ }
+ if c.QuietAfter[o.Region].Before(deadline) {
+ c.QuietAfter[o.Region] = deadline
+ }
+ if err := hostclaim.CompleteObligation(w.Op, w.ID)(c); err != nil {
+ return err
+ }
+ if c.Holder == nil && len(c.Obligations) == 0 && c.Op.Kind == hostclaim.KindParking && c.Op.Intent["park_sweep"] == "v1" {
+ return hostclaim.FreeAfterPark(c.Op.ID)(c)
+ }
+ return nil
+ })
+ if err != nil {
+ return err
+ }
+ if err := e.after("completion"); err != nil {
+ return err
+ }
+ return held.Release(ctx)
+}
diff --git a/internal/park/remote_test.go b/internal/park/remote_test.go
new file mode 100644
index 000000000..0fd8bc541
--- /dev/null
+++ b/internal/park/remote_test.go
@@ -0,0 +1,471 @@
+package park
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/lease/leasetest"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type testLease struct {
+ released int
+ failed bool
+}
+
+func (l *testLease) Check(context.Context) error {
+ if l.failed {
+ return ErrChanged
+ }
+ return nil
+}
+func (l *testLease) Release(context.Context) error { l.released++; return nil }
+func (*testLease) Close() {}
+
+type testLeases struct{ held *testLease }
+
+func (l testLeases) Acquire(context.Context, string, string) (HeldLease, error) { return l.held, nil }
+
+type fixture struct {
+ e Engine
+ b *leasetest.VersionedBucket
+ w Work
+ now time.Time
+ ob hostclaim.Obligation
+ held *testLease
+}
+
+func newFixture(t *testing.T) *fixture {
+ t.Helper()
+ ctx := t.Context()
+ now := time.Date(2026, 10, 5, 12, 0, 0, 0, time.UTC)
+ host := "shop.bab.example.com"
+ inc, err := hostclaim.Incarnation("bab", "jam", host, now)
+ if err != nil {
+ t.Fatal(err)
+ }
+ b := leasetest.NewVersionedBucket()
+ b.AddBucket("data")
+ b.PageSize = 1
+ f := &fixture{b: b, w: Work{host, "ob", "op"}, now: now.Add(16 * time.Minute), held: &testLease{}}
+ f.ob = hostclaim.Obligation{ID: "ob", Op: "op", Kind: "park-replica", Region: "ie", Retirement: "rt1", Owner: "region:ie", Park: &hostclaim.ParkScope{Location: "instances/" + host + "/" + inc, Incarnation: inc, EvidenceAt: now, RequestLimit: time.Second}}
+ st := release.NewDirStore(t.TempDir())
+ _, err = hostclaim.Update(ctx, st, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
+ c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/bab/jam/instances/" + host + ".json", Account: "bab", App: "jam", Incarnation: inc, CreatedAt: now, Location: f.ob.Park.Location, Stamp: 1}
+ c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}
+ c.Obligations = []hostclaim.Obligation{f.ob}
+ c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ serial := 1
+ f.e = Engine{Store: b, Claims: st, Bucket: "data", Region: "ie", Leases: testLeases{f.held}, Now: func() time.Time { return f.now }, NewID: func() string { serial++; return fmt.Sprintf("rt%d", serial) }, Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil }, View: func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
+ }}
+ return f
+}
+func (f *fixture) seed() []lease.ObjectVersion {
+ p := f.ob.Park.Location + "/"
+ f.b.Put("data", p+"0000/a", []byte("old"))
+ f.b.Put("data", p+"0000/a", []byte("new"))
+ f.b.BareDelete("data", p+"0000/a")
+ f.b.Put("data", p+"lock.json", []byte("lease"))
+ vs, _ := f.e.list(context.Background(), f.w.Host, f.ob.Park.Location)
+ return vs
+}
+func (f *fixture) done(t *testing.T) {
+ t.Helper()
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(obs.Claim.Obligations) != 0 {
+ t.Fatal("obligation still open")
+ }
+ if got := obs.Claim.QuietAfter["ie"]; !got.Equal(f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)) {
+ t.Fatal(got)
+ }
+ vs, err := f.e.list(t.Context(), f.w.Host, f.ob.Park.Location)
+ if err != nil || len(vs) != 0 {
+ t.Fatalf("residue: %v %v", vs, err)
+ }
+}
+func TestRemoteCrashResume(t *testing.T) {
+ for _, tc := range []struct {
+ step string
+ n int
+ }{{"copy", 1}, {"copy", 2}, {"manifest", 1}, {"delete", 1}, {"delete", 2}, {"delete", 3}, {"verify", 1}, {"before-completion", 1}, {"completion", 1}} {
+ t.Run(fmt.Sprintf("%s-%d", tc.step, tc.n), func(t *testing.T) {
+ f := newFixture(t)
+ source := f.seed()
+ stop := errors.New("crash")
+ n := 0
+ f.e.After = func(s string) error {
+ if s == tc.step {
+ n++
+ if n == tc.n {
+ return stop
+ }
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, stop) {
+ t.Fatalf("crash: %v", err)
+ }
+ if tc.step != "completion" && f.held.released != 0 {
+ t.Fatal("released before ack")
+ }
+ f.e.After = nil
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ f.done(t)
+ m, err := f.e.manifest(t.Context(), f.w, f.ob)
+ if err != nil || m == nil || len(m.Delete) != len(source) || len(m.Archive) != 2 {
+ t.Fatalf("manifest: %#v %v", m, err)
+ }
+ for _, a := range m.Archive {
+ body, err := f.b.GetVersion(t.Context(), "data", a.Destination.Key, a.Destination.VersionID)
+ if err != nil || int64(len(body)) != a.Source.Size {
+ t.Fatalf("copy: %q %v", body, err)
+ }
+ }
+ if f.b.LatestID("data", f.ob.Park.Location+"/lock.json") == "" {
+ t.Fatal("lock deleted")
+ }
+ })
+ }
+}
+func TestStorageFailuresKeepObligation(t *testing.T) {
+ for _, op := range []string{"list", "get", "copy", "put", "delete"} {
+ t.Run(op, func(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ if op == "get" {
+ f.e.After = func(s string) error {
+ if s == "manifest" {
+ return ErrPending
+ }
+ return nil
+ }
+ _ = f.e.Run(t.Context(), f.w)
+ f.e.After = nil
+ }
+ failure := errors.New("injected")
+ f.b.FailOn = func(got, key string) error {
+ if got == op {
+ return failure
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); err == nil {
+ t.Fatal("failure acknowledged")
+ }
+ obs, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || len(obs.Claim.Obligations) != 1 || len(obs.Claim.QuietAfter) != 0 {
+ t.Fatalf("claim: %v %v", obs, err)
+ }
+ f.b.FailOn = nil
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ f.done(t)
+ })
+ }
+}
+func TestPageFailureDoesNotAct(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ before := f.b.Snapshot()
+ f.b.FailListAfterPages = 1
+ if err := f.e.Run(t.Context(), f.w); err == nil {
+ t.Fatal("short listing accepted")
+ }
+ if f.b.Snapshot() != before {
+ t.Fatal("mutated on listing failure")
+ }
+}
+func TestHorizonLatePut(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
+ f.now = deadline
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
+ t.Fatal(err)
+ }
+ late := f.b.Put("data", f.ob.Park.Location+"/0000/late", []byte("late"))
+ f.now = deadline.Add(time.Nanosecond)
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
+ t.Fatal(err)
+ }
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if obs.Claim.Obligations[0].Retirement == f.ob.Retirement {
+ t.Fatal("manifest reused")
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ f.done(t)
+ if _, err := f.b.GetVersion(t.Context(), "data", f.ob.Park.Location+"/0000/late", late); !errors.Is(err, lease.ErrMirrorNotFound) {
+ t.Fatal(err)
+ }
+}
+func TestVerificationListingStartsAfterHorizon(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ deadline := f.ob.Park.EvidenceAt.Add(SignatureAge + time.Second)
+ f.now = deadline
+ f.e.After = func(s string) error {
+ if s == "verify" {
+ f.now = deadline.Add(time.Hour)
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrPending) {
+ t.Fatal(err)
+ }
+}
+func TestManifestRetryDoesNotCopy(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ f.e.After = func(s string) error {
+ if s == "delete" {
+ return ErrPending
+ }
+ return nil
+ }
+ _ = f.e.Run(t.Context(), f.w)
+ f.e.After = nil
+ f.b.FailOn = func(op, key string) error {
+ if op == "copy" {
+ t.Error("retry copied")
+ return ErrUnsafe
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+}
+func TestCASConflictDoesNotAcknowledge(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ f.e.After = func(s string) error {
+ if s == "before-completion" {
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Lost = append(c.Lost, "other-box"); return nil })
+ return err
+ }
+ return nil
+ }
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, hostclaim.ErrConflict) {
+ t.Fatal(err)
+ }
+ if f.held.released != 0 {
+ t.Fatal("released unacknowledged lease")
+ }
+ f.e.After = nil
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+}
+func TestLiveGenerationCannotBeParked(t *testing.T) {
+ f := newFixture(t)
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.seed()
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: hostclaim.RootGeneration, GenerationOf: func(Ref) (string, bool) { return hostclaim.RootGeneration, true }}, nil
+ }
+ before := f.b.Snapshot()
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrUnsafe) {
+ t.Fatal(err)
+ }
+ if before != f.b.Snapshot() {
+ t.Fatal("live data changed")
+ }
+}
+func TestBoundScopeAndQuietAfter(t *testing.T) {
+ f := newFixture(t)
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ later := f.now.Add(time.Hour)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations[0].Park = nil
+ c.QuietAfter = map[string]time.Time{"ie": later}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ if err := f.e.Bind(t.Context(), f.w, *f.ob.Park); err != nil {
+ t.Fatal(err)
+ }
+ bad := *f.ob.Park
+ bad.EvidenceAt = bad.EvidenceAt.Add(time.Second)
+ if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ obs, _ = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if !obs.Claim.QuietAfter["ie"].Equal(later) {
+ t.Fatal("quiet_after regressed")
+ }
+}
+func TestBindSurvivesSerialization(t *testing.T) {
+ 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[0].Park = nil
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ scope := *f.ob.Park
+ scope.EvidenceAt = time.Now() // Carries a monotonic reading that JSON drops.
+ if err := f.e.Bind(t.Context(), f.w, scope); err != nil {
+ t.Fatal(err)
+ }
+ before, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, at := range []time.Time{scope.EvidenceAt, scope.EvidenceAt.UTC()} {
+ same := scope
+ same.EvidenceAt = at
+ if err := f.e.Bind(t.Context(), f.w, same); err != nil {
+ t.Fatal(err)
+ }
+ }
+ for _, change := range []func(*hostclaim.ParkScope){
+ func(p *hostclaim.ParkScope) { p.Location += "/other" },
+ func(p *hostclaim.ParkScope) { p.Incarnation = strings.Repeat("f", 64) },
+ func(p *hostclaim.ParkScope) { p.RequestLimit += time.Second },
+ func(p *hostclaim.ParkScope) { p.EvidenceAt = p.EvidenceAt.Add(time.Nanosecond) },
+ } {
+ bad := scope
+ change(&bad)
+ if err := f.e.Bind(t.Context(), f.w, bad); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ }
+ after, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || after.ETag != before.ETag {
+ t.Fatalf("bind changed persisted scope: %v", err)
+ }
+}
+
+func TestLegacyLocationDoesNotIncludeIncarnationChildren(t *testing.T) {
+ f := newFixture(t)
+ key := f.ob.Park.Location + "/0000/data"
+ f.b.Put("data", key, []byte("new incarnation"))
+ vs, err := f.e.list(t.Context(), f.w.Host, "instances/"+f.w.Host)
+ if err != nil || len(vs) != 0 {
+ t.Fatalf("%v %v", vs, err)
+ }
+ for _, v := range []string{"a/b", "a@b", "a+b", "a b", "a?b"} {
+ key := destination(f.w.Host, f.ob, lease.ObjectVersion{Key: key, VersionID: v})
+ if strings.HasSuffix(key, "@"+v) {
+ t.Errorf("version not escaped: %s", v)
+ }
+ }
+}
+
+func TestDeleteParksStoppedAuthoritativeLocation(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "ie"}, nil
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ f.done(t)
+}
+func TestForeignRootGenerationRetirement(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Obligations[0].Kind = "park-generation"
+ c.Obligations[0].Generation = hostclaim.RootGeneration
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ return View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk", GenerationOf: func(Ref) (string, bool) { return "", true }}, nil
+ }
+ if err := f.e.Run(t.Context(), f.w); err != nil {
+ t.Fatal(err)
+ }
+ f.done(t)
+}
+
+func TestLeaseLossAndAuthorizationRefuse(t *testing.T) {
+ for _, phase := range []string{"before-first-effect", "before-completion", "authorize"} {
+ t.Run(phase, func(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ before := f.b.Snapshot()
+ switch phase {
+ case "before-first-effect":
+ f.held.failed = true
+ case "before-completion":
+ f.e.After = func(s string) error {
+ if s == phase {
+ f.held.failed = true
+ }
+ return nil
+ }
+ case "authorize":
+ f.e.Authorize = func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return ErrChanged }
+ }
+ if err := f.e.Run(t.Context(), f.w); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if len(obs.Claim.Obligations) != 1 || f.held.released != 0 {
+ t.Fatal("acknowledged without authority")
+ }
+ if phase != "before-completion" && f.b.Snapshot() != before {
+ t.Fatal("acted without authority")
+ }
+ })
+ }
+}
+func TestMalformedManifestNeverDeletes(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ _, err := f.b.PutIfAbsent(t.Context(), "data", archiveBase(f.w.Host, f.ob)+"manifest.json", []byte(`{"V":1,"Host":"other.example"}`))
+ if err != nil {
+ t.Fatal(err)
+ }
+ before := f.b.Snapshot()
+ if err := f.e.Run(t.Context(), f.w); err == nil {
+ t.Fatal("accepted malformed manifest")
+ }
+ if before != f.b.Snapshot() {
+ t.Fatal("deleted from bad manifest")
+ }
+}
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
new file mode 100644
index 000000000..bc19207b1
--- /dev/null
+++ b/internal/park/sweep.go
@@ -0,0 +1,256 @@
+package park
+
+import (
+ "context"
+ "fmt"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// View is a moves-owned, completely read snapshot bound to one claim ETag.
+// GenerationOf is a pure classifier for any exact reference, including a
+// version already deleted by a previous attempt. It does no storage I/O.
+// Pinned includes the exact noncurrent versions needed by the installed root.
+// MarkerKnown means its read succeeded; an absent marker may denote @root.
+type View struct {
+ ClaimETag, Region, AuthoritativeRegion string
+ MarkerKnown bool
+ MarkerGeneration string
+ Marker Ref
+ Pinned map[Ref]bool
+ GenerationOf func(Ref) (string, bool)
+}
+type Alert struct{ Location, Reason string }
+type Candidate struct{ Location, Incarnation, Generation string }
+type Decision struct {
+ Candidates []Candidate
+ Alerts []Alert
+}
+
+func generation(s string) string {
+ if s == "" {
+ return hostclaim.RootGeneration
+ }
+ return s
+}
+
+func protected(c *hostclaim.Claim, v View) (map[string]bool, error) {
+ live := map[string]bool{}
+ if c.Holder == nil || c.Op.Kind == hostclaim.KindParking {
+ return live, nil
+ }
+ if v.AuthoritativeRegion == "" {
+ return nil, ErrUnsafe
+ }
+ if v.Region != v.AuthoritativeRegion {
+ return live, nil
+ }
+ if !v.MarkerKnown || v.GenerationOf == nil {
+ return nil, ErrUnsafe
+ }
+ for r, on := range v.Pinned {
+ if on {
+ g, ok := v.GenerationOf(r)
+ if !ok {
+ return nil, ErrUnsafe
+ }
+ live[generation(g)] = true
+ }
+ }
+ h := c.Holder
+ current := generation(h.Generation)
+ marker := generation(v.MarkerGeneration)
+ live[current] = true
+ live[marker] = true
+ if current == marker {
+ return live, nil
+ }
+ // Walk the recorded chronological predecessor chain backwards. Warm
+ // admissions do not form cold unfinished links and cannot bridge a gap.
+ want, region := current, v.AuthoritativeRegion
+ for i := len(h.Admissions) - 1; i >= 0; i-- {
+ a := h.Admissions[i]
+ if generation(a.Generation) != want || a.Region != region {
+ continue
+ }
+ if a.Covered || 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 want == marker {
+ if region != "" && region != v.AuthoritativeRegion {
+ return nil, ErrUnsafe
+ }
+ return live, nil
+ }
+ }
+ return nil, ErrUnsafe
+}
+
+// Decide uses a complete namespace version listing supplied by Discover.
+// LegacyIncarnation is a caller-supplied, evidenced archive label only. It is
+// never used to discover locations, and its absence refuses legacy cleanup.
+func Decide(obs hostclaim.Observed, v View, versions []lease.ObjectVersion, legacyIncarnation string) (Decision, error) {
+ var d Decision
+ if obs.Absent || obs.Claim == nil || obs.ETag == "" || v.ClaimETag != obs.ETag || v.Region == "" {
+ return d, ErrUnsafe
+ }
+ c := obs.Claim
+ if err := c.Validate(); err != nil {
+ return d, err
+ }
+ if !c.Free() && c.Op.Kind != hostclaim.KindServing {
+ return d, nil
+ }
+ for _, o := range c.Obligations {
+ if o.Region == v.Region {
+ return d, nil
+ }
+ }
+ 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"})
+ }
+ groups := map[Candidate]bool{}
+ seen := map[Ref]bool{}
+ for _, x := range versions {
+ loc, err := locationOf(c.Host, x.Key)
+ if err != nil || x.VersionID == "" || seen[ref(x)] {
+ return Decision{}, ErrUnsafe
+ }
+ seen[ref(x)] = true
+ if isLock(c.Host, loc, x.Key) {
+ continue
+ }
+ candidate := Candidate{Location: loc}
+ if tail := strings.TrimPrefix(loc, "instances/"+c.Host+"/"); tail != loc {
+ candidate.Incarnation = tail
+ } else {
+ candidate.Incarnation = legacyIncarnation
+ }
+ if c.Holder != nil && loc == c.Holder.Location && v.Region == v.AuthoritativeRegion {
+ if liveErr != nil {
+ continue
+ }
+ if x.Key == v.Marker.Key || v.Pinned[ref(x)] {
+ continue
+ }
+ g, ok := v.GenerationOf(ref(x))
+ if !ok {
+ return Decision{}, ErrUnsafe
+ }
+ g = generation(g)
+ if live[g] {
+ continue
+ }
+ candidate.Generation = g
+ candidate.Incarnation = c.Holder.Incarnation
+ }
+ if !hex64(candidate.Incarnation) {
+ return Decision{}, ErrUnsafe
+ }
+ groups[candidate] = true
+ }
+ for c := range groups {
+ d.Candidates = append(d.Candidates, c)
+ }
+ sort.Slice(d.Candidates, func(i, j int) bool {
+ a, b := d.Candidates[i], d.Candidates[j]
+ if a.Location == b.Location {
+ return a.Generation < b.Generation
+ }
+ return a.Location < b.Location
+ })
+ return d, nil
+}
+
+// Discover expands both namespaces across every version page. This is a
+// conservative superset of delimiter discovery: deriving first child prefixes
+// from the complete version listing also discovers marker-only locations.
+func (e Engine) Discover(ctx context.Context, host string) ([]lease.ObjectVersion, error) {
+ var out []lease.ObjectVersion
+ for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} {
+ vs, err := e.Store.ListVersions(ctx, e.Bucket, p)
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, vs...)
+ }
+ return out, 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.
+func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (Decision, error) {
+ var empty Decision
+ if e.View == nil || e.NewID == nil || e.Authorize == nil {
+ return empty, ErrUnsafe
+ }
+ obs, err := hostclaim.Read(ctx, e.Claims, host)
+ if err != nil {
+ return empty, err
+ }
+ v, err := e.View(ctx, obs)
+ if err != nil {
+ return empty, err
+ }
+ vs, err := e.Discover(ctx, host)
+ if err != nil {
+ return empty, err
+ }
+ d, err := Decide(obs, v, vs, legacyIncarnation)
+ if err != nil || len(d.Candidates) == 0 {
+ return d, err
+ }
+ var op string
+ if obs.Claim.Holder != nil {
+ op = fmt.Sprintf("session:%d", obs.Claim.Holder.Session)
+ } else {
+ op = e.NewID()
+ if !token(op) {
+ return d, ErrUnsafe
+ }
+ }
+ chores := make([]hostclaim.Obligation, 0, len(d.Candidates))
+ for _, x := range d.Candidates {
+ o := hostclaim.Obligation{ID: e.NewID(), Op: op, Kind: "park-replica", Region: e.Region, Generation: x.Generation, Retirement: e.NewID(), Owner: "region:" + e.Region, Park: &hostclaim.ParkScope{Location: x.Location, Incarnation: x.Incarnation, EvidenceAt: evidenceAt, RequestLimit: requestLimit}}
+ if x.Generation != "" {
+ o.Kind = "park-generation"
+ }
+ if !token(o.ID) || validateScope(host, o) != nil {
+ return d, ErrUnsafe
+ }
+ if err := e.Authorize(ctx, obs, o); err != nil {
+ return d, err
+ }
+ chores = append(chores, o)
+ }
+ _, err = hostclaim.Update(ctx, e.Claims, obs, func(c *hostclaim.Claim) error {
+ if c.Free() {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: op, Intent: map[string]string{"park_sweep": "v1"}}
+ }
+ c.Obligations = append(c.Obligations, chores...)
+ return nil
+ })
+ if err != nil {
+ return d, err
+ }
+ for _, o := range chores {
+ if err := e.Run(ctx, Work{host, o.ID, o.Op}); err != nil {
+ return d, fmt.Errorf("sweep %s: %w", o.ID, err)
+ }
+ }
+ return d, nil
+}
diff --git a/internal/park/sweep_test.go b/internal/park/sweep_test.go
new file mode 100644
index 000000000..b6fdf9fdc
--- /dev/null
+++ b/internal/park/sweep_test.go
@@ -0,0 +1,212 @@
+package park
+
+import (
+ "context"
+ "errors"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+func sweepFixture(t *testing.T) (*fixture, hostclaim.Observed, View, []lease.ObjectVersion) {
+ f := newFixture(t)
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ obs, 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 = "g3"
+ c.Holder.Session = 3
+ c.Holder.Admissions = []hostclaim.AdmissionRecord{
+ {Session: 2, Generation: "g2", Predecessor: "g1", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
+ {Session: 3, Generation: "g3", Predecessor: "g2", Mode: "cold", Region: "ie", PredecessorRegion: "ie"},
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ p := f.ob.Park.Location + "/"
+ for _, g := range []string{"g0", "g1", "g2", "g3"} {
+ f.b.Put("data", p+g+"/data", []byte(g))
+ }
+ f.b.Put("data", p+"generation.json", []byte("older"))
+ markerID := f.b.Put("data", p+"generation.json", []byte("g1"))
+ pinned := f.b.Put("data", p+"pinned/data", []byte("baseline"))
+ f.b.Put("data", p+"pinned/data", []byte("newer"))
+ classify := func(r Ref) (string, bool) {
+ s := strings.TrimPrefix(r.Key, p)
+ g, _, ok := strings.Cut(s, "/")
+ if r.Key == p+"generation.json" {
+ return "@marker", true
+ }
+ return g, ok
+ }
+ v := View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "ie", MarkerKnown: true, MarkerGeneration: "g1", Marker: Ref{p + "generation.json", markerID}, Pinned: map[Ref]bool{{p + "pinned/data", pinned}: true}, GenerationOf: classify}
+ vs, err := f.e.Discover(t.Context(), f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ return f, obs, v, vs
+}
+func TestSweepProtectsUnfinishedRunAndPins(t *testing.T) {
+ f, obs, v, vs := sweepFixture(t)
+ d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
+ if err != nil || len(d.Alerts) != 0 || len(d.Candidates) != 1 || d.Candidates[0].Generation != "g0" {
+ t.Fatalf("%+v %v", d, err)
+ }
+ f.e.View = func(_ context.Context, o hostclaim.Observed) (View, error) {
+ x := v
+ x.ClaimETag = o.ETag
+ return x, nil
+ }
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err != nil {
+ t.Fatal(err)
+ }
+ for _, x := range vs {
+ if strings.Contains(x.Key, "/g0/") {
+ continue
+ }
+ if _, err := f.b.GetVersion(t.Context(), "data", x.Key, x.VersionID); err != nil {
+ t.Fatalf("live version lost: %+v %v", x, err)
+ }
+ }
+}
+func TestSweepDisagreementAndBusyRefuse(t *testing.T) {
+ f, obs, v, vs := sweepFixture(t)
+ v.MarkerGeneration = "unrelated"
+ d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
+ if err != nil || len(d.Alerts) != 1 || len(d.Candidates) != 0 {
+ t.Fatalf("%+v %v", d, err)
+ }
+ obs.Claim.Op.Kind = hostclaim.KindMoving
+ obs.Claim.Op.ID = "moving"
+ d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
+ if err != nil || len(d.Candidates) != 0 {
+ t.Fatalf("%+v %v", d, err)
+ }
+}
+func TestSweepDiscoversOldNamespaceAndMoves(t *testing.T) {
+ f, obs, v, vs := sweepFixture(t)
+ old := strings.Repeat("a", 64)
+ other := strings.Repeat("b", 64)
+ for _, key := range []string{"instances/" + f.w.Host + "/" + old + "/0000/a", "moves/" + f.w.Host + "/" + other + "/g-op/a", "instances/" + f.w.Host + "/0000/legacy", "moves/" + f.w.Host + "/g-old/staging"} {
+ id := f.b.Put("data", key, []byte("old"))
+ vs = append(vs, lease.ObjectVersion{Key: key, VersionID: id})
+ }
+ d, err := Decide(obs, v, vs, f.ob.Park.Incarnation)
+ if err != nil || len(d.Candidates) != 4 {
+ t.Fatalf("%+v %v", d, err)
+ }
+ // Capped retired history is irrelevant to discovery.
+ obs.Claim.RetiredIncarnations = nil
+ v.Region = "us"
+ d, err = Decide(obs, v, vs, f.ob.Park.Incarnation)
+ if err != nil || len(d.Candidates) != 4 {
+ t.Fatalf("foreign location not residue: %+v %v", d, err)
+ }
+}
+func TestSweepFreeHostQueuesAndAcknowledges(t *testing.T) {
+ f := newFixture(t)
+ f.seed()
+ obs, _ := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error { c.Obligations = nil; return hostclaim.FreeAfterPark("op")(c) })
+ if err != nil {
+ t.Fatal(err)
+ }
+ d, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second)
+ if err != nil || len(d.Candidates) != 1 {
+ t.Fatalf("%+v %v", d, err)
+ }
+ obs, err = hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || !obs.Claim.Free() || len(obs.Claim.Obligations) != 0 {
+ t.Fatalf("%+v %v", obs, err)
+ }
+}
+func TestSweepOperationOwnership(t *testing.T) {
+ for _, serving := range []bool{true, false} {
+ name := "holderless"
+ if serving {
+ name = "serving"
+ }
+ t.Run(name, func(t *testing.T) {
+ f, obs, v, _ := sweepFixture(t)
+ if !serving {
+ _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "free"}
+ return hostclaim.FreeAfterPark("free")(c)
+ })
+ 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
+ }
+ ids := []string{"generated-op", "chore", "retirement"}
+ wantOp, wantCalls := "generated-op", 3
+ if serving {
+ ids, wantOp, wantCalls = ids[1:], "session:3", 2
+ }
+ calls := 0
+ f.e.NewID = func() string {
+ if calls >= len(ids) {
+ t.Fatal("unexpected generated operation id")
+ }
+ id := ids[calls]
+ calls++
+ return id
+ }
+ authorized := false
+ f.e.Authorize = func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
+ authorized = true
+ if o.Op != wantOp {
+ return ErrUnsafe
+ }
+ return nil
+ }
+ // Stop after queueing, before Run can acknowledge the persisted chore.
+ 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)
+ }
+ if !authorized || calls != wantCalls {
+ t.Fatalf("authorized=%v ids=%d", authorized, calls)
+ }
+ queued, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if len(queued.Claim.Obligations) != 1 || queued.Claim.Obligations[0].Op != wantOp {
+ t.Fatalf("wrong chore owner: %+v", queued.Claim)
+ }
+ if serving {
+ if queued.Claim.Op.Kind != hostclaim.KindServing || queued.Claim.Holder.Session != 3 {
+ t.Fatal("serving state changed")
+ }
+ } else if queued.Claim.Holder != nil || queued.Claim.Op.ID != wantOp || queued.Claim.Op.Intent["park_sweep"] != "v1" {
+ t.Fatal("wrong holderless owner")
+ }
+ })
+ }
+}
+
+func TestSweepPageFailureAndStaleView(t *testing.T) {
+ f, obs, v, vs := sweepFixture(t)
+ before := f.b.Snapshot()
+ f.b.FailListAfterPages = 1
+ if _, err := f.e.Sweep(t.Context(), f.w.Host, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); err == nil {
+ t.Fatal("short sweep acted")
+ }
+ if before != f.b.Snapshot() {
+ t.Fatal("mutation")
+ }
+ v.ClaimETag = "old"
+ if _, err := Decide(obs, v, vs, f.ob.Park.Incarnation); !errors.Is(err, ErrUnsafe) {
+ t.Fatal(err)
+ }
+}