amadan

carlos / platform Public

hostops, host, console: legacy repair only for the census owner, backfill off the promote path, quiet when no plan is adopting (Opus review I1, M1-M4, M6)

Co-Authored-By: Codex <noreply@openai.com>
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
codex pushed by claude 9fac55eabc34e453aa5c99940c1eab087cdb0045 parent 99b7742
11 files changed, +532 −53
  • cmd/console/main.go +1 −1
  • internal/console/hostclaimbackfill.go +4 −21
  • internal/console/hostclaimbackfill_test.go +84 −0
  • internal/host/agent.go +3 −6
  • internal/host/claimbackfill.go +4 −4
  • internal/host/claimbackfill_test.go +230 −0
  • internal/hostops/backfill.go +11 −11
  • internal/hostops/backfill_controller.go +8 −4
  • internal/hostops/backfill_test.go +78 −5
  • internal/hostops/operations.go +13 −1
  • internal/hostops/operations_test.go +96 −0

CI passed — run details

diff --git a/cmd/console/main.go b/cmd/console/main.go
index dc28c9523..fad2dd10f 100644
--- a/cmd/console/main.go
+++ b/cmd/console/main.go
@@ -393,7 +393,7 @@ func serve() {
// Same placement rule as the alert watcher: started here, never in
// NewServer, so constructing a Server stays goroutine-free.
if err := srv.ConfigureClaimBackfill(); err != nil {
- log.Fatalf("console: claim recovery: %v", err)
+ log.Printf("console: claim recovery: %v", err)
}
go srv.StartClaimRecoveryPass(context.Background())
go srv.StartEconomicsPass(context.Background())
diff --git a/internal/console/hostclaimbackfill.go b/internal/console/hostclaimbackfill.go
index f4936d0c4..92fc5881b 100644
--- a/internal/console/hostclaimbackfill.go
+++ b/internal/console/hostclaimbackfill.go
@@ -5,10 +5,10 @@ import (
"crypto/rand"
"encoding/hex"
"errors"
+ "log"
"time"
"github.com/carlosframework/platform/internal/hostactors"
- "github.com/carlosframework/platform/internal/hostclaim"
"github.com/carlosframework/platform/internal/hostops"
)
@@ -26,7 +26,9 @@ func (s *Server) ConfigureClaimBackfill() error {
return nil
}
if s.cfg.Releases == nil || s.cfg.Bucket == "" || len(s.cfg.ReleasePub) == 0 {
- return hostops.ErrContract
+ s.cfg.HostClaimOps = nil
+ log.Printf("console: claim recovery: %v", hostops.ErrContract)
+ return nil
}
if s.cfg.HostClaimOps == nil {
d := hostactors.Directory{Store: s.cfg.Releases, PublicKey: s.cfg.ReleasePub, Bucket: s.cfg.Bucket}
@@ -61,22 +63,3 @@ func (s *Server) RunClaimRecoveryPass(ctx context.Context) error {
}
return errors.Join(errs...)
}
-
-// A door may only consume a claim after the backfill has established it.
-func claimedRecord(ctx context.Context, c *hostops.Controller, key string) (hostclaim.Observed, error) {
- r, err := hostops.ReadRecord(ctx, c.Store, key)
- if 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 {
- return obs, hostops.ErrPending
- }
- if obs.Claim.Holder == nil || obs.Claim.Holder.Key != key {
- return obs, hostops.ErrIdentity
- }
- return obs, nil
-}
diff --git a/internal/console/hostclaimbackfill_test.go b/internal/console/hostclaimbackfill_test.go
new file mode 100644
index 000000000..e51440e14
--- /dev/null
+++ b/internal/console/hostclaimbackfill_test.go
@@ -0,0 +1,84 @@
+package console
+
+import (
+ "bytes"
+ "context"
+ "crypto/ed25519"
+ "crypto/rand"
+ "log"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/hostops"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestConfigureClaimBackfillMissingContractDoesNotFailStartup(t *testing.T) {
+ for _, missing := range []string{"store", "bucket", "public-key"} {
+ t.Run(missing, func(t *testing.T) {
+ s := &Server{cfg: Config{HostClaim: true, Releases: release.NewDirStore(t.TempDir()), Bucket: "bucket", ReleasePub: make(ed25519.PublicKey, ed25519.PublicKeySize)}}
+ switch missing {
+ case "store":
+ s.cfg.Releases = nil
+ case "bucket":
+ s.cfg.Bucket = ""
+ case "public-key":
+ s.cfg.ReleasePub = nil
+ }
+ var logs bytes.Buffer
+ previous := log.Writer()
+ log.SetOutput(&logs)
+ t.Cleanup(func() { log.SetOutput(previous) })
+ if err := s.ConfigureClaimBackfill(); err != nil {
+ t.Errorf("configuration failed startup: %v", err)
+ }
+ if s.cfg.HostClaimOps != nil || !strings.Contains(logs.String(), hostops.ErrContract.Error()) {
+ t.Fatalf("ops=%v log=%q", s.cfg.HostClaimOps, logs.String())
+ }
+ s.StartClaimRecoveryPass(context.Background())
+ })
+ }
+}
+
+func TestClaimBackfillQuietWithoutAdoptingPlan(t *testing.T) {
+ for _, phase := range []string{"absent", "collecting", "done", "future"} {
+ t.Run(phase, func(t *testing.T) {
+ pub, key, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ store := release.NewDirStore(t.TempDir())
+ if phase != "absent" {
+ raw, err := release.SignRootEnvelope(release.RootSigner(key), hostops.BackfillPlan{V: 1, ID: "backfill", Phase: phase, CreatedAt: time.Now()})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := store.Put(context.Background(), hostops.BackfillKey, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ s := &Server{cfg: Config{HostClaim: true, Releases: store, Bucket: "bucket", ReleasePub: pub}}
+ if err := s.ConfigureClaimBackfill(); err != nil {
+ t.Fatal(err)
+ }
+ ops := s.cfg.HostClaimOps
+ if ops == nil {
+ t.Fatal("backfill not configured")
+ }
+ if err := s.ConfigureClaimBackfill(); err != nil || s.cfg.HostClaimOps != ops {
+ t.Fatal("controller replaced", err)
+ }
+ var logs bytes.Buffer
+ previous := log.Writer()
+ log.SetOutput(&logs)
+ t.Cleanup(func() { log.SetOutput(previous) })
+ ctx, cancel := context.WithCancel(context.Background())
+ cancel()
+ s.StartClaimRecoveryPass(ctx)
+ if logs.Len() != 0 {
+ t.Fatalf("inactive backfill logged: %s", &logs)
+ }
+ })
+ }
+}
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 2c6f45185..239978987 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -2116,11 +2116,7 @@ func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *schedu
}
defer func(start time.Time) { logSlow("serial pass stages", start, strings.Join(took, ", ")) }(time.Now())
- pcfg := provisionTickConfig(os.Getenv)
- if err := reconcileBackfill(ctx, writes, store, pcfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET")); err != nil {
- log.Printf("agent: claim backfill: %v", err)
- }
- if err := reconcileSerialized(ctx, store, reg, pcfg); err != nil {
+ if err := reconcileSerialized(ctx, store, reg, provisionTickConfig(os.Getenv)); err != nil {
log.Printf("agent: fast converge reconcile: %v", err)
}
mark("reconcile")
@@ -2683,6 +2679,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config) {
return
}
cfg := provisionTickConfig(os.Getenv)
+ backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET"))
log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)
// Platform-only: a fleet box (CARLOS_FLEET set) never reads other
// fleets' bindings — it only ever WRITES its own (fleetbind.go's
@@ -2696,7 +2693,7 @@ func instancesLoop(reg *registry.Registry, writes hostwrite.Config) {
for {
time.Sleep(delay)
delay = 5*time.Minute + rand.N(time.Minute)
- if err := reconcileBackfill(ctx, writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET")); err != nil {
+ if err := backfill(ctx); err != nil {
log.Printf("agent: claim backfill: %v", err)
}
if err := reconcileSerialized(ctx, store, reg, cfg); err != nil {
diff --git a/internal/host/claimbackfill.go b/internal/host/claimbackfill.go
index d751062f8..51f7e7a2d 100644
--- a/internal/host/claimbackfill.go
+++ b/internal/host/claimbackfill.go
@@ -13,12 +13,12 @@ import (
"github.com/carlosframework/platform/internal/release"
)
-func reconcileBackfill(ctx context.Context, writes hostwrite.Config, s release.Store, pub ed25519.PublicKey, bucket string) error {
+func newClaimBackfill(writes hostwrite.Config, s release.Store, pub ed25519.PublicKey, bucket string) func(context.Context) error {
if !writes.Enabled {
- return nil
+ return func(context.Context) error { return nil }
}
if s == nil || len(pub) == 0 || bucket == "" {
- return hostops.ErrContract
+ return func(context.Context) error { return hostops.ErrContract }
}
d := hostactors.Directory{Store: s, PublicKey: pub, Bucket: bucket}
c := hostops.NewBackfillController(s, d, time.Now, func() string {
@@ -28,5 +28,5 @@ func reconcileBackfill(ctx context.Context, writes hostwrite.Config, s release.S
}
return hex.EncodeToString(b[:])
})
- return c.ReconcileClaims(ctx)
+ return c.ReconcileClaims
}
diff --git a/internal/host/claimbackfill_test.go b/internal/host/claimbackfill_test.go
new file mode 100644
index 000000000..c2d57e2a9
--- /dev/null
+++ b/internal/host/claimbackfill_test.go
@@ -0,0 +1,230 @@
+package host
+
+import (
+ "context"
+ "crypto/ed25519"
+ "crypto/rand"
+ "crypto/sha256"
+ "encoding/hex"
+ "encoding/json"
+ "errors"
+ "go/ast"
+ "go/parser"
+ "go/token"
+ "strings"
+ "testing"
+ "time"
+
+ "github.com/carlosframework/platform/internal/domains"
+ "github.com/carlosframework/platform/internal/hostactors"
+ "github.com/carlosframework/platform/internal/hostclaim"
+ "github.com/carlosframework/platform/internal/hostops"
+ "github.com/carlosframework/platform/internal/hostwrite"
+ "github.com/carlosframework/platform/internal/provision"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func TestClaimBackfillRunsOnlyOnInstancesCadence(t *testing.T) {
+ file, err := parser.ParseFile(token.NewFileSet(), "agent.go", nil, 0)
+ if err != nil {
+ t.Fatal(err)
+ }
+ for _, decl := range file.Decls {
+ fn, ok := decl.(*ast.FuncDecl)
+ if !ok {
+ continue
+ }
+ switch fn.Name.Name {
+ case "fastConvergeLoop":
+ ast.Inspect(fn.Body, func(n ast.Node) bool {
+ call, ok := n.(*ast.CallExpr)
+ if !ok {
+ return true
+ }
+ id, ok := call.Fun.(*ast.Ident)
+ if ok && (id.Name == "reconcileBackfill" || id.Name == "newClaimBackfill") {
+ t.Error("backfill runs on the serial promote path")
+ }
+ return true
+ })
+ case "instancesLoop":
+ constructors := 0
+ for _, stmt := range fn.Body.List {
+ loop, inLoop := stmt.(*ast.ForStmt)
+ var node ast.Node = stmt
+ if inLoop {
+ node = loop.Body
+ }
+ ast.Inspect(node, func(n ast.Node) bool {
+ call, ok := n.(*ast.CallExpr)
+ if !ok {
+ return true
+ }
+ id, ok := call.Fun.(*ast.Ident)
+ if ok && id.Name == "newClaimBackfill" {
+ constructors++
+ if inLoop {
+ t.Error("backfill controller constructed per tick")
+ }
+ }
+ return true
+ })
+ }
+ if constructors != 1 {
+ t.Errorf("instances loop has %d persistent backfill constructors, want 1", constructors)
+ }
+ }
+ }
+}
+
+type backfillPassStore struct {
+ release.Store
+ gets, censusReads int
+ failClaim error
+}
+
+func (s *backfillPassStore) Get(ctx context.Context, key string) ([]byte, string, error) {
+ s.gets++
+ if strings.HasPrefix(key, "control/host-census/") {
+ s.censusReads++
+ }
+ return s.Store.Get(ctx, key)
+}
+
+func (s *backfillPassStore) Put(ctx context.Context, key string, body []byte, tag string) (string, error) {
+ if strings.HasPrefix(key, "control/hosts/") && s.failClaim != nil {
+ err := s.failClaim
+ s.failClaim = nil
+ return "", err
+ }
+ return s.Store.Put(ctx, key, body, tag)
+}
+
+func TestClaimBackfillPassInertAndMissingContract(t *testing.T) {
+ store := &backfillPassStore{Store: release.NewDirStore(t.TempDir())}
+ ctx := context.Background()
+ if err := newClaimBackfill(hostwrite.Config{}, store, nil, "")(ctx); err != nil || store.gets != 0 {
+ t.Fatal("flag off accessed backfill", err, store.gets)
+ }
+ for _, missing := range []string{"store", "key", "bucket"} {
+ var s release.Store = store
+ pub := make(ed25519.PublicKey, ed25519.PublicKeySize)
+ bucket := "bucket"
+ switch missing {
+ case "store":
+ s = nil
+ case "key":
+ pub = nil
+ case "bucket":
+ bucket = ""
+ }
+ if err := newClaimBackfill(hostwrite.Config{Enabled: true}, s, pub, bucket)(ctx); !errors.Is(err, hostops.ErrContract) || store.gets != 0 {
+ t.Fatalf("missing %s: err=%v reads=%d", missing, err, store.gets)
+ }
+ }
+}
+
+func TestClaimBackfillPassKeepsCensusBetweenRetries(t *testing.T) {
+ ctx := context.Background()
+ store := &backfillPassStore{Store: release.NewDirStore(t.TempDir())}
+ pub, key, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ boxPub, boxKey, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ now := time.Now().UTC()
+ put := func(path string, value any, signed bool) {
+ t.Helper()
+ raw, err := json.Marshal(value)
+ if signed {
+ raw, err = release.SignRootEnvelope(release.RootSigner(key), value)
+ }
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := store.Put(ctx, path, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ enrollment := strings.Repeat("a", 32)
+ roster := hostactors.Roster{V: 1, DeploymentBucket: "bucket", Revision: 1,
+ Regions: map[string]hostactors.Region{"ie": {}},
+ Boxes: map[string]hostactors.Box{"box-a": {CensusPublicKey: boxPub, Enrollment: enrollment, Namespace: "platform", BoxID: "box-a", Label: "box-a", Region: "ie", Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: "ie", InstanceID: "i-box-a"}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}},
+ Consoles: map[string]hostactors.Console{}, Holds: map[string]hostactors.Reservation{}, Activations: map[string]hostactors.Reservation{}}
+ put(hostactors.Key, roster, true)
+ r, err := hostops.PrepareInstance(provision.Record{Schema: 1, Account: "bab", App: "jam", Host: "legacy.example.com", Slug: "shop", Channel: "edge", Region: "ie", Acquisition: "acquisition"}, now, "ie")
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := store.Put(ctx, r.Holder.Key, r.Body, ""); err != nil {
+ t.Fatal(err)
+ }
+ r, err = hostops.ReadRecord(ctx, store, r.Holder.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ inc, err := r.Incarnation()
+ if err != nil {
+ t.Fatal(err)
+ }
+ put(domains.ClaimKey(r.Holder.Host), domains.Claim{Schema: 1, Host: r.Holder.Host, Account: "bab", App: "jam", Acquisition: "acquisition", RecordIncarnation: inc, OwnershipState: "committed"}, false)
+ report := hostops.CensusReport{Box: "box-a", Enrollment: enrollment, BootID: "boot", Reservation: "backfill", DirectoryRevision: 1, CapturedAt: now, Complete: true, Installations: []hostops.CensusInstallation{{Host: r.Holder.Host, Account: "bab", App: "jam", Family: "instance", Key: r.Holder.Key, Incarnation: inc, Local: true, Route: true}}}
+ canonical, err := json.Marshal(report)
+ if err != nil {
+ t.Fatal(err)
+ }
+ reportPath := "control/host-census/backfill/box-a.json"
+ put(reportPath, hostops.CensusEnvelope{Report: report, Signature: ed25519.Sign(boxKey, canonical)}, false)
+ raw, _, err := store.Get(ctx, reportPath)
+ if err != nil {
+ t.Fatal(err)
+ }
+ digest := sha256.Sum256(raw)
+ put(hostops.BackfillKey, hostops.BackfillPlan{V: 1, ID: "backfill", Phase: "adopting", DirectoryRevision: 1, CreatedAt: now, Reports: map[string]string{"box-a": hex.EncodeToString(digest[:])}, Hosts: map[string]hostops.BackfillHost{r.Holder.Host: {Key: r.Holder.Key, ETag: r.ETag, Acquisition: "acquisition", Region: "ie", Box: "box-a"}}}, true)
+ store.censusReads = 0
+ pass := newClaimBackfill(hostwrite.Config{Enabled: true}, store, pub, "bucket")
+ fail := errors.New("claim write failed")
+ store.failClaim = fail
+ if err := pass(ctx); !errors.Is(err, fail) {
+ t.Fatalf("first pass: %v", err)
+ }
+ if err := pass(ctx); err != nil {
+ t.Fatalf("retry: %v", err)
+ }
+ if store.censusReads != 1 {
+ t.Fatalf("census reads=%d, want one across both passes", store.censusReads)
+ }
+ obs, err := hostclaim.Read(ctx, store, r.Holder.Host)
+ if err != nil || obs.Absent || obs.Claim.Holder.Acquisition != "acquisition" || obs.Claim.Op.Exec != "box-a" {
+ t.Fatalf("retry did not adopt: %+v %v", obs, err)
+ }
+}
+
+func TestClaimBackfillPassQuietWithoutAdoptingPlan(t *testing.T) {
+ for _, phase := range []string{"absent", "collecting", "done", "future"} {
+ t.Run(phase, func(t *testing.T) {
+ ctx := context.Background()
+ pub, key, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ store := &backfillPassStore{Store: release.NewDirStore(t.TempDir())}
+ if phase != "absent" {
+ raw, err := release.SignRootEnvelope(release.RootSigner(key), hostops.BackfillPlan{V: 1, ID: "backfill", Phase: phase, CreatedAt: time.Now()})
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := store.Put(ctx, hostops.BackfillKey, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ pass := newClaimBackfill(hostwrite.Config{Enabled: true}, store, pub, "bucket")
+ if err := pass(ctx); err != nil || store.gets != 1 {
+ t.Fatalf("inactive backfill: err=%v reads=%d", err, store.gets)
+ }
+ })
+ }
+}
diff --git a/internal/hostops/backfill.go b/internal/hostops/backfill.go
index b90607198..e20b19823 100644
--- a/internal/hostops/backfill.go
+++ b/internal/hostops/backfill.go
@@ -126,12 +126,22 @@ func CheckHostQuarantine(ctx context.Context, s release.Store, host string) erro
return nil
}
func (b *Backfill) read(ctx context.Context) (BackfillPlan, string, error) {
+ p, tag, err := b.readPlan(ctx)
+ if err != nil {
+ return p, "", err
+ }
+ if p.Phase != "adopting" {
+ return p, "", ErrContract
+ }
+ return p, tag, nil
+}
+func (b *Backfill) readPlan(ctx context.Context) (BackfillPlan, string, error) {
var p BackfillPlan
raw, tag, err := b.Store.Get(ctx, BackfillKey)
if err != nil {
return p, "", err
}
- if tag == "" || release.VerifyRootEnvelope(b.Directory.PublicKey, raw, &p) != nil || p.V != 1 || !safeSegment(p.ID) || p.Phase != "adopting" || p.CreatedAt.IsZero() || p.Hosts == nil || p.Reports == nil {
+ if tag == "" || release.VerifyRootEnvelope(b.Directory.PublicKey, raw, &p) != nil || p.V != 1 || !safeSegment(p.ID) || p.CreatedAt.IsZero() || (p.Phase == "adopting" && (p.Hosts == nil || p.Reports == nil)) {
return p, "", ErrContract
}
return p, tag, nil
@@ -326,16 +336,6 @@ func (b *Backfill) Adoption(ctx context.Context, r Record) (region, box string,
return
}
}
- eligible := 0
- for _, candidate := range census.Roster.Roster.Boxes {
- if candidate.State == "active" && candidate.Region == selected.Region && candidate.Namespace == enrolled.Namespace && candidate.Pool == enrolled.Pool {
- eligible++
- }
- }
- if eligible != 1 {
- err = ErrContract
- return
- }
raw, _, e := b.Store.Get(ctx, domains.ClaimKey(r.Holder.Host))
if e != nil {
err = e
diff --git a/internal/hostops/backfill_controller.go b/internal/hostops/backfill_controller.go
index f6dcbe5ca..08ebd8567 100644
--- a/internal/hostops/backfill_controller.go
+++ b/internal/hostops/backfill_controller.go
@@ -50,12 +50,16 @@ func ReconcileClaims(ctx context.Context, c Controller, adopt func(context.Conte
backfill = provider.ClaimBackfill()
}
if backfill != nil {
- if _, _, err := backfill.read(ctx); err != nil {
- if errors.Is(err, os.ErrNotExist) {
- return ErrPending
- }
+ plan, _, err := backfill.readPlan(ctx)
+ if errors.Is(err, os.ErrNotExist) {
+ return nil
+ }
+ if err != nil {
return err
}
+ if plan.Phase != "adopting" {
+ return nil
+ }
}
apps, err := Apps(ctx, c.Store)
if err != nil {
diff --git a/internal/hostops/backfill_test.go b/internal/hostops/backfill_test.go
index 32ca43ff7..496d55747 100644
--- a/internal/hostops/backfill_test.go
+++ b/internal/hostops/backfill_test.go
@@ -20,14 +20,14 @@ import (
)
func TestReconcileFrozenBackfillAndQuarantine(t *testing.T) {
- testReconcileFrozenBackfill(t, false)
+ testReconcileFrozenBackfill(t, false, "")
}
func TestReconcileFrozenBackfillRepairsLegacyHolder(t *testing.T) {
- testReconcileFrozenBackfill(t, true)
+ testReconcileFrozenBackfill(t, true, "")
}
-func testReconcileFrozenBackfill(t *testing.T, legacy bool) {
+func testReconcileFrozenBackfill(t *testing.T, legacy bool, scenario string) {
t.Helper()
ctx := context.Background()
f := newFixture(t)
@@ -52,6 +52,20 @@ func testReconcileFrozenBackfill(t *testing.T, legacy bool) {
}
roster.Boxes[ref] = hostactors.Box{CensusPublicKey: boxPub, Enrollment: en, Namespace: "platform", BoxID: ref, Label: ref, Region: region, Pool: "default", Cloud: hostactors.Cloud{Provider: "aws", Account: "123", Region: region, InstanceID: "i-" + ref}, EvidenceSHA256: strings.Repeat("c", 64), State: "active"}
}
+ if scenario == "two-box-region" || scenario == "two-owners" {
+ box := roster.Boxes["box-b"]
+ box.Region, box.Cloud.Region = "ie", "ie"
+ roster.Boxes["box-b"] = box
+ }
+ if scenario == "inactive-owner" || scenario == "wrong-region" {
+ box := roster.Boxes["box-a"]
+ if scenario == "inactive-owner" {
+ box.State = "retired"
+ } else {
+ box.Region, box.Cloud.Region = "us", "us"
+ }
+ roster.Boxes["box-a"] = box
+ }
put := func(k string, v any, signed bool) string {
t.Helper()
var raw []byte
@@ -89,7 +103,7 @@ func testReconcileFrozenBackfill(t *testing.T, legacy bool) {
t.Fatal(err)
}
put(domains.ClaimKey(r.Holder.Host), domains.Claim{Schema: 1, Host: r.Holder.Host, Account: "bab", App: "jam", Acquisition: "ownership-acquisition", RecordIncarnation: inc, OwnershipState: "committed"}, false)
- if err = c.ReconcileClaims(ctx); !errors.Is(err, ErrPending) {
+ if err = c.ReconcileClaims(ctx); err != nil {
t.Fatalf("unfrozen adoption: %v", err)
}
obs, err := hostclaim.Read(ctx, f.store, r.Holder.Host)
@@ -116,7 +130,7 @@ func testReconcileFrozenBackfill(t *testing.T, legacy bool) {
p := BackfillPlan{V: 1, ID: "backfill", Phase: "adopting", DirectoryRevision: 1, CreatedAt: f.now, Reports: map[string]string{}, Hosts: map[string]BackfillHost{r.Holder.Host: {Key: r.Holder.Key, ETag: r.ETag, Acquisition: "ownership-acquisition", Region: "ie", Box: "box-a"}}}
for ref, b := range roster.Boxes {
report := CensusReport{Box: ref, Enrollment: b.Enrollment, BootID: "boot", Reservation: p.ID, DirectoryRevision: 1, CapturedAt: f.now, Complete: true, Installations: []CensusInstallation{}}
- if ref == "box-a" {
+ if ref == "box-a" || scenario == "two-owners" {
report.Installations = append(report.Installations, CensusInstallation{Host: r.Holder.Host, Account: "bab", App: "jam", Family: "instance", Key: r.Holder.Key, Incarnation: inc, Local: true, Route: true})
}
key, _ := reportKey(p.ID, ref)
@@ -182,6 +196,16 @@ func testReconcileFrozenBackfill(t *testing.T, legacy bool) {
if _, e = f.store.Put(ctx, BackfillKey, signed, manifestTag); e != nil {
t.Fatal(e)
}
+ if scenario == "two-owners" || scenario == "inactive-owner" || scenario == "wrong-region" {
+ if err = c.ReconcileClaims(ctx); !errors.Is(err, ErrIdentity) {
+ t.Fatalf("ineligible census owner: %v", err)
+ }
+ obs, err = hostclaim.Read(ctx, f.store, r.Holder.Host)
+ if err != nil || !obs.Absent {
+ t.Fatalf("ineligible owner adopted: %+v %v", obs, err)
+ }
+ return
+ }
if err = c.ReconcileClaims(ctx); err != nil {
t.Fatal(err)
}
@@ -282,3 +306,52 @@ func TestBackfillAuthorityCannotGainMutationMethods(t *testing.T) {
t.Fatal("backfill gained effect authority")
}
}
+
+func TestReconcileBackfillCensusOwnerEligibility(t *testing.T) {
+ for _, scenario := range []string{"two-box-region", "two-owners", "inactive-owner", "wrong-region"} {
+ t.Run(scenario, func(t *testing.T) {
+ testReconcileFrozenBackfill(t, false, scenario)
+ })
+ }
+}
+
+func TestReconcileBackfillPlanPhases(t *testing.T) {
+ for _, phase := range []string{"absent", "collecting", "done", "future", "adopting", "bad-signature", "read-error"} {
+ t.Run(phase, func(t *testing.T) {
+ f := newFixture(t)
+ pub, key, err := ed25519.GenerateKey(rand.Reader)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if phase != "absent" && phase != "read-error" {
+ p := BackfillPlan{V: 1, ID: "backfill", Phase: phase, CreatedAt: f.now, Hosts: map[string]BackfillHost{}}
+ raw, err := release.SignRootEnvelope(release.RootSigner(key), p)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err = f.store.Put(f.ctx, BackfillKey, raw, ""); err != nil {
+ t.Fatal(err)
+ }
+ }
+ if phase == "bad-signature" {
+ pub = make(ed25519.PublicKey, ed25519.PublicKeySize)
+ }
+ if phase == "read-error" {
+ f.store.failGet = BackfillKey
+ }
+ c := NewBackfillController(f.store, hostactors.Directory{Store: f.store, PublicKey: pub, Bucket: "bucket"}, f.c.Now, f.c.NewID)
+ lists, adopts := 0, 0
+ f.store.afterList = func(string, []string) error { lists++; return nil }
+ err = ReconcileClaims(f.ctx, c, func(context.Context, string) (hostclaim.Observed, error) { adopts++; return hostclaim.Observed{}, nil })
+ wantError := phase == "adopting" || phase == "bad-signature" || phase == "read-error"
+ if (err != nil) != wantError || lists != 0 || adopts != 0 {
+ t.Fatalf("phase %s: err=%v lists=%d adopts=%d", phase, err, lists, adopts)
+ }
+ if !wantError && phase != "absent" {
+ if _, _, _, _, err := c.Authority.Adoption(f.ctx, Record{}); err == nil {
+ t.Fatal("inactive plan granted direct adoption")
+ }
+ }
+ })
+ }
+}
diff --git a/internal/hostops/operations.go b/internal/hostops/operations.go
index aea6021d2..8313b80e6 100644
--- a/internal/hostops/operations.go
+++ b/internal/hostops/operations.go
@@ -4,6 +4,7 @@ import (
"context"
"errors"
"os"
+ "slices"
"time"
"github.com/carlosframework/platform/internal/hostclaim"
@@ -155,9 +156,20 @@ func (c Controller) Adopt(ctx context.Context, r Record) (hostclaim.Observed, er
if !matches(r, &prior) {
return obs, ErrIdentity
}
- if _, _, _, _, err := c.Authority.Adoption(ctx, r); err != nil {
+ _, box, _, _, err := c.Authority.Adoption(ctx, r)
+ if err != nil {
return obs, err
}
+ if obs.Claim.Op.Kind != hostclaim.KindServing || obs.Claim.Op.Exec != box || !slices.Contains(obs.Claim.Touched.Boxes, box) {
+ return obs, ErrIdentity
+ }
+ fresh, err := ReadRecord(ctx, c.Store, r.Holder.Key)
+ if err != nil {
+ return obs, err
+ }
+ if fresh.ETag != r.ETag {
+ return obs, ErrPending
+ }
return hostclaim.Update(ctx, c.Store, obs, func(next *hostclaim.Claim) error {
if next.Holder == nil || next.Holder.Acquisition != "" {
return ErrPending
diff --git a/internal/hostops/operations_test.go b/internal/hostops/operations_test.go
index e9526e34c..42ebf9d11 100644
--- a/internal/hostops/operations_test.go
+++ b/internal/hostops/operations_test.go
@@ -2,6 +2,7 @@ package hostops
import (
"context"
+ "encoding/json"
"errors"
"fmt"
"os"
@@ -531,3 +532,98 @@ func TestFenceReleasePreservesHistoryAndRefusesAnotherPurge(t *testing.T) {
t.Fatal("lost history")
}
}
+
+type repairAuthority struct {
+ Authority
+ afterAdoption func()
+}
+
+func (a repairAuthority) Adoption(ctx context.Context, r Record) (string, string, []string, []string, error) {
+ region, box, regions, boxes, err := a.Authority.Adoption(ctx, r)
+ if a.afterAdoption != nil {
+ a.afterAdoption()
+ }
+ return region, box, regions, boxes, err
+}
+
+func TestAdoptLegacyRepairRequiresCensusIdentityAndFreshRecord(t *testing.T) {
+ for _, scenario := range []string{"wrong-executor", "not-serving", "owner-not-touched", "changed-record", "missing-record", "record-read-error"} {
+ t.Run(scenario, func(t *testing.T) {
+ f := newFixture(t)
+ before := f.serving(t, "legacy.example.com")
+ var err error
+ before, err = hostclaim.Update(f.ctx, f.store, before, func(c *hostclaim.Claim) error {
+ switch scenario {
+ case "wrong-executor":
+ c.Op.Exec = "box-b"
+ c.Touched.Boxes = append(c.Touched.Boxes, "box-b")
+ case "not-serving":
+ c.Op.Kind, c.Op.ID = hostclaim.KindMoving, "move"
+ case "owner-not-touched":
+ c.Touched.Boxes = []string{"box-b"}
+ }
+ return nil
+ })
+ if err != nil {
+ t.Fatal(err)
+ }
+ r, err := ReadRecord(f.ctx, f.store, before.Claim.Holder.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ var body provision.Record
+ if err := json.Unmarshal(r.Body, &body); err != nil {
+ t.Fatal(err)
+ }
+ body.Acquisition = "ownership-acquisition"
+ raw, err := json.Marshal(body)
+ if err != nil {
+ t.Fatal(err)
+ }
+ if _, err := f.store.Put(f.ctx, r.Holder.Key, raw, r.ETag); err != nil {
+ t.Fatal(err)
+ }
+ r, err = ReadRecord(f.ctx, f.store, r.Holder.Key)
+ if err != nil {
+ t.Fatal(err)
+ }
+ want := ErrIdentity
+ if scenario == "changed-record" {
+ want = ErrPending
+ } else if scenario == "missing-record" {
+ want = os.ErrNotExist
+ }
+ f.c.Authority = repairAuthority{Authority: f.auth, afterAdoption: func() {
+ switch scenario {
+ case "changed-record":
+ if _, err := f.store.Put(f.ctx, r.Holder.Key, append(raw, ' '), r.ETag); err != nil {
+ t.Fatal(err)
+ }
+ case "missing-record":
+ if err := f.store.DeleteIfMatch(f.ctx, r.Holder.Key, r.ETag); err != nil {
+ t.Fatal(err)
+ }
+ case "record-read-error":
+ f.store.failGet = r.Holder.Key
+ }
+ }}
+ writes := 0
+ f.store.beforePut = func(string) error { writes++; return nil }
+ _, err = f.c.Adopt(f.ctx, r)
+ if scenario == "record-read-error" {
+ if err == nil || err.Error() != "read failed" {
+ t.Errorf("repair error = %v, want record read failure", err)
+ }
+ } else if !errors.Is(err, want) {
+ t.Errorf("repair error = %v, want %v", err, want)
+ }
+ if scenario == "changed-record" {
+ writes--
+ }
+ after, readErr := hostclaim.Read(f.ctx, f.store, before.Host)
+ if readErr != nil || writes != 0 || after.ETag != before.ETag || !reflect.DeepEqual(after.Claim, before.Claim) {
+ t.Fatalf("refused repair changed claim: writes=%d, read=%v", writes, readErr)
+ }
+ })
+ }
+}