amadan

carlos / platform Public

fleet bindings: holding box publishes to bucket, platform agent stamps routes

Fleet boxes are outbound-only and can never write the platform's registry
directly, so a box's activator publishes a FleetBinding bucket object
(fleets/<fleet>/bindings/<host>.json) on every successful wake and once
more at startup for already-resident routes. The platform's adopt loop
reads every fleet's bindings back (newest BoundAt wins, corrupt objects
skipped) and stamps registry.Route.FleetBox to match.

- internal/host/fleetbind.go: FleetBinding, WriteFleetBinding,
  ReconcileFleetBindings, fleetBindingHook, publishResidentFleetBindings.
- internal/activator/activator.go: Service.OnWake hook, fired async off
  wake()'s success path so a slow/failing publish can never delay a
  request.
- internal/host/agent.go: wires OnWake + the startup pass when CARLOS_FLEET
  is set; domainsLoop calls ReconcileFleetBindings on platform boxes.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
Paul Campbell 363005dda9bba6c9e86f92fb67bd54a64e404b05 parent b768cbb
5 files changed, +676 −0
  • internal/activator/activator.go +22 −0
  • internal/activator/activator_test.go +53 −0
  • internal/host/agent.go +47 −0
  • internal/host/fleetbind.go +219 −0
  • internal/host/fleetbind_test.go +335 −0
diff --git a/internal/activator/activator.go b/internal/activator/activator.go
index 9014bfe37..cb29b9fe2 100644
--- a/internal/activator/activator.go
+++ b/internal/activator/activator.go
@@ -375,6 +375,19 @@ type Service struct {
sweepStop chan struct{}
sweepDone chan struct{}
+
+ // OnWake, when set, is called with the host after EVERY successful wake
+ // — fleet boxes use it to publish a binding object to the deployment
+ // bucket (customer-fleets plan, Task 6) naming this box as the current
+ // holder of host's lease; nil (every non-fleet deployment) is a pure
+ // no-op. Deliberately a callback rather than a fleet-aware field on
+ // Config: the activator has no notion of "fleet" anywhere else in this
+ // package, and threading FleetName/BoxLabel through Config would leak a
+ // customer-fleets concept into a package that otherwise only knows
+ // hosts, leases, and processes. wake() invokes it in its own goroutine
+ // (see below) so a slow or failing publish can never delay a request
+ // that's waiting on this same wake.
+ OnWake func(host string)
}
// New wires the state machine to its side-effect backend and due-time store.
@@ -592,6 +605,15 @@ func (a *Service) wake(e *actEntry) {
// cycle. Fresh wake, so it can't be stale yet.
e.spawnVersion = w.spawnVersion
e.staleSince = time.Time{}
+ if a.OnWake != nil {
+ // Fire-and-forget, off this goroutine: e.host is read once here,
+ // before this goroutine can be reused for anything else — the
+ // hook must never share e.host's own goroutine's fate (a wedged
+ // bucket Put must not hold e.ready closed any longer than the
+ // wake itself already takes).
+ host := e.host
+ go a.OnWake(host)
+ }
}
a.mu.Unlock()
close(e.ready)
diff --git a/internal/activator/activator_test.go b/internal/activator/activator_test.go
index d14d19014..90f7818e0 100644
--- a/internal/activator/activator_test.go
+++ b/internal/activator/activator_test.go
@@ -585,6 +585,59 @@ func TestActivatorWakeCoalesces(t *testing.T) {
}
}
+// OnWake (customer-fleets plan, Task 6: a fleet box's binding publish) must
+// fire after a successful wake, named with the right host, and — the whole
+// point of running it via `go a.OnWake(host)` rather than calling it inline —
+// must NOT be able to hold the wake open: beginRequest returns and releases
+// long before the hook is allowed to finish.
+func TestActivatorWakeCallsOnWakeHookAsync(t *testing.T) {
+ a, _, _, _ := testActivator(t, time.Minute)
+ block := make(chan struct{})
+ hookHost := make(chan string, 1)
+ a.OnWake = func(host string) {
+ <-block // would hang beginRequest forever if invoked synchronously
+ hookHost <- host
+ }
+
+ done := make(chan struct{})
+ go func() {
+ release, err := a.beginRequest(context.Background(), testRoute("hook.example.com"))
+ if err != nil {
+ t.Errorf("beginRequest: %v", err)
+ return
+ }
+ release()
+ close(done)
+ }()
+
+ select {
+ case <-done:
+ case <-time.After(2 * time.Second):
+ t.Fatal("beginRequest never returned — OnWake appears to be blocking the wake path")
+ }
+
+ close(block)
+ select {
+ case host := <-hookHost:
+ if host != "hook.example.com" {
+ t.Fatalf("OnWake called with %q, want hook.example.com", host)
+ }
+ case <-time.After(2 * time.Second):
+ t.Fatal("OnWake was never called after a successful wake")
+ }
+}
+
+// A nil OnWake (every non-fleet deployment, and every test above this one)
+// must be a pure no-op — no crash, wake behaves exactly as before Task 6.
+func TestActivatorNilOnWakeIsNoop(t *testing.T) {
+ a, _, _, _ := testActivator(t, time.Minute)
+ release, err := a.beginRequest(context.Background(), testRoute("nohook.example.com"))
+ if err != nil {
+ t.Fatalf("beginRequest: %v", err)
+ }
+ release()
+}
+
// The wake threads the route's replication class into Replicate() as the
// litestream sync interval — the per-plan durability ladder's seam. Unset
// (every pre-ladder route) must thread 0, i.e. litestream's default: a deploy
diff --git a/internal/host/agent.go b/internal/host/agent.go
index d71df4650..ce95c264e 100644
--- a/internal/host/agent.go
+++ b/internal/host/agent.go
@@ -177,6 +177,30 @@ func RunAgent(args []string) {
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)
+ }
+ }
+
go adoptLoop(reg)
go canaryAdoptLoop(reg)
go statusLoop(reg)
@@ -893,6 +917,14 @@ 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)
@@ -900,6 +932,21 @@ 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)
+ }
+ }
}
}
diff --git a/internal/host/fleetbind.go b/internal/host/fleetbind.go
new file mode 100644
index 000000000..a5ba263b7
--- /dev/null
+++ b/internal/host/fleetbind.go
@@ -0,0 +1,219 @@
+package host
+
+import (
+ "context"
+ "encoding/json"
+ "errors"
+ "log"
+ "os"
+ "path"
+ "time"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+// 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
+// (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
+}
+
+// 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.
+//
+// Deliberately not routed through release.NewPrefixStore's
+// CARLOS_STATUS_PREFIX rebasing: that wrapper exists so a fleet box's own
+// status/metrics/alerts heartbeats land under its private
+// "fleets/<name>/" corner instead of the platform's shared paths — but a
+// binding is not a heartbeat, it is a cross-box MESSAGE the platform box
+// must read back, and it already carries its own fleet name (b.Fleet) in
+// the struct. Both WriteFleetBinding and ReconcileFleetBindings take an
+// UNPREFIXED release.Store and compute this absolute key themselves, so
+// there is exactly one key format to keep in sync, and a fleet box's own
+// CARLOS_STATUS_PREFIX (which does apply to its status/metrics stores) has
+// no bearing on where its bindings land.
+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 {
+ body, err := json.Marshal(b)
+ if err != nil {
+ return err
+ }
+ key := fleetBindingKey(b.Fleet, b.Host)
+ etag, err := currentBindingEtag(ctx, s, key)
+ if err != nil {
+ return err
+ }
+ 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)
+ if err != nil {
+ return err
+ }
+ if _, err := s.Put(ctx, key, body, etag); err != nil {
+ return 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.
+//
+// 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.
+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 {
+ log.Printf("host: fleet bindings: list %s: %v", dir, err)
+ continue
+ }
+ for _, key := range keys {
+ body, _, err := s.Get(ctx, key)
+ if err != nil {
+ log.Printf("host: fleet bindings: skipping %s: read failed: %v", key, err)
+ continue
+ }
+ var b FleetBinding
+ if err := json.Unmarshal(body, &b); err != nil {
+ 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
+ }
+ }
+ }
+ 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).
+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 {
+ log.Printf("agent: fleet binding for %s: %v", host, err)
+ }
+ }
+}
+
+// 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:
+// - 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
+// wake, which may be a long time coming, or never for an always-awake
+// sidecar host;
+// - a route that isn't activator-managed at all (Hibernate unset, or a
+// unit-backed Backing) never goes through activator.wake in the first
+// place, so it would never get a binding published any other way.
+//
+// Best-effort and non-fatal per host: a failed List aborts the whole pass
+// (logged; the next restart tries again), but one host's write failure does
+// not stop the rest of the pass from covering everyone else.
+func publishResidentFleetBindings(store release.Store, fleet, box string, reg *registry.Registry) {
+ routes, err := reg.List()
+ if err != nil {
+ log.Printf("agent: fleet bindings startup pass: list routes: %v", err)
+ return
+ }
+ ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
+ defer cancel()
+ now := time.Now().Unix()
+ for _, rt := range routes {
+ b := FleetBinding{Host: rt.Host, Fleet: fleet, Box: box, BoundAt: now}
+ if err := WriteFleetBinding(ctx, store, b); 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
new file mode 100644
index 000000000..69c6166e7
--- /dev/null
+++ b/internal/host/fleetbind_test.go
@@ -0,0 +1,335 @@
+package host
+
+import (
+ "bytes"
+ "context"
+ "encoding/json"
+ "errors"
+ "log"
+ "path/filepath"
+ "strings"
+ "testing"
+
+ "github.com/carlosframework/platform/internal/registry"
+ "github.com/carlosframework/platform/internal/release"
+)
+
+func testRegistry(t *testing.T) *registry.Registry {
+ t.Helper()
+ reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db"))
+ if err != nil {
+ t.Fatalf("open registry: %v", err)
+ }
+ t.Cleanup(func() { reg.Close() })
+ return reg
+}
+
+// --- 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}
+
+ if err := WriteFleetBinding(ctx, s, b); err != nil {
+ t.Fatalf("WriteFleetBinding: %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 != b {
+ t.Fatalf("round-tripped binding = %+v, want %+v", got, b)
+ }
+}
+
+// A box republishes to the SAME key on every wake and once more at agent
+// startup — a bare create-only Put would fail every write after the first
+// with ErrCASConflict, which is exactly the bug the read-etag-then-Put dance
+// (mirroring status.WriteBox) exists to avoid.
+func TestWriteFleetBindingOverwritesExistingKey(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ 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 {
+ t.Fatalf("first write: %v", err)
+ }
+ if err := WriteFleetBinding(ctx, s, second); err != nil {
+ t.Fatalf("second write (must replace, not conflict): %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 != 200 {
+ t.Fatalf("BoundAt = %d, want the second write's 200", 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 {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := WriteFleetBinding(ctx, s, FleetBinding{
+ Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
+ }); err != nil {
+ t.Fatalf("WriteFleetBinding: %v", err)
+ }
+
+ if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
+ t.Fatalf("ReconcileFleetBindings: %v", err)
+ }
+
+ rt, err := reg.Get("one.example.com")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if rt.FleetBox != "amber-gliding" {
+ t.Fatalf("FleetBox = %q, want amber-gliding", rt.FleetBox)
+ }
+}
+
+// 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.
+func TestReconcileFleetBindingsLeavesNonFleetRoutesAlone(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ reg := testRegistry(t)
+ if err := reg.Upsert(registry.Route{Host: "static.example.com", Kind: "static"}); err != nil {
+ t.Fatalf("upsert: %v", err)
+ }
+
+ if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
+ t.Fatalf("ReconcileFleetBindings: %v", err)
+ }
+
+ rt, err := reg.Get("static.example.com")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if rt.FleetBox != "" {
+ t.Fatalf("FleetBox = %q, want empty (untouched)", rt.FleetBox)
+ }
+}
+
+// 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) {
+ 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 {
+ t.Fatalf("upsert: %v", err)
+ }
+ 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)
+ }
+ 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)
+ }
+
+ 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 != "new-box" {
+ t.Fatalf("FleetBox = %q, want new-box (BoundAt 900 beats 500)", rt.FleetBox)
+ }
+}
+
+// A corrupt binding object must be logged and skipped, never fatal — the
+// loadBoxes precedent (internal/status/fleet.go) — and must not stop the
+// pass from stamping every OTHER host's valid binding.
+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 {
+ t.Fatalf("upsert good: %v", err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "bad.example.com", Kind: "instance"}); 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 {
+ t.Fatalf("seed corrupt object: %v", err)
+ }
+ if err := WriteFleetBinding(ctx, s, FleetBinding{
+ Host: "good.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
+ }); err != nil {
+ t.Fatalf("WriteFleetBinding: %v", err)
+ }
+
+ if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
+ t.Fatalf("ReconcileFleetBindings: %v", err)
+ }
+
+ good, err := reg.Get("good.example.com")
+ if err != nil {
+ t.Fatalf("Get good: %v", err)
+ }
+ if good.FleetBox != "amber-gliding" {
+ t.Fatalf("good.FleetBox = %q, want amber-gliding despite the corrupt sibling", good.FleetBox)
+ }
+ bad, err := reg.Get("bad.example.com")
+ if err != nil {
+ t.Fatalf("Get bad: %v", err)
+ }
+ if bad.FleetBox != "" {
+ t.Fatalf("bad.FleetBox = %q, want empty (corrupt binding never stamped)", bad.FleetBox)
+ }
+}
+
+// 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).
+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 {
+ t.Fatalf("upsert: %v", err)
+ }
+ if err := WriteFleetBinding(ctx, s, FleetBinding{
+ Host: "one.example.com", Fleet: "pi", Box: "amber-gliding", BoundAt: 1000,
+ }); err != nil {
+ t.Fatalf("WriteFleetBinding: %v", err)
+ }
+ for i := 0; i < 2; i++ {
+ if err := ReconcileFleetBindings(ctx, s, reg); err != nil {
+ t.Fatalf("ReconcileFleetBindings pass %d: %v", i, err)
+ }
+ }
+ rt, err := reg.Get("one.example.com")
+ if err != nil {
+ t.Fatalf("Get: %v", err)
+ }
+ if rt.FleetBox != "amber-gliding" {
+ t.Fatalf("FleetBox = %q, want amber-gliding", rt.FleetBox)
+ }
+}
+
+// --- publishResidentFleetBindings (agent-startup pass) ------------------------
+
+func TestPublishResidentFleetBindingsCoversEveryLocalRoute(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 one: %v", err)
+ }
+ if err := reg.Upsert(registry.Route{Host: "two.example.com", Kind: "instance"}); err != nil {
+ t.Fatalf("upsert two: %v", err)
+ }
+
+ publishResidentFleetBindings(s, "pi", "amber-gliding", reg)
+
+ for _, host := range []string{"one.example.com", "two.example.com"} {
+ body, _, err := s.Get(ctx, fleetBindingKey("pi", host))
+ if err != nil {
+ t.Fatalf("Get %s: %v", host, err)
+ }
+ var b FleetBinding
+ if err := json.Unmarshal(body, &b); err != nil {
+ t.Fatalf("unmarshal %s: %v", host, err)
+ }
+ if b.Host != host || b.Fleet != "pi" || b.Box != "amber-gliding" || b.BoundAt == 0 {
+ t.Fatalf("binding for %s = %+v, want host/fleet/box populated and a nonzero BoundAt", host, b)
+ }
+ }
+}
+
+// A registry with no routes at all must not error or panic — a startup pass
+// on a freshly-provisioned fleet box, before any host has been assigned.
+func TestPublishResidentFleetBindingsEmptyRegistryIsNoop(t *testing.T) {
+ s := release.NewDirStore(t.TempDir())
+ reg := testRegistry(t)
+ publishResidentFleetBindings(s, "pi", "amber-gliding", reg)
+ keys, err := s.List(context.Background(), "fleets/")
+ if err != nil {
+ t.Fatalf("List: %v", err)
+ }
+ if len(keys) != 0 {
+ t.Fatalf("keys = %v, want none written for an empty registry", keys)
+ }
+}
+
+// --- fleetBindingHook (fire-and-forget wake publish) --------------------------
+
+// errPutStore fails every Put while otherwise behaving like a normal
+// DirStore (Get/List/ListDirs/Delete pass through) — enough to exercise
+// WriteFleetBinding's error path without a broken embedded nil Store.
+type errPutStore struct{ release.Store }
+
+func (errPutStore) Put(ctx context.Context, key string, body []byte, ifMatch string) (string, error) {
+ return "", errors.New("boom: put refused")
+}
+
+func TestFleetBindingHookLogsErrorWithoutPanicking(t *testing.T) {
+ var buf bytes.Buffer
+ prev := log.Writer()
+ log.SetOutput(&buf)
+ defer log.SetOutput(prev)
+
+ store := errPutStore{Store: release.NewDirStore(t.TempDir())}
+ hook := fleetBindingHook(store, "pi", "amber-gliding")
+
+ hook("one.example.com") // must not panic even though every Put fails
+
+ if !strings.Contains(buf.String(), "fleet binding") {
+ t.Fatalf("expected the Put failure to be logged, got: %q", buf.String())
+ }
+}
+
+func TestFleetBindingHookWritesBinding(t *testing.T) {
+ ctx := context.Background()
+ s := release.NewDirStore(t.TempDir())
+ hook := fleetBindingHook(s, "pi", "amber-gliding")
+
+ hook("woken.example.com")
+
+ body, _, err := s.Get(ctx, fleetBindingKey("pi", "woken.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)
+ }
+ if b.Host != "woken.example.com" || b.Fleet != "pi" || b.Box != "amber-gliding" {
+ t.Fatalf("binding = %+v, want host/fleet/box populated from the hook's closure", b)
+ }
+}