amadan
fleets: final-review wave — edge-route step restored, apex region, bounded loads, label identity
C1: restore the pre-fleets `carlos add <host>` edge-side step in infra/dev/region-pi.md and scripts/seed-pi-demo.sh — placement stamps Region but does not yet mint a platform-edge route, so the edge needs this per-hostname step until fleet-route materialization lands. I1: mintedRoute carries src.Region onto the platform apex, and drops Socket when the source is fleet-placed (a fleet box's socket path is meaningless off-box). I2: enableTunnelsFromEnv logs an explicit WARNING when the hub arms with zero static clients, so a typo'd CARLOS_TUNNEL_CLIENTS is distinguishable from a deliberate fleet-channels-only box. I3: status/fleet.go loadBoxes bounds an untrusted (customer-writable, in the fleet case) prefix: caps listed keys at 200, skips any object body over 1MB, logging both. I4: console/fleets.go handleAPIPlace's instance-record walk tolerates a record retired between List and its own write (os.ErrNotExist), logging and continuing instead of 502ing the whole placement. I5: metrics_loop.go publishes under the same resolved fleet-box identity statusLoop uses, not status.Label(machineID)'s hashed pseudonym, extracted as metricsBoxLabel for testability. M1/M2: routeCols doc comment corrected to 32 names; fleet_e2e_test.go's detach stay-down window widened 10x100ms -> 20x100ms.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
14 files changed,
+365
−14
infra/dev/region-pi.md+17 −1internal/console/fleets.go+11 −0internal/console/fleets_test.go+81 −0internal/domains/reconcile.go+15 −0internal/domains/reconcile_internal_test.go+45 −0internal/edge/edge.go+9 −0internal/edge/fleet_e2e_test.go+5 −2internal/edge/tunnel_test.go+15 −1internal/host/metrics_loop.go+34 −8internal/host/metrics_loop_test.go+49 −0internal/registry/registry.go+1 −1internal/status/fleet.go+24 −0internal/status/status_test.go+52 −0scripts/seed-pi-demo.sh+7 −1
diff --git a/infra/dev/region-pi.md b/infra/dev/region-pi.md| index be74ce399..07cad7a9c 100644 |
| --- a/infra/dev/region-pi.md |
| +++ b/infra/dev/region-pi.md |
| @@ -40,7 +40,23 @@ re-running an already-seeded batch is safe but not free, since the limiter |
| counts attempts, not successes, so only re-run a batch that actually |
| failed). Requires a carlos CLI build carrying PR #52's `apps create` verb — |
| seeding is blocked until #52 merges and a release ships. Apps are placed at |
| -create via `-place fleet/pi`. |
| +create via `-place fleet/pi`; records provision on the fleet's boxes |
| +automatically once the Pi is joined. |
| + |
| +Placement does NOT yet produce a route in the platform edge's own registry: |
| +the Pi adopts fleet-placed records into ITS OWN registry (that's what makes |
| +it serve them), but the platform edge routes from its own `reg.List()`, and |
| +nothing creates that row there — `apps place`/`apps create -place` stamps |
| +Region, it does not mint a route. Until fleet-route materialization lands |
| +(follow-up), the edge-side step from before fleets existed is still |
| +required: for each seeded hostname, on the dev edge box, run |
| + |
| + carlos add <hostname> |
| + |
| +with NO `-tunnel` flag — placement already supplies the channel once the |
| +route row exists and `apps place`/`apps create -place` has stamped Region, |
| +so passing `-tunnel` here would fight that. This is a one-time step per |
| +hostname; the dev edge's registry watcher picks up the new row within ~2s. |
| Target: ≥5 resident + ≥50 parked. |
diff --git a/internal/console/fleets.go b/internal/console/fleets.go| index 7d3a910fa..50d191d08 100644 |
| --- a/internal/console/fleets.go |
| +++ b/internal/console/fleets.go |
| @@ -30,6 +30,7 @@ import ( |
| "log" |
| "net/http" |
| "net/url" |
| + "os" |
| "regexp" |
| "sort" |
| "strings" |
| @@ -575,6 +576,16 @@ func (s *Server) handleAPIPlace(w http.ResponseWriter, r *http.Request) { |
| continue |
| } |
| if err := provision.SetRecordRegion(ctx, s.cfg.Releases, acct.Sqid, app, host, target); err != nil { |
| + if errors.Is(err, os.ErrNotExist) { |
| + // The record was retired (DeleteRecord) between the List |
| + // above and this write reaching it — not a failure of this |
| + // call, just a target that's gone. Every OTHER error still |
| + // fails the whole place: this is the one case where "the |
| + // thing we were about to touch already doesn't exist" is |
| + // indistinguishable from success. |
| + log.Printf("console: place %s/%s: instance %s vanished mid-walk, skipping", acct.Sqid, app, host) |
| + continue |
| + } |
| log.Printf("console: place %s/%s: stamp instance %s: %v", acct.Sqid, app, host, err) |
| jsonError(w, http.StatusBadGateway, "could not update every instance — try again") |
| return |
diff --git a/internal/console/fleets_test.go b/internal/console/fleets_test.go| index 6de0f03f8..860351ca6 100644 |
| --- a/internal/console/fleets_test.go |
| +++ b/internal/console/fleets_test.go |
| @@ -15,9 +15,11 @@ import ( |
| "context" |
| "crypto/sha256" |
| "encoding/hex" |
| + "errors" |
| "io" |
| "net/http" |
| "net/url" |
| + "os" |
| "strings" |
| "testing" |
| "time" |
| @@ -408,6 +410,85 @@ func TestPlaceStampsRoutesAndInstances(t *testing.T) { |
| } |
| } |
| +// vanishOnGetStore wraps a release.Store and deletes ONE key out from under |
| +// itself the first time that key is Get — a stand-in for I4's race: an |
| +// instance record retired (DeleteRecord) in the gap between handleAPIPlace's |
| +// List of a prefix and its per-key SetRecordRegion write reaching that same |
| +// key. Every other key, and every other method, is the embedded Store's real |
| +// behavior untouched. |
| +type vanishOnGetStore struct { |
| + release.Store |
| + vanish string |
| + deleted bool |
| +} |
| + |
| +func (v *vanishOnGetStore) Get(ctx context.Context, key string) ([]byte, string, error) { |
| + if key == v.vanish && !v.deleted { |
| + v.deleted = true |
| + if err := v.Store.Delete(ctx, key); err != nil { |
| + return nil, "", err |
| + } |
| + } |
| + return v.Store.Get(ctx, key) |
| +} |
| + |
| +// TestPlaceSkipsVanishedRecordMidWalk is I4 (final customer-fleets review): |
| +// handleAPIPlace's instance-record walk lists a prefix, then writes each key |
| +// it found. If one of those records is deleted in the gap between the list |
| +// and its own write — SetRecordRegion returns an os.ErrNotExist-wrapped |
| +// error — that must be a logged skip, not a 502 that fails the whole |
| +// placement (and not silently swallowing some OTHER error, which still must |
| +// fail the call). The surviving record must still get stamped. |
| +func TestPlaceSkipsVanishedRecordMidWalk(t *testing.T) { |
| + fx, token := fleetFixture(t, "code-place-vanish") |
| + now := time.Now() |
| + if _, err := fx.st.CreateAppIn(OpsAccountID, "titogo", "paul@keymail.dev", now); err != nil { |
| + t.Fatal(err) |
| + } |
| + reg := openTestReg(t, fx) |
| + for _, host := range []string{"acme.go.tito.io", "ghost.go.tito.io"} { |
| + if err := reg.Upsert(registry.Route{Host: host, Kind: "instance", App: "titogo", Account: fx.sqid}); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if status, body := postJSON(t, fleetsURL(fx, fx.sqid)+"/new", |
| + map[string]string{"name": "pi", "bucket": "pi-fleet-data"}, token); status != http.StatusOK { |
| + t.Fatalf("create fleet = %d %v", status, body) |
| + } |
| + |
| + ctx := context.Background() |
| + for _, host := range []string{"acme.go.tito.io", "ghost.go.tito.io"} { |
| + rec := provision.Record{Schema: 1, Host: host, Slug: strings.SplitN(host, ".", 2)[0], App: "titogo", |
| + Account: fx.sqid, Channel: "stable", CreatedAt: now.UTC()} |
| + if err := provision.CreateRecord(ctx, fx.rel, rec); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + |
| + // Swap the server's store for one that deletes ghost's record the |
| + // instant handleAPIPlace's walk tries to read it — after the List that |
| + // found it, exactly the race SetRecordRegion's os.ErrNotExist-wrapped |
| + // return exists to name. |
| + ghostKey := provision.RecordKey(fx.sqid, "titogo", "ghost.go.tito.io") |
| + fx.srv.cfg.Releases = &vanishOnGetStore{Store: fx.rel, vanish: ghostKey} |
| + |
| + status, body := postJSON(t, placeURL(fx, "titogo"), map[string]string{"target": "fleet/pi"}, token) |
| + if status != http.StatusOK { |
| + t.Fatalf("place = %d %v, want 200 (a vanished record mid-walk must not 502 the call)", status, body) |
| + } |
| + |
| + got, err := provision.GetRecord(ctx, fx.rel, fx.sqid, "titogo", "acme.go.tito.io") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if got.Region != "fleet/pi" { |
| + t.Fatalf("surviving record Region = %q, want fleet/pi — a vanished sibling must not stop the rest of the walk", got.Region) |
| + } |
| + if _, err := provision.GetRecord(ctx, fx.rel, fx.sqid, "titogo", "ghost.go.tito.io"); !errors.Is(err, os.ErrNotExist) { |
| + t.Fatalf("ghost record: %v, want it to stay gone", err) |
| + } |
| +} |
| + |
| // TestPlaceRefusesForeignFleet: placing onto a fleet owned by a DIFFERENT |
| // account is refused (the same byte-identical 404 apiFleet gives elsewhere), |
| // and nothing about the app's routes or instances is touched. |
diff --git a/internal/domains/reconcile.go b/internal/domains/reconcile.go| index d96c9a1ff..f879b461e 100644 |
| --- a/internal/domains/reconcile.go |
| +++ b/internal/domains/reconcile.go |
| @@ -655,6 +655,17 @@ func firstReadyUpstream(routes []registry.Route, k appKey) *registry.Route { |
| // this path (IMPORTANT 3), so a binding is no longer silently lost here — |
| // but a row this function rebuilds is still not one the route verb hands an |
| // operator, which is why the CLI refusal exists on top of the preservation. |
| +// |
| +// Region is copied from src (customer-fleets follow-up): a fleet-placed |
| +// source's minted apex must carry the same "fleet/<name>" Region so the |
| +// platform edge dispatches it through the fleet's channel rather than |
| +// treating it as unconstrained. When src is fleet-placed |
| +// (registry.FleetPlacement(src.Region)), Socket is deliberately left empty |
| +// even though it was just copied above the src.Kind == "static" carve-out |
| +// would otherwise leave it — a fleet-placed source's Socket is a path on |
| +// the FLEET BOX, not this platform process, so copying it verbatim onto the |
| +// minted apex would give it a dead local socket instead of letting it ride |
| +// the fleet dispatch Region now enables. |
| func mintedRoute(host, app, account string, src registry.Route) registry.Route { |
| rt := registry.Route{ |
| Host: host, |
| @@ -666,10 +677,14 @@ func mintedRoute(host, app, account string, src registry.Route) registry.Route { |
| Status: "ready", |
| Platform: true, |
| Account: account, |
| + Region: src.Region, |
| } |
| if src.Kind == "static" { |
| rt.Channel = src.Channel |
| } |
| + if _, ok := registry.FleetPlacement(src.Region); ok { |
| + rt.Socket = "" |
| + } |
| return rt |
| } |
diff --git a/internal/domains/reconcile_internal_test.go b/internal/domains/reconcile_internal_test.go| index 3b1db6def..a354f3d1c 100644 |
| --- a/internal/domains/reconcile_internal_test.go |
| +++ b/internal/domains/reconcile_internal_test.go |
| @@ -18,6 +18,8 @@ import ( |
| "path/filepath" |
| "testing" |
| "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| ) |
| // writeWildcardPEM writes a self-signed cert for name valid until notAfter, |
| @@ -112,3 +114,46 @@ func TestCertExpiryWildcardFallback(t *testing.T) { |
| t.Fatalf("hostCertExpiry with no files = %d, want 0", got) |
| } |
| } |
| + |
| +// TestMintedRouteCarriesRegionAndDropsSocketWhenFleetPlaced: a minted |
| +// platform apex must carry its source's Region so the edge dispatches it |
| +// through the same fleet channel — but when that Region is a fleet |
| +// placement, the source's Socket is a path on the FLEET BOX, not this |
| +// platform process, so it must NOT be copied onto the mint (I1, final |
| +// customer-fleets review). |
| +func TestMintedRouteCarriesRegionAndDropsSocketWhenFleetPlaced(t *testing.T) { |
| + src := registry.Route{ |
| + Kind: "instance", |
| + Socket: "/run/carlos/acme.sock", |
| + Region: registry.PlacementTarget("pi"), |
| + Status: "ready", |
| + App: "acme", |
| + Account: "abs", |
| + } |
| + rt := mintedRoute("acme.abs.oncarlos.com", "acme", "abs", src) |
| + if rt.Region != src.Region { |
| + t.Fatalf("mintedRoute Region = %q, want %q (fleet-placed source)", rt.Region, src.Region) |
| + } |
| + if rt.Socket != "" { |
| + t.Fatalf("mintedRoute Socket = %q, want empty — a fleet-placed source's socket lives on the fleet box, not here", rt.Socket) |
| + } |
| + |
| + // A plain (non-fleet) Region is still copied, and Socket is still copied |
| + // too, unchanged from today's behavior — only the fleet-placed case |
| + // clears it. |
| + src2 := registry.Route{ |
| + Kind: "instance", |
| + Socket: "/run/carlos/acme.sock", |
| + Region: "ie", |
| + Status: "ready", |
| + App: "acme", |
| + Account: "abs", |
| + } |
| + rt2 := mintedRoute("acme.abs.oncarlos.com", "acme", "abs", src2) |
| + if rt2.Region != "ie" { |
| + t.Fatalf("mintedRoute Region = %q, want %q (plain region)", rt2.Region, "ie") |
| + } |
| + if rt2.Socket != src2.Socket { |
| + t.Fatalf("mintedRoute Socket = %q, want %q — a non-fleet source's socket is still copied", rt2.Socket, src2.Socket) |
| + } |
| +} |
diff --git a/internal/edge/edge.go b/internal/edge/edge.go| index 599720865..abb2ad6fd 100644 |
| --- a/internal/edge/edge.go |
| +++ b/internal/edge/edge.go |
| @@ -1345,6 +1345,15 @@ func enableTunnelsFromEnv(e *Edge) error { |
| return fmt.Errorf("reload after EnableTunnels: %w", err) |
| } |
| log.Printf("edge: tunnel hub armed on %s (%d static client(s))", host, len(clients)) |
| + if len(clients) == 0 { |
| + // Zero static clients is legitimate (doc comment above), but it is |
| + // ALSO exactly what a typo'd CARLOS_TUNNEL_CLIENTS var name produces |
| + // (the env var silently reads empty rather than failing to parse) — |
| + // an operator staring at logs needs this distinguishable from "I |
| + // meant to run fleet-channels-only," not inferred from a client |
| + // count buried in the line above. |
| + log.Printf("edge: WARNING: no static tunnel clients configured — fleet channels only") |
| + } |
| return nil |
| } |
diff --git a/internal/edge/fleet_e2e_test.go b/internal/edge/fleet_e2e_test.go| index 343fa7fdc..b48fcaaec 100644 |
| --- a/internal/edge/fleet_e2e_test.go |
| +++ b/internal/edge/fleet_e2e_test.go |
| @@ -573,8 +573,11 @@ func TestTwoBoxFleetEndToEnd(t *testing.T) { |
| // And it STAYS down: its agent re-dials every few hundred ms and is |
| // refused every time, because a detached box is gone from the auth table |
| // entirely. A one-shot close that let the box straight back in would pass |
| - // the line above and fail here. |
| - for i := 0; i < 10; i++ { |
| + // the line above and fail here. 20x100ms (2s, M2 final-review widening |
| + // from 10x100ms/1s) gives a slow CI runner's re-dial loop more than one |
| + // chance to prove it stays refused, not just that it hasn't reconnected |
| + // yet. |
| + for i := 0; i < 20; i++ { |
| if got := hub.State(e2eFleet + "/box-b"); got != "" { |
| t.Fatalf("(d) detached box reconnected (state %q) — detach must revoke the credential, not just close the socket", got) |
| } |
diff --git a/internal/edge/tunnel_test.go b/internal/edge/tunnel_test.go| index fdc017273..4c48acfba 100644 |
| --- a/internal/edge/tunnel_test.go |
| +++ b/internal/edge/tunnel_test.go |
| @@ -1,11 +1,14 @@ |
| package edge |
| import ( |
| + "bytes" |
| "crypto/sha256" |
| "encoding/hex" |
| + "log" |
| "net" |
| "net/http" |
| "net/http/httptest" |
| + "os" |
| "strings" |
| "testing" |
| @@ -399,16 +402,27 @@ func TestEnableTunnelsFromEnv(t *testing.T) { |
| // A host with no STATIC clients is valid: fleet channels (SetDynamicAuth, |
| // wired by Reload from the registry) can populate the hub with no static |
| // env config at all — the hub still has to exist for those to land on. |
| - t.Run("host with no static clients is valid", func(t *testing.T) { |
| + // I2 (final customer-fleets review): zero static clients is legitimate |
| + // (fleet-channels-only), but it's also exactly what a typo'd |
| + // CARLOS_TUNNEL_CLIENTS var name silently produces — an explicit warning |
| + // must land in the log so the two are distinguishable there, not just |
| + // inferable from a client count. |
| + t.Run("host with no static clients is valid but warns", func(t *testing.T) { |
| t.Setenv("CARLOS_TUNNEL_HOST", "tunnel.test") |
| t.Setenv("CARLOS_TUNNEL_CLIENTS", "") |
| e := newEdge() |
| + var buf bytes.Buffer |
| + log.SetOutput(&buf) |
| + t.Cleanup(func() { log.SetOutput(os.Stderr) }) |
| if err := enableTunnelsFromEnv(e); err != nil { |
| t.Fatalf("CARLOS_TUNNEL_HOST with no static clients: %v", err) |
| } |
| if e.tunnels == nil { |
| t.Fatal("hub not armed with no static clients configured") |
| } |
| + if !strings.Contains(buf.String(), "WARNING: no static tunnel clients configured") { |
| + t.Fatalf("log = %q, want a WARNING about no static tunnel clients", buf.String()) |
| + } |
| }) |
| t.Run("invalid channel name", func(t *testing.T) { |
| t.Setenv("CARLOS_TUNNEL_HOST", "tunnel.test") |
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go| index 9a07cebae..18d677cf2 100644 |
| --- a/internal/host/metrics_loop.go |
| +++ b/internal/host/metrics_loop.go |
| @@ -690,15 +690,41 @@ func loadAlertState(path, box string) metrics.AlertState { |
| return st |
| } |
| +// metricsBoxLabel resolves the label metrics/alerts publish under — pulled |
| +// out of metricsLoop as its own pure seam (I5, final customer-fleets review) |
| +// so it is testable without a bucket, mirroring how statusTick takes an |
| +// explicit labelOverride rather than re-deriving it inline. |
| +// |
| +// It must be the SAME resolved identity statusLoop uses |
| +// (fleetidentity.go's FleetIdentity.Label, the operator-chosen |
| +// fleet_boxes.label the edge actually dials), NEVER status.Label(machineID)'s |
| +// hashed pseudonym, when this is a fleet box. Diverging from statusLoop here |
| +// would put this box's heartbeat under one label |
| +// (fleets/<fleet>/status/boxes/<label>.json) and its metrics under a |
| +// DIFFERENT one (status.Label's hash), which breaks the fleet-channel |
| +// identity a customer's monitoring page keys everything by — see |
| +// fleetidentity.go's doc comment for why the hashed pseudonym is wrong here |
| +// at all. An invalid (not merely absent) identity degrades to the hashed |
| +// label rather than disabling metrics outright, mirroring statusLoop's own |
| +// degrade-don't-disable choice. |
| +func metricsBoxLabel(getenv func(string) string, machineID string) string { |
| + if identity, err := resolveFleetIdentity(getenv); err == nil { |
| + return identity.Label |
| + } else if !errors.Is(err, errNotFleetBox) { |
| + log.Printf("agent: %v — metrics will publish under the hashed pseudonym instead", err) |
| + } |
| + return status.Label(machineID) |
| +} |
| + |
| // metricsLoop runs the metrics sampler's 60s tick forever, once armed — thin |
| // and untested by the same convention as statusLoop/adoptLoop: everything |
| -// with behavior lives in sampler.tick, metricsFlush, and alertTick, all |
| -// exercised above with fixtures. Missing status/deployment bucket config |
| -// disables the loop with one log line, the same "a box with no store is |
| -// still a valid deployment" shape every other optional loop in this package |
| -// uses. Flush runs every 5th tick (5 minutes at the 60s tick cadence); |
| -// alertTick runs every tick, since a threshold breach deserves faster |
| -// detection than a metrics flush does. |
| +// with behavior lives in sampler.tick, metricsFlush, alertTick, and |
| +// metricsBoxLabel, all exercised above with fixtures. Missing |
| +// status/deployment bucket config disables the loop with one log line, the |
| +// same "a box with no store is still a valid deployment" shape every other |
| +// optional loop in this package uses. Flush runs every 5th tick (5 minutes |
| +// at the 60s tick cadence); alertTick runs every tick, since a threshold |
| +// breach deserves faster detection than a metrics flush does. |
| func metricsLoop() { |
| // This loop is always on and shares a process with the agent that serves |
| // this box's traffic, so a panic in it — an unreadable cgroup shape, a |
| @@ -733,7 +759,7 @@ func metricsLoop() { |
| // comment's CARLOS_STATUS_PREFIX entry. |
| s := release.NewPrefixStore(raw, os.Getenv("CARLOS_STATUS_PREFIX")) |
| - box := status.Label(readMachineID()) |
| + box := metricsBoxLabel(os.Getenv, readMachineID()) |
| statePath := filepath.Join(dataDir(), "monitor", "alerts.json") |
| st := loadAlertState(statePath, box) |
| rules := metrics.RulesFromEnv(os.Getenv) |
diff --git a/internal/host/metrics_loop_test.go b/internal/host/metrics_loop_test.go| index c016b58d4..7ad672f3a 100644 |
| --- a/internal/host/metrics_loop_test.go |
| +++ b/internal/host/metrics_loop_test.go |
| @@ -15,6 +15,7 @@ import ( |
| "github.com/carlosframework/platform/internal/metrics" |
| "github.com/carlosframework/platform/internal/release" |
| + "github.com/carlosframework/platform/internal/status" |
| ) |
| // TestSamplerTickAndFlush is the brief's Step-1 test, verbatim: two ticks |
| @@ -869,3 +870,51 @@ func TestLogRejectedAlertOverrides(t *testing.T) { |
| t.Errorf("an accepted or unset override was logged: %q", out) |
| } |
| } |
| + |
| +// TestMetricsBoxLabelUsesFleetIdentity is I5 (final customer-fleets |
| +// review): metricsLoop's published label must be the SAME resolved |
| +// fleet-box identity statusLoop uses (fleetidentity.go's |
| +// FleetIdentity.Label), never status.Label(machineID)'s hashed pseudonym, |
| +// whenever CARLOS_FLEET is set — otherwise a fleet box's metrics/alerts |
| +// would land under a different key than its own heartbeat. |
| +func TestMetricsBoxLabelUsesFleetIdentity(t *testing.T) { |
| + const machineID = "test-machine-id" |
| + |
| + t.Run("no fleet env falls back to the hashed pseudonym", func(t *testing.T) { |
| + got := metricsBoxLabel(envFunc(map[string]string{}), machineID) |
| + if want := status.Label(machineID); got != want { |
| + t.Fatalf("box label = %q, want the hashed pseudonym %q", got, want) |
| + } |
| + }) |
| + |
| + t.Run("fleet env resolves to the fleet-box label", func(t *testing.T) { |
| + got := metricsBoxLabel(envFunc(map[string]string{ |
| + "CARLOS_FLEET": "pi", |
| + "CARLOS_FLEET_BOX": "pi-box", |
| + }), machineID) |
| + if got != "pi-box" { |
| + t.Fatalf("box label = %q, want %q (the fleet-box label, not a hash of %q)", got, "pi-box", machineID) |
| + } |
| + if got == status.Label(machineID) { |
| + t.Fatalf("box label collided with the hashed pseudonym — fleet identity was not used") |
| + } |
| + }) |
| + |
| + t.Run("invalid fleet identity degrades to the hashed pseudonym and warns", func(t *testing.T) { |
| + var buf bytes.Buffer |
| + prev := log.Writer() |
| + log.SetOutput(&buf) |
| + defer log.SetOutput(prev) |
| + |
| + got := metricsBoxLabel(envFunc(map[string]string{ |
| + "CARLOS_FLEET": "pi", |
| + // no CARLOS_FLEET_BOX, no CARLOS_TUNNEL_NAME to derive one from. |
| + }), machineID) |
| + if want := status.Label(machineID); got != want { |
| + t.Fatalf("box label = %q, want the hashed pseudonym %q on invalid identity", got, want) |
| + } |
| + if !strings.Contains(buf.String(), "hashed pseudonym") { |
| + t.Errorf("invalid fleet identity logged nothing: %q", buf.String()) |
| + } |
| + }) |
| +} |
diff --git a/internal/registry/registry.go b/internal/registry/registry.go| index 1c136de22..e5fb58955 100644 |
| --- a/internal/registry/registry.go |
| +++ b/internal/registry/registry.go |
| @@ -433,7 +433,7 @@ func (r *Registry) Accounts() ([]Account, error) { |
| } |
| // routeCols is the column list every route read and the one route write |
| -// share — 30 names, in this exact order. scanRoute's destinations, Upsert's |
| +// share — 32 names, in this exact order. scanRoute's destinations, Upsert's |
| // placeholder row and Upsert's argument list all count from here: adding a |
| // column means adding it in FOUR places, and a miscount is a silent column |
| // shift rather than an error (TestUpsertRoundTripsAllColumns is what catches |
diff --git a/internal/status/fleet.go b/internal/status/fleet.go| index eaf12a48a..0cd184e7a 100644 |
| --- a/internal/status/fleet.go |
| +++ b/internal/status/fleet.go |
| @@ -194,11 +194,31 @@ func BuildFleet(ctx context.Context, s release.Store, routes []registry.Route, n |
| // that fails to Get or fails to unmarshal is logged and skipped — one box's |
| // bad write must never take the whole fleet page down, per the plan's |
| // "corrupt snapshot = logged skip, never fatal" rule. |
| +// maxFleetBoxKeys bounds how many objects loadBoxes will ever process from |
| +// one prefix listing, and maxFleetBoxBodyBytes bounds how large any single |
| +// one of those objects may be before its body is read. Both exist for the |
| +// SAME reason: the prefix loadBoxes reads (status/boxes/, and — via |
| +// release.NewPrefixStore — a customer fleet's own fleets/<name>/status/boxes/) |
| +// is written to by boxes this operator does not control in the fleet case. |
| +// A hostile or badly-misbehaving box could park tens of thousands of keys, |
| +// or one huge body, under its own prefix; without a cap this function would |
| +// list/fetch/unmarshal all of it on every call, feeding both the operator |
| +// page (platform prefix) and every fleet's own page (customer prefix) — |
| +// unbounded memory from data this process does not trust. |
| +const ( |
| + maxFleetBoxKeys = 200 |
| + maxFleetBoxBodyBytes = 1 << 20 // 1MB |
| +) |
| + |
| func loadBoxes(ctx context.Context, s release.Store) ([]Box, error) { |
| keys, err := s.List(ctx, "status/boxes/") |
| if err != nil { |
| return nil, err |
| } |
| + if len(keys) > maxFleetBoxKeys { |
| + log.Printf("status: fleet: %d keys under status/boxes/, capping at %d", len(keys), maxFleetBoxKeys) |
| + keys = keys[:maxFleetBoxKeys] |
| + } |
| var boxes []Box |
| for _, key := range keys { |
| body, _, err := s.Get(ctx, key) |
| @@ -206,6 +226,10 @@ func loadBoxes(ctx context.Context, s release.Store) ([]Box, error) { |
| log.Printf("status: fleet: skipping %s: read failed: %v", key, err) |
| continue |
| } |
| + if len(body) > maxFleetBoxBodyBytes { |
| + log.Printf("status: fleet: skipping %s: body %d bytes exceeds %d byte cap", key, len(body), maxFleetBoxBodyBytes) |
| + continue |
| + } |
| var b Box |
| if err := json.Unmarshal(body, &b); err != nil { |
| log.Printf("status: fleet: skipping %s: corrupt snapshot: %v", key, err) |
diff --git a/internal/status/status_test.go b/internal/status/status_test.go| index 8c73ed3eb..fcd26f773 100644 |
| --- a/internal/status/status_test.go |
| +++ b/internal/status/status_test.go |
| @@ -5,6 +5,7 @@ import ( |
| "context" |
| "encoding/json" |
| "errors" |
| + "fmt" |
| "strings" |
| "testing" |
| "time" |
| @@ -504,6 +505,57 @@ func TestFleetSkipsCorruptSnapshot(t *testing.T) { |
| } |
| } |
| +// TestFleetLoadBoxesCapsKeyCount is I3 (final customer-fleets review): |
| +// status/boxes/ is customer-writable in the fleet case (via |
| +// CARLOS_STATUS_PREFIX), so loadBoxes must bound how many keys it will ever |
| +// process from one listing rather than trusting an untrusted prefix's size. |
| +// Seed one more box than the cap and confirm only the cap's worth land in |
| +// the merged Fleet. |
| +func TestFleetLoadBoxesCapsKeyCount(t *testing.T) { |
| + s := teststore(t) |
| + now := time.Now() |
| + const keyCap = 200 |
| + for i := 0; i < keyCap+1; i++ { |
| + label := fmt.Sprintf("box-%04d", i) |
| + if err := status.WriteBox(context.Background(), s, boxFixture(label, "plenty", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + f, err := status.BuildFleet(context.Background(), s, nil, now, 5*time.Minute) |
| + if err != nil { |
| + t.Fatalf("BuildFleet must not fail on an oversized key listing: %v", err) |
| + } |
| + if len(f.Boxes) != keyCap { |
| + t.Fatalf("boxes = %d, want exactly the %d-key cap (seeded %d)", len(f.Boxes), keyCap, keyCap+1) |
| + } |
| +} |
| + |
| +// TestFleetLoadBoxesSkipsOversizedBody is I3's other half: a single huge |
| +// object under status/boxes/ (a hostile or misbehaving box parking a large |
| +// body on its own customer-writable prefix) must be skipped, logged, and |
| +// must not prevent a sibling box's snapshot from loading. |
| +func TestFleetLoadBoxesSkipsOversizedBody(t *testing.T) { |
| + s := teststore(t) |
| + now := time.Now() |
| + if err := status.WriteBox(context.Background(), s, boxFixture("good", "plenty", now)); err != nil { |
| + t.Fatal(err) |
| + } |
| + huge := make([]byte, (1<<20)+1) // one byte over the 1MB cap |
| + for i := range huge { |
| + huge[i] = ' ' |
| + } |
| + if _, err := s.Put(context.Background(), "status/boxes/huge.json", huge, ""); err != nil { |
| + t.Fatal(err) |
| + } |
| + f, err := status.BuildFleet(context.Background(), s, nil, now, 5*time.Minute) |
| + if err != nil { |
| + t.Fatalf("BuildFleet must not fail on an oversized body: %v", err) |
| + } |
| + if len(f.Boxes) != 1 || f.Boxes[0].Label != "good" { |
| + t.Errorf("boxes = %+v, want only the good one (oversized body skipped)", f.Boxes) |
| + } |
| +} |
| + |
| // TestPublishFleetWritesAndOverwrites mirrors WriteBox's own overwrite test: |
| // PublishFleet lands status/fleet.json on every publish pass (every 5 |
| // minutes, per the plan), so a second call must overwrite cleanly rather |
diff --git a/scripts/seed-pi-demo.sh b/scripts/seed-pi-demo.sh| index f1374bcec..35336acd1 100755 |
| --- a/scripts/seed-pi-demo.sh |
| +++ b/scripts/seed-pi-demo.sh |
| @@ -43,5 +43,11 @@ for i in $(seq "$START" $((START + COUNT - 1))); do |
| done |
| echo |
| -echo "Apps are fleet-placed at create. Instances land on the fleet's boxes" |
| +echo "Apps are fleet-placed at create. Records provision on the fleet's boxes" |
| echo "automatically once the Pi is joined (runbook: infra/dev/region-pi.md)." |
| +echo |
| +echo "Placement does not yet mint a platform-edge route: on the dev edge box," |
| +echo "for each app above, run 'carlos add <hostname-of-app>' (NO -tunnel flag" |
| +echo "— placement already supplies the channel once the route row exists and" |
| +echo "this script's -place stamped Region) until fleet-route materialization" |
| +echo "lands (runbook: infra/dev/region-pi.md)." |