amadan

carlos / platform Public

hostclaim, hostbox: enrolled identity, durable wake retries and reconcile admission (claim Plan 6 Task 7)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 5b33faac3d73a584f9667881e34399847e08c48e parent bf63e25
9 files changed, +693 −0
  • internal/hostbox/identity.go +106 −0
  • internal/hostbox/identity_test.go +36 −0
  • internal/hostbox/reconcile.go +133 −0
  • internal/hostbox/reconcile_test.go +41 −0
  • internal/hostbox/work.go +13 −0
  • internal/hostclaim/claim.go +8 −0
  • internal/hostclaim/generation.go +6 −0
  • internal/hostclaim/wakeretry.go +193 −0
  • internal/hostclaim/wakeretry_test.go +157 −0
diff --git a/internal/hostbox/identity.go b/internal/hostbox/identity.go
new file mode 100644
index 000000000..36e56a0c2
--- /dev/null
+++ b/internal/hostbox/identity.go
@@ -0,0 +1,106 @@
+package hostbox
+
+import (
+ "context"
+ "crypto/x509"
+
+ "github.com/carlosframework/platform/internal/boxid"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+)
+
+// Actor is the verified projection of the sole control/host-actors.json roster.
+// Roster authenticates that document and its revision before returning a row.
+type Actor struct {
+ Box, Enrollment string
+ Revision uint64
+ Region, Bucket, Namespace string
+ Cloud CloudIdentity
+ Eligible bool
+ Quarantined bool
+}
+type CloudIdentity struct{ Provider, Account, Region, InstanceID string }
+type Roster interface {
+ Actor(context.Context, string) (Actor, error)
+}
+type BoxEvidence interface {
+ Load(context.Context, string) (boxid.Record, []byte, *boxid.Bind, error)
+}
+
+type IdentityProvider struct {
+ Roster Roster
+ Evidence BoxEvidence
+ Account, Box, Region, Bucket, Namespace string
+ Certs map[string]*x509.Certificate
+}
+
+func (p IdentityProvider) Verify(ctx context.Context) (Actor, error) {
+ if p.Roster == nil || p.Evidence == nil || p.Account == "" || p.Box == "" || p.Region == "" || p.Bucket == "" || p.Namespace == "" || len(p.Certs) == 0 {
+ return Actor{}, ErrUnsupported
+ }
+ a, err := p.Roster.Actor(ctx, p.Box)
+ if err != nil {
+ return Actor{}, err
+ }
+ if a.Box != p.Box || a.Enrollment == "" || a.Revision == 0 || !a.Eligible || a.Quarantined || a.Region != p.Region || a.Bucket != p.Bucket || a.Namespace != p.Namespace || a.Cloud.Provider != "aws" || a.Cloud.Account != p.Account || a.Cloud.Region != p.Region || a.Cloud.InstanceID == "" {
+ return Actor{}, ErrChanged
+ }
+ rec, raw, bind, err := p.Evidence.Load(ctx, p.Box)
+ if err != nil {
+ return Actor{}, err
+ }
+ if rec.Box != p.Box {
+ return Actor{}, ErrChanged
+ }
+ id, err := (boxid.Verifier{Account: p.Account, Certs: p.Certs}).Verify(rec, raw, bind)
+ if err != nil {
+ return Actor{}, err
+ }
+ if id.AccountID != a.Cloud.Account || id.Region != a.Cloud.Region || id.InstanceID != a.Cloud.InstanceID {
+ return Actor{}, ErrChanged
+ }
+ return a, nil
+}
+
+// LocalAuthority is Plan 8's nonblocking local-state reader. It checks its own
+// boot-clock deadline, loss latch, controller identity, quarantine and fences.
+// No second execution lease state machine is introduced here.
+type LocalAuthority interface {
+ Check(context.Context, Actor, *hostclaim.Claim, Selection) error
+}
+
+func (p IdentityProvider) Local(authority LocalAuthority) func(context.Context, *hostclaim.Claim, Selection) error {
+ return func(ctx context.Context, c *hostclaim.Claim, s Selection) error {
+ if authority == nil {
+ return ErrUnsupported
+ }
+ actor, err := p.Verify(ctx)
+ if err != nil {
+ return err
+ }
+ if s.Box != actor.Box || s.Region != actor.Region {
+ return ErrChanged
+ }
+ return authority.Check(ctx, actor, c, s)
+ }
+}
+
+// Startup evaluates the switch before constructing any authority provider.
+func Startup(ctx context.Context, getenv func(string) string, construct func(context.Context) (hostwrite.Selector, error)) (hostwrite.Config, error) {
+ c := hostwrite.Config{Enabled: hostwrite.Enabled(getenv)}
+ if !c.Enabled {
+ return c, nil
+ }
+ if construct == nil {
+ return c, hostwrite.ErrUnavailable
+ }
+ selector, err := construct(ctx)
+ if err != nil {
+ return c, err
+ }
+ if selector == nil {
+ return c, hostwrite.ErrUnavailable
+ }
+ c.Selector = selector
+ return c, nil
+}
diff --git a/internal/hostbox/identity_test.go b/internal/hostbox/identity_test.go
new file mode 100644
index 000000000..3782fa162
--- /dev/null
+++ b/internal/hostbox/identity_test.go
@@ -0,0 +1,36 @@
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "testing"
+)
+
+func TestStartupFlagOnceAndInert(t *testing.T) {
+ for _, value := range []string{"", " ", "0", "1"} {
+ reads, calls := 0, 0
+ cfg, err := Startup(context.Background(), func(name string) string {
+ reads++
+ if name != "CARLOS_HOST_CLAIM" {
+ t.Fatal(name)
+ }
+ return value
+ }, func(context.Context) (hostwrite.Selector, error) { calls++; return nil, hostwrite.ErrUnavailable })
+ on := value == "0" || value == "1"
+ if reads != 1 || cfg.Enabled != on || (calls == 1) != on || (err != nil) != on {
+ t.Fatalf("%q: %+v %v %d %d", value, cfg, err, reads, calls)
+ }
+ }
+}
+
+type refuseRoster struct{}
+
+func (refuseRoster) Actor(context.Context, string) (Actor, error) {
+ return Actor{}, errors.New("unverified roster")
+}
+func TestIdentityHasNoRegionOnlyFallback(t *testing.T) {
+ if _, err := (IdentityProvider{Box: "box", Region: "ie", Roster: refuseRoster{}}).Verify(context.Background()); err == nil {
+ t.Fatal("identity without boxid evidence")
+ }
+}
diff --git a/internal/hostbox/reconcile.go b/internal/hostbox/reconcile.go
new file mode 100644
index 000000000..7cf5d4a59
--- /dev/null
+++ b/internal/hostbox/reconcile.go
@@ -0,0 +1,133 @@
+package hostbox
+
+import (
+ "context"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostguard"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/park"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type RecordSelection struct {
+ Host, Key, ETag, Incarnation, Family string
+ Layout string
+ Paths []string
+ DryRun bool
+}
+type Reconciler struct {
+ Claims release.Store
+ Guard hostguard.Guard
+ Local park.Local
+ Writes hostwrite.Config
+ Box, Region string
+ Now func() time.Time
+ // CheckRecord directly checks the selected record ETag and ownership acquisition.
+ CheckRecord func(context.Context, RecordSelection) error
+ // CheckLocal includes complete census and journal recovery before initialization.
+ CheckLocal func(context.Context, hostclaim.Observed, RecordSelection) error
+ // Adopt is Plan 5's #232 backfill coordinator, never a second adoption rule.
+ Adopt func(context.Context, RecordSelection) (hostclaim.Observed, error)
+}
+type activationReader struct {
+ r Reconciler
+ selected RecordSelection
+}
+
+func (a activationReader) Read(ctx context.Context, host string) (hostclaim.Observed, error) {
+ if host != a.selected.Host {
+ return hostclaim.Observed{}, ErrChanged
+ }
+ obs, err := hostclaim.Read(ctx, a.r.Claims, host)
+ if err != nil {
+ return obs, err
+ }
+ if err := a.r.check(ctx, obs, a.selected); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ return obs, nil
+}
+func (r Reconciler) check(ctx context.Context, obs hostclaim.Observed, selected RecordSelection) error {
+ if r.Claims == nil || r.CheckRecord == nil || r.CheckLocal == nil || r.Now == nil || r.Box == "" || r.Region == "" || selected.ETag == "" || obs.Absent || obs.Claim == nil || obs.Claim.Holder == nil {
+ return ErrUnsupported
+ }
+ h := obs.Claim.Holder
+ if obs.Host != selected.Host || h.Key != selected.Key || h.Incarnation != selected.Incarnation || h.Family != selected.Family || h.Stamp == 0 {
+ return ErrChanged
+ }
+ if err := Serving(obs.Claim, Selection{Box: r.Box}); err != nil {
+ return err
+ }
+ if until, ok := obs.Claim.QuietAfter[r.Region]; ok && !r.Now().After(until) {
+ return ErrChanged
+ }
+ if err := r.CheckRecord(ctx, selected); err != nil {
+ return err
+ }
+ return r.CheckLocal(ctx, obs, selected)
+}
+func (r Reconciler) Install(ctx context.Context, selected RecordSelection, write func() error) error {
+ if !r.Writes.Enabled {
+ if selected.DryRun {
+ return nil
+ }
+ return write()
+ }
+ if write == nil || r.Claims == nil {
+ return ErrUnsupported
+ }
+ obs, err := hostclaim.Read(ctx, r.Claims, selected.Host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent {
+ if selected.DryRun {
+ return ErrUnsupported
+ }
+ if r.Adopt == nil {
+ return ErrUnsupported
+ }
+ obs, err = r.Adopt(ctx, selected)
+ if err != nil {
+ return err
+ }
+ }
+ if err := r.check(ctx, obs, selected); err != nil {
+ return err
+ }
+ if selected.DryRun {
+ return nil
+ }
+ // Activate's participant callback is not an admission validator for executors.
+ // The reader validates EVERY fresh read while Activate holds hostguard.
+ guard := r.Guard
+ guard.Claims = activationReader{r: r, selected: selected}
+ if err := guard.Activate(ctx, selected.Host, r.Box, nil); err != nil {
+ return err
+ }
+ captured, err := hostguard.Capture(guard.DataDir, selected.Host)
+ if err != nil {
+ return err
+ }
+ selector, ok := r.Writes.Selector.(interface {
+ SelectAt(context.Context, string, string, hostguard.Captured) (hostwrite.Work, error)
+ })
+ if !ok {
+ return ErrUnsupported
+ }
+ work, err := selector.SelectAt(ctx, selected.Host, "reconcile", captured)
+ if err != nil {
+ return err
+ }
+ local := r.Local
+ local.Guard = guard
+ check := func(c *hostclaim.Claim) error {
+ return r.check(ctx, hostclaim.Observed{Host: selected.Host, Claim: c}, selected)
+ }
+ if err := local.RecordInventory(ctx, selected.Host, captured, selected.Layout, selected.Paths, check); err != nil {
+ return err
+ }
+ return work.With(ctx, hostwrite.Identity{}, write)
+}
diff --git a/internal/hostbox/reconcile_test.go b/internal/hostbox/reconcile_test.go
new file mode 100644
index 000000000..71f87c248
--- /dev/null
+++ b/internal/hostbox/reconcile_test.go
@@ -0,0 +1,41 @@
+//go:build unix
+
+package hostbox
+
+import (
+ "context"
+ "errors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/park"
+ "testing"
+ "time"
+)
+
+func TestReconcileDryRunAndExecutorValidation(t *testing.T) {
+ source, store, obs := workFixture(t)
+ h := obs.Claim.Holder
+ selected := RecordSelection{Host: obs.Host, Key: h.Key, ETag: "record-etag", Incarnation: h.Incarnation, Family: h.Family, Layout: "flat", Paths: []string{"shop.example.db"}, DryRun: true}
+ checks, writes := 0, 0
+ r := Reconciler{Claims: store, Guard: source.Guard, Local: park.Local{Guard: source.Guard, Claims: store, DataDir: source.Guard.DataDir, Box: "box"}, Writes: hostwrite.Config{Enabled: true, Selector: source}, Box: "box", Region: "ie", Now: func() time.Time { return time.Unix(100, 0) }, CheckRecord: func(context.Context, RecordSelection) error { return nil }, CheckLocal: func(context.Context, hostclaim.Observed, RecordSelection) error { checks++; return nil }}
+ if err := r.Install(context.Background(), selected, func() error { writes++; return nil }); err != nil {
+ t.Fatal(err)
+ }
+ if writes != 0 || checks != 1 {
+ t.Fatal(writes, checks)
+ }
+ selected.DryRun = false
+ r.CheckLocal = func(context.Context, hostclaim.Observed, RecordSelection) error {
+ checks++
+ if checks >= 3 {
+ return errors.New("quarantined after precheck")
+ }
+ return nil
+ }
+ if err := r.Install(context.Background(), selected, func() error { writes++; return nil }); err == nil {
+ t.Fatal("executor bypassed fresh local check")
+ }
+ if writes != 0 {
+ t.Fatal("row materialized")
+ }
+}
diff --git a/internal/hostbox/work.go b/internal/hostbox/work.go
index 66ac634c2..ad9fa22c7 100644
--- a/internal/hostbox/work.go
+++ b/internal/hostbox/work.go
@@ -89,3 +89,16 @@ func Serving(c *hostclaim.Claim, selected Selection) error {
}
return nil
}
+
+// SelectAt captures an already chosen installation without relabelling it.
+func (s WorkSource) SelectAt(ctx context.Context, host, action string, captured hostguard.Captured) (hostwrite.Work, error) {
+ work, err := s.Select(ctx, host, action)
+ if err != nil {
+ return nil, err
+ }
+ selected := work.(selectedWork)
+ if selected.selection.Captured != captured {
+ return nil, ErrChanged
+ }
+ return work, nil
+}
diff --git a/internal/hostclaim/claim.go b/internal/hostclaim/claim.go
index 43ed0bb84..4773e9539 100644
--- a/internal/hostclaim/claim.go
+++ b/internal/hostclaim/claim.go
@@ -58,6 +58,7 @@ type Holder struct {
Session uint64 `json:"session,omitempty"`
WriterBox string `json:"writer_box,omitempty"`
Admissions []AdmissionRecord `json:"admissions,omitempty"`
+ WakeRetry *WakeRetry `json:"wake_retry,omitempty"`
}
type AdmissionRecord struct {
@@ -253,6 +254,13 @@ func (c *Claim) Validate() error {
return fmt.Errorf("%w: admission without region", ErrMalformed)
}
}
+ if err := h.WakeRetry.validate(); err != nil {
+ return err
+ }
+ if h.WakeRetry != nil && h.WakeRetry.State == "attempting" && h.WakeRetry.Attempt.Session != 0 &&
+ (h.WakeRetry.Attempt.Session != h.Session || h.WakeRetry.Attempt.Box != h.WriterBox) {
+ return fmt.Errorf("%w: wake retry attempt is not the live writer session", ErrMalformed)
+ }
}
if len(c.RetiredIncarnations) > retiredCap {
return fmt.Errorf("%w: retired_incarnations over cap", ErrMalformed)
diff --git a/internal/hostclaim/generation.go b/internal/hostclaim/generation.go
index 973e71dac..33d5cc4f2 100644
--- a/internal/hostclaim/generation.go
+++ b/internal/hostclaim/generation.go
@@ -40,6 +40,12 @@ func AdmitWriter(in Admission) func(*Claim) error {
return fmt.Errorf("%w: op kind %q cannot admit a writer", ErrWrongState, c.Op.Kind)
}
h := c.Holder
+ if c.Op.Kind == KindServing && h.WakeRetry != nil {
+ r := h.WakeRetry
+ if r.State != "attempting" || r.Attempt == nil || r.Attempt.Session != 0 || r.Attempt.Box != in.Box {
+ return fmt.Errorf("%w: serving writer admission has no matching retry reservation", ErrWrongState)
+ }
+ }
if h.Session != in.ObservedSession || h.Generation != in.ObservedGeneration {
return fmt.Errorf("%w: writer session or generation moved since it was observed", ErrWrongState)
}
diff --git a/internal/hostclaim/wakeretry.go b/internal/hostclaim/wakeretry.go
new file mode 100644
index 000000000..cd68cc933
--- /dev/null
+++ b/internal/hostclaim/wakeretry.go
@@ -0,0 +1,193 @@
+package hostclaim
+
+import (
+ "fmt"
+ "time"
+)
+
+const WakeRetryLimit = 5
+
+var wakeRetryDelays = [...]time.Duration{time.Second, 2 * time.Second, 4 * time.Second, 8 * time.Second, 16 * time.Second}
+
+type WakeRetry struct {
+ EpisodeID string `json:"episode_id"`
+ AttemptsReserved int `json:"attempts_reserved"`
+ State string `json:"state"`
+ NotBefore time.Time `json:"not_before,omitempty"`
+ Attempt *WakeRetryAttempt `json:"attempt,omitempty"`
+ ResetRequest string `json:"reset_request,omitempty"`
+}
+
+type WakeRetryAttempt struct {
+ ID string `json:"id"`
+ Box string `json:"box"`
+ Enrollment string `json:"enrollment"`
+ BootID string `json:"boot_id"`
+ ControllerID string `json:"controller_id"`
+ Session uint64 `json:"session,omitempty"`
+ InvocationID string `json:"invocation_id"`
+ LaunchInputs string `json:"launch_inputs"`
+}
+
+func (r *WakeRetry) validate() error {
+ if r == nil {
+ return nil
+ }
+ if r.EpisodeID == "" || r.AttemptsReserved < 0 || r.AttemptsReserved > WakeRetryLimit {
+ return fmt.Errorf("%w: wake retry episode", ErrMalformed)
+ }
+ switch r.State {
+ case "open":
+ if r.Attempt != nil || r.AttemptsReserved >= WakeRetryLimit {
+ return fmt.Errorf("%w: open wake retry", ErrMalformed)
+ }
+ case "attempting":
+ if r.AttemptsReserved == 0 || r.Attempt == nil || !r.NotBefore.IsZero() || !r.Attempt.valid() {
+ return fmt.Errorf("%w: attempting wake retry", ErrMalformed)
+ }
+ case "failed":
+ if r.AttemptsReserved != WakeRetryLimit || r.Attempt != nil || r.NotBefore.IsZero() {
+ return fmt.Errorf("%w: failed wake retry", ErrMalformed)
+ }
+ default:
+ return fmt.Errorf("%w: wake retry state %q", ErrMalformed, r.State)
+ }
+ return nil
+}
+
+func (a *WakeRetryAttempt) valid() bool {
+ return a.ID != "" && a.Box != "" && a.Enrollment != "" && a.BootID != "" &&
+ a.ControllerID != "" && a.InvocationID != "" && a.LaunchInputs != ""
+}
+
+type WakeRetryReservation struct {
+ HolderIncarnation string
+ EpisodeID string
+ Attempt WakeRetryAttempt
+ Now time.Time
+}
+
+// ReserveWakeRetryAttempt spends one serving retry after the caller has
+// established all read-only prerequisites. Session zero denotes the first
+// admission and must be bound with BindWakeRetrySession before app code runs.
+func ReserveWakeRetryAttempt(in WakeRetryReservation) func(*Claim) error {
+ return func(c *Claim) error {
+ if c.Holder == nil || c.Op.Kind != KindServing || c.Holder.Incarnation != in.HolderIncarnation ||
+ in.Now.IsZero() || !in.Attempt.valid() || (in.Attempt.Session != 0 && in.Attempt.Session != c.Holder.Session) {
+ return ErrWrongState
+ }
+ r := c.Holder.WakeRetry
+ if r == nil {
+ if in.EpisodeID == "" {
+ return ErrWrongState
+ }
+ r = &WakeRetry{EpisodeID: in.EpisodeID, State: "open"}
+ c.Holder.WakeRetry = r
+ } else if in.EpisodeID != "" && in.EpisodeID != r.EpisodeID {
+ return ErrWrongState
+ }
+ if r.State != "open" || r.AttemptsReserved >= WakeRetryLimit || in.Now.Before(r.NotBefore) {
+ return ErrWrongState
+ }
+ r.AttemptsReserved++
+ r.State, r.NotBefore = "attempting", time.Time{}
+ a := in.Attempt
+ r.Attempt = &a
+ return nil
+ }
+}
+
+// BindWakeRetrySession attaches the cold admission resulting from a first
+// reservation. It is required before that invocation may execute app code.
+func BindWakeRetrySession(incarnation, episode, attempt string, session uint64) func(*Claim) error {
+ return func(c *Claim) error {
+ if c.Holder == nil || c.Holder.Incarnation != incarnation || session == 0 || c.Holder.Session != session {
+ return ErrWrongState
+ }
+ r := c.Holder.WakeRetry
+ if r == nil || r.EpisodeID != episode || r.State != "attempting" || r.Attempt == nil ||
+ r.Attempt.ID != attempt || r.Attempt.Session != 0 {
+ return ErrWrongState
+ }
+ r.Attempt.Session = session
+ return nil
+ }
+}
+
+type WakeRetryResult struct {
+ HolderIncarnation string
+ EpisodeID string
+ Attempt WakeRetryAttempt
+ Now time.Time
+}
+
+func sameWakeAttempt(a *WakeRetryAttempt, b WakeRetryAttempt) bool { return a != nil && *a == b }
+
+// FailWakeRetryAttempt settles the exact reserved attempt. A crash or
+// ambiguous result uses this transition too; the reservation is never refunded.
+func FailWakeRetryAttempt(in WakeRetryResult) func(*Claim) error {
+ return func(c *Claim) error {
+ if c.Holder == nil || c.Holder.Incarnation != in.HolderIncarnation || in.Now.IsZero() {
+ return ErrWrongState
+ }
+ r := c.Holder.WakeRetry
+ if r == nil || r.EpisodeID != in.EpisodeID || r.State != "attempting" || !sameWakeAttempt(r.Attempt, in.Attempt) {
+ return ErrWrongState
+ }
+ r.Attempt = nil
+ r.NotBefore = in.Now.Add(wakeRetryDelays[r.AttemptsReserved-1])
+ if r.AttemptsReserved == WakeRetryLimit {
+ r.State = "failed"
+ } else {
+ r.State = "open"
+ }
+ return nil
+ }
+}
+
+type WakeRetryHealth struct {
+ WakeRetryResult
+ ReadinessPassed bool
+}
+
+// ClearWakeRetryHealthy accepts only a readiness result for the exact live
+// reservation. The caller is responsible for performing the backend checks.
+func ClearWakeRetryHealthy(in WakeRetryHealth) func(*Claim) error {
+ return func(c *Claim) error {
+ if !in.ReadinessPassed || c.Holder == nil || c.Holder.Incarnation != in.HolderIncarnation ||
+ in.Attempt.Session == 0 || c.Holder.Session != in.Attempt.Session {
+ return ErrWrongState
+ }
+ r := c.Holder.WakeRetry
+ if r == nil || r.EpisodeID != in.EpisodeID || r.State != "attempting" || !sameWakeAttempt(r.Attempt, in.Attempt) {
+ return ErrWrongState
+ }
+ c.Holder.WakeRetry = nil
+ return nil
+ }
+}
+
+type WakeRetryReset struct {
+ HolderIncarnation string
+ ObservedEpisode string
+ NewEpisodeID string
+ OperatorRequest string
+ AttemptSettled bool
+}
+
+// ResetWakeRetry is the operator-authorized primitive. Authorization and any
+// required fencing happen outside this package and are asserted by the input.
+func ResetWakeRetry(in WakeRetryReset) func(*Claim) error {
+ return func(c *Claim) error {
+ if c.Holder == nil || c.Holder.Incarnation != in.HolderIncarnation || in.ObservedEpisode == "" ||
+ in.NewEpisodeID == "" || in.NewEpisodeID == in.ObservedEpisode || in.OperatorRequest == "" {
+ return ErrWrongState
+ }
+ r := c.Holder.WakeRetry
+ if r == nil || r.EpisodeID != in.ObservedEpisode || (r.State == "attempting" && !in.AttemptSettled) {
+ return ErrWrongState
+ }
+ c.Holder.WakeRetry = &WakeRetry{EpisodeID: in.NewEpisodeID, State: "open", ResetRequest: in.OperatorRequest}
+ return nil
+ }
+}
diff --git a/internal/hostclaim/wakeretry_test.go b/internal/hostclaim/wakeretry_test.go
new file mode 100644
index 000000000..b7ec7e108
--- /dev/null
+++ b/internal/hostclaim/wakeretry_test.go
@@ -0,0 +1,157 @@
+package hostclaim
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func retryAttempt(id string, session uint64) WakeRetryAttempt {
+ return WakeRetryAttempt{ID: id, Box: "adi-knitting", Enrollment: "enroll-1", BootID: "boot-1",
+ ControllerID: "controller-1", Session: session, InvocationID: "invoke-" + id, LaunchInputs: "sha256-inputs-" + id}
+}
+
+func TestWakeRetryDurableBudgetAndBackoff(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ inc := o.Claim.Holder.Incarnation
+ now := time.Date(2026, 10, 6, 12, 0, 0, 0, time.UTC)
+
+ for n := 1; n <= WakeRetryLimit; n++ {
+ a := retryAttempt(string(rune('0'+n)), o.Claim.Holder.Session)
+ var err error
+ o, err = Update(ctx, s, o, ReserveWakeRetryAttempt(WakeRetryReservation{
+ HolderIncarnation: inc, EpisodeID: "episode-1", Attempt: a, Now: now,
+ }))
+ if err != nil {
+ t.Fatalf("reserve %d: %v", n, err)
+ }
+ if o.Claim.Holder.WakeRetry.AttemptsReserved != n || o.Claim.Holder.WakeRetry.State != "attempting" {
+ t.Fatalf("reserved %d = %+v", n, o.Claim.Holder.WakeRetry)
+ }
+ // A reread proves the budget is stored in the claim rather than process memory.
+ if o, err = Read(ctx, s, testHost); err != nil {
+ t.Fatal(err)
+ }
+ o, err = Update(ctx, s, o, FailWakeRetryAttempt(WakeRetryResult{
+ HolderIncarnation: inc, EpisodeID: "episode-1", Attempt: a, Now: now,
+ }))
+ if err != nil {
+ t.Fatalf("fail %d: %v", n, err)
+ }
+ wantState := "open"
+ if n == WakeRetryLimit {
+ wantState = "failed"
+ }
+ if r := o.Claim.Holder.WakeRetry; r.State != wantState || !r.NotBefore.Equal(now.Add(wakeRetryDelays[n-1])) {
+ t.Fatalf("settled %d = %+v", n, r)
+ }
+ if _, err := Update(ctx, s, o, ReserveWakeRetryAttempt(WakeRetryReservation{HolderIncarnation: inc, Attempt: retryAttempt("early", 0), Now: now})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("early reserve %d = %v", n, err)
+ }
+ now = now.Add(wakeRetryDelays[n-1])
+ }
+ if _, err := Update(ctx, s, o, ReserveWakeRetryAttempt(WakeRetryReservation{HolderIncarnation: inc, Attempt: retryAttempt("six", 0), Now: now})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("sixth reserve = %v", err)
+ }
+ if _, err := Update(ctx, s, o, AdmitWriter(Admission{Region: "ie", PredecessorRegion: "ie", Box: "adi-knitting", Mode: "cold", NewGeneration: "new-after-failure", PredecessorCertified: true})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("serving admission after exhausted budget = %v", err)
+ }
+}
+
+func TestWakeRetryColdAdmissionBindingAndExactHealth(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ inc := o.Claim.Holder.Incarnation
+ a := retryAttempt("first", 0)
+ now := time.Date(2026, 10, 6, 13, 0, 0, 0, time.UTC)
+ var err error
+ o, err = Update(ctx, s, o, ReserveWakeRetryAttempt(WakeRetryReservation{HolderIncarnation: inc, EpisodeID: "episode", Attempt: a, Now: now}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ o, err = Update(ctx, s, o, AdmitWriter(Admission{Region: "ie", PredecessorRegion: "ie", Box: "adi-knitting", Mode: "cold", NewGeneration: "generation-1", PredecessorCertified: true}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ session := o.Claim.Holder.Session
+ if o.Claim.Holder.WakeRetry.AttemptsReserved != 1 { // cold admission did not reset the episode
+ t.Fatalf("episode after admission = %+v", o.Claim.Holder.WakeRetry)
+ }
+ if _, err := Update(ctx, s, o, ClearWakeRetryHealthy(WakeRetryHealth{WakeRetryResult: WakeRetryResult{HolderIncarnation: inc, EpisodeID: "episode", Attempt: a}, ReadinessPassed: true})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("health before session binding = %v", err)
+ }
+ o, err = Update(ctx, s, o, BindWakeRetrySession(inc, "episode", "first", session))
+ if err != nil {
+ t.Fatal(err)
+ }
+ a.Session = session
+ wrong := a
+ wrong.InvocationID = "other-invocation"
+ if _, err := Update(ctx, s, o, ClearWakeRetryHealthy(WakeRetryHealth{WakeRetryResult: WakeRetryResult{HolderIncarnation: inc, EpisodeID: "episode", Attempt: wrong}, ReadinessPassed: true})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("wrong invocation health = %v", err)
+ }
+ if _, err := Update(ctx, s, o, ClearWakeRetryHealthy(WakeRetryHealth{WakeRetryResult: WakeRetryResult{HolderIncarnation: inc, EpisodeID: "episode", Attempt: a}})); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("failed readiness = %v", err)
+ }
+ o, err = Update(ctx, s, o, ClearWakeRetryHealthy(WakeRetryHealth{WakeRetryResult: WakeRetryResult{HolderIncarnation: inc, EpisodeID: "episode", Attempt: a}, ReadinessPassed: true}))
+ if err != nil || o.Claim.Holder.WakeRetry != nil {
+ t.Fatalf("healthy clear = %+v, %v", o.Claim.Holder.WakeRetry, err)
+ }
+}
+
+func TestWakeRetryOperatorResetRequiresSettledAttempt(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ inc := o.Claim.Holder.Incarnation
+ a := retryAttempt("active", o.Claim.Holder.Session)
+ var err error
+ o, err = Update(ctx, s, o, ReserveWakeRetryAttempt(WakeRetryReservation{HolderIncarnation: inc, EpisodeID: "old", Attempt: a, Now: time.Now()}))
+ if err != nil {
+ t.Fatal(err)
+ }
+ reset := WakeRetryReset{HolderIncarnation: inc, ObservedEpisode: "old", NewEpisodeID: "new", OperatorRequest: "request-42"}
+ if _, err := Update(ctx, s, o, ResetWakeRetry(reset)); !errors.Is(err, ErrWrongState) {
+ t.Fatalf("unsettled reset = %v", err)
+ }
+ reset.AttemptSettled = true
+ o, err = Update(ctx, s, o, ResetWakeRetry(reset))
+ if err != nil {
+ t.Fatal(err)
+ }
+ if got := o.Claim.Holder.WakeRetry; got.EpisodeID != "new" || got.State != "open" || got.AttemptsReserved != 0 || got.ResetRequest != "request-42" {
+ t.Fatalf("reset episode = %+v", got)
+ }
+}
+
+func TestWakeRetryValidationAndWireShape(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ o := servingClaim(t, ctx, s)
+ bad := []WakeRetry{
+ {EpisodeID: "e", State: "open", AttemptsReserved: 5},
+ {EpisodeID: "e", State: "attempting", AttemptsReserved: 1},
+ {EpisodeID: "e", State: "failed", AttemptsReserved: 4, NotBefore: time.Now()},
+ {EpisodeID: "e", State: "unknown"},
+ }
+ for i := range bad {
+ if _, err := Update(ctx, s, o, func(c *Claim) error { r := bad[i]; c.Holder.WakeRetry = &r; return nil }); !errors.Is(err, ErrMalformed) {
+ t.Errorf("bad %d = %v", i, err)
+ }
+ }
+ raw, err := json.Marshal(o.Claim.Holder)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if bytes.Contains(raw, []byte(`"wake_retry"`)) {
+ t.Fatalf("legacy holder gained wake_retry: %s", raw)
+ }
+}