amadan

carlos / platform Public

hostbox: writer transitions and regional parking composition (claim Plan 6 Task 4)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 9046d841c29137d89bbfe9a640c4c6a5cf8504a4 parent 8e5ebc6
4 files changed, +197 −0
  • internal/hostbox/park.go +60 −0
  • internal/hostbox/park_test.go +20 −0
  • internal/hostbox/writer.go +61 −0
  • internal/hostbox/writer_test.go +56 −0
diff --git a/internal/hostbox/park.go b/internal/hostbox/park.go
new file mode 100644
index 000000000..ee6b7b0bc
--- /dev/null
+++ b/internal/hostbox/park.go
@@ -0,0 +1,60 @@
+package hostbox
+
+import (
+ "context"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/park"
+)
+
+type Regional struct {
+ Engine park.Engine
+ // Owner checks complete deployment membership and this worker's assignment.
+ Owner func(context.Context, hostclaim.Observed) error
+}
+
+func (r Regional) observe(ctx context.Context, host string) (hostclaim.Observed, error) {
+ e := r.Engine
+ if e.Claims == nil || e.Store == nil || e.Leases == nil || e.View == nil || e.Authorize == nil || e.Now == nil || e.NewID == nil || e.Bucket == "" || e.Region == "" || r.Owner == nil {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
+ obs, err := hostclaim.Read(ctx, e.Claims, host)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if obs.Absent || obs.Claim == nil {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
+ if err := r.Owner(ctx, obs); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ return obs, nil
+}
+
+func (r Regional) Run(ctx context.Context, work park.Work) error {
+ obs, err := r.observe(ctx, work.Host)
+ if err != nil {
+ return err
+ }
+ for _, ob := range obs.Claim.Obligations {
+ if ob.ID != work.ID || ob.Op != work.Op {
+ continue
+ }
+ if ob.Region != r.Engine.Region || ob.Owner != "region:"+r.Engine.Region || (ob.Kind != "park-replica" && ob.Kind != "park-generation") {
+ return ErrUnsupported
+ }
+ return r.Engine.Run(ctx, work)
+ }
+ return nil // Already acknowledged; no mutation.
+}
+
+func (r Regional) Sweep(ctx context.Context, host, legacyIncarnation string, evidenceAt time.Time, requestLimit time.Duration) (park.Decision, error) {
+ if _, err := r.observe(ctx, host); err != nil {
+ return park.Decision{}, err
+ }
+ if evidenceAt.IsZero() || requestLimit <= 0 {
+ return park.Decision{}, ErrUnsupported
+ }
+ return r.Engine.Sweep(ctx, host, legacyIncarnation, evidenceAt, requestLimit)
+}
diff --git a/internal/hostbox/park_test.go b/internal/hostbox/park_test.go
new file mode 100644
index 000000000..c19c1bfc1
--- /dev/null
+++ b/internal/hostbox/park_test.go
@@ -0,0 +1,20 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/park"
+)
+
+func TestRegionalMissingAuthorityNeverRuns(t *testing.T) {
+ r := Regional{}
+ if err := r.Run(context.Background(), park.Work{Host: "shop.example", ID: "id", Op: "op"}); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+ if _, err := r.Sweep(context.Background(), "shop.example", "", time.Now(), time.Second); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+}
diff --git a/internal/hostbox/writer.go b/internal/hostbox/writer.go
new file mode 100644
index 000000000..484b45cfd
--- /dev/null
+++ b/internal/hostbox/writer.go
@@ -0,0 +1,61 @@
+package hostbox
+
+import (
+ "context"
+ "fmt"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// Writer supplies transition adapters, not stop, certificate or marker evidence.
+type Writer struct {
+ Claims release.Store
+ Data genselect.Reader
+ Bucket, Region, Box string
+ AdmissionEvidence func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error
+ PublicationEvidence func(context.Context, hostclaim.Observed) error
+ NewID func() string
+}
+
+func (w Writer) Baseline(ctx context.Context, observed hostclaim.Observed) (*genselect.Plan, error) {
+ if w.Data == nil || w.Bucket == "" || w.Region == "" || observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil {
+ return nil, ErrUnsupported
+ }
+ return genselect.Select(ctx, w.Data, genselect.Request{Bucket: w.Bucket, Region: w.Region, Host: observed.Host, Holder: *observed.Claim.Holder})
+}
+
+func (w Writer) Admit(ctx context.Context, observed hostclaim.Observed, in hostclaim.Admission, baseline *genselect.Plan) (hostclaim.Observed, error) {
+ if w.Claims == nil || w.AdmissionEvidence == nil || w.Box == "" || w.Region == "" || observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || in.Box != w.Box || in.Region != w.Region {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
+ if baseline != nil && baseline.MaxTXID != in.RemoteMaxTXID {
+ return hostclaim.Observed{}, ErrChanged
+ }
+ if err := w.AdmissionEvidence(ctx, observed, in, baseline); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ // Exactly one attempt. A conflict requires new evidence, not an internal retry.
+ return hostclaim.Update(ctx, w.Claims, observed, hostclaim.AdmitWriter(in))
+}
+
+func (w Writer) Published(ctx context.Context, observed hostclaim.Observed) (hostclaim.Observed, error) {
+ if w.Claims == nil || w.PublicationEvidence == nil || w.NewID == nil || w.Box == "" || w.Region == "" || observed.Absent || observed.Claim == nil || observed.Claim.Holder == nil || observed.Claim.Op.Exec != w.Box {
+ return hostclaim.Observed{}, ErrUnsupported
+ }
+ if err := w.PublicationEvidence(ctx, observed); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ c := observed.Claim
+ return hostclaim.Update(ctx, w.Claims, observed, hostclaim.PublishGeneration(c.Op.ID, c.Op.Phase, c.Holder.Session, c.Holder.Generation, w.Region, w.NewID))
+}
+
+func GenerationPrefix(holder hostclaim.Holder) (string, error) {
+ generation := holder.Generation
+ if holder.Session == 0 || generation == "" || generation == hostclaim.RootGeneration || generation == "." || generation == ".." || strings.ContainsAny(generation, "/\\\x00") || holder.Location == "" {
+ return "", ErrUnsupported
+ }
+ return fmt.Sprintf("%s/g-%s/", strings.TrimSuffix(holder.Location, "/"), generation), nil
+}
diff --git a/internal/hostbox/writer_test.go b/internal/hostbox/writer_test.go
new file mode 100644
index 000000000..b3d19ae74
--- /dev/null
+++ b/internal/hostbox/writer_test.go
@@ -0,0 +1,56 @@
+//go:build unix
+
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/genselect"
+ "github.com/carlosframework/platform/internal/hostclaim"
+)
+
+func TestWriterRefusesMissingEvidenceAndStaleCAS(t *testing.T) {
+ _, store, obs := workFixture(t)
+ ctx := context.Background()
+ in := hostclaim.Admission{Box: "box", Mode: "cold", NewGeneration: "next", Region: "ie", PredecessorRegion: "ie"}
+ writer := Writer{Claims: store, Box: "box", Region: "ie"}
+ if _, err := writer.Admit(ctx, obs, in, nil); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+ checks := 0
+ writer.AdmissionEvidence = func(context.Context, hostclaim.Observed, hostclaim.Admission, *genselect.Plan) error {
+ checks++
+ return nil
+ }
+ admitted, err := writer.Admit(ctx, obs, in, nil)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if admitted.Claim.Holder.Session <= obs.Claim.Holder.Session || admitted.Claim.Holder.Generation != "next" {
+ t.Fatal(admitted)
+ }
+ if _, err := writer.Admit(ctx, obs, in, nil); err == nil {
+ t.Fatal("stale observation admitted")
+ }
+ if checks != 2 {
+ t.Fatal(checks)
+ }
+ current, err := hostclaim.Read(ctx, store, obs.Host)
+ if err != nil || current.Claim.Holder.Session != admitted.Claim.Holder.Session {
+ t.Fatal(current, err)
+ }
+ if _, err := writer.Published(ctx, admitted); !errors.Is(err, ErrUnsupported) {
+ t.Fatal(err)
+ }
+ writer.NewID = func() string { return "chore" }
+ writer.PublicationEvidence = func(context.Context, hostclaim.Observed) error { return ErrChanged }
+ if _, err := writer.Published(ctx, admitted); !errors.Is(err, ErrChanged) {
+ t.Fatal(err)
+ }
+ prefix, err := GenerationPrefix(*admitted.Claim.Holder)
+ if err != nil || prefix != "instances/shop.example/g-next/" {
+ t.Fatal(prefix, err)
+ }
+}