| 1 | // Ported from lchat activator.go (the s3Backend half) and the exec/env/health |
| 2 | // helpers it leaned on from lchat control.go and versions.go. |
| 3 | package activator |
| 4 | |
| 5 | import ( |
| 6 | "context" |
| 7 | "encoding/json" |
| 8 | "errors" |
| 9 | "fmt" |
| 10 | "io" |
| 11 | "io/fs" |
| 12 | "log" |
| 13 | "maps" |
| 14 | "net" |
| 15 | "net/http" |
| 16 | "os" |
| 17 | "os/exec" |
| 18 | "path/filepath" |
| 19 | "sort" |
| 20 | "strings" |
| 21 | "sync" |
| 22 | "syscall" |
| 23 | "time" |
| 24 | |
| 25 | "github.com/carlosframework/platform/internal/applog" |
| 26 | "github.com/carlosframework/platform/internal/cgroup" |
| 27 | "github.com/carlosframework/platform/internal/edge" |
| 28 | "github.com/carlosframework/platform/internal/instancectl" |
| 29 | "github.com/carlosframework/platform/internal/instancekey" |
| 30 | "github.com/carlosframework/platform/internal/instanceupdates" |
| 31 | "github.com/carlosframework/platform/internal/lease" |
| 32 | "github.com/carlosframework/platform/internal/moves" |
| 33 | "github.com/carlosframework/platform/internal/registry" |
| 34 | "github.com/carlosframework/platform/internal/release" |
| 35 | "github.com/carlosframework/platform/internal/safefile" |
| 36 | "github.com/carlosframework/platform/internal/schedule" |
| 37 | "github.com/carlosframework/platform/internal/seeds" |
| 38 | "github.com/carlosframework/platform/internal/unixdial" |
| 39 | ) |
| 40 | |
| 41 | // spawnHealthTimeout bounds how long a spawned instance may take to bind its |
| 42 | // socket and answer. A var so tests can shrink the failure path. |
| 43 | var spawnHealthTimeout = 30 * time.Second |
| 44 | |
| 45 | // ErrSeedUnproven is seedUnit's answer when a seed's cgroup never reports |
| 46 | // empty (design §6.2): only a SIGKILL survivor reaches this, and while its |
| 47 | // cgroup stays populated it may still be writing the database, so the |
| 48 | // caller must treat this the way stop() treats ErrStillRunning and leave |
| 49 | // the lease for its own TTL to expire rather than release it over a write |
| 50 | // that might still be happening. |
| 51 | var ErrSeedUnproven = errors.New("activator: seed unproven; processes remain") |
| 52 | |
| 53 | // seedUnprovenSuffix names a flat seed's unproven-cleanup obligation, kept |
| 54 | // beside the database (<db>.seed-unproven): a seed failed, its processes |
| 55 | // were not proven gone, and so its database was kept. While the file is |
| 56 | // there no wake treats the database as warm (Restore). It does not end in |
| 57 | // ".db", so the box-wide litestream glob never picks it up. |
| 58 | const seedUnprovenSuffix = ".seed-unproven" |
| 59 | |
| 60 | // writeSeedObligationFile is the obligation's durable write: a var so a |
| 61 | // test can make it fail (a full disk). |
| 62 | var writeSeedObligationFile = safefile.WriteFile |
| 63 | |
| 64 | // quarantineSuffix names where a restore that failed its integrity check is |
| 65 | // parked. It deliberately does NOT end in ".db": the box-wide litestream |
| 66 | // generator globs <data>/*.db, and a quarantined corpse must never be picked |
| 67 | // up for replication. |
| 68 | const quarantineSuffix = ".corrupt" |
| 69 | |
| 70 | // ExecBackend is the production Backend: leases + restore via the lease |
| 71 | // package's litestream primitives, processes via exec of the same versioned |
| 72 | // binary a systemd unit would run — `<OptDir>/live/<host>` with the |
| 73 | // `<DataDir>/<host>.env` EnvironmentFile. |
| 74 | type ExecBackend struct { |
| 75 | Cfg Config |
| 76 | // OptDir is the on-box install root: the version store and the |
| 77 | // per-instance "live" symlinks live under it (/opt/carlos in production; |
| 78 | // a struct field, never hardcoded, so tests point it at a temp dir). |
| 79 | OptDir string |
| 80 | // DataDir holds each instance's DB and env files (<host>.db, <host>.env, |
| 81 | // <host>-sidecar.env) — /data in production. |
| 82 | DataDir string |
| 83 | // Materialize, when non-nil, stages a binary and points the host's live |
| 84 | // symlink at it via an external mechanism instead of the backend doing |
| 85 | // filesystem work itself — Eleven needed this because its router ran |
| 86 | // sandboxed (ProtectSystem=strict) and couldn't write to /opt itself. |
| 87 | // With it nil, a missing live symlink is an error: Eleven fell back to |
| 88 | // staging its OWN binary (instance and control binary were one program), |
| 89 | // but a carlos app is a separate binary from the release store — there is |
| 90 | // nothing sensible to invent here, and the adopt pass owns the symlinks. |
| 91 | Materialize func(ctx context.Context, host string) error |
| 92 | // Mirror, when non-nil, arms Restore's staging fallback (issue #331): |
| 93 | // a host with no replica under the ordinary instances/ prefix may have |
| 94 | // one under moves/<host>/ — a residence move's staged copy, pushed by |
| 95 | // the SOURCE box. Restore then copies it into place (own bucket, own |
| 96 | // IAM), restores as normal, and deletes the staging copy once CheckDB |
| 97 | // vouches for what it restored. Nil = the fallback does not exist, |
| 98 | // byte-identical to the pre-move behaviour. |
| 99 | Mirror lease.MirrorStore |
| 100 | // Phase1, set by the agent from moves.Enabled, retires the staging |
| 101 | // fallback above: Restore never promotes a moves/<host>/ tree, because |
| 102 | // under Phase 1 the staged copy is the target's to read through the |
| 103 | // landing, and a restore that promoted it would be a second, older |
| 104 | // source of truth (moves design §5.3, M15). Off, the legacy promotion |
| 105 | // is exactly as it was until Plan 6 Task 8 removes it. |
| 106 | Phase1 bool |
| 107 | // Region is the box's CARLOS placement code (CARLOS_REGION), handed to |
| 108 | // every exec child as its own CARLOS_REGION (issue #329, titogo's |
| 109 | // residence design: the app reads where it lives — e.g. to clear a move |
| 110 | // freeze after landing). Agent-owned in BOTH directions, exactly like |
| 111 | // the control-channel vars below: set from what the agent knows, and |
| 112 | // STRIPPED when the agent knows nothing — a tenant's config env must not |
| 113 | // be able to tell an app it runs somewhere it does not. |
| 114 | Region string |
| 115 | // ReplicaHomeKeys answers, for a replica row's host, the SHA-256 of its |
| 116 | // home's effective CARLOS_INSTANCE_KEY (instance-key plan rulings 26 and |
| 117 | // 30). O8 will supply it from the home's published env item; until then |
| 118 | // the agent reads the box's list (/etc/carlos/replica-home-keys), which |
| 119 | // only the pre-roll check writes, again at every call. Nil, or false for |
| 120 | // a host, means the home key cannot be verified, and a replica launch |
| 121 | // that would be handed a key is refused (replicaKeyGate). |
| 122 | ReplicaHomeKeys func(host string) (sha256 string, ok bool) |
| 123 | // ConfiguredKey reads rt's configuration from the bucket now and |
| 124 | // returns the SHA-256 of the CARLOS_INSTANCE_KEY it renders, "" for |
| 125 | // none (appconfig.ConfiguredKeyDigest). Two readers ask it. The mint |
| 126 | // asks before it mints, so a key the configuration sets is never |
| 127 | // shadowed by one minted while that configuration was still on its |
| 128 | // way to the box (ensureInstanceKey; ruling 5): a key the bucket sets |
| 129 | // and the box has not delivered defers the launch, and nil is a box |
| 130 | // with no configuration source, where nothing is on its way. The |
| 131 | // replica gate asks at every launch whose key is configured, and |
| 132 | // refuses a replica whose configured key is not that digest: a |
| 133 | // configuration change is in flight (ruling 27); there nil refuses |
| 134 | // every configured key. |
| 135 | ConfiguredKey func(ctx context.Context, rt registry.Route) (sha256 string, err error) |
| 136 | // Cache, when non-nil, adds the purge verb to each spawned instance's |
| 137 | // control socket so it can invalidate its own cached responses right |
| 138 | // after a write. Nil disables that ONE verb, not the socket: the control |
| 139 | // channel is bound for every exec child either way (§7.2), because the |
| 140 | // scheduled-work verbs need it on boxes with no edge cache too. |
| 141 | Cache *edge.Cache |
| 142 | // Cgroups, when non-nil, places every spawned child — instance, sidecar, |
| 143 | // seed — in a per-host cgroup under the agent's delegated subtree so the |
| 144 | // metrics sampler can account it (exec-child accounting spec). Nil — no |
| 145 | // delegation: old unit file, tests, darwin — spawns exactly as before. |
| 146 | Cgroups *cgroup.Manager |
| 147 | // Tier, when non-nil, picks the memory cap for a route's carlos-exec@ |
| 148 | // cgroup (memory-pressure spec §2.3). The box wiring reads the staged |
| 149 | // manifest of the live version; tests leave it nil and get the default |
| 150 | // tier. Instance, seed and sidecar share the cgroup, so they share the |
| 151 | // tier — a seed's first-boot cost is that instance's work. |
| 152 | Tier func(registry.Route) cgroup.Limit |
| 153 | // GenContainment selects the generation layout (zero-downtime spec |
| 154 | // §7.3, §7.6): each instance spawn gets its own child cgroup g<n> (or |
| 155 | // the carlos-instance-gen@<id>:<n> unit for a nested host) under a |
| 156 | // parent that holds no processes. False, the zero value, is today's |
| 157 | // layout byte for byte: one cgroup per host and carlos-instance@. Set |
| 158 | // only by plan 5's CARLOS_SWAP wiring. |
| 159 | GenContainment bool |
| 160 | // TierOf is one named version's tier, read from that version's staged |
| 161 | // manifest (zero-downtime spec §7.3). A generation's limits are its own |
| 162 | // artifact's, never whatever the live link names at the moment. Nil |
| 163 | // falls back to Tier (tests, and every flag-off spawn). |
| 164 | TierOf func(rt registry.Route, version string) cgroup.Limit |
| 165 | // Logs, when non-nil, gets each instance and seed child's stdout/stderr |
| 166 | // teed in line-wise for `carlos logs` (spec §5) — journald keeps every |
| 167 | // byte it got before. Nil (tests, a box without the shipping loop) |
| 168 | // spawns exactly as before: fd passthrough, no pipes. |
| 169 | Logs EventSink |
| 170 | // Timers, when non-nil, mounts the scheduled-work verbs on each spawned |
| 171 | // instance's control socket so an app can set and clear its own one-shot |
| 172 | // timers (spec 2026-08-23-scheduled-work-design.md §7.2). The box wiring |
| 173 | // passes the registry. Nil — tests, and any box whose agent predates the |
| 174 | // feature — binds the control socket without those verbs, so an app that |
| 175 | // asks gets 404 rather than a socket that isn't there. |
| 176 | Timers schedule.TimerStore |
| 177 | // UpdatesReader answers the control socket's read-only GET /updates |
| 178 | // (instance update settings spec §5.1); nil mounts nothing. |
| 179 | UpdatesReader func(host string) (instanceupdates.Report, error) |
| 180 | // LayoutFacts supplies the box facts the classification of §8.1 needs |
| 181 | // beyond the row: journals and peers. Zero value: never fenced, never |
| 182 | // conflict. Named apart from the Layout method (the Classifier |
| 183 | // interface's Layout(rt) (Layout, string)) so a struct field and a |
| 184 | // method can coexist on the same type. |
| 185 | LayoutFacts LayoutInputs |
| 186 | // Control is the root instance-control channel a nested wake talks to |
| 187 | // (design §5.3). Nil refuses every nested wake with "instance control |
| 188 | // channel not configured" rather than dialing nothing. A flat or |
| 189 | // foreign host never needs it, and a nil Control does not touch them. |
| 190 | Control Control |
| 191 | // UnitState answers `systemctl show` for the observer (design §6.5). |
| 192 | // Nil: systemctlUnitState, the real systemctl binary. |
| 193 | UnitState UnitState |
| 194 | // CgroupEvents reads a unit's cgroup.events for the observer and for a |
| 195 | // seed's cgroup-empty wait. Nil: cgroupEventsUnder("/sys/fs/cgroup"), |
| 196 | // or CgroupRoot when set. |
| 197 | CgroupEvents CgroupEvents |
| 198 | // CgroupRoot overrides where cgroup v2 is mounted, for the default |
| 199 | // CgroupEvents reader. Nil (empty): /sys/fs/cgroup. |
| 200 | CgroupRoot string |
| 201 | // NestedOK gates a nested wake on the box facts a unit test can't see: |
| 202 | // the isolation mounts, the control service's own /ready. Nil: no |
| 203 | // gate (every unit test and any box that predates the check). Wired in |
| 204 | // production by Task 6. |
| 205 | NestedOK func() error |
| 206 | // Seeds reads a seed's snapshot from this region's data bucket |
| 207 | // (seeded-instances spec §3.4). Nil: no seed can be placed here, and a |
| 208 | // seeded row's wake fails rather than starting empty. |
| 209 | Seeds SeedObjects |
| 210 | // ClaimArmed is CARLOS_HOST_CLAIM on this box. The seed fallback is the |
| 211 | // plain path only; with claims armed it refuses (spec §3.4, #250). |
| 212 | ClaimArmed bool |
| 213 | // OnSeedPlaced records a seed placed (Task 4): the record's CAS to |
| 214 | // placed, then the snapshot's delete. Nil: not recorded here. |
| 215 | OnSeedPlaced func(ctx context.Context, account, app, host, id string) error |
| 216 | // ConfigGate answers whether this box can provide rt's config (amadan |
| 217 | // #213): nil means yes, an error means refuse. Acquire asks it BEFORE |
| 218 | // lease.Acquire, so a box that would start the instance without its |
| 219 | // secrets never holds the lease, and a keyed sibling takes it instead. |
| 220 | // The gate owns its own policy, logging and operator-visible record of |
| 221 | // the refusal; this backend only honours the answer. Nil: no gate (every |
| 222 | // unit test, and a box with no deploy store to read config from). |
| 223 | ConfigGate func(ctx context.Context, rt registry.Route) error |
| 224 | // LiveExecGate answers whether rt may start build, the build a start |
| 225 | // from the live link is about to run: nil means yes, an error refuses |
| 226 | // the start and comes back as it is. Only a box that declined |
| 227 | // CARLOS_SWAP=overlap and honours failure records or hold-ran markers |
| 228 | // from an earlier armed run sets it (internal/host, through |
| 229 | // Service.SetLiveExecGate, before Start), so the plan-time refusal is |
| 230 | // repeated at the real launch with the build actually being started |
| 231 | // (Astra R1 #6, the R2/R3 declined residuals). It is asked by a flat |
| 232 | // instance's or sidecar's live-link spawn (which then execs the staged |
| 233 | // path it resolved, not the link), a nested instance's or sidecar's |
| 234 | // live-link start and a seed from the live link (each pins the link |
| 235 | // first, declinedLivePin, and runs the file it pinned), and a worker's |
| 236 | // spawn from the live link. Nil, on every other box: each of |
| 237 | // those spawns is exactly as it was. It reads only memory, the row it |
| 238 | // is handed and the live link, and writes nothing: the activator asks |
| 239 | // it under its own lock too (convergeWorkerLocked). |
| 240 | LiveExecGate func(rt registry.Route, build string) error |
| 241 | // ObservePoll is how often the exit observer (unitproc.go) and a |
| 242 | // seed's cgroup-empty wait re-check unit state. 0: one second. A field |
| 243 | // so tests can shrink the wait without a package-global var shared |
| 244 | // across every test in the suite. |
| 245 | ObservePoll time.Duration |
| 246 | // RunDirOverride is test-only: when set, layoutInputs() classifies |
| 247 | // against it instead of provision.SocketDir and pathsFor resolves a |
| 248 | // nested run dir and log socket beneath it, so a test can exercise the |
| 249 | // nested layout without writing to /run/carlos. |
| 250 | RunDirOverride string |
| 251 | // ScheduleFence runs a schedule write made on the control socket of |
| 252 | // host's generation gen (the number the listener was bound for, 0 for |
| 253 | // a spawn that is not a generation's), unless the activator has |
| 254 | // switched away from that generation, and reports whether it ran |
| 255 | // (zero-downtime spec §7.10, final review F8). Wired to |
| 256 | // Service.ScheduleWrite by the agent. Nil (tests, a box before this |
| 257 | // wiring) refuses nothing. |
| 258 | ScheduleFence func(host string, gen uint64, write func()) bool |
| 259 | // layoutFn, when set, replaces Classify in StartCmd's layout check; a |
| 260 | // unit test's way to stand a temp directory in for a nested host. |
| 261 | layoutFn func(rt registry.Route) (Layout, string) |
| 262 | // Strict, when non-nil and true for a host, makes every spawn of that |
| 263 | // host strict (moves design §5.3): the child starts only inside a group |
| 264 | // prepareStrict prepared, which is drained (KillAndEmpty) and closed |
| 265 | // when the child exits, and a spawn that cannot be made strict is |
| 266 | // refused with ErrStrictRequired. The host package sets it from the hold |
| 267 | // view: a host whose record has a move in flight, with the switch on |
| 268 | // (Task 14). Nil, or false for a host, is today's spawn exactly. |
| 269 | Strict func(host string) bool |
| 270 | // prepareStrict is the strict path's seam. Nil: Cgroups.PrepareStrict, |
| 271 | // and a nil Cgroups under Strict is ErrStrictRequired. Tests replace |
| 272 | // this seam, never the Cgroups field. |
| 273 | prepareStrict func(host string, limit cgroup.Limit) (strictProc, error) |
| 274 | // killGen is the spawn path's own generation kill (startChildAt's |
| 275 | // failed health wait). Nil: KillGen. Tests replace this seam, never the |
| 276 | // Cgroups field. |
| 277 | killGen func(ctx context.Context, rt registry.Route, n uint64) error |
| 278 | // cgroupKill is seedDescendantsGone's kill and proof: child "" is the |
| 279 | // host cgroup (KillHost), any other name one of its children. Nil: the |
| 280 | // cgroup manager's own. Tests replace this seam, never the Cgroups |
| 281 | // field. |
| 282 | cgroupKill func(ctx context.Context, rt registry.Route, child string) error |
| 283 | // openHostCleanup is EmptyHost's open of the host cgroup. Nil: |
| 284 | // Cgroups.OpenCleanup. Tests replace this seam, never the Cgroups |
| 285 | // field. |
| 286 | openHostCleanup func(host string) (*cgroup.Cleanup, error) |
| 287 | // strictPreflight is NextSpawnStrict's seam for PrepareStrict's runtime |
| 288 | // preflight. Nil: Cgroups.StrictPreflight, the very function |
| 289 | // PrepareStrict runs, so S2 and the spawn cannot disagree. |
| 290 | strictPreflight func() error |
| 291 | // strictGates holds one hostGate per host that has spawned with Strict |
| 292 | // set. strictMu guards only the map's lookup and insert, never a start |
| 293 | // or any filesystem work, so a stalled start blocks only its own host. |
| 294 | strictMu sync.Mutex |
| 295 | strictGates map[string]*hostGate |
| 296 | // strictHook, when set, is called at named points of the strict |
| 297 | // cleanup path. Test-only: it makes interleavings deterministic. |
| 298 | strictHook func(point string) |
| 299 | // launchHook, when set, is called by a staged spawn once its |
| 300 | // preparation is done, immediately before its exec gate (runExecGate): |
| 301 | // flat, inside the cgroup manager's before-start hook, so once per |
| 302 | // start attempt (a PID-move retry asks again); nested, before root's |
| 303 | // start call. Test-only: it lands a lease loss, a fence or a |
| 304 | // supersession after the preparation and before the launch (final |
| 305 | // review F1). |
| 306 | launchHook func(host string) |
| 307 | } |
| 308 | |
| 309 | // hostGate serialises everything that touches one host's group once Strict |
| 310 | // is wired (moves design §5.3): the ownership check and an ordinary start, |
| 311 | // a strict spawn taking ownership, the exit path's directory removal, a |
| 312 | // retry's removal, and S2's read. owner, and owner.pending, are read and |
| 313 | // written only with mu held. While owner is set, from a strict spawn until |
| 314 | // its cleanup succeeds (retries included), no other spawn of the host may |
| 315 | // use the group and nothing but the owner may remove it. |
| 316 | type hostGate struct { |
| 317 | mu sync.Mutex |
| 318 | owner *strictOwner |
| 319 | } |
| 320 | |
| 321 | // gate returns host's gate, creating it. |
| 322 | func (b *ExecBackend) gate(host string) *hostGate { |
| 323 | b.strictMu.Lock() |
| 324 | defer b.strictMu.Unlock() |
| 325 | if b.strictGates == nil { |
| 326 | b.strictGates = map[string]*hostGate{} |
| 327 | } |
| 328 | g := b.strictGates[host] |
| 329 | if g == nil { |
| 330 | g = &hostGate{} |
| 331 | b.strictGates[host] = g |
| 332 | } |
| 333 | return g |
| 334 | } |
| 335 | |
| 336 | // lookupGate returns host's gate, or nil if it never had one. |
| 337 | func (b *ExecBackend) lookupGate(host string) *hostGate { |
| 338 | b.strictMu.Lock() |
| 339 | defer b.strictMu.Unlock() |
| 340 | return b.strictGates[host] |
| 341 | } |
| 342 | |
| 343 | func (b *ExecBackend) hook(point string) { |
| 344 | if b.strictHook != nil { |
| 345 | b.strictHook(point) |
| 346 | } |
| 347 | } |
| 348 | |
| 349 | // ErrStrictGroupInUse is a spawn refused because a strict child of the same |
| 350 | // host is alive in the host's group: an ordinary child placed beside it |
| 351 | // would be killed by its exit drain. |
| 352 | var ErrStrictGroupInUse = errors.New("activator: a strict child of this host is still running in its group") |
| 353 | |
| 354 | // ErrStrictCleanupPending is a spawn refused because the host's last strict |
| 355 | // group has not been cleaned up: its retained handle still owns the group, |
| 356 | // and a retry's drain would kill anything placed in it. |
| 357 | var ErrStrictCleanupPending = errors.New("activator: this host's last strict group is still being cleaned up") |
| 358 | |
| 359 | // ErrStrictRequired is a spawn refused because its host is in a move and |
| 360 | // the child could not be started under strict containment (moves design |
| 361 | // §5.3): no delegated cgroup subtree, a populated or unverifiable group, or |
| 362 | // a strict start that failed. Nothing was spawned. |
| 363 | var ErrStrictRequired = errors.New("activator: a host in a move spawns only under strict containment") |
| 364 | |
| 365 | // strictProc is what a strict spawn holds for the life of its child. |
| 366 | // *cgroup.Prepared satisfies it on every build (strict_linux.go, |
| 367 | // strict_other.go). |
| 368 | type strictProc interface { |
| 369 | Start(*exec.Cmd) error |
| 370 | KillAndEmpty(context.Context) error |
| 371 | Close() error |
| 372 | } |
| 373 | |
| 374 | var _ strictProc = (*cgroup.Prepared)(nil) |
| 375 | |
| 376 | // strictDrainTimeout bounds each KillAndEmpty a strict group's cleanup runs. |
| 377 | var strictDrainTimeout = 30 * time.Second |
| 378 | |
| 379 | // strictCleanupBackoff is the bounded retry schedule for a strict group |
| 380 | // whose cleanup failed on exit: one more attempt after each wait. When it |
| 381 | // runs out the handle is still kept, and the host's strict spawns stay |
| 382 | // refused until the agent restarts: a group that never drained may still |
| 383 | // hold a process nothing accounts for. |
| 384 | var strictCleanupBackoff = []time.Duration{time.Second, 5 * time.Second, 15 * time.Second, 30 * time.Second, time.Minute, 2 * time.Minute, 5 * time.Minute, 10 * time.Minute} |
| 385 | |
| 386 | // execLogLineMax caps one teed log line (spec §5): longer is truncated with |
| 387 | // a marker, never split into phantom lines. |
| 388 | const execLogLineMax = 8 << 10 |
| 389 | |
| 390 | // execDrainGrace bounds how long a dead child's held-open output pipes can |
| 391 | // delay its Wait. Teeing turns the child's stdout/stderr into PIPES (with a |
| 392 | // plain *os.File the fd passed straight through), and os/exec's Wait then |
| 393 | // waits for the read side to drain — which a grandchild holding the write |
| 394 | // end keeps open forever. Same discipline, same reason as seedDrainGrace; |
| 395 | // without it, an exited instance would look alive to the activator for as |
| 396 | // long as some stray grandchild lived, and Stop would hang on p.done. |
| 397 | var execDrainGrace = 5 * time.Second |
| 398 | |
| 399 | // teeStd wires cmd's stdout/stderr: straight to the agent's own streams |
| 400 | // (journald under systemd) exactly as before, PLUS — only when b.Logs is |
| 401 | // set — a line-splitting tee into the log sink tagged as rt's app output. |
| 402 | func (b *ExecBackend) teeStd(cmd *exec.Cmd, rt registry.Route) { |
| 403 | if b.Logs == nil { |
| 404 | cmd.Stdout = os.Stdout |
| 405 | cmd.Stderr = os.Stderr |
| 406 | return |
| 407 | } |
| 408 | acct, app := release.RouteAccount(rt), rt.App |
| 409 | emit := func(line string) { b.Logs.App(acct, app, rt.Host, line) } |
| 410 | cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit)) |
| 411 | cmd.Stderr = io.MultiWriter(os.Stderr, applog.NewLineWriter(execLogLineMax, emit)) |
| 412 | cmd.WaitDelay = execDrainGrace |
| 413 | } |
| 414 | |
| 415 | var _ Backend = (*ExecBackend)(nil) |
| 416 | |
| 417 | // tier is Tier with the nil default applied. |
| 418 | func (b *ExecBackend) tier(rt registry.Route) cgroup.Limit { |
| 419 | if b.Tier == nil { |
| 420 | return cgroup.TierDefault |
| 421 | } |
| 422 | return b.Tier(rt) |
| 423 | } |
| 424 | |
| 425 | // spawnInCgroup is the ONE seam all three child shapes — seed, instance, |
| 426 | // sidecar — start through: the route's cgroup, the route's tier. A spawn |
| 427 | // site that bypassed it would compile and silently ship uncapped children; |
| 428 | // keeping the tier plumbing here makes that a diff a reviewer can see and |
| 429 | // TestSpawnInCgroupAppliesTheTier can pin. |
| 430 | // |
| 431 | // Under a move (Strict) it is also the one place a spawn is made strict, so |
| 432 | // no child shape can start a host's process outside the strict group. The |
| 433 | // returned settle runs after the child has been reaped: for a strict spawn |
| 434 | // it drains and closes the group, for a plain one it does nothing. |
| 435 | // |
| 436 | // A strict handle owns the host's group from the spawn until its cleanup |
| 437 | // succeeds (hostGate.owner). While it does, every other spawn of the host |
| 438 | // is refused, an ordinary one included: the check and an ordinary spawn's |
| 439 | // start happen under the host's gate, so no strict group can be prepared |
| 440 | // for the host in between. With Strict nil (a box with no moves wiring) |
| 441 | // none of this runs and no lock is taken: the spawn is main's exactly. |
| 442 | func (b *ExecBackend) spawnInCgroup(rt registry.Route, build func() *exec.Cmd) (*exec.Cmd, func(), error) { |
| 443 | return b.spawnInCgroupGated(rt, cgroup.Limit{}, build, nil) |
| 444 | } |
| 445 | |
| 446 | // spawnInCgroupGated is spawnInCgroup with a before-start hook (final review |
| 447 | // F1), asked immediately before each start attempt: the cgroup manager's |
| 448 | // (StartGated), or, strict, after the strict group is prepared. A refusal |
| 449 | // starts nothing and is cleaned up as a failed start is. A nil before is |
| 450 | // spawnInCgroup exactly. |
| 451 | // |
| 452 | // tier is the artifact's own (a recorded or staged build's, ruling |
| 453 | // R-TIER); zero is b.tier(rt), the live link's, which is every caller but |
| 454 | // a staged spawn. |
| 455 | func (b *ExecBackend) spawnInCgroupGated(rt registry.Route, tier cgroup.Limit, build func() *exec.Cmd, before func() error) (*exec.Cmd, func(), error) { |
| 456 | if b.Cfg.HostWrites.Enabled { |
| 457 | return nil, nil, ErrClaimIsolationUnavailable |
| 458 | } |
| 459 | if tier == (cgroup.Limit{}) { |
| 460 | tier = b.tier(rt) |
| 461 | } |
| 462 | if b.Strict == nil { |
| 463 | cmd, err := b.Cgroups.StartGated(rt.Host, tier, build, before) |
| 464 | return cmd, func() {}, err |
| 465 | } |
| 466 | strict := b.Strict(rt.Host) |
| 467 | g := b.gate(rt.Host) |
| 468 | g.mu.Lock() |
| 469 | if o := g.owner; o != nil { |
| 470 | pending := o.pending |
| 471 | g.mu.Unlock() |
| 472 | cause := ErrStrictGroupInUse |
| 473 | if pending { |
| 474 | cause = ErrStrictCleanupPending |
| 475 | } |
| 476 | if strict { |
| 477 | return nil, nil, fmt.Errorf("%s: %w: %w", rt.Host, ErrStrictRequired, cause) |
| 478 | } |
| 479 | return nil, nil, fmt.Errorf("%s: %w", rt.Host, cause) |
| 480 | } |
| 481 | if !strict { |
| 482 | cmd, err := b.Cgroups.StartGated(rt.Host, tier, build, before) |
| 483 | g.mu.Unlock() |
| 484 | return cmd, func() {}, err |
| 485 | } |
| 486 | o := &strictOwner{} |
| 487 | g.owner = o |
| 488 | g.mu.Unlock() |
| 489 | sp, err := b.prepareStrictProc(rt.Host, tier) |
| 490 | if err != nil { |
| 491 | b.disown(rt.Host, o, false) |
| 492 | return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err) |
| 493 | } |
| 494 | o.sp = sp |
| 495 | cmd := build() |
| 496 | if before != nil { |
| 497 | if err := before(); err != nil { |
| 498 | // Nothing started: the strict group goes as after a failed start. |
| 499 | b.settleStrict(rt.Host, o) |
| 500 | return nil, nil, err |
| 501 | } |
| 502 | } |
| 503 | if err := sp.Start(cmd); err != nil { |
| 504 | b.settleStrict(rt.Host, o) |
| 505 | return nil, nil, fmt.Errorf("%s: %w: %v", rt.Host, ErrStrictRequired, err) |
| 506 | } |
| 507 | return cmd, func() { b.settleStrict(rt.Host, o) }, nil |
| 508 | } |
| 509 | |
| 510 | // prepareStrictProc is prepareStrict with its default applied. |
| 511 | func (b *ExecBackend) prepareStrictProc(host string, limit cgroup.Limit) (strictProc, error) { |
| 512 | if b.prepareStrict != nil { |
| 513 | return b.prepareStrict(host, limit) |
| 514 | } |
| 515 | if b.Cgroups == nil { |
| 516 | return nil, errors.New("no delegated cgroup subtree") |
| 517 | } |
| 518 | p, err := b.Cgroups.PrepareStrict(host, limit) |
| 519 | if err != nil { |
| 520 | return nil, err |
| 521 | } |
| 522 | return p, nil |
| 523 | } |
| 524 | |
| 525 | // strictOwner is the strict handle that owns a host's group. pending |
| 526 | // (guarded by the host gate's mu) is set once its first cleanup attempt |
| 527 | // failed and retries own the group. drained records a KillAndEmpty that |
| 528 | // already succeeded: the handle refuses a second one, so a retry after a |
| 529 | // failed Close only closes. sp and drained are touched only by the one |
| 530 | // goroutine running the owner's cleanup at a time. |
| 531 | type strictOwner struct { |
| 532 | sp strictProc |
| 533 | pending bool |
| 534 | drained bool |
| 535 | } |
| 536 | |
| 537 | // attempt is one cleanup attempt: drain (unless already drained), then |
| 538 | // close. It runs outside the host gate (a drain may take strictDrainTimeout); |
| 539 | // the owner it belongs to keeps every other user of the group out meanwhile. |
| 540 | func (o *strictOwner) attempt(host string) error { |
| 541 | if !o.drained { |
| 542 | ctx, cancel := context.WithTimeout(context.Background(), strictDrainTimeout) |
| 543 | err := o.sp.KillAndEmpty(ctx) |
| 544 | cancel() |
| 545 | if err != nil { |
| 546 | log.Printf("activator: %s: strict group did not drain: %v", host, err) |
| 547 | return err |
| 548 | } |
| 549 | o.drained = true |
| 550 | } |
| 551 | if err := o.sp.Close(); err != nil { |
| 552 | log.Printf("activator: %s: strict group not closed: %v", host, err) |
| 553 | return err |
| 554 | } |
| 555 | return nil |
| 556 | } |
| 557 | |
| 558 | // settleStrict drains a strict child's group and closes the handle. On |
| 559 | // success the owner lets go and the exit path's cleanupHost removes the |
| 560 | // group as before. On a failure the owner stays, marked pending, retried on |
| 561 | // strictCleanupBackoff: until a retry succeeds every spawn of the host is |
| 562 | // refused and only the owner may remove the group directory, because the |
| 563 | // handle's identity is that directory and removing it would strand the |
| 564 | // handle for ever. |
| 565 | func (b *ExecBackend) settleStrict(host string, o *strictOwner) { |
| 566 | if o.attempt(host) == nil { |
| 567 | b.disown(host, o, false) |
| 568 | return |
| 569 | } |
| 570 | b.hook("cleanup-failed") |
| 571 | g := b.gate(host) |
| 572 | g.mu.Lock() |
| 573 | o.pending = true |
| 574 | g.mu.Unlock() |
| 575 | go b.retryStrictCleanup(host, o, append([]time.Duration(nil), strictCleanupBackoff...)) |
| 576 | } |
| 577 | |
| 578 | // disown ends o's ownership of host's group. removeGroup removes the group |
| 579 | // directory first, under the host gate, so no spawn can recreate it in |
| 580 | // between: the retry owner's last act. |
| 581 | func (b *ExecBackend) disown(host string, o *strictOwner, removeGroup bool) { |
| 582 | g := b.gate(host) |
| 583 | g.mu.Lock() |
| 584 | defer g.mu.Unlock() |
| 585 | if g.owner != o { |
| 586 | return |
| 587 | } |
| 588 | if removeGroup { |
| 589 | b.Cgroups.CleanupHost(host) |
| 590 | } |
| 591 | g.owner = nil |
| 592 | } |
| 593 | |
| 594 | // cleanupHost is the exit path's CleanupHost. With Strict nil it is main's |
| 595 | // call, with no lock. Otherwise it runs under the host gate and removes |
| 596 | // nothing while any strict owner, live or pending, holds the group: the |
| 597 | // owner removes it once its handle is closed. |
| 598 | func (b *ExecBackend) cleanupHost(host string) { |
| 599 | if b.Strict == nil { |
| 600 | b.Cgroups.CleanupHost(host) |
| 601 | return |
| 602 | } |
| 603 | g := b.gate(host) |
| 604 | g.mu.Lock() |
| 605 | defer g.mu.Unlock() |
| 606 | if g.owner != nil { |
| 607 | return |
| 608 | } |
| 609 | b.hook("exit-cleanup-checked") |
| 610 | b.Cgroups.CleanupHost(host) |
| 611 | } |
| 612 | |
| 613 | // retryStrictCleanup retries o's cleanup once after each wait in schedule, |
| 614 | // a copy taken when the cleanup was queued, and removes the group directory |
| 615 | // when one succeeds. |
| 616 | func (b *ExecBackend) retryStrictCleanup(host string, o *strictOwner, schedule []time.Duration) { |
| 617 | for i, wait := range schedule { |
| 618 | time.Sleep(wait) |
| 619 | if o.attempt(host) == nil { |
| 620 | b.disown(host, o, true) |
| 621 | log.Printf("activator: %s: strict group cleaned up on retry %d", host, i+1) |
| 622 | return |
| 623 | } |
| 624 | } |
| 625 | log.Printf("activator: %s: strict group still not cleaned up after %d retries; spawns of this host stay refused until the agent restarts", host, len(schedule)) |
| 626 | } |
| 627 | |
| 628 | // strictCleanupPending reports, under the host gate, whether host's group |
| 629 | // is owned by a strict handle whose cleanup is being retried. |
| 630 | func (b *ExecBackend) strictCleanupPending(host string) bool { |
| 631 | g := b.lookupGate(host) |
| 632 | if g == nil { |
| 633 | return false |
| 634 | } |
| 635 | g.mu.Lock() |
| 636 | defer g.mu.Unlock() |
| 637 | return g.owner != nil && g.owner.pending |
| 638 | } |
| 639 | |
| 640 | // NextSpawnStrict is the source's S2 input (moves design §7.3, §5.3): will |
| 641 | // host's next spawn be strict? OK only when Strict names the host, the box |
| 642 | // has a delegated cgroup subtree, PrepareStrict's own runtime preflight |
| 643 | // passes (the agent placed in that subtree or its supervise child, on a |
| 644 | // real cgroup2 mount, on a build that supports it), and no earlier strict |
| 645 | // group of the host awaits cleanup. The per-host group state PrepareStrict |
| 646 | // also judges (a populated group) is the idle stop's to clear. |
| 647 | func (b *ExecBackend) NextSpawnStrict(host string) moves.Check { |
| 648 | switch { |
| 649 | case b.Strict == nil || !b.Strict(host): |
| 650 | return moves.Check{Reason: "S2: the next spawn of this host is not strict"} |
| 651 | case b.Cgroups.Root() == "": |
| 652 | return moves.Check{Reason: "S2: this box has no delegated cgroup subtree, so it cannot spawn strictly"} |
| 653 | case b.strictCleanupPending(host): |
| 654 | return moves.Check{Reason: "S2: this host's last strict group has not been cleaned up yet"} |
| 655 | } |
| 656 | preflight := b.strictPreflight |
| 657 | if preflight == nil { |
| 658 | preflight = b.Cgroups.StrictPreflight |
| 659 | } |
| 660 | if err := preflight(); err != nil { |
| 661 | if errors.Is(err, cgroup.ErrStrictRefused) { |
| 662 | return moves.Check{Reason: "S2: the agent is not placed in its delegated cgroup subtree, so it cannot spawn strictly"} |
| 663 | } |
| 664 | return moves.Check{Reason: "S2: this box cannot spawn strictly"} |
| 665 | } |
| 666 | return moves.Check{OK: true} |
| 667 | } |
| 668 | |
| 669 | func (b *ExecBackend) liveDir() string { return filepath.Join(b.OptDir, "live") } |
| 670 | func (b *ExecBackend) liveLink(host string) string { return filepath.Join(b.liveDir(), host) } |
| 671 | |
| 672 | // LiveVersion reports the version host's live symlink currently points at, or |
| 673 | // "" (no error) if the host hasn't been put under version management yet. It |
| 674 | // doubles as the Service's convergence-sweep seam — New picks it up via an |
| 675 | // optional interface. |
| 676 | // |
| 677 | // Delegated to release.LiveVersion rather than reading the parent directory, |
| 678 | // because the version store gained an account level: a migrated tenant's link |
| 679 | // is <opt>/versions/<ver>/<acct>/<app>, and "the parent dir" there is the |
| 680 | // account SQID. ConvergeDecision compares this against the route's version, |
| 681 | // so returning a sqid would convince it that every migrated tenant was |
| 682 | // converged forever — a tenant that silently stops taking deploys. One |
| 683 | // definition of "which segment is the version", in one place. |
| 684 | func (b *ExecBackend) LiveVersion(host string) (string, error) { |
| 685 | return release.LiveVersion(b.OptDir, host) |
| 686 | } |
| 687 | |
| 688 | // Acquire claims the per-host single-writer lease. A contested lease is |
| 689 | // re-surfaced as errLeaseHeld (never a litestream type — the lease package's |
| 690 | // *HeldError is this package's only view of it) so the edge can 503 with a |
| 691 | // Retry-After that tracks the holder's expiry. |
| 692 | func (b *ExecBackend) Acquire(ctx context.Context, rt registry.Route) (Lease, error) { |
| 693 | host := rt.Host |
| 694 | // Admission before the lease: the instance's config has to be |
| 695 | // providable from THIS box, or it must not be the one that starts it. |
| 696 | // Here, not in doWake, because this is the only place an instance or |
| 697 | // sidecar lease is taken; every wake source reaches it. |
| 698 | if err := b.AdmitConfig(ctx, rt); err != nil { |
| 699 | return nil, err |
| 700 | } |
| 701 | held, err := lease.Acquire(ctx, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region) |
| 702 | if err != nil { |
| 703 | var he *lease.HeldError |
| 704 | if errors.As(err, &he) { |
| 705 | retry := time.Until(he.ExpiresAt) |
| 706 | if retry < time.Second { |
| 707 | retry = time.Second |
| 708 | } |
| 709 | return nil, &errLeaseHeld{holder: he.Owner, retryAfter: retry} |
| 710 | } |
| 711 | return nil, err |
| 712 | } |
| 713 | return &execLease{held: held, defaultDB: b.pathsFor(rt).dbPath}, nil |
| 714 | } |
| 715 | |
| 716 | // AdmitConfig asks the config gate whether this box can provide rt's config: |
| 717 | // nil means yes (or no gate), an *errConfigRefused means no. Acquire calls it |
| 718 | // before the lease; the activator calls it again before an in-place worker |
| 719 | // respawn, which keeps its lease and so never passes through Acquire. |
| 720 | func (b *ExecBackend) AdmitConfig(ctx context.Context, rt registry.Route) error { |
| 721 | if b.ConfigGate == nil { |
| 722 | return nil |
| 723 | } |
| 724 | if err := b.ConfigGate(ctx, rt); err != nil { |
| 725 | return &errConfigRefused{err: err} |
| 726 | } |
| 727 | return nil |
| 728 | } |
| 729 | |
| 730 | // execLease adapts *lease.Held to the Lease seam. |
| 731 | type execLease struct { |
| 732 | held *lease.Held |
| 733 | defaultDB string // used when the route carries no DBPath |
| 734 | } |
| 735 | |
| 736 | func (l *execLease) Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (Rep, error) { |
| 737 | if dbPath == "" { |
| 738 | dbPath = l.defaultDB |
| 739 | } |
| 740 | rep, err := l.held.Replicate(ctx, dbPath, syncInterval) |
| 741 | if err != nil { |
| 742 | return nil, err |
| 743 | } |
| 744 | return rep, nil |
| 745 | } |
| 746 | |
| 747 | // The Rep execLease.Replicate hands back is lease.Held's *lease.Replicator, |
| 748 | // and SeedPlaced records a seed placed only through its SyncAndWait. Pinned |
| 749 | // here so a change to either fails the build, not every seeded wake. |
| 750 | var ( |
| 751 | _ func(*lease.Held, context.Context, string, time.Duration) (*lease.Replicator, error) = (*lease.Held).Replicate |
| 752 | _ interface{ SyncAndWait(context.Context) error } = (*lease.Replicator)(nil) |
| 753 | ) |
| 754 | |
| 755 | func (l *execLease) Release(ctx context.Context) error { return l.held.Release(ctx) } |
| 756 | |
| 757 | // RetryRelease is releaseRetrier: the held lease's conditional retry of a |
| 758 | // release that failed (#292). |
| 759 | func (l *execLease) RetryRelease(ctx context.Context) error { return l.held.RetryRelease(ctx) } |
| 760 | |
| 761 | // Err is the held lease's liveness (leaseHealth): a failed renewal, a |
| 762 | // spent handle or a poisoned one. |
| 763 | func (l *execLease) Err() error { return l.held.Err() } |
| 764 | |
| 765 | // Restore puts the row's database in place for a wake. A row whose seed is |
| 766 | // placeable (ready, swapped or placed) or committed goes through the seed |
| 767 | // fallback (seedplace.go), which never reports fresh; every other row |
| 768 | // restores plainly. A seed column that does not parse fails the wake: |
| 769 | // reading it as "no seed" would boot a seeded instance empty. |
| 770 | func (b *ExecBackend) Restore(ctx context.Context, rt registry.Route) (bool, error) { |
| 771 | src, err := seeds.ParseRowSeed(rt.SeedSource) |
| 772 | if err != nil { |
| 773 | return false, err |
| 774 | } |
| 775 | if seedRestores(src) { |
| 776 | return b.seededRestore(ctx, rt, src) |
| 777 | } |
| 778 | return b.restorePlain(ctx, rt) |
| 779 | } |
| 780 | |
| 781 | // restorePlain is Restore for a row with no seed to place: the warm cache, |
| 782 | // then the staging fallback, then the host's own replica, and fresh only |
| 783 | // when there is none of the three. |
| 784 | func (b *ExecBackend) restorePlain(ctx context.Context, rt registry.Route) (bool, error) { |
| 785 | host, dbPath := rt.Host, b.pathsFor(rt).dbPath |
| 786 | // A failed seed whose processes were not proven gone left its database |
| 787 | // and an obligation (Task 8): it is answered before the database can |
| 788 | // count as a warm cache, and before the marker check below can purge it. |
| 789 | if err := b.settleSeedObligation(ctx, rt, dbPath); err != nil { |
| 790 | return false, err |
| 791 | } |
| 792 | if err := reconcileSeedMarker(rt, dbPath); err != nil { |
| 793 | return false, err |
| 794 | } |
| 795 | if _, err := os.Stat(dbPath); err == nil { |
| 796 | return false, nil // warm cache — see doWake |
| 797 | } else if !os.IsNotExist(err) { |
| 798 | return false, err |
| 799 | } |
| 800 | // The DB is gone, so any leftover litestream sidecar state (meta dirs, |
| 801 | // WAL/SHM) describes a database that no longer exists — litestream's own |
| 802 | // recovery guidance is to delete it. Clear it so the restored copy starts |
| 803 | // a clean lineage. Two meta dirs: litestream's default suffix ("-litestream", |
| 804 | // its MetaDirSuffix — spelled out because this package never imports |
| 805 | // litestream) as any box-wide backup replicator would use, and the lease |
| 806 | // package's activator-scoped ".activator-<db>-litestream". |
| 807 | PurgeLocalDBArtifacts(dbPath, false) |
| 808 | |
| 809 | // The staging fallback (issue #331): when a move staged a replica |
| 810 | // here, promote it (server-side, own bucket) and restore from the |
| 811 | // promoted copy. Staging wins when present (promoteStagedReplica, |
| 812 | // review round 1): a staged copy exists only while an inbound move is |
| 813 | // landing, so it is newer than any lineage the own prefix kept from an |
| 814 | // earlier stay on this box (the legacy cut does not delete the own |
| 815 | // prefix it leaves behind), and the promotion's mirror replaces it. |
| 816 | var promoted map[string]lease.MirrorObject |
| 817 | if b.Mirror != nil && !b.Phase1 { |
| 818 | var err error |
| 819 | if promoted, err = b.promoteStagedReplica(ctx, host); err != nil { |
| 820 | // HARD failure, same rule as restoreReplica's own error paths |
| 821 | // (review round 1): "continuing without it" here would fall |
| 822 | // through to a FRESH boot + seed, and once that empty database |
| 823 | // replicates, the staged data is shadowed forever. A failed |
| 824 | // wake retries; a seeded ghost does not. |
| 825 | return false, fmt.Errorf("staging fallback for %s: %w", host, err) |
| 826 | } |
| 827 | } |
| 828 | |
| 829 | restored, err := restoreReplica(ctx, dbPath, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), b.Cfg.Region) |
| 830 | if err != nil { |
| 831 | // "We couldn't tell" is never reported as "there was nothing there": |
| 832 | // fresh stays false on every error path, so a bucket outage can't |
| 833 | // route a live instance into the seed. |
| 834 | return false, err |
| 835 | } |
| 836 | if restored { |
| 837 | // Prove the restored file before anything serves from it. A replica |
| 838 | // can come back damaged — a torn page, a truncated object, a lineage |
| 839 | // that never had a snapshot — and the alternatives to failing here are |
| 840 | // both worse than a down route: serving a corrupt database, or (if |
| 841 | // this returned fresh) running the route's seed over recoverable data. |
| 842 | if err := lease.CheckDB(dbPath); err != nil { |
| 843 | q := dbPath + quarantineSuffix |
| 844 | _ = os.Remove(q) // keep exactly one: the newest failure is the evidence |
| 845 | if mvErr := os.Rename(dbPath, q); mvErr != nil { |
| 846 | // Quarantine failed, so the live path may still hold the bad |
| 847 | // file — remove it rather than leave a warm cache that the |
| 848 | // next wake would boot with no check at all. |
| 849 | _ = os.Remove(dbPath) |
| 850 | log.Printf("activator: %s restored CORRUPT (%v) and could not be quarantined (%v) — removed", host, err, mvErr) |
| 851 | } else { |
| 852 | log.Printf("activator: %s restored CORRUPT from s3://%s/%s: %v — quarantined at %s, instance NOT started", |
| 853 | host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), err, q) |
| 854 | } |
| 855 | for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| 856 | _ = os.Remove(p) |
| 857 | } |
| 858 | // The next wake re-attempts the restore rather than refusing on |
| 859 | // sight of the quarantined file: a replaced or repaired replica |
| 860 | // must recover with no operator action. The price is a repeated |
| 861 | // restore for as long as the replica stays broken, which is the |
| 862 | // right way round — a permanently corrupt replica is an alarm |
| 863 | // someone is answering, not a steady state to optimise for. |
| 864 | return false, fmt.Errorf("restored replica for %s is corrupt: %w", host, err) |
| 865 | } |
| 866 | log.Printf("activator: restored %s from s3://%s/%s", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| 867 | // A clean restore of a copy promoted THIS call is the moment the |
| 868 | // staging has served its purpose; a corrupt one (above) |
| 869 | // deliberately LEFT it in place as evidence, and staging that was |
| 870 | // never consulted (an abort's leftovers under a live own lineage) |
| 871 | // is not this restore's to touch. |
| 872 | if len(promoted) > 0 { |
| 873 | b.dropStagedReplica(ctx, promoted) |
| 874 | } |
| 875 | } else { |
| 876 | // Never replicated: a brand-new instance boots fresh and creates its |
| 877 | // own DB — the register-only create path (provisioning as a metadata |
| 878 | // write) landing on its first wake. This — DB absent AND no replica — |
| 879 | // is the ONLY fresh boot, and the one moment the route's seed runs. |
| 880 | log.Printf("activator: no replica for %s in s3://%s/%s — booting fresh", host, b.Cfg.Bucket, b.Cfg.hibernatePrefix(host)) |
| 881 | } |
| 882 | return !restored, nil |
| 883 | } |
| 884 | |
| 885 | // restoreReplica is lease.Restore behind a package var so the fresh-detection |
| 886 | // matrix (replica present / absent / unreachable) is testable without S3. |
| 887 | var restoreReplica = lease.Restore |
| 888 | |
| 889 | // Seed runs the route's first-boot initialization, once, before the instance |
| 890 | // has ever been spawned (doWake calls it only when Restore reported fresh). |
| 891 | // |
| 892 | // THE CONTRACT, FOR APP AUTHORS: the platform invokes your binary with your |
| 893 | // manifest's `seed` argv followed by `--db <path>` — you create your schema |
| 894 | // and starter content in that database and exit 0. (`titogo seed` is the shape |
| 895 | // this was named for.) It runs under the same merged env layers the instance |
| 896 | // itself gets, so the credentials and config in your env file are there; it |
| 897 | // does NOT get the purge-channel vars, which describe a serving process's |
| 898 | // cache and mean nothing to a one-shot init — and it does not get them even |
| 899 | // if your own env file sets them, those two names being the agent's to |
| 900 | // author. A non-zero exit fails the wake: |
| 901 | // a half-seeded instance must not serve. |
| 902 | // Seed dispatches by layout, the same way Start does: a nested row seeds |
| 903 | // through its own launch file and unit, a fenced or conflicting row |
| 904 | // refuses, and a flat or foreign row seeds through today's exec child. |
| 905 | func (b *ExecBackend) Seed(ctx context.Context, rt registry.Route) error { |
| 906 | p := b.pathsFor(rt) |
| 907 | switch p.layout { |
| 908 | case LayoutNested: |
| 909 | return b.seedUnit(ctx, rt, nil, cgroup.Limit{}) |
| 910 | case LayoutFenced, LayoutConflict: |
| 911 | return fmt.Errorf("%s: refusing to seed: %s", rt.Host, p.reason) |
| 912 | } |
| 913 | return b.seedChild(ctx, rt, b.liveLink(rt.Host), cgroup.Limit{}) |
| 914 | } |
| 915 | |
| 916 | // SeedExe is Seed on the identity a wake recorded (settings spec §4.2): |
| 917 | // id.StagedPath is checked against id.Digest first, and a copy that |
| 918 | // changed seeds nothing (never the live link instead). A flat or foreign |
| 919 | // row execs that file by path. A nested row asks root to run the seed unit |
| 920 | // on it, which root validates and binds in place of the live link. The |
| 921 | // seed runs at id's own tier (GenTier, ruling R-TIER), never the tier of |
| 922 | // whatever the live link names by then. |
| 923 | func (b *ExecBackend) SeedExe(ctx context.Context, rt registry.Route, id release.StagedIdentity) error { |
| 924 | if err := release.VerifyStaged(id); err != nil { |
| 925 | return fmt.Errorf("%s: not seeded with %s: %w", rt.Host, id.Version, err) |
| 926 | } |
| 927 | tier := b.GenTier(rt, id.Version) |
| 928 | p := b.pathsFor(rt) |
| 929 | switch p.layout { |
| 930 | case LayoutNested: |
| 931 | return b.seedUnit(ctx, rt, stagedExeOf(id), tier) |
| 932 | case LayoutFenced, LayoutConflict: |
| 933 | return fmt.Errorf("%s: refusing to seed: %s", rt.Host, p.reason) |
| 934 | } |
| 935 | return b.seedChild(ctx, rt, id.StagedPath, tier) |
| 936 | } |
| 937 | |
| 938 | // seedArgv parses rt.Seed's JSON argv array and appends the --db flag, the |
| 939 | // composition both spawn paths need identically (design §6.2). An empty |
| 940 | // declared seed (an empty array, len 0) returns a nil argv, which both |
| 941 | // callers treat as "nothing to run" (doWake only calls Seed for a |
| 942 | // non-empty rt.Seed; this stays defensive rather than assuming that). |
| 943 | func seedArgv(rt registry.Route, dbPath string) ([]string, error) { |
| 944 | var seedArgs []string |
| 945 | if err := json.Unmarshal([]byte(rt.Seed), &seedArgs); err != nil { |
| 946 | // Never a silent skip: a mangled seed column must fail the wake |
| 947 | // loudly, not boot an empty instance that looks healthy. |
| 948 | return nil, fmt.Errorf("parse seed argv for %s (%q): %w", rt.Host, rt.Seed, err) |
| 949 | } |
| 950 | if len(seedArgs) == 0 { |
| 951 | return nil, nil |
| 952 | } |
| 953 | return append(seedArgs, "--db", dbPath), nil |
| 954 | } |
| 955 | |
| 956 | // removeSeedDB deletes a seed's own half-built database and its WAL/SHM |
| 957 | // siblings, proven out of reach of its processes: the cleanup both spawn |
| 958 | // paths and the next wake's discharge run when, and only when, the |
| 959 | // database is the seed's own work (seedCreatedTheDB). Shared so the three |
| 960 | // cannot drift on which files "the database" means. A file already gone |
| 961 | // is fine; any other failure is returned, and the directory is synced so |
| 962 | // the removals are durable before the caller removes the obligation. |
| 963 | func removeSeedDB(dbPath string) error { |
| 964 | for _, p := range []string{dbPath, dbPath + "-wal", dbPath + "-shm"} { |
| 965 | if err := os.Remove(p); err != nil && !errors.Is(err, fs.ErrNotExist) { |
| 966 | return fmt.Errorf("remove the seed's %s: %w", p, err) |
| 967 | } |
| 968 | } |
| 969 | if err := syncDir(filepath.Dir(dbPath)); err != nil { |
| 970 | return fmt.Errorf("sync %s after removing the seed's database: %w", filepath.Dir(dbPath), err) |
| 971 | } |
| 972 | return nil |
| 973 | } |
| 974 | |
| 975 | // seedChild is Seed's flat body. exe is the file the seed execs: the live |
| 976 | // link (Seed), which is materialized first if it is missing, or a staged |
| 977 | // path SeedExe verified, which is exec'd by path whatever the link names. |
| 978 | // With LiveExecGate set the live link is pinned instead (declinedLivePin) |
| 979 | // and the file it named is exec'd. tier is that file's own; zero is the |
| 980 | // live tier (b.tier). |
| 981 | func (b *ExecBackend) seedChild(ctx context.Context, rt registry.Route, exe string, tier cgroup.Limit) error { |
| 982 | host := rt.Host |
| 983 | paths := b.pathsFor(rt) |
| 984 | argv, err := seedArgv(rt, paths.dbPath) |
| 985 | if err != nil { |
| 986 | return err |
| 987 | } |
| 988 | if argv == nil { |
| 989 | return nil // defensive: doWake only calls us for a non-empty seed |
| 990 | } |
| 991 | if exe == b.liveLink(host) { |
| 992 | if b.LiveExecGate != nil { |
| 993 | // A declined box seeds the build its gate passed, by path, at |
| 994 | // that build's tier (declinedLivePin), never the link. Before the |
| 995 | // database stat below: a refusal runs nothing and so can never |
| 996 | // count as a failed seed whose database is removed. |
| 997 | id, err := b.declinedLivePin(ctx, rt) |
| 998 | if err != nil { |
| 999 | return err |
| 1000 | } |
| 1001 | exe, tier = id.StagedPath, b.GenTier(rt, id.Version) |
| 1002 | } else if err := b.ensureLiveLink(ctx, host); err != nil { |
| 1003 | return err |
| 1004 | } |
| 1005 | } |
| 1006 | // Mint before the env is read, exactly as Start does — a seed is the FIRST |
| 1007 | // thing a fresh instance runs, so this is usually where the token comes |
| 1008 | // into existence, and instanceEnv's contract is that a seed sees the same |
| 1009 | // environment the instance will. |
| 1010 | if err := EnsureAdminToken(paths.envFile); err != nil { |
| 1011 | return fmt.Errorf("admin token for %s: %w", host, err) |
| 1012 | } |
| 1013 | // The instance key, before the env is read for the same reason: a seed |
| 1014 | // is usually the first process of a fresh instance (L21). |
| 1015 | if _, err := b.ensureInstanceKeyFor(ctx, rt, paths); err != nil { |
| 1016 | return fmt.Errorf("instance key for %s: %w", host, err) |
| 1017 | } |
| 1018 | fileEnv, keyLayer, err := b.instanceEnvLayer(rt) |
| 1019 | if err != nil { |
| 1020 | return err |
| 1021 | } |
| 1022 | // A seed has no control socket, so a replica's seed is gated as a |
| 1023 | // launch without one (ruling 26), on the layer this same read found. |
| 1024 | if err := b.replicaKeyGate(rt, fileEnv, keyLayer, false); err != nil { |
| 1025 | return err |
| 1026 | } |
| 1027 | dbPath := paths.dbPath |
| 1028 | |
| 1029 | // Whether the database existed BEFORE the seed ran decides whether a |
| 1030 | // failed seed may clean up after itself (below). On the wake path it is |
| 1031 | // always absent — Restore reported fresh. |
| 1032 | _, statErr := os.Stat(dbPath) |
| 1033 | seedOwnsDB := seedCreatedTheDB(statErr) |
| 1034 | |
| 1035 | ctx, cancel := context.WithTimeout(ctx, seedTimeout) |
| 1036 | defer cancel() |
| 1037 | var stderr capBuffer |
| 1038 | log.Printf("activator: seeding %s: %s %s", host, exe, strings.Join(argv, " ")) |
| 1039 | // The seed shares the instance's carlos-exec@<host> cgroup: first-boot |
| 1040 | // cost is that instance's work. No CleanupHost here — the instance spawn |
| 1041 | // reuses the dir moments later, and its exit path (or the next agent |
| 1042 | // start's sweep) removes it. |
| 1043 | child, _, perr := b.placeFor(spawnSeed, 0) |
| 1044 | if perr != nil { |
| 1045 | return perr |
| 1046 | } |
| 1047 | // The obligation goes on disk, durably, before the seed's first |
| 1048 | // instruction (Task 8 fix round 1): from here until its processes are |
| 1049 | // proven gone, nothing may treat the database as warm, whatever happens |
| 1050 | // to the seed, its disk or this agent. A write that fails refuses the |
| 1051 | // seed with nothing run. |
| 1052 | if err := recordSeedObligation(host, dbPath, seedObligation{ownsDB: seedOwnsDB, layout: LayoutFlat}); err != nil { |
| 1053 | return err |
| 1054 | } |
| 1055 | cmd, settle, err := b.spawnInCgroupAtTier(rt, child, false, tier, func() *exec.Cmd { |
| 1056 | cmd := exec.CommandContext(ctx, exe, argv...) |
| 1057 | // The control-channel vars are agent-owned in BOTH directions here too |
| 1058 | // (Start's discipline, see withoutVars), and here they are always |
| 1059 | // STRIPPED: the seed gets no control listener of its own, so fileEnv is |
| 1060 | // tenant-supplied and without this a tenant's own env file could hand |
| 1061 | // its seed a CARLOS_CONTROL_SOCKET the agent never opened and a |
| 1062 | // CARLOS_PURGE for a channel that does not exist — and the seed has no |
| 1063 | // serving cache to purge, nor any business setting timers, in any case. |
| 1064 | // This is what makes the contract above true rather than merely |
| 1065 | // intended. |
| 1066 | cmd.Env = childEnv(seedVars(fileEnv, host)) |
| 1067 | // Like the instance child, the seed logs through the agent's own streams |
| 1068 | // (journald under systemd) — and its stderr is ALSO teed into a bounded |
| 1069 | // buffer, because the wake error this returns is the whole diagnosis an |
| 1070 | // operator gets for a host that refuses to come up. With Logs set the |
| 1071 | // seed's output ships to `carlos logs` too: it is app output, and on a |
| 1072 | // failed first wake it is the diagnosis the MEMBER gets. |
| 1073 | cmd.Stdout = os.Stdout |
| 1074 | cmd.Stderr = io.MultiWriter(os.Stderr, &stderr) |
| 1075 | if b.Logs != nil { |
| 1076 | acct, app := release.RouteAccount(rt), rt.App |
| 1077 | emit := func(line string) { b.Logs.App(acct, app, host, line) } |
| 1078 | cmd.Stdout = io.MultiWriter(os.Stdout, applog.NewLineWriter(execLogLineMax, emit)) |
| 1079 | cmd.Stderr = io.MultiWriter(os.Stderr, &stderr, applog.NewLineWriter(execLogLineMax, emit)) |
| 1080 | } |
| 1081 | // WaitDelay is what makes seedTimeout real. Because Stderr is a plain |
| 1082 | // io.Writer (the tee), os/exec pipes it and Wait waits for that pipe to |
| 1083 | // close — and a seed that spawned its own grandchild holds the write end |
| 1084 | // open long after the timeout killed the seed itself. Without this, a |
| 1085 | // wedged seed would hold the lease until the wake's own 5m budget expired. |
| 1086 | cmd.WaitDelay = seedDrainGrace |
| 1087 | return cmd |
| 1088 | }, pendingRecordHook(ctx)) // the wake's record, at the exec (amadan #272) |
| 1089 | spawned := err == nil |
| 1090 | if spawned { |
| 1091 | err = cmd.Wait() |
| 1092 | settle() |
| 1093 | } |
| 1094 | // WaitDelay reports on the SUCCESS path too: a seed that exits 0 after |
| 1095 | // backgrounding something (a helper, a stray `&`) leaves the inherited |
| 1096 | // stderr pipe open, and Wait then returns ErrWaitDelay INSTEAD of nil |
| 1097 | // even though the process exited 0. Treating that as a failure would |
| 1098 | // fail the wake AND delete the database the seed had just correctly |
| 1099 | // built — leaving the host fresh again next wake, i.e. permanently |
| 1100 | // un-wakeable. The exit status is the verdict; the leaked pipe is a |
| 1101 | // log line. |
| 1102 | exitedZero := spawned && (err == nil || (errors.Is(err, exec.ErrWaitDelay) && cmd.ProcessState != nil && cmd.ProcessState.Success())) |
| 1103 | var failure error |
| 1104 | if !exitedZero { |
| 1105 | if errors.Is(ctx.Err(), context.DeadlineExceeded) { |
| 1106 | err = fmt.Errorf("timed out after %s and was killed: %w", seedTimeout, err) |
| 1107 | } |
| 1108 | failure = fmt.Errorf("seeding %s: %w", host, err) |
| 1109 | if tail := strings.TrimSpace(stderr.String()); tail != "" { |
| 1110 | failure = fmt.Errorf("seeding %s: %w: %s", host, err, tail) |
| 1111 | } |
| 1112 | } |
| 1113 | // What the seed left is proven gone before its database is touched |
| 1114 | // (Task 8). CommandContext and Wait speak for the leader only, and a |
| 1115 | // process it forked keeps writing the database, a removed one included. |
| 1116 | // A failed seed's cgroup is killed and proven empty: flag on its role |
| 1117 | // child, flag off carlos-exec@<host>, which holds nothing else yet (a |
| 1118 | // fresh host, before the spawn, under the wake's lease). A successful |
| 1119 | // seed's role child is too (flag on): a helper that outlived the seed |
| 1120 | // would otherwise run on beside the instance, out of reach of any |
| 1121 | // generation kill. Flag off, a successful seed's helper stays in the |
| 1122 | // host cgroup, where the terminal stop's proof (stopHostProven) finds |
| 1123 | // it. A spawn that failed ran nothing. |
| 1124 | if spawned && (failure != nil || child != "") { |
| 1125 | pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| 1126 | perr := b.seedDescendantsGone(pctx, rt, child) |
| 1127 | pcancel() |
| 1128 | if perr != nil { |
| 1129 | // The database is kept, as seedUnitFailed keeps it: removing it |
| 1130 | // under a live writer is the hazard itself. The obligation |
| 1131 | // written ahead stays, so every later wake refuses it as warm |
| 1132 | // until the proof holds (Restore), and the caller leaves the |
| 1133 | // lease to its TTL (ErrSeedUnproven). |
| 1134 | if failure == nil { |
| 1135 | failure = fmt.Errorf("seed for %s: %w: processes remain after a successful seed: %v", host, ErrSeedUnproven, perr) |
| 1136 | } else { |
| 1137 | failure = fmt.Errorf("%w: %w (the seed's processes remain after stopping it; its database was left in place: %v)", failure, ErrSeedUnproven, perr) |
| 1138 | } |
| 1139 | log.Printf("activator: %s: the seed's processes are not proven gone; its database is kept and %s refuses it as warm until they are", host, dbPath+seedUnprovenSuffix) |
| 1140 | return failure |
| 1141 | } |
| 1142 | b.Cgroups.RemoveChild(host, child) // the seed's own child only; no-op for "" (flag off) |
| 1143 | } |
| 1144 | // From here the seed's processes are proven gone (or a spawn that |
| 1145 | // failed ran nothing, or a flag-off success's helper is the terminal |
| 1146 | // stop's to prove), so the obligation is discharged. A failure takes |
| 1147 | // the half-built database with it first. Failing the wake is only half |
| 1148 | // the promise: the file a crashed seed left behind would look like a |
| 1149 | // warm cache to the NEXT wake (DB present, so not fresh, so no seed) and |
| 1150 | // the instance would boot on it, which is the half-seeded instance this |
| 1151 | // whole path exists to prevent. Only a database that wasn't there when |
| 1152 | // we started is ours to remove, and under the lease, with the seed's |
| 1153 | // processes proven gone above, nothing else can have written it. A |
| 1154 | // discharge that fails keeps the obligation and fails the wake while |
| 1155 | // nothing serves; the next wake proves and removes again. A successful |
| 1156 | // flag-off seed's failed discharge first proves its processes gone, since |
| 1157 | // no instance will (ErrSeedUnproven otherwise). |
| 1158 | if derr := dischargeSeedObligation(dbPath, failure != nil && seedOwnsDB); derr != nil { |
| 1159 | unprovedSuccess := spawned && failure == nil && child == "" // the proof above was skipped |
| 1160 | if failure == nil { |
| 1161 | failure = fmt.Errorf("seed for %s: %w", host, derr) |
| 1162 | } else { |
| 1163 | failure = fmt.Errorf("%w; and %w", failure, derr) |
| 1164 | } |
| 1165 | if unprovedSuccess { |
| 1166 | // A successful flag-off seed has no proof of its own: its helper |
| 1167 | // is left for the instance's terminal stop to prove. With the |
| 1168 | // discharge failed the wake fails, so no instance will run to |
| 1169 | // prove it, and releasing the lease over it would be the hazard |
| 1170 | // itself (fix wave 1, item 2): the proof is tried now. Unproven, |
| 1171 | // the seed is ErrSeedUnproven and the lease is left to its TTL. |
| 1172 | pctx, pcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| 1173 | perr := b.seedDescendantsGone(pctx, rt, child) |
| 1174 | pcancel() |
| 1175 | if perr != nil { |
| 1176 | log.Printf("activator: %s: the seed's obligation could not be discharged and its processes are not proven gone; %s stays", host, dbPath+seedUnprovenSuffix) |
| 1177 | return fmt.Errorf("%w: %w: the seed's processes are not proven gone: %v", failure, ErrSeedUnproven, perr) |
| 1178 | } |
| 1179 | } |
| 1180 | return failure |
| 1181 | } |
| 1182 | if failure == nil { |
| 1183 | if err != nil { |
| 1184 | log.Printf("activator: seeded %s (it exited 0 but left a child holding its output pipe open for more than %s)", host, seedDrainGrace) |
| 1185 | } else { |
| 1186 | log.Printf("activator: seeded %s", host) |
| 1187 | } |
| 1188 | return nil |
| 1189 | } |
| 1190 | return failure |
| 1191 | } |
| 1192 | |
| 1193 | // seedDescendantsGone kills what a flat seed left and proves it gone: flag |
| 1194 | // on, its own role child (child); flag off, carlos-exec@<host> (""), which |
| 1195 | // holds nothing else yet: a fresh host, before the spawn, under the wake's |
| 1196 | // lease. An absent cgroup is empty. A box with no delegated cgroup |
| 1197 | // (cgroup.ErrNoDelegation) ran the seed unaccounted, with nothing that |
| 1198 | // could reach a descendant, so the leader's exit is the only proof there |
| 1199 | // is (ledger 127): it logs and answers nil. So is a kernel with no |
| 1200 | // cgroup.kill (cgroup.ErrNoKill). Any other failure is returned, and the |
| 1201 | // caller keeps the database (ErrSeedUnproven). |
| 1202 | func (b *ExecBackend) seedDescendantsGone(ctx context.Context, rt registry.Route, child string) error { |
| 1203 | var err error |
| 1204 | switch { |
| 1205 | case b.cgroupKill != nil: |
| 1206 | err = b.cgroupKill(ctx, rt, child) |
| 1207 | case child == "": |
| 1208 | err = b.KillHost(ctx, rt) |
| 1209 | default: |
| 1210 | err = b.killChild(ctx, rt.Host, child) |
| 1211 | } |
| 1212 | if errors.Is(err, cgroup.ErrNoDelegation) { |
| 1213 | log.Printf("activator: %s: the seed ran with no delegated cgroup; its leader's exit is the only proof there is", rt.Host) |
| 1214 | return nil |
| 1215 | } |
| 1216 | if errors.Is(err, cgroup.ErrNoKill) { |
| 1217 | // A kernel before 5.14 can kill nothing in a cgroup (EmptyHost's |
| 1218 | // rule): read as unproven, an obligation could never be discharged |
| 1219 | // on such a box and the row would stay down for good. |
| 1220 | log.Printf("activator: %s: this kernel has no cgroup.kill; the seed's leader's exit is the only proof there is", rt.Host) |
| 1221 | return nil |
| 1222 | } |
| 1223 | return err |
| 1224 | } |
| 1225 | |
| 1226 | // seedObligation is what <db>.seed-unproven records, written ahead of a |
| 1227 | // seed (Task 8 fix round 1): before the seed's first instruction runs, so |
| 1228 | // a seed that dies of a full disk, or an agent that dies mid-seed, cannot |
| 1229 | // leave a partial database with nothing to refuse it. It is removed only |
| 1230 | // once the seed's processes are proven gone (dischargeSeedObligation). |
| 1231 | // |
| 1232 | // ownsDB is seedCreatedTheDB at the seed's start: only then is the |
| 1233 | // database the obligation's to remove. layout is where the seed ran: |
| 1234 | // LayoutNested is root's seed unit in the host slice, not |
| 1235 | // carlos-exec@<host>, so its proof is that unit's (nestedSeedGone); |
| 1236 | // LayoutFlat is an exec child; "" is an obligation that does not say. |
| 1237 | type seedObligation struct { |
| 1238 | ownsDB bool |
| 1239 | layout Layout |
| 1240 | } |
| 1241 | |
| 1242 | // The obligation's body: three lines, the first a version. Anything a |
| 1243 | // reader does not recognise reads as the cautious answer: a database that |
| 1244 | // was not the seed's (kept), and the layout the row reads now. |
| 1245 | const ( |
| 1246 | seedObligationHeader = "carlos seed obligation v1" |
| 1247 | seedObligationOwned = "db=owned-by-seed" |
| 1248 | seedObligationKept = "db=there-before-seed" |
| 1249 | seedObligationFlat = "seed=flat" |
| 1250 | seedObligationNested = "seed=nested" |
| 1251 | ) |
| 1252 | |
| 1253 | func (o seedObligation) body() []byte { |
| 1254 | db, seed := seedObligationKept, seedObligationFlat |
| 1255 | if o.ownsDB { |
| 1256 | db = seedObligationOwned |
| 1257 | } |
| 1258 | if o.layout == LayoutNested { |
| 1259 | seed = seedObligationNested |
| 1260 | } |
| 1261 | return []byte(seedObligationHeader + "\n" + db + "\n" + seed + "\n") |
| 1262 | } |
| 1263 | |
| 1264 | // recordSeedObligation writes the obligation ahead of a seed, durably |
| 1265 | // (safefile: file fsync, rename, dir fsync). One that is already there is |
| 1266 | // an earlier seed not yet answered (Restore answers it before any seed of |
| 1267 | // a wake): it is never overwritten, since that could lose what it says the |
| 1268 | // earlier seed owned, and the seed is refused with ErrSeedUnproven. A write |
| 1269 | // that fails refuses the seed as an ordinary error: nothing ran. |
| 1270 | func recordSeedObligation(host, dbPath string, o seedObligation) error { |
| 1271 | ob := dbPath + seedUnprovenSuffix |
| 1272 | if _, err := os.Lstat(ob); err == nil { |
| 1273 | return fmt.Errorf("%s: refusing to seed: %w: an earlier seed's obligation %s is not answered", host, ErrSeedUnproven, ob) |
| 1274 | } else if !errors.Is(err, fs.ErrNotExist) { |
| 1275 | return fmt.Errorf("%s: refusing to seed: read the seed obligation %s: %w", host, ob, err) |
| 1276 | } |
| 1277 | if err := writeSeedObligationFile(ob, o.body(), 0o600); err != nil { |
| 1278 | return fmt.Errorf("%s: refusing to seed: record its obligation %s: %w", host, ob, err) |
| 1279 | } |
| 1280 | return nil |
| 1281 | } |
| 1282 | |
| 1283 | // readSeedObligation reads dbPath's obligation. present is false only on a |
| 1284 | // definite "not there"; any other failure to look is an error, never "no |
| 1285 | // obligation". |
| 1286 | // |
| 1287 | // The name sits beside the database, on ground the instance itself can |
| 1288 | // write, so it is opened as the env files are (openOwnEnvNoFollow): a |
| 1289 | // symlink or a FIFO planted there is refused in the open itself, never |
| 1290 | // followed or waited on. Such an entry was not written by a seed, so it |
| 1291 | // reads as an obligation that does not say (the zero value): it is proven |
| 1292 | // and removed like any other, and the database beside it is kept, which |
| 1293 | // neither hangs the wake nor holds the row down. |
| 1294 | func readSeedObligation(dbPath string) (o seedObligation, present bool, err error) { |
| 1295 | f, err := openOwnEnvNoFollow(dbPath+seedUnprovenSuffix, os.O_RDONLY, 0) |
| 1296 | if errors.Is(err, fs.ErrNotExist) { |
| 1297 | return seedObligation{}, false, nil |
| 1298 | } |
| 1299 | if errors.Is(err, errOwnEnvSymlink) || errors.Is(err, errOwnEnvNotRegular) { |
| 1300 | return seedObligation{}, true, nil |
| 1301 | } |
| 1302 | if err != nil { |
| 1303 | return seedObligation{}, true, err |
| 1304 | } |
| 1305 | raw, err := io.ReadAll(io.LimitReader(f, 4<<10)) |
| 1306 | f.Close() |
| 1307 | if err != nil { |
| 1308 | return seedObligation{}, true, err |
| 1309 | } |
| 1310 | lines := strings.Split(string(raw), "\n") |
| 1311 | if len(lines) >= 3 && lines[0] == seedObligationHeader { |
| 1312 | o.ownsDB = lines[1] == seedObligationOwned |
| 1313 | switch lines[2] { |
| 1314 | case seedObligationNested: |
| 1315 | o.layout = LayoutNested |
| 1316 | case seedObligationFlat: |
| 1317 | o.layout = LayoutFlat |
| 1318 | } |
| 1319 | } |
| 1320 | return o, true, nil |
| 1321 | } |
| 1322 | |
| 1323 | // dischargeSeedObligation runs once a seed's processes are proven gone: |
| 1324 | // the database files go first when removeDB (removeSeedDB, every error |
| 1325 | // checked and the directory synced), and only then the obligation, synced |
| 1326 | // too, so a crash between the two leaves the obligation, never a database |
| 1327 | // without it. Any failure keeps the obligation, and the next wake proves |
| 1328 | // and removes again. |
| 1329 | func dischargeSeedObligation(dbPath string, removeDB bool) error { |
| 1330 | if removeDB { |
| 1331 | if err := removeSeedDB(dbPath); err != nil { |
| 1332 | return err |
| 1333 | } |
| 1334 | } |
| 1335 | ob := dbPath + seedUnprovenSuffix |
| 1336 | if err := os.Remove(ob); err != nil && !errors.Is(err, fs.ErrNotExist) { |
| 1337 | return fmt.Errorf("remove the seed obligation %s: %w", ob, err) |
| 1338 | } |
| 1339 | if err := syncDir(filepath.Dir(dbPath)); err != nil { |
| 1340 | return fmt.Errorf("sync %s after removing the seed obligation: %w", filepath.Dir(dbPath), err) |
| 1341 | } |
| 1342 | return nil |
| 1343 | } |
| 1344 | |
| 1345 | // settleSeedObligation answers an earlier seed's obligation before anything |
| 1346 | // treats dbPath as warm. With none there it does nothing. Otherwise the |
| 1347 | // seed's processes are proven gone first (seedObligationProof). A proof |
| 1348 | // that fails refuses the wake with ErrSeedUnproven and removes nothing. |
| 1349 | // One that holds discharges the obligation: a database the seed owned goes |
| 1350 | // with it, so the wake goes on as a first boot and the seed runs again; a |
| 1351 | // database that was there before the seed stays, and is warm. The |
| 1352 | // obligation is a file, so it survives an agent restart. |
| 1353 | func (b *ExecBackend) settleSeedObligation(ctx context.Context, rt registry.Route, dbPath string) error { |
| 1354 | host := rt.Host |
| 1355 | o, present, err := readSeedObligation(dbPath) |
| 1356 | if !present { |
| 1357 | return nil |
| 1358 | } |
| 1359 | if err != nil { |
| 1360 | // "Couldn't tell" is never "no obligation", and never "proven |
| 1361 | // gone" either (fix wave 1, item 2): ErrSeedUnproven leaves the |
| 1362 | // lease to its TTL. A nested wake runs no host proof before this, |
| 1363 | // so nothing else says the seed's processes are gone. |
| 1364 | return fmt.Errorf("%s: %w: read the seed obligation %s: %w", host, ErrSeedUnproven, dbPath+seedUnprovenSuffix, err) |
| 1365 | } |
| 1366 | if err := b.seedObligationProof(ctx, rt, o); err != nil { |
| 1367 | return fmt.Errorf("%s: %w: an earlier seed's processes are not proven gone, so its database is not used: %v", host, ErrSeedUnproven, err) |
| 1368 | } |
| 1369 | if err := dischargeSeedObligation(dbPath, o.ownsDB); err != nil { |
| 1370 | return fmt.Errorf("%s: after an earlier seed proven gone: %w", host, err) |
| 1371 | } |
| 1372 | if o.ownsDB { |
| 1373 | log.Printf("activator: %s: the earlier seed's processes are proven gone; its database is removed and the seed runs again", host) |
| 1374 | } else { |
| 1375 | log.Printf("activator: %s: the earlier seed's processes are proven gone; the database that was there before it is kept", host) |
| 1376 | } |
| 1377 | return nil |
| 1378 | } |
| 1379 | |
| 1380 | // seedObligationProof proves an obligation's seed gone, by where it ran. A |
| 1381 | // nested seed ran as root's seed unit in the host slice: that unit is |
| 1382 | // stopped and its cgroup proven empty (nestedSeedGone). A flat seed ran in |
| 1383 | // carlos-exec@<host>, flag off in it and flag on in its role child, so the |
| 1384 | // host cgroup is proven, whatever CARLOS_SWAP says now: it holds the role |
| 1385 | // child too, so it covers the seed whichever way it ran. An obligation that |
| 1386 | // does not say (an unknown body) is proven by the layout the row reads now. |
| 1387 | // |
| 1388 | // Killing there is safe: this runs under the wake's lease, before anything |
| 1389 | // of this wake has run, and no other entry of the host is live (an unproven |
| 1390 | // failed wake or stray sidecar refuses the wake before the lease). |
| 1391 | func (b *ExecBackend) seedObligationProof(ctx context.Context, rt registry.Route, o seedObligation) error { |
| 1392 | layout := o.layout |
| 1393 | if layout == "" { |
| 1394 | layout = b.pathsFor(rt).layout |
| 1395 | } |
| 1396 | if layout == LayoutNested { |
| 1397 | return b.nestedSeedGone(ctx, rt) |
| 1398 | } |
| 1399 | pctx, cancel := context.WithTimeout(ctx, actStopTimeout) |
| 1400 | defer cancel() |
| 1401 | return b.seedDescendantsGone(pctx, rt, "") |
| 1402 | } |
| 1403 | |
| 1404 | // nestedSeedGone stops root's seed unit for rt and proves its cgroup empty |
| 1405 | // (waitSeedCgroupEmpty), each bounded and detached from ctx's own |
| 1406 | // cancellation, which may be why the seed failed. A failed stop is logged: |
| 1407 | // the empty cgroup is the proof. |
| 1408 | func (b *ExecBackend) nestedSeedGone(ctx context.Context, rt registry.Route) error { |
| 1409 | if b.Control == nil { |
| 1410 | return fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| 1411 | } |
| 1412 | sctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl) |
| 1413 | if err := b.Control.Stop(sctx, rt.Host, instancectl.RoleSeed); err != nil { |
| 1414 | log.Printf("activator: %s: stop seed unit: %v", rt.Host, err) |
| 1415 | } |
| 1416 | cancel() |
| 1417 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| 1418 | defer wcancel() |
| 1419 | return b.waitSeedCgroupEmpty(wctx, rt) |
| 1420 | } |
| 1421 | |
| 1422 | // seedCreatedTheDB decides, from the stat taken just BEFORE a seed ran, |
| 1423 | // whether the database at that path can only be the seed's own work — the |
| 1424 | // single condition under which a failed seed's cleanup may delete it. |
| 1425 | // |
| 1426 | // Fail-safe by construction: ONLY a definite "not there" says yes. A stat that |
| 1427 | // merely failed — a permission error, an I/O error, a path that stopped being |
| 1428 | // traversable — means "there might be data here", and a guard on deleting live |
| 1429 | // data must never arm itself because we couldn't look. |
| 1430 | func seedCreatedTheDB(statErr error) bool { return os.IsNotExist(statErr) } |
| 1431 | |
| 1432 | // seedTimeout bounds a first-boot seed run — a wedged seed must fail the wake |
| 1433 | // (which releases the lease) rather than sit on it. seedDrainGrace is how long |
| 1434 | // Wait then waits for the killed seed's output pipes to close before giving up |
| 1435 | // on them (cmd.WaitDelay). Both vars so tests can shrink them. |
| 1436 | var ( |
| 1437 | seedTimeout = 2 * time.Minute |
| 1438 | seedDrainGrace = 10 * time.Second |
| 1439 | ) |
| 1440 | |
| 1441 | // capBuffer keeps at most capBufferMax bytes of what's written to it and |
| 1442 | // discards the rest, so quoting a child's stderr in an error can never turn a |
| 1443 | // chatty seed into an unbounded string. Locked because it is read after |
| 1444 | // the seed's Wait returns while os/exec's own copy goroutine may still be |
| 1445 | // writing: on a WaitDelay expiry (the wedged-seed path) Wait returns without |
| 1446 | // waiting for that copy to finish. |
| 1447 | type capBuffer struct { |
| 1448 | mu sync.Mutex |
| 1449 | b []byte |
| 1450 | } |
| 1451 | |
| 1452 | const capBufferMax = 4 << 10 |
| 1453 | |
| 1454 | func (c *capBuffer) Write(p []byte) (int, error) { |
| 1455 | c.mu.Lock() |
| 1456 | defer c.mu.Unlock() |
| 1457 | if room := capBufferMax - len(c.b); room > 0 { |
| 1458 | if len(p) < room { |
| 1459 | room = len(p) |
| 1460 | } |
| 1461 | c.b = append(c.b, p[:room]...) |
| 1462 | } |
| 1463 | return len(p), nil |
| 1464 | } |
| 1465 | |
| 1466 | func (c *capBuffer) String() string { |
| 1467 | c.mu.Lock() |
| 1468 | defer c.mu.Unlock() |
| 1469 | return string(c.b) |
| 1470 | } |
| 1471 | |
| 1472 | // seedUnit is Seed's nested body: the same contract as seedChild, namely |
| 1473 | // the merged env, the CARLOS_ADMIN_TOKEN mint, and the cleanup on a failure |
| 1474 | // this call itself caused, run through the root instance-control channel |
| 1475 | // instead of a direct exec (design §6.2). Argv and env come from the same |
| 1476 | // seedArgv and withoutVars calls seedChild uses, so the two cannot drift. |
| 1477 | // |
| 1478 | // A nil exe seeds on the live link (Seed). A staged exe is the build the |
| 1479 | // wake recorded (SeedExe): root validates it and binds the seed unit's app |
| 1480 | // mount to it, so the live link need not exist and a repoint of it |
| 1481 | // changes nothing. A control channel that cannot name it refuses the seed. |
| 1482 | // With LiveExecGate set a nil exe pins the live link (declinedLivePin) and |
| 1483 | // seeds that file the same way. tier is that build's own (ruling R-TIER); |
| 1484 | // zero is the live tier. |
| 1485 | func (b *ExecBackend) seedUnit(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) error { |
| 1486 | if b.Cfg.HostWrites.Enabled { |
| 1487 | return ErrClaimIsolationUnavailable |
| 1488 | } |
| 1489 | host := rt.Host |
| 1490 | if err := b.nestedGate(rt); err != nil { |
| 1491 | return err |
| 1492 | } |
| 1493 | if exe == nil && b.LiveExecGate != nil { |
| 1494 | // A declined box names the build its gate passed to root, at that |
| 1495 | // build's tier (declinedLivePin), as SeedExe does. Before anything |
| 1496 | // is prepared and before the database stat below, so a refusal asks |
| 1497 | // root for nothing and removes nothing. |
| 1498 | id, err := b.declinedLivePin(ctx, rt) |
| 1499 | if err != nil { |
| 1500 | return err |
| 1501 | } |
| 1502 | exe, tier = stagedExeOf(id), b.GenTier(rt, id.Version) |
| 1503 | } |
| 1504 | if tier == (cgroup.Limit{}) { |
| 1505 | tier = b.tier(rt) |
| 1506 | } |
| 1507 | var xc StagedExeControl |
| 1508 | if exe == nil { |
| 1509 | if err := b.ensureLiveLink(ctx, host); err != nil { |
| 1510 | return err |
| 1511 | } |
| 1512 | } else { |
| 1513 | var err error |
| 1514 | if xc, err = b.stagedExeControl(); err != nil { |
| 1515 | return fmt.Errorf("%s: not seeded with %s: %w", host, exe.Version, err) |
| 1516 | } |
| 1517 | } |
| 1518 | p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSeed) |
| 1519 | if err != nil { |
| 1520 | return err |
| 1521 | } |
| 1522 | defer logs.Close() |
| 1523 | dbPath := p.dbPath |
| 1524 | seedArgs, err := seedArgv(rt, dbPath) |
| 1525 | if err != nil { |
| 1526 | return err |
| 1527 | } |
| 1528 | if seedArgs == nil { |
| 1529 | return nil // defensive: doWake only calls us for a non-empty seed |
| 1530 | } |
| 1531 | argv := append([]string{b.liveLink(host)}, seedArgs...) |
| 1532 | env := childEnv(seedVars(fileEnv, host)) |
| 1533 | if err := b.writeLaunch(p, instancectl.RoleSeed, host, argv, env); err != nil { |
| 1534 | return err |
| 1535 | } |
| 1536 | // Whether the database existed BEFORE the seed ran decides whether a |
| 1537 | // failed seed may clean up after itself below, seedChild's rule, |
| 1538 | // unchanged for the unit path. |
| 1539 | _, statErr := os.Stat(dbPath) |
| 1540 | seedOwnsDB := seedCreatedTheDB(statErr) |
| 1541 | // The obligation is written ahead, as seedChild's is, before root is |
| 1542 | // asked to run anything; it names the seed unit as the proof to ask. |
| 1543 | if err := recordSeedObligation(host, dbPath, seedObligation{ownsDB: seedOwnsDB, layout: LayoutNested}); err != nil { |
| 1544 | return err |
| 1545 | } |
| 1546 | // Then the wake's pending record (amadan #272), the last step before |
| 1547 | // root is asked, as the flat seed's comes at its exec after its |
| 1548 | // obligation (Astra review of zdd-owed-items, P2): an obligation that |
| 1549 | // cannot be written records nothing. A record that fails, or a recheck |
| 1550 | // that refuses after it, runs no seed, so the obligation just written |
| 1551 | // is discharged and the database, which no seed touched, is kept. |
| 1552 | if _, err := runPendingRecord(ctx); err != nil { |
| 1553 | cause := fmt.Errorf("seed for %s: %w", host, err) |
| 1554 | if derr := dischargeSeedObligation(dbPath, false); derr != nil { |
| 1555 | return fmt.Errorf("%w; and %w", cause, derr) |
| 1556 | } |
| 1557 | return cause |
| 1558 | } |
| 1559 | |
| 1560 | log.Printf("activator: seeding %s: %s", host, strings.Join(argv, " ")) |
| 1561 | // The client already bounds a seed call at instancectl.DeadlineSeed. |
| 1562 | // Wrapping ctx any tighter here (seedTimeout, at 2m, is shorter) would |
| 1563 | // guarantee the failure path on every seed that legitimately runs |
| 1564 | // longer than that. Wrapping at the SAME deadline is redundant with the |
| 1565 | // client's own timeout but harmless, and keeps this call visibly |
| 1566 | // bounded without second-guessing how long a seed may take. |
| 1567 | sctx, cancel := context.WithTimeout(ctx, instancectl.DeadlineSeed) |
| 1568 | var res instancectl.SeedResponse |
| 1569 | if exe != nil { |
| 1570 | res, err = xc.SeedExe(sctx, host, tier.MemoryMax, *exe) |
| 1571 | } else { |
| 1572 | res, err = b.Control.Seed(sctx, host, tier.MemoryMax) |
| 1573 | } |
| 1574 | cancel() |
| 1575 | if errors.Is(err, instancectl.ErrStagedExe) { |
| 1576 | // Root refused the recorded build before it laid out or started |
| 1577 | // anything (403), so no seed ran: nothing to stop, and the |
| 1578 | // database is not the seed's work. Its obligation is discharged. |
| 1579 | cause := fmt.Errorf("seed for %s: %w", host, err) |
| 1580 | if derr := dischargeSeedObligation(dbPath, false); derr != nil { |
| 1581 | return fmt.Errorf("%w; and %w", cause, derr) |
| 1582 | } |
| 1583 | return cause |
| 1584 | } |
| 1585 | if err != nil { |
| 1586 | return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s: %w", host, err)) |
| 1587 | } |
| 1588 | if res.Result != "success" { |
| 1589 | return b.seedUnitFailed(ctx, rt, dbPath, seedOwnsDB, fmt.Errorf("seed for %s failed: exit %s (%s)", host, res.Status, res.Result)) |
| 1590 | } |
| 1591 | // A successful seed is not done until its cgroup is empty (design |
| 1592 | // §6.2): the control service already waited before answering, but the |
| 1593 | // activator's own proof is what lets the wake go on to Start with no |
| 1594 | // seed process still holding the database. Unproven, the obligation |
| 1595 | // written ahead stays. |
| 1596 | wctx, wcancel := context.WithTimeout(context.WithoutCancel(ctx), actStopTimeout) |
| 1597 | emptyErr := b.waitSeedCgroupEmpty(wctx, rt) |
| 1598 | wcancel() |
| 1599 | if emptyErr != nil { |
| 1600 | return fmt.Errorf("seed for %s: %w: processes remain after a successful seed: %v", host, ErrSeedUnproven, emptyErr) |
| 1601 | } |
| 1602 | if derr := dischargeSeedObligation(dbPath, false); derr != nil { |
| 1603 | return fmt.Errorf("seed for %s: %w", host, derr) |
| 1604 | } |
| 1605 | log.Printf("activator: seeded %s", host) |
| 1606 | return nil |
| 1607 | } |
| 1608 | |
| 1609 | // seedUnitFailed is seedUnit's shared failure tail: on EITHER failure |
| 1610 | // shape, whether Control.Seed itself errored or it answered with a |
| 1611 | // non-success result, the seed unit is stopped and its cgroup is proven |
| 1612 | // empty (nestedSeedGone) BEFORE any database cleanup runs, never after. |
| 1613 | // Running the cleanup on the bare result instead could race a seed process |
| 1614 | // SIGKILL hadn't actually reaped yet, exactly the class of bug Stop's own |
| 1615 | // ErrStillRunning discipline exists to rule out elsewhere. |
| 1616 | // The database goes only when seedOwnsDB AND the proof held, and then the |
| 1617 | // obligation (dischargeSeedObligation); when the proof failed, nothing is |
| 1618 | // removed, the obligation stays and the returned error says so, because a |
| 1619 | // database is not safe to delete out from under a process that might still |
| 1620 | // be writing it. |
| 1621 | func (b *ExecBackend) seedUnitFailed(ctx context.Context, rt registry.Route, dbPath string, seedOwnsDB bool, cause error) error { |
| 1622 | if emptyErr := b.nestedSeedGone(ctx, rt); emptyErr != nil { |
| 1623 | return fmt.Errorf("%w: %w (the seed's processes remain after stopping it; its database was left in place: %v)", cause, ErrSeedUnproven, emptyErr) |
| 1624 | } |
| 1625 | if derr := dischargeSeedObligation(dbPath, seedOwnsDB); derr != nil { |
| 1626 | return fmt.Errorf("%w; and %w", cause, derr) |
| 1627 | } |
| 1628 | return cause |
| 1629 | } |
| 1630 | |
| 1631 | // waitSeedCgroupEmpty polls the seed unit's cgroup.events until it reports |
| 1632 | // empty or roleEmptyBound elapses, honouring ctx and b.ObservePoll like the |
| 1633 | // exit observer (unitproc.go). A timeout or a read error is returned to the |
| 1634 | // caller as a log line, not a failure: it is the deciding result's |
| 1635 | // confidence that is at stake here, not the wake itself. |
| 1636 | func (b *ExecBackend) waitSeedCgroupEmpty(ctx context.Context, rt registry.Route) error { |
| 1637 | return b.waitRoleCgroupEmpty(ctx, rt, instancectl.RoleSeed) |
| 1638 | } |
| 1639 | |
| 1640 | // roleEmptyBound is how long waitRoleCgroupEmpty waits for a role's cgroup |
| 1641 | // to empty: actStopTimeout. A var so tests can shrink it. |
| 1642 | var roleEmptyBound = actStopTimeout |
| 1643 | |
| 1644 | // waitRoleCgroupEmpty polls role's unit's cgroup.events until it reports |
| 1645 | // empty or roleEmptyBound elapses, honouring ctx and b.ObservePoll. |
| 1646 | func (b *ExecBackend) waitRoleCgroupEmpty(ctx context.Context, rt registry.Route, role instancectl.Role) error { |
| 1647 | events := b.CgroupEvents |
| 1648 | if events == nil { |
| 1649 | events = cgroupEventsUnder(b.cgroupRoot()) |
| 1650 | } |
| 1651 | poll := b.ObservePoll |
| 1652 | if poll <= 0 { |
| 1653 | poll = time.Second |
| 1654 | } |
| 1655 | slice := instancectl.SliceName(rt.Host) |
| 1656 | unit := instancectl.UnitName(role, rt.Host) |
| 1657 | bound := roleEmptyBound |
| 1658 | deadline := time.Now().Add(bound) |
| 1659 | for { |
| 1660 | exists, populated, err := events(slice, unit) |
| 1661 | if err != nil { |
| 1662 | return err |
| 1663 | } |
| 1664 | if !exists || !populated { |
| 1665 | return nil |
| 1666 | } |
| 1667 | if time.Now().After(deadline) { |
| 1668 | return fmt.Errorf("processes remained after %s", bound) |
| 1669 | } |
| 1670 | select { |
| 1671 | case <-ctx.Done(): |
| 1672 | return ctx.Err() |
| 1673 | case <-time.After(poll): |
| 1674 | } |
| 1675 | } |
| 1676 | } |
| 1677 | |
| 1678 | // NextDue asks the (still-running) instance for its earliest pending scheduled |
| 1679 | // delivery over its socket. The endpoint is admin-gated like every instance |
| 1680 | // data endpoint, so present the instance's own admin token — read from the |
| 1681 | // same env file the spawn path hands the child (the activator holds it either |
| 1682 | // way). |
| 1683 | func (b *ExecBackend) NextDue(ctx context.Context, rt registry.Route) (int64, error) { |
| 1684 | return b.NextDueVia(ctx, rt, rt.Socket) |
| 1685 | } |
| 1686 | |
| 1687 | // NextDueVia is NextDue dialling socket instead of rt.Socket: the activator |
| 1688 | // passes the instance's current generation socket (AppSocket, zero-downtime |
| 1689 | // spec §7.1), because with generations on nothing listens on the canonical |
| 1690 | // path. With them off the activator passes rt.Socket, which is NextDue. |
| 1691 | func (b *ExecBackend) NextDueVia(ctx context.Context, rt registry.Route, socket string) (int64, error) { |
| 1692 | paths := b.pathsFor(rt) |
| 1693 | // A nested host's env file lives on tenant-writable ground (design |
| 1694 | // §3.2), the same as instanceEnv's read of it: go through instancefs |
| 1695 | // rather than a plain read, so a name planted there cannot be anything |
| 1696 | // but a regular file, on pain of an error instead of a hang. |
| 1697 | read := readEnvFile |
| 1698 | if paths.layout == LayoutNested { |
| 1699 | read = func(path string) (map[string]string, error) { |
| 1700 | return readEnvFileBeneath(filepath.Dir(path), filepath.Base(path)) |
| 1701 | } |
| 1702 | } |
| 1703 | fileEnv, err := read(paths.envFile) |
| 1704 | if err != nil { |
| 1705 | return 0, fmt.Errorf("read env file: %w", err) |
| 1706 | } |
| 1707 | token := fileEnv["CARLOS_ADMIN_TOKEN"] |
| 1708 | if token == "" { |
| 1709 | return 0, errors.New("no CARLOS_ADMIN_TOKEN in env file") |
| 1710 | } |
| 1711 | req, err := http.NewRequestWithContext(ctx, "GET", "http://carlos/api/next-due", nil) |
| 1712 | if err != nil { |
| 1713 | return 0, err |
| 1714 | } |
| 1715 | req.Header.Set("Authorization", "Bearer "+token) |
| 1716 | resp, err := socketClient(socket).Do(req) |
| 1717 | if err != nil { |
| 1718 | return 0, err |
| 1719 | } |
| 1720 | defer resp.Body.Close() |
| 1721 | if resp.StatusCode != http.StatusOK { |
| 1722 | return 0, fmt.Errorf("GET /api/next-due: HTTP %d", resp.StatusCode) |
| 1723 | } |
| 1724 | var body struct { |
| 1725 | Due int64 `json:"due"` |
| 1726 | } |
| 1727 | if err := json.NewDecoder(io.LimitReader(resp.Body, 1<<12)).Decode(&body); err != nil { |
| 1728 | return 0, fmt.Errorf("decode /api/next-due: %w", err) |
| 1729 | } |
| 1730 | return body.Due, nil |
| 1731 | } |
| 1732 | |
| 1733 | // ensureLiveLink verifies host's live symlink (<OptDir>/live/<host> → a staged |
| 1734 | // version) exists before the exec, routing a missing one through the |
| 1735 | // Materialize hook when configured. A materialize failure (e.g. the staging |
| 1736 | // mechanism unreachable) must propagate — the exact live symptom found |
| 1737 | // 2026-07-13 in Eleven (a 503 "instance failed to wake"), never silently |
| 1738 | // swallowed into "looks fine, wakes anyway." |
| 1739 | func (b *ExecBackend) ensureLiveLink(ctx context.Context, host string) error { |
| 1740 | bin := b.liveLink(host) |
| 1741 | if _, err := os.Stat(bin); err == nil { |
| 1742 | return nil |
| 1743 | } |
| 1744 | if b.Materialize != nil { |
| 1745 | if err := b.Materialize(ctx, host); err != nil { |
| 1746 | return fmt.Errorf("materialize live symlink for %s: %w", host, err) |
| 1747 | } |
| 1748 | return nil |
| 1749 | } |
| 1750 | return fmt.Errorf("no live binary for %s (%s missing and no Materialize hook)", host, bin) |
| 1751 | } |
| 1752 | |
| 1753 | // LiveIdentity is the one file a cold wake will run (settings spec §4.2): |
| 1754 | // the live link, materialized first if it is missing (ensureLiveLink, so |
| 1755 | // the Materialize hook and its root-stage relay run exactly as a wake's |
| 1756 | // spawn would run them), then resolved ONCE into the file at the end of |
| 1757 | // its chain and that file's digest, as StartWorkerStaged does. Version is |
| 1758 | // "" for a link outside the version store; the caller decides whether that |
| 1759 | // may run (Service.recordBuild refuses it while a build state is kept). |
| 1760 | func (b *ExecBackend) LiveIdentity(ctx context.Context, rt registry.Route) (release.StagedIdentity, error) { |
| 1761 | if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| 1762 | return release.StagedIdentity{}, err |
| 1763 | } |
| 1764 | id, err := release.ResolveLive(b.OptDir, rt.Host) |
| 1765 | if err != nil { |
| 1766 | return release.StagedIdentity{}, fmt.Errorf("%s: %w", rt.Host, err) |
| 1767 | } |
| 1768 | return id, nil |
| 1769 | } |
| 1770 | |
| 1771 | // declinedLivePin is a live-link start's identity on a box with |
| 1772 | // LiveExecGate set (a declined box that honours failure records or hold-ran |
| 1773 | // markers): live/<host> materialized and resolved ONCE, hashed |
| 1774 | // (LiveIdentity), and the gate asked about that version. The caller runs |
| 1775 | // exactly that file: a flat start execs it by path, and a nested one names |
| 1776 | // it to root (StartExe, StartGenExe, StartSidecarExe, SeedExe), which |
| 1777 | // validates it and binds it in place of the link. So the build the gate |
| 1778 | // passed is the build that runs, and root repointing the link after this |
| 1779 | // read changes nothing. |
| 1780 | // |
| 1781 | // A link outside the version store names no build the gate could judge or |
| 1782 | // root could validate (instancectl.validateStagedExe), and is refused in |
| 1783 | // recordBuild's words, as the activator already refuses every exec row's |
| 1784 | // unversioned capture while a build state is kept (ruling |
| 1785 | // R-EXEC-NOVERSION). Nothing has been prepared, launched or asked of root |
| 1786 | // when it returns an error. |
| 1787 | func (b *ExecBackend) declinedLivePin(ctx context.Context, rt registry.Route) (release.StagedIdentity, error) { |
| 1788 | id, err := b.LiveIdentity(ctx, rt) |
| 1789 | if err != nil { |
| 1790 | return release.StagedIdentity{}, err |
| 1791 | } |
| 1792 | if id.Version == "" { |
| 1793 | log.Printf("activator: %s not started: the executable its live link names (%s) is outside the version store, so this box cannot check it against its holds; move the instance onto a promoted build", rt.Host, id.StagedPath) |
| 1794 | return release.StagedIdentity{}, fmt.Errorf("%s: %w: the executable is outside the version store", rt.Host, errBuildStateUnwritten) |
| 1795 | } |
| 1796 | if err := b.LiveExecGate(rt, id.Version); err != nil { |
| 1797 | return release.StagedIdentity{}, err |
| 1798 | } |
| 1799 | return id, nil |
| 1800 | } |
| 1801 | |
| 1802 | // BuildIdentity is buildIdentifier's answer: build's staged executable of |
| 1803 | // rt's app on this box, at the path root's validation expects |
| 1804 | // (release.VersionPath under the row's account; instancectl's |
| 1805 | // validateStagedExe), hashed now. It reads no link. An invalid version, an |
| 1806 | // app name that is not one path segment, or a copy that is not a regular |
| 1807 | // file it can read names nothing. Whether build may run is the caller's |
| 1808 | // question (the live exec gate), not this one's. |
| 1809 | func (b *ExecBackend) BuildIdentity(rt registry.Route, build string) (release.StagedIdentity, error) { |
| 1810 | if !release.ValidVersion(build) { |
| 1811 | return release.StagedIdentity{}, fmt.Errorf("%s: %q is not a build version", rt.Host, build) |
| 1812 | } |
| 1813 | if rt.App == "" || rt.App == "." || rt.App == ".." || filepath.Base(rt.App) != rt.App { |
| 1814 | return release.StagedIdentity{}, fmt.Errorf("%s: app %q names no staged copy", rt.Host, rt.App) |
| 1815 | } |
| 1816 | path := release.VersionPath(b.OptDir, release.RouteAccount(rt), build, rt.App) |
| 1817 | fi, err := os.Lstat(path) |
| 1818 | if err != nil { |
| 1819 | return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %w", rt.Host, build, err) |
| 1820 | } |
| 1821 | if !fi.Mode().IsRegular() { |
| 1822 | return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %s is not a regular file", rt.Host, build, path) |
| 1823 | } |
| 1824 | sum, err := release.FileSHA256(path) |
| 1825 | if err != nil { |
| 1826 | return release.StagedIdentity{}, fmt.Errorf("%s: build %s: %w", rt.Host, build, err) |
| 1827 | } |
| 1828 | return release.StagedIdentity{Version: build, StagedPath: path, Digest: sum}, nil |
| 1829 | } |
| 1830 | |
| 1831 | // Start dispatches a wake by layout (design §8.1): a nested row starts as a |
| 1832 | // unit through the root instance-control channel, a fenced or conflicting |
| 1833 | // row refuses outright, and a flat or foreign row runs today's exec child. |
| 1834 | func (b *ExecBackend) Start(ctx context.Context, rt registry.Route) (Proc, string, error) { |
| 1835 | p := b.pathsFor(rt) |
| 1836 | switch p.layout { |
| 1837 | case LayoutNested: |
| 1838 | return b.startUnitAt(ctx, rt, rt.Socket, 0, nil, cgroup.Limit{}) |
| 1839 | case LayoutFenced, LayoutConflict: |
| 1840 | return nil, "", fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason) |
| 1841 | } |
| 1842 | return b.startChildAt(ctx, rt, rt.Socket, 0, "", cgroup.Limit{}) |
| 1843 | } |
| 1844 | |
| 1845 | // StartGen is Start for generation n (zero-downtime spec §7.1). perGen false |
| 1846 | // is Start exactly: the canonical socket, the canonical control socket, the |
| 1847 | // same argv. perGen true binds the generation's own socket: GenSocketPath |
| 1848 | // for a flat row, g<n>/app.sock in the instance run dir for a nested one |
| 1849 | // (rev 19, rule 1). Only the directory differs from the flag-off launch: |
| 1850 | // the file name, the rest of the argv and the agent-owned environment are |
| 1851 | // the same (rule 2). A foreign row keeps its canonical socket either way: |
| 1852 | // it is outside the two layouts the spec gives generation paths to, and |
| 1853 | // keeping it is the conservative reading (plan 1, "Spec questions"). So |
| 1854 | // does a flat row whose generation path would not fit a unix socket path |
| 1855 | // (genLayout, rule 4). |
| 1856 | // |
| 1857 | // The stale-file rule (spec §7.1): before the app binds, the generation's |
| 1858 | // directory is made and the file at THIS generation's own path is removed |
| 1859 | // (left there by an earlier agent process, whose counter started at 1 like |
| 1860 | // this one's). A sibling generation's path is never touched. |
| 1861 | func (b *ExecBackend) StartGen(ctx context.Context, rt registry.Route, n uint64, perGen bool) (Proc, Gen, error) { |
| 1862 | return b.startGenFrom(ctx, rt, n, perGen, nil, cgroup.Limit{}) |
| 1863 | } |
| 1864 | |
| 1865 | // StartGenExe is StartGen on the identity a wake recorded (settings spec |
| 1866 | // §4.2): the same generation, socket and tier, but the process runs |
| 1867 | // id.StagedPath, checked against id.Digest first. A copy that changed is a |
| 1868 | // failed start, and nothing runs (never the live link instead). A flat or |
| 1869 | // foreign row execs the file by path. A nested row asks root to bind it in |
| 1870 | // place of the live link: the original run unit while GenContainment is |
| 1871 | // off (generation 0), generation n's unit while it is on. The handle names |
| 1872 | // id's build, path, digest, demand and tier, so a sidecar can bind the same |
| 1873 | // file. |
| 1874 | // |
| 1875 | // The tier is id's own (GenTier, as StartStaged takes it; ruling R-TIER), |
| 1876 | // never b.tier(rt): that reads whatever the live link names when it is |
| 1877 | // asked, and root may have repointed the link since the record, which |
| 1878 | // would run A's bytes under B's limits. |
| 1879 | func (b *ExecBackend) StartGenExe(ctx context.Context, rt registry.Route, n uint64, perGen bool, id release.StagedIdentity) (Proc, Gen, error) { |
| 1880 | if err := release.VerifyStaged(id); err != nil { |
| 1881 | return nil, Gen{}, fmt.Errorf("%s: not started on %s: %w", rt.Host, id.Version, err) |
| 1882 | } |
| 1883 | tier := b.GenTier(rt, id.Version) |
| 1884 | proc, g, err := b.startGenFrom(ctx, rt, n, perGen, &id, tier) |
| 1885 | if err != nil { |
| 1886 | return proc, g, err |
| 1887 | } |
| 1888 | g.Build, g.StagedPath, g.Digest, g.Demand, g.Tier = id.Version, id.StagedPath, id.Digest, id.Demand, tier |
| 1889 | return proc, g, nil |
| 1890 | } |
| 1891 | |
| 1892 | // startGenFrom is StartGen's body. A nil id is the live link (StartGen); a |
| 1893 | // non-nil one is the verified file StartGenExe names. tier is that file's |
| 1894 | // own; zero is the live tier (b.tier). |
| 1895 | func (b *ExecBackend) startGenFrom(ctx context.Context, rt registry.Route, n uint64, perGen bool, id *release.StagedIdentity, tier cgroup.Limit) (Proc, Gen, error) { |
| 1896 | var ( |
| 1897 | flatExe string |
| 1898 | unitExe *instancectl.StagedExe |
| 1899 | ) |
| 1900 | if id != nil { |
| 1901 | flatExe, unitExe = id.StagedPath, stagedExeOf(*id) |
| 1902 | } |
| 1903 | p := b.pathsFor(rt) |
| 1904 | sock := rt.Socket |
| 1905 | if perGen { |
| 1906 | switch genLayout(p.layout, rt.Host) { |
| 1907 | case LayoutFlat: |
| 1908 | sock = b.flatGenSocket(rt.Host, n) |
| 1909 | case LayoutNested: |
| 1910 | sock = nestedGenSocketIn(p.runDir, n) |
| 1911 | case LayoutForeign: |
| 1912 | if p.layout == LayoutFlat { |
| 1913 | log.Printf("activator: %s: a generation socket path would exceed %d bytes; generation %d keeps the canonical socket", rt.Host, maxUnixPath, n) |
| 1914 | } |
| 1915 | } |
| 1916 | } |
| 1917 | var ( |
| 1918 | proc Proc |
| 1919 | ver string |
| 1920 | err error |
| 1921 | ) |
| 1922 | switch p.layout { |
| 1923 | case LayoutNested: |
| 1924 | // prepareAt makes sock's generation directory and removes the |
| 1925 | // stale file at sock for the nested layout. |
| 1926 | proc, ver, err = b.startUnitAt(ctx, rt, sock, n, unitExe, tier) |
| 1927 | case LayoutFenced, LayoutConflict: |
| 1928 | return nil, Gen{}, fmt.Errorf("%s: refusing to wake: %s", rt.Host, p.reason) |
| 1929 | default: |
| 1930 | if sock == rt.Socket { |
| 1931 | // A link left at the canonical path by an agent that ran with |
| 1932 | // generations on would make the app's bind fail. That holds for |
| 1933 | // every spawn on the canonical socket, not only a flag-off one: |
| 1934 | // a foreign row, or a flat row too long for a generation path |
| 1935 | // (rev 19 rule 4) whose box ran the old naming, binds it with |
| 1936 | // perGen true too (Astra, review of 19c07b2b), as StartStaged's |
| 1937 | // foreign arm already handles. |
| 1938 | if err := clearLeftoverCanonicalLink(rt.Socket); err != nil { |
| 1939 | return nil, Gen{}, fmt.Errorf("%s: leftover canonical link: %w", rt.Host, err) |
| 1940 | } |
| 1941 | } |
| 1942 | if sock != rt.Socket { |
| 1943 | if err := clearOwnGenSocket(sock); err != nil { |
| 1944 | return nil, Gen{}, fmt.Errorf("%s: generation socket: %w", rt.Host, err) |
| 1945 | } |
| 1946 | } |
| 1947 | proc, ver, err = b.startChildAt(ctx, rt, sock, n, flatExe, tier) |
| 1948 | } |
| 1949 | g := Gen{N: n, Socket: sock, Control: sock + ".control"} |
| 1950 | if err != nil { |
| 1951 | // The handle goes back with the error (P2-13): a spawn that failed |
| 1952 | // may have left generation n running, and its number is what the |
| 1953 | // caller proves it stopped by (KillGen). Refusals above spawned |
| 1954 | // nothing and answer no handle. |
| 1955 | return proc, g, err |
| 1956 | } |
| 1957 | g.Build = ver |
| 1958 | return proc, g, nil |
| 1959 | } |
| 1960 | |
| 1961 | // flatGenSocket is GenSocketPath, under RunDirOverride/g in a test. |
| 1962 | func (b *ExecBackend) flatGenSocket(host string, n uint64) string { |
| 1963 | if b.RunDirOverride != "" { |
| 1964 | return genSocketPathIn(filepath.Join(b.RunDirOverride, "g"), host, n) |
| 1965 | } |
| 1966 | return GenSocketPath(host, n) |
| 1967 | } |
| 1968 | |
| 1969 | // clearOwnGenSocket makes the flat generation's own directory, |
| 1970 | // /run/carlos/g/<n>, and removes a stale file at sock, and only at sock (rev |
| 1971 | // 19, rule 3). The directory is 0750 and inherits /run/carlos's setgid |
| 1972 | // carlos-sockets group, as the nested run dirs do. |
| 1973 | func clearOwnGenSocket(sock string) error { |
| 1974 | if err := os.MkdirAll(filepath.Dir(sock), 0o750|os.ModeSetgid); err != nil { |
| 1975 | return err |
| 1976 | } |
| 1977 | if err := os.Remove(sock); err != nil && !errors.Is(err, os.ErrNotExist) { |
| 1978 | return err |
| 1979 | } |
| 1980 | return nil |
| 1981 | } |
| 1982 | |
| 1983 | // startChildAt is Start's exec-child body (flat and foreign layouts): |
| 1984 | // exec of the versioned binary directly, exactly as before nested layouts |
| 1985 | // existed. sock is the app socket this spawn binds: rt.Socket, or a |
| 1986 | // generation socket (StartGen). n is the generation this spawn is (StartGen's |
| 1987 | // n; 0 from Start). It only places the spawn's cgroup, and only under |
| 1988 | // GenContainment (placeFor). |
| 1989 | // |
| 1990 | // exe is "" for the live link, or the staged file a wake recorded |
| 1991 | // (StartGenExe). spawnChildFrom leaves a staged spawn's cache purge to the |
| 1992 | // swap's switch, but a wake's staged spawn is the host's only process, so |
| 1993 | // its purge happens here, at the same moment, before the health wait. |
| 1994 | // tier is exe's own; zero is the live tier (b.tier). |
| 1995 | func (b *ExecBackend) startChildAt(ctx context.Context, rt registry.Route, sock string, n uint64, exe string, tier cgroup.Limit) (Proc, string, error) { |
| 1996 | p, spawnVer, err := b.spawnChildFrom(ctx, rt, sock, n, exe, tier) |
| 1997 | if err != nil { |
| 1998 | return p, "", err |
| 1999 | } |
| 2000 | if exe != "" && b.Cache != nil { |
| 2001 | if purged := b.Cache.PurgeHost(rt.Host); purged > 0 { |
| 2002 | log.Printf("activator: purged %d cached response(s) for %s on spawn", purged, rt.Host) |
| 2003 | } |
| 2004 | } |
| 2005 | if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil { |
| 2006 | stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| 2007 | defer cancel() |
| 2008 | _ = p.Stop(stopCtx) |
| 2009 | if b.GenContainment && n != 0 { |
| 2010 | // The spawn was placed in g<n>, and Stop signals only the |
| 2011 | // leader: a process it forked can live on there (spec §7.3). |
| 2012 | // The generation's own kill ends it and proves the child |
| 2013 | // empty. Unproven, the process goes back with an error that |
| 2014 | // says so, and the caller keeps the lease until the sweep |
| 2015 | // proves it (P2-13, as startUnitAt does). |
| 2016 | if kerr := b.killGenAfterFailedStart(ctx, rt, n); kerr != nil { |
| 2017 | log.Printf("activator: %s: kill generation %d after a failed health wait: %v", rt.Host, n, kerr) |
| 2018 | return p, "", fmt.Errorf("%w; %w: kill generation %d after a failed health wait: %v", err, errGenUnproven, n, kerr) |
| 2019 | } |
| 2020 | } |
| 2021 | // The handle goes back with the health error, as startUnitAt's does |
| 2022 | // (P2-13). The leader's exit is not the end of what it started: with |
| 2023 | // no g<n>, a process it forked lives on in carlos-exec@<host>, and |
| 2024 | // the caller's failed-wake cleanup proves that cgroup empty before |
| 2025 | // the lease goes (Task 15 fix round 1, ruling a; Astra IMPL-3 F6). |
| 2026 | // The leader is already stopped, so the cleanup's Stop is a no-op. |
| 2027 | return p, "", err |
| 2028 | } |
| 2029 | b.normalizeSocketMode(rt, sock, p) |
| 2030 | return p, spawnVer, nil |
| 2031 | } |
| 2032 | |
| 2033 | // killGenAfterFailedStart is KillGen for a spawn path's own cleanup, on a |
| 2034 | // budget the caller's context cannot cancel: a wake cancelled by a fence or |
| 2035 | // Shutdown still proves what it started. nil is that proof. |
| 2036 | func (b *ExecBackend) killGenAfterFailedStart(ctx context.Context, rt registry.Route, n uint64) error { |
| 2037 | kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl) |
| 2038 | defer cancel() |
| 2039 | if b.killGen != nil { |
| 2040 | return b.killGen(kctx, rt, n) |
| 2041 | } |
| 2042 | return b.KillGen(kctx, rt, n) |
| 2043 | } |
| 2044 | |
| 2045 | // spawnChildFrom is the flat spawn with the executable and the tier named, |
| 2046 | // and nothing after the exec: it returns the moment the process runs. Empty |
| 2047 | // exe is the live link, which is every wake (startChildAt adds today's |
| 2048 | // health wait). A staged path is a swap's successor or a pinned wake |
| 2049 | // (zero-downtime spec §7.5 step 3): the bytes the attempt chose, whatever |
| 2050 | // root does to the live link meanwhile, with the caller owning the health |
| 2051 | // deadline and the termination. tier is that artifact's own tier; zero is |
| 2052 | // the live tier (b.tier). n is the generation number plan 2's containment |
| 2053 | // places the child by. |
| 2054 | func (b *ExecBackend) spawnChildFrom(ctx context.Context, rt registry.Route, sock string, n uint64, exe string, tier cgroup.Limit) (Proc, string, error) { |
| 2055 | host := rt.Host |
| 2056 | paths := b.pathsFor(rt) |
| 2057 | dbPath := paths.dbPath |
| 2058 | child, ownTier, err := b.placeFor(spawnInstance, n) |
| 2059 | if err != nil { |
| 2060 | return nil, "", err |
| 2061 | } |
| 2062 | |
| 2063 | // Exec exactly what a systemd unit would: the per-instance live symlink |
| 2064 | // (<OptDir>/live/<host> → a staged version), same --socket/--db flags, |
| 2065 | // same EnvironmentFile. |
| 2066 | bin := exe |
| 2067 | if bin == "" { |
| 2068 | bin = b.liveLink(host) |
| 2069 | if err := b.ensureLiveLink(ctx, host); err != nil { |
| 2070 | return nil, "", err |
| 2071 | } |
| 2072 | } |
| 2073 | |
| 2074 | // Mint the instance's admin token BEFORE the env is read for the spawn, so |
| 2075 | // the child is handed the same secret the agent will present back to it |
| 2076 | // (NextDue, the scheduled-work tick) and the same one its control socket |
| 2077 | // checks bearers against. A failure here fails the wake on purpose: an |
| 2078 | // instance with no token can neither answer /api/next-due nor accept a |
| 2079 | // tick, and a wake that quietly skipped it would report a healthy instance |
| 2080 | // that silently never runs its scheduled work — the #110 kind of lie. |
| 2081 | if err := EnsureAdminToken(paths.envFile); err != nil { |
| 2082 | return nil, "", fmt.Errorf("admin token for %s: %w", host, err) |
| 2083 | } |
| 2084 | // The instance key is minted here for the same reason and fails the |
| 2085 | // wake the same way: an app that takes its key from the platform |
| 2086 | // refuses to start without one (L21). It returns only once the key is |
| 2087 | // durable, so no child starts with a key a power cut could take back. |
| 2088 | if _, err := b.ensureInstanceKeyFor(ctx, rt, paths); err != nil { |
| 2089 | return nil, "", fmt.Errorf("instance key for %s: %w", host, err) |
| 2090 | } |
| 2091 | |
| 2092 | // One read gives the env and the layer its CARLOS_INSTANCE_KEY came |
| 2093 | // from, for the replica gate below (ruling 29). |
| 2094 | fileEnv, keyLayer, err := b.instanceEnvLayer(rt) |
| 2095 | if err != nil { |
| 2096 | return nil, "", err |
| 2097 | } |
| 2098 | |
| 2099 | // The control socket reaches the child through its ENVIRONMENT, never as a |
| 2100 | // CLI flag: every app binary built before this feature existed parses its |
| 2101 | // flags with flag.ExitOnError, which exits 2 on an unrecognized flag — so |
| 2102 | // passing --control-socket would refuse to start every app that hasn't |
| 2103 | // also shipped support for it, a fleet-wide outage on agent upgrade. An |
| 2104 | // environment variable a binary doesn't know about is simply ignored. |
| 2105 | controlSocket := sock + ".control" |
| 2106 | // Bind the control channel BEFORE the spawn, so CARLOS_CONTROL_SOCKET can |
| 2107 | // follow the OUTCOME rather than the intention. It used to be set from |
| 2108 | // "b.Cache != nil" and the listener started afterwards; now that the |
| 2109 | // channel exists for every exec child, the only thing that can make the |
| 2110 | // variable a lie is a bind that failed (windows has no secureUnixListen at |
| 2111 | // all, and a stale path can refuse), and binding first is what lets the |
| 2112 | // else-branch below strip the name in exactly that case. |
| 2113 | // |
| 2114 | // A bind failure is still NOT fatal to the wake — same reasoning as before, |
| 2115 | // the instance serves requests either way; it just goes without its control |
| 2116 | // verbs until the next restart, a stuck-stale cache self-heals via the |
| 2117 | // safety-net TTL, and a missed tick is retried by the runner. |
| 2118 | envPath := paths.envFile |
| 2119 | // Read on EVERY call, never captured: the env file can be rewritten under a |
| 2120 | // running instance (`carlos env`, a re-mint on a later spawn), and a token |
| 2121 | // cached at bind time would keep authorising the old secret. |
| 2122 | tokenReader := func() (string, error) { |
| 2123 | env, err := readEnvFile(envPath) |
| 2124 | if err != nil { |
| 2125 | return "", err |
| 2126 | } |
| 2127 | return env["CARLOS_ADMIN_TOKEN"], nil |
| 2128 | } |
| 2129 | var mount func(*http.ServeMux) |
| 2130 | if b.Timers != nil || b.UpdatesReader != nil { |
| 2131 | mount = func(mux *http.ServeMux) { |
| 2132 | if b.Timers != nil { |
| 2133 | schedule.MountGated(mux, rt.Host, b.Timers, tokenReader, time.Now, b.writeFence(rt.Host, n)) |
| 2134 | } |
| 2135 | if b.UpdatesReader != nil { |
| 2136 | instanceupdates.Mount(mux, rt.Host, tokenReader, b.UpdatesReader) |
| 2137 | } |
| 2138 | } |
| 2139 | } |
| 2140 | ctl, err := edge.ListenControl(controlSocket, rt.Host, b.Cache, mount) |
| 2141 | if err != nil { |
| 2142 | log.Printf("activator: control listener for %s: %v (continuing without it)", rt.Host, err) |
| 2143 | ctl = nil |
| 2144 | } |
| 2145 | // A replica is started only with its home's key, verified, or with |
| 2146 | // none and a control socket its helper can refuse on (ruling 26). |
| 2147 | if err := b.replicaKeyGate(rt, fileEnv, keyLayer, ctl != nil); err != nil { |
| 2148 | if ctl != nil { |
| 2149 | ctl.Close() |
| 2150 | } |
| 2151 | return nil, "", err |
| 2152 | } |
| 2153 | // Both control-channel vars are AGENT-OWNED in BOTH directions: set when the |
| 2154 | // channel exists, and REMOVED when it does not. The removal half is not |
| 2155 | // decoration. fileEnv is tenant-supplied — <host>.config.env comes from the |
| 2156 | // bucket via `carlos env`, <host>.env is the instance's own file — so |
| 2157 | // without the else branches a tenant could put CARLOS_PURGE=header (or worse, |
| 2158 | // CARLOS_CONTROL_SOCKET=/some/path) in its own config and have it survive |
| 2159 | // into the child on a box with NO cache. The app would then be told a purge |
| 2160 | // channel exists when none does: an app that trusts the var skips its own |
| 2161 | // invalidation and serves stale content forever, and a socket path a tenant |
| 2162 | // chose is a name the agent never vouched for. |
| 2163 | // |
| 2164 | // "A variable naming a channel must not be set where that channel does |
| 2165 | // nothing" still holds — the two names just answer to different channels |
| 2166 | // now. CARLOS_CONTROL_SOCKET names the SOCKET, which exists for every exec |
| 2167 | // child once the bind above succeeded (the purge verb is only one of the |
| 2168 | // things served on it; the scheduled-work verbs are served there on a box |
| 2169 | // with no cache at all). CARLOS_PURGE names the PURGE channel, which still |
| 2170 | // exists only where there is a cache to purge. Both are set from what was |
| 2171 | // actually created, and stripped otherwise; neither is set from a tenant's |
| 2172 | // say-so. |
| 2173 | // |
| 2174 | // CARLOS_PURGE is ADVISORY: it tells the app that this box's edge reads the |
| 2175 | // Carlos-Purge response header (internal/edge/purge_header.go), so the app |
| 2176 | // can log or branch on it — but the channel works whether or not the app |
| 2177 | // ever reads the variable, because a header the edge understands is inert |
| 2178 | // when the app never emits it and honoured when it does. That is what makes |
| 2179 | // the rollout ordering harmless. It does not replace the socket channel; |
| 2180 | // both are live for exec-backed instances, and converging them is deferred |
| 2181 | // (spec §10.3). |
| 2182 | childVars := b.agentOwnedVars(rt, controlSocket, fileEnv, ctl != nil) |
| 2183 | // Record the build we're about to exec — read the symlink immediately before |
| 2184 | // the start so it names the version the kernel will resolve. Reading it here, |
| 2185 | // rather than back in wake() after the health wait, closes a real window: a |
| 2186 | // concurrent ship's repoint during the up-to-30s wait would otherwise make |
| 2187 | // us record the NEW target while this child runs the OLD binary, and the |
| 2188 | // sweep would then think it's already converged and never cycle it. The |
| 2189 | // residual sub-exec race (a repoint between this read and the start) is in |
| 2190 | // the safe direction — we'd record the older version, so the sweep cycles |
| 2191 | // once too often rather than never. Empty on a read error: the sweep won't |
| 2192 | // cycle. |
| 2193 | // |
| 2194 | // A staged spawn names its build from the attempt; the caller fills |
| 2195 | // Gen.Build. Reading the live link here would name whatever root |
| 2196 | // installed, not what this process runs. |
| 2197 | // |
| 2198 | // With LiveExecGate set (a declined box that honours failure records or |
| 2199 | // hold-ran markers, Task 9), the live link is resolved ONCE here, path |
| 2200 | // only, with no hash (release.ResolveLivePath, fix round 1 ruling 4b), |
| 2201 | // the gate is asked about that build at the |
| 2202 | // launch, and the staged file the link named is what is exec'd, not the |
| 2203 | // link. So the build the gate passed is the build that runs: a repoint |
| 2204 | // between the gate and the exec changes nothing, where exec'ing the link |
| 2205 | // would run whatever it named by then (plan review #6). A link that |
| 2206 | // cannot be resolved fails the wake, as a missing one does. With the gate |
| 2207 | // nil this is today's read and today's exec of the link. |
| 2208 | spawnVer := "" |
| 2209 | var liveGate func() error |
| 2210 | if exe == "" && b.LiveExecGate != nil { |
| 2211 | id, err := release.ResolveLivePath(b.OptDir, host) |
| 2212 | if err != nil { |
| 2213 | if ctl != nil { |
| 2214 | ctl.Close() |
| 2215 | } |
| 2216 | return nil, "", fmt.Errorf("%s: %w", host, err) |
| 2217 | } |
| 2218 | bin, spawnVer = id.StagedPath, id.Version |
| 2219 | liveGate = func() error { return b.LiveExecGate(rt, id.Version) } |
| 2220 | } else if exe == "" { |
| 2221 | spawnVer, _ = b.LiveVersion(host) |
| 2222 | } |
| 2223 | // Spawned through the cgroup manager (nil-tolerant: no delegation means |
| 2224 | // a plain start, exactly as before) so the child lands in |
| 2225 | // carlos-exec@<host> and the sampler can account it. build may run |
| 2226 | // twice — a flat start's PID-move fallback after a failed placement — |
| 2227 | // so everything about the Cmd is configured inside it. |
| 2228 | // |
| 2229 | // The caller's exec gate (final review F1) is the manager's before-start |
| 2230 | // hook (launchGate): asked after every step above (the token, the config |
| 2231 | // read, the control socket's bind) AND the cgroup preparation (the |
| 2232 | // parent, the child, its limits, and the shim's placement), immediately |
| 2233 | // before each exec, the flat PID-move fallback's included. The lease, a |
| 2234 | // fence or the attempt can change during any of it. A refusal runs |
| 2235 | // nothing of the command; the manager drops |
| 2236 | // the child cgroup and the error path below unbinds the listener, as for |
| 2237 | // a failed spawn. With no gate on ctx (every live-link spawn) there is no |
| 2238 | // hook and the spawn is exactly as it was, unless LiveExecGate is set: |
| 2239 | // then its check of the resolved build (liveGate) is the hook, asked at |
| 2240 | // the same point. |
| 2241 | cmd, settle, err := b.spawnInCgroupAtTier(rt, child, ownTier, tier, func() *exec.Cmd { |
| 2242 | cmd := exec.Command(bin, "--socket", sock, "--db", dbPath) |
| 2243 | cmd.Env = childEnv(childVars) |
| 2244 | // The child logs through the agent's own stderr/stdout (journald |
| 2245 | // under systemd); there's no per-instance unit to own a log stream |
| 2246 | // anymore. With Logs set the same bytes are ALSO teed line-wise |
| 2247 | // into the log sink (`carlos logs`); teeStd owns the WaitDelay |
| 2248 | // that piping entails. |
| 2249 | b.teeStd(cmd, rt) |
| 2250 | return cmd |
| 2251 | }, b.launchGateWith(ctx, host, liveGate)) |
| 2252 | if err != nil { |
| 2253 | // The listener was bound before the spawn; nothing will ever answer on |
| 2254 | // it now, so unbind it rather than leaving a live socket for a process |
| 2255 | // that does not exist. |
| 2256 | if ctl != nil { |
| 2257 | ctl.Close() |
| 2258 | } |
| 2259 | return nil, "", err |
| 2260 | } |
| 2261 | |
| 2262 | p := &childProc{cmd: cmd, done: make(chan struct{}), ctl: ctl} |
| 2263 | // A staged successor spawns while the incumbent still serves, and its |
| 2264 | // purge happens at the switch (GenSwitched), not here. |
| 2265 | if b.Cache != nil && exe == "" { |
| 2266 | // A fresh process for this host is exactly the moment its cached |
| 2267 | // content becomes suspect: a deploy/redeploy means new code and |
| 2268 | // potentially new pages, and nothing else would invalidate them until |
| 2269 | // their TTL ran out. Drop the whole host's entries before the new |
| 2270 | // process serves its first request. |
| 2271 | if n := b.Cache.PurgeHost(rt.Host); n > 0 { |
| 2272 | log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host) |
| 2273 | } |
| 2274 | } |
| 2275 | go func() { |
| 2276 | p.err = cmd.Wait() |
| 2277 | // The member-facing record of the exit: a crash shows an "exited" |
| 2278 | // line with no "stopping" line before it — that pairing, not the |
| 2279 | // exit code alone, is what tells a member crash from lifecycle. |
| 2280 | if p.err != nil { |
| 2281 | emitEvent(b.Logs, rt, "warn", "instance exited: "+RedactPaths(p.err.Error())) |
| 2282 | } else { |
| 2283 | emitEvent(b.Logs, rt, "info", "instance exited (exit 0)") |
| 2284 | } |
| 2285 | // A strict child's group is drained and closed before anything |
| 2286 | // reads this exit as a stop (p.done closes below). |
| 2287 | settle() |
| 2288 | b.Cgroups.RemoveChild(host, child) // no-op for "" (flag off) |
| 2289 | // Last exiter wins: rmdir refuses while the sidecar (same host |
| 2290 | // cgroup) is still alive, and that failure is tolerated. |
| 2291 | b.cleanupHost(host) |
| 2292 | close(p.done) |
| 2293 | }() |
| 2294 | return p, spawnVer, nil |
| 2295 | } |
| 2296 | |
| 2297 | // agentOwnedNames are the variables agentOwnedVars decides: it sets or |
| 2298 | // removes each of them on every launch, whatever the instance's files say. |
| 2299 | // moves.AgentOwnedVars is the same set (TestAgentOwnedVarsAreMovesBoxOwned |
| 2300 | // pins both against agentOwnedVars itself). |
| 2301 | var agentOwnedNames = []string{"CARLOS_CONTROL_SOCKET", "CARLOS_PURGE", "CARLOS_REGION", "CARLOS_HOST"} |
| 2302 | |
| 2303 | // agentOwnedVars composes the child's agent-owned environment variables, |
| 2304 | // namely the control socket, the purge channel, and the region, on top of |
| 2305 | // fileEnv, exactly as startChild did inline before the unit spawn path |
| 2306 | // needed the same composition. Extracted so the two paths cannot drift |
| 2307 | // (design §7.2): haveControl is whether this call's control listener |
| 2308 | // actually bound. childVars must follow the OUTCOME of the bind, not the |
| 2309 | // intention, per the long comment this replaced, still true here. |
| 2310 | func (b *ExecBackend) agentOwnedVars(rt registry.Route, control string, fileEnv map[string]string, haveControl bool) map[string]string { |
| 2311 | childVars := fileEnv |
| 2312 | if haveControl { |
| 2313 | childVars = withVar(childVars, "CARLOS_CONTROL_SOCKET", control) |
| 2314 | } else { |
| 2315 | childVars = withoutVars(childVars, "CARLOS_CONTROL_SOCKET") |
| 2316 | } |
| 2317 | if b.Cache != nil { |
| 2318 | childVars = withVar(childVars, "CARLOS_PURGE", "header") |
| 2319 | } else { |
| 2320 | childVars = withoutVars(childVars, "CARLOS_PURGE") |
| 2321 | } |
| 2322 | if b.Region != "" { |
| 2323 | childVars = withVar(childVars, "CARLOS_REGION", b.Region) |
| 2324 | } else { |
| 2325 | childVars = withoutVars(childVars, "CARLOS_REGION") |
| 2326 | } |
| 2327 | // The hostname, always, from the row (discussion #256 item 2). An app |
| 2328 | // used to learn it from its socket's file name, /run/carlos/<host>.sock, |
| 2329 | // and the nested layout's app.sock names nothing; three instances |
| 2330 | // crashed at start over that on the first roll. Agent-owned like the |
| 2331 | // three above: a tenant's env file does not get to tell an app it is |
| 2332 | // some other host. |
| 2333 | return withVar(childVars, "CARLOS_HOST", rt.Host) |
| 2334 | } |
| 2335 | |
| 2336 | // seedVars is the seed's environment over fileEnv: the control-channel |
| 2337 | // variables stripped (a seed has no listener for either; see seedChild's |
| 2338 | // comment), and CARLOS_HOST set, since a seed initialising a host's |
| 2339 | // database is as entitled to know which host as the instance is. One |
| 2340 | // composer for both seed paths, as agentOwnedVars is for both run paths. |
| 2341 | func seedVars(fileEnv map[string]string, host string) map[string]string { |
| 2342 | return withVar(withoutVars(fileEnv, "CARLOS_CONTROL_SOCKET", "CARLOS_PURGE"), "CARLOS_HOST", host) |
| 2343 | } |
| 2344 | |
| 2345 | // normalizeSocketMode is the PLATFORM's 0660 on a spawned instance's socket |
| 2346 | // (issue #103), not the app's. Extracted from startChild so the unit path |
| 2347 | // (whose own bind is the trampoline's, not this process's exec) applies the |
| 2348 | // identical rule rather than a second copy that can drift. So a deployment's |
| 2349 | // enrolled cross-user consumers (the carlos-sockets group, group-inherited |
| 2350 | // from the setgid /run/carlos, see host.tmpfilesConf) can dial, no matter |
| 2351 | // what the child's umask or its own post-bind chmod produced. titogo |
| 2352 | // tightened its socket to 0600 and stranded the sibling router that proxies |
| 2353 | // to it. Called after the health wait, so the app's own bind/chmod has |
| 2354 | // already happened and this wins. |
| 2355 | // |
| 2356 | // The two guards make "only a socket THIS backend's child bound" true |
| 2357 | // rather than merely intended. waitHealthy tolerates a squatter (its own |
| 2358 | // contract: it only dials), so a child that lost the bind race with a |
| 2359 | // stray supervisor (EADDRINUSE, exits) can still see the health wait |
| 2360 | // succeed off the SQUATTER's answers; a dead child means the socket is |
| 2361 | // not ours to widen (that route is the #17 external-socket advisory's |
| 2362 | // territory). The Lstat keeps the chmod from ever following a |
| 2363 | // non-socket: a symlink planted at the route's path must not become a |
| 2364 | // mode change somewhere else. On a box with no consumers group the |
| 2365 | // socket's group is just `carlos`, so the wider mode grants nobody |
| 2366 | // anything. All three branches log and never fail the wake: a box with |
| 2367 | // no consumers must not lose a healthy wake to a normalization it |
| 2368 | // doesn't need. |
| 2369 | func (b *ExecBackend) normalizeSocketMode(rt registry.Route, sock string, proc Proc) { |
| 2370 | host := rt.Host |
| 2371 | if proc.Exited() { |
| 2372 | log.Printf("activator: %s: child exited during the health wait (external process serving its socket?) — not normalizing the socket mode", host) |
| 2373 | } else if fi, statErr := os.Lstat(sock); statErr != nil || fi.Mode()&os.ModeSocket == 0 { |
| 2374 | log.Printf("activator: %s: %s is not a socket we can normalize (stat: %v)", host, sock, statErr) |
| 2375 | } else if err := os.Chmod(sock, 0o660); err != nil { |
| 2376 | log.Printf("activator: normalize socket mode for %s: %v (enrolled consumers would get permission denied)", host, err) |
| 2377 | } |
| 2378 | } |
| 2379 | |
| 2380 | // spawnUnitFrom is the nested spawn with the executable and the tier named, |
| 2381 | // and nothing after root has started the unit: it returns the moment the |
| 2382 | // unit runs, as spawnChildFrom does for a flat host. A nil exe is the live |
| 2383 | // link, which is every wake (startUnitAt adds today's health wait). A staged |
| 2384 | // exe is a swap's successor or a pinned wake (zero-downtime spec §7.5 step |
| 2385 | // 3): root validates the path and mounts it in place of the live link, so |
| 2386 | // whatever happens to the live link meanwhile changes nothing, and the |
| 2387 | // caller owns the health deadline and the termination. tier is that |
| 2388 | // artifact's own tier; zero is the live tier (b.tier). With LiveExecGate |
| 2389 | // set a nil exe pins the live link (declinedLivePin) and starts that file as |
| 2390 | // a staged one, at its own tier. |
| 2391 | // |
| 2392 | // n is the generation this spawn is (StartGen's n; 0 from Start). Under |
| 2393 | // GenContainment it selects the carlos-instance-gen@<id>:<n> unit and its |
| 2394 | // generation-keyed files (unitGen); otherwise it is ignored. A failure |
| 2395 | // whose kill could not prove the unit empty wraps errGenUnproven. |
| 2396 | func (b *ExecBackend) spawnUnitFrom(ctx context.Context, rt registry.Route, sock string, n uint64, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, string, error) { |
| 2397 | // The launch's own split, for the journal (wake latency, pass 2). |
| 2398 | var steps wakeSteps |
| 2399 | steps.begin(time.Now()) |
| 2400 | if err := b.nestedGate(rt); err != nil { |
| 2401 | return nil, "", err |
| 2402 | } |
| 2403 | gen, err := b.unitGen(n) |
| 2404 | if err != nil { |
| 2405 | return nil, "", err |
| 2406 | } |
| 2407 | pinned := "" |
| 2408 | if exe == nil && b.LiveExecGate != nil { |
| 2409 | // A declined box that honours failure records or hold-ran markers |
| 2410 | // (Task 9) pins the live link once and names that file to root at |
| 2411 | // its own tier (declinedLivePin), so root binds the build the gate |
| 2412 | // passed and a repoint after this read changes nothing. Before |
| 2413 | // anything is prepared: a refusal binds no listener and writes no |
| 2414 | // launch file. |
| 2415 | id, err := b.declinedLivePin(ctx, rt) |
| 2416 | if err != nil { |
| 2417 | return nil, "", err |
| 2418 | } |
| 2419 | exe, tier, pinned = stagedExeOf(id), b.GenTier(rt, id.Version), id.Version |
| 2420 | } else if exe == nil { |
| 2421 | // A staged exe is mounted in place of the live link, which then |
| 2422 | // need not exist (as spawnChildFrom does for a flat host). |
| 2423 | if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| 2424 | return nil, "", err |
| 2425 | } |
| 2426 | } |
| 2427 | steps.mark(time.Now(), "build") |
| 2428 | var prep wakeSteps |
| 2429 | p, ctl, logs, fileEnv, keyLayer, err := b.prepareKeyedAt(rt, instancectl.RoleRun, sock, gen, &prep) |
| 2430 | if err != nil { |
| 2431 | return nil, "", err |
| 2432 | } |
| 2433 | fail := func(err error) (Proc, string, error) { |
| 2434 | if ctl != nil { |
| 2435 | ctl.Close() |
| 2436 | } |
| 2437 | logs.Close() |
| 2438 | return nil, "", err |
| 2439 | } |
| 2440 | childVars := b.agentOwnedVars(rt, sock+".control", fileEnv, ctl != nil) |
| 2441 | argv := []string{b.liveLink(rt.Host), "--socket", sock, "--db", p.dbPath} |
| 2442 | steps.mark(time.Now(), "prepare ("+prep.String()+")") |
| 2443 | if err := b.writeLaunchFor(p, instancectl.RoleRun, gen, rt.Host, argv, childEnv(childVars)); err != nil { |
| 2444 | return fail(err) |
| 2445 | } |
| 2446 | steps.mark(time.Now(), "launch file") |
| 2447 | // Read the symlink immediately before the start, same fail-safe |
| 2448 | // direction as startChild: a concurrent repoint at worst records the |
| 2449 | // older version (one extra convergence cycle), never the newer. A |
| 2450 | // pinned start names the build it pinned, which is what root binds. |
| 2451 | spawnVer := pinned |
| 2452 | if spawnVer == "" { |
| 2453 | spawnVer, _ = b.LiveVersion(rt.Host) |
| 2454 | } |
| 2455 | // The caller's exec gate (final review F1), right before root is asked |
| 2456 | // to start the unit: the config read, the listeners and the launch file |
| 2457 | // above all take time. A refusal starts nothing; the generation's own |
| 2458 | // launch file (it carries the instance's environment) goes, and the |
| 2459 | // listeners close, as after root's own refusal (ErrStagedExe below). |
| 2460 | // With no gate on ctx (every live-link spawn) this does nothing. |
| 2461 | if b.launchHook != nil { |
| 2462 | b.launchHook(rt.Host) |
| 2463 | } |
| 2464 | err = runExecGate(ctx) |
| 2465 | if err != nil { |
| 2466 | if gen != 0 { |
| 2467 | _ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen))) |
| 2468 | } |
| 2469 | return fail(err) |
| 2470 | } |
| 2471 | steps.mark(time.Now(), "gate") |
| 2472 | inv, err := b.startRole(ctx, rt, instancectl.RoleRun, gen, exe, tier) |
| 2473 | steps.mark(time.Now(), "root start") |
| 2474 | // Past this call root may have started the unit: every failure below |
| 2475 | // but ErrStagedExe (root's refusal before it laid out or started |
| 2476 | // anything) and errNoStagedExeControl (root was never asked: the |
| 2477 | // channel cannot name the file) is marked startUncertain, since the |
| 2478 | // build may have run before it failed or exited (final review wave 2, |
| 2479 | // N4). |
| 2480 | if errors.Is(err, errNoStagedExeControl) { |
| 2481 | // Nothing was sent to root. Proven not run, so never a failure |
| 2482 | // record. A generation's own launch file goes as after a refusal. |
| 2483 | if gen != 0 { |
| 2484 | _ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen))) |
| 2485 | } |
| 2486 | return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err)) |
| 2487 | } |
| 2488 | if errors.Is(err, instancectl.ErrStagedExe) { |
| 2489 | // Root refused the staged build before it laid out or started |
| 2490 | // anything, so there is nothing to kill and nothing unproven. The |
| 2491 | // generation's own launch file, which this start wrote and which |
| 2492 | // carries the instance's environment, goes too (final review T9). |
| 2493 | // The legacy launch.run.json (gen 0) is shared with the original |
| 2494 | // template and is left as it is. |
| 2495 | if gen != 0 { |
| 2496 | _ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen))) |
| 2497 | } |
| 2498 | return fail(fmt.Errorf("%s: start unit: %w", rt.Host, err)) |
| 2499 | } |
| 2500 | if gen != 0 && (err != nil || inv == "") { |
| 2501 | if err == nil { |
| 2502 | // The unit ran and exited before root observed it: the build |
| 2503 | // crashed at its start (errStartedThenExited, ruling 206). |
| 2504 | err = exitedUnobserved{unit: unitName(instancectl.RoleRun, rt.Host, gen)} |
| 2505 | } |
| 2506 | if kerr := b.killFailedGen(ctx, rt, gen); kerr != nil { |
| 2507 | // Nothing is released: the generation may still be running. |
| 2508 | return nil, "", markStartUncertain(fmt.Errorf("%s: start unit: %w; %w: %v", rt.Host, err, errGenUnproven, kerr)) |
| 2509 | } |
| 2510 | _ = os.Remove(filepath.Join(p.dataDir, instancectl.GenLaunchFileName(gen))) |
| 2511 | return fail(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err))) |
| 2512 | } |
| 2513 | if err != nil { |
| 2514 | // The original run unit (generation 0). Root was asked to start |
| 2515 | // it, and a start job systemd queued (behind an edge restart, say) |
| 2516 | // outlives this call's cancellation, so the wake's cleanup stops |
| 2517 | // the unit and proves it gone before the lease goes (#292). |
| 2518 | return fail(markLegacyStartUncertain(markStartUncertain(fmt.Errorf("%s: start unit: %w", rt.Host, err)))) |
| 2519 | } |
| 2520 | if inv == "" { |
| 2521 | // The server read the invocation after the start and found none: |
| 2522 | // the unit already exited (plan 2 final review). Nothing is |
| 2523 | // running, so there is nothing to stop; the wake failed. |
| 2524 | return fail(markStartUncertain(fmt.Errorf("%s: %w", rt.Host, exitedUnobserved{unit: unitName(instancectl.RoleRun, rt.Host, gen)}))) |
| 2525 | } |
| 2526 | started := tier |
| 2527 | if started == (cgroup.Limit{}) { |
| 2528 | started = b.tier(rt) |
| 2529 | } |
| 2530 | log.Printf("activator: %s: started %s memory.max=%d (%s)", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen), started.MemoryMax, steps.String()) |
| 2531 | up := b.newUnitProcFor(rt, instancectl.RoleRun, gen, p, inv, ctl, logs) |
| 2532 | // The file environment this generation launched with, kept on its |
| 2533 | // handle: a run against it composes its environment from this, never |
| 2534 | // from the files config sync has delivered since (merge ruling M3). |
| 2535 | // It carries the generation's CARLOS_INSTANCE_KEY, minted or |
| 2536 | // configured before this launch, and the layer that key came from is |
| 2537 | // kept with it, decided by the same read (instance-key ruling 29), so |
| 2538 | // a run's command passes the replica gate on exactly what it is given. |
| 2539 | up.launchEnv, up.launchKeyLayer = maps.Clone(fileEnv), keyLayer |
| 2540 | if b.Cache != nil { |
| 2541 | // A fresh process for this host is exactly the moment its cached |
| 2542 | // content becomes suspect, same rule as startChild. |
| 2543 | if n := b.Cache.PurgeHost(rt.Host); n > 0 { |
| 2544 | log.Printf("activator: purged %d cached response(s) for %s on spawn", n, rt.Host) |
| 2545 | } |
| 2546 | } |
| 2547 | return up, spawnVer, nil |
| 2548 | } |
| 2549 | |
| 2550 | // startUnitAt is Start's nested body (design §6): the same admin token, env |
| 2551 | // composition and health gate as startChild, but the process itself is |
| 2552 | // started by the root instance-control channel from a launch file rather |
| 2553 | // than exec'd directly. The agent has no privilege to enter the tenant's |
| 2554 | // namespace, only the control service does. It is spawnUnitFrom, then the |
| 2555 | // health wait. exe is nil for the live link, or the staged file a wake |
| 2556 | // recorded (StartGenExe), which root binds in place of the link. tier is |
| 2557 | // exe's own; zero is the live tier (b.tier). |
| 2558 | func (b *ExecBackend) startUnitAt(ctx context.Context, rt registry.Route, sock string, n uint64, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, string, error) { |
| 2559 | began := time.Now() |
| 2560 | p, spawnVer, err := b.spawnUnitFrom(ctx, rt, sock, n, exe, tier) |
| 2561 | if err != nil { |
| 2562 | return p, "", err |
| 2563 | } |
| 2564 | up := p.(*unitProc) |
| 2565 | launched := time.Now() |
| 2566 | if err := waitHealthyCtx(ctx, sock, spawnHealthTimeout); err != nil { |
| 2567 | // Set before the Stop call below, closing the same race Stop's own |
| 2568 | // internal set leaves open by a few instructions: observe runs |
| 2569 | // concurrently from the moment newUnitProc started it, and setting |
| 2570 | // this here rather than waiting for it to happen inside up.Stop |
| 2571 | // means the observer can never see a foreign invocation appear and |
| 2572 | // fire its own redundant stop for a unit we already know we are |
| 2573 | // about to stop. |
| 2574 | up.stopRequested.Store(true) |
| 2575 | stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| 2576 | defer cancel() |
| 2577 | if serr := up.Stop(stopCtx); serr != nil { |
| 2578 | // The health error, not this one, is the wake's failure. The |
| 2579 | // observer stands down on its own once the stop actually |
| 2580 | // lands (stopRequested above), so an ErrStillRunning here |
| 2581 | // just means it had not landed yet within actStopTimeout, not |
| 2582 | // that the wake should be retried differently. |
| 2583 | log.Printf("activator: %s: stop after failed health wait: %v", rt.Host, serr) |
| 2584 | // The unit may still run, so its process goes back with the |
| 2585 | // health error and the caller's failed-wake cleanup proves it |
| 2586 | // stopped before the lease goes (P2-13). Under GenContainment |
| 2587 | // the error also says so (errGenUnproven), so that cleanup |
| 2588 | // asks for the proof with generations off as well. |
| 2589 | if up.gen != 0 { |
| 2590 | return up, "", fmt.Errorf("%w; %w: stop after failed health wait: %v", err, errGenUnproven, serr) |
| 2591 | } |
| 2592 | return up, "", err |
| 2593 | } |
| 2594 | return nil, "", err |
| 2595 | } |
| 2596 | // The spawn's own split, for the journal: the unit's launch (instance |
| 2597 | // control's half included) and then the app's boot to its socket. |
| 2598 | log.Printf("activator: %s: %s launched in %s, healthy %s later", rt.Host, unitName(instancectl.RoleRun, rt.Host, n), |
| 2599 | launched.Sub(began).Round(time.Millisecond), time.Since(launched).Round(time.Millisecond)) |
| 2600 | b.normalizeSocketMode(rt, sock, up) |
| 2601 | return up, spawnVer, nil |
| 2602 | } |
| 2603 | |
| 2604 | // LegacyRunGone stops rt's original run unit (carlos-instance@, generation |
| 2605 | // 0) and proves it gone (#292, Astra P1): the failed-wake cleanup's proof |
| 2606 | // before an aborted start's lease goes back. The stop goes through |
| 2607 | // instance-control's /stop, which runs `systemctl stop`. systemd keeps one |
| 2608 | // job per unit and a stop job replaces a queued start job, so a successful |
| 2609 | // stop also means that no start job is still queued, even while the edge |
| 2610 | // restart that queued it is in progress. Then the unit must read inactive |
| 2611 | // (or failed) with its cgroup empty or gone, the facts the exit observer |
| 2612 | // trusts (observe, design §6.5). Bounded by ctx, which the caller gives a |
| 2613 | // budget of its own; anything short of the proof is an error, and the |
| 2614 | // caller keeps the lease. |
| 2615 | func (b *ExecBackend) LegacyRunGone(ctx context.Context, rt registry.Route) error { |
| 2616 | if b.Control == nil { |
| 2617 | return fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| 2618 | } |
| 2619 | unit := unitName(instancectl.RoleRun, rt.Host, 0) |
| 2620 | if err := b.Control.Stop(ctx, rt.Host, instancectl.RoleRun); err != nil { |
| 2621 | return fmt.Errorf("stop %s: %w", unit, err) |
| 2622 | } |
| 2623 | state := b.UnitState |
| 2624 | if state == nil { |
| 2625 | state = systemctlUnitState |
| 2626 | } |
| 2627 | events := b.CgroupEvents |
| 2628 | if events == nil { |
| 2629 | events = cgroupEventsUnder(b.cgroupRoot()) |
| 2630 | } |
| 2631 | poll := b.ObservePoll |
| 2632 | if poll <= 0 { |
| 2633 | poll = time.Second |
| 2634 | } |
| 2635 | slice := instancectl.SliceName(rt.Host) |
| 2636 | var last error |
| 2637 | for { |
| 2638 | active, _, err := state(ctx, unit) |
| 2639 | switch { |
| 2640 | case err != nil: |
| 2641 | last = fmt.Errorf("unit state: %w", err) |
| 2642 | case active != "inactive" && active != "failed": |
| 2643 | last = fmt.Errorf("unit is %s", active) |
| 2644 | default: |
| 2645 | exists, populated, eerr := events(slice, unit) |
| 2646 | switch { |
| 2647 | case eerr != nil: |
| 2648 | last = fmt.Errorf("cgroup.events: %w", eerr) |
| 2649 | case exists && populated: |
| 2650 | last = errors.New("processes remain in its cgroup") |
| 2651 | default: |
| 2652 | return nil |
| 2653 | } |
| 2654 | } |
| 2655 | select { |
| 2656 | case <-ctx.Done(): |
| 2657 | return fmt.Errorf("%s not proven stopped: %v", unit, last) |
| 2658 | case <-time.After(poll): |
| 2659 | } |
| 2660 | } |
| 2661 | } |
| 2662 | |
| 2663 | // killFailedGen ends a generation whose start failed or answered no |
| 2664 | // invocation. The start may have succeeded with its answer lost, so the |
| 2665 | // generation is killed and proven empty (StopGen with kill) on a budget the |
| 2666 | // caller's context cannot cancel. nil is that proof. With no generation |
| 2667 | // channel StartGen was never sent, so nothing was started. |
| 2668 | func (b *ExecBackend) killFailedGen(ctx context.Context, rt registry.Route, gen uint64) error { |
| 2669 | gc, err := b.genControl() |
| 2670 | if err != nil { |
| 2671 | return nil |
| 2672 | } |
| 2673 | kctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), instancectl.DeadlineControl) |
| 2674 | defer cancel() |
| 2675 | if err := gc.StopGen(kctx, rt.Host, gen, true); err != nil { |
| 2676 | log.Printf("activator: %s: kill %s after a failed start: %v", rt.Host, unitName(instancectl.RoleRun, rt.Host, gen), err) |
| 2677 | return err |
| 2678 | } |
| 2679 | return nil |
| 2680 | } |
| 2681 | |
| 2682 | // nestedGate is the pre-flight check every nested spawn path runs first: no |
| 2683 | // root instance-control channel configured, or the box itself is not fit |
| 2684 | // to run an isolated instance (NestedOK's own mounts and /ready check), |
| 2685 | // refuses before anything else runs. Shared so seedUnit and |
| 2686 | // startSidecarUnit gate on NestedOK exactly like startUnit does. |
| 2687 | func (b *ExecBackend) nestedGate(rt registry.Route) error { |
| 2688 | if b.Control == nil { |
| 2689 | return fmt.Errorf("%s: instance control channel not configured", rt.Host) |
| 2690 | } |
| 2691 | if b.NestedOK != nil { |
| 2692 | if err := b.NestedOK(); err != nil { |
| 2693 | return fmt.Errorf("%s: refusing to wake an isolated instance: %w", rt.Host, err) |
| 2694 | } |
| 2695 | } |
| 2696 | return nil |
| 2697 | } |
| 2698 | |
| 2699 | // instanceEnv composes the host's two EnvironmentFile layers, both optional |
| 2700 | // like a unit's "-" prefix: the sync-owned bucket config (<host>.config.env) |
| 2701 | // underneath, the instance's own file (<host>.env) on top — it wins conflicts, |
| 2702 | // see mergeEnv. Shared by Start and Seed precisely so a seed sees the SAME |
| 2703 | // environment the instance will: a seed that read different credentials, or a |
| 2704 | // different database URL, than the process it initializes would be a trap. |
| 2705 | // (Start layers its agent-owned control-channel vars on top of this; those |
| 2706 | // belong to a serving process, not to a one-shot init. Both callers mint the |
| 2707 | // instance's CARLOS_ADMIN_TOKEN into <host>.env first, so "the same |
| 2708 | // environment" includes the same secret.) |
| 2709 | func (b *ExecBackend) instanceEnv(rt registry.Route) (map[string]string, error) { |
| 2710 | cfgEnv, ownEnv, err := b.instanceEnvParts(rt) |
| 2711 | if err != nil { |
| 2712 | return nil, err |
| 2713 | } |
| 2714 | return mergeEnv(cfgEnv, ownEnv), nil |
| 2715 | } |
| 2716 | |
| 2717 | // instanceEnvParts is instanceEnv's two reads, before they are merged: |
| 2718 | // the sync-owned config layer and the instance's own file. Extracted so |
| 2719 | // LaunchEnvironment can see which names the own file shadows; instanceEnv |
| 2720 | // is exactly mergeEnv over these two, as it was. |
| 2721 | func (b *ExecBackend) instanceEnvParts(rt registry.Route) (cfgEnv, ownEnv map[string]string, err error) { |
| 2722 | paths := b.pathsFor(rt) |
| 2723 | // A nested host's files live under a tenant-writable directory (design |
| 2724 | // §3.2): read them without following a symlink. A flat host's files sit |
| 2725 | // outside any tenant's reach, so the plain read stays exactly as it was. |
| 2726 | read := readEnvFile |
| 2727 | if paths.layout == LayoutNested { |
| 2728 | read = func(path string) (map[string]string, error) { |
| 2729 | return readEnvFileBeneath(filepath.Dir(path), filepath.Base(path)) |
| 2730 | } |
| 2731 | } |
| 2732 | cfgEnv, err = read(paths.configEnv) |
| 2733 | if err != nil && !os.IsNotExist(err) { |
| 2734 | return nil, nil, fmt.Errorf("read config env file: %w", err) |
| 2735 | } |
| 2736 | ownEnv, err = read(paths.envFile) |
| 2737 | if err != nil && !os.IsNotExist(err) { |
| 2738 | return nil, nil, fmt.Errorf("read env file: %w", err) |
| 2739 | } |
| 2740 | return cfgEnv, ownEnv, nil |
| 2741 | } |
| 2742 | |
| 2743 | // LaunchEnvironment is the composed environment the next Start of rt |
| 2744 | // would spawn with: childEnv over agentOwnedVars over instanceEnv, the |
| 2745 | // exact composition Start uses (agentOwnedVars(rt, fileEnv, haveControl)), |
| 2746 | // decoded back to a map. agentOwnedVars REMOVES tenant-provided |
| 2747 | // CARLOS_CONTROL_SOCKET, CARLOS_PURGE and CARLOS_REGION when the agent |
| 2748 | // does not provide them, and always sets CARLOS_HOST, so a report over |
| 2749 | // mergeEnv alone would misstate the environment. haveControl is true: startChild calls |
| 2750 | // edge.ListenControl for EVERY exec child, and b.Control is the nested |
| 2751 | // root-control client, which says nothing about the flat channel, so |
| 2752 | // before the spawn the honest composition is "the agent will bind it"; a |
| 2753 | // bind that fails at spawn is logged there and the probe's spawn-time |
| 2754 | // check compares the actual outcome. CARLOS_PURGE follows b.Cache != nil |
| 2755 | // exactly as agentOwnedVars does. |
| 2756 | // |
| 2757 | // shadowed is the sorted names present in both the rendered configuration |
| 2758 | // file and the instance's own <host>.env: the override mergeEnv applies |
| 2759 | // (moves spec §7.3 T3 reports it), collected here because this is the only |
| 2760 | // layer that sees both files. The one exception is mergeEnv's: an empty |
| 2761 | // CARLOS_INSTANCE_KEY in the own file overrides nothing, so it is not |
| 2762 | // listed. The move's launch report (§7.3, codex R2-8) |
| 2763 | // is computed over exactly this env, so a target compares what would run, |
| 2764 | // not a reconstruction of it. The values are raw: the caller HMACs them |
| 2765 | // and must never log or return them. |
| 2766 | func (b *ExecBackend) LaunchEnvironment(rt registry.Route) (map[string]string, []string, error) { |
| 2767 | env, _, shadowed, err := b.LaunchLayers(rt) |
| 2768 | return env, shadowed, err |
| 2769 | } |
| 2770 | |
| 2771 | // LaunchLayers is LaunchEnvironment with the names whose running value the |
| 2772 | // instance's own layers supplied: every name of the rendered configuration |
| 2773 | // file and the instance's own <host>.env, as mergeEnv merged them, except |
| 2774 | // the names agentOwnedVars decides (agentOwnedNames), whose running value |
| 2775 | // is the agent's whatever the files say. A name in the composed env that is |
| 2776 | // not in instance came from the agent: its own variables or its process |
| 2777 | // environment (childEnv). The move's env pin needs the difference |
| 2778 | // (moves.ReportLaunch): a box variable such as HOSTNAME is the box's when |
| 2779 | // inherited and the instance's when its files set it. |
| 2780 | func (b *ExecBackend) LaunchLayers(rt registry.Route) (env map[string]string, instance map[string]bool, shadowed []string, err error) { |
| 2781 | cfgEnv, ownEnv, err := b.instanceEnvParts(rt) |
| 2782 | if err != nil { |
| 2783 | return nil, nil, nil, err |
| 2784 | } |
| 2785 | for k := range ownEnv { |
| 2786 | if _, rendered := cfgEnv[k]; rendered { |
| 2787 | if k == instancekey.Var && ownEnv[k] == "" { |
| 2788 | continue // an empty own key shadows nothing (mergeEnv) |
| 2789 | } |
| 2790 | shadowed = append(shadowed, k) |
| 2791 | } |
| 2792 | } |
| 2793 | sort.Strings(shadowed) |
| 2794 | fileEnv := mergeEnv(cfgEnv, ownEnv) |
| 2795 | instance = make(map[string]bool, len(fileEnv)) |
| 2796 | for k := range fileEnv { |
| 2797 | instance[k] = true |
| 2798 | } |
| 2799 | for _, k := range agentOwnedNames { |
| 2800 | delete(instance, k) |
| 2801 | } |
| 2802 | out := map[string]string{} |
| 2803 | for _, kv := range childEnv(b.agentOwnedVars(rt, rt.Socket+".control", fileEnv, true)) { |
| 2804 | if i := strings.IndexByte(kv, '='); i > 0 { |
| 2805 | out[kv[:i]] = kv[i+1:] |
| 2806 | } |
| 2807 | } |
| 2808 | return out, instance, shadowed, nil |
| 2809 | } |
| 2810 | |
| 2811 | // childEnv is the child's complete environment: the agent's own, minus |
| 2812 | // every CARLOS_* variable, with the instance's file values layered on top. |
| 2813 | // One entry per key, last value wins: os/exec deduplicates cmd.Env that |
| 2814 | // way, but on the unit path the trampoline hands this list to execve as |
| 2815 | // is, duplicates and all, and a Go child then keeps the FIRST occurrence |
| 2816 | // (design §6.2, codex R1-16). Composing it deduplicated here keeps the |
| 2817 | // two spawn paths identical. |
| 2818 | func childEnv(fileEnv map[string]string) []string { |
| 2819 | order := make([]string, 0, 32) |
| 2820 | values := make(map[string]string, 32) |
| 2821 | set := func(k, v string) { |
| 2822 | if _, seen := values[k]; !seen { |
| 2823 | order = append(order, k) |
| 2824 | } |
| 2825 | values[k] = v |
| 2826 | } |
| 2827 | for _, kv := range os.Environ() { |
| 2828 | if strings.HasPrefix(kv, "CARLOS_") { |
| 2829 | continue |
| 2830 | } |
| 2831 | k, v, ok := strings.Cut(kv, "=") |
| 2832 | if !ok { |
| 2833 | continue |
| 2834 | } |
| 2835 | set(k, v) |
| 2836 | } |
| 2837 | keys := make([]string, 0, len(fileEnv)) |
| 2838 | for k := range fileEnv { |
| 2839 | keys = append(keys, k) |
| 2840 | } |
| 2841 | sort.Strings(keys) |
| 2842 | for _, k := range keys { |
| 2843 | set(k, fileEnv[k]) |
| 2844 | } |
| 2845 | env := make([]string, 0, len(order)) |
| 2846 | for _, k := range order { |
| 2847 | env = append(env, k+"="+values[k]) |
| 2848 | } |
| 2849 | return env |
| 2850 | } |
| 2851 | |
| 2852 | // withVar returns a copy of env with key=value added — a copy, so the |
| 2853 | // caller's map (an EnvironmentFile's parsed contents) is never mutated. |
| 2854 | func withVar(env map[string]string, key, value string) map[string]string { |
| 2855 | out := make(map[string]string, len(env)+1) |
| 2856 | for k, v := range env { |
| 2857 | out[k] = v |
| 2858 | } |
| 2859 | out[key] = value |
| 2860 | return out |
| 2861 | } |
| 2862 | |
| 2863 | // withoutVars returns a copy of env with keys removed — the mirror of withVar, |
| 2864 | // and the half that makes an agent-owned variable actually agent-owned. Absent |
| 2865 | // it, "the agent decides whether this var is set" is only true when the agent |
| 2866 | // sets it; a tenant supplying the same name in its own env file wins by |
| 2867 | // default. Same copy discipline as withVar: the caller's map is never mutated. |
| 2868 | // |
| 2869 | // This package keeps its own agent-owned list inline (the two names above, |
| 2870 | // at Start's control-channel branches) rather than sharing one with |
| 2871 | // internal/appconfig's agentOwned (agentvars.go), which the unit-tenant sync |
| 2872 | // path uses for the same purpose — the two lists must not drift apart. |
| 2873 | func withoutVars(env map[string]string, keys ...string) map[string]string { |
| 2874 | out := make(map[string]string, len(env)) |
| 2875 | for k, v := range env { |
| 2876 | out[k] = v |
| 2877 | } |
| 2878 | for _, k := range keys { |
| 2879 | delete(out, k) |
| 2880 | } |
| 2881 | return out |
| 2882 | } |
| 2883 | |
| 2884 | // mergeEnv composes the child's env-file layers: base (the sync-owned |
| 2885 | // <host>.config.env, bucket-derived) under overlay (the instance's own |
| 2886 | // <host>.env). The overlay wins every conflict but one — CARLOS_ADMIN_TOKEN |
| 2887 | // lives only in the instance file and must never be clobberable from |
| 2888 | // the bucket, and a box-local override deliberately beats central |
| 2889 | // config. The exception: an empty CARLOS_INSTANCE_KEY in the overlay |
| 2890 | // supplies no key, so a non-empty one in base stands (instancekey.Effective). |
| 2891 | // Returns a fresh map; inputs are never mutated. |
| 2892 | func mergeEnv(base, overlay map[string]string) map[string]string { |
| 2893 | out := make(map[string]string, len(base)+len(overlay)) |
| 2894 | for k, v := range base { |
| 2895 | out[k] = v |
| 2896 | } |
| 2897 | for k, v := range overlay { |
| 2898 | out[k] = v |
| 2899 | } |
| 2900 | // An empty CARLOS_INSTANCE_KEY in the instance's own file supplies no |
| 2901 | // key, so it never shadows a configured one: the child is never handed |
| 2902 | // an empty key while a non-empty one exists (instance-key plan ruling |
| 2903 | // 13). instancekey.Effective is the rule the key's mint decides by. |
| 2904 | if key, _ := instancekey.Effective(base, overlay); key != "" { |
| 2905 | out[instancekey.Var] = key |
| 2906 | } |
| 2907 | return out |
| 2908 | } |
| 2909 | |
| 2910 | // HasSidecar: a host has a sidecar iff its sidecar EnvironmentFile exists. |
| 2911 | // Derived from the filesystem on purpose, the way systemd units would do it: |
| 2912 | // no registry column to drift from what's actually staged. |
| 2913 | // |
| 2914 | // This is the LEGACY file-staged co-spawn — Eleven runs on it today, so it |
| 2915 | // is not going anywhere yet. The NATIVE path is a route with |
| 2916 | // registry.Route.IsSidecar true: its own registry row, spawned by |
| 2917 | // StartWorker instead of StartSidecar, with no parent instance and no file |
| 2918 | // on disk. A host may carry one mechanism or the other but never both — the |
| 2919 | // provisioning reconciler's mint-time check (sidecar_reconcile.go) refuses |
| 2920 | // to mint a native row for a host this method already reports true for. |
| 2921 | // Retiring this file path is a follow-up gated on Eleven's own migration to |
| 2922 | // native sidecar records. |
| 2923 | func (b *ExecBackend) HasSidecar(rt registry.Route) bool { |
| 2924 | _, err := os.Stat(b.pathsFor(rt).sidecarEnv) |
| 2925 | return err == nil |
| 2926 | } |
| 2927 | |
| 2928 | // StartSidecar dispatches by layout, the same way Start does. |
| 2929 | func (b *ExecBackend) StartSidecar(ctx context.Context, rt registry.Route) (Proc, error) { |
| 2930 | p := b.pathsFor(rt) |
| 2931 | switch p.layout { |
| 2932 | case LayoutNested: |
| 2933 | return b.startSidecarUnit(ctx, rt) |
| 2934 | case LayoutFenced, LayoutConflict: |
| 2935 | return nil, fmt.Errorf("%s: refusing to start sidecar: %s", rt.Host, p.reason) |
| 2936 | } |
| 2937 | return b.startSidecarChild(ctx, rt) |
| 2938 | } |
| 2939 | |
| 2940 | // startSidecarChild execs the host's live symlink with `sidecar run` under |
| 2941 | // the sidecar's EnvironmentFile, CARLOS_*-clean like the instance spawn. No |
| 2942 | // health wait — the sidecar is a pure outbound client with no socket to |
| 2943 | // probe; its own connect/backoff loop absorbs a slow instance. The live |
| 2944 | // symlink exists by now: the instance spawn checked it. The ctx is unused (an |
| 2945 | // exec doesn't block); it's here for the backend seam's symmetry. |
| 2946 | // |
| 2947 | // This arm serves the LEGACY co-spawn only (see HasSidecar) — StartWorker is |
| 2948 | // the native spawn path for a standalone registry.Route.IsSidecar row. |
| 2949 | // |
| 2950 | // With LiveExecGate set (a declined box that honours failure records or |
| 2951 | // hold-ran markers; Task 9 fix round 1, ruling 1), the link is resolved once, |
| 2952 | // without hashing (release.ResolveLivePath), the gate is asked about that |
| 2953 | // build, and the file it resolved is exec'd, not the link. The legs leave a |
| 2954 | // held instance running on its old build while root repoints the link, and |
| 2955 | // every respawn of its sidecar (the wake tail, late attachment, the sweep's |
| 2956 | // supervision) comes through here: exec'ing the link would run the held |
| 2957 | // build beside it. A refusal starts nothing, and the supervisor's backoff |
| 2958 | // keeps the sidecar down. With the gate nil it is today's exec of the link. |
| 2959 | func (b *ExecBackend) startSidecarChild(ctx context.Context, rt registry.Route) (Proc, error) { |
| 2960 | if b.LiveExecGate == nil { |
| 2961 | return b.startSidecarChildFrom(ctx, rt, b.liveLink(rt.Host)) |
| 2962 | } |
| 2963 | id, err := release.ResolveLivePath(b.OptDir, rt.Host) |
| 2964 | if err != nil { |
| 2965 | return nil, fmt.Errorf("%s: sidecar: %w", rt.Host, err) |
| 2966 | } |
| 2967 | if err := b.LiveExecGate(rt, id.Version); err != nil { |
| 2968 | return nil, err |
| 2969 | } |
| 2970 | return b.startSidecarChildFrom(ctx, rt, id.StagedPath) |
| 2971 | } |
| 2972 | |
| 2973 | // startSidecarChildFrom is startSidecarChild with the binary named. That is |
| 2974 | // the live link for today's StartSidecar, and a generation's captured |
| 2975 | // staged path for StartSidecarGen (zero-downtime spec §7.5). Placement is |
| 2976 | // plan 2's: under GenContainment the sidecar lands in role-sidecar, with no |
| 2977 | // limit of its own, under the generation parent's backstop (§7.3, plan 2 |
| 2978 | // D4). The parent holds no process and keeps memory.oom.group=0. |
| 2979 | func (b *ExecBackend) startSidecarChildFrom(_ context.Context, rt registry.Route, bin string) (Proc, error) { |
| 2980 | fileEnv, err := readEnvFile(b.pathsFor(rt).sidecarEnv) |
| 2981 | if err != nil { |
| 2982 | return nil, fmt.Errorf("read sidecar env: %w", err) |
| 2983 | } |
| 2984 | // The sidecar runs the app binary with no control socket, so on a |
| 2985 | // replica host it is gated as the worker is (ruling 26): its home's key, |
| 2986 | // verified, or nothing runs. A key in its own file is checked too. |
| 2987 | if err := b.replicaKeyGate(rt, fileEnv, sidecarKeyLayer(fileEnv), false); err != nil { |
| 2988 | return nil, err |
| 2989 | } |
| 2990 | child, _, err := b.placeFor(spawnSidecar, 0) |
| 2991 | if err != nil { |
| 2992 | return nil, err |
| 2993 | } |
| 2994 | // Same host cgroup as the instance: a sidecar's load is that instance's |
| 2995 | // work, and attributing it elsewhere would be a lie in the other |
| 2996 | // direction. |
| 2997 | cmd, settle, err := b.spawnInCgroupAt(rt, child, false, func() *exec.Cmd { |
| 2998 | cmd := exec.Command(bin, "sidecar", "run") |
| 2999 | cmd.Env = childEnv(fileEnv) |
| 3000 | // Like the instance child: logs through the agent's own streams |
| 3001 | // (journald under systemd), teed into the log sink when armed — a |
| 3002 | // sidecar's output is the app's output. |
| 3003 | b.teeStd(cmd, rt) |
| 3004 | return cmd |
| 3005 | }) |
| 3006 | if err != nil { |
| 3007 | return nil, err |
| 3008 | } |
| 3009 | p := &childProc{cmd: cmd, done: make(chan struct{})} |
| 3010 | go func() { |
| 3011 | p.err = cmd.Wait() |
| 3012 | settle() |
| 3013 | b.Cgroups.RemoveChild(rt.Host, child) // no-op for "" (flag off) |
| 3014 | b.cleanupHost(rt.Host) |
| 3015 | close(p.done) |
| 3016 | }() |
| 3017 | return p, nil |
| 3018 | } |
| 3019 | |
| 3020 | // StartWorker is StartWorkerStaged onto whatever live/<host> names now. Its |
| 3021 | // string result is that build's version, as it always was. |
| 3022 | func (b *ExecBackend) StartWorker(ctx context.Context, rt registry.Route, extra map[string]string) (Proc, string, error) { |
| 3023 | p, id, err := b.StartWorkerStaged(ctx, rt, extra, nil) |
| 3024 | return p, id.Version, err |
| 3025 | } |
| 3026 | |
| 3027 | // StartWorkerStaged execs a worker with the route's declared argv (rt.Args, a |
| 3028 | // JSON array: a sidecar's `["sidecar","run"]` today) under the instance's own |
| 3029 | // env layers, CARLOS_*-clean like every other child this backend spawns, and |
| 3030 | // with NO health wait: a worker binds nothing, so there is nothing |
| 3031 | // waitHealthyCtx could dial. extra (the per-wake sidecar token) is applied |
| 3032 | // last and wins over the tenant's own <host>.env. |
| 3033 | // |
| 3034 | // What it execs is a staged build BY PATH, never live/<host> (zero-downtime |
| 3035 | // deploys spec §7.9): |
| 3036 | // - onto == nil (a wake, or a converge onto a new build): the live link is |
| 3037 | // ensured, read ONCE, and the file it names is hashed and exec'd, once |
| 3038 | // LiveExecGate (when set) has passed its build; |
| 3039 | // - onto != nil (a crash or max_run recycle): onto.StagedPath is checked |
| 3040 | // against onto.Digest (release.VerifyStaged) and exec'd. A copy that has |
| 3041 | // changed is refused with release.ErrStagedCopyChanged and nothing runs. |
| 3042 | // |
| 3043 | // The identity returned is the build exec'd, the agent's serving identity for |
| 3044 | // this child. What the child later reports over the agent socket is display |
| 3045 | // metadata only. |
| 3046 | // |
| 3047 | // Between resolving that identity and the exec it runs ctx's before-exec hook |
| 3048 | // (runBeforeExec), through which the activator publishes the build first; a |
| 3049 | // hook error means nothing runs (ruling R19 item 3). ctx's exec gate |
| 3050 | // (withExecGate), the caller's ownership check, is the cgroup manager's |
| 3051 | // before-start hook (launchGate), as a flat instance spawn's is: asked after |
| 3052 | // the cgroup's preparation and before each start attempt, and its refusal |
| 3053 | // starts nothing and comes back as it is (Astra R3 §5 item 4). |
| 3054 | // |
| 3055 | // A worker on a replica row passes replicaKeyGate as a seed does, with no |
| 3056 | // control socket, before that hook: its home's key, verified, or nothing |
| 3057 | // runs (instance-key plan ruling 26). A worker never mints a key. |
| 3058 | func (b *ExecBackend) StartWorkerStaged(ctx context.Context, rt registry.Route, extra map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) { |
| 3059 | var argv []string |
| 3060 | if err := json.Unmarshal([]byte(rt.Args), &argv); err != nil || len(argv) == 0 { |
| 3061 | return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: no runnable argv in args %q", rt.Host, rt.Args) |
| 3062 | } |
| 3063 | var id release.StagedIdentity |
| 3064 | if onto == nil { |
| 3065 | if err := b.ensureLiveLink(ctx, rt.Host); err != nil { |
| 3066 | return nil, release.StagedIdentity{}, err |
| 3067 | } |
| 3068 | got, err := release.ResolveLive(b.OptDir, rt.Host) |
| 3069 | if err != nil { |
| 3070 | return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err) |
| 3071 | } |
| 3072 | id = got |
| 3073 | // A declined box that honours failure records or hold-ran markers |
| 3074 | // (Task 9) asks about the build just resolved, which is the build |
| 3075 | // exec'd (by its staged path), before it is published or anything is |
| 3076 | // prepared. A refusal runs nothing and comes back as it is. A recycle |
| 3077 | // onto the captured build (onto set) is no move onto the live link, |
| 3078 | // and is not asked. |
| 3079 | if b.LiveExecGate != nil { |
| 3080 | if err := b.LiveExecGate(rt, id.Version); err != nil { |
| 3081 | return nil, release.StagedIdentity{}, err |
| 3082 | } |
| 3083 | } |
| 3084 | } else { |
| 3085 | if err := release.VerifyStaged(*onto); err != nil { |
| 3086 | return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: not recycled onto %s: %w", rt.Host, onto.Version, err) |
| 3087 | } |
| 3088 | id = *onto |
| 3089 | } |
| 3090 | |
| 3091 | // One read gives the env and the layer its CARLOS_INSTANCE_KEY came |
| 3092 | // from, for the replica gate below (ruling 29). A worker never mints |
| 3093 | // (ruling 12): the key is whatever the env files already deliver. |
| 3094 | fileEnv, keyLayer, err := b.instanceEnvLayer(rt) |
| 3095 | if err != nil { |
| 3096 | return nil, release.StagedIdentity{}, err |
| 3097 | } |
| 3098 | for k, v := range extra { |
| 3099 | fileEnv = withVar(fileEnv, k, v) |
| 3100 | } |
| 3101 | // A worker has no control socket, so a replica row's worker is gated as |
| 3102 | // its seed is (ruling 26): its home's key, verified, or refused. extra is |
| 3103 | // agent-issued and has no business carrying a key; one that did would |
| 3104 | // not be the key the layer describes, so it is refused as of unknown |
| 3105 | // origin. |
| 3106 | if _, ok := extra[instancekey.Var]; ok { |
| 3107 | keyLayer = instancekey.LayerNone |
| 3108 | } |
| 3109 | if err := b.replicaKeyGate(rt, fileEnv, keyLayer, false); err != nil { |
| 3110 | return nil, release.StagedIdentity{}, err |
| 3111 | } |
| 3112 | |
| 3113 | // The identity is resolved and nothing has run: the activator publishes |
| 3114 | // the build now, before the exec, and a failure means no child (ruling |
| 3115 | // R19 item 3). |
| 3116 | if err := runBeforeExec(ctx, id); err != nil { |
| 3117 | return nil, release.StagedIdentity{}, fmt.Errorf("worker %s: %w", rt.Host, err) |
| 3118 | } |
| 3119 | |
| 3120 | // Through spawnInCgroup, not Cgroups.Start directly: that seam is what |
| 3121 | // applies the route's resource tier, and a worker is a long-lived child |
| 3122 | // of a tenant's app like any other. Bypassing it compiles and silently |
| 3123 | // ships an uncapped process — the exact failure spawnInCgroup's own |
| 3124 | // comment names. |
| 3125 | // |
| 3126 | // Gated (spawnInCgroupGated): ownership can be lost while the cgroup is |
| 3127 | // prepared and the process placed, or between a failed placement and |
| 3128 | // its PID-move fallback, so the caller's gate is asked at each exec, not |
| 3129 | // once before all of it. |
| 3130 | // With no gate on ctx (an unplanned wake's spawn passes none) launchGate |
| 3131 | // is nil and this is spawnInCgroup exactly. The zero tier is |
| 3132 | // spawnInCgroup's too: the route's live tier (b.tier), as a worker has |
| 3133 | // always had. |
| 3134 | cmd, settle, err := b.spawnInCgroupGated(rt, cgroup.Limit{}, func() *exec.Cmd { |
| 3135 | cmd := exec.Command(id.StagedPath, argv...) |
| 3136 | cmd.Env = childEnv(fileEnv) |
| 3137 | b.teeStd(cmd, rt) |
| 3138 | return cmd |
| 3139 | }, b.launchGate(ctx, rt.Host)) |
| 3140 | if err != nil { |
| 3141 | return nil, release.StagedIdentity{}, err |
| 3142 | } |
| 3143 | p := &childProc{cmd: cmd, done: make(chan struct{})} |
| 3144 | go func() { |
| 3145 | p.err = cmd.Wait() |
| 3146 | // Mirrors Start's exit record (#5 review, closed here): a worker's |
| 3147 | // crash is a sidecar's health word ("Crashed") the moment it lands, |
| 3148 | // so the member needs the SAME explanation an instance crash gets, |
| 3149 | // not silence. This fires once per actual exit, and the caller's |
| 3150 | // own doubling-backoff restart (2s up to a minute, activator.go) |
| 3151 | // already bounds how often an exit can happen at all — a |
| 3152 | // crash-looping worker cannot make this any noisier than the |
| 3153 | // restart log line it already earns internally. |
| 3154 | if p.err != nil { |
| 3155 | emitEvent(b.Logs, rt, "warn", "worker exited: "+RedactPaths(p.err.Error())) |
| 3156 | } else { |
| 3157 | emitEvent(b.Logs, rt, "info", "worker exited (exit 0)") |
| 3158 | } |
| 3159 | settle() |
| 3160 | b.cleanupHost(rt.Host) |
| 3161 | close(p.done) |
| 3162 | }() |
| 3163 | return p, id, nil |
| 3164 | } |
| 3165 | |
| 3166 | // startSidecarUnit is StartSidecar's nested body: a sidecar has no health |
| 3167 | // wait either way, and no control listener of its own. Sidecar env is a |
| 3168 | // single layer, read straight from its own file, never merged with the |
| 3169 | // instance's two-layer config. |
| 3170 | func (b *ExecBackend) startSidecarUnit(ctx context.Context, rt registry.Route) (Proc, error) { |
| 3171 | return b.startSidecarUnitFrom(ctx, rt, nil, cgroup.Limit{}) |
| 3172 | } |
| 3173 | |
| 3174 | // startSidecarUnitFrom is startSidecarUnit with the executable and the tier |
| 3175 | // named. A nil exe is the live link (today's start), pinned instead on a box |
| 3176 | // with LiveExecGate set (declinedLivePin). A staged exe is the serving |
| 3177 | // generation's captured build (zero-downtime spec §7.5), which root |
| 3178 | // validates before it lays out the mount, and tier that generation's own. |
| 3179 | func (b *ExecBackend) startSidecarUnitFrom(ctx context.Context, rt registry.Route, exe *instancectl.StagedExe, tier cgroup.Limit) (Proc, error) { |
| 3180 | if b.Cfg.HostWrites.Enabled { |
| 3181 | return nil, ErrClaimIsolationUnavailable |
| 3182 | } |
| 3183 | if err := b.nestedGate(rt); err != nil { |
| 3184 | return nil, err |
| 3185 | } |
| 3186 | if exe == nil && b.LiveExecGate != nil { |
| 3187 | // A declined box that honours failure records or hold-ran markers |
| 3188 | // (Task 9 fix round 1, ruling 1) pins the live link once and names |
| 3189 | // that file to root at its own tier (declinedLivePin), as the nested |
| 3190 | // instance start does (spawnUnitFrom): root binds the build the gate |
| 3191 | // passed, never the link. Before the launch file is written, so a |
| 3192 | // refusal leaves nothing behind. |
| 3193 | id, err := b.declinedLivePin(ctx, rt) |
| 3194 | if err != nil { |
| 3195 | return nil, err |
| 3196 | } |
| 3197 | exe, tier = stagedExeOf(id), b.GenTier(rt, id.Version) |
| 3198 | } |
| 3199 | p, _, logs, fileEnv, err := b.prepare(rt, instancectl.RoleSidecar) |
| 3200 | if err != nil { |
| 3201 | return nil, err |
| 3202 | } |
| 3203 | argv := []string{b.liveLink(rt.Host), "sidecar", "run"} |
| 3204 | if err := b.writeLaunch(p, instancectl.RoleSidecar, rt.Host, argv, childEnv(fileEnv)); err != nil { |
| 3205 | logs.Close() |
| 3206 | return nil, err |
| 3207 | } |
| 3208 | var inv string |
| 3209 | if exe != nil { |
| 3210 | // The serving generation's captured staged build, which root |
| 3211 | // validates before it lays out the mount, at that generation's own |
| 3212 | // tier, never the live link's. |
| 3213 | gc, gerr := b.genControl() |
| 3214 | if gerr != nil { |
| 3215 | logs.Close() |
| 3216 | return nil, gerr |
| 3217 | } |
| 3218 | if tier == (cgroup.Limit{}) { |
| 3219 | tier = b.tier(rt) |
| 3220 | } |
| 3221 | inv, err = gc.StartSidecarExe(ctx, rt.Host, tier.MemoryMax, *exe) |
| 3222 | } else { |
| 3223 | inv, err = b.Control.Start(ctx, rt.Host, instancectl.RoleSidecar, b.tier(rt).MemoryMax) |
| 3224 | } |
| 3225 | if err != nil { |
| 3226 | logs.Close() |
| 3227 | return nil, fmt.Errorf("%s: start sidecar unit: %w", rt.Host, err) |
| 3228 | } |
| 3229 | if inv == "" { |
| 3230 | logs.Close() |
| 3231 | return nil, fmt.Errorf("%s: %s exited before it was observed", rt.Host, instancectl.UnitName(instancectl.RoleSidecar, rt.Host)) |
| 3232 | } |
| 3233 | return b.newUnitProc(rt, instancectl.RoleSidecar, p, inv, nil, logs), nil |
| 3234 | } |
| 3235 | |
| 3236 | // socketClient returns an HTTP client that dials an instance's unix socket |
| 3237 | // (request URLs use the placeholder host http://carlos). |
| 3238 | func socketClient(socket string) *http.Client { |
| 3239 | return &http.Client{ |
| 3240 | Timeout: 3 * time.Second, |
| 3241 | Transport: &http.Transport{ |
| 3242 | DialContext: func(ctx context.Context, _, _ string) (net.Conn, error) { |
| 3243 | return unixdial.Checked(ctx, &net.Dialer{}, socket) |
| 3244 | }, |
| 3245 | }, |
| 3246 | } |
| 3247 | } |
| 3248 | |
| 3249 | // waitHealthy blocks until the instance answers an HTTP request on its socket, |
| 3250 | // or the timeout elapses. It retries, so it's robust to the restart window |
| 3251 | // (socket briefly absent, or a stale socket left by an abruptly-killed old |
| 3252 | // process) — and a broken binary that never becomes healthy fails the wake. |
| 3253 | func waitHealthy(socket string, timeout time.Duration) error { |
| 3254 | deadline := time.Now().Add(timeout) |
| 3255 | client := socketClient(socket) |
| 3256 | var last error |
| 3257 | var poll time.Duration |
| 3258 | for time.Now().Before(deadline) { |
| 3259 | resp, err := client.Get("http://carlos/") |
| 3260 | if err == nil { |
| 3261 | resp.Body.Close() |
| 3262 | if resp.StatusCode < 500 { |
| 3263 | return nil |
| 3264 | } |
| 3265 | last = fmt.Errorf("unhealthy: HTTP %d", resp.StatusCode) |
| 3266 | } else { |
| 3267 | last = err |
| 3268 | } |
| 3269 | poll = nextHealthPoll(poll) |
| 3270 | time.Sleep(poll) |
| 3271 | } |
| 3272 | return fmt.Errorf("instance not healthy within %s: %v", timeout, last) |
| 3273 | } |
| 3274 | |
| 3275 | // The wait between health probes. A request is held open for the whole of |
| 3276 | // a wake, and a fixed 150 ms gap cost it 75 ms on average after the app was |
| 3277 | // already listening. A refused dial costs microseconds, so the wait starts |
| 3278 | // short and doubles to a ceiling that keeps a slow boot from being probed |
| 3279 | // in a busy loop. |
| 3280 | const ( |
| 3281 | healthPollMin = 5 * time.Millisecond |
| 3282 | healthPollMax = 50 * time.Millisecond |
| 3283 | ) |
| 3284 | |
| 3285 | // nextHealthPoll is the wait after one of d; zero gives the first. |
| 3286 | func nextHealthPoll(d time.Duration) time.Duration { |
| 3287 | return min(max(2*d, healthPollMin), healthPollMax) |
| 3288 | } |
| 3289 | |
| 3290 | // waitHealthyCtx is waitHealthy bounded by ctx as well as its own timeout, so |
| 3291 | // an aborted wake doesn't keep probing a socket nobody wants anymore. |
| 3292 | func waitHealthyCtx(ctx context.Context, socket string, timeout time.Duration) error { |
| 3293 | done := make(chan error, 1) |
| 3294 | go func() { done <- waitHealthy(socket, timeout) }() |
| 3295 | select { |
| 3296 | case err := <-done: |
| 3297 | return err |
| 3298 | case <-ctx.Done(): |
| 3299 | return ctx.Err() |
| 3300 | } |
| 3301 | } |
| 3302 | |
| 3303 | // readEnvFile parses a KEY=value EnvironmentFile (blank lines and #comments |
| 3304 | // ignored). |
| 3305 | func readEnvFile(path string) (map[string]string, error) { |
| 3306 | raw, err := os.ReadFile(path) |
| 3307 | if err != nil { |
| 3308 | return nil, err |
| 3309 | } |
| 3310 | return parseEnvBytes(raw), nil |
| 3311 | } |
| 3312 | |
| 3313 | // childProc is a spawned child: SIGTERM starts the instance's own graceful |
| 3314 | // drain; SIGKILL only if it overstays. |
| 3315 | type childProc struct { |
| 3316 | cmd *exec.Cmd |
| 3317 | done chan struct{} |
| 3318 | err error |
| 3319 | ctl *edge.ControlListener // nil only if the pre-spawn bind failed |
| 3320 | } |
| 3321 | |
| 3322 | func (p *childProc) Exited() bool { |
| 3323 | select { |
| 3324 | case <-p.done: |
| 3325 | return true |
| 3326 | default: |
| 3327 | return false |
| 3328 | } |
| 3329 | } |
| 3330 | |
| 3331 | // Crashed answers the activator's procCrasher seam: a DEFINITELY non-zero |
| 3332 | // exit status. Read from ProcessState rather than from p.err because a clean |
| 3333 | // exit whose pipes outlive it comes back as exec.ErrWaitDelay (cmd.WaitDelay |
| 3334 | // is set on every child here) — an error that says nothing about the exit |
| 3335 | // status, and reading it as a crash would restart a worker that finished. |
| 3336 | // Anything this cannot establish reads as NOT a crash, which is procCrashed's |
| 3337 | // own default and for its reasons. |
| 3338 | func (p *childProc) Crashed() bool { |
| 3339 | if !p.Exited() { |
| 3340 | return false |
| 3341 | } |
| 3342 | st := p.cmd.ProcessState // safe: written before done closed, read after |
| 3343 | return st != nil && !st.Success() |
| 3344 | } |
| 3345 | |
| 3346 | func (p *childProc) Stop(ctx context.Context) error { |
| 3347 | // The control listener OUTLIVES the drain, and the ordering is the whole |
| 3348 | // point. Closing it first — which is what this did — unbound and unlinked |
| 3349 | // the socket before the child had even seen SIGTERM, so an app that |
| 3350 | // registers or re-asserts its once-timers on the way down (spec §7.2's |
| 3351 | // lifetime decision — the socket brackets the PROCESS, bound before the |
| 3352 | // spawn and closed after the exit — serving §7.1's re-assert story) got |
| 3353 | // connection-refused on the channel the platform had just told it about, |
| 3354 | // during precisely the window it was told to use. Signal first, close |
| 3355 | // after the process is gone. |
| 3356 | // |
| 3357 | // Deferred, so it is unconditional on every path out of here — the |
| 3358 | // already-exited return, the lost-race return, the kill-timeout return, |
| 3359 | // and even a Signal error we hand back to the caller. A listener left |
| 3360 | // bound to a socket no process is behind is the stale-socket wake trap |
| 3361 | // (#129) waiting to happen, and this is the only close a childProc ever |
| 3362 | // gets — Start's own is on the path where the spawn failed and no |
| 3363 | // childProc was ever built. |
| 3364 | defer func() { |
| 3365 | if p.ctl != nil { |
| 3366 | p.ctl.Close() |
| 3367 | } |
| 3368 | }() |
| 3369 | if p.Exited() { |
| 3370 | return nil |
| 3371 | } |
| 3372 | if err := p.cmd.Process.Signal(syscall.SIGTERM); err != nil { |
| 3373 | if errors.Is(err, os.ErrProcessDone) { // lost the race with its own exit |
| 3374 | <-p.done |
| 3375 | return nil |
| 3376 | } |
| 3377 | return err |
| 3378 | } |
| 3379 | select { |
| 3380 | case <-p.done: |
| 3381 | return nil |
| 3382 | case <-ctx.Done(): |
| 3383 | _ = p.cmd.Process.Kill() |
| 3384 | <-p.done |
| 3385 | return fmt.Errorf("instance did not exit on SIGTERM; killed") |
| 3386 | } |
| 3387 | } |
| 3388 | |
| 3389 | // movesPrefix is where a residence move stages a replica in the TARGET's |
| 3390 | // data bucket (instance-move spec): moves/<host>/ beside the ordinary |
| 3391 | // instances/ prefix. A free function so the host's move worker and this |
| 3392 | // package spell it identically. |
| 3393 | func MovesPrefix(host string) string { return "moves/" + host + "/" } |
| 3394 | |
| 3395 | // MoveReceiptsKey is where a residence move's mirror keeps its copy |
| 3396 | // receipts (#175, lease.Receipts) in the SOURCE box's own data bucket. The |
| 3397 | // source cannot read the target's moves/ prefix (its grant there is |
| 3398 | // push-only), so the receipts cannot live beside the staged copy. This key |
| 3399 | // is under neither MovesPrefix nor the hibernate prefix, so no restore, |
| 3400 | // promotion or staging cleanup ever sees it. |
| 3401 | func MoveReceiptsKey(host string) string { return "move-receipts/" + host + ".json" } |
| 3402 | |
| 3403 | // PurgeLocalDBArtifacts removes the local litestream footprint next to a |
| 3404 | // DB — WAL, SHM and both meta dirs — and, when full is true, the DB file |
| 3405 | // itself. Restore's pre-restore cleanup uses the partial form (the DB's |
| 3406 | // absence is what got us here); a residence move's cut uses the full form: |
| 3407 | // the activator's warm-cache trust is exactly right for crashes and exactly |
| 3408 | // wrong after a cut, where a stale local file would resurrect a moved-away |
| 3409 | // instance's history. The full form removes the database's <db>.seed |
| 3410 | // marker too (final review carry item): a marker that outlived its |
| 3411 | // database would name a seed for whatever database is put there later. |
| 3412 | func PurgeLocalDBArtifacts(dbPath string, full bool) { |
| 3413 | for _, p := range []string{dbPath + "-wal", dbPath + "-shm"} { |
| 3414 | _ = os.Remove(p) |
| 3415 | } |
| 3416 | _ = os.RemoveAll(dbPath + "-litestream") |
| 3417 | _ = os.RemoveAll(filepath.Join(filepath.Dir(dbPath), ".activator-"+filepath.Base(dbPath)+"-litestream")) |
| 3418 | if full { |
| 3419 | _ = os.Remove(dbPath) |
| 3420 | _ = os.Remove(dbPath + SeedMarkerSuffix) |
| 3421 | } |
| 3422 | } |
| 3423 | |
| 3424 | // promoteStagedReplica copies moves/<host>/ over instances/<host>/ in this |
| 3425 | // box's OWN bucket whenever anything is staged: staging wins when present, |
| 3426 | // whatever the ordinary prefix holds (an exact mirror, so own keys the |
| 3427 | // staging lacks are deleted). A no-op (nil error) when there is nothing |
| 3428 | // staged. |
| 3429 | func (b *ExecBackend) promoteStagedReplica(ctx context.Context, host string) (map[string]lease.MirrorObject, error) { |
| 3430 | // STAGING WINS when present (review round 1). A staged replica exists |
| 3431 | // only while an inbound move is landing or has just landed: it was |
| 3432 | // mirrored from the then-live data at cut, so it is by construction |
| 3433 | // newer than any lineage this box kept from a PREVIOUS residence — the |
| 3434 | // boomerang case (ie→uk→ie) that "own lineage never loses" restored |
| 3435 | // months-stale data from. An abandoned move's staging can't linger to |
| 3436 | // poison this: abort is two-phase, and the source deletes staging |
| 3437 | // before the move clears (MoveAborting). |
| 3438 | // |
| 3439 | // The trailing slash on the guard listing matters: hibernatePrefix has |
| 3440 | // none (path.Join), and a slashless List would let app.example.com's |
| 3441 | // keys answer for app.example.co. |
| 3442 | staged, err := b.Mirror.List(ctx, b.Cfg.Bucket, MovesPrefix(host)) |
| 3443 | if err != nil { |
| 3444 | return nil, err |
| 3445 | } |
| 3446 | if len(staged) == 0 { |
| 3447 | return nil, nil |
| 3448 | } |
| 3449 | log.Printf("activator: %s: promoting %d staged object(s) from %s (residence move landing)", host, len(staged), MovesPrefix(host)) |
| 3450 | // No receipts: a staged object is the source box's single-part copy, |
| 3451 | // and an in-bucket copy of it keeps its ETag, so plain ETag equality |
| 3452 | // is exact here. A deferred object (staging overwritten mid-promote, |
| 3453 | // which nothing should do once the move has cut) fails the wake |
| 3454 | // rather than restoring from a promotion that is not byte-final. |
| 3455 | round, err := lease.MirrorPrefix(ctx, b.Mirror, b.Cfg.Bucket, MovesPrefix(host), b.Cfg.Bucket, b.Cfg.hibernatePrefix(host), lease.Receipts{}) |
| 3456 | if err != nil { |
| 3457 | return nil, err |
| 3458 | } |
| 3459 | if round.Deferred > 0 { |
| 3460 | return nil, fmt.Errorf("promote staged replica for %s: %d staged object(s) changed during the copy; retrying on the next wake", host, round.Deferred) |
| 3461 | } |
| 3462 | return staged, nil |
| 3463 | } |
| 3464 | |
| 3465 | // dropStagedReplica deletes the staged keys promote just returned — no |
| 3466 | // third listing of a prefix this restore has already walked twice. |
| 3467 | // Best-effort: a leftover staging copy is storage, not a correctness |
| 3468 | // problem. |
| 3469 | func (b *ExecBackend) dropStagedReplica(ctx context.Context, staged map[string]lease.MirrorObject) { |
| 3470 | for k := range staged { |
| 3471 | _ = b.Mirror.Delete(ctx, b.Cfg.Bucket, k) |
| 3472 | } |
| 3473 | } |
| 3474 | |
amadan