amadan

carlos / platform Public

hostbox, park: regional obligations and serving/free sweeps, aggregate leases (claim Plan 6 Task 10)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 92934eaa4e3e3f8b3b45b179b6dd495cdb31a2c3 parent 1ecd770
8 files changed, +1049 −0
  • internal/hostbox/aggregateleases.go +108 −0
  • internal/hostbox/aggregateleases_test.go +83 −0
  • internal/hostbox/regionalworker.go +288 −0
  • internal/hostbox/regionalworker_test.go +303 −0
  • internal/hostbox/requeststore.go +99 −0
  • internal/hostbox/requeststore_test.go +59 −0
  • internal/park/binding_test.go +51 −0
  • internal/park/sweep.go +58 −0
diff --git a/internal/hostbox/aggregateleases.go b/internal/hostbox/aggregateleases.go
new file mode 100644
index 000000000..87f135329
--- /dev/null
+++ b/internal/hostbox/aggregateleases.go
@@ -0,0 +1,108 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "sort"
+
+ "github.com/carlosframework/platform/internal/park"
+)
+
+// LeaseSet resolves every lock which can write a retirement scope. Production
+// must include the replica location and any moves staging location able to
+// write it; returning an empty set is a refusal, not permission.
+type LeaseSet func(context.Context, string, string) ([]string, error)
+
+// AggregateLeases adapts a renewing lease implementation to park.Leases.
+// Constituents are acquired in lexical order so competing workers cannot
+// deadlock by discovering the same set in a different order.
+type AggregateLeases struct {
+ Leases park.Leases
+ Resolve LeaseSet
+}
+
+func (a AggregateLeases) Acquire(ctx context.Context, bucket, scope string) (park.HeldLease, error) {
+ if ctx == nil || a.Leases == nil || a.Resolve == nil || bucket == "" || scope == "" {
+ return nil, ErrUnsupported
+ }
+ names, err := a.Resolve(ctx, bucket, scope)
+ if err != nil {
+ return nil, err
+ }
+ sort.Strings(names)
+ uniq := names[:0]
+ hasScope := false
+ for _, name := range names {
+ if name == "" {
+ return nil, ErrUnsupported
+ }
+ if len(uniq) == 0 || uniq[len(uniq)-1] != name {
+ uniq = append(uniq, name)
+ }
+ if name == scope {
+ hasScope = true
+ }
+ }
+ if len(uniq) == 0 || !hasScope {
+ return nil, ErrUnsupported
+ }
+ held := &aggregateHeld{}
+ for _, name := range uniq {
+ h, acquireErr := a.Leases.Acquire(ctx, bucket, name)
+ if acquireErr != nil {
+ return nil, errors.Join(acquireErr, held.releaseAndClose(ctx))
+ }
+ if h == nil {
+ return nil, errors.Join(ErrUnsupported, held.releaseAndClose(ctx))
+ }
+ held.leases = append(held.leases, h)
+ }
+ return held, nil
+}
+
+type aggregateHeld struct {
+ leases []park.HeldLease
+ closed bool
+}
+
+func (h *aggregateHeld) Check(ctx context.Context) error {
+ if h == nil || h.closed || len(h.leases) == 0 {
+ return ErrChanged
+ }
+ var errs []error
+ for _, lease := range h.leases {
+ errs = append(errs, lease.Check(ctx))
+ }
+ return errors.Join(errs...)
+}
+
+func (h *aggregateHeld) Release(ctx context.Context) error {
+ if h == nil || h.closed || len(h.leases) == 0 {
+ return ErrChanged
+ }
+ var errs []error
+ for i := len(h.leases) - 1; i >= 0; i-- {
+ errs = append(errs, h.leases[i].Release(ctx))
+ }
+ return errors.Join(errs...)
+}
+
+func (h *aggregateHeld) Close() {
+ if h == nil || h.closed {
+ return
+ }
+ h.closed = true
+ for i := len(h.leases) - 1; i >= 0; i-- {
+ h.leases[i].Close()
+ }
+}
+
+func (h *aggregateHeld) releaseAndClose(ctx context.Context) error {
+ var errs []error
+ for i := len(h.leases) - 1; i >= 0; i-- {
+ errs = append(errs, h.leases[i].Release(ctx))
+ h.leases[i].Close()
+ }
+ h.closed = true
+ return errors.Join(errs...)
+}
diff --git a/internal/hostbox/aggregateleases_test.go b/internal/hostbox/aggregateleases_test.go
new file mode 100644
index 000000000..3867a9891
--- /dev/null
+++ b/internal/hostbox/aggregateleases_test.go
@@ -0,0 +1,83 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "reflect"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/park"
+)
+
+type recordedLease struct {
+ name string
+ checks, releases int
+ closed int
+ checkErr, releaseErr error
+}
+
+func (l *recordedLease) Check(context.Context) error { l.checks++; return l.checkErr }
+func (l *recordedLease) Release(context.Context) error { l.releases++; return l.releaseErr }
+func (l *recordedLease) Close() { l.closed++ }
+
+type recordingLeases struct {
+ names []string
+ held map[string]*recordedLease
+ failAt string
+}
+
+func (l *recordingLeases) Acquire(_ context.Context, _, name string) (park.HeldLease, error) {
+ l.names = append(l.names, name)
+ if name == l.failAt {
+ return nil, errors.New("acquire failed")
+ }
+ h := &recordedLease{name: name}
+ l.held[name] = h
+ return h, nil
+}
+
+func TestAggregateLeasesOrderHealthReleaseAndClose(t *testing.T) {
+ base := &recordingLeases{held: map[string]*recordedLease{}}
+ a := AggregateLeases{Leases: base, Resolve: func(context.Context, string, string) ([]string, error) {
+ return []string{"moves/h/op", "instances/h/inc", "moves/h/op"}, nil
+ }}
+ held, err := a.Acquire(t.Context(), "data", "instances/h/inc")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if want := []string{"instances/h/inc", "moves/h/op"}; !reflect.DeepEqual(base.names, want) {
+ t.Fatalf("acquire order = %v, want %v", base.names, want)
+ }
+ base.held["moves/h/op"].checkErr = errors.New("lost")
+ if err := held.Check(t.Context()); err == nil || base.held["instances/h/inc"].checks != 1 || base.held["moves/h/op"].checks != 1 {
+ t.Fatalf("aggregate check = %v", err)
+ }
+ if err := held.Release(t.Context()); err != nil {
+ t.Fatal(err)
+ }
+ held.Close()
+ for name, h := range base.held {
+ if h.releases != 1 || h.closed != 1 {
+ t.Fatalf("%s: releases=%d closes=%d", name, h.releases, h.closed)
+ }
+ }
+}
+
+func TestAggregateLeasesPartialAcquireUnwindsOnlyAcquired(t *testing.T) {
+ base := &recordingLeases{held: map[string]*recordedLease{}, failAt: "b"}
+ a := AggregateLeases{Leases: base, Resolve: func(context.Context, string, string) ([]string, error) {
+ return []string{"scope", "b", "a"}, nil
+ }}
+ if _, err := a.Acquire(t.Context(), "data", "scope"); err == nil {
+ t.Fatal("acquire succeeded")
+ }
+ if want := []string{"a", "b"}; !reflect.DeepEqual(base.names, want) {
+ t.Fatalf("attempts = %v, want %v", base.names, want)
+ }
+ if h := base.held["a"]; h.releases != 1 || h.closed != 1 {
+ t.Fatalf("acquired lease not unwound: %+v", h)
+ }
+ if _, ok := base.held["scope"]; ok {
+ t.Fatal("unattempted lease was acquired")
+ }
+}
diff --git a/internal/hostbox/regionalworker.go b/internal/hostbox/regionalworker.go
new file mode 100644
index 000000000..8171abf6e
--- /dev/null
+++ b/internal/hostbox/regionalworker.go
@@ -0,0 +1,288 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "sort"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/lease"
+ "github.com/carlosframework/platform/internal/park"
+)
+
+// RegionalEnrollment must verify current enrollment, cloud identity, region,
+// bucket access and retirement assignment on every call. Construction-time
+// configuration is not authority.
+type RegionalEnrollment interface {
+ VerifyRegional(context.Context, string, string, time.Duration) error
+}
+
+// RegionalEvidence returns immutable, durable evidence covering every writer
+// of residue represented by this exact claim observation. EvidenceAt is the
+// original evidence time, never a retry time. LegacyIncarnation is required
+// when the legacy instances/<host>/ namespace contains residue.
+type RegionalEvidence interface {
+ VerifyRetirement(context.Context, hostclaim.Observed, []lease.ObjectVersion) (legacyIncarnation string, evidenceAt time.Time, err error)
+}
+
+// RegionalObligationEvidence binds legacy publication chores before their
+// first storage effect. It is required only while callers still publish
+// obligations without the immutable ParkScope.
+type RegionalObligationEvidence interface {
+ VerifyObligation(context.Context, hostclaim.Observed, hostclaim.Obligation) (hostclaim.ParkScope, error)
+}
+
+// RegionalWorker is one pass of the regional retirement worker. Production
+// wiring must provide cloud-roster verification and a durable evidence reader;
+// this adapter deliberately has no permissive defaults.
+type RegionalWorker struct {
+ Regional Regional
+ Enrollment RegionalEnrollment
+ Evidence RegionalEvidence
+}
+
+func (w RegionalWorker) validate(ctx context.Context) error {
+ e := w.Regional.Engine
+ if ctx == nil || w.Enrollment == nil || w.Evidence == nil || e.Store == nil || e.Claims == nil || e.Bucket == "" || e.Region == "" {
+ return ErrUnsupported
+ }
+ return w.Enrollment.VerifyRegional(ctx, e.Region, e.Bucket, RequestLimit)
+}
+
+// Census returns every host named by either regional data namespace. The
+// VersionedStore contract makes each call a complete paginated listing.
+func (w RegionalWorker) Census(ctx context.Context) ([]string, error) {
+ if err := w.validate(ctx); err != nil {
+ return nil, err
+ }
+ hosts := map[string]bool{}
+ for _, prefix := range []string{"instances/", "moves/"} {
+ versions, err := w.Regional.Engine.Store.ListVersions(ctx, w.Regional.Engine.Bucket, prefix)
+ if err != nil {
+ return nil, err
+ }
+ for _, v := range versions {
+ if !strings.HasPrefix(v.Key, prefix) {
+ return nil, ErrUnsupported
+ }
+ tail := strings.TrimPrefix(v.Key, prefix)
+ host, rest, ok := strings.Cut(tail, "/")
+ key, keyErr := hostclaim.Key(host)
+ if !ok || host == "" || rest == "" || v.VersionID == "" || keyErr != nil || key == "" {
+ return nil, ErrUnsupported
+ }
+ hosts[host] = true
+ }
+ }
+ claimKeys, err := w.Regional.Engine.Claims.List(ctx, "control/hosts/")
+ if err != nil {
+ return nil, err
+ }
+ for _, key := range claimKeys {
+ tail, ok := strings.CutPrefix(key, "control/hosts/")
+ if !ok {
+ return nil, ErrUnsupported
+ }
+ // List is recursive; evidence and exec objects are not claim records.
+ if strings.Contains(tail, "/") {
+ continue
+ }
+ if !strings.HasSuffix(tail, ".json") {
+ return nil, ErrUnsupported
+ }
+ host := strings.TrimSuffix(tail, ".json")
+ want, keyErr := hostclaim.Key(host)
+ if keyErr != nil || want != key {
+ return nil, ErrUnsupported
+ }
+ obs, readErr := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
+ if readErr != nil {
+ return nil, readErr
+ }
+ for _, o := range obs.Claim.Obligations {
+ if o.Region == w.Regional.Engine.Region && o.Owner == "region:"+w.Regional.Engine.Region {
+ hosts[host] = true
+ break
+ }
+ }
+ }
+ out := make([]string, 0, len(hosts))
+ for host := range hosts {
+ out = append(out, host)
+ }
+ sort.Strings(out)
+ return out, nil
+}
+
+// Run performs one bounded pass. Existing chores run in every claim state;
+// serving/free claims start sweeps; holderless park_sweep:v1 operations admit
+// residue from another region under the persisted operation. A missing claim is
+// refused because legacy/backfill disposition is a separate authority.
+func (w RegionalWorker) Run(ctx context.Context) error {
+ hosts, err := w.Census(ctx)
+ if err != nil {
+ return err
+ }
+ var errs []error
+ for _, host := range hosts {
+ if err := ctx.Err(); err != nil {
+ errs = append(errs, err)
+ break
+ }
+ if err := w.runHost(ctx, host); err != nil {
+ errs = append(errs, err)
+ }
+ }
+ return errors.Join(errs...)
+}
+
+// Loop runs a startup pass immediately, then on each serial notification or
+// one-minute reconciliation tick. Cancelling ctx stops and joins the loop.
+func (w RegionalWorker) Loop(ctx context.Context, serial <-chan struct{}, report func(error)) error {
+ if ctx == nil {
+ return ErrUnsupported
+ }
+ run := func() {
+ if err := w.Run(ctx); err != nil && report != nil {
+ report(err)
+ }
+ }
+ run()
+ ticker := time.NewTicker(time.Minute)
+ defer ticker.Stop()
+ for {
+ select {
+ case <-ctx.Done():
+ return ctx.Err()
+ case _, ok := <-serial:
+ if !ok {
+ serial = nil
+ continue
+ }
+ run()
+ case <-ticker.C:
+ run()
+ }
+ }
+}
+
+func (w RegionalWorker) runHost(ctx context.Context, host string) error {
+ if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
+ return err
+ }
+ obs, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent || obs.Claim == nil {
+ return ErrUnsupported
+ }
+ pending := false
+ works := append([]hostclaim.Obligation(nil), obs.Claim.Obligations...)
+ for _, selected := range works {
+ obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
+ if err != nil {
+ return err
+ }
+ var o hostclaim.Obligation
+ found := false
+ for _, current := range obs.Claim.Obligations {
+ if current.ID == selected.ID && current.Op == selected.Op {
+ o, found = current, true
+ break
+ }
+ }
+ if !found {
+ continue
+ }
+ if o.Owner != "region:"+w.Regional.Engine.Region || o.Region != w.Regional.Engine.Region {
+ continue
+ }
+ if o.Kind != "park-replica" && o.Kind != "park-generation" {
+ return ErrUnsupported
+ }
+ if o.Park != nil && o.Park.RequestLimit != RequestLimit {
+ return ErrUnsupported
+ }
+ if o.Park == nil {
+ provider, ok := w.Evidence.(RegionalObligationEvidence)
+ if !ok {
+ return ErrUnsupported
+ }
+ scope, err := provider.VerifyObligation(ctx, obs, o)
+ if err != nil {
+ return err
+ }
+ if scope.EvidenceAt.IsZero() || scope.RequestLimit != RequestLimit {
+ return ErrUnsupported
+ }
+ if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
+ return err
+ }
+ if err := w.Regional.Engine.Bind(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}, scope); err != nil {
+ return err
+ }
+ o.Park = &scope
+ }
+ if o.Kind == "park-replica" && o.Park != nil && strings.HasPrefix(o.Park.Location, "moves/"+host+"/") {
+ // park-staging uses the existing park-replica shape and moves scope.
+ }
+ if err := w.Regional.Run(ctx, park.Work{Host: host, ID: o.ID, Op: o.Op}); err != nil {
+ if errors.Is(err, park.ErrPending) {
+ pending = true
+ continue
+ }
+ return err
+ }
+ }
+
+ // Chore execution may have changed or freed the claim. Re-read before a
+ // possible sweep and require fresh enrollment plus evidence for that read.
+ obs, err = hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
+ if err != nil {
+ return err
+ }
+ joining := obs.Claim != nil && obs.Claim.Holder == nil && obs.Claim.Op.Kind == hostclaim.KindParking &&
+ obs.Claim.Op.ID != "" && obs.Claim.Op.Exec == "" && obs.Claim.Op.Intent["park_sweep"] == "v1"
+ if obs.Absent || obs.Claim == nil || (!obs.Claim.Free() && obs.Claim.Op.Kind != hostclaim.KindServing && !joining) {
+ return nil
+ }
+ if pending {
+ return nil
+ }
+ if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
+ return err
+ }
+ versions, err := w.Regional.Engine.Discover(ctx, host)
+ if err != nil {
+ return err
+ }
+ legacy, evidenceAt, err := w.Evidence.VerifyRetirement(ctx, obs, versions)
+ if err != nil {
+ return err
+ }
+ if evidenceAt.IsZero() {
+ return ErrUnsupported
+ }
+ current, err := hostclaim.Read(ctx, w.Regional.Engine.Claims, host)
+ if err != nil {
+ return err
+ }
+ if current.Absent || current.ETag != obs.ETag {
+ return ErrChanged
+ }
+ if err := w.Enrollment.VerifyRegional(ctx, w.Regional.Engine.Region, w.Regional.Engine.Bucket, RequestLimit); err != nil {
+ return err
+ }
+ verified, err := w.Regional.observe(ctx, host)
+ if err != nil {
+ return err
+ }
+ if verified.ETag != obs.ETag {
+ return ErrChanged
+ }
+ _, err = w.Regional.Engine.SweepObserved(ctx, obs, versions, legacy, evidenceAt, RequestLimit)
+ return err
+}
diff --git a/internal/hostbox/regionalworker_test.go b/internal/hostbox/regionalworker_test.go
new file mode 100644
index 000000000..5c83ccf09
--- /dev/null
+++ b/internal/hostbox/regionalworker_test.go
@@ -0,0 +1,303 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "fmt"
+ "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/park"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type enrollmentFunc func(context.Context, string, string, time.Duration) error
+
+func (f enrollmentFunc) VerifyRegional(ctx context.Context, region, bucket string, limit time.Duration) error {
+ return f(ctx, region, bucket, limit)
+}
+
+type evidenceFunc func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error)
+
+func (f evidenceFunc) VerifyRetirement(ctx context.Context, o hostclaim.Observed, v []lease.ObjectVersion) (string, time.Time, error) {
+ return f(ctx, o, v)
+}
+
+type workerLease struct{}
+
+func (workerLease) Check(context.Context) error { return nil }
+func (workerLease) Release(context.Context) error { return nil }
+func (workerLease) Close() {}
+
+type workerLeases struct{}
+
+func (workerLeases) Acquire(context.Context, string, string) (park.HeldLease, error) {
+ return workerLease{}, nil
+}
+
+func TestRegionalWorkerRealEngineCensusSweepAndBound(t *testing.T) {
+ ctx := t.Context()
+ evidenceAt := time.Date(2026, 10, 6, 10, 0, 0, 0, time.UTC)
+ host := "app.example"
+ inc, err := hostclaim.Incarnation("acct", "app", host, evidenceAt)
+ if err != nil {
+ t.Fatal(err)
+ }
+ location := "instances/" + host + "/" + inc
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ bucket.PageSize = 1
+ bucket.Put("data", location+"/0000/db", []byte("old"))
+ // The moves namespace participates in the top-level census independently.
+ bucket.Put("data", "moves/"+host+"/landing/lock.json", []byte("lease"))
+ claims := release.NewDirStore(t.TempDir())
+ _, err = hostclaim.Update(ctx, claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
+ c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/acct/app/instances/" + host + ".json", Account: "acct", App: "app", Incarnation: inc, CreatedAt: evidenceAt, Location: location, Stamp: 1, Session: 1}
+ c.Op = hostclaim.Op{Kind: hostclaim.KindServing, Exec: "box"}
+ c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ seq := 0
+ var limits []time.Duration
+ engine := park.Engine{
+ Store: bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: workerLeases{},
+ Now: func() time.Time { return evidenceAt.Add(18 * time.Minute) },
+ NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
+ View: func(_ context.Context, o hostclaim.Observed) (park.View, error) {
+ return park.View{ClaimETag: o.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
+ },
+ Authorize: func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
+ if o.Park != nil {
+ limits = append(limits, o.Park.RequestLimit)
+ }
+ return nil
+ },
+ }
+ worker := RegionalWorker{
+ Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }},
+ Enrollment: enrollmentFunc(func(_ context.Context, region, gotBucket string, limit time.Duration) error {
+ if region != "ie" || gotBucket != "data" {
+ t.Fatalf("binding = %s/%s", region, gotBucket)
+ }
+ if limit != 120*time.Second {
+ t.Fatalf("enforced request limit = %v", limit)
+ }
+ return nil
+ }),
+ Evidence: evidenceFunc(func(_ context.Context, o hostclaim.Observed, versions []lease.ObjectVersion) (string, time.Time, error) {
+ if o.ETag == "" || len(versions) != 2 {
+ t.Fatalf("evidence observation = etag %q, versions %d", o.ETag, len(versions))
+ }
+ return inc, evidenceAt, nil
+ }),
+ }
+ if err := worker.Run(ctx); err != nil {
+ t.Fatal(err)
+ }
+ if len(limits) == 0 {
+ t.Fatal("authorization did not see a scoped chore")
+ }
+ for _, limit := range limits {
+ if limit != RequestLimit {
+ t.Fatalf("request limit = %v", limit)
+ }
+ }
+}
+
+func TestRegionalWorkerRefusesMissingAuthorityEvidenceAndUnknownKinds(t *testing.T) {
+ base := RegionalWorker{}
+ if _, err := base.Census(t.Context()); !errors.Is(err, ErrUnsupported) {
+ t.Fatalf("missing providers: %v", err)
+ }
+
+ claims := release.NewDirStore(t.TempDir())
+ host := "unknown.example"
+ _, err := hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
+ c.Obligations = []hostclaim.Obligation{{ID: "id", Op: "op", Kind: "park-superseded", Region: "ie", Owner: "region:ie"}}
+ return nil
+ })
+ if err == nil {
+ // Claim validation itself currently refuses this pending-decision kind,
+ // which is the desired boundary. If it becomes schema-valid, runHost's
+ // explicit kind check below remains the adapter refusal.
+ w := RegionalWorker{Regional: Regional{Engine: park.Engine{Claims: claims, Region: "ie", Bucket: "data"}}, Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil })}
+ if got := w.runHost(t.Context(), host); !errors.Is(got, ErrUnsupported) {
+ t.Fatalf("unknown kind: %v", got)
+ }
+ }
+}
+
+func TestRegionalWorkerCensusIncludesQueuedClaimWithoutResidue(t *testing.T) {
+ for name, namespaces := range map[string][]string{
+ "direct-only": nil,
+ "evidence": {"evidence/fence/receipt.json"},
+ "exec": {"exec/controller.json"},
+ "both": {"evidence/fence/receipt.json", "exec/controller.json"},
+ } {
+ t.Run(name, func(t *testing.T) {
+ claims := release.NewDirStore(t.TempDir())
+ host := "queued.example"
+ created := time.Unix(1, 0).UTC()
+ inc, err := hostclaim.Incarnation("bab", "app", host, created)
+ if err != nil {
+ t.Fatal(err)
+ }
+ _, err = hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
+ c.Holder = &hostclaim.Holder{Family: "instance", Key: "apps/bab/app/instances/" + host + ".json", Account: "bab", App: "app", Incarnation: inc, CreatedAt: created, Location: "instances/" + host + "/" + inc, Stamp: 1}
+ c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "op"}
+ c.Obligations = []hostclaim.Obligation{{ID: "id", Op: "op", Kind: "park-replica", Region: "ie", Owner: "region:ie", Retirement: "retire", Park: &hostclaim.ParkScope{Location: c.Holder.Location, Incarnation: inc, EvidenceAt: time.Unix(1, 0).UTC(), RequestLimit: RequestLimit}}}
+ c.Touched = hostclaim.Touched{Complete: true, Regions: []string{"ie"}}
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, namespace := range namespaces {
+ for _, nestedHost := range []string{host, "nested-only.example"} {
+ key := "control/hosts/" + nestedHost + "/" + namespace
+ if _, err := claims.Put(t.Context(), key, []byte(`{"not_a_claim":true}`), ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ }
+ keys, err := claims.List(t.Context(), "control/hosts/")
+ if err != nil || len(keys) != 1+2*len(namespaces) {
+ t.Fatalf("recursive claim-store listing = %v, %v", keys, err)
+ }
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ w := RegionalWorker{
+ Regional: Regional{Engine: park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie"}},
+ Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }),
+ Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
+ return "", time.Time{}, errors.New("unused")
+ }),
+ }
+ hosts, err := w.Census(t.Context())
+ if err != nil || len(hosts) != 1 || hosts[0] != host {
+ t.Fatalf("census = %v, %v", hosts, err)
+ }
+ })
+ }
+}
+
+func TestRegionalWorkerLoopRunsStartupAndJoinsCancellation(t *testing.T) {
+ ctx, cancel := context.WithCancel(t.Context())
+ claims := release.NewDirStore(t.TempDir())
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ started := make(chan struct{}, 1)
+ w := RegionalWorker{
+ Regional: Regional{Engine: park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie"}},
+ Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error {
+ select {
+ case started <- struct{}{}:
+ default:
+ }
+ return nil
+ }),
+ Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
+ return "", time.Time{}, errors.New("unused")
+ }),
+ }
+ done := make(chan error, 1)
+ go func() { done <- w.Loop(ctx, nil, nil) }()
+ <-started
+ cancel()
+ if err := <-done; !errors.Is(err, context.Canceled) {
+ t.Fatalf("loop = %v", err)
+ }
+}
+
+func TestRegionalWorkerDoesNotBindEvidenceToLateVersion(t *testing.T) {
+ host := "late.example"
+ inc := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+ claims := release.NewDirStore(t.TempDir())
+ if _, err := hostclaim.Update(t.Context(), claims, hostclaim.Observed{Host: host, Absent: true}, func(*hostclaim.Claim) error { return nil }); err != nil {
+ t.Fatal(err)
+ }
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ key := "instances/" + host + "/old/data"
+ bucket.Put("data", key, []byte("evidenced"))
+ seq := 0
+ engine := park.Engine{
+ Store: bucket, Claims: claims, Bucket: "data", Region: "ie", Leases: workerLeases{},
+ Now: func() time.Time { return time.Unix(2000, 0).UTC() },
+ NewID: func() string { seq++; return fmt.Sprintf("id%d", seq) },
+ View: func(_ context.Context, obs hostclaim.Observed) (park.View, error) {
+ return park.View{ClaimETag: obs.ETag, Region: "ie", AuthoritativeRegion: "uk"}, nil
+ },
+ Authorize: func(context.Context, hostclaim.Observed, hostclaim.Obligation) error { return nil },
+ }
+ w := RegionalWorker{
+ Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }},
+ Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }),
+ Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
+ bucket.Put("data", key, []byte("late"))
+ return inc, time.Unix(1, 0).UTC(), nil
+ }),
+ }
+ if err := w.Run(t.Context()); !errors.Is(err, park.ErrChanged) {
+ t.Fatalf("Run = %v", err)
+ }
+ obs, err := hostclaim.Read(t.Context(), claims, host)
+ if err != nil || len(obs.Claim.Obligations) != 0 {
+ t.Fatalf("late residue queued under old evidence: %+v, %v", obs.Claim.Obligations, err)
+ }
+}
+
+func TestRegionalWorkerJoinsHolderlessSweepFromAnotherRegion(t *testing.T) {
+ ctx := t.Context()
+ host := "joining.example"
+ claims := release.NewDirStore(t.TempDir())
+ at := time.Unix(1, 0).UTC()
+ inc := "aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+ uk := hostclaim.Obligation{ID: "uk-chore", Op: "original-op", Kind: "park-replica", Region: "uk", Owner: "region:uk", Retirement: "uk-retire", Park: &hostclaim.ParkScope{Location: "instances/" + host + "/" + inc, Incarnation: inc, EvidenceAt: at, RequestLimit: RequestLimit}}
+ if _, err := hostclaim.Update(ctx, claims, hostclaim.Observed{Host: host, Absent: true}, func(c *hostclaim.Claim) error {
+ c.Op = hostclaim.Op{Kind: hostclaim.KindParking, ID: "original-op", Intent: map[string]string{"park_sweep": "v1"}}
+ c.Obligations = []hostclaim.Obligation{uk}
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ bucket := leasetest.NewVersionedBucket()
+ bucket.AddBucket("data")
+ bucket.Put("data", "instances/"+host+"/"+inc+"/old", []byte("residue"))
+ seq := 0
+ seen := false
+ engine := park.Engine{Claims: claims, Store: bucket, Bucket: "data", Region: "ie", Leases: workerLeases{}, Now: func() time.Time { return at.Add(18 * time.Minute) }, NewID: func() string { seq++; return fmt.Sprintf("new%d", seq) }, View: func(_ context.Context, o hostclaim.Observed) (park.View, error) {
+ return park.View{ClaimETag: o.ETag, Region: "ie"}, nil
+ }, Authorize: func(_ context.Context, _ hostclaim.Observed, o hostclaim.Obligation) error {
+ if o.Op != "original-op" {
+ t.Fatalf("replaced operation: %+v", o)
+ }
+ seen = true
+ return nil
+ }}
+ w := RegionalWorker{Regional: Regional{Engine: engine, Owner: func(context.Context, hostclaim.Observed) error { return nil }}, Enrollment: enrollmentFunc(func(context.Context, string, string, time.Duration) error { return nil }), Evidence: evidenceFunc(func(context.Context, hostclaim.Observed, []lease.ObjectVersion) (string, time.Time, error) {
+ return inc, at, nil
+ })}
+ if err := w.Run(ctx); err != nil {
+ t.Fatal(err)
+ }
+ obs, err := hostclaim.Read(ctx, claims, host)
+ if err != nil || !seen || obs.Claim.Op.ID != "original-op" {
+ t.Fatalf("join: seen=%v obs=%+v err=%v", seen, obs, err)
+ }
+ found := false
+ for _, o := range obs.Claim.Obligations {
+ if o.ID == uk.ID {
+ found = true
+ }
+ }
+ if !found {
+ t.Fatal("other region's chore lost")
+ }
+}
diff --git a/internal/hostbox/requeststore.go b/internal/hostbox/requeststore.go
new file mode 100644
index 000000000..c1c02eea3
--- /dev/null
+++ b/internal/hostbox/requeststore.go
@@ -0,0 +1,99 @@
+package hostbox
+
+import (
+ "context"
+ "io"
+ "sync"
+
+ "github.com/carlosframework/platform/internal/lease"
+)
+
+// RequestStore bounds complete SDK operations, including retries. Check must
+// inspect current captured authority and aggregate lease health. The supplied
+// context must also be cancelled by the operation owner on lease loss.
+type RequestStore struct {
+ Store lease.VersionedStore
+ Check func(context.Context) error
+}
+
+func (s RequestStore) request(ctx context.Context) (context.Context, context.CancelFunc, error) {
+ if ctx == nil || s.Store == nil || s.Check == nil {
+ return nil, nil, ErrUnsupported
+ }
+ bounded, cancel := context.WithTimeout(ctx, RequestLimit)
+ if err := s.Check(bounded); err != nil {
+ cancel()
+ return nil, nil, err
+ }
+ if err := bounded.Err(); err != nil {
+ cancel()
+ return nil, nil, err
+ }
+ return bounded, cancel, nil
+}
+func (s RequestStore) ListVersions(ctx context.Context, bucket, prefix string) ([]lease.ObjectVersion, error) {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer cancel()
+ return s.Store.ListVersions(ctx, bucket, prefix)
+}
+func (s RequestStore) GetVersion(ctx context.Context, bucket, key, version string) ([]byte, error) {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return nil, err
+ }
+ defer cancel()
+ return s.Store.GetVersion(ctx, bucket, key, version)
+}
+func (s RequestStore) OpenVersion(ctx context.Context, bucket, key, version string) (io.ReadCloser, error) {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return nil, err
+ }
+ r, err := s.Store.OpenVersion(ctx, bucket, key, version)
+ if err != nil {
+ cancel()
+ return nil, err
+ }
+ if r == nil {
+ cancel()
+ return nil, ErrUnsupported
+ }
+ return &requestReader{ReadCloser: r, cancel: cancel}, nil
+}
+func (s RequestStore) CopyVersion(ctx context.Context, sb, sk, sv, db, dk string, createOnly bool) (string, string, error) {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return "", "", err
+ }
+ defer cancel()
+ return s.Store.CopyVersion(ctx, sb, sk, sv, db, dk, createOnly)
+}
+func (s RequestStore) DeleteVersion(ctx context.Context, bucket, key, version string) error {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return err
+ }
+ defer cancel()
+ return s.Store.DeleteVersion(ctx, bucket, key, version)
+}
+func (s RequestStore) PutIfAbsent(ctx context.Context, bucket, key string, body []byte) (string, error) {
+ ctx, cancel, err := s.request(ctx)
+ if err != nil {
+ return "", err
+ }
+ defer cancel()
+ return s.Store.PutIfAbsent(ctx, bucket, key, body)
+}
+
+type requestReader struct {
+ io.ReadCloser
+ cancel context.CancelFunc
+ once sync.Once
+}
+
+func (r *requestReader) Close() error { err := r.ReadCloser.Close(); r.once.Do(r.cancel); return err }
+
+var _ lease.VersionedStore = RequestStore{}
diff --git a/internal/hostbox/requeststore_test.go b/internal/hostbox/requeststore_test.go
new file mode 100644
index 000000000..33223f93a
--- /dev/null
+++ b/internal/hostbox/requeststore_test.go
@@ -0,0 +1,59 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/lease"
+ "io"
+ "strings"
+ "testing"
+ "time"
+)
+
+type budgetStore struct {
+ lease.VersionedStore
+ seen context.Context
+ calls int
+}
+
+func (s *budgetStore) PutIfAbsent(ctx context.Context, _, _ string, _ []byte) (string, error) {
+ s.seen = ctx
+ s.calls++
+ return "version", nil
+}
+func (s *budgetStore) OpenVersion(ctx context.Context, _, _, _ string) (io.ReadCloser, error) {
+ s.seen = ctx
+ s.calls++
+ return io.NopCloser(strings.NewReader("version")), nil
+}
+func TestRequestStoreBudgetAndReadLifetime(t *testing.T) {
+ raw := &budgetStore{}
+ s := RequestStore{Store: raw, Check: func(context.Context) error { return nil }}
+ if _, err := s.PutIfAbsent(context.Background(), "b", "key", nil); err != nil {
+ t.Fatal(err)
+ }
+ deadline, ok := raw.seen.Deadline()
+ if !ok || time.Until(deadline) > RequestLimit {
+ t.Fatal(deadline)
+ }
+ if !errors.Is(raw.seen.Err(), context.Canceled) {
+ t.Fatal("request context not released")
+ }
+ r, err := s.OpenVersion(context.Background(), "b", "key", "v")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if raw.seen.Err() != nil {
+ t.Fatal("stream cancelled before read")
+ }
+ if err := r.Close(); err != nil {
+ t.Fatal(err)
+ }
+ if !errors.Is(raw.seen.Err(), context.Canceled) {
+ t.Fatal("stream context leaked")
+ }
+ s.Check = func(context.Context) error { return ErrChanged }
+ if _, err := s.PutIfAbsent(context.Background(), "b", "key", nil); err == nil || raw.calls != 2 {
+ t.Fatal(err, raw.calls)
+ }
+}
diff --git a/internal/park/binding_test.go b/internal/park/binding_test.go
new file mode 100644
index 000000000..70306003c
--- /dev/null
+++ b/internal/park/binding_test.go
@@ -0,0 +1,51 @@
+package park
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func TestSweepObservedRefusesChangedCompleteCensus(t *testing.T) {
+ f, obs, view, versions := sweepFixture(t)
+ f.e.View = func(_ context.Context, current hostclaim.Observed) (View, error) {
+ v := view
+ v.ClaimETag = current.ETag
+ return v, nil
+ }
+ // A later version of an already-listed key must be detected too; checking
+ // only current keys would silently attach old evidence to this residue.
+ f.b.Put("data", versions[0].Key, []byte("late noncurrent version"))
+ if _, err := f.e.SweepObserved(t.Context(), obs, versions, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
+ t.Fatalf("SweepObserved = %v", err)
+ }
+ current, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || len(current.Claim.Obligations) != 0 {
+ t.Fatalf("claim mutated: %+v, %v", current.Claim.Obligations, err)
+ }
+}
+
+func TestSweepObservedRefusesChangedClaim(t *testing.T) {
+ f, obs, view, versions := sweepFixture(t)
+ f.e.View = func(_ context.Context, current hostclaim.Observed) (View, error) {
+ v := view
+ v.ClaimETag = current.ETag
+ return v, nil
+ }
+ if _, err := hostclaim.Update(t.Context(), f.e.Claims, obs, func(c *hostclaim.Claim) error {
+ c.Op.Phase = "changed"
+ return nil
+ }); err != nil {
+ t.Fatal(err)
+ }
+ if _, err := f.e.SweepObserved(t.Context(), obs, versions, f.ob.Park.Incarnation, f.ob.Park.EvidenceAt, time.Second); !errors.Is(err, ErrChanged) {
+ t.Fatalf("SweepObserved = %v", err)
+ }
+ current, err := hostclaim.Read(t.Context(), f.e.Claims, f.w.Host)
+ if err != nil || len(current.Claim.Obligations) != 0 {
+ t.Fatalf("claim mutated: %+v, %v", current.Claim.Obligations, err)
+ }
+}
diff --git a/internal/park/sweep.go b/internal/park/sweep.go
index 8768139c1..4339e3a27 100644
--- a/internal/park/sweep.go
+++ b/internal/park/sweep.go
@@ -3,6 +3,7 @@ package park
import (
"context"
"fmt"
+ "reflect"
"sort"
"strings"
"time"
@@ -243,14 +244,59 @@ func parkedGenerations(c *hostclaim.Claim, v View, chores []hostclaim.Obligation
// 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) {
+ return e.sweep(ctx, hostclaim.Observed{Host: host}, nil, legacyIncarnation, evidenceAt, requestLimit, false)
+}
+
+// SweepObserved binds retirement evidence to the exact claim observation and
+// complete version census from which it was verified. Any claim or namespace
+// change requires new evidence; it is never folded into the old timestamp.
+func (e Engine) SweepObserved(ctx context.Context, expected hostclaim.Observed, versions []lease.ObjectVersion, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (Decision, error) {
+ if expected.Host == "" || expected.Absent || expected.Claim == nil || expected.ETag == "" {
+ return Decision{}, ErrUnsafe
+ }
+ return e.sweep(ctx, expected, versions, legacyIncarnation, evidenceAt, requestLimit, true)
+}
+
+func sameCensus(a, b []lease.ObjectVersion) bool {
+ if len(a) != len(b) {
+ return false
+ }
+ indexed := make(map[Ref]lease.ObjectVersion, len(a))
+ for _, version := range a {
+ r := ref(version)
+ if r.Key == "" || r.VersionID == "" {
+ return false
+ }
+ if _, duplicate := indexed[r]; duplicate {
+ return false
+ }
+ indexed[r] = version
+ }
+ seen := make(map[Ref]bool, len(b))
+ for _, version := range b {
+ r := ref(version)
+ want, ok := indexed[r]
+ if !ok || seen[r] || !reflect.DeepEqual(want, version) {
+ return false
+ }
+ seen[r] = true
+ }
+ return true
+}
+
+func (e Engine) sweep(ctx context.Context, expected hostclaim.Observed, expectedVersions []lease.ObjectVersion, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration, bound bool) (Decision, error) {
var empty Decision
if e.View == nil || e.NewID == nil || e.Authorize == nil {
return empty, ErrUnsafe
}
+ host := expected.Host
obs, err := hostclaim.Read(ctx, e.Claims, host)
if err != nil {
return empty, err
}
+ if bound && (obs.ETag != expected.ETag || !reflect.DeepEqual(obs.Claim, expected.Claim)) {
+ return empty, ErrChanged
+ }
v, err := e.View(ctx, obs)
if err != nil {
return empty, err
@@ -259,6 +305,9 @@ func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evide
if err != nil {
return empty, err
}
+ if bound && !sameCensus(vs, expectedVersions) {
+ return empty, ErrChanged
+ }
d, err := Decide(obs, v, vs, legacyIncarnation)
if err != nil || len(d.Candidates) == 0 {
return d, err
@@ -292,6 +341,15 @@ func (e Engine) Sweep(ctx context.Context, host, legacyIncarnation string, evide
if err != nil {
return d, err
}
+ if bound {
+ fresh, discoverErr := e.Discover(ctx, host)
+ if discoverErr != nil {
+ return d, discoverErr
+ }
+ if !sameCensus(fresh, expectedVersions) {
+ return d, ErrChanged
+ }
+ }
_, 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 {