host: steering converger rides the housekeeper
The duty that wires internal/steering into the agent: its own loop beside mintMeshKeyLoop (fleet-skip, openDeployStore, 5min + jitter ticks of runHousekeeperDuties), plus ConvergeDeployment — the package's one exported entry point, which folds registry rows and raw edges-file bodies and converges in a single call a caller cannot hand a set the fold did not build. Reads FAIL CLOSED, the durable half of Task 5's shrink-guard finding: a control/edges/ listing that errors, ANY edges file that lists but will not Get, or a registry read that fails returns the error WITHOUT converging. A short fold deletes the records of every box that went missing — an NXDOMAIN caused by a bucket blip. A file that reads but does not PARSE is the other class entirely: a content problem, skipped and logged, the pass carries on. Dormant on every box today: CARLOS_STEERING_ZONE_ID unset (or malformed — loudly, never a crash loop) means one log line at arm and a duty that returns nil before touching the closure holding the store, the registry and the AWS client. Zero reads, zero API calls, structurally.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
6 files changed,
+781
−1
internal/host/agent.go+5 −0internal/host/edgesfile.go+9 −1internal/host/steering_loop.go+196 −0internal/host/steering_loop_test.go+347 −0internal/steering/deployment.go+73 −0internal/steering/deployment_test.go+151 −0
diff --git a/internal/host/agent.go b/internal/host/agent.go| index 3f9e5dda4..f1ba38497 100644 |
| --- a/internal/host/agent.go |
| +++ b/internal/host/agent.go |
| @@ -329,6 +329,11 @@ func RunAgent(args []string) { |
| // only inside retentionLoop would then silently never run fleet-wide. |
| // See mintMeshKeyLoop's own comment (housekeeper.go). |
| go mintMeshKeyLoop() |
| + // The DNS steering converger (multi-region ingress spec, Part 4) — its |
| + // own loop for mintMeshKeyLoop's exact reason, and dormant on every box |
| + // today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and |
| + // nothing else, ever. See steering_loop.go. |
| + go steeringLoop(reg) |
| // The sweep reads the registry only through this closure — the stamped |
| // bespoke units to sample beside the carlos- ones, re-read each tick. |
| go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }) |
diff --git a/internal/host/edgesfile.go b/internal/host/edgesfile.go| index 1ff616d8a..8b18bb192 100644 |
| --- a/internal/host/edgesfile.go |
| +++ b/internal/host/edgesfile.go |
| @@ -20,7 +20,15 @@ import ( |
| // the fleet's per-box files into a deployment-wide host→home map. v1 writes |
| // only `role: "home"` — the anticipated read-local role values are Task 3's |
| // concern, not this file's. |
| -func edgesFileKey(box string) string { return "control/edges/" + box + ".json" } |
| +// |
| +// edgesFilePrefix is the flat prefix those files live under. It is named |
| +// rather than spelled twice because there is now a READER on this side too — |
| +// steering_loop.go lists it to fold the fleet's files (internal/edge's |
| +// homeMap has its own copy, on the far side of an import boundary this |
| +// package does not cross). |
| +const edgesFilePrefix = "control/edges/" |
| + |
| +func edgesFileKey(box string) string { return edgesFilePrefix + box + ".json" } |
| // edgeHost is the per-host entry in edgesFile.Hosts. An object, not a bare |
| // string, is deliberately load-bearing: it is read-local anticipation — |
diff --git a/internal/host/steering_loop.go b/internal/host/steering_loop.go| new file mode 100644 |
| index 000000000..743f6a43f |
| --- /dev/null |
| +++ b/internal/host/steering_loop.go |
| @@ -0,0 +1,196 @@ |
| +package host |
| + |
| +// steering_loop.go — the DNS steering converger's housekeeper loop |
| +// (multi-region ingress spec, Part 4; plan 2026-08-23-multi-region-steering |
| +// Task 6). The converger itself lives in internal/steering; this file is the |
| +// wiring: what it reads, when it runs, and what it refuses to run on. |
| + |
| +import ( |
| + "context" |
| + "fmt" |
| + "log" |
| + "math/rand/v2" |
| + "os" |
| + "sort" |
| + "strings" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/steering" |
| +) |
| + |
| +// loadEdgesFileBodies reads every box's edges file under control/edges/ and |
| +// returns the raw bodies in listing order (sorted by key, so the fold |
| +// downstream is stable tick to tick). |
| +// |
| +// FAIL-CLOSED, and this is the whole point of the function: a listing that |
| +// errors, or ANY object that lists but will not Get, returns an ERROR and no |
| +// bodies. The alternative — carrying on with what came back — is the SHORT |
| +// READ that Task 5's shrink guard was added to survive: the fold would be |
| +// missing a live box's edge, every record belonging to that box would look |
| +// managed-but-unwanted, and the pass would delete them. NXDOMAIN for a live |
| +// steered host, caused by a bucket blip rather than by anything an operator |
| +// did. |
| +// |
| +// This is deliberately STRICTER than internal/edge's homeMap, which reads the |
| +// same prefix and tolerates a single unreadable object by carrying that box's |
| +// previous parse forward. homeMap can do that because it HOLDS a last-good |
| +// snapshot; this pass holds nothing between ticks, so "tolerate" here would |
| +// mean "silently converge on less". Even a raced delete (list saw the key, |
| +// Get says it is gone — a box retired mid-pass) fails the pass: the cost is |
| +// one skipped tick, five minutes later the listing no longer names it, and |
| +// the withdrawal happens on complete information instead of a race. |
| +// |
| +// A key that is not <box>.json directly under the prefix is not read at all — |
| +// the same key-shape filter homeMap applies. A stray object under the prefix |
| +// is not this deployment's edges file, and must not be able to fail the pass |
| +// forever. |
| +func loadEdgesFileBodies(ctx context.Context, s release.Store) ([][]byte, error) { |
| + keys, err := s.List(ctx, edgesFilePrefix) |
| + if err != nil { |
| + return nil, fmt.Errorf("steering: list %s: %w", edgesFilePrefix, err) |
| + } |
| + sort.Strings(keys) |
| + out := make([][]byte, 0, len(keys)) |
| + for _, key := range keys { |
| + name, ok := strings.CutPrefix(key, edgesFilePrefix) |
| + if !ok || !strings.HasSuffix(name, ".json") || strings.Contains(name, "/") { |
| + continue |
| + } |
| + body, _, err := s.Get(ctx, key) |
| + if err != nil { |
| + return nil, fmt.Errorf("steering: read %s: %w", key, err) |
| + } |
| + out = append(out, body) |
| + } |
| + return out, nil |
| +} |
| + |
| +// steeringPass is one converge attempt: read both inputs, then converge. |
| +// |
| +// Both reads fail closed (see loadEdgesFileBodies) and BOTH must succeed |
| +// before converge is called even once — a routes query that failed is not |
| +// "no steered hosts" any more than a failed listing is "no edges". converge |
| +// is injected rather than called directly so this contract is testable |
| +// without an AWS client: the real one is the closure steeringLoop builds |
| +// around steering.ConvergeDeployment. |
| +// |
| +// listRoutes is the registry's List, injected for the same reason. |
| +func steeringPass(ctx context.Context, s release.Store, listRoutes func() ([]registry.Route, error), converge func(context.Context, []registry.Route, [][]byte) error) error { |
| + edges, err := loadEdgesFileBodies(ctx, s) |
| + if err != nil { |
| + return err |
| + } |
| + routes, err := listRoutes() |
| + if err != nil { |
| + return fmt.Errorf("steering: list routes: %w", err) |
| + } |
| + return converge(ctx, routes, edges) |
| +} |
| + |
| +// steeringDuty wraps one pass as a housekeeper duty. |
| +// |
| +// DORMANCY (zoneID == "") is enforced here, at the top, and it is a |
| +// structural claim rather than a promise: pass is the closure holding the |
| +// store handle, the registry and the AWS client, so a dormant duty that never |
| +// calls it makes zero reads and zero API calls by construction. Every box in |
| +// production today is in exactly this state. |
| +// |
| +// DUTY-STATUS HONESTY — read this before trusting a green tick. A nil return |
| +// from this duty does NOT mean the zone converged. internal/steering's two |
| +// safety refusals (the empty-desired-set refusal and the proportional shrink |
| +// guard) both log loudly and return nil, deliberately: the duty cannot fix |
| +// either, and marking it failed every five minutes would bury the one line |
| +// that says what actually happened. So a refused pass is indistinguishable |
| +// from a clean one at THIS layer, by design, and the operator surface for |
| +// steering is the JOURNAL — `journalctl -u carlos -g "^steering:"` — not the |
| +// housekeeper's error log. |
| +// |
| +// The case with no escape hatch, recorded for the rollout notes: a GENUINE |
| +// withdrawal of more than half the managed records (a real multi-box |
| +// retirement) is refused on every tick, forever, with no flag to override it. |
| +// Recovery is an operator deleting the records by hand, or withdrawing the |
| +// boxes in batches under the limit. Threading a status back up through the |
| +// duty was considered and left out of this task on purpose — it changes the |
| +// housekeeperDuty contract for every duty, and the journal already carries |
| +// the truth. |
| +func steeringDuty(zoneID string, pass func(context.Context) error) housekeeperDuty { |
| + return housekeeperDuty{"steering converge", func(ctx context.Context) error { |
| + if zoneID == "" { |
| + return nil |
| + } |
| + return pass(ctx) |
| + }} |
| +} |
| + |
| +// steeringLoop converges the deployment's steering zone every 5 minutes + |
| +// jitter, on the one box holding the housekeeper lease. |
| +// |
| +// Its shape is mintMeshKeyLoop's, deliberately and for the same reason |
| +// (housekeeper.go's own comment): its OWN loop, gated on nothing but |
| +// deployment-store presence, never a duty folded into retentionLoop's tick — |
| +// that loop never arms at all on a box missing CARLOS_RELEASE_PUBKEY, and |
| +// such a box still WINS the housekeeper election, so a steering duty living |
| +// there would silently never converge, fleet-wide, with no error anywhere. |
| +// |
| +// Env, read ONCE at arm (domainsLoop's read-once precedent): |
| +// |
| +// - CARLOS_STEERING_ZONE_ID — the zone. Unset, blank, or malformed and the |
| +// loop still ARMS but its duty is dormant: one log line here, and nothing |
| +// else, ever. Malformed is logged loudly by SteeringZoneID itself and is |
| +// dormancy, never a crash loop — a typo must not take a box's agent down. |
| +// - PLATFORM_APPS_DOMAIN — the domain the synthetic edge.<apps-domain> |
| +// steering hostname is built from. Empty is not fatal: the fold logs it |
| +// and mints no steering hostname (steered per-host records are |
| +// unaffected). |
| +// |
| +// FIRST REAL BEHAVIOUR, when a deployment does set the zone id while still |
| +// running ONE box: the steering hostname and every steered host get a |
| +// single-edge latency record — the same answers today's static DNS gives — |
| +// plus one Route53 health check per EIP, which is what adds rung-2 |
| +// withdrawal for steered hosts even single-box. Nothing about it needs a |
| +// second region to be correct. |
| +func steeringLoop(reg *registry.Registry) { |
| + ctx := context.Background() |
| + if shouldSkipPublisherElection(os.Getenv) { |
| + log.Printf("agent: steering loop is a platform-box job — skipped on a fleet box") |
| + return |
| + } |
| + store, err := openDeployStore(ctx) |
| + if err != nil { |
| + log.Printf("agent: steering loop disabled: %v", err) |
| + return |
| + } |
| + |
| + zoneID := steering.SteeringZoneID() |
| + var pass func(context.Context) error |
| + if zoneID == "" { |
| + log.Printf("agent: steering: zone unset — dormant (CARLOS_STEERING_ZONE_ID is unset or unusable; the duty reads nothing and calls no AWS API)") |
| + log.Printf("agent: steering loop armed (5min + jitter) — dormant") |
| + } else { |
| + // Constructed only on a deployment that actually steers: the default |
| + // credential chain is the box instance profile, and a dormant box has |
| + // no business resolving credentials at all. |
| + client, err := steering.NewClient(ctx) |
| + if err != nil { |
| + log.Printf("agent: steering loop disabled: %v", err) |
| + return |
| + } |
| + appsDomain := strings.TrimSpace(os.Getenv("PLATFORM_APPS_DOMAIN")) |
| + pass = func(ctx context.Context) error { |
| + return steeringPass(ctx, store, reg.List, func(ctx context.Context, routes []registry.Route, edges [][]byte) error { |
| + return steering.ConvergeDeployment(ctx, client, zoneID, routes, edges, appsDomain) |
| + }) |
| + } |
| + log.Printf("agent: steering loop armed (5min + jitter), zone %s, apps domain %q — converges only while this box holds the housekeeper lease", zoneID, appsDomain) |
| + } |
| + |
| + duty := steeringDuty(zoneID, pass) |
| + delay := time.Minute |
| + for { |
| + time.Sleep(delay) |
| + delay = 5*time.Minute + rand.N(time.Minute) |
| + runHousekeeperDuties(housekeeperGate, duty) |
| + } |
| +} |
diff --git a/internal/host/steering_loop_test.go b/internal/host/steering_loop_test.go| new file mode 100644 |
| index 000000000..60c923620 |
| --- /dev/null |
| +++ b/internal/host/steering_loop_test.go |
| @@ -0,0 +1,347 @@ |
| +package host |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// ---- fakes ----------------------------------------------------------------- |
| + |
| +// failingListStore is a store whose control/edges/ LISTING is broken — the |
| +// control-bucket blip that must never reach the converger as "no edges". |
| +type failingListStore struct { |
| + release.Store |
| + err error |
| +} |
| + |
| +func (f failingListStore) List(context.Context, string) ([]string, error) { return nil, f.err } |
| + |
| +// failingGetStore lists fine but refuses ONE named key — the half-read that |
| +// would otherwise fold to a short desired set and delete the missing box's |
| +// records. |
| +type failingGetStore struct { |
| + release.Store |
| + bad string |
| + err error |
| +} |
| + |
| +func (f failingGetStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == f.bad { |
| + return nil, "", f.err |
| + } |
| + return f.Store.Get(ctx, key) |
| +} |
| + |
| +// convergeRecorder captures what the pass handed the converger, and whether |
| +// it was called at all — the assertion every fail-closed test makes. |
| +type convergeRecorder struct { |
| + calls int |
| + routes []registry.Route |
| + edges [][]byte |
| + err error |
| +} |
| + |
| +func (c *convergeRecorder) converge(_ context.Context, routes []registry.Route, edges [][]byte) error { |
| + c.calls++ |
| + c.routes = routes |
| + c.edges = edges |
| + return c.err |
| +} |
| + |
| +func steeringTestStore(t *testing.T, files map[string][]byte) release.Store { |
| + t.Helper() |
| + s := release.NewDirStore(t.TempDir()) |
| + for key, body := range files { |
| + if _, err := s.Put(context.Background(), key, body, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + return s |
| +} |
| + |
| +func okRoutes(routes ...registry.Route) func() ([]registry.Route, error) { |
| + return func() ([]registry.Route, error) { return routes, nil } |
| +} |
| + |
| +// ---- the pass: fail-closed reads (H2) -------------------------------------- |
| + |
| +// A healthy pass hands the converger every edges file body under the prefix, |
| +// in listing order, plus the registry's routes. |
| +func TestSteeringPassReadsEdgesFilesAndRoutes(t *testing.T) { |
| + quietLog(t) |
| + s := steeringTestStore(t, map[string][]byte{ |
| + "control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`), |
| + "control/edges/oregon.json": []byte(`{"v":1,"box":"oregon"}`), |
| + }) |
| + rec := &convergeRecorder{} |
| + |
| + if err := steeringPass(context.Background(), s, okRoutes(registry.Route{Host: "app.example.com"}), rec.converge); err != nil { |
| + t.Fatalf("steeringPass: %v", err) |
| + } |
| + if rec.calls != 1 { |
| + t.Fatalf("converge calls = %d, want 1", rec.calls) |
| + } |
| + if len(rec.edges) != 2 { |
| + t.Fatalf("edges = %d bodies, want 2", len(rec.edges)) |
| + } |
| + // Listing order is sorted, so the fold is deterministic tick to tick. |
| + if !strings.Contains(string(rec.edges[0]), "dublin") || !strings.Contains(string(rec.edges[1]), "oregon") { |
| + t.Fatalf("edges = %q, want dublin then oregon (sorted)", rec.edges) |
| + } |
| + if len(rec.routes) != 1 || rec.routes[0].Host != "app.example.com" { |
| + t.Fatalf("routes = %+v, want the registry's one row", rec.routes) |
| + } |
| +} |
| + |
| +// H2, half one: a broken LISTING must not converge. A converger handed "no |
| +// edges" folds to an empty desired set; Converge's own refusal would catch |
| +// that particular case, but a listing that came back SHORT would not be |
| +// caught at all — so the read fails the duty instead, before any of it. |
| +func TestSteeringPassFailsClosedOnListError(t *testing.T) { |
| + quietLog(t) |
| + boom := errors.New("bucket down") |
| + s := failingListStore{Store: steeringTestStore(t, nil), err: boom} |
| + rec := &convergeRecorder{} |
| + |
| + err := steeringPass(context.Background(), s, okRoutes(), rec.converge) |
| + if err == nil { |
| + t.Fatal("steeringPass returned nil on a failed edges listing") |
| + } |
| + if !errors.Is(err, boom) { |
| + t.Fatalf("err = %v, want it to wrap the store error", err) |
| + } |
| + if rec.calls != 0 { |
| + t.Fatalf("converge ran %d times after a failed listing, want 0 — never converge on a partial fold", rec.calls) |
| + } |
| +} |
| + |
| +// H2, half two, and the one the shrink guard was written for: the listing |
| +// succeeded, so the pass KNOWS how many boxes there are — and one of their |
| +// files would not come back. Converging on the rest is exactly the short read |
| +// that deletes a live edge's records. |
| +func TestSteeringPassFailsClosedOnGetError(t *testing.T) { |
| + quietLog(t) |
| + boom := errors.New("object unreadable") |
| + s := failingGetStore{ |
| + Store: steeringTestStore(t, map[string][]byte{ |
| + "control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`), |
| + "control/edges/oregon.json": []byte(`{"v":1,"box":"oregon"}`), |
| + }), |
| + bad: "control/edges/oregon.json", |
| + err: boom, |
| + } |
| + rec := &convergeRecorder{} |
| + |
| + err := steeringPass(context.Background(), s, okRoutes(), rec.converge) |
| + if err == nil { |
| + t.Fatal("steeringPass returned nil with one edges file unreadable") |
| + } |
| + if !errors.Is(err, boom) { |
| + t.Fatalf("err = %v, want it to wrap the store error", err) |
| + } |
| + if !strings.Contains(err.Error(), "control/edges/oregon.json") { |
| + t.Fatalf("err = %v, want the unreadable key named", err) |
| + } |
| + if rec.calls != 0 { |
| + t.Fatalf("converge ran %d times after a failed get, want 0", rec.calls) |
| + } |
| +} |
| + |
| +// H2, half three: the registry read is the other input, and the same rule |
| +// holds — a routes query that failed is not "no steered hosts". |
| +func TestSteeringPassFailsClosedOnRegistryError(t *testing.T) { |
| + quietLog(t) |
| + boom := errors.New("registry locked") |
| + s := steeringTestStore(t, map[string][]byte{"control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`)}) |
| + rec := &convergeRecorder{} |
| + |
| + err := steeringPass(context.Background(), s, func() ([]registry.Route, error) { return nil, boom }, rec.converge) |
| + if err == nil { |
| + t.Fatal("steeringPass returned nil on a failed registry read") |
| + } |
| + if !errors.Is(err, boom) { |
| + t.Fatalf("err = %v, want it to wrap the registry error", err) |
| + } |
| + if rec.calls != 0 { |
| + t.Fatalf("converge ran %d times after a failed registry read, want 0", rec.calls) |
| + } |
| +} |
| + |
| +// The distinction H2 turns on: a file that READS but does not PARSE is a |
| +// CONTENT problem, and the host loader must not pre-judge it — the body goes |
| +// down to the fold, which skips it and says so (steering's own test). Only |
| +// READ failures fail the duty. |
| +func TestSteeringPassPassesUnparseableBodiesDown(t *testing.T) { |
| + quietLog(t) |
| + s := steeringTestStore(t, map[string][]byte{ |
| + "control/edges/dublin.json": []byte("{not json"), |
| + }) |
| + rec := &convergeRecorder{} |
| + |
| + if err := steeringPass(context.Background(), s, okRoutes(), rec.converge); err != nil { |
| + t.Fatalf("steeringPass: %v — an unparseable body is not a read failure", err) |
| + } |
| + if rec.calls != 1 { |
| + t.Fatalf("converge calls = %d, want 1 — parse failures are the fold's business", rec.calls) |
| + } |
| + if len(rec.edges) != 1 { |
| + t.Fatalf("edges = %d bodies, want the unparseable one delivered verbatim", len(rec.edges)) |
| + } |
| +} |
| + |
| +// Objects under the prefix that this deployment's edges loop did not write |
| +// are not read at all — the same key-shape filter internal/edge's homeMap |
| +// applies. A stray object must not be able to fail the pass forever. |
| +func TestSteeringPassIgnoresForeignKeysUnderThePrefix(t *testing.T) { |
| + quietLog(t) |
| + s := steeringTestStore(t, map[string][]byte{ |
| + "control/edges/dublin.json": []byte(`{"v":1,"box":"dublin"}`), |
| + "control/edges/README.txt": []byte("hello"), |
| + "control/edges/nested/inner.json": []byte(`{"v":1,"box":"nested"}`), |
| + }) |
| + rec := &convergeRecorder{} |
| + |
| + if err := steeringPass(context.Background(), s, okRoutes(), rec.converge); err != nil { |
| + t.Fatalf("steeringPass: %v", err) |
| + } |
| + if len(rec.edges) != 1 { |
| + t.Fatalf("edges = %d bodies, want only control/edges/<box>.json", len(rec.edges)) |
| + } |
| +} |
| + |
| +// ---- the duty: dormancy and the housekeeper gate --------------------------- |
| + |
| +// The dormancy proof: with no zone id the duty returns nil having called |
| +// NOTHING — the pass closure holds every read and every AWS call, so "the |
| +// pass never ran" is "zero reads, zero API calls" structurally. |
| +func TestSteeringDutyDormantDoesNoWork(t *testing.T) { |
| + quietLog(t) |
| + ran := 0 |
| + duty := steeringDuty("", func(context.Context) error { ran++; return nil }) |
| + |
| + if err := duty.run(context.Background()); err != nil { |
| + t.Fatalf("dormant duty returned %v, want nil", err) |
| + } |
| + if ran != 0 { |
| + t.Fatalf("dormant duty ran the pass %d times, want 0", ran) |
| + } |
| +} |
| + |
| +func TestSteeringDutyRunsThePassWhenZoneSet(t *testing.T) { |
| + quietLog(t) |
| + ran := 0 |
| + boom := errors.New("converge failed") |
| + duty := steeringDuty("Z0ABCDEF1234", func(context.Context) error { ran++; return boom }) |
| + |
| + if err := duty.run(context.Background()); !errors.Is(err, boom) { |
| + t.Fatalf("duty error = %v, want the pass's error propagated to the housekeeper log", err) |
| + } |
| + if ran != 1 { |
| + t.Fatalf("pass ran %d times, want 1", ran) |
| + } |
| +} |
| + |
| +// The housekeeper contract: one box converges the deployment's zone, not |
| +// every box. A non-holder does nothing and says nothing. |
| +func TestSteeringDutyRunsOnlyOnTheHousekeeper(t *testing.T) { |
| + quietLog(t) |
| + ran := 0 |
| + duty := steeringDuty("Z0ABCDEF1234", func(context.Context) error { ran++; return nil }) |
| + |
| + runHousekeeperDuties(noGate, duty) |
| + if ran != 0 { |
| + t.Fatalf("pass ran %d times on a non-holder, want 0", ran) |
| + } |
| + runHousekeeperDuties(heldGate, duty) |
| + if ran != 1 { |
| + t.Fatalf("pass ran %d times on the holder, want 1", ran) |
| + } |
| +} |
| + |
| +// ---- the loop: arming ------------------------------------------------------ |
| + |
| +// Mirrors TestMintMeshKeyLoopIsAPlatformBoxJob: a fleet box must skip before |
| +// ever opening a store. |
| +func TestSteeringLoopIsAPlatformBoxJob(t *testing.T) { |
| + const skipLine = "agent: steering loop is a platform-box job — skipped on a fleet box" |
| + |
| + logs := captureLog(t) |
| + door := newDoorFake(t) |
| + for k, v := range fleetDoorEnv(door.url) { |
| + t.Setenv(k, v) |
| + } |
| + t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_DIR", "") |
| + |
| + steeringLoop(nil) // must return; a fleet box never reaches the loop |
| + |
| + if got := logs(); !strings.Contains(got, skipLine) { |
| + t.Fatalf("log = %q, want the skip line verbatim", got) |
| + } |
| + if hits := door.seen(); len(hits) != 0 { |
| + t.Fatalf("door saw %+v, want zero — the skip must land BEFORE the store opens", hits) |
| + } |
| +} |
| + |
| +// Arming independence, the mesh-key loop's own regression guard applied here: |
| +// a box missing CARLOS_RELEASE_PUBKEY (retentionLoop's gate, unrelated to |
| +// steering) still WINS the housekeeper election, so a steering duty gated on |
| +// that var would silently never converge fleet-wide. |
| +func TestSteeringLoopArmsWithoutReleasePubkey(t *testing.T) { |
| + t.Setenv("CARLOS_FLEET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir()) |
| + t.Setenv("CARLOS_RELEASE_PUBKEY", "") // deliberately unset — the whole point |
| + t.Setenv("CARLOS_STEERING_ZONE_ID", "") |
| + |
| + buf := captureRacyLog(t) |
| + go steeringLoop(nil) |
| + |
| + waitFor(t, "the steering loop to arm despite no CARLOS_RELEASE_PUBKEY", func() bool { |
| + return strings.Contains(buf.String(), "agent: steering loop armed") |
| + }) |
| +} |
| + |
| +// Today's boxes: the env is unset, so the loop arms and says so ONCE, and its |
| +// duty is the dormant one — no store reads, no registry reads, no AWS client |
| +// ever constructed. |
| +func TestSteeringLoopDormantWhenZoneUnset(t *testing.T) { |
| + t.Setenv("CARLOS_FLEET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir()) |
| + t.Setenv("CARLOS_STEERING_ZONE_ID", "") |
| + |
| + buf := captureRacyLog(t) |
| + go steeringLoop(nil) |
| + |
| + waitFor(t, "the dormant line", func() bool { |
| + return strings.Contains(buf.String(), "steering: zone unset — dormant") |
| + }) |
| + if n := strings.Count(buf.String(), "steering: zone unset — dormant"); n != 1 { |
| + t.Fatalf("dormant line logged %d times, want exactly 1 (at arm)", n) |
| + } |
| +} |
| + |
| +// A malformed zone id is dormancy too, not a crash loop: SteeringZoneID logs |
| +// it loudly and returns "", and the loop arms dormant on that. |
| +func TestSteeringLoopDormantWhenZoneMalformed(t *testing.T) { |
| + t.Setenv("CARLOS_FLEET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "") |
| + t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir()) |
| + t.Setenv("CARLOS_STEERING_ZONE_ID", "/hostedzone/Z0ABCDEF1234") |
| + |
| + buf := captureRacyLog(t) |
| + go steeringLoop(nil) |
| + |
| + waitFor(t, "the dormant line for a malformed zone id", func() bool { |
| + return strings.Contains(buf.String(), "steering: zone unset — dormant") |
| + }) |
| + if !strings.Contains(buf.String(), "is not a bare hosted-zone id") { |
| + t.Fatalf("log = %q, want the malformed id named loudly", buf.String()) |
| + } |
| +} |
diff --git a/internal/steering/deployment.go b/internal/steering/deployment.go| new file mode 100644 |
| index 000000000..9a7a74198 |
| --- /dev/null |
| +++ b/internal/steering/deployment.go |
| @@ -0,0 +1,73 @@ |
| +package steering |
| + |
| +// deployment.go — this package's ONE exported converge entry point. |
| +// |
| +// Everything else here is unexported by design: desiredState is a pure fold |
| +// (steering.go) and Converge takes a set that only that fold can build |
| +// (route53.go), so a caller outside this package cannot assemble the argument |
| +// Converge needs. ConvergeDeployment is the seam that closes that gap for the |
| +// host housekeeper (internal/host/steering_loop.go) — it takes what a BOX has |
| +// (registry rows and raw edges-file bodies), folds, and converges, in one |
| +// call that cannot be told to converge a set the fold did not produce. |
| +// |
| +// Note on the client parameter's type. route53API stays unexported even |
| +// though this function is exported: every method in it is an exported SDK |
| +// method name, so any type implementing them satisfies it from any package — |
| +// Go checks method sets, not the interface's own visibility. The host caller |
| +// therefore needs nothing more than NewClient (route53.go), whose concrete |
| +// *route53.Client it can pass straight in, and no test outside this package |
| +// ever needs to fake AWS. Exporting the interface would add surface with no |
| +// consumer; the constructor is the seam, and it already exists. |
| + |
| +import ( |
| + "context" |
| + "log" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// ConvergeDeployment folds this deployment's routes and edges files into the |
| +// desired steering state and converges the zone to it. |
| +// |
| +// edgesFiles are the RAW bodies of control/edges/<box>.json, one per armed |
| +// box, in the caller's listing order (the host loader sorts by key, so the |
| +// order — and therefore the fold — is stable tick to tick). |
| +// |
| +// The two failure classes an edges file can have are handled in DIFFERENT |
| +// places on purpose, and the split is the durable half of Task 5's shrink- |
| +// guard finding: |
| +// |
| +// - a file that could not be READ (the listing failed, or a Get failed) is |
| +// a VISIBILITY problem: the deployment cannot see all of its edges, so a |
| +// fold built from what came back is SHORT, and a short fold deletes the |
| +// records of every box that went missing. That case must never reach this |
| +// function — the caller fails closed and returns the error instead (see |
| +// internal/host/steering_loop.go's loadEdgesFileBodies). |
| +// - a file that was read but does not PARSE is a CONTENT problem, and it is |
| +// handled HERE by skipping it and saying so. A single box publishing a |
| +// corrupt or future-shaped body must not wedge the whole deployment's |
| +// steering, and the skip is bounded: the box simply contributes no |
| +// records this pass, which is the same as being unarmed. |
| +// |
| +// The distinction is not cosmetic. Both end in "one fewer edge in the fold", |
| +// but only the read failure means the deployment does not KNOW what it has — |
| +// and Converge's own guards (the empty-set refusal and the proportional |
| +// shrink limit) are the last line, not the first. This is the first. |
| +func ConvergeDeployment(ctx context.Context, client route53API, zoneID string, routes []registry.Route, edgesFiles [][]byte, appsDomain string) error { |
| + edges := make([]edgesFileView, 0, len(edgesFiles)) |
| + for i, body := range edgesFiles { |
| + f, err := parseEdgesFileView(body) |
| + if err != nil { |
| + // The bodies arrive without their keys (the caller hands over |
| + // []byte by design, so this package never touches a store), so |
| + // the position is the handle an operator gets: bodies are passed |
| + // in the caller's sorted listing order, making "file 2 of 3" the |
| + // 2nd key under control/edges/. The length disambiguates an empty |
| + // object from a truncated one. |
| + log.Printf("steering: skipping edges file %d of %d (%d bytes): %v — that box contributes no records this pass", i+1, len(edgesFiles), len(body), err) |
| + continue |
| + } |
| + edges = append(edges, f) |
| + } |
| + return Converge(ctx, client, zoneID, desiredState(routes, edges, appsDomain)) |
| +} |
diff --git a/internal/steering/deployment_test.go b/internal/steering/deployment_test.go| new file mode 100644 |
| index 000000000..313a3fad4 |
| --- /dev/null |
| +++ b/internal/steering/deployment_test.go |
| @@ -0,0 +1,151 @@ |
| +package steering |
| + |
| +import ( |
| + "context" |
| + "encoding/json" |
| + "strings" |
| + "testing" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| +) |
| + |
| +// edgesBody renders one box's edges file the way internal/host's |
| +// edgesFilePass writes it (schema v1) — the bytes the housekeeper duty reads |
| +// out of control/edges/<box>.json and hands to ConvergeDeployment verbatim. |
| +func edgesBody(t *testing.T, box, eip, region string) []byte { |
| + t.Helper() |
| + body, err := json.Marshal(map[string]any{ |
| + "v": 1, |
| + "box": box, |
| + "eip": eip, |
| + "region": region, |
| + "generation": "2026-08-23T09:00:00Z", |
| + "hosts": map[string]any{"app.example.com": map[string]string{"role": "home"}}, |
| + }) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return body |
| +} |
| + |
| +// TestConvergeDeploymentFoldsAndConverges is the entry point's whole job: two |
| +// boxes' edges files plus one steered route become the four latency records |
| +// (2 names x 2 edges) the zone should hold. |
| +func TestConvergeDeploymentFoldsAndConverges(t *testing.T) { |
| + captureLog(t) |
| + f := newFake() |
| + |
| + err := ConvergeDeployment(context.Background(), f, testZone, |
| + []registry.Route{steeredRoute("app.example.com")}, |
| + [][]byte{ |
| + edgesBody(t, "dublin", "1.2.3.4", "eu-west-1"), |
| + edgesBody(t, "oregon", "5.6.7.8", "us-west-2"), |
| + }, |
| + "example.com") |
| + if err != nil { |
| + t.Fatalf("ConvergeDeployment: %v", err) |
| + } |
| + |
| + want := []string{ |
| + "app.example.com.|A|dublin", |
| + "app.example.com.|A|oregon", |
| + "edge.example.com.|A|dublin", |
| + "edge.example.com.|A|oregon", |
| + } |
| + got := recordNames(f.records) |
| + if len(got) != len(want) { |
| + t.Fatalf("records = %v, want %v", got, want) |
| + } |
| + for i := range want { |
| + if got[i] != want[i] { |
| + t.Fatalf("records = %v, want %v", got, want) |
| + } |
| + } |
| + if f.nCreateCheck != 2 { |
| + t.Errorf("health checks created = %d, want one per distinct EIP (2)", f.nCreateCheck) |
| + } |
| +} |
| + |
| +// TestConvergeDeploymentSkipsUnparseableEdgesFiles: a file that LISTS and |
| +// READS but does not PARSE is a content problem the fold tolerates — it is |
| +// skipped, named in the log, and the pass converges on the rest. (A file that |
| +// fails to READ is a different class entirely and never reaches here: the |
| +// duty fails closed before calling this function at all — see |
| +// internal/host/steering_loop.go's loadEdgesFileBodies.) |
| +func TestConvergeDeploymentSkipsUnparseableEdgesFiles(t *testing.T) { |
| + logs := captureLog(t) |
| + f := newFake() |
| + |
| + err := ConvergeDeployment(context.Background(), f, testZone, |
| + []registry.Route{steeredRoute("app.example.com")}, |
| + [][]byte{ |
| + []byte("{not json at all"), |
| + edgesBody(t, "dublin", "1.2.3.4", "eu-west-1"), |
| + []byte(`{"box":"ancient","eip":"9.9.9.9","region":"eu-west-1"}`), // no "v" — unversioned |
| + }, |
| + "example.com") |
| + if err != nil { |
| + t.Fatalf("ConvergeDeployment: %v", err) |
| + } |
| + |
| + // Only dublin folded in: 2 names x 1 edge. |
| + if got := recordNames(f.records); len(got) != 2 { |
| + t.Fatalf("records = %v, want the 2 dublin records only", got) |
| + } |
| + out := logs.String() |
| + if !strings.Contains(out, "skipping edges file") { |
| + t.Fatalf("log = %q, want a skip line naming the unparseable files", out) |
| + } |
| + if strings.Count(out, "skipping edges file") != 2 { |
| + t.Fatalf("log = %q, want exactly 2 skip lines (bad json + unversioned)", out) |
| + } |
| +} |
| + |
| +// TestConvergeDeploymentAllEdgesUnparseableRefusesTeardown is the composition |
| +// that matters most: every edges file unreadable-as-content folds to an EMPTY |
| +// desired set, and Converge's empty-set refusal must catch it. A pass that |
| +// obeyed it would delete the deployment's entire managed record set because |
| +// of a schema problem. |
| +func TestConvergeDeploymentAllEdgesUnparseableRefusesTeardown(t *testing.T) { |
| + logs := captureLog(t) |
| + f := newFake() |
| + // Seed the zone with a managed record so a teardown would be visible. |
| + mustConverge(t, f, []desiredRecord{dublinRec}) |
| + before := len(f.records) |
| + writesBefore := f.writes() |
| + |
| + err := ConvergeDeployment(context.Background(), f, testZone, |
| + []registry.Route{steeredRoute("app.example.com")}, |
| + [][]byte{[]byte("garbage"), []byte("{}")}, |
| + "example.com") |
| + if err != nil { |
| + t.Fatalf("ConvergeDeployment: %v", err) |
| + } |
| + if len(f.records) != before { |
| + t.Fatalf("records = %d, want %d — an empty fold must never tear the zone down", len(f.records), before) |
| + } |
| + if f.writes() != writesBefore { |
| + t.Fatalf("writes = %d, want %d (zero new writes)", f.writes(), writesBefore) |
| + } |
| + if !strings.Contains(logs.String(), "refusing to remove any record") { |
| + t.Fatalf("log = %q, want the empty-set refusal", logs.String()) |
| + } |
| +} |
| + |
| +// TestConvergeDeploymentRefusesBadZoneID: the zone-id gate is Converge's, and |
| +// the entry point must not paper over it — a malformed id returns an error |
| +// (which the duty logs) rather than converging against something else. |
| +func TestConvergeDeploymentRefusesBadZoneID(t *testing.T) { |
| + captureLog(t) |
| + f := newFake() |
| + err := ConvergeDeployment(context.Background(), f, "/hostedzone/Z0ABC", |
| + []registry.Route{steeredRoute("app.example.com")}, |
| + [][]byte{edgesBody(t, "dublin", "1.2.3.4", "eu-west-1")}, |
| + "example.com") |
| + if err == nil { |
| + t.Fatal("ConvergeDeployment accepted a non-bare hosted-zone id") |
| + } |
| + if f.writes() != 0 { |
| + t.Fatalf("writes = %d, want 0", f.writes()) |
| + } |
| +} |