| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "log" |
| 8 | "os" |
| 9 | "path" |
| 10 | "time" |
| 11 | |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | "github.com/carlosframework/platform/internal/release" |
| 14 | ) |
| 15 | |
| 16 | // FleetBinding is the deployment-bucket object a fleet box publishes to say |
| 17 | // "as of BoundAt, Box (the box's fleet_boxes.label — resolveFleetIdentity, |
| 18 | // NEVER status.Label(machineID)) is the box currently serving Host". Fleet |
| 19 | // boxes are outbound-only — they dial the platform edge over a reverse |
| 20 | // tunnel and can never write the platform box's registry directly |
| 21 | // (customer-fleets design spec, "Routing to N boxes: the lease stamp") — so |
| 22 | // this bucket object is the ONLY channel a fleet box has to tell the |
| 23 | // platform which of its boxes currently holds a given hostname's lease. |
| 24 | // ReconcileFleetBindings is the platform-side reader that turns it back |
| 25 | // into registry.Route.FleetBox. |
| 26 | type FleetBinding struct { |
| 27 | Host string `json:"host"` |
| 28 | Fleet string `json:"fleet"` |
| 29 | Box string `json:"box"` // the holding box's fleet_boxes.label |
| 30 | // BoundAt is unix MILLISECONDS (not seconds) — a wire-shape choice made |
| 31 | // free at Task 6's fix round, before any binding object has ever shipped |
| 32 | // to production: two wakes racing on different boxes (or a wake racing |
| 33 | // the startup pass on the SAME box) can land inside the same second, and |
| 34 | // millisecond resolution is what lets bindingBlocked's logical-clock |
| 35 | // guard actually distinguish them instead of comparing equal. |
| 36 | BoundAt int64 `json:"bound_at"` |
| 37 | } |
| 38 | |
| 39 | // fleetBindingKey is the one place the object's key is spelled, shared by |
| 40 | // the writer and the reader below: fleets/<fleet>/bindings/<host>.json, |
| 41 | // ABSOLUTE in the PLATFORM'S deployment bucket — a fleet box has TWO buckets |
| 42 | // (its own fleet/data bucket the activator leases and restores instances |
| 43 | // against, and the platform's shared deployment bucket every box, fleet or |
| 44 | // not, adopts releases from and now publishes/reads bindings in) and this |
| 45 | // key lives in the latter, never the former. |
| 46 | // |
| 47 | // Deliberately not routed through release.NewPrefixStore's |
| 48 | // CARLOS_STATUS_PREFIX rebasing: that wrapper exists so a fleet box's own |
| 49 | // status/metrics/alerts heartbeats land under its private |
| 50 | // "fleets/<name>/" corner instead of the platform's shared paths — but a |
| 51 | // binding is not a heartbeat, it is a cross-box MESSAGE the platform box |
| 52 | // must read back, and it already carries its own fleet name (b.Fleet) in |
| 53 | // the struct. Both WriteFleetBinding and ReconcileFleetBindings take an |
| 54 | // UNPREFIXED release.Store and compute this absolute key themselves, so |
| 55 | // there is exactly one key format to keep in sync, and a fleet box's own |
| 56 | // CARLOS_STATUS_PREFIX (which does apply to its status/metrics stores) has |
| 57 | // no bearing on where its bindings land. |
| 58 | func fleetBindingKey(fleet, host string) string { |
| 59 | return path.Join("fleets", fleet, "bindings", host+".json") |
| 60 | } |
| 61 | |
| 62 | // bindingBlocked reports whether writing b over the binding currently stored |
| 63 | // at its key (existing, nil if none) must be skipped. Two independent |
| 64 | // guards, either one enough to block: |
| 65 | // |
| 66 | // - Non-stealing: a non-takeover write (the startup pass, |
| 67 | // publishResidentFleetBindings — a bare "this box has this route |
| 68 | // resident" observation, not lease-backed) must never overwrite a |
| 69 | // binding that already names a DIFFERENT box. Only a wake-success write |
| 70 | // (takeover=true) may take over — it is backed by having just won the |
| 71 | // per-host S3 lease, which is the actual arbiter of who may serve a |
| 72 | // host, so it is allowed to restamp regardless of who was there before. |
| 73 | // - Logical clock, SAME BOX ONLY: a binding already stored FROM THE SAME |
| 74 | // BOX with a strictly greater BoundAt must never be overwritten by an |
| 75 | // older write from that same box — this holds regardless of takeover, |
| 76 | // and fixes a physical-write-race hazard: two concurrent Puts to the |
| 77 | // same key from one box (a retried wake publish racing its own |
| 78 | // predecessor, or any other overlap) must not let whichever one's HTTP |
| 79 | // round-trip happens to land second win just because it arrived later. |
| 80 | // Scoped to existing.Box == b.Box deliberately (fix round 2, Opus |
| 81 | // review): BoundAt is each BOX's OWN wall clock, never a shared one — |
| 82 | // comparing it across two DIFFERENT boxes compares clock skew, not |
| 83 | // recency. Before this scoping, a dead box A's stale binding stamped |
| 84 | // with a clock running ahead of surviving box B's would silently and |
| 85 | // PERMANENTLY block B's lease-backed takeover write (the startup pass |
| 86 | // can't help either — it's non-stealing by the guard above), stranding |
| 87 | // the edge on its sorted-fallback channel instead of the real lease |
| 88 | // holder for as long as the skew persists. A cross-box binding is |
| 89 | // already fully governed by the non-stealing guard above (blocked |
| 90 | // unless takeover=true); the clock only ever needs to arbitrate a BOX |
| 91 | // against its own past self. |
| 92 | func bindingBlocked(existing *FleetBinding, b FleetBinding, takeover bool) bool { |
| 93 | if existing == nil { |
| 94 | return false |
| 95 | } |
| 96 | if !takeover && existing.Box != b.Box { |
| 97 | return true |
| 98 | } |
| 99 | return existing.Box == b.Box && existing.BoundAt > b.BoundAt |
| 100 | } |
| 101 | |
| 102 | // getExistingBinding reads key's current binding + etag for a replace-only |
| 103 | // Put, folding "doesn't exist yet" into (nil, "", nil) — a create, with |
| 104 | // neither guard in bindingBlocked applicable. A corrupt existing object is |
| 105 | // treated as absent (logged) rather than failing the write outright: the |
| 106 | // same "never fatal" spirit ReconcileFleetBindings' reader-side skip uses, |
| 107 | // and it ensures a legitimate new write is never permanently stuck behind a |
| 108 | // broken object at its own key. |
| 109 | func getExistingBinding(ctx context.Context, s release.Store, key string) (*FleetBinding, string, error) { |
| 110 | body, etag, err := s.Get(ctx, key) |
| 111 | if err != nil { |
| 112 | if errors.Is(err, os.ErrNotExist) { |
| 113 | return nil, "", nil |
| 114 | } |
| 115 | return nil, "", err |
| 116 | } |
| 117 | var b FleetBinding |
| 118 | if err := json.Unmarshal(body, &b); err != nil { |
| 119 | log.Printf("host: fleet binding: existing object at %s is corrupt, treating as absent: %v", key, err) |
| 120 | return nil, etag, nil |
| 121 | } |
| 122 | return &b, etag, nil |
| 123 | } |
| 124 | |
| 125 | // WriteFleetBinding publishes b to the PLATFORM'S deployment bucket, |
| 126 | // replacing whatever is currently at its key — UNLESS bindingBlocked says |
| 127 | // the existing binding must win (see its doc comment for the two guards). |
| 128 | // s must be an UNPREFIXED store handle — see fleetBindingKey's doc comment. |
| 129 | // |
| 130 | // takeover distinguishes the two callers: fleetBindingHook (the activator's |
| 131 | // OnWake, a lease-backed wake success) passes true and may claim a binding |
| 132 | // away from any other box; publishResidentFleetBindings (the agent-startup |
| 133 | // pass, a bare "this route is resident here" observation with no lease |
| 134 | // behind it) passes false and must never steal a binding another box |
| 135 | // currently holds. |
| 136 | // |
| 137 | // The read-etag-then-Put dance mirrors status.WriteBox, and for the same |
| 138 | // reason: a box republishes to the SAME key on every wake and once more at |
| 139 | // startup, so a bare create-only Put (ifMatch "") would fail every write |
| 140 | // after the first with ErrCASConflict. |
| 141 | func WriteFleetBinding(ctx context.Context, s release.Store, b FleetBinding, takeover bool) error { |
| 142 | body, err := json.Marshal(b) |
| 143 | if err != nil { |
| 144 | return err |
| 145 | } |
| 146 | key := fleetBindingKey(b.Fleet, b.Host) |
| 147 | |
| 148 | existing, etag, err := getExistingBinding(ctx, s, key) |
| 149 | if err != nil { |
| 150 | return err |
| 151 | } |
| 152 | if bindingBlocked(existing, b, takeover) { |
| 153 | return nil |
| 154 | } |
| 155 | |
| 156 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 157 | if !errors.Is(err, release.ErrCASConflict) { |
| 158 | return err |
| 159 | } |
| 160 | // Lost the race against a concurrent write to the same key — |
| 161 | // re-read once, re-apply BOTH guards against the fresher state (it |
| 162 | // may now block a write that was fine against the stale read), and |
| 163 | // retry — the same single-retry contract status.WriteBox uses. |
| 164 | existing, etag, err = getExistingBinding(ctx, s, key) |
| 165 | if err != nil { |
| 166 | return err |
| 167 | } |
| 168 | if bindingBlocked(existing, b, takeover) { |
| 169 | return nil |
| 170 | } |
| 171 | if _, err := s.Put(ctx, key, body, etag); err != nil { |
| 172 | return err |
| 173 | } |
| 174 | } |
| 175 | return nil |
| 176 | } |
| 177 | |
| 178 | // ReconcileFleetBindings is the platform-side half of fleet routing: for |
| 179 | // every fleets/<fleet>/bindings/<host>.json object published under any |
| 180 | // fleet, if the route registered for that host is actually PLACED on that |
| 181 | // SAME fleet (registry.FleetPlacement(rt.Region) == the binding's own |
| 182 | // Fleet) and its FleetBox doesn't already match, stamp it via SetFleetBox. |
| 183 | // |
| 184 | // The fleet-placement check is a v1 trust-boundary requirement, not a |
| 185 | // nicety: fleet boxes are outbound-only, but nothing at the bucket layer |
| 186 | // stops fleet A's box from publishing a binding that names a host actually |
| 187 | // placed on fleet B (a same-named host collision, or a stale/rewritten |
| 188 | // object) — without this check that binding could silently override fleet |
| 189 | // B's own routing for a host fleet A's box has no business serving. There |
| 190 | // is deliberately no "newest wins" tie-break for a cross-fleet collision: |
| 191 | // two bindings for the exact same (fleet, host) pair cannot exist (the key |
| 192 | // IS fleet+host), and a binding whose fleet doesn't match the route's own |
| 193 | // placement is rejected outright, never preferred by timestamp — an earlier |
| 194 | // version of this function used a cross-fleet "newest BoundAt" reduction |
| 195 | // before this check existed; it's gone because the placement check makes it |
| 196 | // both wrong (a malicious/stale binding could out-race a legitimate one by |
| 197 | // BoundAt alone) and unnecessary (same-fleet duplicates are structurally |
| 198 | // impossible). |
| 199 | // |
| 200 | // s must be an UNPREFIXED store handle (see fleetBindingKey) — this walks |
| 201 | // the same absolute "fleets/" keys WriteFleetBinding writes, and is only |
| 202 | // ever called on platform boxes, which never set CARLOS_STATUS_PREFIX in |
| 203 | // the first place. |
| 204 | // |
| 205 | // Corrupt objects and per-key read failures are logged and skipped, never |
| 206 | // fatal — the loadBoxes precedent (internal/status/fleet.go:197-216): one |
| 207 | // bad binding must not blank out routing for every other fleet host. A |
| 208 | // missing route, or one not placed on the binding's own fleet, is silently |
| 209 | // skipped too — neither is this binding's to stamp. |
| 210 | func ReconcileFleetBindings(ctx context.Context, s release.Store, reg *registry.Registry) error { |
| 211 | fleetDirs, err := s.ListDirs(ctx, "fleets/") |
| 212 | if err != nil { |
| 213 | return err |
| 214 | } |
| 215 | for _, dir := range fleetDirs { |
| 216 | keys, err := s.List(ctx, dir+"bindings/") |
| 217 | if err != nil { |
| 218 | log.Printf("host: fleet bindings: list %s: %v", dir, err) |
| 219 | continue |
| 220 | } |
| 221 | for _, key := range keys { |
| 222 | body, _, err := s.Get(ctx, key) |
| 223 | if err != nil { |
| 224 | log.Printf("host: fleet bindings: skipping %s: read failed: %v", key, err) |
| 225 | continue |
| 226 | } |
| 227 | var b FleetBinding |
| 228 | if err := json.Unmarshal(body, &b); err != nil { |
| 229 | log.Printf("host: fleet bindings: skipping %s: corrupt binding: %v", key, err) |
| 230 | continue |
| 231 | } |
| 232 | rt, err := reg.Get(b.Host) |
| 233 | if err != nil { |
| 234 | log.Printf("host: fleet bindings: read route %s: %v", b.Host, err) |
| 235 | continue |
| 236 | } |
| 237 | if rt == nil { |
| 238 | continue // no matching route on this platform box |
| 239 | } |
| 240 | // Never stamp a sidecar. SetFleetBox on a fleet-placed route is |
| 241 | // precisely what turns the edge's fleet dispatch arm into a live |
| 242 | // proxy, so this reader is the last gate on the chain and it |
| 243 | // refuses on kind rather than trusting its input. |
| 244 | // |
| 245 | // Redundant with residentBindingHosts on the publishing side, and |
| 246 | // deliberately so: the two halves run on DIFFERENT machines and |
| 247 | // upgrade independently. A fleet box still on an older binary |
| 248 | // keeps publishing sidecar bindings into the bucket for the whole |
| 249 | // of a mixed-version roll, and this platform box is what reads |
| 250 | // them. A binding is an untrusted claim written by another host; |
| 251 | // the route row is local truth. |
| 252 | if rt.IsSidecar() { |
| 253 | continue |
| 254 | } |
| 255 | if fleet, ok := registry.FleetPlacement(rt.Region); !ok || fleet != b.Fleet { |
| 256 | // The route isn't placed on THIS binding's fleet at all — |
| 257 | // v1 trust boundary: refuse to let it stamp (see doc |
| 258 | // comment above). |
| 259 | continue |
| 260 | } |
| 261 | if rt.FleetBox == b.Box { |
| 262 | continue |
| 263 | } |
| 264 | if err := reg.SetFleetBox(b.Host, b.Box); err != nil { |
| 265 | log.Printf("host: fleet bindings: stamp %s -> %s: %v", b.Host, b.Box, err) |
| 266 | } |
| 267 | } |
| 268 | } |
| 269 | return nil |
| 270 | } |
| 271 | |
| 272 | // fleetBindingHook builds the activator's OnWake hook for a fleet box: on |
| 273 | // every successful wake, publish a FleetBinding naming this box (identified |
| 274 | // by fleet/box, resolveFleetIdentity's result — the channel label, never a |
| 275 | // hashed pseudonym) as the current holder, with takeover=true since a |
| 276 | // wake's lease acquisition is exactly the authority that may claim a |
| 277 | // binding away from whoever held it before. activator.Service already |
| 278 | // invokes OnWake in its own goroutine (see activator.go's wake), so this |
| 279 | // function runs off the wake path entirely — a slow or failing bucket Put |
| 280 | // can never delay or fail a request. Errors are logged only: a binding is |
| 281 | // eventually-consistent by design (the design spec's failure sequence — box |
| 282 | // goes dark, heartbeat ages, lease expires, the next request wakes a |
| 283 | // surviving box, which restamps — is exactly the fallback that covers a |
| 284 | // dropped write here). |
| 285 | // |
| 286 | // The hook takes the whole ROUTE, which is what the activator's wake path has |
| 287 | // in hand anyway, and refuses on its Kind: a sidecar host must never have a |
| 288 | // binding published for it, and this is the write path that carries TAKEOVER |
| 289 | // semantics, so an errant one here does not merely appear — it claims the host |
| 290 | // away from whoever legitimately held it. The filter lives here rather than |
| 291 | // resting on "the activator never wakes a sidecar": that is true of the |
| 292 | // activator TODAY, and it is a fact a later task in this same branch removes |
| 293 | // (the sidecar worker goes through the same wake() that fires OnWake). A guard |
| 294 | // that depends on a caller's current behaviour is not a guard. |
| 295 | // |
| 296 | // Taking the route also removes the registry read this refusal used to need. |
| 297 | // That read could fail, and a security-relevant check whose input can fail |
| 298 | // must then decide which way to fail: it published anyway (fail OPEN), because |
| 299 | // a dropped binding is a routing outage and the fallback that covers it is |
| 300 | // slow. The route the wake already holds is the same local truth the read was |
| 301 | // fetching, so the refusal is now total and depends on no I/O at all. |
| 302 | func fleetBindingHook(store release.Store, fleet, box string) func(rt registry.Route) { |
| 303 | return func(rt registry.Route) { |
| 304 | if rt.IsSidecar() { |
| 305 | return |
| 306 | } |
| 307 | ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 308 | defer cancel() |
| 309 | b := FleetBinding{Host: rt.Host, Fleet: fleet, Box: box, BoundAt: time.Now().UnixMilli()} |
| 310 | if err := WriteFleetBinding(ctx, store, b, true); err != nil { |
| 311 | log.Printf("agent: fleet binding for %s: %v", rt.Host, err) |
| 312 | } |
| 313 | } |
| 314 | } |
| 315 | |
| 316 | // publishResidentFleetBindings runs once at fleet-box agent startup: every |
| 317 | // route already present in this box's local registry gets a binding |
| 318 | // written immediately, without waiting for an activator wake, with |
| 319 | // takeover=false — this is a bare "this route is resident here" observation |
| 320 | // with no lease behind it, so it must never steal a binding a wake-success |
| 321 | // write already holds for a DIFFERENT box (see WriteFleetBinding/ |
| 322 | // bindingBlocked). Two gaps this closes that a wake-only publish would |
| 323 | // miss: |
| 324 | // - an agent restart discards activator.Service's in-memory entries, but |
| 325 | // an already-running exec child (or a unit-backed tenant) keeps serving |
| 326 | // — its host would otherwise have no fresh binding until its NEXT idle |
| 327 | // wake, which may be a long time coming, or never for a host that is |
| 328 | // always awake; |
| 329 | // - a route that isn't activator-managed at all (Hibernate unset, or a |
| 330 | // unit-backed Backing) never goes through activator.wake in the first |
| 331 | // place, so it would never get a binding published any other way. |
| 332 | // |
| 333 | // Best-effort and non-fatal per host: a failed List aborts the whole pass |
| 334 | // (logged; the next restart tries again), but one host's write failure does |
| 335 | // not stop the rest of the pass from covering everyone else. |
| 336 | func publishResidentFleetBindings(store release.Store, fleet, box string, reg *registry.Registry) { |
| 337 | routes, err := reg.List() |
| 338 | if err != nil { |
| 339 | log.Printf("agent: fleet bindings startup pass: list routes: %v", err) |
| 340 | return |
| 341 | } |
| 342 | ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second) |
| 343 | defer cancel() |
| 344 | now := time.Now().UnixMilli() |
| 345 | for _, host := range residentBindingHosts(routes) { |
| 346 | b := FleetBinding{Host: host, Fleet: fleet, Box: box, BoundAt: now} |
| 347 | if err := WriteFleetBinding(ctx, store, b, false); err != nil { |
| 348 | log.Printf("agent: fleet binding for %s: %v", host, err) |
| 349 | } |
| 350 | } |
| 351 | } |
| 352 | |
| 353 | // residentBindingHosts is the set of hosts this box advertises as served — |
| 354 | // the pure host-selection half of publishResidentFleetBindings, split out so |
| 355 | // the refusal below is directly testable. |
| 356 | // |
| 357 | // Sidecar routes (Kind "sidecar") are excluded. Nothing routes to a sidecar, |
| 358 | // so a binding claiming a box serves it is a lie the platform side then acts |
| 359 | // on: ReconcileFleetBindings stamps SetFleetBox from it, and a fleet-placed |
| 360 | // route is exactly what the edge's fleet dispatch arm turns into a live |
| 361 | // internet-facing proxy. Filtered at the publisher rather than at the |
| 362 | // reconciler because this is the only place that decides what a box CLAIMS — |
| 363 | // the claim is what must never exist. |
| 364 | func residentBindingHosts(routes []registry.Route) []string { |
| 365 | var out []string |
| 366 | for _, rt := range routes { |
| 367 | if rt.Host == "" || rt.IsSidecar() { |
| 368 | continue |
| 369 | } |
| 370 | out = append(out, rt.Host) |
| 371 | } |
| 372 | return out |
| 373 | } |
| 374 | |
amadan