amadan

carlos / platform Public

host: fleet boxes speak the store door; retention stays a platform job (spec F5/F6)

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 0121890713838cafbbadefac0612bc25382b4f53 parent c601012
3 files changed, +506 −25
  • internal/host/agent.go +143 −17
  • internal/host/agent_test.go +353 −0
  • internal/host/metrics_loop.go +10 −8
diff --git a/internal/host/agent.go b/internal/host/agent.go
index 11d17dc30..d22e0bdcf 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -68,6 +68,19 @@
// — an invalid result refuses to arm the wake-time/startup binding
// publish (fleetbind.go) loudly, and degrades the heartbeat's label back
// to the hashed pseudonym rather than disabling it outright.
+// - CARLOS_TUNNEL_URL / CARLOS_TUNNEL_TOKEN — the tunnel env
+// tunnel_client.go already dials with, read a SECOND time here (spec
+// 2026-08-11-fleet-api-surface, F5) as the fleet store door's address
+// and credential: on a fleet box with NO bucket env set, openDeployStore
+// and the status/metrics store both become internal/fleetapi clients
+// speaking release.Store over the edge's /store/ routes as
+// "<fleet>/<label>", so the box needs no AWS credentials on the
+// platform's deployment bucket at all. A bucket still set WINS
+// everywhere (the migration fallback): every platform box, and any
+// grandfathered fleet box that still carries bucket credentials, keeps
+// direct S3 byte for byte. Missing tunnel env on a fleet box with no
+// bucket is a loud error, not a silent degrade — that box has no way to
+// reach a store at all.
// - STATUS_BOX_REGION — an operator-set, coarse region word (e.g.
// "EU · Ireland") carried into status.Probes.Region and published
// verbatim on this box's snapshot; "" omits the field. Deliberately
@@ -127,6 +140,7 @@ import (
"github.com/carlosframework/platform/internal/cgroup"
"github.com/carlosframework/platform/internal/domains"
"github.com/carlosframework/platform/internal/edge"
+ "github.com/carlosframework/platform/internal/fleetapi"
"github.com/carlosframework/platform/internal/lease"
"github.com/carlosframework/platform/internal/provision"
"github.com/carlosframework/platform/internal/registry"
@@ -1455,6 +1469,19 @@ func instancesLoop(reg *registry.Registry) {
// box that cannot verify pointers must not delete anything.
func retentionLoop(reg *registry.Registry) {
ctx := context.Background()
+ // Platform boxes only (spec F6). Retention prunes the PLATFORM's release
+ // objects — a customer's box scheduling deletes of them was already wrong
+ // under direct S3, and through the store door it is refused outright (the
+ // door's write/delete ACL scopes a fleet to fleets/<fleet>/**), so arming
+ // here would buy nothing but an hourly loop of 404s. Reuses
+ // shouldSkipPublisherElection because the question is exactly "is this a
+ // fleet box", the same one instancesLoop asks — and it is deliberately the
+ // bare CARLOS_FLEET test, not resolveFleetIdentity: a fleet box whose
+ // LABEL is misconfigured is still a fleet box, and must still not prune.
+ if shouldSkipPublisherElection(os.Getenv) {
+ log.Printf("agent: retention loop is a platform-box job — skipped on a fleet box (spec F6)")
+ return
+ }
store, err := openDeployStore(ctx)
if err != nil {
log.Printf("agent: retention loop disabled: %v", err)
@@ -1549,11 +1576,12 @@ func shouldSkipPublisherElection(getenv func(string) string) bool {
// other tick failure.
func statusLoop(reg *registry.Registry) {
ctx := context.Background()
+ // bucket is read for the lease key alone (lease.Acquire below), which a
+ // fleet box never reaches: shouldSkipPublisherElection short-circuits the
+ // election before it, and a fleet box is the only shape that gets a store
+ // with no bucket behind it (openStatusStore). Every other box resolves a
+ // real bucket here exactly as before.
bucket := resolveStatusBucket(os.Getenv)
- if bucket == "" {
- log.Printf("agent: status loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set (CARLOS_DEPLOYMENT_DIR deployments don't publish status yet — set CARLOS_STATUS_BUCKET explicitly)")
- return
- }
prefix := os.Getenv("CARLOS_STATUS_PREFIX")
region := os.Getenv("CARLOS_STATUS_REGION")
boxRegion := os.Getenv("STATUS_BOX_REGION")
@@ -1579,9 +1607,13 @@ func statusLoop(reg *registry.Registry) {
log.Printf("agent: %v — heartbeat will publish under the hashed pseudonym instead", err)
}
- s, err := release.NewS3Store(ctx, bucket)
+ s, target, err := openStatusStore(ctx, os.Getenv)
if err != nil {
- log.Printf("agent: status loop disabled: %v", err)
+ if errors.Is(err, errNoStatusStore) {
+ log.Printf("agent: status loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set (CARLOS_DEPLOYMENT_DIR deployments don't publish status yet — set CARLOS_STATUS_BUCKET explicitly)")
+ } else {
+ log.Printf("agent: status loop disabled: %v", err)
+ }
return
}
// store is what statusTick actually reads/writes through: identical to s
@@ -1590,12 +1622,13 @@ func statusLoop(reg *registry.Registry) {
// release.NewPrefixStore and the package doc comment's
// CARLOS_STATUS_PREFIX entry.
store := release.NewPrefixStore(s, prefix)
- // The effective bucket is worth a log line on its own: it's easy to set
- // CARLOS_DEPLOYMENT_BUCKET and CARLOS_STATUS_BUCKET to different values
- // by accident (or on purpose) and end up with the agent writing status
- // to a bucket the edge/console never look at — this line is the one
- // place an operator can confirm which bucket actually won.
- log.Printf("agent: status loop armed (%s + jitter), publishing to bucket %s", status.SnapshotPeriod, bucket)
+ // The effective destination is worth a log line on its own: it's easy to
+ // set CARLOS_DEPLOYMENT_BUCKET and CARLOS_STATUS_BUCKET to different
+ // values by accident (or on purpose) and end up with the agent writing
+ // status to a bucket the edge/console never look at — this line is the
+ // one place an operator can confirm which bucket (or, on a bucketless
+ // fleet box, which door identity) actually won.
+ log.Printf("agent: status loop armed (%s + jitter), publishing to %s", status.SnapshotPeriod, target)
// canaryStore/canaryPub resolve the release deployment config
// countExpiredCanaries needs to answer §5's "how many canaries on this
@@ -1920,19 +1953,112 @@ func bounceFn(svc *activator.Service) func(host string) bool {
}
// openDeployStore opens the release store the box adopts from:
-// CARLOS_DEPLOYMENT_BUCKET (S3) or CARLOS_DEPLOYMENT_DIR (a directory —
-// tests and single-box no-AWS deployments). Mirrors cli.go's openStore, but
-// returns the error instead of dying: the agent degrades, the CLI exits.
+// CARLOS_DEPLOYMENT_BUCKET (S3), CARLOS_DEPLOYMENT_DIR (a directory — tests
+// and single-box no-AWS deployments), or — on a fleet box with neither set —
+// the edge's store door over the tunnel env (spec F5, openFleetDoorStore).
+// Mirrors cli.go's openStore, but returns the error instead of dying: the
+// agent degrades, the CLI exits.
func openDeployStore(ctx context.Context) (release.Store, error) {
- if b := os.Getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
+ return openDeployStoreEnv(ctx, os.Getenv)
+}
+
+// openDeployStoreEnv is openDeployStore with the environment injected — the
+// same getenv seam resolveStatusBucket already uses in this file, so the
+// selection rule can be tested without mutating process-wide env.
+//
+// Order is the whole contract (spec F5): a bucket or dir still set WINS,
+// before fleet identity is even consulted, so every platform box and every
+// grandfathered fleet box that still has bucket credentials keeps exactly
+// today's store, byte for byte. Only a box that resolves a fleet identity
+// AND has no bucket/dir at all reaches the door — that is the cutover
+// switch, per box, no flag day: drop the bucket var and the box is on the
+// API; put it back and it is off.
+func openDeployStoreEnv(ctx context.Context, getenv func(string) string) (release.Store, error) {
+ if b := getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
return release.NewS3Store(ctx, b)
}
- if d := os.Getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
+ if d := getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
return release.NewDirStore(d), nil
}
+ identity, err := resolveFleetIdentity(getenv)
+ switch {
+ case err == nil:
+ return openFleetDoorStore(getenv, identity)
+ case !errors.Is(err, errNotFleetBox):
+ // CARLOS_FLEET is set but the identity is unusable, and there is no
+ // bucket to fall back to: this box has no store at all. Return the
+ // identity error itself rather than the generic "no bucket" line —
+ // the actual fix is the label, and the generic message would send an
+ // operator hunting for a bucket var that is deliberately absent.
+ return nil, err
+ }
return nil, errors.New("neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set")
}
+// openFleetDoorStore builds the fleetapi client a bucketless fleet box adopts
+// through: the edge origin it already tunnels to (CARLOS_TUNNEL_URL), its
+// channel identity ("<fleet>/<label>" — the same string tunnel_client.go
+// registers with and edge.fleetChannelFor dials), and the tunnel bearer token.
+//
+// Missing tunnel env here is an ERROR, never a silent no-store: the caller
+// established that this box is a fleet box with no bucket, so there is no
+// other store it could possibly be meant to use, and a quiet nil would leave
+// every loop in this file reporting "not configured" about a box that is in
+// fact configured — just with one variable missing.
+func openFleetDoorStore(getenv func(string) string, identity FleetIdentity) (release.Store, error) {
+ base := strings.TrimSpace(getenv("CARLOS_TUNNEL_URL"))
+ token := strings.TrimSpace(getenv("CARLOS_TUNNEL_TOKEN"))
+ name := identity.Fleet + "/" + identity.Label
+ if base == "" || token == "" {
+ return nil, fmt.Errorf("fleet box %s has no deployment bucket and cannot reach the store door: both CARLOS_TUNNEL_URL and CARLOS_TUNNEL_TOKEN must be set", name)
+ }
+ return fleetapi.New(base, name, token)
+}
+
+// errNoStatusStore is openStatusStore's "this deployment simply hasn't
+// configured one" sentinel — distinct from a real construction failure, so
+// each loop can keep its own long-standing disabled-log wording for the
+// unconfigured case while sharing one selection rule.
+var errNoStatusStore = errors.New("no status store configured")
+
+// openStatusStore resolves the store statusLoop and metricsLoop publish
+// through, plus a human phrase naming where that is for their armed log
+// lines. Same F5 rule as openDeployStoreEnv, read through
+// resolveStatusBucket: an explicit CARLOS_STATUS_BUCKET (else the
+// CARLOS_DEPLOYMENT_BUCKET fallback) WINS and keeps direct S3; only a fleet
+// box with neither set speaks the door.
+//
+// LEDGER NOTE (spec F5): the door serves the DEPLOYMENT store. A fleet box
+// that sets CARLOS_STATUS_BUCKET to some OTHER bucket therefore keeps direct
+// S3 for status/metrics while its releases may come through the door — that
+// is the bucket-set-wins rule doing its job, not a bug: an operator who named
+// a bucket explicitly gets that bucket.
+//
+// The CARLOS_STATUS_PREFIX rebase both loops apply (release.NewPrefixStore)
+// is deliberately NOT applied here — it stays at each call site, unchanged,
+// and wraps whichever store this returns.
+func openStatusStore(ctx context.Context, getenv func(string) string) (release.Store, string, error) {
+ if bucket := resolveStatusBucket(getenv); bucket != "" {
+ s, err := release.NewS3Store(ctx, bucket)
+ if err != nil {
+ return nil, "", err
+ }
+ return s, "bucket " + bucket, nil
+ }
+ identity, err := resolveFleetIdentity(getenv)
+ switch {
+ case err == nil:
+ s, derr := openFleetDoorStore(getenv, identity)
+ if derr != nil {
+ return nil, "", derr
+ }
+ return s, "the fleet store door as " + identity.Fleet + "/" + identity.Label, nil
+ case !errors.Is(err, errNotFleetBox):
+ return nil, "", err
+ }
+ return nil, "", errNoStatusStore
+}
+
// RunAdopt is the `carlos adopt` verb: one immediate adoption pass — the
// carlos-adopt.timer unit's ExecStart, and an operator's "converge now".
// Runs as root (see carlos-adopt.service): this is the ONLY place the
diff --git a/internal/host/agent_test.go b/internal/host/agent_test.go
index 00917f7a4..33b06e9a9 100644
--- a/internal/host/agent_test.go
+++ b/internal/host/agent_test.go
@@ -7,12 +7,15 @@ import (
"encoding/base64"
"errors"
"log"
+ "net/http"
+ "net/http/httptest"
"os"
"path/filepath"
"runtime"
"slices"
"sort"
"strings"
+ "sync"
"testing"
"time"
@@ -1198,3 +1201,353 @@ func TestFastSweepBouncesAWarmHibernatingCanary(t *testing.T) {
}
}
}
+
+// ---- store selection: bucket, dir, or the fleet store door (spec F5) -------
+
+// doorHit is one request the fake door saw, reduced to the parts these tests
+// reason about: which call it was, and the two headers that prove WHICH box
+// made it (fleetapi sends the tunnel token and channel name on every request).
+type doorHit struct{ Method, Path, Auth, Name string }
+
+// doorFake stands in for the edge's store door. It answers the single GET
+// these tests drive and records every request, so a test can assert both
+// halves of the F5 rule: that a bucketless fleet box's store really does
+// speak to the door under its channel identity, and — the regression pin that
+// matters more — that a bucket-configured box never sends it anything.
+type doorFake struct {
+ url string
+ mu sync.Mutex
+ hits []doorHit
+}
+
+func newDoorFake(t *testing.T) *doorFake {
+ t.Helper()
+ d := &doorFake{}
+ srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+ d.mu.Lock()
+ d.hits = append(d.hits, doorHit{r.Method, r.URL.Path, r.Header.Get("Authorization"), r.Header.Get("Carlos-Tunnel-Name")})
+ d.mu.Unlock()
+ if r.Method == http.MethodGet && r.URL.Path == "/store/o/control/serial" {
+ w.Header().Set("ETag", `"e1"`)
+ _, _ = w.Write([]byte("42"))
+ return
+ }
+ http.NotFound(w, r)
+ }))
+ t.Cleanup(srv.Close)
+ d.url = srv.URL
+ return d
+}
+
+func (d *doorFake) seen() []doorHit {
+ d.mu.Lock()
+ defer d.mu.Unlock()
+ return append([]doorHit(nil), d.hits...)
+}
+
+// fleetDoorEnv is the env a fleet box that has been CUT OVER carries: fleet
+// identity plus tunnel address/credential, and deliberately no bucket and no
+// AWS credentials at all — the whole point of the door.
+func fleetDoorEnv(doorURL string) map[string]string {
+ return map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_FLEET_BOX": "pi-box",
+ "CARLOS_TUNNEL_URL": doorURL,
+ "CARLOS_TUNNEL_TOKEN": "tok-123",
+ }
+}
+
+// awsEnvForS3Store makes release.NewS3Store constructible in a test without
+// network or a developer's real profile — the same shape internal/release's
+// own S3 tests use. Nothing is ever called on the resulting client here; these
+// tests only assert which store TYPE the selection built.
+func awsEnvForS3Store(t *testing.T) {
+ t.Helper()
+ t.Setenv("AWS_ACCESS_KEY_ID", "k")
+ t.Setenv("AWS_SECRET_ACCESS_KEY", "s")
+ t.Setenv("AWS_SESSION_TOKEN", "")
+ t.Setenv("AWS_REGION", "eu-west-1")
+ t.Setenv("AWS_EC2_METADATA_DISABLED", "true")
+}
+
+// TestOpenDeployStoreSelection pins spec F5's whole rule for the store every
+// converge loop on the box reads through. The migration fallback is the
+// load-bearing half: a bucket or dir still set must win OUTRIGHT, before fleet
+// identity is even consulted, or rolling this branch out would silently move
+// every already-working fleet box onto a door it may not be able to reach.
+func TestOpenDeployStoreSelection(t *testing.T) {
+ ctx := context.Background()
+ dir := t.TempDir()
+
+ t.Run("fleet box with no bucket speaks the door", func(t *testing.T) {
+ door := newDoorFake(t)
+ s, err := openDeployStoreEnv(ctx, envFunc(fleetDoorEnv(door.url)))
+ if err != nil {
+ t.Fatalf("openDeployStoreEnv: %v", err)
+ }
+ body, etag, err := s.Get(ctx, "control/serial")
+ if err != nil {
+ t.Fatalf("Get through the door: %v", err)
+ }
+ if string(body) != "42" || etag != `"e1"` {
+ t.Fatalf("Get = %q, %q; want \"42\", \"e1\"", body, etag)
+ }
+ hits := door.seen()
+ if len(hits) != 1 {
+ t.Fatalf("door saw %d requests, want 1: %+v", len(hits), hits)
+ }
+ if hits[0].Path != "/store/o/control/serial" {
+ t.Fatalf("door path = %q, want /store/o/control/serial", hits[0].Path)
+ }
+ // The identity on the wire is the channel identity the edge dials —
+ // "<fleet>/<label>", never the hashed status pseudonym.
+ if hits[0].Name != "pi/pi-box" {
+ t.Fatalf("Carlos-Tunnel-Name = %q, want pi/pi-box", hits[0].Name)
+ }
+ if hits[0].Auth != "Bearer tok-123" {
+ t.Fatalf("Authorization = %q, want the tunnel token", hits[0].Auth)
+ }
+ })
+
+ t.Run("dir set wins on a fleet box, door untouched", func(t *testing.T) {
+ door := newDoorFake(t)
+ env := fleetDoorEnv(door.url)
+ env["CARLOS_DEPLOYMENT_DIR"] = dir
+ s, err := openDeployStoreEnv(ctx, envFunc(env))
+ if err != nil {
+ t.Fatalf("openDeployStoreEnv: %v", err)
+ }
+ if _, ok := s.(*release.DirStore); !ok {
+ t.Fatalf("store = %T, want *release.DirStore — a configured dir must win", s)
+ }
+ if _, _, err := s.Get(ctx, "control/serial"); !errors.Is(err, os.ErrNotExist) {
+ t.Fatalf("Get = %v, want the dir store's not-exist", err)
+ }
+ if hits := door.seen(); len(hits) != 0 {
+ t.Fatalf("door saw %+v, want zero requests — a dir-configured box must never reach it", hits)
+ }
+ })
+
+ t.Run("bucket set wins on a fleet box, door untouched", func(t *testing.T) {
+ awsEnvForS3Store(t)
+ door := newDoorFake(t)
+ env := fleetDoorEnv(door.url)
+ env["CARLOS_DEPLOYMENT_BUCKET"] = "carlos-deploy"
+ s, err := openDeployStoreEnv(ctx, envFunc(env))
+ if err != nil {
+ t.Fatalf("openDeployStoreEnv: %v", err)
+ }
+ if _, ok := s.(*release.S3Store); !ok {
+ t.Fatalf("store = %T, want *release.S3Store — a configured bucket must win", s)
+ }
+ if hits := door.seen(); len(hits) != 0 {
+ t.Fatalf("door saw %+v, want zero requests — a bucket-configured box must never reach it", hits)
+ }
+ })
+
+ t.Run("platform box is unchanged", func(t *testing.T) {
+ s, err := openDeployStoreEnv(ctx, envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": dir}))
+ if err != nil {
+ t.Fatalf("openDeployStoreEnv: %v", err)
+ }
+ if _, ok := s.(*release.DirStore); !ok {
+ t.Fatalf("store = %T, want *release.DirStore", s)
+ }
+ _, err = openDeployStoreEnv(ctx, envFunc(map[string]string{}))
+ if err == nil || !strings.Contains(err.Error(), "neither CARLOS_DEPLOYMENT_BUCKET nor CARLOS_DEPLOYMENT_DIR is set") {
+ t.Fatalf("unconfigured platform box err = %v, want today's message verbatim", err)
+ }
+ })
+
+ t.Run("fleet box with no bucket and no tunnel env fails loudly", func(t *testing.T) {
+ for _, missing := range []string{"CARLOS_TUNNEL_URL", "CARLOS_TUNNEL_TOKEN"} {
+ env := fleetDoorEnv("https://tunnel.example.com")
+ delete(env, missing)
+ _, err := openDeployStoreEnv(ctx, envFunc(env))
+ if err == nil {
+ t.Fatalf("missing %s: want an error, got a store", missing)
+ }
+ if !strings.Contains(err.Error(), "pi/pi-box") || !strings.Contains(err.Error(), missing) {
+ t.Fatalf("missing %s: err = %v, want it to name the box and the variable", missing, err)
+ }
+ // The token must never be quoted back into an error a log line or
+ // a status page could carry.
+ if strings.Contains(err.Error(), "tok-123") {
+ t.Fatalf("err leaks the tunnel token: %v", err)
+ }
+ }
+ })
+
+ t.Run("unusable fleet identity with no bucket names the identity", func(t *testing.T) {
+ _, err := openDeployStoreEnv(ctx, envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_TUNNEL_NAME": "legacy-box", // no "pi/" prefix: underivable label
+ "CARLOS_TUNNEL_URL": "https://tunnel.example.com",
+ "CARLOS_TUNNEL_TOKEN": "tok-123",
+ }))
+ if err == nil || !strings.Contains(err.Error(), "CARLOS_TUNNEL_NAME") {
+ t.Fatalf("err = %v, want the identity error, not the generic no-bucket line", err)
+ }
+ })
+}
+
+// TestOpenStatusStoreSelection pins the same rule for the status/metrics
+// store, including the ledger call it makes deliberately: the door serves the
+// DEPLOYMENT store, so a fleet box that names its own CARLOS_STATUS_BUCKET
+// keeps direct S3 for status — bucket-set-wins, on purpose.
+func TestOpenStatusStoreSelection(t *testing.T) {
+ ctx := context.Background()
+
+ t.Run("fleet box with no bucket speaks the door", func(t *testing.T) {
+ door := newDoorFake(t)
+ s, target, err := openStatusStore(ctx, envFunc(fleetDoorEnv(door.url)))
+ if err != nil {
+ t.Fatalf("openStatusStore: %v", err)
+ }
+ if _, _, err := s.Get(ctx, "control/serial"); err != nil {
+ t.Fatalf("Get through the door: %v", err)
+ }
+ if hits := door.seen(); len(hits) != 1 || hits[0].Name != "pi/pi-box" {
+ t.Fatalf("door hits = %+v, want one request as pi/pi-box", hits)
+ }
+ if !strings.Contains(target, "pi/pi-box") {
+ t.Fatalf("armed-log target = %q, want it to name the door identity", target)
+ }
+ })
+
+ t.Run("status bucket set on a fleet box keeps direct S3", func(t *testing.T) {
+ awsEnvForS3Store(t)
+ door := newDoorFake(t)
+ env := fleetDoorEnv(door.url)
+ env["CARLOS_STATUS_BUCKET"] = "carlos-status"
+ s, target, err := openStatusStore(ctx, envFunc(env))
+ if err != nil {
+ t.Fatalf("openStatusStore: %v", err)
+ }
+ if _, ok := s.(*release.S3Store); !ok {
+ t.Fatalf("store = %T, want *release.S3Store", s)
+ }
+ if target != "bucket carlos-status" {
+ t.Fatalf("target = %q, want %q", target, "bucket carlos-status")
+ }
+ if hits := door.seen(); len(hits) != 0 {
+ t.Fatalf("door saw %+v, want zero requests", hits)
+ }
+ })
+
+ t.Run("deployment bucket fallback still applies", func(t *testing.T) {
+ awsEnvForS3Store(t)
+ door := newDoorFake(t)
+ env := fleetDoorEnv(door.url)
+ env["CARLOS_DEPLOYMENT_BUCKET"] = "carlos-deploy"
+ s, target, err := openStatusStore(ctx, envFunc(env))
+ if err != nil {
+ t.Fatalf("openStatusStore: %v", err)
+ }
+ if _, ok := s.(*release.S3Store); !ok {
+ t.Fatalf("store = %T, want *release.S3Store", s)
+ }
+ if target != "bucket carlos-deploy" {
+ t.Fatalf("target = %q, want %q", target, "bucket carlos-deploy")
+ }
+ if hits := door.seen(); len(hits) != 0 {
+ t.Fatalf("door saw %+v, want zero requests", hits)
+ }
+ })
+
+ t.Run("platform box with no bucket stays unconfigured", func(t *testing.T) {
+ _, _, err := openStatusStore(ctx, envFunc(map[string]string{"CARLOS_DEPLOYMENT_DIR": t.TempDir()}))
+ if !errors.Is(err, errNoStatusStore) {
+ t.Fatalf("err = %v, want errNoStatusStore — a DIR deployment publishes no status", err)
+ }
+ })
+
+ t.Run("fleet box with no bucket and no tunnel env fails loudly", func(t *testing.T) {
+ env := fleetDoorEnv("https://tunnel.example.com")
+ delete(env, "CARLOS_TUNNEL_TOKEN")
+ _, _, err := openStatusStore(ctx, envFunc(env))
+ if err == nil || errors.Is(err, errNoStatusStore) {
+ t.Fatalf("err = %v, want a loud construction error, not the unconfigured sentinel", err)
+ }
+ })
+}
+
+// TestRetentionLoopIsAPlatformBoxJob pins spec F6. A customer's box pruning
+// the PLATFORM's releases was wrong under direct S3 and is refused outright
+// through the door (its write ACL scopes a fleet to fleets/<fleet>/**), so a
+// fleet box must not arm the loop at all — one log line, then out, before it
+// ever opens a store.
+func TestRetentionLoopIsAPlatformBoxJob(t *testing.T) {
+ const skipLine = "agent: retention loop is a platform-box job — skipped on a fleet box (spec F6)"
+
+ t.Run("fleet box skips before arming", func(t *testing.T) {
+ 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", "")
+
+ retentionLoop(nil) // must return; a fleet box never reaches the loop
+
+ if got := logs(); !strings.Contains(got, skipLine) {
+ t.Fatalf("log = %q, want the F6 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)
+ }
+ })
+
+ t.Run("platform box still arms", func(t *testing.T) {
+ pub, _, err := ed25519.GenerateKey(nil)
+ if err != nil {
+ t.Fatalf("generate key: %v", err)
+ }
+ t.Setenv("CARLOS_FLEET", "")
+ t.Setenv("CARLOS_DEPLOYMENT_BUCKET", "")
+ t.Setenv("CARLOS_DEPLOYMENT_DIR", t.TempDir())
+ t.Setenv("CARLOS_RELEASE_PUBKEY", base64.StdEncoding.EncodeToString(pub))
+
+ // retentionLoop never returns once armed, so the arming log line is
+ // what proves it: run it in its own goroutine (it parks in a
+ // time.Sleep for a minute, long past the end of this test) against a
+ // synchronized sink, and wait for the line to appear.
+ var mu sync.Mutex
+ var buf bytes.Buffer
+ old, flags := log.Writer(), log.Flags()
+ log.SetOutput(writerFunc(func(p []byte) (int, error) {
+ mu.Lock()
+ defer mu.Unlock()
+ return buf.Write(p)
+ }))
+ log.SetFlags(0)
+ t.Cleanup(func() { log.SetOutput(old); log.SetFlags(flags) })
+
+ go retentionLoop(nil)
+
+ deadline := time.Now().Add(5 * time.Second)
+ for {
+ mu.Lock()
+ got := buf.String()
+ mu.Unlock()
+ if strings.Contains(got, "agent: retention loop armed") {
+ if strings.Contains(got, skipLine) {
+ t.Fatalf("platform box took the fleet skip: %q", got)
+ }
+ return
+ }
+ if time.Now().After(deadline) {
+ t.Fatalf("retention loop never armed on a platform box; log = %q", got)
+ }
+ time.Sleep(10 * time.Millisecond)
+ }
+ })
+}
+
+// writerFunc adapts a func to io.Writer — the log sink above needs its own
+// mutex because the loop under test keeps running (and may keep logging)
+// after the test returns, which a bare bytes.Buffer would race on.
+type writerFunc func(p []byte) (int, error)
+
+func (f writerFunc) Write(p []byte) (int, error) { return f(p) }
diff --git a/internal/host/metrics_loop.go b/internal/host/metrics_loop.go
index 140b69dd8..9e0cb945a 100644
--- a/internal/host/metrics_loop.go
+++ b/internal/host/metrics_loop.go
@@ -740,14 +740,16 @@ func metricsLoop(execRoot string) {
}()
ctx := context.Background()
- bucket := resolveStatusBucket(os.Getenv)
- if bucket == "" {
- log.Printf("agent: metrics loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
- return
- }
- raw, err := release.NewS3Store(ctx, bucket)
+ // Same selection rule statusLoop uses, in one place for both (agent.go's
+ // openStatusStore): a status/deployment bucket wins and keeps direct S3;
+ // a fleet box with neither set publishes through the edge's store door.
+ raw, target, err := openStatusStore(ctx, os.Getenv)
if err != nil {
- log.Printf("agent: metrics loop disabled: %v", err)
+ if errors.Is(err, errNoStatusStore) {
+ log.Printf("agent: metrics loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set")
+ } else {
+ log.Printf("agent: metrics loop disabled: %v", err)
+ }
return
}
// s is what metricsFlush/alertTick actually read/write through: the same
@@ -769,7 +771,7 @@ func metricsLoop(execRoot string) {
logRejectedAlertOverrides(os.Getenv)
webhookURL := os.Getenv("CARLOS_ALERT_WEBHOOK")
- log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to bucket %s", bucket)
+ log.Printf("agent: metrics loop armed (60s ticks, 5min flush), publishing to %s", target)
sm := &sampler{}
probes := realSampleProbes(execRoot)