amadan

carlos / platform Public

hostops: record, transition and recovery primitives (claim Plan 5 Task 2)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 3f1db4ea1a578e6e24dbcf4eb7c3236fb4537d94 parent 56ecb9e
5 files changed, +899 −0
  • internal/hostops/operations.go +267 −0
  • internal/hostops/purge.go +196 −0
  • internal/hostops/record.go +173 −0
  • internal/hostops/recovery.go +216 −0
  • internal/hostops/remote.go +47 −0
diff --git a/internal/hostops/operations.go b/internal/hostops/operations.go
new file mode 100644
index 000000000..4409ef9b5
--- /dev/null
+++ b/internal/hostops/operations.go
@@ -0,0 +1,267 @@
+package hostops
+
+import (
+ "context"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// These callbacks are authority providers, not optional best-effort hints.
+// Production construction waits for the decisions named in this plan.
+type Authority interface {
+ Create(context.Context, hostclaim.Observed, Record, string) error
+ Executor(context.Context, hostclaim.Observed, Record) (string, error)
+ StartDelete(context.Context, hostclaim.Observed, Record) error
+ FirstPhase(context.Context, hostclaim.Observed) error
+ Inventory(context.Context, hostclaim.Observed) ([]string, []string, error)
+ Adoption(context.Context, Record) (region, box string, otherRegions, otherBoxes []string, err error)
+ PurgeRetirement(context.Context, hostclaim.Observed) error
+}
+
+type Controller struct {
+ Store release.Store
+ Authority Authority
+ Now func() time.Time
+ NewID func() string
+}
+
+func (c Controller) ready() error {
+ if c.Store == nil || c.Authority == nil || c.Now == nil || c.NewID == nil {
+ return ErrContract
+ }
+ return nil
+}
+
+// Attempt consumes the claim observation taken BEFORE app registration.
+// Create's caller owns registration renewal/removal. A failed CAS is never retried here.
+func (c Controller) Attempt(ctx context.Context, before hostclaim.Observed, r Record, id, by string) (hostclaim.Observed, error) {
+ if err := c.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ if before.Host != r.Holder.Host || r.Holder.CreatedAt.IsZero() || id == "" {
+ return hostclaim.Observed{}, ErrIdentity
+ }
+ // Last pre-CAS read of registration, open admission and purge reservation.
+ if err := c.Authority.Create(ctx, before, r, id); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ obs, err := hostclaim.Update(ctx, c.Store, before, hostclaim.BeginCreate(r.Holder, r.Body, id, by, c.Now()))
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ // No raw record effect precedes the durable claim intent.
+ if _, err = c.Store.Put(ctx, r.Holder.Key, r.Body, ""); err != nil {
+ if !errors.Is(err, release.ErrCASConflict) {
+ return obs, err
+ }
+ existing, rerr := ReadRecord(ctx, c.Store, r.Holder.Key)
+ if rerr != nil {
+ return obs, rerr
+ }
+ if !matches(existing, obs.Claim.Holder) {
+ removed, rerr := RemoveOrphan(ctx, c.Store, existing)
+ if rerr != nil {
+ return obs, rerr
+ }
+ if !removed {
+ return obs, ErrPending
+ }
+ if _, err = c.Store.Put(ctx, r.Holder.Key, r.Body, ""); err != nil {
+ return obs, err
+ }
+ }
+ }
+ return c.Complete(ctx, obs.Host, id)
+}
+
+func (c Controller) Complete(ctx context.Context, host, id string) (hostclaim.Observed, error) {
+ if err := c.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, host)
+ if err != nil {
+ return obs, err
+ }
+ if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindCreating || obs.Claim.Op.ID != id {
+ return obs, ErrPending
+ }
+ r, err := ReadRecord(ctx, c.Store, obs.Claim.Holder.Key)
+ if err != nil {
+ return obs, err
+ }
+ if !matches(r, obs.Claim.Holder) {
+ return obs, ErrIdentity
+ }
+ box, err := c.Authority.Executor(ctx, obs, r)
+ if err != nil {
+ return obs, err
+ }
+ return hostclaim.Update(ctx, c.Store, obs, hostclaim.CompleteCreate(id, box))
+}
+
+func (c Controller) RecoverCreate(ctx context.Context, host string) error {
+ if err := c.ready(); err != nil {
+ return err
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent || obs.Claim.Op.Kind != hostclaim.KindCreating {
+ return nil
+ }
+ if obs.Claim.Op.StartedAt.IsZero() || !c.Now().After(obs.Claim.Op.StartedAt.Add(5*time.Minute)) {
+ return ErrPending
+ }
+ r, err := ReadRecord(ctx, c.Store, obs.Claim.Holder.Key)
+ if errors.Is(err, os.ErrNotExist) {
+ _, err = hostclaim.Update(ctx, c.Store, obs, hostclaim.CancelCreate(obs.Claim.Op.ID, c.Now()))
+ return err
+ }
+ if err != nil {
+ return err
+ }
+ if !matches(r, obs.Claim.Holder) {
+ return ErrPending
+ }
+ _, err = c.Complete(ctx, host, obs.Claim.Op.ID)
+ return err
+}
+
+// Adopt is the shared IO wrapper around the one hostclaim.Adopt transition.
+func (c Controller) Adopt(ctx context.Context, r Record) (hostclaim.Observed, error) {
+ if err := c.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, r.Holder.Host)
+ if err != nil {
+ return obs, err
+ }
+ if !obs.Absent {
+ if !matches(r, obs.Claim.Holder) {
+ return obs, ErrIdentity
+ }
+ return obs, nil
+ }
+ region, box, regions, boxes, err := c.Authority.Adoption(ctx, r)
+ if err != nil {
+ return obs, err
+ }
+ // Re-read the exact record after discovery; never adopt a stale listing body.
+ fresh, err := ReadRecord(ctx, c.Store, r.Holder.Key)
+ if err != nil {
+ return obs, err
+ }
+ if fresh.ETag != r.ETag {
+ return obs, ErrPending
+ }
+ n := fresh.Holder
+ n.Region = region
+ return hostclaim.Update(ctx, c.Store, obs, hostclaim.Adopt(n, box, regions, boxes, c.NewID(), c.NewID))
+}
+
+// BeginRetire only starts the claim operation. Its successful return does not
+// mean the record is gone or the host is free; the caller must run/retry Retire.
+func (c Controller) BeginRetire(ctx context.Context, key, by, expectedInc string) (hostclaim.Observed, error) {
+ if err := c.ready(); err != nil {
+ return hostclaim.Observed{}, err
+ }
+ r, err := ReadRecord(ctx, c.Store, key)
+ if err != nil {
+ return hostclaim.Observed{}, err
+ }
+ inc, ierr := r.Incarnation()
+ if ierr != nil {
+ return hostclaim.Observed{}, ierr
+ }
+ if expectedInc == "" || inc != expectedInc {
+ return hostclaim.Observed{}, ErrIdentity
+ }
+ obs, err := c.Adopt(ctx, r)
+ if err != nil {
+ return obs, err
+ }
+ if err := c.Authority.StartDelete(ctx, obs, r); err != nil {
+ return obs, err
+ }
+ regions, boxes, err := c.Authority.Inventory(ctx, obs)
+ if err != nil {
+ return obs, err
+ }
+ return hostclaim.Update(ctx, c.Store, obs, hostclaim.BeginDelete(c.NewID(), by, r.ETag, c.Now(), regions, boxes))
+}
+
+func (c Controller) Retire(ctx context.Context, host, id string) error {
+ if err := c.ready(); err != nil {
+ return err
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent || obs.Claim.Op.ID != id {
+ return ErrPending
+ }
+ op := obs.Claim.Op
+ if op.Kind != hostclaim.KindDeleting && op.Kind != hostclaim.KindPurging {
+ return ErrPending
+ }
+ if obs.Claim.Holder == nil || op.Intent["key"] != obs.Claim.Holder.Key || op.Intent["etag"] == "" {
+ return ErrIdentity
+ }
+ if err := c.Authority.FirstPhase(ctx, obs); err != nil {
+ return err
+ }
+ if op.Kind == hostclaim.KindPurging {
+ if err := c.Authority.PurgeRetirement(ctx, obs); err != nil {
+ return err
+ }
+ }
+ regions, boxes, err := c.Authority.Inventory(ctx, obs)
+ if err != nil {
+ return err
+ }
+ // Validate record identity before deleting even the stored version.
+ r, rerr := ReadRecord(ctx, c.Store, op.Intent["key"])
+ if rerr != nil && !errors.Is(rerr, os.ErrNotExist) {
+ return rerr
+ }
+ if rerr == nil {
+ if r.ETag != op.Intent["etag"] || !matches(r, obs.Claim.Holder) {
+ return ErrPending
+ }
+ if err := release.DeleteIfMatch(ctx, c.Store, r.Holder.Key, r.ETag); err != nil {
+ return err
+ }
+ }
+ // A success/412 alone never certifies absence. Re-read before parking.
+ _, _, err = c.Store.Get(ctx, op.Intent["key"])
+ if !errors.Is(err, os.ErrNotExist) {
+ if err != nil {
+ return err
+ }
+ return ErrPending
+ }
+ // The old observation binds the entire sequence to this operation.
+ _, err = hostclaim.Update(ctx, c.Store, obs, hostclaim.ToParking(id, regions, boxes, c.NewID))
+ return err
+}
+
+func (c Controller) Free(ctx context.Context, host, id string) error {
+ if err := c.ready(); err != nil {
+ return err
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent {
+ return ErrPending
+ }
+ _, err = hostclaim.Update(ctx, c.Store, obs, hostclaim.FreeAfterPark(id))
+ return err
+}
diff --git a/internal/hostops/purge.go b/internal/hostops/purge.go
new file mode 100644
index 000000000..ba2ae03d5
--- /dev/null
+++ b/internal/hostops/purge.go
@@ -0,0 +1,196 @@
+package hostops
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "os"
+ "time"
+
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+type Purge struct {
+ ID string `json:"id"`
+ Phase string `json:"phase"`
+}
+
+var phases = []string{"reserved", "adopted", "claimed", "torn-down", "retired", "done"}
+
+func phaseIndex(s string) int {
+ for i, p := range phases {
+ if s == p {
+ return i
+ }
+ }
+ return -1
+}
+func PurgeKey(acct, app string) (string, error) {
+ if !controlname.ValidAccount(acct) || !controlname.ValidApp(app) {
+ return "", ErrIdentity
+ }
+ return "apps/" + acct + "/" + app + "/purge.json", nil
+}
+func ReadPurge(ctx context.Context, s release.Store, acct, app string) (Purge, string, error) {
+ key, err := PurgeKey(acct, app)
+ if err != nil {
+ return Purge{}, "", err
+ }
+ body, etag, err := s.Get(ctx, key)
+ if err != nil {
+ return Purge{}, "", err
+ }
+ var p Purge
+ if err := decode(body, &p); err != nil {
+ return Purge{}, "", err
+ }
+ if p.ID == "" || phaseIndex(p.Phase) < 0 || etag == "" {
+ return Purge{}, "", ErrIdentity
+ }
+ return p, etag, nil
+}
+func ReservePurge(ctx context.Context, s release.Store, acct, app, id string) (Purge, string, error) {
+ key, err := PurgeKey(acct, app)
+ if err != nil || id == "" {
+ return Purge{}, "", ErrIdentity
+ }
+ p := Purge{ID: id, Phase: "reserved"}
+ body, err := json.Marshal(p)
+ if err != nil {
+ return Purge{}, "", err
+ }
+ etag, err := s.Put(ctx, key, body, "")
+ if errors.Is(err, release.ErrCASConflict) {
+ existing, tag, rerr := ReadPurge(ctx, s, acct, app)
+ if rerr != nil {
+ return Purge{}, "", rerr
+ }
+ if existing.ID != id {
+ return existing, tag, ErrPending
+ }
+ return existing, tag, nil
+ }
+ return p, etag, err
+}
+
+// AdvancePurge preserves additional journal fields owned by the final durable
+// host-set schema. Preconditions for the next phase belong to the coordinator.
+func AdvancePurge(ctx context.Context, s release.Store, acct, app, id, from, to string) error {
+ if phaseIndex(from) < 0 || phaseIndex(to) != phaseIndex(from)+1 {
+ return ErrIdentity
+ }
+ key, err := PurgeKey(acct, app)
+ if err != nil {
+ return err
+ }
+ body, etag, err := s.Get(ctx, key)
+ if err != nil {
+ return err
+ }
+ var p Purge
+ if err := decode(body, &p); err != nil {
+ return err
+ }
+ if p.ID != id || p.Phase != from || etag == "" {
+ return ErrPending
+ }
+ var fields map[string]json.RawMessage
+ if err := decode(body, &fields); err != nil {
+ return err
+ }
+ fields["phase"], err = json.Marshal(to)
+ if err != nil {
+ return err
+ }
+ body, err = json.Marshal(fields)
+ if err != nil {
+ return err
+ }
+ _, err = s.Put(ctx, key, body, etag)
+ return err
+}
+
+// Only the completed coordinator may call this, after all fence-release writes.
+func RemovePurge(ctx context.Context, s release.Store, acct, app, id string) error {
+ p, etag, err := ReadPurge(ctx, s, acct, app)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
+ return err
+ }
+ if p.ID != id || p.Phase != "done" {
+ return ErrPending
+ }
+ key, err := PurgeKey(acct, app)
+ if err != nil {
+ return err
+ }
+ return release.DeleteIfMatch(ctx, s, key, etag)
+}
+
+func FencePurge(id string, now time.Time) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if id == "" || now.IsZero() {
+ return ErrIdentity
+ }
+ if !c.Free() || len(c.Obligations) != 0 {
+ return ErrPending
+ }
+ c.Op = hostclaim.Op{Kind: hostclaim.KindPurging, ID: id, StartedAt: now.UTC()}
+ return nil
+ }
+}
+func ReleasePurgeFence(id string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if id == "" || c.Holder != nil || c.Op.Kind != hostclaim.KindPurging || c.Op.ID != id || len(c.Obligations) != 0 {
+ return ErrPending
+ }
+ c.Op = hostclaim.Op{}
+ return nil
+ }
+}
+func BeginPurge(id, by string, now time.Time, regions, boxes []string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if id == "" || now.IsZero() {
+ return ErrIdentity
+ }
+ if err := hostclaim.StartGuard(c, regions, boxes); err != nil {
+ return err
+ }
+ c.Op = hostclaim.Op{Kind: hostclaim.KindPurging, ID: id, By: by, StartedAt: now.UTC(), Exec: c.Op.Exec, Participants: c.Op.Participants}
+ return nil
+ }
+}
+func PurgeRecordIntent(id, key, etag string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if id == "" || etag == "" || c.Op.Kind != hostclaim.KindPurging || c.Op.ID != id || c.Holder == nil || c.Holder.Key != key {
+ return ErrIdentity
+ }
+ if c.Op.Intent == nil {
+ c.Op.Intent = map[string]string{}
+ }
+ if old := c.Op.Intent["etag"]; old != "" && old != etag {
+ return ErrPending
+ }
+ c.Op.Intent["key"], c.Op.Intent["etag"] = key, etag
+ return nil
+ }
+}
+
+// No completion transition is supplied until placement's executor/evidence
+// contract is resolved. This start transition alone grants no new execution.
+func BeginPlacement(id, by, target string, now time.Time, regions, boxes []string) func(*hostclaim.Claim) error {
+ return func(c *hostclaim.Claim) error {
+ if id == "" || now.IsZero() {
+ return ErrIdentity
+ }
+ if err := hostclaim.StartGuard(c, regions, boxes); err != nil {
+ return err
+ }
+ c.Op = hostclaim.Op{Kind: hostclaim.KindPlacing, ID: id, By: by, StartedAt: now.UTC(), Exec: c.Op.Exec, Participants: c.Op.Participants, Intent: map[string]string{"target": target}}
+ return nil
+ }
+}
diff --git a/internal/hostops/record.go b/internal/hostops/record.go
new file mode 100644
index 000000000..7616db1b0
--- /dev/null
+++ b/internal/hostops/record.go
@@ -0,0 +1,173 @@
+package hostops
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "io"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/controlname"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+var (
+ ErrPending = errors.New("hostops: pending")
+ ErrContract = errors.New("hostops: required contract unavailable")
+ ErrIdentity = errors.New("hostops: identity mismatch")
+)
+
+type Record struct {
+ Holder hostclaim.NewHolder
+ Body []byte
+ ETag string
+}
+
+func decode(body []byte, out any) error {
+ d := json.NewDecoder(bytes.NewReader(body))
+ if err := d.Decode(out); err != nil {
+ return err
+ }
+ if _, err := d.Token(); err != io.EOF {
+ return ErrIdentity
+ }
+ return nil
+}
+
+func ReadRecord(ctx context.Context, s release.Store, key string) (Record, error) {
+ parts := strings.Split(key, "/")
+ if len(parts) != 5 || parts[0] != "apps" || !controlname.ValidAccount(parts[1]) || !controlname.ValidApp(parts[2]) || !strings.HasSuffix(parts[4], ".json") {
+ return Record{}, ErrIdentity
+ }
+ body, etag, err := s.Get(ctx, key)
+ if err != nil {
+ return Record{}, err
+ }
+ if etag == "" {
+ return Record{}, ErrIdentity
+ }
+ n := hostclaim.NewHolder{Key: key, Account: parts[1], App: parts[2]}
+ switch parts[3] {
+ case "instances":
+ var r provision.Record
+ if err := decode(body, &r); err != nil {
+ return Record{}, err
+ }
+ if r.Account != n.Account || r.App != n.App || key != provision.RecordKey(r.Account, r.App, r.Host) || !controlname.ValidHostShape(r.Host) {
+ return Record{}, ErrIdentity
+ }
+ // DecodeRecord is also the ordinary instance reader's shape validator.
+ if _, err := provision.DecodeRecord(body); err != nil {
+ return Record{}, err
+ }
+ n.Family, n.Host, n.RecordRegion, n.CreatedAt = "instance", r.Host, r.Region, r.CreatedAt
+ case "sidecars":
+ var r provision.SidecarRecord
+ if err := decode(body, &r); err != nil {
+ return Record{}, err
+ }
+ if r.Account != n.Account || r.App != n.App || !controlname.ValidLabel(r.Name) || key != provision.SidecarRecordKey(r.Account, r.App, r.Name) || r.Host != provision.SidecarHost(r.Name, r.App, r.Account) {
+ return Record{}, ErrIdentity
+ }
+ n.Family, n.Host, n.Name, n.RecordRegion, n.CreatedAt = "sidecar", r.Host, r.Name, r.Region, r.CreatedAt
+ default:
+ return Record{}, ErrIdentity
+ }
+ return Record{Holder: n, Body: body, ETag: etag}, nil
+}
+
+func (r Record) Incarnation() (string, error) {
+ n := r.Holder
+ return hostclaim.Incarnation(n.Account, n.App, n.Host, n.CreatedAt)
+}
+
+func matches(r Record, h *hostclaim.Holder) bool {
+ if h == nil {
+ return false
+ }
+ n := r.Holder
+ inc, err := r.Incarnation()
+ return err == nil && n.Key == h.Key && n.Family == h.Family && n.Account == h.Account && n.App == h.App && n.Name == h.Name && n.CreatedAt.Equal(h.CreatedAt) && inc == h.Incarnation
+}
+
+// The caller reads the record first. This function performs the later claim read.
+func RemoveOrphan(ctx context.Context, s release.Store, r Record) (bool, error) {
+ inc, err := r.Incarnation()
+ if err != nil {
+ return false, err
+ }
+ obs, err := hostclaim.Read(ctx, s, r.Holder.Host)
+ if err != nil {
+ return false, err
+ }
+ if !hostclaim.IsOrphan(inc, obs) {
+ return false, nil
+ }
+ if err := release.DeleteIfMatch(ctx, s, r.Holder.Key, r.ETag); err != nil {
+ return false, err
+ }
+ return true, nil
+}
+
+func Records(ctx context.Context, s release.Store, acct, app string) ([]Record, error) {
+ if !controlname.ValidAccount(acct) || !controlname.ValidApp(app) {
+ return nil, ErrIdentity
+ }
+ var out []Record
+ for _, prefix := range []string{provision.RecordsPrefix(acct, app), provision.SidecarsPrefix(acct, app)} {
+ keys, err := s.List(ctx, prefix)
+ if err != nil {
+ return nil, err
+ }
+ for _, key := range keys {
+ tail, ok := strings.CutPrefix(key, prefix)
+ if !ok {
+ return nil, ErrIdentity
+ }
+ if strings.Contains(tail, "/") || !strings.HasSuffix(tail, ".json") {
+ continue
+ }
+ r, err := ReadRecord(ctx, s, key)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ return nil, err
+ }
+ out = append(out, r)
+ }
+ }
+ return out, nil
+}
+
+// PrepareInstance accepts only an unstamped internal request, never a caller time.
+func PrepareInstance(r provision.Record, now time.Time, region string) (Record, error) {
+ if !r.CreatedAt.IsZero() || now.IsZero() || region == "" || len(r.Maintenance) != 0 {
+ return Record{}, ErrIdentity
+ }
+ r.CreatedAt = now.UTC()
+ body, err := json.Marshal(r)
+ if err != nil {
+ return Record{}, err
+ }
+ n := hostclaim.NewHolder{Family: "instance", Key: provision.RecordKey(r.Account, r.App, r.Host), Account: r.Account, App: r.App, Host: r.Host, RecordRegion: r.Region, Region: region, CreatedAt: r.CreatedAt}
+ return Record{Holder: n, Body: body}, nil
+}
+
+func PrepareSidecar(r provision.SidecarRecord, now time.Time, region string) (Record, error) {
+ if !r.CreatedAt.IsZero() || now.IsZero() || region == "" {
+ return Record{}, ErrIdentity
+ }
+ r.CreatedAt = now.UTC()
+ body, err := json.Marshal(r)
+ if err != nil {
+ return Record{}, err
+ }
+ n := hostclaim.NewHolder{Family: "sidecar", Key: provision.SidecarRecordKey(r.Account, r.App, r.Name), Account: r.Account, App: r.App, Name: r.Name, Host: r.Host, RecordRegion: r.Region, Region: region, CreatedAt: r.CreatedAt}
+ return Record{Holder: n, Body: body}, nil
+}
diff --git a/internal/hostops/recovery.go b/internal/hostops/recovery.go
new file mode 100644
index 000000000..c134a69b1
--- /dev/null
+++ b/internal/hostops/recovery.go
@@ -0,0 +1,216 @@
+package hostops
+
+import (
+ "context"
+ "errors"
+ "os"
+ "strings"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// ListClaims is discovery only. Every action subsequently reads its own key.
+func ListClaims(ctx context.Context, s release.Store) ([]string, error) {
+ keys, err := s.List(ctx, "control/hosts/")
+ if err != nil {
+ return nil, err
+ }
+ var hosts []string
+ for _, key := range keys {
+ tail, ok := strings.CutPrefix(key, "control/hosts/")
+ if !ok {
+ return nil, ErrIdentity
+ }
+ if strings.Contains(tail, "/") || !strings.HasSuffix(tail, ".json") {
+ continue
+ }
+ host := strings.TrimSuffix(tail, ".json")
+ if _, err := hostclaim.Key(host); err != nil {
+ return nil, err
+ }
+ hosts = append(hosts, host)
+ }
+ return hosts, nil
+}
+
+type Recovery struct {
+ SweepAdmissions func(context.Context) error
+ Controller Controller
+ // These are required integrations, not no-op defaults.
+ Purge func(context.Context, hostclaim.Observed) error
+ Placement func(context.Context, hostclaim.Observed) error
+ SweepRecords func(context.Context) error
+ SweepPurges func(context.Context) error
+}
+
+func (r Recovery) Pass(ctx context.Context) error {
+ c := r.Controller
+ if err := c.ready(); err != nil {
+ return err
+ }
+ if r.Purge == nil || r.Placement == nil || r.SweepRecords == nil || r.SweepPurges == nil || r.SweepAdmissions == nil {
+ return ErrContract
+ }
+ hosts, err := ListClaims(ctx, c.Store)
+ if err != nil {
+ return err
+ }
+ var all []error
+ for _, host := range hosts {
+ if err := ctx.Err(); err != nil {
+ return errors.Join(append(all, err)...)
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, host)
+ if err != nil {
+ all = append(all, err)
+ continue
+ }
+ if obs.Absent {
+ continue
+ }
+ op := obs.Claim.Op
+ // Zero time is unknown age, never old enough. Freeing empty parking is safe
+ // independently of age; this includes Plan 4's holderless sweep retries.
+ if op.Kind == hostclaim.KindParking && len(obs.Claim.Obligations) == 0 {
+ err = c.Free(ctx, host, op.ID)
+ } else if op.Kind != "" && op.Kind != hostclaim.KindServing && !op.StartedAt.IsZero() && c.Now().After(op.StartedAt.Add(5*time.Minute)) {
+ switch op.Kind {
+ case hostclaim.KindCreating:
+ err = c.RecoverCreate(ctx, host)
+ case hostclaim.KindDeleting:
+ err = c.Retire(ctx, host, op.ID)
+ case hostclaim.KindPurging:
+ err = r.Purge(ctx, obs)
+ case hostclaim.KindPlacing:
+ err = r.Placement(ctx, obs)
+ }
+ }
+ if err != nil && !errors.Is(err, ErrPending) {
+ all = append(all, err)
+ }
+ }
+ // Records have their own uncapped scan: late orphans outlive the one-hour
+ // cancellation hint; records without claims need the shared adoption path.
+ if err := r.SweepRecords(ctx); err != nil {
+ all = append(all, err)
+ }
+ // Reservations can exist without any claim (crash just after reserve).
+ if err := r.SweepPurges(ctx); err != nil {
+ all = append(all, err)
+ }
+ if err := r.SweepAdmissions(ctx); err != nil {
+ all = append(all, err)
+ }
+ return errors.Join(all...)
+}
+
+type App struct{ Account, Name string }
+
+func Apps(ctx context.Context, s release.Store) ([]App, error) {
+ accounts, err := s.ListDirs(ctx, "apps/")
+ if err != nil {
+ return nil, err
+ }
+ var out []App
+ for _, prefix := range accounts {
+ account := strings.TrimSuffix(strings.TrimPrefix(prefix, "apps/"), "/")
+ if prefix != "apps/"+account+"/" {
+ return nil, ErrIdentity
+ }
+ // The retained legacy layout is apps/<app>/..., so its first segment is
+ // an app name, not necessarily an account. Skip names that are valid apps
+ // but invalid accounts before treating this as a qualified account root.
+ // A letters-only legacy name can pass both validators, so preserve its
+ // children as candidates and let the exact record and purge parsers
+ // validate them. The path alone cannot tell that case from a real account
+ // containing apps with those legal names; never infer deletion authority
+ // over retained legacy content from this prefix shape.
+ if !release.ValidAccount(account) {
+ if release.ValidApp(account) && prefix == release.LegacyAppPrefix(account) {
+ continue
+ }
+ return nil, ErrIdentity
+ }
+ apps, err := s.ListDirs(ctx, prefix)
+ if err != nil {
+ return nil, err
+ }
+ for _, ap := range apps {
+ app := strings.TrimSuffix(strings.TrimPrefix(ap, prefix), "/")
+ if _, err := PurgeKey(account, app); err != nil || ap != prefix+app+"/" {
+ return nil, ErrIdentity
+ }
+ out = append(out, App{Account: account, Name: app})
+ }
+ }
+ return out, nil
+}
+
+func (c Controller) SweepAllRecords(ctx context.Context) error {
+ if err := c.ready(); err != nil {
+ return err
+ }
+ apps, err := Apps(ctx, c.Store)
+ if err != nil {
+ return err
+ }
+ var all []error
+ for _, app := range apps {
+ records, err := Records(ctx, c.Store, app.Account, app.Name)
+ if err != nil {
+ all = append(all, err)
+ continue
+ }
+ for _, listed := range records {
+ record, err := ReadRecord(ctx, c.Store, listed.Holder.Key)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ all = append(all, err)
+ continue
+ }
+ obs, err := hostclaim.Read(ctx, c.Store, record.Holder.Host)
+ if err != nil {
+ all = append(all, err)
+ continue
+ }
+ if obs.Absent {
+ _, err = c.Adopt(ctx, record)
+ } else {
+ _, err = RemoveOrphan(ctx, c.Store, record)
+ }
+ if err != nil {
+ all = append(all, err)
+ }
+ }
+ }
+ return errors.Join(all...)
+}
+
+func SweepAllPurges(ctx context.Context, s release.Store, replay func(context.Context, App, Purge) error) error {
+ if replay == nil {
+ return ErrContract
+ }
+ apps, err := Apps(ctx, s)
+ if err != nil {
+ return err
+ }
+ var all []error
+ for _, app := range apps {
+ p, _, err := ReadPurge(ctx, s, app.Account, app.Name)
+ if errors.Is(err, os.ErrNotExist) {
+ continue
+ }
+ if err != nil {
+ all = append(all, err)
+ continue
+ }
+ if err := replay(ctx, app, p); err != nil {
+ all = append(all, err)
+ }
+ }
+ return errors.Join(all...)
+}
diff --git a/internal/hostops/remote.go b/internal/hostops/remote.go
new file mode 100644
index 000000000..e1c44b737
--- /dev/null
+++ b/internal/hostops/remote.go
@@ -0,0 +1,47 @@
+package hostops
+
+import (
+ "context"
+
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/park"
+)
+
+type ParkScopeProvider func(context.Context, hostclaim.Observed, hostclaim.Obligation) (hostclaim.ParkScope, error)
+
+// RunOwnedRemote is called only by the appointed regional worker. It never
+// takes another region's chore or acknowledges local work on another box.
+func RunOwnedRemote(ctx context.Context, e *park.Engine, w park.Work, scope ParkScopeProvider) error {
+ if e == nil || e.Claims == nil || e.Region == "" || scope == nil || e.Authorize == nil || e.View == nil || e.Leases == nil {
+ return ErrContract
+ }
+ obs, err := hostclaim.Read(ctx, e.Claims, w.Host)
+ if err != nil {
+ return err
+ }
+ if obs.Absent {
+ return ErrPending
+ }
+ for _, ob := range obs.Claim.Obligations {
+ if ob.ID != w.ID || ob.Op != w.Op {
+ continue
+ }
+ if ob.Region != e.Region || ob.Owner != "region:"+e.Region || (ob.Kind != "park-replica" && ob.Kind != "park-generation") {
+ return ErrIdentity
+ }
+ if err := e.Authorize(ctx, obs, ob); err != nil {
+ return err
+ }
+ if ob.Park == nil {
+ p, err := scope(ctx, obs, ob)
+ if err != nil {
+ return err
+ }
+ if err := e.Bind(ctx, w, p); err != nil {
+ return err
+ }
+ }
+ return e.Run(ctx, w)
+ }
+ return ErrPending
+}