amadan

carlos / platform Public

fleet bindings: channel-label identity, non-stealing startup, fleet-scoped reconcile on instancesLoop

Fix round addressing Opus review. Controller ruling: a fleet box's identity
is the operator-chosen fleet_boxes.label (the channel identity the edge
dials), never status.Label(machineID) (the public status page's hashed
pseudonym) — resolveFleetIdentity (internal/host/fleetidentity.go) is the
one place CARLOS_FLEET/CARLOS_FLEET_BOX/CARLOS_TUNNEL_NAME are read and
validated, used by both the binding publish and the fleet-box heartbeat.

- WriteFleetBinding gained a takeover bool: a wake-success write may claim a
  binding from another box (lease-backed authority); the agent-startup pass
  may not (bindingBlocked). BoundAt also gates writes on its own (a stored
  binding with a greater BoundAt always wins), and is now unix milliseconds.
- ReconcileFleetBindings now requires registry.FleetPlacement(rt.Region) to
  match the binding's own Fleet — the v1 trust-boundary check that replaced
  an unsound cross-fleet "newest BoundAt wins" reduction.
- svc.OnWake is now wired before svc.Start() (was a data race).
- ReconcileFleetBindings moved from domainsLoop to instancesLoop, which has
  no PLATFORM_APPS_DOMAIN-shaped inert gate.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell fba2357401b7ed79d602c321353d9cc9e23aad77 parent fff285c
7 files changed, +750 −163
  • README.md +12 −0
  • internal/host/agent.go +114 −46
  • internal/host/fleetbind.go +164 −80
  • internal/host/fleetbind_test.go +257 −32
  • internal/host/fleetidentity.go +84 −0
  • internal/host/fleetidentity_test.go +96 −0
  • internal/host/host_test.go +23 −5
diff --git a/README.md b/README.md
index eaeea0d05..eaf0100d2 100644
--- a/README.md
+++ b/README.md
@@ -366,6 +366,18 @@ no single reader owns the whole contract:
the cross-box publish lease and never merges/publishes `status/fleet.json`
— that feed belongs to the platform's own boxes only. Unset (every
platform box today) is unaffected.
+- `CARLOS_FLEET_BOX` — this box's operator-chosen `fleet_boxes.label`
+ (customer-fleets plan, Task 6 fix round). This — NOT the public status
+ page's hashed machine-id pseudonym — is the box's real identity: the
+ channel name the edge dials is `<fleet>/<label>`, so both the fleet
+ binding a box publishes on wake and its own heartbeat filename
+ (`fleets/<fleet>/status/boxes/<label>.json`) use it. Optional: defaults to
+ `CARLOS_TUNNEL_NAME` with the `<fleet>/` prefix stripped, since a fleet
+ box's tunnel name already has this shape. Both the fleet name and the
+ resolved label are validated with `registry.ValidTunnelName`; a fleet box
+ whose identity fails to resolve refuses (loudly) to arm binding-publish at
+ all, and its heartbeat falls back to the hashed pseudonym instead of going
+ dark.
- `STATUS_BOX_REGION` — also read by `carlos agent`'s status loop: an
operator-set, coarse region word (e.g. `"EU · Ireland"`) carried onto
this box's snapshot verbatim; unset omits the field. Free text the
diff --git a/internal/host/agent.go b/internal/host/agent.go
index ce95c264e..327091236 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -58,6 +58,16 @@
// no-AWS deployment) has nothing for the fallback to resolve to, so
// DIR-based deployments do not publish fleet status yet; give them
// their own CARLOS_STATUS_BUCKET explicitly if they want the page.
+// - CARLOS_FLEET_BOX — this box's operator-chosen fleet_boxes.label (the
+// channel identity the edge dials, "<fleet>/<label>", NEVER the public
+// status page's hashed pseudonym). Optional: defaults to
+// CARLOS_TUNNEL_NAME with the "<fleet>/" prefix stripped, since a fleet
+// box's tunnel name already IS "<fleet>/<label>" (see
+// internal/host/tunnel_client.go). resolveFleetIdentity is the one
+// place both env vars are read and validated (registry.ValidTunnelName)
+// — 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.
// - 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
@@ -156,6 +166,34 @@ func RunAgent(args []string) {
if err != nil {
log.Fatalf("agent: %v", err)
}
+
+ // Fleet box identity (customer-fleets plan, Task 6 fix round — controller
+ // ruling): resolved ONCE, here, before the activator is even constructed.
+ // identErr is errNotFleetBox for the ordinary platform-box case
+ // (CARLOS_FLEET unset — no log, nothing to arm) or names the specific
+ // validation failure for a fleet box whose identity doesn't resolve to a
+ // usable channel label, in which case the fleet machinery — OnWake and
+ // the startup pass, both — stays disarmed (loud log) rather than run
+ // with a label the edge could never dial. fleetStore is opened here too
+ // (once) so it can be threaded into svc.OnWake BEFORE svc.Start() below —
+ // assigning it after Start() would race the sweeper goroutine's own use
+ // of the field.
+ var fleetStore release.Store
+ identity, identErr := resolveFleetIdentity(os.Getenv)
+ switch {
+ case errors.Is(identErr, errNotFleetBox):
+ // Ordinary platform box; nothing to arm.
+ case identErr != nil:
+ log.Printf("agent: %v — refusing to arm fleet machinery", identErr)
+ default:
+ fs, err := openDeployStore(context.Background())
+ if err != nil {
+ log.Printf("agent: CARLOS_FLEET=%s set but deploy store unavailable: %v — fleet bindings disabled", identity.Fleet, err)
+ } else {
+ fleetStore = fs
+ }
+ }
+
var act edge.Activator
var svc *activator.Service
if cfg != nil {
@@ -172,33 +210,32 @@ func RunAgent(args []string) {
},
Unit: &activator.UnitBackend{OptDir: release.OptDir()},
}, reg) // the registry satisfies activator.DueStore directly
+ if fleetStore != nil {
+ // MUST be assigned before Start() below spawns the sweeper
+ // goroutine, which reads svc.OnWake on every successful wake from
+ // then on — setting it any later is a data race.
+ svc.OnWake = fleetBindingHook(fleetStore, identity.Fleet, identity.Label)
+ }
svc.Start()
act = svc
log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
}
- // Fleet box binding publish (customer-fleets plan, Task 6): CARLOS_FLEET
- // set marks this agent as a fleet box (Task 5's marker, reused here — see
- // this file's package doc comment). Arms two publish paths against the
- // SAME store handle: the activator's OnWake hook (one publish per
- // successful wake) and a one-shot startup pass over every route already
- // resident in this box's local registry (covers an agent restart, and
- // routes that never go through activator.wake at all). Both write
- // through fleetbind.go's WriteFleetBinding, an UNPREFIXED store — see its
- // doc comment for why this deliberately does NOT go through the
- // CARLOS_STATUS_PREFIX-wrapped store statusLoop/metricsLoop use.
- if fleet := strings.TrimSpace(os.Getenv("CARLOS_FLEET")); fleet != "" {
- boxLabel := status.Label(readMachineID())
- if fleetStore, err := openDeployStore(context.Background()); err != nil {
- log.Printf("agent: CARLOS_FLEET=%s set but deploy store unavailable: %v — fleet bindings disabled", fleet, err)
- } else {
- if svc != nil {
- svc.OnWake = fleetBindingHook(fleetStore, fleet, boxLabel)
- } else {
- log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", fleet)
- }
- go publishResidentFleetBindings(fleetStore, fleet, boxLabel, reg)
+ // Fleet box binding publish, startup half (customer-fleets plan, Task 6):
+ // a one-shot pass over every route already resident in this box's local
+ // registry, covering an agent restart (activator.Service's in-memory
+ // entries are gone, but an already-running child isn't) and routes that
+ // never go through activator.wake at all. Non-authoritative
+ // (publishResidentFleetBindings calls WriteFleetBinding with
+ // takeover=false) — see fleetbind.go's doc comments for why a bare
+ // "this box has this route resident" pass must never steal a binding a
+ // wake-success write (lease-backed, authoritative) already holds for a
+ // DIFFERENT box.
+ if fleetStore != nil {
+ if svc == nil {
+ log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", identity.Fleet)
}
+ go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
}
go adoptLoop(reg)
@@ -917,14 +954,6 @@ func domainsLoop(reg *registry.Registry, acmeCacheFlag string) {
} else if cfg.ReleasePub == nil {
log.Printf("agent: domains loop auto-purge disabled: CARLOS_RELEASE_PUBKEY is not set — trashed apps will be hidden but never destroyed")
}
- // Platform-only: a fleet box (CARLOS_FLEET set) never reads other fleets'
- // bindings — it only ever WRITES its own (fleetbind.go's
- // fleetBindingHook/publishResidentFleetBindings, wired in RunAgent).
- // Reusing shouldSkipPublisherElection rather than re-testing the env var
- // here: both questions are exactly "is this box a fleet box", so one
- // predicate, already covered by TestShouldSkipPublisherElection, answers
- // both.
- isFleetBox := shouldSkipPublisherElection(os.Getenv)
delay := time.Minute
for {
time.Sleep(delay)
@@ -932,21 +961,6 @@ func domainsLoop(reg *registry.Registry, acmeCacheFlag string) {
if err := domains.Reconcile(ctx, store, reg, cfg); err != nil {
log.Printf("agent: domains pass: %v", err)
}
- // Fleet binding materialization (customer-fleets plan, Task 6): turn
- // every fleet's published fleets/*/bindings/*.json object into this
- // platform box's own registry.Route.FleetBox, on the SAME store and
- // cadence domainsLoop already has open — see ReconcileFleetBindings'
- // doc comment (fleetbind.go) for the corrupt-object/newest-wins
- // contract. Riding this loop rather than a new one is a real
- // coupling: it means fleet-binding reconciliation is inert whenever
- // PLATFORM_APPS_DOMAIN is unset (this loop's own inert gate, above),
- // even though bindings have nothing to do with custom domains. Flagged
- // for review rather than fixed unasked — see this task's report.
- if !isFleetBox {
- if err := ReconcileFleetBindings(ctx, store, reg); err != nil {
- log.Printf("agent: fleet bindings reconcile: %v", err)
- }
- }
}
}
@@ -962,6 +976,16 @@ func domainsLoop(reg *registry.Registry, acmeCacheFlag string) {
// one log line, the same "inert until configured" shape adoptLoop and
// domainsLoop use for their own missing config. Per-pass errors are logged;
// the loop lives on.
+//
+// Also runs ReconcileFleetBindings (customer-fleets plan, Task 6) on the same
+// cadence, platform boxes only: turns every fleet's published
+// fleets/*/bindings/*.json object into this platform box's own
+// registry.Route.FleetBox — see fleetbind.go's doc comment for the
+// fleet-placement/corrupt-object contract. Deliberately homed HERE rather
+// than domainsLoop (an earlier pass rode domainsLoop's ticker, since it
+// already had a store+reg in scope) — that coupled binding materialization
+// to PLATFORM_APPS_DOMAIN being set, which has nothing to do with fleets;
+// this loop has no such gate and shares the same store already opened above.
func instancesLoop(reg *registry.Registry) {
ctx := context.Background()
store, err := openDeployStore(ctx)
@@ -971,6 +995,14 @@ func instancesLoop(reg *registry.Registry) {
}
cfg := provisionTickConfig(os.Getenv)
log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)
+ // Platform-only: a fleet box (CARLOS_FLEET set) never reads other
+ // fleets' bindings — it only ever WRITES its own (fleetbind.go's
+ // fleetBindingHook/publishResidentFleetBindings, wired in RunAgent).
+ // Reusing shouldSkipPublisherElection rather than re-testing the env var
+ // here: both questions are exactly "is this box a fleet box", so one
+ // predicate, already covered by TestShouldSkipPublisherElection, answers
+ // both.
+ isFleetBox := shouldSkipPublisherElection(os.Getenv)
delay := time.Minute
for {
time.Sleep(delay)
@@ -978,6 +1010,11 @@ func instancesLoop(reg *registry.Registry) {
if err := provision.Reconcile(ctx, store, reg, cfg); err != nil {
log.Printf("agent: instances pass: %v", err)
}
+ if !isFleetBox {
+ if err := ReconcileFleetBindings(ctx, store, reg); err != nil {
+ log.Printf("agent: fleet bindings reconcile: %v", err)
+ }
+ }
}
}
@@ -1059,6 +1096,26 @@ func statusLoop(reg *registry.Registry) {
boxRegion := os.Getenv("STATUS_BOX_REGION")
skipElection := shouldSkipPublisherElection(os.Getenv)
+ // Fleet boxes heartbeat under the operator-chosen channel label
+ // (fleet_boxes.label — the identity the edge dials via
+ // edge.fleetChannelFor), NEVER the public status page's hashed
+ // pseudonym (status.Label(machineID)): that anonymization exists so a
+ // customer's machine id is never recoverable from the PUBLIC
+ // status/fleet.json feed, but a fleet box's own snapshot never reaches
+ // that feed at all (skipElection, above) — it only ever lands at
+ // fleets/<fleet>/status/boxes/<label>.json, an owner-facing path no
+ // public page reads, so the anonymization rationale simply doesn't
+ // apply here. An invalid identity degrades to the hashed label rather
+ // than disabling the heartbeat outright (RunAgent already refuses,
+ // loudly, to arm the binding-publish machinery that actually needs the
+ // label to be dial-able).
+ fleetLabel := ""
+ if identity, err := resolveFleetIdentity(os.Getenv); err == nil {
+ fleetLabel = identity.Label
+ } else if !errors.Is(err, errNotFleetBox) {
+ log.Printf("agent: %v — heartbeat will publish under the hashed pseudonym instead", err)
+ }
+
s, err := release.NewS3Store(ctx, bucket)
if err != nil {
log.Printf("agent: status loop disabled: %v", err)
@@ -1178,7 +1235,7 @@ func statusLoop(reg *registry.Registry) {
}
}
- if err := statusTick(ctx, store, reg, p, publish); err != nil {
+ if err := statusTick(ctx, store, reg, p, publish, fleetLabel); err != nil {
log.Printf("agent: status tick: %v", err)
}
@@ -1200,7 +1257,15 @@ func statusLoop(reg *registry.Registry) {
// only run when publish is true, which for a DirStore-backed dev/test caller
// with no lease in play at all just means "always true, no election needed"
// (a lone box is its own only publisher).
-func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p status.Probes, publish bool) error {
+//
+// labelOverride, when non-empty, replaces the box's hashed-pseudonym label
+// (status.CollectBox derives one from p.MachineID by default) with the given
+// value before WriteBox — statusLoop passes a fleet box's channel label
+// (resolveFleetIdentity) here so its heartbeat lands at
+// fleets/<fleet>/status/boxes/<label>.json under the SAME label the edge
+// dials, not an opaque pseudonym only the public status page ever wanted.
+// "" (every platform box) leaves CollectBox's own label untouched.
+func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p status.Probes, publish bool, labelOverride string) error {
routes, err := reg.List()
if err != nil {
return err
@@ -1208,6 +1273,9 @@ func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p
now := time.Now()
box := status.CollectBox(routes, p, now)
+ if labelOverride != "" {
+ box.Label = labelOverride
+ }
if err := status.WriteBox(ctx, s, box); err != nil {
return err
}
diff --git a/internal/host/fleetbind.go b/internal/host/fleetbind.go
index a5ba263b7..9c03b3f6c 100644
--- a/internal/host/fleetbind.go
+++ b/internal/host/fleetbind.go
@@ -14,24 +14,35 @@ import (
)
// FleetBinding is the deployment-bucket object a fleet box publishes to say
-// "as of BoundAt, Box (a status.Label) is the box currently serving Host".
-// Fleet boxes are outbound-only — they dial the platform edge over a
-// reverse tunnel and can never write the platform box's registry directly
+// "as of BoundAt, Box (the box's fleet_boxes.label — resolveFleetIdentity,
+// NEVER status.Label(machineID)) is the box currently serving Host". Fleet
+// boxes are outbound-only — they dial the platform edge over a reverse
+// tunnel and can never write the platform box's registry directly
// (customer-fleets design spec, "Routing to N boxes: the lease stamp") — so
// this bucket object is the ONLY channel a fleet box has to tell the
// platform which of its boxes currently holds a given hostname's lease.
// ReconcileFleetBindings is the platform-side reader that turns it back
// into registry.Route.FleetBox.
type FleetBinding struct {
- Host string `json:"host"`
- Fleet string `json:"fleet"`
- Box string `json:"box"` // status.Label of the holding box
- BoundAt int64 `json:"bound_at"` // unix seconds
+ Host string `json:"host"`
+ Fleet string `json:"fleet"`
+ Box string `json:"box"` // the holding box's fleet_boxes.label
+ // BoundAt is unix MILLISECONDS (not seconds) — a wire-shape choice made
+ // free at Task 6's fix round, before any binding object has ever shipped
+ // to production: two wakes racing on different boxes (or a wake racing
+ // the startup pass on the SAME box) can land inside the same second, and
+ // millisecond resolution is what lets bindingBlocked's logical-clock
+ // guard actually distinguish them instead of comparing equal.
+ BoundAt int64 `json:"bound_at"`
}
// fleetBindingKey is the one place the object's key is spelled, shared by
// the writer and the reader below: fleets/<fleet>/bindings/<host>.json,
-// ABSOLUTE in the deployment bucket.
+// ABSOLUTE in the PLATFORM'S deployment bucket — a fleet box has TWO buckets
+// (its own fleet/data bucket the activator leases and restores instances
+// against, and the platform's shared deployment bucket every box, fleet or
+// not, adopts releases from and now publishes/reads bindings in) and this
+// key lives in the latter, never the former.
//
// Deliberately not routed through release.NewPrefixStore's
// CARLOS_STATUS_PREFIX rebasing: that wrapper exists so a fleet box's own
@@ -48,34 +59,102 @@ func fleetBindingKey(fleet, host string) string {
return path.Join("fleets", fleet, "bindings", host+".json")
}
-// WriteFleetBinding publishes b to the deployment bucket, replacing whatever
-// is currently at its key (create-or-replace). s must be an UNPREFIXED
-// store handle — see fleetBindingKey's doc comment. The read-etag-then-Put
-// dance mirrors status.WriteBox exactly, and for the same reason: a box
-// republishes to the SAME key on every wake and once more at startup, so a
-// bare create-only Put (ifMatch "") would fail every write after the first
-// with ErrCASConflict.
-func WriteFleetBinding(ctx context.Context, s release.Store, b FleetBinding) error {
+// bindingBlocked reports whether writing b over the binding currently stored
+// at its key (existing, nil if none) must be skipped. Two independent
+// guards, either one enough to block:
+//
+// - Non-stealing: a non-takeover write (the startup pass,
+// publishResidentFleetBindings — a bare "this box has this route
+// resident" observation, not lease-backed) must never overwrite a
+// binding that already names a DIFFERENT box. Only a wake-success write
+// (takeover=true) may take over — it is backed by having just won the
+// per-host S3 lease, which is the actual arbiter of who may serve a
+// host, so it is allowed to restamp regardless of who was there before.
+// - Logical clock: a binding with a strictly greater BoundAt already
+// stored must never be overwritten by an older one — this holds
+// regardless of takeover, and fixes a physical-write-race hazard: two
+// concurrent Puts to the same key (a retried wake publish racing its own
+// predecessor, or any other overlap) must not let whichever one's HTTP
+// round-trip happens to land second win just because it arrived later.
+func bindingBlocked(existing *FleetBinding, b FleetBinding, takeover bool) bool {
+ if existing == nil {
+ return false
+ }
+ if !takeover && existing.Box != b.Box {
+ return true
+ }
+ return existing.BoundAt > b.BoundAt
+}
+
+// getExistingBinding reads key's current binding + etag for a replace-only
+// Put, folding "doesn't exist yet" into (nil, "", nil) — a create, with
+// neither guard in bindingBlocked applicable. A corrupt existing object is
+// treated as absent (logged) rather than failing the write outright: the
+// same "never fatal" spirit ReconcileFleetBindings' reader-side skip uses,
+// and it ensures a legitimate new write is never permanently stuck behind a
+// broken object at its own key.
+func getExistingBinding(ctx context.Context, s release.Store, key string) (*FleetBinding, string, error) {
+ body, etag, err := s.Get(ctx, key)
+ if err != nil {
+ if errors.Is(err, os.ErrNotExist) {
+ return nil, "", nil
+ }
+ return nil, "", err
+ }
+ var b FleetBinding
+ if err := json.Unmarshal(body, &b); err != nil {
+ log.Printf("host: fleet binding: existing object at %s is corrupt, treating as absent: %v", key, err)
+ return nil, etag, nil
+ }
+ return &b, etag, nil
+}
+
+// WriteFleetBinding publishes b to the PLATFORM'S deployment bucket,
+// replacing whatever is currently at its key — UNLESS bindingBlocked says
+// the existing binding must win (see its doc comment for the two guards).
+// s must be an UNPREFIXED store handle — see fleetBindingKey's doc comment.
+//
+// takeover distinguishes the two callers: fleetBindingHook (the activator's
+// OnWake, a lease-backed wake success) passes true and may claim a binding
+// away from any other box; publishResidentFleetBindings (the agent-startup
+// pass, a bare "this route is resident here" observation with no lease
+// behind it) passes false and must never steal a binding another box
+// currently holds.
+//
+// The read-etag-then-Put dance mirrors status.WriteBox, and for the same
+// reason: a box republishes to the SAME key on every wake and once more at
+// startup, so a bare create-only Put (ifMatch "") would fail every write
+// after the first with ErrCASConflict.
+func WriteFleetBinding(ctx context.Context, s release.Store, b FleetBinding, takeover bool) error {
body, err := json.Marshal(b)
if err != nil {
return err
}
key := fleetBindingKey(b.Fleet, b.Host)
- etag, err := currentBindingEtag(ctx, s, key)
+
+ existing, etag, err := getExistingBinding(ctx, s, key)
if err != nil {
return err
}
+ if bindingBlocked(existing, b, takeover) {
+ return nil
+ }
+
if _, err := s.Put(ctx, key, body, etag); err != nil {
if !errors.Is(err, release.ErrCASConflict) {
return err
}
- // Lost the race against a concurrent republish (another wake, or the
- // startup pass, writing the same key) — re-read once and retry, the
- // same single-retry contract status.WriteBox uses.
- etag, err = currentBindingEtag(ctx, s, key)
+ // Lost the race against a concurrent write to the same key —
+ // re-read once, re-apply BOTH guards against the fresher state (it
+ // may now block a write that was fine against the stale read), and
+ // retry — the same single-retry contract status.WriteBox uses.
+ existing, etag, err = getExistingBinding(ctx, s, key)
if err != nil {
return err
}
+ if bindingBlocked(existing, b, takeover) {
+ return nil
+ }
if _, err := s.Put(ctx, key, body, etag); err != nil {
return err
}
@@ -83,50 +162,43 @@ func WriteFleetBinding(ctx context.Context, s release.Store, b FleetBinding) err
return nil
}
-// currentBindingEtag reads key's current etag for a replace-only Put,
-// folding "doesn't exist yet" into "" (create-only) — the same convention
-// status.currentEtag and metrics' currentAlertEtag each carry locally rather
-// than sharing.
-func currentBindingEtag(ctx context.Context, s release.Store, key string) (string, error) {
- _, etag, err := s.Get(ctx, key)
- if err != nil {
- if errors.Is(err, os.ErrNotExist) {
- return "", nil
- }
- return "", err
- }
- return etag, nil
-}
-
-// ReconcileFleetBindings is the platform-side half of fleet routing: list
-// every fleet's fleets/<fleet>/bindings/*.json object, and for each host
-// whose newest binding names a fleet box THIS platform box doesn't already
-// have stamped, call registry.SetFleetBox to catch it up.
+// ReconcileFleetBindings is the platform-side half of fleet routing: for
+// every fleets/<fleet>/bindings/<host>.json object published under any
+// fleet, if the route registered for that host is actually PLACED on that
+// SAME fleet (registry.FleetPlacement(rt.Region) == the binding's own
+// Fleet) and its FleetBox doesn't already match, stamp it via SetFleetBox.
+//
+// The fleet-placement check is a v1 trust-boundary requirement, not a
+// nicety: fleet boxes are outbound-only, but nothing at the bucket layer
+// stops fleet A's box from publishing a binding that names a host actually
+// placed on fleet B (a same-named host collision, or a stale/rewritten
+// object) — without this check that binding could silently override fleet
+// B's own routing for a host fleet A's box has no business serving. There
+// is deliberately no "newest wins" tie-break for a cross-fleet collision:
+// two bindings for the exact same (fleet, host) pair cannot exist (the key
+// IS fleet+host), and a binding whose fleet doesn't match the route's own
+// placement is rejected outright, never preferred by timestamp — an earlier
+// version of this function used a cross-fleet "newest BoundAt" reduction
+// before this check existed; it's gone because the placement check makes it
+// both wrong (a malicious/stale binding could out-race a legitimate one by
+// BoundAt alone) and unnecessary (same-fleet duplicates are structurally
+// impossible).
//
// s must be an UNPREFIXED store handle (see fleetBindingKey) — this walks
// the same absolute "fleets/" keys WriteFleetBinding writes, and is only
// ever called on platform boxes, which never set CARLOS_STATUS_PREFIX in
// the first place.
//
-// Newest BoundAt wins when more than one fleet's binding names the same
-// host — a bare hostname collision across fleets should never happen in
-// practice, but nothing enforces it at the bucket layer, so ties are broken
-// deterministically rather than by map/listing order. A host with NO
-// binding object at all (a non-fleet route, or a fleet route this platform
-// box hasn't heard from yet) is left completely alone — this function only
-// ever calls SetFleetBox for a host it found a binding for.
-//
// Corrupt objects and per-key read failures are logged and skipped, never
// fatal — the loadBoxes precedent (internal/status/fleet.go:197-216): one
// bad binding must not blank out routing for every other fleet host. A
-// missing route (the binding names a host this platform box doesn't carry)
-// is silently skipped too — it isn't this platform box's route to stamp.
+// missing route, or one not placed on the binding's own fleet, is silently
+// skipped too — neither is this binding's to stamp.
func ReconcileFleetBindings(ctx context.Context, s release.Store, reg *registry.Registry) error {
fleetDirs, err := s.ListDirs(ctx, "fleets/")
if err != nil {
return err
}
- best := make(map[string]FleetBinding)
for _, dir := range fleetDirs {
keys, err := s.List(ctx, dir+"bindings/")
if err != nil {
@@ -144,42 +216,50 @@ func ReconcileFleetBindings(ctx context.Context, s release.Store, reg *registry.
log.Printf("host: fleet bindings: skipping %s: corrupt binding: %v", key, err)
continue
}
- if cur, ok := best[b.Host]; !ok || b.BoundAt > cur.BoundAt {
- best[b.Host] = b
+ rt, err := reg.Get(b.Host)
+ if err != nil {
+ log.Printf("host: fleet bindings: read route %s: %v", b.Host, err)
+ continue
+ }
+ if rt == nil {
+ continue // no matching route on this platform box
+ }
+ if fleet, ok := registry.FleetPlacement(rt.Region); !ok || fleet != b.Fleet {
+ // The route isn't placed on THIS binding's fleet at all —
+ // v1 trust boundary: refuse to let it stamp (see doc
+ // comment above).
+ continue
+ }
+ if rt.FleetBox == b.Box {
+ continue
+ }
+ if err := reg.SetFleetBox(b.Host, b.Box); err != nil {
+ log.Printf("host: fleet bindings: stamp %s -> %s: %v", b.Host, b.Box, err)
}
- }
- }
- for host, b := range best {
- rt, err := reg.Get(host)
- if err != nil {
- log.Printf("host: fleet bindings: read route %s: %v", host, err)
- continue
- }
- if rt == nil || rt.FleetBox == b.Box {
- continue
- }
- if err := reg.SetFleetBox(host, b.Box); err != nil {
- log.Printf("host: fleet bindings: stamp %s -> %s: %v", host, b.Box, err)
}
}
return nil
}
// fleetBindingHook builds the activator's OnWake hook for a fleet box: on
-// every successful wake, publish a FleetBinding naming this box as the
-// current holder. activator.Service already invokes OnWake in its own
-// goroutine (see activator.go's wake), so this function runs off the wake
-// path entirely — a slow or failing bucket Put can never delay or fail a
-// request. Errors are logged only: a binding is eventually-consistent by
-// design (the design spec's failure sequence — box goes dark, heartbeat
-// ages, lease expires, the next request wakes a surviving box, which
-// restamps — is exactly the fallback that covers a dropped write here).
+// every successful wake, publish a FleetBinding naming this box (identified
+// by fleet/box, resolveFleetIdentity's result — the channel label, never a
+// hashed pseudonym) as the current holder, with takeover=true since a
+// wake's lease acquisition is exactly the authority that may claim a
+// binding away from whoever held it before. activator.Service already
+// invokes OnWake in its own goroutine (see activator.go's wake), so this
+// function runs off the wake path entirely — a slow or failing bucket Put
+// can never delay or fail a request. Errors are logged only: a binding is
+// eventually-consistent by design (the design spec's failure sequence — box
+// goes dark, heartbeat ages, lease expires, the next request wakes a
+// surviving box, which restamps — is exactly the fallback that covers a
+// dropped write here).
func fleetBindingHook(store release.Store, fleet, box string) func(host string) {
return func(host string) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
- b := FleetBinding{Host: host, Fleet: fleet, Box: box, BoundAt: time.Now().Unix()}
- if err := WriteFleetBinding(ctx, store, b); err != nil {
+ b := FleetBinding{Host: host, Fleet: fleet, Box: box, BoundAt: time.Now().UnixMilli()}
+ if err := WriteFleetBinding(ctx, store, b, true); err != nil {
log.Printf("agent: fleet binding for %s: %v", host, err)
}
}
@@ -187,8 +267,12 @@ func fleetBindingHook(store release.Store, fleet, box string) func(host string)
// publishResidentFleetBindings runs once at fleet-box agent startup: every
// route already present in this box's local registry gets a binding
-// written immediately, without waiting for an activator wake. Two gaps this
-// closes that a wake-only publish would miss:
+// written immediately, without waiting for an activator wake, with
+// takeover=false — this is a bare "this route is resident here" observation
+// with no lease behind it, so it must never steal a binding a wake-success
+// write already holds for a DIFFERENT box (see WriteFleetBinding/
+// bindingBlocked). Two gaps this closes that a wake-only publish would
+// miss:
// - an agent restart discards activator.Service's in-memory entries, but
// an already-running exec child (or a unit-backed tenant) keeps serving
// — its host would otherwise have no fresh binding until its NEXT idle
@@ -209,10 +293,10 @@ func publishResidentFleetBindings(store release.Store, fleet, box string, reg *r
}
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
defer cancel()
- now := time.Now().Unix()
+ now := time.Now().UnixMilli()
for _, rt := range routes {
b := FleetBinding{Host: rt.Host, Fleet: fleet, Box: box, BoundAt: now}
- if err := WriteFleetBinding(ctx, store, b); err != nil {
+ if err := WriteFleetBinding(ctx, store, b, false); err != nil {
log.Printf("agent: fleet binding for %s: %v", rt.Host, err)
}
}
diff --git a/internal/host/fleetbind_test.go b/internal/host/fleetbind_test.go
index 69c6166e7..a8854b727 100644
--- a/internal/host/fleetbind_test.go
+++ b/internal/host/fleetbind_test.go
@@ -24,14 +24,57 @@ func testRegistry(t *testing.T) *registry.Registry {
return reg
}
+// --- Critical 1 (Opus review, fix round): channel-label identity -------------
+
+// TestPublishedBindingBoxMatchesTunnelChannelName is the reviewer-named test:
+// a published binding's Box, joined with its Fleet as "fleet/box", must
+// equal a channel name edge.fleetChannelFor's rule 1 would actually resolve
+// — that function's whole check is
+// `bound := fleet + "/" + boxLabel; hub.State(bound) == "connected"`, and a
+// box only ever shows up connected in the tunnel hub under the exact string
+// its own tunnel client presented as CARLOS_TUNNEL_NAME
+// (internal/host/tunnel_client.go's tunnelLoop dials with that env var
+// verbatim). This exercises the full pipeline this task wires —
+// resolveFleetIdentity -> fleetBindingHook -> WriteFleetBinding — and
+// asserts its published Box is dial-able, not just present.
+func TestPublishedBindingBoxMatchesTunnelChannelName(t *testing.T) {
+ const tunnelName = "pi/amber-gliding" // what this box's own tunnel client presents
+ id, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_TUNNEL_NAME": tunnelName,
+ }))
+ if err != nil {
+ t.Fatalf("resolveFleetIdentity: %v", err)
+ }
+
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ hook := fleetBindingHook(s, id.Fleet, id.Label)
+ hook("one.example.com")
+
+ body, _, err := s.Get(ctx, fleetBindingKey(id.Fleet, "one.example.com"))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var b FleetBinding
+ if err := json.Unmarshal(body, &b); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+
+ channel := b.Fleet + "/" + b.Box
+ if channel != tunnelName {
+ t.Fatalf("published binding's fleet/box = %q, want %q (this box's own CARLOS_TUNNEL_NAME) — a mismatch is a binding the edge could never dial", channel, tunnelName)
+ }
+}
+
// --- WriteFleetBinding --------------------------------------------------------
func TestWriteFleetBindingRoundTrips(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
- b := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1700000000}
+ b := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1700000000000}
- if err := WriteFleetBinding(ctx, s, b); err != nil {
+ if err := WriteFleetBinding(ctx, s, b, true); err != nil {
t.Fatalf("WriteFleetBinding: %v", err)
}
@@ -58,10 +101,10 @@ func TestWriteFleetBindingOverwritesExistingKey(t *testing.T) {
first := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 100}
second := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 200}
- if err := WriteFleetBinding(ctx, s, first); err != nil {
+ if err := WriteFleetBinding(ctx, s, first, true); err != nil {
t.Fatalf("first write: %v", err)
}
- if err := WriteFleetBinding(ctx, s, second); err != nil {
+ if err := WriteFleetBinding(ctx, s, second, true); err != nil {
t.Fatalf("second write (must replace, not conflict): %v", err)
}
@@ -78,18 +121,110 @@ func TestWriteFleetBindingOverwritesExistingKey(t *testing.T) {
}
}
+// Critical 2 (Opus review, fix round): a non-takeover write (the startup
+// pass's shape) must never steal a binding that already names a DIFFERENT
+// box — only a takeover=true (wake-success) write may do that.
+func TestWriteFleetBindingNonTakeoverNeverStealsAnotherBox(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ held := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "box-a", BoundAt: 100}
+ if err := WriteFleetBinding(ctx, s, held, true); err != nil {
+ t.Fatalf("seed box-a's binding: %v", err)
+ }
+
+ // box-b observes the SAME host resident locally (e.g. a stale local
+ // route not yet cleaned up) and tries a non-takeover (startup-pass)
+ // write with a LATER BoundAt — later in time must still lose, because
+ // takeover=false is a non-authoritative observation, not a lease.
+ attempt := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "box-b", BoundAt: 999}
+ if err := WriteFleetBinding(ctx, s, attempt, false); err != nil {
+ t.Fatalf("non-takeover write: %v", err)
+ }
+
+ body, _, err := s.Get(ctx, "fleets/pi/bindings/one.example.com.json")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got FleetBinding
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if got.Box != "box-a" {
+ t.Fatalf("Box = %q, want box-a (non-takeover write must not steal)", got.Box)
+ }
+}
+
+// A takeover=true write (a real wake success) MAY claim a binding away from
+// a different box — this is the counterpart proving the non-stealing rule
+// above is specific to non-takeover callers, not a blanket "first box wins".
+func TestWriteFleetBindingTakeoverClaimsFromAnotherBox(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ held := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "box-a", BoundAt: 100}
+ if err := WriteFleetBinding(ctx, s, held, true); err != nil {
+ t.Fatalf("seed box-a's binding: %v", err)
+ }
+
+ claim := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "box-b", BoundAt: 200}
+ if err := WriteFleetBinding(ctx, s, claim, true); err != nil {
+ t.Fatalf("takeover write: %v", err)
+ }
+
+ body, _, err := s.Get(ctx, "fleets/pi/bindings/one.example.com.json")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got FleetBinding
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if got.Box != "box-b" {
+ t.Fatalf("Box = %q, want box-b (a wake-success takeover must be able to claim)", got.Box)
+ }
+}
+
+// Important 5 (Opus review, fix round): a stored binding with a strictly
+// greater BoundAt must never be overwritten by an older one, even by the
+// SAME box, and even with takeover=true — a physical write race must not
+// let older information win just because its Put happens to land second.
+func TestWriteFleetBindingSkipsWhenStoredIsLogicallyNewer(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ newer := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 2000}
+ if err := WriteFleetBinding(ctx, s, newer, true); err != nil {
+ t.Fatalf("seed newer binding: %v", err)
+ }
+
+ older := FleetBinding{Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000}
+ if err := WriteFleetBinding(ctx, s, older, true); err != nil {
+ t.Fatalf("write older binding: %v", err)
+ }
+
+ body, _, err := s.Get(ctx, "fleets/pi/bindings/one.example.com.json")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got FleetBinding
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if got.BoundAt != 2000 {
+ t.Fatalf("BoundAt = %d, want 2000 (the older write must be skipped)", got.BoundAt)
+ }
+}
+
// --- ReconcileFleetBindings ---------------------------------------------------
func TestReconcileFleetBindingsStampsMatchingRoute(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
reg := testRegistry(t)
- if err := reg.Upsert(registry.Route{Host: "one.example.com", Kind: "instance"}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "one.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
t.Fatalf("upsert: %v", err)
}
if err := WriteFleetBinding(ctx, s, FleetBinding{
Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
- }); err != nil {
+ }, true); err != nil {
t.Fatalf("WriteFleetBinding: %v", err)
}
@@ -109,7 +244,7 @@ func TestReconcileFleetBindingsStampsMatchingRoute(t *testing.T) {
// A route with no published binding at all — a non-fleet route, or a fleet
// route this platform box hasn't heard from yet — must be left completely
// alone: ReconcileFleetBindings only ever calls SetFleetBox for a host it
-// found a binding for.
+// found a MATCHING binding for.
func TestReconcileFleetBindingsLeavesNonFleetRoutesAlone(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
@@ -131,28 +266,55 @@ func TestReconcileFleetBindingsLeavesNonFleetRoutesAlone(t *testing.T) {
}
}
-// Two different fleets both naming the same host (a bare hostname collision
-// the bucket layer itself never prevents) must resolve to the binding with
-// the greater BoundAt, regardless of listing/iteration order — writing the
-// OLDER one last (over a map, whose iteration order Go deliberately
-// randomizes) is what actually exercises the "compare, don't just take the
-// latest write" logic rather than an accidental last-one-wins.
-func TestReconcileFleetBindingsPrefersNewestBoundAt(t *testing.T) {
+// Important 4 (Opus review, fix round): a binding published under fleet
+// "eleven" must never stamp a route actually PLACED on fleet "pi" — even
+// with a far newer BoundAt than anything "pi" published. This is the v1
+// trust-boundary structural check that replaced an earlier "newest BoundAt
+// wins across fleets" reduction, which this scenario would have wrongly let
+// the mismatched, newer binding win.
+func TestReconcileFleetBindingsRejectsMismatchedFleetPlacement(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
reg := testRegistry(t)
- if err := reg.Upsert(registry.Route{Host: "shared.example.com", Kind: "instance"}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "shared.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
t.Fatalf("upsert: %v", err)
}
+ // A binding published under the WRONG fleet, naming the same host, with
+ // a much later BoundAt.
if err := WriteFleetBinding(ctx, s, FleetBinding{
- Host: "shared.example.com", Fleet: "pi", Box: "old-box", BoundAt: 500,
- }); err != nil {
- t.Fatalf("write pi binding: %v", err)
+ Host: "shared.example.com", Fleet: "eleven", Box: "eleven-box", BoundAt: 9_000_000_000_000,
+ }, true); err != nil {
+ t.Fatalf("write eleven binding: %v", err)
+ }
+
+ if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
+ t.Fatalf("ReconcileFleetBindings: %v", err)
+ }
+
+ rt, err := reg.Get("shared.example.com")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if rt.FleetBox != "" {
+ t.Fatalf("FleetBox = %q, want empty — a mismatched fleet's binding must never stamp this route", rt.FleetBox)
+ }
+}
+
+// The positive counterpart: a binding published under the route's OWN fleet
+// (Region "fleet/pi") DOES get applied, proving the rejection above is about
+// the mismatch specifically, not FleetPlacement/ReconcileFleetBindings being
+// broadly broken.
+func TestReconcileFleetBindingsAppliesMatchingFleetPlacement(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ reg := testRegistry(t)
+ if err := reg.Upsert(registry.Route{Host: "shared.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
+ t.Fatalf("upsert: %v", err)
}
if err := WriteFleetBinding(ctx, s, FleetBinding{
- Host: "shared.example.com", Fleet: "eleven", Box: "new-box", BoundAt: 900,
- }); err != nil {
- t.Fatalf("write eleven binding: %v", err)
+ Host: "shared.example.com", Fleet: "pi", Box: "pi-box", BoundAt: 1000,
+ }, true); err != nil {
+ t.Fatalf("write pi binding: %v", err)
}
if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
@@ -163,8 +325,8 @@ func TestReconcileFleetBindingsPrefersNewestBoundAt(t *testing.T) {
if err != nil {
t.Fatalf("Get: %v", err)
}
- if rt.FleetBox != "new-box" {
- t.Fatalf("FleetBox = %q, want new-box (BoundAt 900 beats 500)", rt.FleetBox)
+ if rt.FleetBox != "pi-box" {
+ t.Fatalf("FleetBox = %q, want pi-box", rt.FleetBox)
}
}
@@ -175,10 +337,10 @@ func TestReconcileFleetBindingsSkipsCorruptJSON(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
reg := testRegistry(t)
- if err := reg.Upsert(registry.Route{Host: "good.example.com", Kind: "instance"}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "good.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
t.Fatalf("upsert good: %v", err)
}
- if err := reg.Upsert(registry.Route{Host: "bad.example.com", Kind: "instance"}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "bad.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
t.Fatalf("upsert bad: %v", err)
}
if _, err := s.Put(ctx, "fleets/pi/bindings/bad.example.com.json", []byte("{ not json"), ""); err != nil {
@@ -186,7 +348,7 @@ func TestReconcileFleetBindingsSkipsCorruptJSON(t *testing.T) {
}
if err := WriteFleetBinding(ctx, s, FleetBinding{
Host: "good.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
- }); err != nil {
+ }, true); err != nil {
t.Fatalf("WriteFleetBinding: %v", err)
}
@@ -212,20 +374,17 @@ func TestReconcileFleetBindingsSkipsCorruptJSON(t *testing.T) {
// Already-stamped, matching FleetBox must not trigger a needless SetFleetBox
// call — proven indirectly by asserting the value survives a second
-// reconcile pass unchanged (a broken idempotence bug would show up as a
-// SetFleetBox call against a host with a matching value, which is harmless
-// here but would be loud with a fake registry; this pins the observable
-// behavior instead).
+// reconcile pass unchanged.
func TestReconcileFleetBindingsIdempotent(t *testing.T) {
ctx := context.Background()
s := release.NewDirStore(t.TempDir())
reg := testRegistry(t)
- if err := reg.Upsert(registry.Route{Host: "one.example.com", Kind: "instance"}); err != nil {
+ if err := reg.Upsert(registry.Route{Host: "one.example.com", Kind: "instance", Region: "fleet/pi"}); err != nil {
t.Fatalf("upsert: %v", err)
}
if err := WriteFleetBinding(ctx, s, FleetBinding{
Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
- }); err != nil {
+ }, true); err != nil {
t.Fatalf("WriteFleetBinding: %v", err)
}
for i := 0; i < 2; i++ {
@@ -287,6 +446,42 @@ func TestPublishResidentFleetBindingsEmptyRegistryIsNoop(t *testing.T) {
}
}
+// Critical 2 (Opus review, fix round): the missing test the reviewer named —
+// the startup pass must be non-stealing. A route resident locally on THIS
+// box (box-b) already has a binding published by a DIFFERENT box (box-a);
+// the startup pass must leave it alone rather than overwrite it, since it
+// has no lease behind it, unlike a wake-success write.
+func TestPublishResidentFleetBindingsDoesNotStealAnotherBoxsBinding(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ reg := testRegistry(t)
+ if err := reg.Upsert(registry.Route{Host: "one.example.com", Kind: "instance"}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := WriteFleetBinding(ctx, s, FleetBinding{
+ Host: "one.example.com", Fleet: "pi", Box: "box-a", BoundAt: 100,
+ }, true); err != nil {
+ t.Fatalf("seed box-a's binding: %v", err)
+ }
+
+ // box-b's startup pass encounters the same host resident in ITS OWN
+ // local registry (e.g. a route not yet cleaned up after a failover) and
+ // must not steal box-a's binding.
+ publishResidentFleetBindings(s, "pi", "box-b", reg)
+
+ body, _, err := s.Get(ctx, fleetBindingKey("pi", "one.example.com"))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got FleetBinding
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if got.Box != "box-a" {
+ t.Fatalf("Box = %q, want box-a (the startup pass must not steal)", got.Box)
+ }
+}
+
// --- fleetBindingHook (fire-and-forget wake publish) --------------------------
// errPutStore fails every Put while otherwise behaving like a normal
@@ -333,3 +528,33 @@ func TestFleetBindingHookWritesBinding(t *testing.T) {
t.Fatalf("binding = %+v, want host/fleet/box populated from the hook's closure", b)
}
}
+
+// fleetBindingHook must publish with takeover=true — a wake success is
+// lease-backed authority to claim a binding away from a different box.
+// Proven end-to-end here (rather than just trusting the source) since a
+// regression (accidentally passing false) would only show up as a silent
+// no-op against an existing binding, never a compile error.
+func TestFleetBindingHookTakesOverFromAnotherBox(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ if err := WriteFleetBinding(ctx, s, FleetBinding{
+ Host: "one.example.com", Fleet: "pi", Box: "old-box", BoundAt: 100,
+ }, true); err != nil {
+ t.Fatalf("seed old-box's binding: %v", err)
+ }
+
+ hook := fleetBindingHook(s, "pi", "new-box")
+ hook("one.example.com")
+
+ body, _, err := s.Get(ctx, fleetBindingKey("pi", "one.example.com"))
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ var got FleetBinding
+ if err := json.Unmarshal(body, &got); err != nil {
+ t.Fatalf("unmarshal: %v", err)
+ }
+ if got.Box != "new-box" {
+ t.Fatalf("Box = %q, want new-box (OnWake's hook must take over)", got.Box)
+ }
+}
diff --git a/internal/host/fleetidentity.go b/internal/host/fleetidentity.go
new file mode 100644
index 000000000..d121c8ce6
--- /dev/null
+++ b/internal/host/fleetidentity.go
@@ -0,0 +1,84 @@
+package host
+
+import (
+ "errors"
+ "fmt"
+ "strings"
+
+ "github.com/carlosframework/platform/internal/registry"
+)
+
+// FleetIdentity is a fleet box's resolved identity: which fleet it belongs
+// to, and which of that fleet's boxes IT is.
+//
+// Label is the operator-chosen fleet_boxes.label — the channel identity the
+// edge actually dials ("<Fleet>/<Label>", edge.fleetChannelFor) — and
+// deliberately NEVER status.Label(machineID). That hashed value is the
+// PUBLIC status page's opaque pseudonym, chosen specifically so a customer's
+// machine id can never be recovered from a feed anyone can view; a fleet
+// box's real identity is a name an operator picked when they ran
+// `carlos fleets add-box`, and nothing else on the platform (the edge's
+// tunnel hub, the registry's fleet_boxes table, a fleet binding object) ever
+// refers to a box any other way. Using the hashed pseudonym here would
+// publish bindings the edge could never dial back.
+type FleetIdentity struct {
+ Fleet string
+ Label string
+}
+
+// errNotFleetBox is resolveFleetIdentity's sentinel for "this agent isn't a
+// fleet box at all" (CARLOS_FLEET unset) — distinct from an invalid
+// identity, which callers must treat very differently: the former is every
+// platform box today (silent, expected), the latter is a fleet box
+// misconfigured badly enough that it must refuse to arm its fleet
+// machinery rather than run under a label nothing else in the system would
+// recognize.
+var errNotFleetBox = errors.New("host: CARLOS_FLEET is not set")
+
+// resolveFleetIdentity reads CARLOS_FLEET and CARLOS_FLEET_BOX (falling
+// back to CARLOS_TUNNEL_NAME, since a fleet box's tunnel name already has
+// the "<fleet>/<label>" shape tunnel_client.go dials with) and validates
+// both halves with registry.ValidTunnelName — the exact grammar
+// fleet_boxes.label, Route.Tunnel, and a fleet name in Route.Region
+// ("fleet/<name>", registry.FleetPlacement) all already share, so a value
+// this accepts is guaranteed usable everywhere else in the system that
+// stores or parses one.
+//
+// Returns (_, errNotFleetBox) when CARLOS_FLEET is unset — the ordinary,
+// silent, every-platform-box-today case; callers must not log anything for
+// it. Any OTHER non-nil error names a genuine misconfiguration (an invalid
+// fleet name, or a box label that isn't set and can't be derived/doesn't
+// validate) — callers must refuse to arm whatever fleet-box behavior they
+// gate on this, loudly, rather than fall back to guessing an identity.
+func resolveFleetIdentity(getenv func(string) string) (FleetIdentity, error) {
+ fleet := strings.TrimSpace(getenv("CARLOS_FLEET"))
+ if fleet == "" {
+ return FleetIdentity{}, errNotFleetBox
+ }
+ if !registry.ValidTunnelName(fleet) {
+ return FleetIdentity{}, fmt.Errorf("CARLOS_FLEET=%q is not a valid fleet name", fleet)
+ }
+
+ label := strings.TrimSpace(getenv("CARLOS_FLEET_BOX"))
+ if label == "" {
+ // A fleet box's CARLOS_TUNNEL_NAME is minted as "<fleet>/<label>"
+ // (plan doc: "FleetChannelAuth: channel name '<fleet>/<label>'"), so
+ // stripping this fleet's own prefix recovers the label without a
+ // second env var in the common case. TrimPrefix no-ops when
+ // CARLOS_TUNNEL_NAME is unset (label stays "" — caught below) or
+ // doesn't carry this exact prefix; ValidTunnelName below is the
+ // backstop, but note it cannot detect every malformed input this way
+ // (a CARLOS_TUNNEL_NAME from a WRONG fleet, or one with no "/" at
+ // all, can still leave behind a string that happens to validate) —
+ // CARLOS_FLEET_BOX is the reliable path for anything but the
+ // standard provisioning shape.
+ tunnelName := strings.TrimSpace(getenv("CARLOS_TUNNEL_NAME"))
+ label = strings.TrimPrefix(tunnelName, fleet+"/")
+ }
+ if label == "" || !registry.ValidTunnelName(label) {
+ return FleetIdentity{}, fmt.Errorf(
+ "fleet box label %q is invalid — set CARLOS_FLEET_BOX, or CARLOS_TUNNEL_NAME=%s/<label>",
+ label, fleet)
+ }
+ return FleetIdentity{Fleet: fleet, Label: label}, nil
+}
diff --git a/internal/host/fleetidentity_test.go b/internal/host/fleetidentity_test.go
new file mode 100644
index 000000000..f55ec6580
--- /dev/null
+++ b/internal/host/fleetidentity_test.go
@@ -0,0 +1,96 @@
+package host
+
+import (
+ "errors"
+ "testing"
+)
+
+func envFunc(m map[string]string) func(string) string {
+ return func(k string) string { return m[k] }
+}
+
+func TestResolveFleetIdentityNotAFleetBox(t *testing.T) {
+ _, err := resolveFleetIdentity(envFunc(map[string]string{}))
+ if !errors.Is(err, errNotFleetBox) {
+ t.Fatalf("err = %v, want errNotFleetBox", err)
+ }
+ // Explicitly empty is the same as unset.
+ _, err = resolveFleetIdentity(envFunc(map[string]string{"CARLOS_FLEET": ""}))
+ if !errors.Is(err, errNotFleetBox) {
+ t.Fatalf("err = %v, want errNotFleetBox for empty CARLOS_FLEET", err)
+ }
+}
+
+func TestResolveFleetIdentityExplicitBoxLabel(t *testing.T) {
+ id, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_FLEET_BOX": "amber-gliding",
+ }))
+ if err != nil {
+ t.Fatalf("resolveFleetIdentity: %v", err)
+ }
+ if id.Fleet != "pi" || id.Label != "amber-gliding" {
+ t.Fatalf("id = %+v, want {pi amber-gliding}", id)
+ }
+}
+
+func TestResolveFleetIdentityDefaultsLabelFromTunnelName(t *testing.T) {
+ id, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_TUNNEL_NAME": "pi/amber-gliding",
+ }))
+ if err != nil {
+ t.Fatalf("resolveFleetIdentity: %v", err)
+ }
+ if id.Fleet != "pi" || id.Label != "amber-gliding" {
+ t.Fatalf("id = %+v, want {pi amber-gliding}", id)
+ }
+}
+
+// An explicit CARLOS_FLEET_BOX must win over CARLOS_TUNNEL_NAME's derived
+// value — the explicit override is the reliable path the doc comment names
+// for anything the standard "<fleet>/<label>" tunnel-name shape can't cover.
+func TestResolveFleetIdentityExplicitBoxLabelWinsOverTunnelName(t *testing.T) {
+ id, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_FLEET_BOX": "explicit-label",
+ "CARLOS_TUNNEL_NAME": "pi/tunnel-derived-label",
+ }))
+ if err != nil {
+ t.Fatalf("resolveFleetIdentity: %v", err)
+ }
+ if id.Label != "explicit-label" {
+ t.Fatalf("Label = %q, want explicit-label (CARLOS_FLEET_BOX must win)", id.Label)
+ }
+}
+
+func TestResolveFleetIdentityInvalidFleetNameRefuses(t *testing.T) {
+ _, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "Not Valid!",
+ "CARLOS_FLEET_BOX": "amber-gliding",
+ }))
+ if err == nil || errors.Is(err, errNotFleetBox) {
+ t.Fatalf("err = %v, want a non-nil, non-errNotFleetBox validation error", err)
+ }
+}
+
+func TestResolveFleetIdentityMissingLabelRefuses(t *testing.T) {
+ // CARLOS_FLEET set, neither CARLOS_FLEET_BOX nor CARLOS_TUNNEL_NAME set
+ // at all — nothing to derive a label from.
+ _, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ }))
+ if err == nil || errors.Is(err, errNotFleetBox) {
+ t.Fatalf("err = %v, want a non-nil, non-errNotFleetBox validation error", err)
+ }
+}
+
+func TestResolveFleetIdentityInvalidLabelRefuses(t *testing.T) {
+ _, err := resolveFleetIdentity(envFunc(map[string]string{
+ "CARLOS_FLEET": "pi",
+ "CARLOS_FLEET_BOX": "not a valid label!",
+ }))
+ if err == nil || errors.Is(err, errNotFleetBox) {
+ t.Fatalf("err = %v, want a non-nil, non-errNotFleetBox validation error", err)
+ }
+}
diff --git a/internal/host/host_test.go b/internal/host/host_test.go
index a74e98e9f..21798c6f4 100644
--- a/internal/host/host_test.go
+++ b/internal/host/host_test.go
@@ -270,7 +270,7 @@ func TestStatusTickWritesBoxAndPublishesOnlyWhenAsked(t *testing.T) {
probes := status.Probes{MachineID: "test-machine"}
ctx := context.Background()
- if err := statusTick(ctx, store, reg, probes, false); err != nil {
+ if err := statusTick(ctx, store, reg, probes, false, ""); err != nil {
t.Fatalf("statusTick (publish=false): %v", err)
}
boxes, err := store.List(ctx, "status/boxes/")
@@ -284,7 +284,7 @@ func TestStatusTickWritesBoxAndPublishesOnlyWhenAsked(t *testing.T) {
t.Error("publish=false must not write status/fleet.json")
}
- if err := statusTick(ctx, store, reg, probes, true); err != nil {
+ if err := statusTick(ctx, store, reg, probes, true, ""); err != nil {
t.Fatalf("statusTick (publish=true): %v", err)
}
for _, key := range []string{"status/fleet.json", "status/index.html", "status/status.css"} {
@@ -343,6 +343,16 @@ func TestShouldSkipPublisherElection(t *testing.T) {
// snapshot at "fleets/pi/status/boxes/<label>.json" in the underlying
// bucket — never at the shared "status/boxes/<label>.json" a platform box
// uses, and never status/fleet.json either.
+//
+// Critical 1b (Opus review, fix round, controller ruling): <label> here MUST
+// be the fleet box's operator-chosen fleet_boxes.label — resolveFleetIdentity's
+// result, passed as statusTick's labelOverride — NEVER
+// status.Label(probes.MachineID)'s hashed pseudonym. The public status
+// page's anonymization rationale for that hash doesn't apply to a path
+// (fleets/<fleet>/status/boxes/…) no public page ever reads; using the
+// hashed value here would leave the heartbeat filename unrelated to the
+// name anything else (the edge's tunnel hub, a fleet binding) calls this
+// box by.
func TestFleetBoxHeartbeatLandsUnderFleetPrefixNeverPublishes(t *testing.T) {
reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
if err != nil {
@@ -355,14 +365,22 @@ func TestFleetBoxHeartbeatLandsUnderFleetPrefixNeverPublishes(t *testing.T) {
under := release.NewDirStore(t.TempDir())
fleetStore := release.NewPrefixStore(under, "fleets/pi/")
ctx := context.Background()
- probes := status.Probes{MachineID: "pi-box"}
+ // MachineID deliberately differs from the fleet-box label passed below —
+ // proving the filename tracks the identity override, not the probe.
+ probes := status.Probes{MachineID: "some-hashed-machine-id"}
// publish=false is what shouldSkipPublisherElection forces statusLoop
// into for a fleet box — lease.Acquire is never even called.
- if err := statusTick(ctx, fleetStore, reg, probes, false); err != nil {
+ const fleetBoxLabel = "amber-gliding"
+ if err := statusTick(ctx, fleetStore, reg, probes, false, fleetBoxLabel); err != nil {
t.Fatalf("statusTick: %v", err)
}
+ // The filename is pinned to the CHANNEL label, not a hash of MachineID.
+ if _, _, err := under.Get(ctx, "fleets/pi/status/boxes/"+fleetBoxLabel+".json"); err != nil {
+ t.Fatalf("Get fleets/pi/status/boxes/%s.json: %v", fleetBoxLabel, err)
+ }
+
boxes, err := under.List(ctx, "fleets/pi/status/boxes/")
if err != nil {
t.Fatal(err)
@@ -856,7 +874,7 @@ func TestStatusPublishNeverReadsMetricsPrefixes(t *testing.T) {
t.Fatal(err)
}
- if err := statusTick(ctx, store, reg, status.Probes{MachineID: "m"}, true); err != nil {
+ if err := statusTick(ctx, store, reg, status.Probes{MachineID: "m"}, true, ""); err != nil {
t.Fatal(err)
}
keys, err := store.List(ctx, "status/")