carlos / platform Public

Clone
git clone https://amadan.net/carlos/platform

Plain git — no account needed to clone.

Download

Download this file

1// The `carlos agent` verb — the single-box role: edge (TLS + host routing) +
2// hibernation activator + hourly release adoption (plus a 2s fast-converge
3// loop, fastConvergeLoop: serial poll + canary/unstaged sweep) + the fleet status
4// page's 5-minute self-report composed in one process. The composition is
5// new here, but each part is a port: the edge from Eleven's router.go, the
6// activator from lchat activator.go, the adopt loop is release.Pass (or,
7// for canaries, release.PassFilter) on the cadence of Eleven's updater
8// timer, and the status loop is internal/status's
9// CollectBox/WriteBox/BuildFleet/PublishFleet run on a timer with one
10// publisher elected via internal/lease.
11//
12// Fleet status page config surface (every env var any part of this file
13// reads for it, gathered in one place since no single function owns the
14// whole contract):
15//
16// - CARLOS_STATUS_BUCKET / CARLOS_STATUS_PREFIX / CARLOS_STATUS_REGION —
17// read by statusLoop. CARLOS_STATUS_BUCKET names the S3 bucket each
18// box writes its own status/boxes/<label>.json snapshot to and (when
19// it wins the lease) publishes status/fleet.json + the viewer assets
20// to. When CARLOS_STATUS_BUCKET is unset, it DEFAULTS to whatever
21// CARLOS_DEPLOYMENT_BUCKET is set to (resolveStatusBucket) — the edge
22// and console both serve the fleet status page out of the same
23// deployment bucket adoptLoop already publishes releases to, so the
24// common single-bucket deployment needs to set only one env var; an
25// explicit CARLOS_STATUS_BUCKET still always wins, for the deployments
26// that deliberately want a second, independent bucket. Both unset ->
27// the loop never starts, logging one line and returning — the same
28// "inert until configured" convention adoptLoop uses for
29// CARLOS_DEPLOYMENT_BUCKET/_DIR, and litestream's own env-gated
30// replication before that. CARLOS_STATUS_REGION feeds lease.Acquire
31// (the single-writer election for the publish half of each tick); it
32// may be empty (falls back to the AWS default chain). CARLOS_STATUS_PREFIX
33// does double duty: it namespaces that same lease key (path.Join(prefix,
34// "status-lease")) AND, since Task 5 of the customer-fleets plan,
35// rebases every key statusLoop's AND metricsLoop's store touches
36// (release.NewPrefixStore) — a fleet box set to
37// CARLOS_STATUS_PREFIX=fleets/<name>/ writes its own status snapshot at
38// fleets/<name>/status/boxes/<label>.json and its metrics/alerts under
39// fleets/<name>/metrics/boxes/... and fleets/<name>/alerts/boxes/...
40// instead of the shared paths every platform box also writes — so
41// neither its status heartbeat nor its metrics/alerts ever collide with
42// (or leak into) the platform's own fleet.json, metrics view, or alert
43// webhook. Empty (every platform box today) is a pure passthrough —
44// EXCEPT on a door-backed fleet box, where empty defaults to
45// "fleets/<fleet>/" because the door admits no other write scope
46// (resolveStatusPrefix). Log shipping is deliberately NOT rebased: see
47// logship.go. A
48// PUBLISHING box with a prefix set would relocate status/fleet.json
49// itself under that prefix too — moot today since nothing sets a prefix
50// on a publishing (non-fleet) box, and a fleet box never publishes at
51// all (CARLOS_FLEET's guard, below).
52// - CARLOS_FLEET — read by statusLoop (Task 5, customer-fleets plan). Set
53// (to the fleet's name — its value is otherwise unused here) marks this
54// agent as a fleet box: shouldSkipPublisherElection short-circuits
55// publish to false BEFORE lease.Acquire is ever called, so a fleet box
56// never contends for the cross-box publish lease and never runs
57// BuildFleet/PublishFleet — it still self-reports via statusTick's
58// unconditional WriteBox, just under its own CARLOS_STATUS_PREFIX.
59// HONEST LIMIT: the CARLOS_DEPLOYMENT_BUCKET fallback only ever
60// resolves to an S3 bucket. A deployment adopting from
61// CARLOS_DEPLOYMENT_DIR (a plain directory — tests, or a single-box
62// no-AWS deployment) has nothing for the fallback to resolve to, so
63// DIR-based deployments do not publish fleet status yet; give them
64// their own CARLOS_STATUS_BUCKET explicitly if they want the page.
65// - CARLOS_FLEET_BOX — this box's operator-chosen fleet_boxes.label (the
66// channel identity the edge dials, "<fleet>/<label>", NEVER the public
67// status page's hashed pseudonym). Optional: defaults to
68// CARLOS_TUNNEL_NAME with the "<fleet>/" prefix stripped, since a fleet
69// box's tunnel name already IS "<fleet>/<label>" (see
70// internal/host/tunnel_client.go). resolveFleetIdentity is the one
71// place both env vars are read and validated (registry.ValidTunnelName)
72// — an invalid result refuses to arm the wake-time/startup binding
73// publish (fleetbind.go) loudly, and degrades the heartbeat's label back
74// to the hashed pseudonym rather than disabling it outright.
75// - CARLOS_TUNNEL_URL / CARLOS_TUNNEL_TOKEN — the tunnel env
76// tunnel_client.go already dials with, read a SECOND time here (spec
77// 2026-08-11-fleet-api-surface, F5) as the fleet store door's address
78// and credential: on a fleet box with NO bucket env set, openDeployStore
79// and the status/metrics store both become internal/fleetapi clients
80// speaking release.Store over the edge's /store/ routes as
81// "<fleet>/<label>", so the box needs no AWS credentials on the
82// platform's deployment bucket at all. A bucket still set WINS
83// everywhere (the migration fallback): every platform box, and any
84// grandfathered fleet box that still carries bucket credentials, keeps
85// direct S3 byte for byte. Missing tunnel env on a fleet box with no
86// bucket is a loud error, not a silent degrade — that box has no way to
87// reach a store at all.
88// - STATUS_BOX_REGION — an operator-set, coarse region word (e.g.
89// "EU · Ireland") carried into status.Probes.Region and published
90// verbatim on this box's snapshot; "" omits the field. Deliberately
91// free text chosen by the operator, never derived from cloud metadata —
92// the public commitment is a human-readable region, not a specific
93// datacenter.
94// - CARLOS_SYSTEM_CHANNEL — platform channel this box follows for
95// self-update, canary|stable (default stable).
96// - CARLOS_LAYOUT_DEFAULT — the instance layout this box converges to
97// when no layout intent is recorded, nested|flat (default nested;
98// isolated-exec design §8.6, "Nested by default"). Nested applies only
99// on a box whose isolation is ready; a recorded intent always wins.
100// - STATUS_HOST / STATUS_PUBLIC — NOT read anywhere in this file. They
101// gate whether ops/tofu fronts the bucket's published status/ prefix
102// with a public hostname; recorded here only so the fleet status
103// page's entire config surface is documented in one place rather than
104// split across this file and the tofu module.
105//
106// The status-lease above is one of TWO elections this process runs, and the
107// only per-tick one. The other is the HOUSEKEEPER lease (housekeeper.go):
108// the same internal/lease mechanism, held long-term, at the fixed
109// deployment-bucket key control/housekeeper-lease/lock.json, electing the one
110// box that performs this deployment's global duties — today the releases
111// retention prune (retentionLoop), from Part 2 of the multi-region spec
112// certificate issuance. It reads CARLOS_DEPLOYMENT_BUCKET (or
113// CARLOS_DEPLOYMENT_DIR, which makes the box its own housekeeper) and the
114// same CARLOS_FLEET guard, and adds no env var of its own.
115//
116// Domains config surface (read by domainsLoop / domainsTickConfig):
117//
118// - PLATFORM_APPS_DOMAIN — the deployment's platform apps domain
119// (flagship: oncarlos.com). UNSET = the domains feature is inert: no
120// platform domains minted, no bucket manifests read, no health stamps —
121// zero behavior change, same litestream-convention gate as the loops
122// above.
123// - CARLOS_WILDCARD_DOMAINS — comma list of static wildcard-cert parent
124// domains for the edge (tito: go.tito.io); per-account parents under
125// PLATFORM_APPS_DOMAIN are derived automatically (edge wildcard.go)
126// - CARLOS_DNS_PROVIDER / CARLOS_DNSIMPLE_TOKEN — DNS-01 backend for
127// wildcard certs: unset/dnsimple needs the token (SSM → unit env),
128// route53 rides the box IAM role; nothing set = wildcards off,
129// per-host HTTP-01 only
130// - CARLOS_ACME_DELEGATION_ZONE — our own zone that delegated wildcard
131// parents' challenge TXTs publish into (spec §4); unset or an invalid
132// shape = delegation off, per-host and direct DNS-01 continue
133// - PLATFORM_EDGE_IPS — comma-separated edge IPs custom-domain DNS checks
134// compare against. Default: this box's public IPv4 from IMDS
135// (best-effort). It is no longer the WHOLE comparison set: each pass
136// unions it with the deployment's LIVE edges — every box's published
137// control/edges/<box>.json whose status snapshot is still fresh
138// (edgesfile.EdgeIPs) — because under latency steering a correctly-
139// pointed domain resolves to whichever region's edge is nearest, and
140// one box's boot-frozen list reads that as "elsewhere". The union only
141// ever WIDENS, so an unreadable bucket degrades to exactly this
142// variable. A pass that ends up knowing no addresses at all stands
143// down rather than guessing "elsewhere" about every host.
144package host
145
146import (
147 "github.com/carlosframework/platform/internal/cliflag"
148
149 "context"
150 "crypto/ecdh"
151 "crypto/ed25519"
152 "errors"
153 "flag"
154 "fmt"
155 "io"
156 "log"
157 "math/rand/v2"
158 "net"
159 "net/http"
160 "os"
161 "os/exec"
162 "os/signal"
163 "path"
164 "path/filepath"
165 "runtime"
166 "slices"
167 "sort"
168 "strings"
169 "sync"
170 "sync/atomic"
171 "syscall"
172 "time"
173
174 "github.com/carlosframework/platform/internal/activator"
175 "github.com/carlosframework/platform/internal/agentsock"
176 "github.com/carlosframework/platform/internal/analytics"
177 "github.com/carlosframework/platform/internal/appconfig"
178 "github.com/carlosframework/platform/internal/apperr"
179 "github.com/carlosframework/platform/internal/applog"
180 "github.com/carlosframework/platform/internal/caps"
181 "github.com/carlosframework/platform/internal/cgroup"
182 "github.com/carlosframework/platform/internal/controlsock"
183 "github.com/carlosframework/platform/internal/domains"
184 "github.com/carlosframework/platform/internal/edge"
185 "github.com/carlosframework/platform/internal/edgesfile"
186 "github.com/carlosframework/platform/internal/fleetapi"
187 "github.com/carlosframework/platform/internal/hostbox"
188 "github.com/carlosframework/platform/internal/hostwrite"
189 "github.com/carlosframework/platform/internal/instancectl"
190 "github.com/carlosframework/platform/internal/instancefs"
191 "github.com/carlosframework/platform/internal/layout"
192 "github.com/carlosframework/platform/internal/lease"
193 "github.com/carlosframework/platform/internal/moves"
194 "github.com/carlosframework/platform/internal/provision"
195 "github.com/carlosframework/platform/internal/registry"
196 "github.com/carlosframework/platform/internal/release"
197 "github.com/carlosframework/platform/internal/rootparticipation"
198 "github.com/carlosframework/platform/internal/rootstate"
199 "github.com/carlosframework/platform/internal/schedule"
200 "github.com/carlosframework/platform/internal/status"
201 "github.com/carlosframework/platform/internal/version"
202)
203
204// dataDir resolves the fleet data directory (CARLOS_DATA_DIR, default /data).
205func dataDir() string {
206 if d := os.Getenv("CARLOS_DATA_DIR"); d != "" {
207 return d
208 }
209 return "/data"
210}
211
212// RunAgent runs edge + activator + adoption in one process. Flags are exactly
213// edge.Run's (-registry, -dev, -http, -https, ...): the args pass through to
214// edge.RunWithActivator, which owns the serving loop and blocks until
215// SIGTERM/SIGINT.
216//
217// With CARLOS_HOST_CLAIM and CARLOS_REGION_HA on, a box that is not yet
218// enrolled with its guard armed runs region HA's joining mode instead, and
219// RunAgent returns once it can host, so systemd starts it again as the
220// claim agent (hostingGate, joining.go).
221//
222// Before either, a box its region's group launched resolves its group
223// reference, whatever the switches say: the loops runAgent starts read
224// the box's label once and keep it (agentEntry, joining.go).
225func RunAgent(args []string) {
226 agentEntry(os.Getenv, resolveGroupBoxReference, groupReferenceBackoff, log.Printf,
227 func() bool { return hostingGate(args) },
228 func() { runAgent(args, productionClaimFactory) })
229}
230
231func runAgent(args []string, buildClaims ClaimFactory) {
232 agentCtx, cancelAgent := context.WithCancel(context.Background())
233 defer cancelAgent()
234 // Startup can stop before children drain; rollout keeps agentCtx until Close.
235 startupCtx, cancelStartup := context.WithCancel(agentCtx)
236 defer cancelStartup()
237 claimCtx, err := claimEntryContext(agentCtx, os.Getenv, openDeployStore)
238 if err != nil {
239 log.Fatalf("agent: host claims: %v", err)
240 }
241 dd := dataDir()
242 regPath := peekRegistryPath(args, dd)
243 // A second handle on the same registry file the edge opens — the registry
244 // is designed for concurrent handles (fresh-connection reads), and this
245 // one feeds the activator's due-time index and the adopt loop.
246 reg, err := registry.Open(regPath)
247 if err != nil {
248 if hint := registryOpenHint(regPath, err); hint != "" {
249 log.Fatalf("agent: open registry %s: %v\nagent: %s", regPath, err, hint)
250 }
251 log.Fatalf("agent: open registry %s: %v", regPath, err)
252 }
253
254 // Constructed once here so the activator's ExecBackend (purges) and the
255 // edge (reads + serves cached hits) share the very same store — see the
256 // package doc comment above about caching applying before the
257 // hibernation branch.
258 cache := edge.NewCache(edge.DefaultCacheMaxEntries, edge.DefaultCacheMaxBodyBytes)
259
260 // Exec-child accounting: claim the cgroup subtree the unit file
261 // delegates to us (Delegate= in carlos-edge.service). Nil when there is
262 // no delegation — an old unit file, a dev run outside systemd, darwin —
263 // in which case children spawn exactly as before and simply have no
264 // per-instance readings.
265 var cgm *cgroup.Manager
266 if hostwrite.Enabled(claimEntryEnv(claimCtx)) {
267 cgm = cgroup.SetupRetaining(log.Printf)
268 } else {
269 cgm = cgroup.Setup(log.Printf)
270 }
271
272 // Alert sinks and the operator paging hook (moves spec §13), built from
273 // the sink env alone and armed here, before armMovesPass and before any
274 // loop that runs a moves pass starts, so a stuck mark never finds paging
275 // unarmed. They do not wait on the metrics store: the metrics loop is
276 // handed this same instance, and a box whose store fails still pages.
277 alerts := armAgentAlerts(agentCtx, os.Getenv, &processAlertCounters, newSNSClient)
278
279 // Log shipping (`carlos logs`): one sink for every producer in this
280 // process, armed only when there is somewhere to ship to — a nil sink
281 // leaves every producer exactly as before (no tees, no pipes, no
282 // events). The box identity is the one metrics publishes under.
283 //
284 // The gate is openStatusStore, not a bucket test: a cut-over fleet box has
285 // no bucket at all and ships through the store door, and gating on the
286 // bucket alone would leave `carlos logs` silently empty for exactly the
287 // boxes an operator can't SSH into. Constructing the store here to throw
288 // it away is deliberate and cheap (neither branch makes a call), and it
289 // asks precisely the question logshipLoop will ask again for itself.
290 var logSink *applog.Sink
291 var errSink *apperr.Sink
292 if _, _, err := openStatusStore(context.Background(), os.Getenv); err == nil {
293 logSink = applog.NewSink(status.Label(readMachineID()), time.Now)
294 errSink = apperr.NewSink(status.Label(readMachineID()), time.Now)
295 }
296
297 cfg, err := activator.ConfigFromEnv()
298 if err != nil {
299 log.Fatalf("agent: %v", err)
300 }
301 claimCtx = withClaimRuntimeResources(claimCtx, ClaimRuntimeResources{Mode: "agent", Registry: reg, Manager: cgm, DataDir: dd, OptDir: release.OptDir(), UnitsDir: "/etc/systemd/system"})
302 // The moves park providers (Plan 5 Task 9a) register before the claim
303 // assembly reads them, then the moves wake source right after them
304 // (Plan 5 Task 9b): only with both switches on. The parks refuse by
305 // name until the wake source registers, which opens the one gate
306 // (movesWakeSourceRegistered), so the parks arm in the same start as
307 // the wake owners. Task 5's expected refusals are quiet in the regional
308 // worker's log and its anomalies page once (movesrefusals.go).
309 movesRefusalLog := newMovesRefusals(metricsBoxLabel(os.Getenv, readMachineID()), PageOperator, log.Printf, time.Now)
310 if parks, err := registerProductionMovesParks(claimCtx, claimEntryEnv(claimCtx), claimStore(claimCtx), *cfg, productionRosterRegion(claimStore(claimCtx), os.Getenv), movesRefusalLog); err != nil {
311 log.Printf("agent: moves: the park providers are not registered: %v", err)
312 } else if err := registerProductionMovesWakeSource(claimEntryEnv(claimCtx), parks, dd); err != nil {
313 log.Printf("agent: moves: the wake source is not registered: %v", err)
314 }
315 // The C8 files manifest provider (moves Task 6) registers under the
316 // claim switch alone: every claim-on serving wake establishes its files
317 // manifest through it (claimfiles.go).
318 if err := registerProductionClaimFiles(claimEntryEnv(claimCtx), dd, reg); err != nil {
319 log.Printf("agent: host claims: the files manifest provider is not registered: %v", err)
320 }
321 if hostwrite.Enabled(claimEntryEnv(claimCtx)) {
322 // Region HA (stage B Task 10): with CARLOS_REGION_HA on, the
323 // relocation owner source is registered before the claim's owners
324 // open. In claim mode the assembly is opened here, as the claim
325 // factory would open it, so the relocation pass, or with region HA
326 // off the completion pass (ruling 82), binds to the same one.
327 if regionHAEnabled(claimEntryEnv(claimCtx)) {
328 if err := registerRelocationSource(claimCtx, dd); err != nil {
329 log.Fatalf("agent: region HA: %v", err)
330 }
331 }
332 if claimCtx, err = ensureProductionClaimAssembly(claimCtx); err != nil {
333 log.Fatalf("agent: host claims: %v", err)
334 }
335 }
336 claims, err := newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
337 if err != nil {
338 log.Fatalf("agent: host claims: %v", err)
339 }
340 if err := configureWakeClaims(claims, cfg); err != nil {
341 log.Fatalf("agent: host claims: %v", err)
342 }
343 // The build-state store (settings spec §4.2), built once after the claim
344 // wiring so it locates claimed hosts through cfg.Claims. nil, nil on a box
345 // with no replica bucket. The activator's BeforeExec hook comes from bs
346 // (buildStateHook, below): a failure here is one operator line, the
347 // agent keeps running, and every spawn that records its build is refused
348 // until it is fixed: every exec-backed wake, worker, sidecar, swap and
349 // claimed start (Tasks 5a-5d). A unit-backed row is started by systemd
350 // on its live link and records nothing (activator identityFor is nil
351 // for it), so it still starts.
352 bs, bsErr := newBuildStates(cfg, reg, release.OptDir())
353 // The status report says this box enforces update settings only while
354 // the build-state store is up (settings spec §4.2; Astra I3 #1).
355 noteBuildStatesInit(bsErr)
356 if bsErr != nil {
357 log.Printf("agent: build state store unavailable: %v. every exec-backed instance, worker and sidecar spawn is refused until it is fixed and the agent restarted (unit-backed rows still start)", bsErr)
358 }
359
360 // Fleet box identity (customer-fleets plan, Task 6 fix round — controller
361 // ruling): resolved ONCE, here, before the activator is even constructed.
362 // identErr is errNotFleetBox for the ordinary platform-box case
363 // (CARLOS_FLEET unset — no log, nothing to arm) or names the specific
364 // validation failure for a fleet box whose identity doesn't resolve to a
365 // usable channel label, in which case the fleet machinery — OnWake and
366 // the startup pass, both — stays disarmed (loud log) rather than run
367 // with a label the edge could never dial. fleetStore is opened here too
368 // (once) so it can be threaded into svc.OnWake BEFORE svc.Start() below —
369 // assigning it after Start() would race the sweeper goroutine's own use
370 // of the field.
371 var fleetStore release.Store
372 identity, identErr := resolveFleetIdentity(os.Getenv)
373 switch {
374 case errors.Is(identErr, errNotFleetBox):
375 // Ordinary platform box; nothing to arm.
376 case identErr != nil:
377 log.Printf("agent: %v — refusing to arm fleet machinery", identErr)
378 default:
379 fs, err := openDeployStore(context.Background())
380 if err != nil {
381 log.Printf("agent: CARLOS_FLEET=%s set but deploy store unavailable: %v — fleet bindings disabled", identity.Fleet, err)
382 } else {
383 fleetStore = fs
384 }
385 }
386
387 // The config guard (amadan #213), armed before the activator and every
388 // reconcile loop that asks it: a box takes an instance's lease, or mints
389 // its record, only if it can provide that instance's config. Armed
390 // whether or not the activator is, because the mint gate is the reconcile
391 // pass's. A box with no deployment store reads it as "nothing sealed
392 // anywhere" and admits as before.
393 guard := newConfigGuard(os.Getenv, openDeployStore)
394 armConfigGuard(guard)
395 guard.Announce()
396
397 var act edge.Activator
398 var svc *activator.Service
399 started := make(chan struct{})
400 var startupDone <-chan struct{}
401 if cfg != nil {
402 // Exec children and carlos-app@ unit tenants share one state machine;
403 // the split backend dispatches per route on Backing. Note the arming
404 // gate is still the hibernate bucket env even though unit tenants
405 // never touch the bucket — one opt-in switch for the whole activator.
406 eb := &activator.ExecBackend{
407 Cfg: *cfg,
408 OptDir: release.OptDir(),
409 DataDir: dd,
410 // The box's placement code, into every exec child's env
411 // (issue #329) — same env the reconcile pass scopes by.
412 Region: strings.TrimSpace(os.Getenv("CARLOS_REGION")),
413 Cache: cache,
414 Materialize: materializeHook(reg, dd),
415 Cgroups: cgm,
416 // The memory tier for each spawned child's cgroup, read off the
417 // staged manifest of the route's live version (memory-pressure
418 // spec §2.3): 512M, or 1G for a family that declares children.
419 Tier: memTier(stagedManifestOfLiveVersion(release.OptDir())),
420 // The same read for one named version, for a generation that runs
421 // a build other than the live link's (zero-downtime spec §7.3).
422 TierOf: memTierOf(func(rt registry.Route, v string) (*release.Manifest, error) {
423 return release.ReadStagedManifest(release.OptDir(), release.RouteAccount(rt), v, rt.App)
424 }),
425 // The scheduled-work verbs on every spawned instance's control
426 // socket write to the SAME registry this agent's converger and
427 // runner read (schedule spec §7.2): an app that asks for a
428 // once-timer from inside itself is asking this box's timer table,
429 // so the row is due-indexed by the next tick with nothing in
430 // between. reg satisfies schedule.TimerStore directly.
431 Timers: reg,
432 // The control socket's read-only GET /updates answers from the
433 // same registry copy (instance update settings spec §5.1).
434 UpdatesReader: updatesReader(reg, release.OptDir()),
435 // Asked inside Acquire, before the lease (amadan #213).
436 ConfigGate: configGate(),
437 }
438 if claims.Writes.Enabled {
439 eb.Timers = schedule.ClaimTimers{TimerStore: reg, Writes: claims.Writes, Context: agentCtx}
440 }
441 if logSink != nil { // a typed-nil in the interface would NOT be nil
442 eb.Logs = logSink
443 }
444 // The verified home keys of this box's replica rows (L21 rulings 26
445 // and 30), read again at every replica launch until O8 reads them
446 // from each home's published env item, and the bucket's
447 // configuration, read fresh before a mint (ruling 5) and at each
448 // replica launch whose key is configured (ruling 27). The claim
449 // path's backend gets the same pair (claimExecBackend).
450 productionInstanceKeyReaders().wire(eb)
451 // The root instance-control channel (design §5.3) and the box facts
452 // a nested wake needs beyond the row (design §8.1, Task 6): the
453 // journal check and the peer list both read this SAME registry
454 // handle, so a nested row's classification here always agrees with
455 // what the rest of this process already knows about the box.
456 eb.Control = instancectl.NewClient(controlsock.Path)
457 eb.LayoutFacts = activator.LayoutInputs{
458 DataDir: dd,
459 // A journal that cannot be read is not proof there isn't one: an
460 // EACCES or EIO on /data/layout must fence the row rather than
461 // wake it as if nothing were in flight, so only a definite
462 // ErrNotExist counts as "no journal". It is the pass's own
463 // predicate, an Lstat, so a symlink planted at the journal path
464 // fences the host rather than being followed.
465 Journal: func(host string) bool { return layout.JournalPresent(dd, host) },
466 Peers: reg.List,
467 }
468 // The layout pass is handed this same gate below, so it never makes
469 // a nested tenant this backend would refuse to start. The gate's
470 // first question is about this process itself: the probe of design
471 // §8.5 runs under the instance templates' unit text and cannot see
472 // the edge's, and RestrictSUIDSGID= on carlos-edge.service once made
473 // every openat2 the agent issues answer ENOSYS (#126). A box in
474 // that state wakes nothing nested and migrates nothing, and says
475 // why, instead of failing on the first host's first open.
476 selfCheck := func() error { return instancefs.SelfCheck(dd) }
477 if err := selfCheck(); err != nil {
478 log.Printf("agent: %v (nested wakes and the layout pass refuse until the unit is fixed)", err)
479 }
480 eb.NestedOK = nestedGate(selfCheck, instancefs.MountNoSymfollow, eb.Control.Ready)
481 // CARLOS_SWAP=overlap turns on generations (plan 1), and with them
482 // plan 3's Swap, before the Service exists, which reads the switch
483 // once in New. Plan 2's per-generation containment goes on with
484 // them: an overlap needs each generation in its own child. A box
485 // that cannot contain generations refuses the flag here and stays
486 // on today's paths (swapArming), so all three stay off.
487 swapOn := swapArming(os.Getenv, eb)
488 activator.SetGenerations(swapOn)
489 eb.GenContainment = swapOn
490 rolloutMaterialize = eb.Materialize
491 svc = activator.New(*cfg, &activator.SplitBackend{
492 Exec: eb,
493 Unit: &activator.UnitBackend{OptDir: release.OptDir()},
494 }, reg) // the registry satisfies activator.DueStore directly
495 // doWake's own re-read of the committed row before it spawns
496 // (design §6.8). The registry satisfies RouteLookup with its own
497 // Get, exactly as DueStore above.
498 svc.Routes = reg
499 // Plans 2: an account's plan can shorten its entries' idle window. With
500 // plans off (no marker, CARLOS_PLAN_LIMITS=off, or no record) the hook
501 // answers 0 and every window is the box's own.
502 svc.SetIdleFor(planIdleFor)
503 // Settings spec §4.2: before any process runs a build against an
504 // instance's database, the build state names it, under the lease. A
505 // store that failed to start refuses every spawn (never "run
506 // unrecorded"). Set before Start, like Routes.
507 svc.BeforeExec = buildStateHook(bs, bsErr)
508 // The held wake check (settings spec §4.3 rule 3): a held instance
509 // never wakes on an installed build other than its target. The copy
510 // is refreshed before every wake that could be held (wakeTarget), and
511 // a target not installed is asked of root through the stage relay,
512 // one throttled requester for every wake. As every stage relay, not
513 // in host-writes mode (ruling R-AGENT-RELAY; release's OnStageBlocked
514 // callers): there a held wake waits for root's own pass. A box with
515 // no build-state store keeps today's wake.
516 if bs != nil {
517 svc.HeldTarget = func(h string) activator.HeldWake { return bs.wakeTarget(context.Background(), h) }
518 if !cfg.HostWrites.Enabled {
519 askStage := stageRequester(dataDir())
520 // A wake waiting for root's decision on new settings names
521 // no build (ruling R-HELD-PURE): one marker for a root pass,
522 // nothing the relay would keep asking for.
523 askPass := stageMarkerRequester(dataDir())
524 svc.RequestStage = func(host, build string) {
525 if build == "" {
526 askPass(host, build, nil)
527 return
528 }
529 askStage(host, build, nil)
530 }
531 }
532 }
533 // Every worker's captured serving build, published for root's pin
534 // set (zero-downtime deploys spec §4.2, ruling R17), whatever
535 // CARLOS_SWAP says. dd is dataDir(), the same data dir root's
536 // RunAdopt and this agent's passes hand release.BakeDir, which is
537 // where the pass reads release.GenerationsPath from. Set before
538 // Start and before any wake.
539 svc.GenerationsFile = release.GenerationsPath(dd)
540 // A retiring generation's schedule writes are refused once a swap
541 // has switched away from it, through each host's write fence, which
542 // the switch takes (zero-downtime spec §7.10). Wired before Start,
543 // like Routes. With swaps off nothing is ever retired or serialized,
544 // so every write applies as before.
545 eb.ScheduleFence = svc.ScheduleWrite
546 if fleetStore != nil {
547 // MUST be assigned before Start() below spawns the sweeper
548 // goroutine, which reads svc.OnWake on every successful wake from
549 // then on — setting it any later is a data race.
550 svc.OnWake = fleetBindingHook(fleetStore, identity.Fleet, identity.Label)
551 }
552 if logSink != nil {
553 svc.Logs = logSink
554 }
555
556 // The startup sweep (design §6.7): a leftover instance unit from a
557 // previous agent process, one this box's registry no longer holds
558 // a live wake for, is stopped before this process starts spawning
559 // its own. Best-effort, like every other startup duty here: a
560 // channel that never answers logs and moves on, and the box is no
561 // worse off than before this existed (a leftover unit that keeps
562 // running is exactly what an agent restart left before this pass).
563 //
564 // Then, on a legacy box, layout recovery before Start can wake
565 // anything, and the instance-key backfill in the background after
566 // it, which this does not wait for (the edge below must listen
567 // within the update gate, whatever the box's size); on a claim
568 // box, Start alone. The order and the branch are activatorStart's.
569 svc.Startup = started
570 startupDone = activatorStart{
571 Configure: func(_ context.Context) {
572 armAlways(svc, reg, dd) // flag on or off
573 var rolloutSink activator.EventSink
574 if logSink != nil { // a typed nil in the interface would not be nil
575 rolloutSink = logSink
576 }
577 armRolloutAndRelay(agentCtx, os.Getenv, svc, reg, dd, rolloutSink, openDeployStore)
578 // The residence-move machinery (issue #331), armed with the
579 // activator: the staging-fallback restore needs a same-region
580 // mirror client, and the move worker needs the service's quiesce
581 // answer. Both are best-effort — a mirror client that cannot
582 // build logs and leaves the fallback dark, exactly like every
583 // other optional duty here.
584 if m, err := lease.NewS3Mirror(agentCtx, cfg.Region, cfg.Region); err != nil {
585 log.Printf("agent: staging-fallback mirror client: %v (restore fallback disarmed)", err)
586 } else {
587 eb.Mirror = m
588 }
589 // Seeded instances (seeded-instances spec §3.4): the seed reader,
590 // the placed recorder, the provision pass's hooks and the claim
591 // flag the seed fallback refuses under. In Configure, like the
592 // mirror, as it was before the startup moved to the background
593 // (#281): before the sweep, before Start, and so before Ready
594 // closes the startup latch. Every wake waits on that latch
595 // (awaitStartup) and the reconcile loops start after it, so no
596 // seeded row is woken, and no seeded record acted on, before
597 // this has run. The import swap and rollback (spec §5.4, §5.6)
598 // fence through this activator, delete the replica under
599 // cfg.Prefix, and trust an unproven stop only after the startup
600 // sweep below (seedSweptAtStart, set in Recover, which also
601 // comes before the latch).
602 wireSeedPlacement(eb, claims.Writes.Enabled, seedDataOpener(cfg.Bucket, cfg.Region),
603 func() (release.Store, error) { return openDeployStore(context.Background()) },
604 seedBox{Fence: svc, Prefix: cfg.Prefix, Swept: seedSweptAtStart.Load, NextFenceGen: fenceGenerations(reg)})
605 armMovesPass(*cfg, dd, reg, svc, eb, claims.Writes, claims.Providers, rolloutSink, moveCarry(bs, bsErr))
606 // carlos run's worker, armed with the moves pass as it was
607 // before the startup moved to the background (#281). Its
608 // passes ride the converge and instances loops, which the
609 // duties start only once Ready has closed the startup
610 // latch: no run is picked up before the startup sweep has
611 // stopped a previous agent's leftover units, the cmd units
612 // among them.
613 armRunsPass(svc, reg, dd, rolloutSink)
614 // The db-reads loop (db region forward spec §7.1): armed
615 // only on a box that can serve; it takes nothing until a
616 // console asks.
617 armDBReads(*cfg)
618 },
619 Ready: func() {
620 close(started)
621 log.Printf("agent: hibernation activator armed (bucket %s, idle %s)", cfg.Bucket, cfg.Idle)
622 },
623 Claims: claims.Writes.Enabled,
624 Sweep: func(ctx context.Context) error {
625 sw, err := eb.Control.Sweep(ctx)
626 if err != nil {
627 log.Printf("agent: instance sweep: %v (nested wakes refuse until the channel answers)", err)
628 } else if len(sw.Units) > 0 {
629 log.Printf("agent: swept %d instance unit(s) left from a previous agent: %v", len(sw.Units), sw.Units)
630 }
631 return err
632 },
633 Recover: func(ctx context.Context, swept bool) *layout.Pass {
634 seedSweptAtStart.Store(swept)
635 pass := newLayoutPass(reg, svc, dd, eb.NestedOK, swept)
636 intent := startupRecoveryIntent(ctx, dd, openDeployStore, metricsBoxLabel(os.Getenv, readMachineID()))
637 recoverLayoutAtStart(ctx, pass, intent)
638 return pass
639 },
640 Routes: reg.List,
641 Route: reg.Get,
642 Keys: eb,
643 Slot: svc.PrepareScheduledToken,
644 Publish: armedLayoutPass.Store,
645 Start: svc.Start,
646 }.Run(startupCtx)
647 act = svc
648
649 // The agent socket: the one channel a sidecar dials back over
650 // (internal/agentsock, Task 7). Constructed only in this branch —
651 // the same condition that constructs the activator — because with
652 // hibernation off there are no workers on the box at all, and
653 // nothing would ever call in. cfg.AgentSocket is never empty here:
654 // activator.ConfigFromEnv defaults it to DefaultAgentSocket.
655 //
656 // The cancel is deferred right here, not called explicitly at the
657 // shutdown site below: RunAgent's defers run LIFO at return, so this
658 // one (registered first) fires AFTER the shutdown block's own
659 // `defer cancel()` for svc.Shutdown's context — i.e. once the
660 // children are already down — while staying the one place this
661 // context's lifetime is decided, which is what satisfies go vet's
662 // lostcancel check.
663 agentSockCtx, agentSockCancel := context.WithCancel(context.Background())
664 defer agentSockCancel()
665 agentSrv := agentsock.New(svc)
666 go serveAgentSocket(agentSockCtx, cfg.AgentSocket, agentSrv.Serve)
667 log.Printf("agent: agent socket armed at %s", cfg.AgentSocket)
668
669 // The boot pass (Task 8): a sidecar that does not hibernate must be
670 // running whenever the agent is — an agent restart otherwise leaves
671 // it down until a request or its due time happens to arrive, and an
672 // always-on worker has no request path at all to wake it. Run off
673 // the startup path (its own goroutine, not blocking RunAgent) since
674 // WakeWorker can block for the settle window per host and this must
675 // never delay the edge coming up; runWorkerBootPass itself wakes
676 // its candidates concurrently, bounded, rather than serially.
677 go func() {
678 select {
679 case <-started:
680 case <-startupCtx.Done():
681 return
682 }
683 routes, err := reg.List()
684 if err != nil {
685 log.Printf("agent: worker boot pass: list routes: %v", err)
686 return
687 }
688 runWorkerBootPass(routes, svc.WakeWorker)
689 }()
690 }
691
692 if svc == nil {
693 close(started)
694 }
695
696 seedEnabled := claims.Writes.Enabled && os.Getenv("CARLOS_HOST_CLAIM_ORIGINS") == "1" && svc != nil
697 safeClaimClose := true
698 var shutdownCtx context.Context
699 var shutdownCancel context.CancelFunc
700 defer func() {
701 if shutdownCancel != nil {
702 defer shutdownCancel()
703 }
704 cancelAgent()
705 if !safeClaimClose {
706 return
707 }
708 base := shutdownCtx
709 if !seedEnabled || base == nil {
710 base = context.Background()
711 }
712 ctx, cancel := context.WithTimeout(base, 5*time.Second)
713 defer cancel()
714 if err := claims.close(ctx); err != nil {
715 log.Printf("agent: regional claims shutdown: %v", err)
716 }
717 }()
718 var stopHousekeeper func()
719 if !claims.Writes.Enabled {
720 // Certificate issuance must not wait for instance recovery.
721 stopHousekeeper = startHousekeeper()
722 }
723 dutiesDone := make(chan struct{})
724 runDuties := func() {
725 defer close(dutiesDone)
726 select {
727 case <-started:
728 case <-startupCtx.Done():
729 return
730 }
731 if startupCtx.Err() != nil {
732 return
733 }
734
735 // Fleet box binding publish, startup half (customer-fleets plan, Task 6):
736 // a one-shot pass over every route already resident in this box's local
737 // registry, covering an agent restart (activator.Service's in-memory
738 // entries are gone, but an already-running child isn't) and routes that
739 // never go through activator.wake at all. Non-authoritative
740 // (publishResidentFleetBindings calls WriteFleetBinding with
741 // takeover=false) — see fleetbind.go's doc comments for why a bare
742 // "this box has this route resident" pass must never steal a binding a
743 // wake-success write (lease-backed, authoritative) already holds for a
744 // DIFFERENT box.
745 if fleetStore != nil {
746 if svc == nil {
747 log.Printf("agent: CARLOS_FLEET=%s set but the hibernation activator is not armed — wake-time fleet bindings will never publish", identity.Fleet)
748 }
749 go publishResidentFleetBindings(fleetStore, identity.Fleet, identity.Label, reg)
750 }
751
752 // Scheduled work (schedule spec §6.2/§6.3), built ONCE here and shared by
753 // all three legs below: the seed and the fast leg in fastConvergeLoop, the
754 // hourly leg in adoptLoop, and the runner started here. One converger is a
755 // correctness requirement, not tidiness — its etag memory is what makes a
756 // declaration change cheap and a run signal exactly-once, and a second
757 // instance elsewhere would replay or swallow one. The three legs run in
758 // two different goroutines and do overlap (an hourly pass mid-flight when
759 // a console bump fires the fast one); Converger.Pass serializes itself on
760 // its own mutex, so no call site here needs one.
761 //
762 // The two halves arm on different conditions — armSchedule
763 // (schedule_wiring.go) is that switch, and carries the why. In short:
764 // declared schedules live in the bucket, so the converger needs the
765 // deploy store; the RUNNER needs only the registry, because once-timers
766 // are registered by instances on their control socket (mounted above,
767 // Timers: reg, whenever the activator is armed) straight into this box's
768 // rows, and a box that mounts the verb must also fire what it accepts.
769 //
770 // Guarded assignment, not a direct pass: logSink is a *applog.Sink, and
771 // a nil one handed to an interface parameter is NOT a nil interface —
772 // the same trap the activator's Logs assignment above avoids.
773 var sink activator.EventSink
774 if logSink != nil {
775 sink = logSink
776 }
777 schedStore, schedErr := openDeployStore(agentCtx)
778 sd := armSchedule(reg, schedStore, schedErr, svc, sink, dd)
779 // Move holds (moves spec §9.4): one gate for the runner, the converger
780 // and a cut's drain, and the converger's freeze. Nothing with the
781 // Phase 1 switch off.
782 armScheduleHolds(sd, armedHoldView())
783 if err := configureScheduleClaims(claims, sd); err != nil {
784 log.Fatalf("agent: host claims: %v", err)
785 }
786 if sd != nil {
787 sd.runner.Start(context.Background()) // runs until the process ends, like every loop below
788 }
789 serialClaims := make(chan struct{}, 1)
790 regionalReport := func(err error) { log.Printf("agent: regional claims: %v", err) }
791 if hostwrite.Enabled(claimEntryEnv(claimCtx)) && moves.Enabled(os.Getenv) {
792 regionalReport = movesRefusalLog.report
793 }
794 if err := claims.startRegional(agentCtx, serialClaims, regionalReport); err != nil {
795 log.Fatalf("agent: regional claims: %v", err)
796 }
797
798 // The relocation pass (stage B Task 10): it says itself what it started.
799 startRelocationPass(agentCtx, claimCtx, claims, reg, svc)
800 // The change index (S3 change-index design §4): armed before the
801 // loops that read through it, with its backstop running from here.
802 // Unset CARLOS_CHANGE_INDEX, or a fleet box: nothing.
803 if v := armChangeIndex(agentCtx, os.Getenv, openDeployStore); v != nil {
804 go recordViewBackstopLoop(agentCtx, v)
805 }
806 go adoptLoop(reg, svc, sd, claims.Writes)
807 go fastConvergeLoop(reg, svc, sd, claims.Writes, serialClaims, bs, claims.Providers.Provision)
808 if logSink != nil {
809 go logshipLoop(logSink, status.Label(readMachineID()))
810 }
811 if errSink != nil {
812 go errshipLoop(errSink, status.Label(readMachineID()))
813 }
814 // The build-state report (settings spec rev 11 §4.4) names the rows
815 // keepsBuildState selects, with host claims as the build-state store
816 // sees them (cfg.HostWrites.Enabled, set by configureWakeClaims).
817 buildStateClaims := cfg != nil && cfg.HostWrites.Enabled
818 collectBuildStates := func() (map[string]status.BuildStatePins, error) { return buildStatePins(reg, buildStateClaims) }
819 go statusLoop(reg, svc, collectBuildStates)
820 // The memory early-warning (memory-pressure spec §2.4): per box, gated
821 // by CARLOS_HOST_METRICS — one log line and nothing else where unset.
822 go memwatchLoop()
823 go domainsLoop(reg, peekAcmeCache(args), claims.Writes.Enabled)
824 go instancesLoop(reg, claims.Writes, bs, claims.Providers.Provision)
825 // The deployment-global election, armed BEFORE the loops it gates so a
826 // single-box deployment is holding well before their first tick (a
827 // minute away) — see housekeeper.go. Not a `go` of its own: it publishes
828 // the process's housekeeper and spawns its loop itself. Its stand-down
829 // runs in the post-drain block at the end of this function.
830 if claims.Writes.Enabled {
831 stopHousekeeper = startHousekeeper()
832 }
833 go retentionLoop(reg, collectBuildStates)
834 go registrySnapshotLoop(reg)
835 go edgesFileLoop(reg)
836 // Region HA enrollment evidence (spec 1 §5.1): once, at start, and only
837 // with CARLOS_REGION_HA on. See boxrecord.go.
838 go boxRecordLoop()
839 // The mesh key's own loop (multi-region backhaul spec, Part 3 Task 2) —
840 // deliberately NOT a duty riding retentionLoop's tick: that loop returns
841 // early, never arming at all, on a box missing CARLOS_RELEASE_PUBKEY,
842 // and such a box still WINS the housekeeper election. A mint duty living
843 // only inside retentionLoop would then silently never run fleet-wide.
844 // See mintMeshKeyLoop's own comment (housekeeper.go).
845 go mintMeshKeyLoop()
846 // The DNS steering converger (multi-region ingress spec, Part 4) — its
847 // own loop for mintMeshKeyLoop's exact reason, and dormant on every box
848 // today: CARLOS_STEERING_ZONE_ID unset means one log line at arm and
849 // nothing else, ever. See steering_loop.go.
850 go steeringLoop()
851 // The sweep reads the registry only through this closure — the stamped
852 // bespoke units to sample beside the carlos- ones, re-read each tick.
853 go metricsLoop(cgm.Root(), func() []string { return stampedUnitNames(reg) }, alerts)
854 go configSyncLoop(reg, svc, dd, claims.Writes)
855 go configGenLoop(reg, svc, dd, claims.Writes)
856 go socketAdvisoryLoop(reg, svc)
857
858 }
859 if claims.Writes.Enabled {
860 runDuties()
861 } else {
862 go runDuties()
863 }
864
865 // Seed handlers borrow this agent's resources and finish before they close.
866 var seedCtx context.Context
867 var seedCancel context.CancelFunc
868 var seedServer *filesSeedServer
869 if seedEnabled {
870 seedCtx, seedCancel = signal.NotifyContext(claimCtx, syscall.SIGTERM, os.Interrupt)
871 defer seedCancel()
872 var seedFinish func()
873 seedCtx, seedFinish = activator.WithCleanupBudget(seedCtx, 30*time.Second)
874 defer seedFinish()
875 stopSeed := context.AfterFunc(startupCtx, seedCancel)
876 defer stopSeed()
877 seedServer, err = startAgentFilesSeed(seedCtx, svc.Startup, svc)
878 if err != nil {
879 log.Printf("Files seed failed: %s.", err)
880 }
881 }
882
883 // The same serving loop as `carlos edge`, with the activator injected.
884 // Caching applies here too: cached hits still avoid the activator, which
885 // is the entire point of checking the cache before the hibernation branch.
886 // The ready hook fires synchronously before listening, so
887 // tunnelLoop is handed off to its own goroutine rather than called here —
888 // calling it directly would block the front door forever.
889 edgeOpts := append([]edge.RunOption{
890 // Certificate issuance is a deployment-global duty, so it runs where
891 // the housekeeper runs (housekeeper.go). The gate goes DOWN into the
892 // edge as a func — this package imports internal/edge and never the
893 // reverse — and it is the GATE, not the bare flag: an ACME order runs
894 // for seconds to minutes, and the context ends the instant this box
895 // stops being the housekeeper. Armed above, so this closure is live
896 // and correct by the time the edge reads it.
897 edge.WithIssuerGate(housekeeperGate),
898 // Whether this box contends in that election AT ALL — the cert
899 // backfill's eligibility (#318). Mode is decided once from the
900 // environment, so the closure is stable; see housekeeperContends
901 // for the fleet-box exclusion it exists to state.
902 edge.WithIssuerContender(housekeeperContends),
903 // This box's own label, for the home map's own-box exclusion
904 // (internal/edge/homemap.go). metricsBoxLabel is the SAME call
905 // edgesFileLoop publishes under, deliberately: the map excludes our
906 // hosts by matching the box field of the file we ourselves wrote, so
907 // the two derivations must be one derivation. Without it the edge
908 // would offer nothing (fail-closed), never dial itself.
909 edge.WithBoxLabel(metricsBoxLabel(os.Getenv, readMachineID())),
910 // The change index (#336): the cert mirror and the home map read
911 // through the process's one watch. A loader, not the watch: on a box
912 // without claim writes the index is armed on runDuties' goroutine,
913 // which may not have run yet. Disarmed, it returns nil for good and
914 // both keep today's full reads.
915 edge.WithChangeWatch(armedChangeWatch.Load),
916 },
917 // The edge-local door (#175 P7), only when CARLOS_EDGE_DOOR names a
918 // socket. It presents this box's public address, the one the edges
919 // file publishes; with none, the edge keeps the door shut and says so.
920 edgeDoorOptions(os.Getenv, readPublicIP)...)
921 // A tunnel client takes the hub's GeoIP answer and mirrors nothing (geoip
922 // spec §5.1); the same rule tunnelLoop arms on.
923 if tunnelEnvComplete(os.Getenv) {
924 edgeOpts = append(edgeOpts, edge.WithGeoTunnelClient())
925 }
926 edge.RunWithActivatorReady(args, act, cache, func(e *edge.Edge) {
927 go tunnelLoop(e)
928 // The edge owns the health probers, so its state table is only
929 // reachable once it exists — statusLoop (already running above) reads
930 // it through this published source on its next tick. Method value, so
931 // the loop always sees the CURRENT probe set: Reload replaces probers
932 // behind this handle without the heartbeat needing to be re-wired.
933 setUpstreamSource(e.UpstreamStates)
934 // The GeoIP block for the heartbeat (geoip spec §5.5), the same way.
935 geoIPSource.Store(e.GeoIPState)
936 geoIPAlertSource.Store(e.GeoIPAlertCode)
937 // The layout pass reloads routing through this as soon as a move
938 // commits, instead of waiting for the 2s registry poll.
939 reload := func() { _ = e.Reload() }
940 layoutReload.Store(&reload)
941 m := edge.NewUsageMeter()
942 e.EnableUsage(m)
943 // Same arming point as the meter: the edge observes an upstream 5xx
944 // on the serving path and files it here (spec 2026-09-07 §3.2), and
945 // errshipLoop above ships what this fills. Guarded, because a typed
946 // nil inside the interface would NOT be nil to the edge.
947 if errSink != nil {
948 e.EnableErrors(errSink)
949 }
950 // The analytics meter (spec 2026-09-07 §3.3): the edge counts
951 // pageviews per (account, app) while telemetryPass (fullPass and
952 // instancesLoop) keeps its switch map current, and usageLoop
953 // drains it into day objects on the same 5-minute tick as usage.
954 am := analytics.NewMeter(dataDir(), status.Label(readMachineID()), time.Now)
955 e.EnableAnalytics(am)
956 setAnalyticsSink(e.SetAnalyticsOn)
957 go usageLoop(m.Drain, am)
958 }, edgeOpts...)
959
960 var seedErr error
961 shutdownCtx, shutdownCancel, seedErr = joinAgentShutdown(seedCtx, seedCancel, seedServer, cancelStartup, startupDone, dutiesDone)
962 if seedErr != nil {
963 safeClaimClose = false
964 log.Printf("Files seed failed: %s.", seedErr)
965 os.Exit(1)
966 }
967
968 // The edge has drained; take the children down too (they're this process's
969 // children — orphaning them would strand their leases until TTL expiry).
970 if svc != nil {
971 ctx := shutdownCtx
972 svc.Shutdown(ctx)
973 // Then the scheduler, under the same budget (spec §8.6, Task 11): an
974 // attempt whose swap Shutdown just cancelled finishes its done(), its
975 // outcome and the end of its intent on disk, and so does a wake event
976 // under way, before this process exits. A planned restart mid-swap so
977 // leaves nothing for the next agent to reconcile.
978 if m := armedRollout(); m != nil {
979 m.Close(ctx)
980 }
981 // The agent socket's own defer (above, at its construction) stops it
982 // once this function returns — after Shutdown, per the LIFO ordering
983 // explained there. That gap is real, not zero-width: svc.Shutdown
984 // returning here and agentSockCancel actually firing are two
985 // different points in the stack unwind, so the socket is technically
986 // still accepting connections for the width of that unwind. It is
987 // harmless: Shutdown has already stopped every entry and set
988 // shuttingDown, so a call landing in the gap finds no live entry —
989 // WorkerToken returns "" (authorize's unmatchable() then refuses it),
990 // and NoteWorkerContact/NoteWorkerReady/SleepWorker are all no-ops on
991 // a nil entry lookup. Nothing in the gap can reactivate a stopped
992 // worker or authorize against a dead token.
993 }
994 // Same principle one level up: this process may hold the DEPLOYMENT's
995 // housekeeper lease, and a restart that walked away from it would leave
996 // control/housekeeper-lease/lock.json standing for up to its 15-minute
997 // TTL — no pruner, and from Part 2 no cert issuer, across an ordinary
998 // roll of a single-box deployment. Stopping cancels every duty in flight
999 // and releases the lock, so the next box (or this one, a second later)
1000 // takes over immediately.
1001 if stopHousekeeper != nil {
1002 stopHousekeeper()
1003 }
1004}
1005
1006// statusSocketProbe is the kernel-table probe the status loop hands
1007// wrapListeningProbe: listeningProbe itself with generations off (or no
1008// activator), and pointed at each managed route's current generation with
1009// them on (currentSocketProbe).
1010func statusSocketProbe(routes []registry.Route, svc *activator.Service) func(string) *bool {
1011 if svc == nil || !svc.GenerationsOn() {
1012 return listeningProbe
1013 }
1014 return currentSocketProbe(routes, svc.AppSocket, listeningProbe)
1015}
1016
1017func configureWakeClaims(w *ClaimWiring, cfg *activator.Config) error {
1018 if w == nil {
1019 return hostwrite.ErrUnavailable
1020 }
1021 if !w.Writes.Enabled {
1022 return nil
1023 }
1024 if cfg == nil || w.Providers.Wake == nil || w.Writes.Selector == nil {
1025 return hostwrite.ErrUnavailable
1026 }
1027 cfg.HostWrites, cfg.Claims = w.Writes, w.Providers.Wake
1028 return nil
1029}
1030
1031func configureScheduleClaims(w *ClaimWiring, sd *scheduleDeps) error {
1032 if w == nil {
1033 return hostwrite.ErrUnavailable
1034 }
1035 if !w.Writes.Enabled {
1036 return nil
1037 }
1038 if sd == nil || sd.runner == nil || sd.conv == nil || w.Writes.Selector == nil {
1039 return hostwrite.ErrUnavailable
1040 }
1041 sd.runner.HostWrites, sd.conv.HostWrites = w.Writes, w.Writes
1042 sd.runner.StartDelivery = w.Providers.Delivery
1043 return nil
1044}
1045
1046// --- Task 8: the agent socket's retry/backoff, and the worker boot pass ----
1047
1048// workersToStart is the boot pass's selection: a sidecar that does not
1049// hibernate must be running whenever the agent is. A hibernating sidecar is
1050// deliberately NOT started here — it waits for its due time, an operator
1051// command, or (later) a delivery signal, which is the whole point of the
1052// class. Status is checked the same way the due-time sweep checks it
1053// (activator.go's sweepOnce): only "ready" (or the legacy "" some pre-status
1054// rows carry) is eligible — a row still provisioning, or draining, must not
1055// be spawned.
1056func workersToStart(routes []registry.Route) []registry.Route {
1057 var out []registry.Route
1058 for _, rt := range routes {
1059 if rt.IsSidecar() && !rt.Hibernate && (rt.Status == "" || rt.Status == "ready") {
1060 out = append(out, rt)
1061 }
1062 }
1063 return out
1064}
1065
1066// workerBootConcurrency bounds how many sidecars the boot pass wakes at
1067// once. WakeWorker blocks for the settle window (activator's sidecarSettle,
1068// 10s in production) whenever a child never calls /v1/ready — a slow or
1069// silent sidecar, or simply one that only survives the window rather than
1070// reporting early — so an unbounded fan-out could still spawn every child on
1071// the box in the same instant (a thundering herd against S3 leases and
1072// restore reads), while a serial pass would cost up to 10 seconds PER
1073// sidecar of agent startup. This bound trades both off: bounded fan-out, not
1074// one-at-a-time.
1075const workerBootConcurrency = 8
1076
1077// runWorkerBootPass wakes every route workersToStart selects out of routes,
1078// concurrently (bounded by workerBootConcurrency) rather than one at a time
1079// — see workerBootConcurrency's doc comment for why serial is too slow and
1080// unbounded is too bursty. wake is svc.WakeWorker in production; a seam here
1081// so the concurrency shape is testable without a real activator.Service. A
1082// wake that errors is logged and does not stop the rest — one sidecar that
1083// can't come up must not strand every other one on the box at boot.
1084func runWorkerBootPass(routes []registry.Route, wake func(registry.Route) error) {
1085 toStart := workersToStart(routes)
1086 if len(toStart) == 0 {
1087 return
1088 }
1089 sem := make(chan struct{}, workerBootConcurrency)
1090 var wg sync.WaitGroup
1091 for _, rt := range toStart {
1092 rt := rt
1093 wg.Add(1)
1094 sem <- struct{}{}
1095 go func() {
1096 defer wg.Done()
1097 defer func() { <-sem }()
1098 if err := wake(rt); err != nil {
1099 log.Printf("agent: worker boot pass: wake %s: %v", rt.Host, err)
1100 }
1101 }()
1102 }
1103 wg.Wait()
1104}
1105
1106// agentSocketRetryBase/Cap bound the agent socket's rebind backoff — vars,
1107// not consts, so a test can shrink them (sidecarSettle's precedent in
1108// internal/activator).
1109var (
1110 agentSocketRetryBase = time.Second
1111 agentSocketRetryCap = 30 * time.Second
1112)
1113
1114// serveAgentSocket runs serve (agentsock.Server.Serve in production) in a
1115// retry loop with doubling backoff, forever, until ctx is cancelled. serve
1116// is a seam so the retry/backoff shape is testable without a real unix
1117// socket.
1118//
1119// A failed bind must NEVER be fatal to the agent: agentsock.Serve
1120// deliberately REFUSES a socket another listener already owns (rather than
1121// unlinking it) to prevent two agents both believing they own the box's
1122// workers, and on an agent restart the old process's listener can take a
1123// few seconds to release — that refusal is expected and transient. An agent
1124// that exited because a socket was briefly busy would take down every route
1125// on the box, so this loop only ever logs and retries; it has no path that
1126// calls log.Fatal or returns an error to a caller who might.
1127//
1128// A clean return from serve (nil, once ctx is cancelled — agentsock.Serve's
1129// own contract) ends the loop immediately with no retry: that is a normal
1130// shutdown, not a failure.
1131func serveAgentSocket(ctx context.Context, path string, serve func(context.Context, string) error) {
1132 backoff := agentSocketRetryBase
1133 for {
1134 err := serve(ctx, path)
1135 if err == nil || ctx.Err() != nil {
1136 return
1137 }
1138 log.Printf("agent: agent socket %s: %v — retrying in %s", path, err, backoff)
1139 timer := time.NewTimer(backoff)
1140 select {
1141 case <-timer.C:
1142 case <-ctx.Done():
1143 timer.Stop()
1144 return
1145 }
1146 backoff *= 2
1147 if backoff > agentSocketRetryCap {
1148 backoff = agentSocketRetryCap
1149 }
1150 }
1151}
1152
1153// nestedGate is what an isolated wake requires of the box (design §8.5,
1154// §8.6): this process able to make the openat2 call instancefs is built on
1155// (self; the edge unit's own hardening is what can take it away, #126),
1156// the three nosymfollow mounts a nested instance's tree depends on, and a
1157// control service that reports ready. The answer is recomputed on
1158// every wake (the mounts are cheap to read; /ready is one short request),
1159// so a box that becomes ready needs no agent restart and a box that stops
1160// being ready refuses the very next wake rather than the stale verdict a
1161// once-at-startup check would give.
1162func nestedGate(self func() error, mounts func(string) (bool, error), ready func(context.Context) error) func() error {
1163 return func() error {
1164 if err := self(); err != nil {
1165 return err
1166 }
1167 for _, m := range []string{"/data/instances", "/run/carlos/instances", "/run/carlos/logs"} {
1168 ok, err := mounts(m)
1169 if err != nil {
1170 return fmt.Errorf("%s: %v", m, err)
1171 }
1172 if !ok {
1173 return fmt.Errorf("%s is not a nosymfollow mount", m)
1174 }
1175 }
1176 ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineReady)
1177 defer cancel()
1178 return ready(ctx)
1179 }
1180}
1181
1182// peekRegistryPath pre-parses the agent args for -registry (default
1183// <dataDir>/registry.db) without disturbing them — edge.RunWithActivator does
1184// the authoritative parse. The set mirrors edge.Run's flags so a full command
1185// line parses cleanly; an unknown flag just ends the peek early (the edge's
1186// own ExitOnError parse then reports it properly).
1187func peekRegistryPath(args []string, dd string) string {
1188 fs := flag.NewFlagSet("agent", flag.ContinueOnError)
1189 fs.SetOutput(io.Discard)
1190 regPath := fs.String("registry", filepath.Join(dd, "registry.db"), "")
1191 fs.String("acme-cache", "", "")
1192 fs.String("https", "", "")
1193 fs.String("http", "", "")
1194 fs.String("fallback", "", "")
1195 fs.Bool("dev", false, "")
1196 fs.Usage = cliflag.Usage(fs, "")
1197 _ = fs.Parse(args)
1198 return *regPath
1199}
1200
1201// materializeHook builds the ExecBackend's cold-wake staging hook, or nil
1202// when this box has no release store/pubkey — nil keeps today's behavior
1203// ("no live binary for <host>" wake failure), the safe degrade for a box
1204// whose routes are versioned by hand or not at all.
1205//
1206// The hook is the adopt pass scoped to one host (release.MaterializeHost),
1207// so a cold wake with nothing staged pays one slow request instead of 503ing
1208// until the hourly pass comes round (issue #77). Everything the pass refuses
1209// it still refuses here: a bake-held head is NOT materialized, and the wake
1210// fails with that reason rather than the head being installed early.
1211//
1212// Store and key are resolved ONCE, at agent start, exactly like adoptLoop's:
1213// a wake is on a request's critical path, and re-reading env plus re-opening
1214// an S3 client per cold wake would put that work there for nothing.
1215//
1216// OnStageBlocked is wired so a cold wake on a hardened box (this sandboxed
1217// process cannot write /opt/carlos) relays the stage it cannot perform to
1218// the root pass (carlos-stage.path) instead of just failing the wake. NO
1219// RestartHibernating here, deliberately: the host is cold, restart stays
1220// nil — the activator exec'ing the freshly-pointed binary IS the restart.
1221// The requester is built once with the hook, so its once-a-minute budget is
1222// the hook's own, same shape as the loops'.
1223func materializeHook(reg *registry.Registry, dd string) func(context.Context, string) error {
1224 store, err := openDeployStore(context.Background())
1225 if err != nil {
1226 log.Printf("agent: cold-wake materialize disabled: %v", err)
1227 return nil
1228 }
1229 pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
1230 if err != nil {
1231 log.Printf("agent: cold-wake materialize disabled: %v", err)
1232 return nil
1233 }
1234 optDir := release.OptDir()
1235 stageReq := stageRequester(dd)
1236 return func(ctx context.Context, host string) error {
1237 // The process's one proof gate (proofask.go), so a cold wake cannot
1238 // adopt an unproven build of a listed app.
1239 return release.MaterializeHost(ctx, reg, store, pub, optDir, dd, host,
1240 release.OnStageBlocked(stageReq), release.WithProofGate(agentProofGate()),
1241 release.WithDigestRatchet(pointerRatchet(false, optDir, dd, os.Getenv)),
1242 updatesPassOption(dd, false))
1243 }
1244}
1245
1246// adoptLoop runs a release-adoption pass roughly hourly (plus jitter, so a
1247// fleet never hits the store in the same second), forever. Missing deployment
1248// config disables the loop with one log line — an agent with no release store
1249// is a valid box (routes managed by hand, adopted by the timer unit, or not
1250// versioned at all).
1251// The activator is taken so the pass can cycle exec-backed instances the way
1252// only this process can (see makeRestartRoute); nil is fine — a box with no
1253// hibernation armed simply has no children to bounce.
1254//
1255// sd carries the scheduled-work converger; this loop is its HOURLY floor (the
1256// fast one rides the serial in fastConvergeLoop). sd.converge is a no-op when
1257// the box has no deploy store — the agent logged that once at arming.
1258func adoptLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config) {
1259 ctx := context.Background()
1260 store, err := openDeployStore(ctx)
1261 if err != nil {
1262 log.Printf("agent: adopt loop disabled: %v", err)
1263 return
1264 }
1265 pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
1266 if err != nil {
1267 log.Printf("agent: adopt loop disabled: %v", err)
1268 return
1269 }
1270 optDir := release.OptDir()
1271 restart := makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), adoptBounceFn(svc))
1272 log.Printf("agent: adopt loop armed (hourly + jitter)")
1273 // First pass shortly after boot (Eleven's OnBootSec=5min, compressed —
1274 // the edge is already up by the time this fires), then hourly + jitter.
1275 delay := time.Minute
1276 for {
1277 time.Sleep(delay)
1278 delay = time.Hour + rand.N(10*time.Minute)
1279 // DO NOT add release.NagExpiredCanaries(...) to this call. This loop
1280 // runs hourly, but it is NOT "the hourly root adopt pass" the option's
1281 // own doc comment names — that's RunAdopt (carlos-adopt.service)
1282 // below, root-privileged and the one place config sync also runs. Two
1283 // hourly callers both nagging would double the log line every hour;
1284 // worse, if this loop's cadence ever changes, a nag wired here changes
1285 // frequency with it silently. One caller, one nag.
1286 //
1287 // DO NOT add release.WithCapabilities(...) to this call either, and
1288 // for a harder reason than the nag's: that option WRITES
1289 // /etc/systemd/system, and this loop runs inside carlos-edge.service,
1290 // which is ProtectSystem=strict with ReadWritePaths=/data /run/carlos.
1291 // It could not converge a drop-in if it tried; all it would produce is
1292 // a permission-denied line every hour. The capability converge belongs
1293 // to RunAdopt (carlos-adopt.service, root) alone — see below.
1294 //
1295 // OnStageBlocked, by contrast, is exactly right here: it fires only
1296 // when this sandboxed process cannot write the version store, and all
1297 // it does is ask the root pass to do the part this one cannot.
1298 //
1299 // RestartHibernating, by contrast again, belongs HERE and not in
1300 // RunAdopt: this process owns the exec children, so it is the only one
1301 // whose restart callback can reach a warm hibernator at all. Without
1302 // it the pass would split on the Hibernate FLAG and hand every exec
1303 // instance the pure repoint — the flag says "may sleep", and a busy
1304 // instance would go on serving the binary its symlink no longer names
1305 // until it happened to go idle. Safe because makeRestartRoute cannot
1306 // start a process on either of its paths; see release.RestartHibernating.
1307 //
1308 // WithStartUnit rides wherever the named-unit restart arm can fire:
1309 // a failed named-unit restart that this pass rolled back must not
1310 // leave the service down (release.WithStartUnit). It asks for no
1311 // authority the restart above did not already need — both are one
1312 // systemctl verb on the same unit — and it cannot wake anything: it
1313 // fires only after a try-restart RETURNED AN ERROR, which a stopped
1314 // unit's exit-0 no-op never does.
1315 //
1316 // DO NOT add the platform self-update here either: the system leg
1317 // lives in the fast loop and re-observes hourly on its own, so an
1318 // hourly call here would only double the pointer read.
1319 //
1320 // agentUnitsOption hands named units to root when this process is
1321 // not root (issue #315), which leaves WithStartUnit and
1322 // WithRollbackRestart to an agent run as root.
1323 opts := append([]release.PassOption{
1324 release.BakeDir(dataDir()), release.OnStageBlocked(stageRequester(dataDir())),
1325 release.RestartHibernating(restart),
1326 release.WithStartUnit(makeStartRoute(reg, systemctlStartUnit)),
1327 release.WithRollbackRestart(agentRollbackUnitRestart(reg, systemdUnits{}, optDir, os.Getenv)),
1328 release.WithProofGate(agentProofGate()), release.WithHostWrites(writes),
1329 release.WithDigestRatchet(pointerRatchet(false, optDir, dataDir(), os.Getenv)),
1330 updatesPassOption(dataDir(), false)}, agentUnitsOption()...)
1331 if err := release.Pass(ctx, reg, store, pub, optDir, restart, opts...); err != nil {
1332 log.Printf("agent: adopt pass: %v", err)
1333 }
1334 // The schedule converge's hourly floor (schedule spec §6.2), AFTER the
1335 // release pass: an app whose schedules.json arrived with a new version
1336 // gets its rows against the version that is actually staged. Die-free
1337 // per app inside Pass — the joined error is one log line, never fatal.
1338 if err := sd.converge(ctx); err != nil {
1339 log.Printf("agent: schedule converge (hourly): %v", err)
1340 }
1341 }
1342}
1343
1344// configSyncLoop delivers app config to exec-backed routes: hourly (plus
1345// a first pass ~1min after boot, like adoptLoop), materialize every
1346// exec-syncable route's <dataDir>/<host>.config.env from the bucket and
1347// bounce running instances whose config changed. Runs IN-PROCESS on
1348// purpose — /data is inside this unit's sandbox and the exec children
1349// are this process's own, so it can both write the files and restart
1350// the readers; the root adopt path keeps the unit-backed half
1351// (/etc/carlos/apps needs root). Inert until CARLOS_SECRETS_KEY is set,
1352// the same convention as every other optional loop.
1353func configSyncLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
1354 raw := os.Getenv("CARLOS_SECRETS_KEY")
1355 if raw == "" {
1356 log.Printf("agent: config sync loop disabled: CARLOS_SECRETS_KEY not set")
1357 return
1358 }
1359 priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
1360 if err != nil {
1361 log.Printf("agent: config sync loop disabled: %v", err)
1362 return
1363 }
1364 ctx := context.Background()
1365 store, err := openDeployStore(ctx)
1366 if err != nil {
1367 log.Printf("agent: config sync loop disabled: %v", err)
1368 return
1369 }
1370 log.Printf("agent: config sync loop armed (hourly + jitter)")
1371 delay := time.Minute
1372 for {
1373 time.Sleep(delay)
1374 delay = time.Hour + rand.N(10*time.Minute)
1375 if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
1376 log.Printf("agent: config sync: %v", err)
1377 }
1378 }
1379}
1380
1381// configSyncMu serializes the two config loops' SyncExec passes against each
1382// other. They can genuinely overlap — hourly pass mid-flight when the 3s
1383// loop notices a fresh generation — and syncExecHost writes every route's
1384// file through a FIXED temp path (<host>.config.env.tmp, syncexec.go), so
1385// two interleaved passes could cross-rename each other's content: the loser
1386// lands STALE bytes after the winner's fresh ones, with the generation cache
1387// already advanced past the change, leaving nothing to re-fire until the
1388// next hourly pass. Serializing in-process closes that outright; the root
1389// adopt path is a different process writing a different namespace
1390// (/etc/carlos/apps) and is not this mutex's problem.
1391var configSyncMu sync.Mutex
1392
1393// syncExecSerialized is the one way either config loop runs SyncExec — the
1394// mutex lives in a shared wrapper, not at the call sites, so a future third
1395// caller cannot forget it.
1396func syncExecSerialized(ctx context.Context, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error {
1397 return syncExecSerializedFiltered(ctx, nil, reg, store, priv, dd, svc, writes...)
1398}
1399
1400// syncExecSerializedFiltered is syncExecSerialized over the routes keep
1401// accepts (nil: every route): the scoped pass's config leg (S3 change-index
1402// design §4.5). The filter is applied where SyncExec chooses its routes
1403// (appconfig.WithRouteFilter), so a route it leaves out costs no read.
1404func syncExecSerializedFiltered(ctx context.Context, keep func(registry.Route) bool, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes ...hostwrite.Config) error {
1405 configSyncMu.Lock()
1406 defer configSyncMu.Unlock()
1407 pending := map[string]string{}
1408 outcomes := map[string]bool{}
1409 opts := []appconfig.SyncOption{
1410 appconfig.SkipFenced(func(host string) bool { return layout.JournalPresent(dd, host) }),
1411 appconfig.WhenInstanceDirAbsent(func(host, dir string) { pending[host] = dir }),
1412 appconfig.WithHostOutcome(func(host string, done bool) { outcomes[host] = done }),
1413 }
1414 if keep != nil {
1415 opts = append(opts, appconfig.WithRouteFilter(keep))
1416 }
1417 for _, cfg := range writes {
1418 opts = append(opts, appconfig.WithHostWrite(cfg))
1419 }
1420 // CARLOS_SWAP=overlap: the scheduler restarts, paced, and acknowledges a
1421 // revision once it is active; refusing: no restart. configSyncMu above
1422 // is what makes ConfigDelivered serial per host.
1423 var err error
1424 if cb, ok := rolloutConfigCallbacks(ctx, store, configLegs{awake: managedAwake(svc, reg), restart: bounce(svc)}); ok {
1425 err = appconfig.SyncExecWithCallbacks(ctx, reg, store, priv, dd, os.Getenv("PLATFORM_APPS_DOMAIN"), cb, opts...)
1426 } else {
1427 err = appconfig.SyncExec(ctx, reg, store, priv, dd, bounce(svc), os.Getenv("PLATFORM_APPS_DOMAIN"), configAckFunc(ctx, store, "agent"), opts...)
1428 }
1429 // A filtered pass names only the routes it kept, so it is never
1430 // complete: a host it left out keeps its wait (pendingDirs.record).
1431 configDirsPending.record(pending, outcomes, err == nil && keep == nil)
1432 return err
1433}
1434
1435// configDirsPending is what the last config pass could not write because a
1436// nested row's instance directory did not exist yet (harness Run 9, finding
1437// 6): host -> the directory. The activator makes that directory at the
1438// first wake, and before this the only pass that came back for the file was
1439// configSyncLoop's hourly one, so a nested instance created with settings
1440// woke and ran without them, its store binding included, for up to an hour.
1441// configGenLoop now looks at these directories on every tick and runs the
1442// pass again as soon as one is there.
1443var configDirsPending pendingDirs
1444
1445type pendingDirs struct {
1446 mu sync.Mutex
1447 dirs map[string]*pendingDir
1448}
1449
1450// pendingDir is one host the config pass waits for: its instance directory,
1451// and, once the directory is there, how many passes since have left the host
1452// undone and how many ticks to let go by before the next retry.
1453type pendingDir struct {
1454 dir string
1455 failures int
1456 waitTicks int
1457}
1458
1459// pendingRetryCap bounds the retry backoff: at most 16 generation-loop ticks
1460// (about a minute) between passes for a host whose directory is there but
1461// whose config a pass keeps leaving undone (a bundle that will not read).
1462const pendingRetryCap = 16
1463
1464// record keeps what one pass found: absent, the hosts whose instance
1465// directory was not there; outcomes, every in-scope host the pass named,
1466// done or not (appconfig.WithHostOutcome). A host stays waited for until a
1467// pass reports it done (fix round 1, Astra's finding 3: a pass that left it
1468// undone on a read failure still returned nil), or until a complete pass no
1469// longer has it in scope (its row is gone or no longer takes a bundle). A
1470// pass that failed part way names fewer hosts than it might have, so it only
1471// adds. A host a pass leaves undone with its directory there backs off,
1472// doubling from the next tick up to pendingRetryCap ticks.
1473func (p *pendingDirs) record(absent map[string]string, outcomes map[string]bool, complete bool) {
1474 p.mu.Lock()
1475 defer p.mu.Unlock()
1476 if p.dirs == nil {
1477 p.dirs = map[string]*pendingDir{}
1478 }
1479 for host, e := range p.dirs {
1480 if _, again := absent[host]; again {
1481 continue
1482 }
1483 done, named := outcomes[host]
1484 switch {
1485 case named && done, !named && complete:
1486 delete(p.dirs, host)
1487 case named:
1488 // The count saturates once the delay reaches the cap, so the
1489 // shift below never overflows however long a host keeps failing.
1490 if 1<<e.failures <= pendingRetryCap {
1491 e.failures++
1492 }
1493 e.waitTicks = min(1<<(e.failures-1), pendingRetryCap) - 1
1494 }
1495 }
1496 for host, dir := range absent {
1497 if e, ok := p.dirs[host]; ok {
1498 e.dir = dir
1499 continue
1500 }
1501 p.dirs[host] = &pendingDir{dir: dir}
1502 }
1503}
1504
1505// appeared names a host whose instance directory a pass waited for, which
1506// exists now, and whose retry is due; ok is false when there is none. Each
1507// call is one tick of the backoff. Lstat only: the directory is the
1508// activator's, and nothing here follows or opens it.
1509func (p *pendingDirs) appeared() (host string, ok bool) {
1510 p.mu.Lock()
1511 defer p.mu.Unlock()
1512 for h, e := range p.dirs {
1513 if _, err := os.Lstat(e.dir); err != nil {
1514 continue
1515 }
1516 if e.waitTicks > 0 {
1517 e.waitTicks--
1518 continue
1519 }
1520 if !ok {
1521 host, ok = h, true
1522 }
1523 }
1524 return host, ok
1525}
1526
1527// configAckFunc builds the callback config sync's two passes (the root
1528// pass's appconfig.Sync and this agent's own appconfig.SyncExec, called
1529// through syncExecSerialized above) invoke for a route they just converged
1530// (relay spec §5, internal/appconfig/configack.go): it stamps this box's
1531// label onto the ack — BoxLabel (metrics_loop.go) is the same identity
1532// every other per-box artifact this platform writes uses, so a config-ack
1533// and this box's heartbeat/metrics rows agree about which box they belong
1534// to — and writes it. A write failure is logged, never returned: an ack is
1535// best-effort telemetry for `carlos relay status`, not something config
1536// delivery itself depends on, and failing the whole sync pass over it would
1537// make a bucket blip on the ack object cost every OTHER route its config.
1538func configAckFunc(ctx context.Context, store release.Store, logPrefix string) func(appconfig.ConfigAck) {
1539 box := BoxLabel()
1540 return func(a appconfig.ConfigAck) {
1541 a.Box = box
1542 if err := appconfig.WriteConfigAck(ctx, store, a); err != nil {
1543 log.Printf("%s: config ack %s: %v", logPrefix, a.Host, err)
1544 }
1545 }
1546}
1547
1548// configAckWriter is configAckFunc for the armed scheduler's ack writer
1549// (rollout.Manager.DeferAck): the same stamp and write, under the context
1550// the writer hands each write, which bounds it (Task 11, Astra R1 Flag
1551// triage T8). configAckFunc itself stays as it is, unbounded, for flag-off
1552// sync and the root pass (plan review #11).
1553func configAckWriter(store release.Store, logPrefix string) func(context.Context, appconfig.ConfigAck) {
1554 box := BoxLabel()
1555 return func(ctx context.Context, a appconfig.ConfigAck) {
1556 a.Box = box
1557 if err := appconfig.WriteConfigAck(ctx, store, a); err != nil {
1558 log.Printf("%s: config ack %s: %v", logPrefix, a.Host, err)
1559 }
1560 }
1561}
1562
1563// reconcileMu serializes the two provisioning loops' Reconcile passes
1564// against each other, the same shape as configSyncMu above. They can
1565// genuinely overlap — instancesLoop's 5-minute pass mid-flight when a
1566// serial bump fires fastConvergeLoop's full pass set — and Reconcile's
1567// removal phase deletes any Provisioned, in-scope route that is absent from
1568// ITS OWN record snapshot, which it lists at the top of the pass
1569// (provision/reconcile.go): a route the newer pass minted between the older
1570// pass's apps/ listing and its registry listing looks exactly like a
1571// retired record to the older pass, which would delete it. The re-mint
1572// minutes later comes back with its capability grants revoked
1573// (registry.Delete purges grants — that is Delete's contract, not a bug).
1574// Serializing in-process closes the race outright; the root adopt path runs
1575// no Reconcile and is not this mutex's problem.
1576var reconcileMu sync.Mutex
1577
1578// reconcileSerialized is the one way any agent loop runs the record-family
1579// reconcile passes — the mutex lives in a shared wrapper, not at the call
1580// sites, so a future third caller cannot forget it. Pinned by
1581// TestReconcileSerializedHoldsTheMutex.
1582//
1583// BOTH families run here, under ONE hold of the mutex: the instance pass and
1584// the sidecar pass. Each pass's removal phase is scoped to its own route Kind
1585// (provision.ReconcileSidecars' doc comment), so they cannot delete each
1586// other's rows — but they read and write the same registry, and the
1587// overlapping-pass hazard the mutex exists for (a route minted by a newer
1588// pass looking exactly like a retired record to an older one) applies within
1589// each family just as much. Running them as one serialized unit also means a
1590// caller cannot converge instances while leaving sidecars a tick behind.
1591//
1592// Errors are JOINED rather than short-circuited: a sidecar-side failure must
1593// not hide an instance-side one, and neither must stop the other running —
1594// each pass is independently safe to run and independently valuable.
1595func reconcileSerialized(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig) error {
1596 return reconcileSerializedWith(ctx, s, reg, cfg, armedMovesWorker.Load())
1597}
1598
1599// reconcileSerializedWith is reconcileSerialized with the moves worker whose
1600// offline hooks the pass takes (movesoffline.go): the armed worker in the
1601// agent, a box's own worker in a test. A nil worker, or one on a box that
1602// cannot take part in an offline move, adds nothing.
1603func reconcileSerializedWith(ctx context.Context, s release.Store, reg *registry.Registry, cfg provision.ReconcileConfig, w *movesWorker) error {
1604 reconcileMu.Lock()
1605 defer reconcileMu.Unlock()
1606 // With the change index armed, the record walks below read the record
1607 // view's snapshot, refreshed here under reconcileMu so a pass's snapshot
1608 // is never older than the previous pass's (S3 change-index design §4.3).
1609 // Writes still go to s, and a key written through the snapshot store is
1610 // read from s afterwards. Disarmed, or when the view cannot vouch, rs is
1611 // s and the pass walks the bucket as before.
1612 rs, _, _ := viewStore(ctx, s)
1613 // A fresh row is minted at the layout this box's intent names
1614 // (ReconcileConfig.Layout): nested on a nested box, so a new tenant
1615 // never runs flat there. Set here, the one place both callers pass
1616 // through, because the intent is read from the store this pass has in
1617 // hand and scoped to the label this box publishes under.
1618 if cfg.Layout == nil {
1619 cfg.Layout = func(ctx context.Context, s release.Store) (string, error) {
1620 return mintLayoutIntentWith(ctx, s, reg.List)
1621 }
1622 }
1623 // Seeded records (seeded-instances spec §3.2): acted on only once the
1624 // activator has armed the data bucket (wireSeedPlacement).
1625 if cfg.Seeds == nil {
1626 cfg.Seeds = armedSeedHooks.Load()
1627 }
1628 // Move holds (moves spec §9.6): participant rows minted held and kept.
1629 // Unchanged with the Phase 1 switch off.
1630 cfg = withMovesHolds(cfg)
1631 // Offline move holds: the hold from the record alone, the keep from the
1632 // view, which is read here first if no listing has been read yet or the
1633 // last one failed, so keep never fails closed on a box whose listing
1634 // reads. That listing reads the bucket, never the record view: what a
1635 // move keeps is decided on live data (S3 change-index final review R1,
1636 // finding 1). Nothing on a box that cannot take part, and nothing where
1637 // Phase 1's hooks are set.
1638 if w != nil && w.cfg.OfflineCapable && w.cfg.Offline != nil {
1639 if !w.cfg.Offline.built() || w.cfg.Offline.failed() {
1640 refreshOfflineView(ctx, s, w.cfg.Offline)
1641 }
1642 cfg = withOfflineHolds(cfg, offlineHold(w.cfg), w.cfg.Offline)
1643 }
1644 // Runs ride this pass's record reads (run spec §5.2): the worker notes
1645 // every in-scope record with a run, and acts after the pass.
1646 cfg = withRunsHook(cfg)
1647 return errors.Join(
1648 provision.Reconcile(ctx, rs, reg, cfg),
1649 provision.ReconcileSidecars(ctx, rs, reg, cfg),
1650 )
1651}
1652
1653// bounce builds the restart callback SyncExec calls for a route whose
1654// rendered config actually changed. Shared by BOTH config loops (hourly
1655// configSyncLoop and the fast configGenLoop) rather than duplicated per
1656// loop, unlike the adopt pair: those two differ from each other (filter,
1657// nag) where these two must not — the delivery must be identical whichever
1658// loop happened to notice a change first, and a divergence would mean an
1659// app converges one way at 3s and another way on the hour. A nil svc (no
1660// activator armed) still returns a working no-op: the files converge for
1661// future spawns, there is simply nothing running to restart.
1662func bounce(svc *activator.Service) func(string) error {
1663 return func(host string) error {
1664 if svc == nil {
1665 return nil
1666 }
1667 if svc.Bounce(host) {
1668 log.Printf("agent: config changed, bounced %s", host)
1669 }
1670 return nil
1671 }
1672}
1673
1674// genAbsent marks an app whose generation object was READ and found
1675// missing, distinguishing it from an app never seen at all — the two are
1676// different in the only way this loop cares about: a never-seen app must
1677// record silently (first sight), while a known-absent app that suddenly
1678// has a generation object is an app whose config was just written for the
1679// first time, which must sync. A real etag can never collide with it: no
1680// store produces one containing a NUL byte (S3 etags are hex/quoted,
1681// DirStore's is hex).
1682const genAbsent = "\x00absent"
1683
1684// genCache is everything configGenLoop remembers between ticks: one
1685// generation etag per config-carrying app, and the previous tick's
1686// route-shape digests. One object rather than two parameters because the
1687// two are remembered for exactly one reason — deciding what "changed"
1688// means — and a caller holding one without the other would silently lose
1689// half the loop's sensitivity.
1690type genCache struct {
1691 etags map[[2]string]string
1692 // exec/unit are appconfig.ConfigRouteFingerprints' two digests as of
1693 // the last pass that computed them; shapeSeen is false until the first
1694 // such pass. The bool is not redundant with an empty string: a box with
1695 // no config-carrying routes at all has a real, stable digest of the
1696 // empty set, and treating that as "never seen" would make its first
1697 // route look like a change on the tick after boot rather than on the
1698 // tick it appeared.
1699 exec, unit string
1700 shapeSeen bool
1701
1702 // gate skips the read of an app whose scope has not moved since its last
1703 // successful read, younger than the backstop (S3 change-index design §5).
1704 // nil, or a gate over a nil or unusable watch, reads every app every
1705 // tick, as before the index. now is the gate's clock (nil: time.Now).
1706 gate *release.ReadGate
1707 now func() time.Time
1708}
1709
1710func newGenCache() *genCache { return &genCache{etags: map[[2]string]string{}} }
1711
1712// genPollPass is one tick of configGenLoop, split out for tests: read the
1713// generation etag of every app the box can converge — exec-config apps and
1714// unit-backed apps alike — plus the registry's own route shape, and report
1715// which HALVES moved since the cache was built.
1716//
1717// Semantics, each of which is a deliberate choice:
1718//
1719// - First sight of an app (no cache entry) records silently and does NOT
1720// sync. Boot must not re-bounce the world — every app would look
1721// "changed" on the first tick after a restart. configSyncLoop's ~1min
1722// first pass owns boot convergence.
1723// - An unchanged etag does nothing.
1724// - Absent that stays absent does nothing, forever: an app last written
1725// by a pre-generation CLI has no generation object and never will
1726// until someone writes its config again. That is legal, not a fault.
1727// - Any movement syncs: the etag changed, or absent -> present.
1728// present -> absent shouldn't happen (generation objects are only ever
1729// created and overwritten, never deleted), but it's treated as a change
1730// too — one extra SyncExec pass is cheap, and pretending we didn't see
1731// it would be a lie about the only thing this function reports.
1732// - ONE bool PER HALF for the whole pass, however many apps moved: the
1733// caller runs one SyncExec (which resolves every route itself) and asks
1734// for at most one root config pass (which sweeps every unit route
1735// itself). Per-app syncs would multiply bucket reads to deliver exactly
1736// the same files, and per-app root requests would ask the same pass to
1737// run several times over.
1738// - The halves are independent: an app whose routes are all exec-backed
1739// sets execChanged alone, an all-unit app sets unitChanged alone, and an
1740// app holding both shapes at once sets both — off the SAME single read,
1741// because its whole config lives in one generation object however its
1742// routes are backed.
1743// - Apps that vanish from the route set lose their cache entries, so a
1744// re-added app is a first sight again rather than a comparison against
1745// an etag remembered from before it left (a route being re-added gets
1746// fresh config from the spawn path anyway).
1747// - The ROUTE SHAPE is watched too, not just the bucket. A route's
1748// Environment binding lives in the box's own registry and moves no etag
1749// anywhere, so without this the loop cannot see a rebinding at all and
1750// delivery waits out configSyncLoop's hourly floor: issue #196, where a
1751// live route served the wrong merge for ten minutes and the only symptom
1752// was the app reporting a credential missing — which reads as a BAD
1753// credential rather than an undelivered one. First sight of the shape is
1754// silent for the same reason first sight of an app is. This covers BOTH
1755// doors onto the column: `carlos route -environment` on the box, and the
1756// provisioning reconciler converging Environment from an instance record
1757// — which is the sanctioned door for a Provisioned row, and which the
1758// issue's own suggested fix (bump the generation from `carlos route`)
1759// would have missed entirely. It costs no bucket read: see
1760// appconfig.ConfigRouteFingerprints.
1761// - A read error that isn't "not found" leaves that app's cache entry
1762// exactly as it was and keeps evaluating the rest: the errors come back
1763// for the caller to aggregate into ONE log line per pass, never one per
1764// app per tick. Clearing the entry instead would turn a recovered blip
1765// into a bogus first sight or a bogus change.
1766// - With the change index armed (cache.gate), an app already in the cache
1767// is read only when its scope moved, its last read failed, or that read
1768// is the backstop's age (S3 change-index design §5). A skipped app keeps
1769// its entry and reports no movement.
1770//
1771// The cache is mutated in place — it lives across ticks in the loop, which
1772// is the whole point (there is nowhere else to remember what "changed"
1773// means).
1774func genPollPass(ctx context.Context, routes []registry.Route, store release.Store, cache *genCache) (execChanged, unitChanged bool, readErrs []error) {
1775 // The registry-side half (issue #196). Free — no bucket read at all —
1776 // and evaluated first so a rebinding converges on the same tick a var
1777 // change would, rather than one behind it.
1778 execFP, unitFP := appconfig.ConfigRouteFingerprints(routes)
1779 if cache.shapeSeen {
1780 execChanged = execChanged || execFP != cache.exec
1781 unitChanged = unitChanged || unitFP != cache.unit
1782 }
1783 cache.exec, cache.unit, cache.shapeSeen = execFP, unitFP, true
1784
1785 execApps := appconfig.ExecConfigApps(routes)
1786 unitApps := appconfig.UnitConfigApps(routes)
1787 // The poll set is the UNION, read once per app: the two sets overlap
1788 // whenever an app has both route shapes, and its config is one object
1789 // either way. inExec/inUnit then say which half a movement belongs to.
1790 inExec := make(map[[2]string]bool, len(execApps))
1791 for _, k := range execApps {
1792 inExec[k] = true
1793 }
1794 inUnit := make(map[[2]string]bool, len(unitApps))
1795 for _, k := range unitApps {
1796 inUnit[k] = true
1797 }
1798 apps := make([][2]string, 0, len(execApps)+len(unitApps))
1799 apps = append(apps, execApps...)
1800 for _, k := range unitApps {
1801 if !inExec[k] {
1802 apps = append(apps, k)
1803 }
1804 }
1805 now := time.Now
1806 if cache.now != nil {
1807 now = cache.now
1808 }
1809 at := now()
1810 live := make(map[[2]string]bool, len(apps))
1811 for _, k := range apps {
1812 live[k] = true
1813 key := appconfig.GenKey(k[0], k[1])
1814 // The gate (spec §5): an app already in the cache is read again only
1815 // when its scope moved, its last read failed, or that read is the
1816 // backstop's age. An app not in the cache is always read: its first
1817 // sight is what a later change is compared against. The generation
1818 // is taken before the read, so a write landing during it stays due.
1819 scope, gated := release.ChangeScope(key)
1820 gen, due := cache.gate.Due(scope, at)
1821 if _, seen := cache.etags[k]; seen && gated && !due {
1822 continue
1823 }
1824 _, etag, err := store.Get(ctx, key)
1825 if gated {
1826 cache.gate.Mark(scope, gen, at, err == nil || errors.Is(err, os.ErrNotExist))
1827 }
1828 switch {
1829 case errors.Is(err, os.ErrNotExist):
1830 etag = genAbsent
1831 case err != nil:
1832 readErrs = append(readErrs, fmt.Errorf("%s/%s: %w", k[0], k[1], err))
1833 continue // cache entry untouched: this app is unknown this tick, not changed
1834 }
1835 prev, seen := cache.etags[k]
1836 cache.etags[k] = etag
1837 if seen && prev != etag {
1838 execChanged = execChanged || inExec[k]
1839 unitChanged = unitChanged || inUnit[k]
1840 }
1841 }
1842 for k := range cache.etags {
1843 if !live[k] {
1844 delete(cache.etags, k)
1845 }
1846 }
1847 return execChanged, unitChanged, readErrs
1848}
1849
1850// configSyncMarker is the file configGenLoop touches to ask a root config
1851// pass to run now: carlos-adopt.path watches it (PathChanged) and starts
1852// carlos-config-sync.service (`carlos adopt -config-only`), which owns the
1853// unit half of config delivery (/etc/carlos/apps + try-restart) that this
1854// unprivileged process cannot — and only that half.
1855// It lives in the data dir, deliberately outside the *.config.env namespace
1856// reapConfigEnv sweeps — the reaper matches that suffix alone and must go
1857// on ignoring this.
1858const configSyncMarker = "config-sync-requested"
1859
1860// requestRootSync touches the marker. The content is a timestamp only so an
1861// operator reading /data can see when the last request went out; systemd
1862// watches the path, not the bytes. Errors are the caller's to log and
1863// swallow: a request that cannot be written costs latency, not correctness,
1864// because carlos-adopt.timer's hourly pass is the floor underneath it.
1865func requestRootSync(dd string) error {
1866 return os.WriteFile(filepath.Join(dd, configSyncMarker),
1867 []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644)
1868}
1869
1870// stageRequestMarker is the release half of configSyncMarker. The agent's
1871// adopt loops run inside carlos-edge.service (ProtectSystem=strict, no /opt
1872// write), so they can SEE a promotion within seconds and stage nothing; the
1873// only process that can write the version store is the root pass on
1874// carlos-adopt.timer, an hour away. Touching this file relays "there is a
1875// version here I cannot stage" to carlos-stage.path, exactly as the config
1876// half relays to carlos-adopt.path. Lives in the data dir beside its config
1877// twin, outside the *.config.env namespace reapConfigEnv sweeps.
1878const stageRequestMarker = "stage-requested"
1879
1880// stageRequestEvery throttles those requests. Unlike the config marker —
1881// touched only when a generation etag MOVES, which is naturally rare — a
1882// blocked stage repeats on every tick of a loop that runs every 2-3 seconds,
1883// so an untouched throttle would ask systemd to start a root pass ~30 times a
1884// minute and each pass would re-block behind the one before it.
1885//
1886// Fifteen seconds, and a request inside the window is DEFERRED to its end,
1887// never dropped (stageThrottle). It was a minute and dropped, and a second
1888// release inside the minute then waited for whatever next moved the serial:
1889// 98 extra seconds on the flagship on 2026-10-06 (#231). A root pass takes
1890// about eight seconds, and the window has to outlast it: carlos-stage.path
1891// watches with PathChanged, and systemd does not act on a change made while
1892// carlos-stage.service is still running.
1893const stageRequestEvery = 15 * time.Second
1894
1895// stageRequester returns a release.OnStageBlocked callback that touches the
1896// marker at most once per stageRequestEvery. The closure holds the clock, so
1897// each loop gets its own budget: the fast loop asking does not silence the
1898// hourly one, and neither can starve the other.
1899//
1900// Failures are logged and swallowed, like requestRootSync's: a request that
1901// cannot be written costs latency, never correctness, because
1902// carlos-adopt.timer remains the floor underneath it.
1903func stageRequester(dd string) func(host, ver string, err error) {
1904 ask := stageMarkerRequester(dd)
1905 return func(host, ver string, err error) {
1906 ask(host, ver, err)
1907 // The stage relay (stagerelay.go) learns the head here and not in
1908 // markerRequester, which mintRequester shares: a proof-mint request
1909 // is not a head that failed to stage.
1910 if r := armedStageRelay.Load(); r != nil {
1911 r.note(host, ver)
1912 }
1913 }
1914}
1915
1916// stageMarkerRequester is stageRequester without the relay: the same marker,
1917// throttle and log lines, and nothing remembered. The moves worker uses it
1918// (movespass.go): its version is a move's held version, not a channel head,
1919// and a held stage neither moves the installed Version nor leaves a refusal,
1920// so the relay would have no way to know the request was answered.
1921func stageMarkerRequester(dd string) func(host, ver string, err error) {
1922 return markerRequester(dd, func(host, ver string, err error) {
1923 if err != nil {
1924 log.Printf("agent: cannot request root stage for %s %s: %v", host, ver, err)
1925 return
1926 }
1927 // One line for every cause the marker relays (Task 12a): a build
1928 // this process cannot stage, an install request only root
1929 // resolves (ErrInstallRequestNeedsRoot), a named unit only root
1930 // restarts (ErrUnitRestartNeedsRoot), and a held wake's target only
1931 // root installs. The throttle does not carry the cause.
1932 log.Printf("agent: %s needs a root pass for %s (a build only root can stage or install, a unit only root can restart, or an install request only root resolves); asked for one", host, ver)
1933 })
1934}
1935
1936// mintRequester is the mint asker's (proofask.go) request: the same marker
1937// and the same throttle shape as stageRequester's, with its own budget and
1938// its own words. The asker logs why it asks; this logs only a failed write.
1939func mintRequester(dd string) func(host, ver string, err error) {
1940 return markerRequester(dd, func(host, ver string, err error) {
1941 if err != nil {
1942 log.Printf("agent: cannot request a root pass to mint a proof for %s %s: %v", host, ver, err)
1943 }
1944 })
1945}
1946
1947// markerRequester touches the stage-request marker through a fresh
1948// stageThrottle and hands said each write's outcome.
1949func markerRequester(dd string, said func(host, ver string, err error)) func(host, ver string, err error) {
1950 t := &stageThrottle{every: stageRequestEvery, now: time.Now, after: func(d time.Duration, f func()) { time.AfterFunc(d, f) },
1951 write: func(host, ver string) {
1952 said(host, ver, os.WriteFile(filepath.Join(dd, stageRequestMarker),
1953 []byte(time.Now().UTC().Format(time.RFC3339)+"\n"), 0o644))
1954 }}
1955 return func(host, ver string, _ error) { t.request(host, ver) }
1956}
1957
1958// stageThrottle writes at most once per every, measured from the end of one
1959// write to the start of the next. While a write is scheduled (busy), every
1960// request rides it: a deferred write owns the window however late its timer
1961// fires. A request inside the window schedules one write for the window's
1962// end, so none waits longer than every after the write before it.
1963type stageThrottle struct {
1964 every time.Duration
1965 now func() time.Time
1966 after func(time.Duration, func())
1967 write func(host, ver string)
1968 mu sync.Mutex
1969 last time.Time // when the last write finished
1970 busy bool
1971}
1972
1973func (t *stageThrottle) request(host, ver string) {
1974 t.mu.Lock()
1975 if t.busy {
1976 t.mu.Unlock()
1977 return
1978 }
1979 t.busy = true
1980 // The write runs under the lock: a request made once the marker has
1981 // changed may concern a pointer root has already read, so it waits for
1982 // the write to finish and schedules the next window, rather than
1983 // riding a write that cannot answer it.
1984 do := func() {
1985 t.mu.Lock()
1986 t.write(host, ver)
1987 t.busy, t.last = false, t.now()
1988 t.mu.Unlock()
1989 }
1990 if since := t.now().Sub(t.last); !t.last.IsZero() && since < t.every {
1991 t.mu.Unlock()
1992 t.after(t.every-since, do)
1993 return
1994 }
1995 t.mu.Unlock()
1996 do()
1997}
1998
1999// configGenLoop is the fast-converge shape applied to issue #44: a config
2000// set should converge like a canary adopts — seconds, not the next hour.
2001// Every ~3s it reads one small object per config-carrying app (the
2002// generation object every config write touches) and, on movement, delivers
2003// the halves it can: ONE SyncExec pass for the whole tick for exec-backed
2004// instances, and one touch of the root-sync marker for unit-backed ones.
2005// Spurious movement costs a pass that finds nothing to write: SyncExec (and
2006// the root pass's appconfig.Sync) bounces only routes whose rendered file
2007// actually changed, so a redundant touch can never cost a restart.
2008//
2009// Gated on CARLOS_SECRETS_KEY exactly as configSyncLoop is, for a reason
2010// beyond convention: without the key SyncExec cannot render an app's
2011// environment at all, so this loop would poll the bucket every 3 seconds
2012// forever to signal no one. openDeployStore failing is the same story with
2013// nothing to read. The gate covers the unit half too, which is an accepted
2014// asymmetry rather than a considered one: a box whose agent lacks the key
2015// gets no fast path for its unit apps either, even though the root pass
2016// that would run holds its own key. In practice both units read the same
2017// /etc/carlos/host.env, so the case is theoretical — and the hourly timer
2018// floor stands regardless.
2019//
2020// Cardinality: unlike fastConvergeLoop's sweep, which is bounded to the
2021// deliberately rare canary/* routes plus still-unstaged ones, this watches
2022// EVERY config-carrying app on the box. That is accepted knowingly — one GET
2023// of the smallest object in the bucket
2024// per app per 3s, against the store fastConvergeLoop already hits every 2s. What
2025// keeps it bounded is that the pass reads generation objects only: no
2026// manifests, no pointers, no per-environment resolution. All of that
2027// happens inside the SyncExec a real change triggers.
2028//
2029// Deliberately NOT done here:
2030//
2031// - No fast first pass. The loop sleeps BEFORE its first tick and the
2032// first sight of every app is silent (see genPollPass) — boot
2033// convergence belongs to configSyncLoop's ~1min first pass, and a box
2034// coming up must not bounce every instance it has because it has never
2035// seen their etags before.
2036// - No unit-backed rendering or restarting. This process still writes
2037// nothing under /etc and restarts no unit (see RunAdopt): a unit app's
2038// movement only touches <dataDir>/config-sync-requested, which
2039// carlos-adopt.path relays into a root carlos-config-sync.service run
2040// (`carlos adopt -config-only`) that does both. The relay reaches ONLY
2041// the config half — never release adoption, never the canary-TTL nag,
2042// which would otherwise fire off an unrelated app's config write
2043// instead of hourly. The relay can drop an event that lands while that
2044// service is mid-run — accepted: the next config write re-touches the
2045// marker, and the timer floor delivers within the hour regardless.
2046// - No hourly floor replaced. configSyncLoop stays exactly as it was: it
2047// catches apps written by older CLIs (no generation object at all),
2048// ticks this loop missed, and the boot-retry gap of issue #27.
2049//
2050// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
2051// whole decision lives in genPollPass, which is tested.
2052func configGenLoop(reg *registry.Registry, svc *activator.Service, dd string, writes hostwrite.Config) {
2053 raw := os.Getenv("CARLOS_SECRETS_KEY")
2054 if raw == "" {
2055 log.Printf("agent: config generation loop disabled: CARLOS_SECRETS_KEY not set")
2056 return
2057 }
2058 priv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
2059 if err != nil {
2060 log.Printf("agent: config generation loop disabled: %v", err)
2061 return
2062 }
2063 ctx := context.Background()
2064 store, err := openDeployStore(ctx)
2065 if err != nil {
2066 log.Printf("agent: config generation loop disabled: %v", err)
2067 return
2068 }
2069 // One cache for the life of the process: what "changed" means is
2070 // exactly "differs from the last tick that could read it". Its gate is
2071 // built here, after RunAgent armed the change index.
2072 cache := newGenCache()
2073 cache.gate = newChangeGate()
2074 log.Printf("agent: config generation loop armed (3s + jitter, exec-config and unit-backed apps)")
2075 for {
2076 time.Sleep(3*time.Second + rand.N(time.Second))
2077 routes, err := reg.List()
2078 if err != nil {
2079 // The registry is a local file this process already holds open;
2080 // a failing List is a broken box, not a passing blip, so it is
2081 // worth a line even at this cadence (the same call is logged per
2082 // tick by statusLoop).
2083 log.Printf("agent: config generation poll: list routes: %v", err)
2084 continue
2085 }
2086 configGenTick(ctx, routes, reg, store, priv, dd, svc, writes, cache)
2087 }
2088}
2089
2090// configGenTick is one tick of configGenLoop over the routes it listed, split
2091// out so a test can run the loop's own decision.
2092func configGenTick(ctx context.Context, routes []registry.Route, reg *registry.Registry, store release.Store, priv *ecdh.PrivateKey, dd string, svc *activator.Service, writes hostwrite.Config, cache *genCache) {
2093 execChanged, unitChanged, readErrs := genPollPass(ctx, routes, store, cache)
2094 if len(readErrs) > 0 {
2095 // ONE line per pass, never one per app: a bucket having a bad
2096 // minute would otherwise flood the journal at 20 lines a minute
2097 // per app. The count carries the scale, the first error carries
2098 // the cause; they are the same cause in practice.
2099 log.Printf("agent: config generation poll: %d apps unreadable: %v", len(readErrs), readErrs[0])
2100 }
2101 if unitChanged {
2102 // Requested first and never fatal: the write is instant while
2103 // syncExecSerialized can block on a serialized pass, so the root
2104 // run gets to start alongside it rather than behind it. A failed
2105 // marker write costs those apps the wait until
2106 // carlos-adopt.timer's next hourly pass — never correctness.
2107 if err := requestRootSync(dd); err != nil {
2108 log.Printf("agent: request root config sync: %v", err)
2109 } else {
2110 log.Printf("agent: config changed for unit-backed app(s), requested root config sync pass")
2111 }
2112 }
2113 // A nested row an earlier pass could not write for because its instance
2114 // directory was not there yet: the first wake has made it, so its
2115 // bundle goes in now (finding 6). The pass bounces the instance onto
2116 // its settings once the wake it is in has finished (Service.Bounce).
2117 if host, ok := configDirsPending.appeared(); ok && !execChanged {
2118 log.Printf("agent: %s: its instance directory is there now, writing its config", host)
2119 execChanged = true
2120 }
2121 if !execChanged {
2122 return
2123 }
2124 if err := syncExecSerialized(ctx, reg, store, priv, dd, svc, writes); err != nil {
2125 log.Printf("agent: config generation sync: %v", err)
2126 }
2127}
2128
2129// Fast-converge cadence and clamps (carlos-deploy spec §3-§4). The tick is
2130// the serial poll's latency promise (~2s from console bump to pass); the
2131// clamp bounds what a burst of console writes can cost (one full pass set
2132// per 5s, with the last bump deferred, never dropped); the backoff bounds
2133// what a permanently-unstageable route can cost (one look per 5min at the
2134// cap, instead of one per tick forever).
2135const (
2136 fastTickEvery = 2 * time.Second
2137 fastFullClamp = 5 * time.Second
2138 fastBackoffBase = 2 * time.Second
2139 fastBackoffCap = 5 * time.Minute
2140
2141 // The restart-failure hold (noteRestartFailed) starts higher and caps
2142 // higher than the staging backoff above, because what it bounds is not
2143 // a retry's COST to this box but a service CYCLE on a tenant's process:
2144 // re-attempting a version whose restart just failed repoints, performs a
2145 // REAL restart of the running old service, fails, rolls back and starts
2146 // it again — retrying that 2 seconds later is near-certain to do it all
2147 // again. 30s → 15m doubling means a persistently broken head costs a
2148 // handful of honest cycles in its first half hour and then four an hour
2149 // from this loop, with the hourly floors' one-each on top — instead of
2150 // one every other second for as long as the broken head stands.
2151 fastRestartBackoffBase = 30 * time.Second
2152 fastRestartBackoffCap = 15 * time.Minute
2153)
2154
2155// fastDeps is one fastConvergeLoop's state and seams — a struct so the tick
2156// is testable without goroutines, env vars, or sleeps.
2157type fastDeps struct {
2158 reg *registry.Registry
2159 store release.Store
2160 pub ed25519.PublicKey
2161 optDir string
2162 restart func(string) error
2163 passOpts []release.PassOption // fastPassOpts — every sweep pass carries all of them
2164 fullPass func(ctx context.Context) // reconcile → adopt → config sync, or a test hook
2165 restartLeg func(ctx context.Context) // the `carlos restart` check (issue #113), or a test hook
2166 wakeLeg func(ctx context.Context) // the sidecar wake check (design: sidecar members), or a test hook
2167 scheduleLeg func(ctx context.Context) // the schedule converge (spec §6.2), or a test hook; nil = not armed
2168 layoutLeg func(ctx context.Context) // the layout pass (isolated-exec design §8.2), or a test hook; nil = not armed
2169 systemLeg *systemLeg // the platform self-update leg (spec §4); nil = not armed
2170 proof *release.ProofGate // D3's gate (agentProofGate); nil = no proof waits to keep
2171 mintLeg func(ctx context.Context) // D3's mint asker (proofask.go); nil = not armed
2172 lastSerial int64
2173 seeded bool
2174 lastFull time.Time
2175 notBefore map[string]time.Time
2176 step map[string]time.Duration
2177 restartHold map[string]*restartFailure // hosts held out of the sweep after a failed restart; see noteRestartFailed
2178 serialBroken bool // last ReadSerial failed and was logged; reset on success
2179 serialNotify chan<- struct{}
2180
2181 // The change index (CARLOS_CHANGE_INDEX=read, S3 change-index design
2182 // §4.5); watch nil leaves the tick exactly as before. The watch is shared
2183 // with the record view, so Poll's moved list is advisory: the tick diffs
2184 // Observed against indexActed, the generations it last acted on, and
2185 // moves that map on only once a pass has covered them.
2186 watch *release.ChangeWatch
2187 scopedPass func(ctx context.Context, moved []string) // passLegs.scoped, or a test hook
2188 fullDue *atomic.Bool // the backstop's request for a full pass; nil = none
2189 indexActed map[string]uint64 // nil until the first good observation seeds it
2190 indexBroken bool // last Poll failed and was logged; reset on success
2191
2192 // canaryGate gates the sweep's canary pointer reads (spec §5): a canary
2193 // route stays a member of every tick, but its group's pointer is read
2194 // only when the app's scope moved, the last read failed or is the
2195 // backstop's age, or a route of the group is not live on the head that
2196 // read returned (canaryHeads). nil: every pointer is read, as before.
2197 canaryGate *release.ReadGate
2198 canaryHeads map[[3]string]canaryRead // (account, app, channel) -> its group's last good read
2199}
2200
2201// canaryRead is one canary group's last good pointer read: the head it
2202// returned, the app scope's generation captured before it, and when. The
2203// gate's mark is per app scope, so another channel of the same app can mark
2204// it; a group is settled only on its own read (final review R1, finding 6).
2205type canaryRead struct {
2206 head string
2207 gen uint64
2208 at time.Time
2209}
2210
2211// restartFailure is one host's cross-pass memo of a version its restart
2212// refused: which version failed, the current hold step, and when the sweep
2213// may try that host again. Kept on the loop (fastDeps), never in
2214// release.Pass — the pass stays stateless; this loop is the caller whose
2215// cadence needs the bound.
2216type restartFailure struct {
2217 ver string
2218 step time.Duration
2219 notBefore time.Time
2220}
2221
2222// fastPassOpts is the option set EVERY fast-converge pass carries — the
2223// serial-triggered full release.Pass and the sweep's release.PassFilter
2224// both, built in one place so neither call site can drift from the other.
2225//
2226// DO NOT add release.NagExpiredCanaries(...) to this set, ever. This
2227// is THE loop release.NagExpiredCanaries' doc comment warns about by
2228// name: every 2-3 seconds, canary/* always in the set. An expired canary sitting
2229// still would nag on every single tick — a journal-flooding poll storm
2230// out of what is supposed to be an hourly, easy-to-skim log line. The
2231// nag belongs to RunAdopt (carlos-adopt.service) alone.
2232//
2233// DO NOT add release.WithCapabilities(...) to this set, ever. Same
2234// sandbox as adoptLoop above (ProtectSystem=strict,
2235// ReadWritePaths=/data /run/carlos), and this one ticks every 2-3
2236// SECONDS: arming the converge here would write nothing and log a
2237// permission denial forever. The accepted consequence, ruled on in
2238// the 2026-08-03 review of the capability design: a freshly promoted
2239// canary runs its new binary against the unit shape the previous
2240// version left until the hourly root pass converges it — up to an
2241// hour. release.adoptHost prints one line per such repoint so the
2242// window is visible in the journal rather than silent.
2243//
2244// OnStageBlocked IS wired here, and this is the loop it matters most
2245// for: a canary is promoted to be picked up in seconds, and staging is
2246// the one step of that pickup this process cannot perform. Throttled
2247// inside stageRequester — this tick is 2-3 seconds, the request is at
2248// most one per stageRequestEvery (the requester is built ONCE per loop, so the budget
2249// is the loop's, not one tick's).
2250//
2251// RestartHibernating IS wired here too, and for the same reason as on
2252// adoptLoop: this process owns the exec children. It matters more on
2253// this loop than on any other — a promotion is bumped or fast-swept
2254// precisely to be picked up in seconds, and the routes this loop reaches
2255// are the warm ones somebody is watching. Pinned by
2256// TestFastSweepBouncesAWarmHibernatingCanary: dropping it would restore
2257// a dead-branch production bug without failing any other test.
2258//
2259// WithStartUnit IS wired here too (start — makeStartRoute in production,
2260// nil in tests that never reach a rollback): any pass whose restart
2261// callback carries the named-unit arm must also carry the start that
2262// undoes that arm's failed restart, or its rollback leaves the service
2263// down and the next pass stamps a never-run build
2264// (release.WithStartUnit). It asks for no authority the try-restart arm
2265// did not already need, and it cannot fire for a sleeper: only an
2266// ERRORED restart reaches it, and every arm that can meet a sleeping
2267// route swallows its failure.
2268//
2269// unitsOption hands named units to root when this process is not root
2270// (issue #315); see agentUnitsOption.
2271func fastPassOpts(dd string, stageReq func(host, ver string, err error), restart, start func(string) error) []release.PassOption {
2272 return append([]release.PassOption{
2273 release.BakeDir(dd),
2274 release.OnStageBlocked(stageReq),
2275 release.RestartHibernating(restart),
2276 release.WithStartUnit(start),
2277 release.WithDigestRatchet(pointerRatchet(false, release.OptDir(), dd, os.Getenv)),
2278 // Settings spec §4: the agent applies root's decisions, never
2279 // resolves. The book is opened per pass, not here: these options are
2280 // built once per agent run.
2281 updatesPassOption(dd, false),
2282 }, agentUnitsOption()...)
2283}
2284
2285// agentRunsAsRoot is the agent's euid test, a seam for tests.
2286var agentRunsAsRoot = func() bool { return os.Geteuid() == 0 }
2287
2288// agentUnitsOption is release.RootRestartsUnits for an agent that is not
2289// root (issue #315). carlos-edge.service runs as User=carlos, and systemd
2290// refuses that user a restart of any unit a route names, so a named-unit
2291// route goes to the root pass through the stage relay before anything moves.
2292// An agent run as root restarts the unit itself, as before.
2293func agentUnitsOption() []release.PassOption {
2294 if agentRunsAsRoot() {
2295 return nil
2296 }
2297 return []release.PassOption{release.RootRestartsUnits()}
2298}
2299
2300// fastMembers computes the fast set for one tick — the routes with a reason
2301// to be looked at FASTER than the hourly floor:
2302//
2303// - canary/* routes, always: a direct-bucket operator promote
2304// (scripts/promote-app.sh, or `carlos promote` against the bucket)
2305// never bumps the serial, so the sweep is the only thing keeping the
2306// "canary adopts in seconds" promise (spec §4 correction).
2307// - unstaged routes (no live symlink), once their backoff window has
2308// passed: a route was just minted, or its stage failed; the sweep
2309// picks it up as soon as there is something to stage.
2310// - routes waiting for a dev proof (D3), each time the proof gate's
2311// cached miss runs out: they have a live version, but their head is
2312// held until its proof arrives.
2313//
2314// Computed BEFORE any store work so an empty set costs zero store reads —
2315// the idle box pays one serial GET per tick and nothing else. A registry
2316// list failure returns the empty set (a broken box, logged per tick by
2317// statusLoop's identical call — not worth a second line every 2s here).
2318func (d *fastDeps) fastMembers(now time.Time) []string {
2319 routes, err := d.reg.List()
2320 if err != nil {
2321 return nil
2322 }
2323 var members []string
2324 seen := map[string]bool{}
2325 for _, rt := range routes {
2326 // release.Adoptable, not a local copy: the sweep must count as
2327 // outstanding work exactly the routes a pass would consider —
2328 // anything else (platform aliases, non-ready routes, routes with no
2329 // app) would be swept forever, since the pass refuses it on shape.
2330 if !release.Adoptable(rt) || seen[rt.Host] {
2331 continue
2332 }
2333 // A host inside its restart-failure hold is outside the sweep
2334 // entirely — canary or unstaged. Sweeping it again at this cadence
2335 // would repoint, cycle the RUNNING old service through a restart
2336 // that is going to fail, roll back and start it again, every 2-3
2337 // seconds for as long as the broken head stands (before the
2338 // rollback-start fix that flap self-terminated after two ticks — by
2339 // stamping the lie instead). The hourly floors still retry held
2340 // hosts; a console promote's serial bump still converges them
2341 // immediately (fullPass is not membership-gated); and a restart that
2342 // SUCCEEDS clears the hold at once (observeRestart).
2343 if d.restartHeld(rt.Host, now) {
2344 continue
2345 }
2346 if isCanaryRoute(rt) {
2347 seen[rt.Host] = true
2348 members = append(members, rt.Host)
2349 continue
2350 }
2351 // D3 (spec §4.5, codex 14): a listed route waiting for a dev proof
2352 // stays in the sweep although it has a live version, whenever its
2353 // 30 s miss has run out, so it adopts soon after the proof arrives
2354 // rather than at the next serial bump or the hourly floor. A wait the
2355 // last sweep left standing (fastTick) is inside a backoff window, and
2356 // sits it out like an unstaged route.
2357 if d.proof.WaitingDue(rt.Host) {
2358 if nb, ok := d.notBefore[rt.Host]; ok && now.Before(nb) {
2359 continue
2360 }
2361 seen[rt.Host] = true
2362 members = append(members, rt.Host)
2363 continue
2364 }
2365 // A LiveVersion error reads as unstaged: the pass is the thing that
2366 // can repair a broken link, and the backoff bounds the retry cost.
2367 if v, err := release.LiveVersion(d.optDir, rt.Host); err == nil && v != "" {
2368 continue // staged: nothing outstanding at this cadence
2369 }
2370 if nb, ok := d.notBefore[rt.Host]; ok && now.Before(nb) {
2371 continue // still inside this host's backoff window
2372 }
2373 seen[rt.Host] = true
2374 members = append(members, rt.Host)
2375 }
2376 return members
2377}
2378
2379// noteUnstaged records that host came through a sweep still unstaged:
2380// double its backoff (base 2s, capped at 5min) and push its next look out.
2381// Held routes cost almost nothing this way: the pass's own bake logic
2382// refuses them and the backoff keeps re-checks rare; the serial-triggered
2383// and hourly passes re-notice them when the window expires.
2384func (d *fastDeps) noteUnstaged(host string, now time.Time) {
2385 if d.step == nil {
2386 d.step = map[string]time.Duration{}
2387 d.notBefore = map[string]time.Time{}
2388 }
2389 s := d.step[host]
2390 if s == 0 {
2391 s = fastBackoffBase
2392 } else if s *= 2; s > fastBackoffCap {
2393 s = fastBackoffCap
2394 }
2395 d.step[host] = s
2396 d.notBefore[host] = now.Add(s)
2397}
2398
2399// proofWaitRenewed reports whether host is on the proof gate's waiting list
2400// with a cached miss still running: the sweep just asked the gate about it
2401// and the gate held it again.
2402func (d *fastDeps) proofWaitRenewed(host string) bool {
2403 if d.proof.WaitingDue(host) {
2404 return false
2405 }
2406 for _, w := range d.proof.Waits() {
2407 if w.Host == host {
2408 return true
2409 }
2410 }
2411 return false
2412}
2413
2414// backoffFor reports host's current backoff step (zero when none recorded).
2415func (d *fastDeps) backoffFor(host string) time.Duration { return d.step[host] }
2416
2417// noteRestartFailed records that host's restart onto ver just failed, and
2418// holds the host out of the sweep for a growing window: same version failing
2419// again doubles the step (capped), a DIFFERENT version starts the hold fresh
2420// at the base — a new promotion is new work, not a continuation of the old
2421// failure. Keyed per host with the version recorded, rather than per
2422// (host, version) pairs accumulating: one live symlink per host means one
2423// version under attempt at a time, and the map stays bounded by the fleet.
2424func (d *fastDeps) noteRestartFailed(host, ver string, now time.Time) {
2425 if d.restartHold == nil {
2426 d.restartHold = map[string]*restartFailure{}
2427 }
2428 m := d.restartHold[host]
2429 if m == nil || m.ver != ver {
2430 m = &restartFailure{ver: ver, step: fastRestartBackoffBase}
2431 } else if m.step *= 2; m.step > fastRestartBackoffCap {
2432 m.step = fastRestartBackoffCap
2433 }
2434 m.notBefore = now.Add(m.step)
2435 d.restartHold[host] = m
2436}
2437
2438// restartHeld reports whether host is inside its restart-failure hold.
2439func (d *fastDeps) restartHeld(host string, now time.Time) bool {
2440 m := d.restartHold[host]
2441 return m != nil && now.Before(m.notBefore)
2442}
2443
2444// observeRestart wraps the restart dispatch so this loop SEES its outcomes —
2445// the seam the restart-failure hold feeds on. On an error the version under
2446// attempt is read from the live symlink (adoptHost repoints BEFORE it
2447// consults the callback, so at callback time the symlink names the version
2448// being tried; the read failing degrades the memo to host-keyed, which only
2449// makes the hold slightly coarser). On success the hold is cleared: the host
2450// is converged, or a recovery promote just landed, and either way there is
2451// nothing left to hold it out of the sweep for.
2452//
2453// The wrapper never SUPPRESSES a restart — suppression inside the callback
2454// would return nil to adoptHost, which would stamp SetVersions over a
2455// process nothing cycled: the exact lie this whole mechanism exists to
2456// prevent. The hold gates sweep MEMBERSHIP (fastMembers) instead, so a held
2457// host is simply not repointed at this cadence at all.
2458//
2459// Single-goroutine by construction, like every other fastDeps map: the
2460// callback runs inside PassFilter/Pass inside fastTick, on the loop's own
2461// goroutine.
2462func (d *fastDeps) observeRestart(restart func(string) error) func(string) error {
2463 return func(host string) error {
2464 err := restart(host)
2465 if err != nil {
2466 ver, _ := release.LiveVersion(d.optDir, host)
2467 d.noteRestartFailed(host, ver, time.Now())
2468 } else {
2469 delete(d.restartHold, host)
2470 }
2471 return err
2472 }
2473}
2474
2475// convergeAdopted finishes the adoptions this process delegated or never saw.
2476//
2477// The agent is sandboxed (carlos-edge.service: ProtectSystem=strict, no /opt
2478// write), so on a real box a new version is staged, pointed live and stamped
2479// by a ROOT pass — carlos-stage.service on this loop's own request, or the
2480// hourly carlos-adopt.service. That pass cannot cycle an exec child
2481// (rootRestartRoute: the children are this process's), and once it has run,
2482// this process's own pass finds the host up to date and never reaches the
2483// restart callback RestartHibernating exists to hand it. The warm child was
2484// left on the old build until the activator sweep's ConvergeGrace ran out:
2485// 110-200s per deploy on the flagship, past `carlos deploy`'s own watch
2486// (discussion #231).
2487//
2488// So, for each warm instance the activator reports drifted, cycle it once the
2489// route is STAMPED with the version its symlink now names. The stamp is the
2490// adopting pass saying it is finished with the host (adoptHost writes it last,
2491// after its restart and any rollback); a drift the stamp does not yet agree
2492// with is an adoption still in flight, and this leg waits a tick for it rather
2493// than race it. execBounceable keeps the usual line between the two halves: a
2494// route whose process the root pass's systemctl arms can reach is never this
2495// leg's to cycle.
2496//
2497// At most limit instances are cycled per call, in host order. One pass can
2498// adopt every warm instance of an app at once — a root pass repoints them all
2499// inside a few seconds — and stopping them together would drain, checkpoint
2500// and later restore them together too: every host of the app blips at the
2501// same moment. The activator sweep spends the same care (one convergence
2502// cycle per sweep); this leg keeps it at its own cadence, so N instances
2503// converge over N ticks rather than in one. Only a cycle that actually
2504// started counts against the limit — a drifted host this leg must wait on
2505// costs the others nothing.
2506//
2507// Never wakes anything — ConvergeAdopted is a stop, and a sleeper execs the
2508// new build at its own next wake. drifted returning nothing is the ordinary
2509// tick and costs no registry read. Returns how many it cycled.
2510func convergeAdopted(drifted func() map[string]string, get func(host string) (*registry.Route, error), converge func(host, ver string) bool, limit int) int {
2511 stale := drifted()
2512 if len(stale) == 0 {
2513 return 0
2514 }
2515 hosts := make([]string, 0, len(stale))
2516 for host := range stale {
2517 hosts = append(hosts, host)
2518 }
2519 sort.Strings(hosts)
2520 n := 0
2521 for _, host := range hosts {
2522 if n >= limit {
2523 break
2524 }
2525 live := stale[host]
2526 rt, err := get(host)
2527 if err != nil || rt == nil || rt.Version != live || !execBounceable(*rt) {
2528 continue
2529 }
2530 if converge(host, live) {
2531 n++
2532 log.Printf("agent: %s was adopted onto %s by another pass — cycled its warm instance onto it", host, live)
2533 }
2534 }
2535 return n
2536}
2537
2538// adoptedCyclesPerTick is convergeAdopted's limit per run of its loop: one
2539// warm instance every adoptedConvergeEvery. A single-instance app — the common
2540// case — converges on the first run after its adoption is stamped; an app with
2541// a dozen warm instances takes under half a minute, one blip at a time.
2542const adoptedCyclesPerTick = 1
2543
2544// adoptedConvergeEvery is how often the converge leg looks for a finished
2545// adoption to act on.
2546const adoptedConvergeEvery = 2 * time.Second
2547
2548// convergeLegFor builds the converge leg over the box's activator. nil for a
2549// box with none armed: there are no exec children to cycle. On a box that
2550// declined CARLOS_SWAP=overlap but honours failure records (heldHolds), a
2551// host whose installed build a record holds for it is passed by: cycling it
2552// would stop it into a wake its Planner refuses (final review F6).
2553func convergeLegFor(svc *activator.Service, reg *registry.Registry) func() {
2554 if svc == nil {
2555 return nil
2556 }
2557 get := reg.Get
2558 if h := heldHolds(); h != nil {
2559 get = h.skipGet(reg.Get)
2560 }
2561 return func() {
2562 convergeAdopted(svc.Drifted, get, svc.ConvergeAdopted, adoptedCyclesPerTick)
2563 }
2564}
2565
2566// runConvergeLoop runs the converge leg every `every` until stop closes (nil:
2567// for the life of the process).
2568//
2569// Its own loop, not a leg of fastTick, and the reason is measured: on the
2570// flagship's Ireland box a fast tick ran 25 to 50 seconds whenever the serial
2571// moved (a full reconcile and adopt pass over ~180 routes), and console
2572// activity across every tenant moves it about once a minute. Riding the tick,
2573// a stamped adoption waited ~48s for its cycle (discussion #231, 2026-10-05).
2574// The leg needs none of the tick's state: it reads the activator and the
2575// registry, both safe to read beside the tick, and its one write is a stop
2576// under the activator's own lock — where the tick's adopt bounce is
2577// version-aware (BounceAdopted), so the two cannot stop one child twice for
2578// one build.
2579func runConvergeLoop(leg func(), every time.Duration, stop <-chan struct{}) {
2580 t := time.NewTicker(every)
2581 defer t.Stop()
2582 for {
2583 select {
2584 case <-stop:
2585 return
2586 case <-t.C:
2587 leg()
2588 }
2589 }
2590}
2591
2592// slowLegAfter is when logSlow speaks: a fast-loop leg that takes longer than
2593// this is worth a line, because the loop's promise is seconds.
2594const slowLegAfter = 5 * time.Second
2595
2596// logSlow logs that what took longer than slowLegAfter, with detail. Meant
2597// for a defer: the loop's latency was invisible in the journal until someone
2598// reconstructed it from timestamps by hand (#231).
2599func logSlow(what string, start time.Time, detail string) {
2600 if d := time.Since(start); d > slowLegAfter {
2601 if detail != "" {
2602 detail = " (" + detail + ")"
2603 }
2604 log.Printf("agent: fast converge: %s took %s%s", what, d.Round(100*time.Millisecond), detail)
2605 }
2606}
2607
2608// fastTick is one fast-converge tick — two independent jobs (spec §3-§4):
2609//
2610// 1. Serial check: one ReadSerial. The FIRST successful read only seeds
2611// lastSerial — a present object at its value (boot must not replay
2612// history; adoptLoop's 1-minute first pass covers boot), an absent one
2613// at zero (absence is not history, and the first-ever bump on a fresh
2614// bucket must fire, not read as replay). Afterward any CHANGE
2615// (not just increments — a reset must not wedge the box) runs the full
2616// pass set, clamped to one per fastFullClamp: a bump inside the window
2617// is deferred by NOT recording the acted serial, so the next tick past
2618// the window re-fires. Read errors are tolerated silently (one log
2619// line per state change, never per tick).
2620// The restart leg (issue #113) rides that same move, AFTER the pass set:
2621// a bump can carry a promote and a restart request at once, and cycling
2622// processes before their new version and config land would cycle them onto
2623// the old ones. It is a separate call rather than a fourth leg inside
2624// fullPass because it is not release convergence — it restarts what is
2625// already correct — and because it must still run on a box with no
2626// CARLOS_SECRETS_KEY, which fullPass's config leg silently skips. The
2627// sidecar wake leg (design: sidecar members) rides the same move
2628// immediately after, for the identical reason: a bump can carry a
2629// provisioning change and a wake request together, and it must run on
2630// a box with no CARLOS_SECRETS_KEY too.
2631// The schedule converge (schedule spec §6.2) is a further leg on that same
2632// move, for both of the same reasons; see the call site. The layout
2633// leg (isolated-exec design §8.2) follows it and only starts a pass in
2634// the background; the system leg's observe comes last.
2635// 2. Fast sweep: fastMembers, and only when non-empty one PassFilter over
2636// exactly those hosts — an empty sweep costs zero store reads. After
2637// the sweep, members still unstaged back off, and so do members whose
2638// proof wait the sweep left standing; other staged members clear their
2639// backoff state.
2640func fastTick(ctx context.Context, d *fastDeps, now time.Time) {
2641 // The index is observed BEFORE any pass this tick runs, so a pass covers
2642 // at least what was observed and indexActed can take it afterwards.
2643 obs, indexOK := d.observeIndex(ctx)
2644 fullRan := false
2645 doc, _, err := release.ReadSerial(ctx, d.store)
2646 switch {
2647 case err != nil:
2648 if !d.serialBroken {
2649 log.Printf("agent: fast converge serial read: %v", err)
2650 d.serialBroken = true
2651 }
2652 case !d.seeded:
2653 // The first successful read seeds — the ABSENT object included,
2654 // which ReadSerial reports as the zero doc. Absence is not history:
2655 // there is nothing to replay, and every deployment's bucket starts
2656 // without the object, so seeding zero is what lets the first-ever
2657 // console bump (0 → 1) fire instead of being swallowed as boot
2658 // noise. A PRESENT object seeds at its current value, and THAT is
2659 // the no-replay rule: the bumps behind it are history a booting box
2660 // must not act on (adoptLoop's 1-minute first pass covers boot).
2661 d.lastSerial, d.seeded = doc.Serial, true
2662 d.serialBroken = false
2663 default:
2664 d.serialBroken = false
2665 if doc.Serial != d.lastSerial && now.Sub(d.lastFull) >= fastFullClamp {
2666 d.runFull(ctx)
2667 fullRan = true
2668 if d.restartLeg != nil {
2669 d.restartLeg(ctx)
2670 }
2671 if d.wakeLeg != nil {
2672 d.wakeLeg(ctx)
2673 }
2674 // The schedule converge rides the same move, after the pass set for
2675 // the same reason the restart leg does: a bump can carry a promote
2676 // and a `carlos schedule set` at once, and converging schedules
2677 // before the version they belong to has staged would write rows
2678 // against the outgoing build. Also a separate leg rather than a
2679 // member of fullPass — it is not release convergence, and it must
2680 // still run on a box with no CARLOS_SECRETS_KEY.
2681 if d.scheduleLeg != nil {
2682 d.scheduleLeg(ctx)
2683 }
2684 // The layout pass converges to the fleet's recorded layout on
2685 // every move, after the pass set has minted and staged whatever
2686 // the move carried, so a freshly minted row is seen. It only
2687 // starts the pass: the pass runs on its own goroutine, because
2688 // it can wait on stops, and nothing above or below may wait on
2689 // it.
2690 if d.layoutLeg != nil {
2691 d.layoutLeg(ctx)
2692 }
2693 if d.serialNotify != nil {
2694 select {
2695 case d.serialNotify <- struct{}{}:
2696 default:
2697 }
2698 }
2699 // The platform self-update leg rides the move too, last: a
2700 // `carlos system update` is a console action like any other, and
2701 // this is what makes a promote noticed in seconds instead of at
2702 // the leg's own hourly floor. It only READS the pointer and sets
2703 // a deadline — the fetch and the request happen in the tick
2704 // below, after the spread, so a serial move costs one Get.
2705 if d.systemLeg != nil {
2706 d.systemLeg.observe(ctx)
2707 }
2708 d.lastSerial = doc.Serial
2709 d.lastFull = now
2710 }
2711 }
2712
2713 // The backstop's full pass (spec §4.4), on this goroutine: requested
2714 // after each rebuild, run once, under the same clamp.
2715 if !fullRan && d.fullDue != nil && d.fullDue.Load() && now.Sub(d.lastFull) >= fastFullClamp {
2716 d.runFull(ctx)
2717 fullRan = true
2718 d.lastFull = now
2719 }
2720 // An index-only change runs the scoped pass (spec §4.5).
2721 d.indexTick(ctx, now, obs, indexOK, fullRan)
2722
2723 // The self-update tick, outside the serial switch and before the sweep's
2724 // early return: it is what runs a spread deadline down and re-touches an
2725 // unanswered request, and neither has anything to do with routes. It
2726 // reads no store unless a deadline has actually come due.
2727 if d.systemLeg != nil {
2728 d.systemLeg.tick(ctx, now)
2729 }
2730
2731 // D3's mint asker: on a proving box, ask for a root pass every 2 minutes
2732 // while a listed build has baked with no proof. Elsewhere it reads one
2733 // absent file and returns.
2734 if d.mintLeg != nil {
2735 d.mintLeg(ctx)
2736 }
2737
2738 members := d.fastMembers(now)
2739 if len(members) == 0 {
2740 return // zero store reads: no PassFilter, no bake-clock write
2741 }
2742 defer logSlow("fast sweep", time.Now(), fmt.Sprintf("%d routes", len(members)))
2743 member := map[string]bool{}
2744 for _, h := range members {
2745 member[h] = true
2746 }
2747 opts := d.passOpts
2748 // The canary gate only on a tick whose own index observation succeeded:
2749 // a tick that could not read the index reads every pointer, as before
2750 // the index. (A failed poll also leaves the watch unknown, so the gate
2751 // would find every group due anyway; this keeps it off whatever another
2752 // caller's poll did in between.)
2753 if d.canaryGate != nil && indexOK {
2754 opts = append(slices.Clip(opts), release.WithPointerGate(d.canaryPointerGate(now)))
2755 }
2756 if err := release.PassFilter(ctx, d.reg, d.store, d.pub, d.optDir, d.restart,
2757 func(rt registry.Route) bool { return member[rt.Host] }, opts...); err != nil {
2758 log.Printf("agent: fast converge sweep: %v", err)
2759 }
2760 for _, h := range members {
2761 switch v, err := release.LiveVersion(d.optDir, h); {
2762 case err != nil || v == "":
2763 d.noteUnstaged(h, now)
2764 // An unstaged route whose wait this sweep renewed is waiting
2765 // for its proof, not failing to stage: spec §4.5 has it rejoin
2766 // the sweep each time its 30 s miss runs out, so its backoff
2767 // stops doubling there.
2768 if d.proofWaitRenewed(h) && d.step[h] > release.ProofMissTTL {
2769 d.step[h] = release.ProofMissTTL
2770 d.notBefore[h] = now.Add(release.ProofMissTTL)
2771 }
2772 case d.proof.WaitingDue(h):
2773 // The sweep neither renewed this host's proof wait (which would
2774 // cache a fresh 30 s miss) nor cleared it: adoptHost returned
2775 // before the gate, say on a channel with no pointer. Nothing will
2776 // change that by the next tick, so it backs off like an unstaged
2777 // route instead of being swept every 2 s (plan 3 ruling 5).
2778 d.noteUnstaged(h, now)
2779 default:
2780 delete(d.notBefore, h)
2781 delete(d.step, h)
2782 }
2783 }
2784}
2785
2786// canaryPointerGate is the sweep's release.PointerGate at now. Only canary
2787// groups are gated; every other group is read as before. A canary group is
2788// skipped when its app scope has not moved since the group's own last good
2789// read (another channel's read of the same app does not count), that read is
2790// younger than the backstop, and every route of the group is live on the
2791// head it returned: then a read could only repeat what the box already
2792// runs. Anything unsettled (an unstaged route, a failed restart rolled back,
2793// a proof wait, a build the update settings chose over the head) keeps
2794// today's read on every tick.
2795func (d *fastDeps) canaryPointerGate(now time.Time) release.PointerGate {
2796 return func(acct, app, channel string, routes []registry.Route) (bool, func(*release.ChannelPointer, error)) {
2797 if !strings.HasPrefix(channel, "canary/") {
2798 return false, nil
2799 }
2800 scope, gated := release.ChangeScope(release.ChannelKey(acct, app, channel))
2801 if !gated {
2802 return false, nil
2803 }
2804 k := [3]string{acct, app, channel}
2805 gen, due := d.canaryGate.Due(scope, now)
2806 if !due && d.canarySettled(k, routes, gen, now) {
2807 return true, nil
2808 }
2809 return false, func(ptr *release.ChannelPointer, err error) {
2810 if d.canaryHeads == nil {
2811 d.canaryHeads = map[[3]string]canaryRead{}
2812 }
2813 if err != nil || ptr == nil {
2814 delete(d.canaryHeads, k)
2815 d.canaryGate.Mark(scope, gen, now, false)
2816 return
2817 }
2818 d.canaryHeads[k] = canaryRead{head: ptr.Version, gen: gen, at: now}
2819 d.canaryGate.Mark(scope, gen, now, true)
2820 }
2821 }
2822}
2823
2824// canarySettled reports whether group k's own last good read is at gen (the
2825// generation Due returned) and younger than the gate's maximum age, and every
2826// route of the group is live on the head that read returned.
2827func (d *fastDeps) canarySettled(k [3]string, routes []registry.Route, gen uint64, now time.Time) bool {
2828 r, ok := d.canaryHeads[k]
2829 if !ok || r.head == "" || len(routes) == 0 || r.gen != gen || now.Sub(r.at) >= d.canaryGate.MaxAge() {
2830 return false
2831 }
2832 for _, rt := range routes {
2833 if v, err := release.LiveVersion(d.optDir, rt.Host); err != nil || v != r.head {
2834 return false
2835 }
2836 }
2837 return true
2838}
2839
2840// runFull runs the full pass set's fullPass, taking any backstop request
2841// first: a full pass that starts after the backstop's rebuild satisfies it.
2842func (d *fastDeps) runFull(ctx context.Context) {
2843 if d.fullDue != nil {
2844 d.fullDue.Store(false)
2845 }
2846 d.fullPass(ctx)
2847}
2848
2849// observeIndex polls the change index and returns its observed generations;
2850// ok is false when no watch is armed or the index is unusable (then the
2851// serial and the floors carry the tick, as before the index). A failure is
2852// logged once per state change, never per tick.
2853func (d *fastDeps) observeIndex(ctx context.Context) (map[string]uint64, bool) {
2854 if d.watch == nil {
2855 return nil, false
2856 }
2857 // Poll's moved list is advisory (shared watch): only Observed counts.
2858 // Bounded: the tick runs on a background context, and a hung index GET
2859 // would otherwise stall the tick and hold the watch's poll lock that the
2860 // edge's loops share. Running out is "unknown", like any other error.
2861 pctx, cancel := context.WithTimeout(ctx, release.ChangeIndexPollTimeout)
2862 _, err := d.watch.Poll(pctx)
2863 cancel()
2864 if err != nil {
2865 if !d.indexBroken {
2866 log.Printf("agent: fast converge change index: %v (scoped passes wait; the serial and the floors carry on)", err)
2867 d.indexBroken = true
2868 }
2869 return nil, false
2870 }
2871 obs, ok := d.watch.Observed()
2872 if ok && d.indexBroken {
2873 log.Printf("agent: fast converge change index readable again")
2874 d.indexBroken = false
2875 }
2876 return obs, ok
2877}
2878
2879// indexTick acts on an index observation: the first one seeds indexActed
2880// without a pass (boot must not replay history; the backstop's first full
2881// pass and the loops' first passes cover boot), a full pass this tick
2882// covers everything observed, and otherwise app and account scopes that
2883// moved since indexActed run one scoped pass, clamped like the full pass:
2884// a change inside the clamp leaves indexActed as it was, so the next tick
2885// past the window acts on it. indexActed moves on only after a pass ran.
2886func (d *fastDeps) indexTick(ctx context.Context, now time.Time, obs map[string]uint64, ok, fullRan bool) {
2887 if !ok {
2888 return
2889 }
2890 if d.indexActed == nil || fullRan {
2891 d.indexActed = obs
2892 return
2893 }
2894 moved := indexScopes(d.indexActed, obs)
2895 if len(moved) == 0 {
2896 d.indexActed = obs // control/ scopes only: not this loop's business
2897 return
2898 }
2899 if now.Sub(d.lastFull) < fastFullClamp || d.scopedPass == nil {
2900 return
2901 }
2902 d.scopedPass(ctx, moved)
2903 d.indexActed = obs
2904 d.lastFull = now
2905}
2906
2907// fastConvergeLoop replaces the canary-only fast adopt loop with the
2908// carlos-deploy fast set (spec §3-§4): a 2s serial poll that turns a console
2909// bump (promote, instance create, config write) into the ordinary verified
2910// pass set — reconcile → adopt → exec config sync — within seconds, plus a
2911// fast sweep of the routes with genuinely outstanding work (canary/* always;
2912// unstaged routes until they drain, with per-host backoff). Everything it
2913// triggers is the same signature-verified, bake-gated work the hourly loops
2914// run; the serial carries no authority, only earliness.
2915//
2916// Setup mirrors adoptLoop's: missing deployment config or pubkey disables
2917// the loop silently (adoptLoop already logged the reason). The config-sync
2918// leg degrades exactly as configSyncLoop does — no CARLOS_SECRETS_KEY, or a
2919// broken one, and the full pass set simply omits that leg (configSyncLoop
2920// already logged that too). Takes the activator for the same reason
2921// adoptLoop does: a fresh version on a warm exec instance must bounce onto
2922// its new binary, not wait for hibernation. Nil is fine.
2923//
2924// Thin and untested by convention, like adoptLoop/statusLoop — the tick's
2925// whole decision lives in fastTick/fastMembers, which are tested.
2926func fastConvergeLoop(reg *registry.Registry, svc *activator.Service, sd *scheduleDeps, writes hostwrite.Config, serialNotify chan<- struct{}, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
2927 ctx := context.Background()
2928 store, err := openDeployStore(ctx)
2929 if err != nil {
2930 declinedHoldsStay("no deploy store")
2931 return // adoptLoop already logged this
2932 }
2933 pub, err := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
2934 if err != nil {
2935 declinedHoldsStay("no release key")
2936 return // ditto
2937 }
2938 var priv *ecdh.PrivateKey
2939 if raw := os.Getenv("CARLOS_SECRETS_KEY"); raw != "" {
2940 if p, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY"); err == nil {
2941 priv = p
2942 }
2943 }
2944 dd := dataDir()
2945 optDir := release.OptDir()
2946 // d is built empty first so the restart dispatch can be wrapped in its
2947 // observer (observeRestart): a failed restart holds its host out of the
2948 // sweep instead of the sweep re-cycling the service every tick. The
2949 // remaining fields are filled in right below.
2950 d := &fastDeps{reg: reg, store: store, pub: pub, optDir: optDir, serialNotify: serialNotify}
2951 restart := d.observeRestart(makeRestartRoute(reg, systemctlTryRestart, agentCheckedUnitRestart(systemdUnits{}, optDir, os.Getenv, nil), adoptBounceFn(svc)))
2952 // Built once, used by both pass shapes below — see fastPassOpts for the
2953 // DO-NOT list and why the stage requester is per-loop, not per-tick.
2954 opts := append(fastPassOpts(dd, stageRequester(dd), restart, makeStartRoute(reg, systemctlStartUnit)), release.WithHostWrites(writes))
2955 // The rollback's restart (D1): checked like the forward one, and a replace
2956 // rather than a start, so a rejected build left up is cycled out.
2957 opts = append(opts, release.WithRollbackRestart(agentRollbackUnitRestart(reg, systemdUnits{}, optDir, os.Getenv)))
2958 // D3: the process's one proof gate rides every pass this loop runs, and
2959 // fastMembers reads its waiting list. The mint asker has its own
2960 // requester, so its 2-minute asks never spend the stage budget and its
2961 // journal line does not claim a stage.
2962 opts = append(opts, release.WithProofGate(agentProofGate()))
2963 d.proof = agentProofGate()
2964 d.mintLeg = (&mintAsker{reg: reg, store: store, sys: systemdUnits{}, optDir: optDir,
2965 policyPath: release.ProofPolicyPath, request: mintRequester(dd), now: time.Now}).tick
2966 // The restart check's memory (issue #113, narrowed requests added by
2967 // discussion #16), seeded ONCE here — two reads per app at arming (the
2968 // app-wide generation and the narrowed request log) and never again except
2969 // on a serial move, which is what keeps an idle box at exactly one serial
2970 // GET per tick. See restartSeed for why the baseline cannot wait for the
2971 // first request to establish itself.
2972 restartMem := newRestartMemory()
2973 // `carlos instances retry` (zero-downtime spec §8.4) reaches the rollout
2974 // manager through the host-retry log (a third read per app at arming,
2975 // and on a move, only where retryHost is set; hostRetryPass) or a legacy
2976 // entry in the narrowed-restart log, or, on a box that declined
2977 // the flag but honours its failure book, that book (Task 10); nil
2978 // otherwise. Each entry's grant is verified against the root key first
2979 // (spec §11).
2980 restartMem.pub = pub
2981 restartMem.retryHost = rolloutRetryHost()
2982 // The declined box's pointer pass (Task 10, ruling 194): `carlos promote
2983 // --retry` clears a record, a signed rollback a marker, read through
2984 // the same verified pointer read the armed scheduler uses, at most
2985 // every rollout.PointerReadEvery.
2986 if h := heldHolds(); h != nil {
2987 go h.pointerLoop(ctx, func(ctx context.Context, acct, app, channel string) (*release.ChannelPointer, error) {
2988 return release.Channel(ctx, store, pub, acct, app, channel)
2989 })
2990 }
2991 restartSeed(ctx, reg, store, dd, restartMem)
2992 restartBounce := bounceFn(svc)
2993 restartWake := wakeFn(svc)
2994 // The sidecar wake leg's own memory, seeded the same way and for the
2995 // same reason: one silent read per served sidecar at arming so the
2996 // FIRST wake request after a boot is acted on rather than swallowed as
2997 // the baseline (sidecarWakeSeed's own doc comment).
2998 sidecarWakeMem := newSidecarWakeMemory(writes.Enabled)
2999 sidecarWakeSeed(ctx, reg, store, sidecarWakeMem)
3000 sidecarWake := wakeFn(svc)
3001 // The moves hold view, built before the converge below: that converge
3002 // runs before the first moves pass, and an unbuilt view freezes every
3003 // host (moves spec §9.4). Nothing with the Phase 1 switch off.
3004 refreshMovesHolds(ctx, store)
3005 // The offline keep view, likewise before any reconcile pass: once it is
3006 // read, keep answers false for every host with no offline move, so the
3007 // first pass removes what it always has (movesoffline.go).
3008 refreshOfflineView(ctx, store, armedOfflineView())
3009 // carlos run's boot pass (run spec §5.2, index item 2): reconcile's
3010 // first pass is a minute away and the CLI expires a run at 60 s.
3011 runRunsBootPass(ctx)
3012 // The schedule converge's own seed, on restartSeed's precedent and for the
3013 // same shape of reason: the leg below runs only on a serial MOVE, so
3014 // without one pass here a box that booted after the bump would carry no
3015 // timer rows (and report no status) until the hourly floor in adoptLoop —
3016 // up to an hour of a schedule silently not running. It also seeds the
3017 // converger's etag memory, which is what makes a run signal older than this
3018 // process read as history rather than as a fresh request to run.
3019 if err := sd.converge(ctx); err != nil {
3020 log.Printf("agent: schedule converge (start): %v", err)
3021 }
3022 d.restart = restart
3023 d.passOpts = opts
3024 // The scheduler's pass with CARLOS_SWAP=overlap armed, nothing while
3025 // managed starts are refused, today's convergeAdopted otherwise.
3026 if leg := fastConvergeLeg(svc, reg); leg != nil {
3027 go runConvergeLoop(leg, adoptedConvergeEvery, nil)
3028 }
3029 // The serial-triggered full pass set, in spec order: mint routes
3030 // (reconcile), then reconcile aliases, then stage them (adopt), then deliver
3031 // their config. Each error is one log line, never fatal — the hourly floors
3032 // stand underneath every leg. Not membership-gated by the restart hold:
3033 // a serial bump is an explicit console action, bounded by whoever keeps
3034 // making it — and it is what lets a recovery promote through the console
3035 // converge immediately even mid-hold.
3036 legs := &passLegs{store: store, reg: reg, pub: pub, optDir: optDir, restart: restart, opts: opts,
3037 priv: priv, dd: dd, svc: svc, writes: writes, bs: bs,
3038 cfg: func() provision.ReconcileConfig {
3039 return configuredClaimProvision(provisionTickConfig(os.Getenv), configure)
3040 }}
3041 d.fullPass = legs.full
3042 // The change index (S3 change-index design §4.5): an index-only change
3043 // runs the scoped pass, and the backstop's request runs the full one,
3044 // both on this loop's goroutine. Nil watch: the tick is as before.
3045 if watch := armedChangeWatch.Load(); watch != nil {
3046 d.watch = watch
3047 d.scopedPass = legs.scoped
3048 d.fullDue = &backstopFullDue
3049 // The sweep's canary pointer reads are gated on the same watch
3050 // (spec §5). Built here, after RunAgent armed the index.
3051 d.canaryGate = newChangeGate()
3052 }
3053 // The restart leg: this process's own children bounced here, everything
3054 // unit-shaped relayed to a root pass. Deliberately NOT inside fullPass
3055 // above — see fastTick's comment for both reasons.
3056 d.restartLeg = func(ctx context.Context) {
3057 routes, err := reg.List()
3058 if err != nil {
3059 log.Printf("agent: restart check: list routes: %v", err)
3060 return
3061 }
3062 restartRequestLeg(ctx, routes, store, dd, restartBounce, restartWake, restartMem)
3063 }
3064 // The sidecar wake leg: apps/<acct>/<app>/sidecar-wake/<name>.json
3065 // (release.SidecarWakeKey) compared and dispatched here, riding the
3066 // same serial move as the restart leg immediately above it — see
3067 // fastTick's comment for why both must run after fullPass and on a
3068 // box with no CARLOS_SECRETS_KEY.
3069 d.wakeLeg = func(ctx context.Context) {
3070 routes, err := reg.List()
3071 if err != nil {
3072 log.Printf("agent: sidecar wake check: list routes: %v", err)
3073 return
3074 }
3075 sidecarWakeLeg(ctx, routes, store, sidecarWakeMem, sidecarWake)
3076 }
3077 // The schedule leg: converge this box's timer rows from the apps'
3078 // declarations. Nil when the box has no scheduled work armed, which
3079 // fastTick tolerates.
3080 d.scheduleLeg = scheduleLegFor(sd)
3081 // The layout leg: this box's label is the one its status publishes
3082 // under, which is the label an operator scopes a layout request to.
3083 layoutBox := metricsBoxLabel(os.Getenv, readMachineID())
3084 layoutPasses := newLayoutLeg(ctx, func(ctx context.Context) bool { return runLayoutLeg(ctx, store, layoutBox) })
3085 d.layoutLeg = layoutPasses.kick
3086 // The platform self-update leg (spec §4): the sandboxed half, which only
3087 // ever fetches into /data and writes a request. It lives here rather than
3088 // in a loop of its own because the spread deadline needs a seconds
3089 // cadence to run down on, and this is the loop that already has one.
3090 d.systemLeg = newSystemLeg(store, pub, os.Getenv, dd, optDir)
3091 // One observe at arming, on restartSeed's precedent: the leg below only
3092 // re-observes on a serial move or at its hourly floor, so a box that
3093 // (re)starts under an already-promoted pointer would otherwise sit on the
3094 // old build for up to an hour with nobody to bump the serial. observe
3095 // handles "no pointer" itself, which is every deployment before its first
3096 // `carlos system update`.
3097 d.systemLeg.observe(ctx)
3098 log.Printf("agent: fast converge loop armed (2s + jitter: serial poll + canary/* + unstaged routes + system pointer)")
3099 for {
3100 time.Sleep(fastTickEvery + rand.N(time.Second))
3101 fastTick(ctx, d, time.Now())
3102 }
3103}
3104
3105// isCanaryRoute reports whether rt follows a canary/* channel — the same
3106// two-segment convention bake.go's DefaultHolds wildcard matches (a bare
3107// "canary" with no slash is NOT covered, same as bake.go: it falls through
3108// to the unknown-channel default there, so it shouldn't fast-poll here
3109// either).
3110func isCanaryRoute(rt registry.Route) bool {
3111 return strings.HasPrefix(rt.Channel, "canary/")
3112}
3113
3114// countExpiredCanaries counts routes (this box's own, platform aliases
3115// excluded — same convention as CollectBox) on a canary/* channel whose
3116// pointer has passed release.CanaryTTL, for the fleet status page's §5
3117// count (internal/status Probes.ExpiredCanaries). A canary route whose
3118// pointer can't be resolved at all — store unavailable, pubkey unusable, or
3119// the specific channel won't verify — nils the WHOLE count rather than
3120// undercounting: "we couldn't check" and "there are none" are different
3121// claims, and this function only ever makes the second one when it's sure.
3122// No canary routes at all is a confident, known 0 — it needs no store to
3123// answer "none" about routes that don't exist.
3124//
3125// Pointers are resolved at most once per distinct (account, app, channel)
3126// among the routes given (the local `cache` map below), mirroring adopt.go's
3127// Pass — this function is called once per 5-minute status tick
3128// (statusLoop), so N routes on this box following the same canary channel
3129// cost exactly ONE release.Channel() store read for that tick, not N
3130// (2026-08-03 review: confirmed and pinned by
3131// TestCountExpiredCanariesReadsEachChannelOnce). The cache is local to one
3132// call/one tick — nothing persists across ticks, which is fine at a
3133// 5-minute cadence.
3134func countExpiredCanaries(ctx context.Context, routes []registry.Route, store release.Store, pub ed25519.PublicKey, now time.Time) *int {
3135 type key struct{ acct, app, channel string }
3136 cache := map[key]*release.ChannelPointer{}
3137 n := 0
3138 for _, rt := range routes {
3139 if rt.Platform || !isCanaryRoute(rt) {
3140 continue
3141 }
3142 if store == nil || len(pub) != ed25519.PublicKeySize {
3143 return nil // a canary route exists but nothing here can resolve its pointer
3144 }
3145 k := key{release.RouteAccount(rt), rt.App, rt.Channel}
3146 ptr, cached := cache[k]
3147 if !cached {
3148 p, err := release.Channel(ctx, store, pub, k.acct, k.app, k.channel)
3149 if err != nil {
3150 return nil // this route's own pointer won't verify — unknown, not zero
3151 }
3152 ptr = p
3153 cache[k] = ptr
3154 }
3155 if release.CanaryExpired(ptr, now) {
3156 n++
3157 }
3158 }
3159 return &n
3160}
3161
3162// peekAcmeCache pre-parses the agent args for -acme-cache the same way
3163// peekRegistryPath peeks -registry: the edge does the authoritative parse,
3164// this just lets the domains loop stamp cert expiry from the SAME cache dir
3165// the edge actually mints into when the operator moved it off the default.
3166// "" = flag not given; domainsTickConfig then derives the edge's own default
3167// ($CARLOS_DATA_DIR/acme).
3168func peekAcmeCache(args []string) string {
3169 fs := flag.NewFlagSet("agent", flag.ContinueOnError)
3170 fs.SetOutput(io.Discard)
3171 fs.String("registry", "", "")
3172 acme := fs.String("acme-cache", "", "")
3173 fs.String("https", "", "")
3174 fs.String("http", "", "")
3175 fs.String("fallback", "", "")
3176 fs.Bool("dev", false, "")
3177 fs.Usage = cliflag.Usage(fs, "")
3178 _ = fs.Parse(args)
3179 return *acme
3180}
3181
3182// domainsTickConfig builds one reconcile pass's Config from the environment
3183// — pure (getenv + the already-read public IP injected) so the whole env
3184// contract is testable without t.Setenv. PLATFORM_APPS_DOMAIN unset returns
3185// the zero Config, on which Reconcile is inert by contract.
3186//
3187// The SECOND return is the parsed PLATFORM_EDGE_IPS list (with the public-IP
3188// fallback applied). It is not a Config field: Config.EdgeIPs is now a func
3189// answering the deployment's LIVE edge set, which needs a deployment store —
3190// something this function must never touch if it is to stay pure and
3191// getenv-injectable. domainsLoop folds the two together once its store is
3192// open. A field only the agent's own wiring reads would be dead weight in
3193// every other caller and every test fixture, hence a return value.
3194//
3195// Resolve is armed for every configured deployment now, no longer gated on
3196// this box knowing an edge IP: a box with no PLATFORM_EDGE_IPS whose FOLD
3197// supplies edges must still stamp. The stand-down when nothing is knowable
3198// moved to where the set is actually known — domains.Reconcile's section 4,
3199// per pass — because "no edge addresses" is now a per-pass fact, not a
3200// boot-time one.
3201//
3202// The error return is ADVISORY, never fatal: it is non-nil only when
3203// CARLOS_RELEASE_PUBKEY was SET but would not parse, and it exists so the
3204// caller can log the real cause. release.ParsePublicKey says the same sentence
3205// for an empty value as for a typo'd one, so an unset key is reported as nil
3206// error and a nil ReleasePub — the two are different operator problems and the
3207// one log line about them must not name the wrong one. The Config is fully
3208// usable either way; only the auto-purge is off.
3209func domainsTickConfig(getenv func(string) string, publicIP string) (domains.Config, []string, error) {
3210 domain := strings.TrimSpace(getenv("PLATFORM_APPS_DOMAIN"))
3211 if domain == "" {
3212 return domains.Config{}, nil, nil
3213 }
3214 var ips []string
3215 for _, ip := range strings.Split(getenv("PLATFORM_EDGE_IPS"), ",") {
3216 if ip = strings.TrimSpace(ip); ip != "" {
3217 ips = append(ips, ip)
3218 }
3219 }
3220 if len(ips) == 0 && publicIP != "" {
3221 ips = []string{publicIP}
3222 }
3223 dd := getenv("CARLOS_DATA_DIR")
3224 if dd == "" {
3225 dd = "/data"
3226 }
3227 cfg := domains.Config{
3228 AppsDomain: domain,
3229 // The same default the edge's -acme-cache flag carries
3230 // (edge.RunWithActivator): $CARLOS_DATA_DIR/acme. domainsLoop
3231 // overrides this when the flag was given explicitly.
3232 AcmeDir: filepath.Join(dd, "acme"),
3233 Now: time.Now,
3234 Resolve: resolveHost,
3235 }
3236 // EdgeIPs is deliberately left nil here — a Config out of this function
3237 // alone stamps nothing. domainsLoop installs the fold once it has a
3238 // store; see the doc comment.
3239
3240 // The root release public key, same source and same parser as adoptLoop's.
3241 // Absence is TOLERATED and leaves ReleasePub nil, which switches the
3242 // auto-purge off: a box that cannot verify an app's production flag must
3243 // not destroy its objects. Everything else about the pass — including
3244 // hiding trashed apps, the reversible half — carries on regardless. A key
3245 // that is SET but unparseable is tolerated the same way, and RETURNED, so
3246 // the operator who typo'd it hears about the typo rather than being told
3247 // the variable is unset.
3248 if raw := strings.TrimSpace(getenv("CARLOS_RELEASE_PUBKEY")); raw != "" {
3249 pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
3250 if err != nil {
3251 return cfg, ips, err
3252 }
3253 cfg.ReleasePub = pub
3254 }
3255 return cfg, ips, nil
3256}
3257
3258// provisionTickConfig builds one instances reconcile pass's Config from the
3259// environment — pure (getenv-injected), mirroring domainsTickConfig's shape.
3260// Unlike domainsTickConfig, there is no inert zero-Config case: an unset
3261// CARLOS_REGION/CARLOS_POOLS is a LEGAL state, not a "feature not
3262// configured" one — see provision.Reconcile's doc comment — a box that
3263// hasn't been told its region/pools still serves only ""-scoped records,
3264// never someone else's residency promise. So this always returns a
3265// populated Config, and CARLOS_DATA_DIR is honored (default "/data")
3266// regardless of whether region/pools are set. CARLOS_INBOUND (design §8) is
3267// the same shape again: unset/"any" is today's behaviour exactly, and it is
3268// a declaration about what work THIS BOX will accept, independent of a
3269// record's own "no-inbound" placement request — see ReconcileConfig.Inbound.
3270func provisionTickConfig(getenv func(string) string) provision.ReconcileConfig {
3271 region := strings.TrimSpace(getenv("CARLOS_REGION"))
3272 var pools []string
3273 for _, p := range strings.Split(getenv("CARLOS_POOLS"), ",") {
3274 if p = strings.TrimSpace(p); p != "" {
3275 pools = append(pools, p)
3276 }
3277 }
3278 dd := getenv("CARLOS_DATA_DIR")
3279 if dd == "" {
3280 dd = "/data"
3281 }
3282 // CARLOS_INBOUND is the same apps-domain-style inert gate as
3283 // CARLOS_REGION/CARLOS_POOLS above: unset (or "any") is today's
3284 // behaviour exactly, a legal "no constraint" state, never a "feature not
3285 // configured" one. Unlike Region above it, this IS folded and validated
3286 // here (provision.NormalizeInbound) rather than left a pure passthrough:
3287 // review fix, Task 10 finding — an un-folded "None"/"NONE" or a typo'd
3288 // value compared exact-equal against InboundNone false, so a box meant
3289 // to be no-inbound would silently mint routed instances it cannot serve.
3290 // NormalizeInbound folds case/whitespace and maps any unrecognised value
3291 // to the restrictive reading (with its own log line), so
3292 // provision.Reconcile/ReconcileSidecars downstream see only the closed
3293 // vocabulary and their exact-equality comparisons stay correct.
3294 inbound := provision.NormalizeInbound(getenv("CARLOS_INBOUND"))
3295 cfg := provision.ReconcileConfig{
3296 Region: region,
3297 Pools: pools,
3298 DataDir: dd,
3299 Now: time.Now,
3300 Inbound: inbound,
3301 // The armed config guard, asked before a mint (amadan #213). Nil
3302 // when none is armed (every test that does not arm one).
3303 ConfigGate: configGate(),
3304 }
3305 // The root release public key, same source and same parser as adoptLoop's
3306 // and domainsTickConfig's. It guards ONE decision in this pass: whether a
3307 // trash marker may retire a production-flagged app's routes (#340).
3308 //
3309 // Absence is tolerated and leaves ReleasePub nil, which retires trashed
3310 // routes UNGUARDED — deliberately the same end state domains.Config
3311 // documents and accepts for the same decision (§1c), because two passes
3312 // reaching opposite conclusions about one app is worse than either
3313 // answer. A key SET but unparseable lands in the same place, so it is
3314 // logged here rather than swallowed: it is a persistent misconfiguration,
3315 // and the operator who typo'd it must hear about the typo rather than
3316 // silently losing the guard.
3317 if raw := strings.TrimSpace(getenv("CARLOS_RELEASE_PUBKEY")); raw != "" {
3318 pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
3319 if err != nil {
3320 log.Printf("agent: instances pass: %v — a trash marker will retire routes without checking the production flag", err)
3321 } else {
3322 cfg.ReleasePub = pub
3323 }
3324 }
3325 return cfg
3326}
3327
3328// resolveHost is the production DNS seam: the default resolver with a hard
3329// 3s timeout, so one dead nameserver can't stall a whole reconcile pass.
3330func resolveHost(host string) ([]string, error) {
3331 ctx, cancel := context.WithTimeout(context.Background(), 3*time.Second)
3332 defer cancel()
3333 return net.DefaultResolver.LookupHost(ctx, host)
3334}
3335
3336// imdsToken fetches an EC2 IMDSv2 session token (60s TTL) — the shared
3337// first step of every IMDS read in this file. Any failure (not on EC2,
3338// IMDS disabled, a transient metadata-endpoint hiccup) returns "", and
3339// every caller already treats a blank token as "no answer", never an
3340// error.
3341func imdsToken(client *http.Client) string {
3342 req, err := http.NewRequest(http.MethodPut, "http://169.254.169.254/latest/api/token", nil)
3343 if err != nil {
3344 return ""
3345 }
3346 req.Header.Set("X-aws-ec2-metadata-token-ttl-seconds", "60")
3347 resp, err := client.Do(req)
3348 if err != nil {
3349 return ""
3350 }
3351 defer resp.Body.Close()
3352 if resp.StatusCode != http.StatusOK {
3353 return ""
3354 }
3355 body, _ := io.ReadAll(resp.Body)
3356 return strings.TrimSpace(string(body))
3357}
3358
3359// imdsLeaf reads one IMDS leaf under the given token, trimmed. The status
3360// code travels back with the body so a caller can tell "empty leaf" from
3361// "leaf did not answer 200" without a second round trip.
3362func imdsLeaf(client *http.Client, token, path string) (string, int) {
3363 req, err := http.NewRequest(http.MethodGet, "http://169.254.169.254"+path, nil)
3364 if err != nil {
3365 return "", 0
3366 }
3367 req.Header.Set("X-aws-ec2-metadata-token", token)
3368 resp, err := client.Do(req)
3369 if err != nil {
3370 return "", 0
3371 }
3372 defer resp.Body.Close()
3373 body, _ := io.ReadAll(resp.Body)
3374 return strings.TrimSpace(string(body)), resp.StatusCode
3375}
3376
3377// readPublicIP best-effort reads this box's public IPv4 from EC2 IMDSv2
3378// (token PUT, then the public-ipv4 leaf). Any failure — not on EC2, IMDS
3379// disabled, no public address — returns "", and the caller treats that as
3380// "unknown", never as an error: a box without a discoverable public IP is a
3381// valid deployment whose DNS checks simply stand down.
3382func readPublicIP() string {
3383 client := &http.Client{Timeout: 2 * time.Second}
3384 token := imdsToken(client)
3385 if token == "" {
3386 return ""
3387 }
3388 ip, status := imdsLeaf(client, token, "/latest/meta-data/public-ipv4")
3389 if status != http.StatusOK {
3390 return ""
3391 }
3392 if net.ParseIP(ip) == nil {
3393 return ""
3394 }
3395 return ip
3396}
3397
3398// edgeDoorOptions is the edge-local door's option (#175 P7): none unless
3399// CARLOS_EDGE_DOOR names a socket path, and then edge.WithEdgeDoor with the
3400// box's public address. publicAddr is a func so the address is read only for
3401// a box that asked for the door: off EC2, readPublicIP can block for its
3402// whole timeout. An empty address still passes through, and the edge logs
3403// that the door is off for want of one.
3404func edgeDoorOptions(getenv func(string) string, publicAddr func() string) []edge.RunOption {
3405 p := strings.TrimSpace(getenv("CARLOS_EDGE_DOOR"))
3406 if p == "" {
3407 return nil
3408 }
3409 return []edge.RunOption{edge.WithEdgeDoor(p, publicAddr())}
3410}
3411
3412// ReadPublicIP is readPublicIP for callers outside this package — the relay
3413// advertises the box's Elastic IP as its relay address and learns it exactly
3414// as the edge does (relay spec §3, "Listeners"). Same contract: "" means
3415// unknown, never an error.
3416func ReadPublicIP() string { return readPublicIP() }
3417
3418// readPlacementRegion best-effort reads this box's AWS region from EC2
3419// IMDSv2 (token PUT, then the placement/region leaf) — the same idiom and
3420// the same "" on any failure contract as readPublicIP: not on EC2, IMDS
3421// disabled, or a metadata-endpoint hiccup all read as "unknown" here, and
3422// the caller (edgesFileRegion) treats that identically to an unset
3423// CARLOS_AWS_REGION override — an edge with neither is simply unarmed for
3424// steering, not an error.
3425func readPlacementRegion() string {
3426 client := &http.Client{Timeout: 2 * time.Second}
3427 token := imdsToken(client)
3428 if token == "" {
3429 return ""
3430 }
3431 region, status := imdsLeaf(client, token, "/latest/meta-data/placement/region")
3432 if status != http.StatusOK {
3433 return ""
3434 }
3435 return region
3436}
3437
3438// ReadPlacementRegion is readPlacementRegion for callers outside this
3439// package. The relay names its region in every heartbeat and checks it
3440// against each app's region list, and it must be the SAME string the edges
3441// file uses — hence the same probe, with the caller applying the same
3442// CARLOS_AWS_REGION override edgesFileRegion applies.
3443func ReadPlacementRegion() string { return readPlacementRegion() }
3444
3445// domainsLoop runs the domains reconcile pass forever on the STATUS tick's
3446// cadence (5min + jitter), not the adopt loop's hour: the console's
3447// custom-domain guidance promises "Checks run every few minutes", and a DNS
3448// fix that takes up to an hour to turn green would read as broken. Thin and
3449// untested by convention (like adoptLoop/statusLoop) — everything with
3450// behavior lives in domainsTickConfig and domains.Reconcile. Missing config
3451// disables the loop with one log line, the same inert-until-configured
3452// shape as its siblings.
3453func domainsLoop(reg *registry.Registry, acmeCacheFlag string, hostClaim bool) {
3454 ctx := context.Background()
3455 // The inert gate is checked BEFORE readPublicIP: an unconfigured
3456 // deployment must make no IMDS call at all (a 2s link-local probe on
3457 // every agent start, on boxes that aren't even on EC2, for a feature
3458 // that is switched off). Inert means inert — no network, not just no
3459 // writes. domainsTickConfig re-checks the same variable; that's the
3460 // authoritative gate, this is the "don't pay for it first" one.
3461 if strings.TrimSpace(os.Getenv("PLATFORM_APPS_DOMAIN")) == "" {
3462 log.Printf("agent: domains loop disabled: PLATFORM_APPS_DOMAIN is not set")
3463 return
3464 }
3465 cfg, envIPs, keyErr := domainsTickConfig(os.Getenv, readPublicIP())
3466 cfg.HostClaim = hostClaim
3467 if cfg.AppsDomain == "" {
3468 log.Printf("agent: domains loop disabled: PLATFORM_APPS_DOMAIN is not set")
3469 return
3470 }
3471 if acmeCacheFlag != "" {
3472 cfg.AcmeDir = acmeCacheFlag
3473 }
3474 store, err := openDeployStore(ctx)
3475 if err != nil {
3476 log.Printf("agent: domains loop disabled: %v", err)
3477 return
3478 }
3479 // The edge set is folded PER PASS, here rather than in
3480 // domainsTickConfig, because it needs the store this line above just
3481 // opened — and because a set read once at arm is the bug this replaces:
3482 // under latency steering a correctly-pointed customer domain resolves to
3483 // whichever region's edge is nearest, which a single box's boot-frozen
3484 // PLATFORM_EDGE_IPS reads as "elsewhere". envIPs stays in the union, so
3485 // a deployment whose bucket is unreadable degrades to exactly the old
3486 // behaviour and never below it. The bound matters: a hung store must
3487 // cost this pass 30s, not the loop forever.
3488 cfg.EdgeIPs = func() []string {
3489 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
3490 defer cancel()
3491 return edgesfile.EdgeIPs(ctx, store, envIPs, time.Now())
3492 }
3493 log.Printf("agent: domains loop armed (5min + jitter), apps domain %s, configured edge IPs %v (the live set is folded from every box's edges file each pass) — health stamps every tick; claims reconcile only while this box holds the housekeeper lease", cfg.AppsDomain, envIPs)
3494 // One line, and it must name the REAL cause: a typo'd key and an absent
3495 // one are different problems with different fixes, and this is the only
3496 // place an operator learns that purging is not happening.
3497 if keyErr != nil {
3498 log.Printf("agent: domains loop auto-purge disabled: %v — trashed apps will be hidden but never destroyed", keyErr)
3499 } else if cfg.ReleasePub == nil {
3500 log.Printf("agent: domains loop auto-purge disabled: CARLOS_RELEASE_PUBKEY is not set — trashed apps will be hidden but never destroyed")
3501 }
3502 duty := domainsDuty(func(ctx context.Context) error {
3503 return domains.Reconcile(ctx, store, reg, cfg)
3504 })
3505 fleetBox := shouldSkipPublisherElection(os.Getenv)
3506 delay := time.Minute
3507 for {
3508 time.Sleep(delay)
3509 delay = 5*time.Minute + rand.N(time.Minute)
3510 // The deployment-global half, gated; then the per-box half on EVERY
3511 // platform box (#303): DNS and TLS health stamp this box's own
3512 // registry from this box's own view, no lease involved, ONCE per
3513 // tick — after the duties so a host section 3 just minted is
3514 // stamped in the same tick it appeared. Never on a fleet box: it
3515 // never contends for the lease for the same reason (housekeeperNever),
3516 // and it publishes no edges file, so it has no standing to judge the
3517 // deployment's DNS. (The old reason — that its EdgeIPs fall back to
3518 // its own address — no longer holds now that the set is folded from
3519 // every box's file; the exclusion stands on the publication rule
3520 // instead, which is the durable one.)
3521 runHousekeeperDuties(housekeeperGate, duty)
3522 if !fleetBox {
3523 if err := domains.StampHealth(ctx, reg, cfg, nil); err != nil {
3524 log.Printf("agent: domains health: %v", err)
3525 }
3526 }
3527 }
3528}
3529
3530// domainsDuty wraps one domains reconcile pass as a housekeeper duty. The
3531// pass is deployment-wide state — materializing routes from claims, minting
3532// platform-domain and canary hosts, purging trashed apps — with one correct
3533// writer, and since multi-region (spec 2026-08-20) every platform box in a
3534// deployment carries PLATFORM_APPS_DOMAIN (the per-account wildcards and the
3535// steering probe need it), so an ungated loop would be N reconcilers racing
3536// over the same records. Gating on the lease is what makes the env safe to
3537// set fleet-wide. (The DNS/TLS health stamps are NOT in this pass: they are
3538// per-box, domains.StampHealth, run by domainsLoop on every box — #303.) Single-box
3539// deployments are unchanged: the one box always holds the lease. A box that
3540// never contends (a fleet box) never runs the pass — which is right, the
3541// pass is a platform-box job.
3542func domainsDuty(pass func(context.Context) error) housekeeperDuty {
3543 return housekeeperDuty{"domains reconcile", func(ctx context.Context) error {
3544 if pass == nil {
3545 return nil
3546 }
3547 return pass(ctx)
3548 }}
3549}
3550
3551// instancesLoop runs the instance-provisioning reconcile pass forever on the
3552// same cadence as domainsLoop (5min + jitter, first tick after 1 minute) —
3553// turning bucket instance records into registry routes for THIS agent's
3554// region/pool scope (provision.Reconcile). Unlike domainsLoop there is NO
3555// apps-domain-style inert gate: an unset CARLOS_REGION/CARLOS_POOLS is a
3556// legal, still-armed state (see provisionTickConfig and
3557// provision.Reconcile's doc comment — a box with no region deliberately
3558// still serves ""-scoped records), so this loop arms whenever the deploy
3559// store opens, full stop. Missing deployment config disables the loop with
3560// one log line, the same "inert until configured" shape adoptLoop and
3561// domainsLoop use for their own missing config. Per-pass errors are logged;
3562// the loop lives on.
3563//
3564// Also runs ReconcileFleetBindings (customer-fleets plan, Task 6) on the same
3565// cadence, platform boxes only: turns every fleet's published
3566// fleets/*/bindings/*.json object into this platform box's own
3567// registry.Route.FleetBox — see fleetbind.go's doc comment for the
3568// fleet-placement/corrupt-object contract. Deliberately homed HERE rather
3569// than domainsLoop (an earlier pass rode domainsLoop's ticker, since it
3570// already had a store+reg in scope) — that coupled binding materialization
3571// to PLATFORM_APPS_DOMAIN being set, which has nothing to do with fleets;
3572// this loop has no such gate and shares the same store already opened above.
3573func instancesLoop(reg *registry.Registry, writes hostwrite.Config, bs *buildStates, configure ...func(*provision.ReconcileConfig)) {
3574 ctx := context.Background()
3575 store, err := openDeployStore(ctx)
3576 if err != nil {
3577 log.Printf("agent: instances loop disabled: %v", err)
3578 return
3579 }
3580 cfg := configuredClaimProvision(provisionTickConfig(os.Getenv), configure)
3581 backfill := newClaimBackfill(writes, store, cfg.ReleasePub, os.Getenv("CARLOS_DEPLOYMENT_BUCKET"))
3582 log.Printf("agent: instances loop armed (5min + jitter), region %q pools %v", cfg.Region, cfg.Pools)
3583 // Platform-only: a fleet box (CARLOS_FLEET set) never reads other
3584 // fleets' bindings — it only ever WRITES its own (fleetbind.go's
3585 // fleetBindingHook/publishResidentFleetBindings, wired in RunAgent).
3586 // Reusing shouldSkipPublisherElection rather than re-testing the env var
3587 // here: both questions are exactly "is this box a fleet box", so one
3588 // predicate, already covered by TestShouldSkipPublisherElection, answers
3589 // both.
3590 isFleetBox := shouldSkipPublisherElection(os.Getenv)
3591 // An offline move's participants keep their reports current between the
3592 // passes below (finding 7); idle, it reads nothing.
3593 go offlineReassertLoop(ctx, store)
3594 delay := time.Minute
3595 for {
3596 time.Sleep(delay)
3597 delay = 5*time.Minute + rand.N(time.Minute)
3598 if err := backfill(ctx); err != nil {
3599 log.Printf("agent: claim backfill: %v", err)
3600 }
3601 if err := reconcileSerialized(ctx, store, reg, cfg); err != nil {
3602 log.Printf("agent: instances pass: %v", err)
3603 }
3604 // Every reconcile refreshes the build-state copies (settings spec
3605 // §4.2), this 5-minute one too: the fast leg above runs only on a
3606 // serial move.
3607 if bs != nil {
3608 bs.refreshAll(ctx)
3609 }
3610 telemetryPass(ctx, store, reg)
3611 plansPass(ctx, store, reg, os.Getenv)
3612 runMovesPass(ctx, store)
3613 runRunsPass(ctx)
3614 deploymentRoutes := edgesfile.DeploymentRoutes(ctx, store, time.Now())
3615 if err := domains.ReconcileAliases(ctx, store, reg, time.Now, deploymentRoutes, &aliasClaimCache); err != nil {
3616 log.Printf("agent: instances alias pass: %v", err)
3617 }
3618 if !isFleetBox {
3619 if err := ReconcileFleetBindings(ctx, store, reg); err != nil {
3620 log.Printf("agent: fleet bindings reconcile: %v", err)
3621 }
3622 }
3623 }
3624}
3625
3626// offlineReassertLoop runs the armed moves worker's offlineReassert every
3627// offlineReassertEvery plus up to 5 s of jitter, for the life of the agent:
3628// while this box takes part in an offline move, its source and sealer reports
3629// stay current for the doors whatever the serial does. Thin and untested by
3630// convention, like instancesLoop; the tick is offlineReassert.
3631func offlineReassertLoop(ctx context.Context, s release.Store) {
3632 for {
3633 time.Sleep(offlineReassertEvery + rand.N(5*time.Second))
3634 if w := armedMovesWorker.Load(); w != nil {
3635 w.offlineReassert(ctx, s)
3636 }
3637 }
3638}
3639
3640// retentionLoop runs the release-retention pass forever on the slow
3641// reconcile cadence (5min + jitter, first tick after 1 minute) — never the
3642// 2s fast pass: pruning is not urgent, and a policy written now is enforced
3643// within minutes, which is the promise the CLI makes ("boxes prune within
3644// ~10 minutes"). Thin and untested by convention (like domainsLoop /
3645// instancesLoop); everything with behavior lives in retentionPass. Missing
3646// deployment config, a missing/unparseable release pubkey, or no box
3647// reports to read (retentionReports: the status store, or on a sole
3648// housekeeper with none this box's own build-state report) disables the
3649// loop with one log line — fail closed, domainsLoop's auto-purge shape: a
3650// box that cannot verify pointers, or cannot learn which builds the build
3651// states name (settings spec rev 11 §4.4), must not delete anything.
3652//
3653// Since the multi-region writer map (spec
3654// 2026-08-20-multi-region-ingress-design.md, Part 1) the pass itself is a
3655// HOUSEKEEPER DUTY: it runs only on the one box holding the deployment's
3656// housekeeper lease. releases/** is deployment-global state, and N boxes
3657// pruning it concurrently is the bug class the spec refuses to have. Every
3658// other box still arms this loop and still ticks — it just prunes nothing,
3659// silently, until the day it wins the lease (housekeeper.go). On a
3660// single-box deployment the holder is always this box, so the behaviour is
3661// unchanged.
3662func retentionLoop(reg *registry.Registry, buildStates func() (map[string]status.BuildStatePins, error)) {
3663 ctx := context.Background()
3664 // Platform boxes only (spec F6). Retention prunes the PLATFORM's release
3665 // objects — a customer's box scheduling deletes of them was already wrong
3666 // under direct S3, and through the store door it is refused outright (the
3667 // door's write/delete ACL scopes a fleet to fleets/<fleet>/**), so arming
3668 // here would buy nothing but an hourly loop of 404s. Reuses
3669 // shouldSkipPublisherElection because the question is exactly "is this a
3670 // fleet box", the same one instancesLoop asks — and it is deliberately the
3671 // bare CARLOS_FLEET test, not resolveFleetIdentity: a fleet box whose
3672 // LABEL is misconfigured is still a fleet box, and must still not prune.
3673 if shouldSkipPublisherElection(os.Getenv) {
3674 log.Printf("agent: retention loop is a platform-box job — skipped on a fleet box (spec F6)")
3675 return
3676 }
3677 store, err := openDeployStore(ctx)
3678 if err != nil {
3679 log.Printf("agent: retention loop disabled: %v", err)
3680 return
3681 }
3682 raw := strings.TrimSpace(os.Getenv("CARLOS_RELEASE_PUBKEY"))
3683 if raw == "" {
3684 log.Printf("agent: retention loop disabled: CARLOS_RELEASE_PUBKEY is not set — retention policies will not be enforced by this box")
3685 return
3686 }
3687 pub, err := release.ParsePublicKey(raw, "CARLOS_RELEASE_PUBKEY")
3688 if err != nil {
3689 log.Printf("agent: retention loop disabled: %v — retention policies will not be enforced by this box", err)
3690 return
3691 }
3692 // The box reports (settings spec rev 11 §4.4: the build-state builds
3693 // every box names) live in the status store, which is not always the
3694 // deployment store above (CARLOS_STATUS_BUCKET wins); a sole
3695 // housekeeper with none reads its own (retentionReports). Without them
3696 // no app can be pruned safely, so the loop does not arm.
3697 reports, err := retentionReports(ctx, os.Getenv, store, buildStates)
3698 if err != nil {
3699 log.Printf("agent: retention loop disabled: the box reports cannot be read (%v). retention policies will not be enforced by this box", err)
3700 return
3701 }
3702 if reports.local != nil {
3703 log.Printf("agent: retention: no status store; as the sole housekeeper this box reads its own build-state report")
3704 }
3705 log.Printf("agent: retention loop armed (5min + jitter) — prunes only while this box holds the housekeeper lease")
3706 // The pass runs under the HOLDING's context, not this loop's: a sweep
3707 // that is halfway through its deletes when renewal fails must stop, not
3708 // finish against a lock another box now owns (housekeeper.go).
3709 prune := housekeeperDuty{"retention pass", func(ctx context.Context) error {
3710 return retentionPass(ctx, store, reports, pub, reg)
3711 }}
3712 delay := time.Minute
3713 for {
3714 time.Sleep(delay)
3715 delay = 5*time.Minute + rand.N(time.Minute)
3716 runHousekeeperDuties(housekeeperGate, prune)
3717 }
3718}
3719
3720// stampedUnitNames is the metrics sweep's one registry read: every distinct
3721// bespoke unit stamped on a route (Route.Unit, the #110 column — an
3722// operator's console.service, tito's titogo-canary@<tenant>.service), which
3723// the box's cgroup walk cannot discover by name convention.
3724//
3725// Called once per 60s tick from metricsLoop's probe, so it is deliberately a
3726// plain List + filter with no cache: the registry is a local SQLite file, one
3727// read a minute is nothing next to the pass loops already reading it, and a
3728// cache would be one more thing to invalidate when an operator stamps a unit.
3729//
3730// A read failure returns nil — this tick's stamped units are lost, the sweep
3731// and the agent are not. The names are returned sorted so the slice handed
3732// down each tick is stable (nothing below depends on the order; a stable one
3733// just makes a log or a test diff mean something).
3734func stampedUnitNames(reg *registry.Registry) []string {
3735 routes, err := reg.List()
3736 if err != nil {
3737 log.Printf("agent: metrics: read stamped units: %v", err)
3738 return nil
3739 }
3740 seen := make(map[string]bool, len(routes))
3741 var out []string
3742 for _, rt := range routes {
3743 if rt.Unit == "" || seen[rt.Unit] {
3744 continue
3745 }
3746 seen[rt.Unit] = true
3747 out = append(out, rt.Unit)
3748 }
3749 sort.Strings(out)
3750 return out
3751}
3752
3753// resolveStatusBucket applies the status/deployment bucket defaulting
3754// documented in this file's package comment: an explicit CARLOS_STATUS_BUCKET
3755// always wins, and when it's unset the status loop falls back to whatever
3756// CARLOS_DEPLOYMENT_BUCKET is set to, so the common single-bucket deployment
3757// is one env var instead of two that must be kept in sync by hand. Returns
3758// "" when neither is set — the loop stays disabled, same as before this fix.
3759// Deliberately does NOT fall back to CARLOS_DEPLOYMENT_DIR: the status page
3760// has no directory-store serving path (the edge/console read status/ out of
3761// an S3 bucket, not a local directory), so a DIR-based deployment simply
3762// doesn't get this fallback — see the "HONEST LIMIT" note in the package
3763// comment above.
3764//
3765// Takes a getenv func rather than reading os.Getenv itself so the defaulting
3766// logic is testable without mutating process-wide environment variables —
3767// t.Setenv serializes tests within a package and the host package already
3768// has other env-reading tests; a plain map/func seam avoids that coupling
3769// entirely.
3770func resolveStatusBucket(getenv func(string) string) string {
3771 return ResolveStatusBucket(getenv)
3772}
3773
3774// ResolveStatusBucket is resolveStatusBucket for callers outside this package
3775// — `carlos status -live` reads the boxes an agent wrote, and it must land on
3776// the SAME bucket the agent's status loop writes to, including the
3777// CARLOS_DEPLOYMENT_BUCKET fallback. Exported as a second name over one
3778// implementation rather than renamed at every internal call site, so the
3779// defaulting rule documented above stays in exactly one function.
3780func ResolveStatusBucket(getenv func(string) string) string {
3781 if b := getenv("CARLOS_STATUS_BUCKET"); b != "" {
3782 return b
3783 }
3784 return getenv("CARLOS_DEPLOYMENT_BUCKET")
3785}
3786
3787// shouldSkipPublisherElection reports whether this tick must never contend
3788// for the cross-box publish lease at all — true exactly when CARLOS_FLEET is
3789// set (Task 5, customer-fleets plan): a fleet box self-reports its own Box
3790// snapshot every tick (statusTick's WriteBox call is unconditional) but must
3791// never win — or even attempt — the election that would have it merge and
3792// publish status/fleet.json, which is the platform's own boxes' shared feed.
3793// Checked BEFORE lease.Acquire is ever called, not after: a fleet box never
3794// makes the network call, not merely one that's discarded on return.
3795func shouldSkipPublisherElection(getenv func(string) string) bool {
3796 return getenv("CARLOS_FLEET") != ""
3797}
3798
3799// upstreamSource holds the live edge's UpstreamStates method, published by
3800// RunAgent's edge-ready callback and read by statusLoop's probe set — the same
3801// shape tunnelStateVal uses for tunnel state (tunnel_client.go), and for the
3802// same reason: the status loop starts BEFORE the edge exists (RunAgent spawns
3803// the loops, then builds and serves the edge), so the two cannot be wired by
3804// passing a handle down.
3805//
3806// A func rather than the *edge.Edge itself, so what this package's tests
3807// inject is the same one value production sets — no fake Edge, no probe
3808// goroutine a test would have no way to stop.
3809var upstreamSource atomic.Value // func() map[string][]status.UpstreamState
3810
3811// geoIPSource is the edge's GeoIP state for the heartbeat, published from
3812// the edge-ready callback like upstreamSource (geoip spec §5.5).
3813var geoIPSource atomic.Value // func() edge.GeoIPState
3814
3815// geoIPAlertSource is the edge's geoip reason code for the box alert rule
3816// (geoip spec §10.4), published from the same ready callback.
3817var geoIPAlertSource atomic.Value // func(time.Time) (float64, bool)
3818
3819// geoIPAlertReading is the rule's reading, or nil before the edge is ready
3820// (no data never fires and never clears).
3821func geoIPAlertReading(now time.Time) *float64 {
3822 fn, _ := geoIPAlertSource.Load().(func(time.Time) (float64, bool))
3823 if fn == nil {
3824 return nil
3825 }
3826 v, known := fn(now)
3827 if !known {
3828 return nil // no data never fires and never clears
3829 }
3830 return &v
3831}
3832
3833// geoIPState reads it, or nil before the edge is ready.
3834func geoIPState() *status.GeoIPState {
3835 fn, _ := geoIPSource.Load().(func() edge.GeoIPState)
3836 if fn == nil {
3837 return nil
3838 }
3839 s := fn()
3840 return &status.GeoIPState{Mode: s.Mode, SHA256: s.SHA256, Type: s.Type, Error: s.Error}
3841}
3842
3843// tunnelEnvComplete is tunnelLoop's own arming rule: all three set.
3844func tunnelEnvComplete(getenv func(string) string) bool {
3845 return getenv("CARLOS_TUNNEL_URL") != "" && getenv("CARLOS_TUNNEL_TOKEN") != "" && getenv("CARLOS_TUNNEL_NAME") != ""
3846}
3847
3848// setUpstreamSource publishes (or, with nil, withdraws) the probe-state
3849// source. Called once, from the edge-ready callback.
3850func setUpstreamSource(fn func() map[string][]status.UpstreamState) {
3851 if fn == nil {
3852 upstreamSource.Store(func() map[string][]status.UpstreamState { return nil })
3853 return
3854 }
3855 upstreamSource.Store(fn)
3856}
3857
3858// upstreamStates reads the current probe table for the heartbeat, or nil when
3859// no source has been published yet.
3860//
3861// nil is the honest answer for every box shape that has no edge probe table to
3862// report: `carlos edge` run without the agent, an agent whose ready callback
3863// hasn't fired yet, and the overwhelmingly common case of a box with no
3864// multi-upstream route at all. All three publish no upstreams key rather than
3865// an empty object (status.CollectBox's omitempty rule) — a box that isn't
3866// probing and a box running code that predates this field stay
3867// indistinguishable on the wire.
3868func upstreamStates() map[string][]status.UpstreamState {
3869 fn, _ := upstreamSource.Load().(func() map[string][]status.UpstreamState)
3870 if fn == nil {
3871 return nil
3872 }
3873 return fn()
3874}
3875
3876// statusLoop runs the fleet status page's 5-minute tick, forever, once armed
3877// — see the package doc comment above for the full env contract. Missing
3878// CARLOS_STATUS_BUCKET disables the loop with one log line, the same
3879// "a box with no status store is still a valid deployment" shape adoptLoop
3880// uses for its own missing config.
3881//
3882// Each tick: gather fresh probes (small OS-reading funcs in probes.go, none
3883// of which this loop's own tests call — statusTick below takes a
3884// status.Probes value, so tests inject fixtures instead of touching /proc or
3885// a syscall), then contend for the cross-box status-lease. Acquire succeeding
3886// means this box is the one publisher for this tick: statusTick both writes
3887// this box's own snapshot AND merges+publishes the fleet-wide feed, and the
3888// lease is released again right after (held only for the publish, not across
3889// the 5-minute gap — the next tick, possibly on a different box, re-acquires
3890// fresh). A *lease.HeldError means another box already won this tick's
3891// election: completely normal contention, not a failure, so it's swallowed
3892// without a log line — every box hits this on every tick it doesn't win, and
3893// logging that every 5 minutes forever would bury the log in noise. Any
3894// OTHER acquire error (a real S3/credential problem) is logged, same as any
3895// other tick failure.
3896func statusLoop(reg *registry.Registry, svc *activator.Service, buildStates func() (map[string]status.BuildStatePins, error)) {
3897 ctx := context.Background()
3898 // bucket is read for the lease key alone (lease.Acquire below), which a
3899 // fleet box never reaches: shouldSkipPublisherElection short-circuits the
3900 // election before it, and a fleet box is the only shape that gets a store
3901 // with no bucket behind it (openStatusStore). Every other box resolves a
3902 // real bucket here exactly as before.
3903 bucket := resolveStatusBucket(os.Getenv)
3904 prefix := resolveStatusPrefix(os.Getenv)
3905 region := os.Getenv("CARLOS_STATUS_REGION")
3906 boxRegion := os.Getenv("STATUS_BOX_REGION")
3907 skipElection := shouldSkipPublisherElection(os.Getenv)
3908
3909 // Fleet boxes heartbeat under the operator-chosen channel label
3910 // (fleet_boxes.label — the identity the edge dials via
3911 // edge.fleetChannelFor), NEVER the public status page's hashed
3912 // pseudonym (status.Label(machineID)): that anonymization exists so a
3913 // customer's machine id is never recoverable from the PUBLIC
3914 // status/fleet.json feed, but a fleet box's own snapshot never reaches
3915 // that feed at all (skipElection, above) — it only ever lands at
3916 // fleets/<fleet>/status/boxes/<label>.json, an owner-facing path no
3917 // public page reads, so the anonymization rationale simply doesn't
3918 // apply here. An invalid identity degrades to the hashed label rather
3919 // than disabling the heartbeat outright (RunAgent already refuses,
3920 // loudly, to arm the binding-publish machinery that actually needs the
3921 // label to be dial-able).
3922 fleetLabel := ""
3923 if identity, err := resolveFleetIdentity(os.Getenv); err == nil {
3924 fleetLabel = identity.Label
3925 } else if !errors.Is(err, errNotFleetBox) {
3926 log.Printf("agent: %v — heartbeat will publish under the hashed pseudonym instead", err)
3927 }
3928
3929 s, target, err := openStatusStore(ctx, os.Getenv)
3930 if err != nil {
3931 if errors.Is(err, errNoStatusStore) {
3932 log.Printf("agent: status loop disabled: neither CARLOS_STATUS_BUCKET nor CARLOS_DEPLOYMENT_BUCKET is set (CARLOS_DEPLOYMENT_DIR deployments don't publish status yet — set CARLOS_STATUS_BUCKET explicitly)")
3933 } else {
3934 log.Printf("agent: status loop disabled: %v", err)
3935 }
3936 return
3937 }
3938 // store is what statusTick actually reads/writes through: identical to s
3939 // (a pure passthrough) unless CARLOS_STATUS_PREFIX is set, in which case
3940 // every key it touches is rebased under that prefix — see
3941 // release.NewPrefixStore and the package doc comment's
3942 // CARLOS_STATUS_PREFIX entry.
3943 store := release.NewPrefixStore(s, prefix)
3944 // The effective destination is worth a log line on its own: it's easy to
3945 // set CARLOS_DEPLOYMENT_BUCKET and CARLOS_STATUS_BUCKET to different
3946 // values by accident (or on purpose) and end up with the agent writing
3947 // status to a bucket the edge/console never look at — this line is the
3948 // one place an operator can confirm which bucket (or, on a bucketless
3949 // fleet box, which door identity) actually won.
3950 log.Printf("agent: status loop armed (%s + jitter), publishing to %s", status.SnapshotPeriod, target)
3951
3952 // canaryStore/canaryPub resolve the release deployment config
3953 // countExpiredCanaries needs to answer §5's "how many canaries on this
3954 // box have expired" — the SAME env pair adoptLoop reads
3955 // (CARLOS_DEPLOYMENT_BUCKET/_DIR + CARLOS_RELEASE_PUBKEY), not
3956 // necessarily the status bucket above, which may deliberately be a
3957 // separate one. Either can come back unusable: no deployment store
3958 // configured, or no pubkey set. That is a normal, valid box shape (the
3959 // same "routes managed by hand" case adoptLoop's own doc comment
3960 // names) — it just means countExpiredCanaries can't answer and reports
3961 // nil, same as every other unmeasured probe in this loop. Resolved once
3962 // here, not re-read every tick, mirroring adoptLoop's own one-time
3963 // resolution.
3964 //
3965 // The error is logged rather than swallowed: "no deployment store" is a
3966 // valid box shape, but since the store can now also fail to build for a
3967 // REAL misconfiguration (a fleet box with no bucket and no tunnel env —
3968 // openFleetDoorStore's loud error), silence here would turn that into an
3969 // unexplained permanently-nil ExpiredCanaries probe.
3970 canaryStore, err := openDeployStore(ctx)
3971 if err != nil {
3972 log.Printf("agent: status loop: expired-canary probe unavailable: %v", err)
3973 }
3974 canaryPub, _ := release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY")
3975
3976 // status.SnapshotPeriod IS this cadence — the const is named for the tick
3977 // below, and every reader that judges a box stale (the published boxes
3978 // table, the console's private host detail, `carlos status -live`) measures
3979 // against it, so changing it here must move them together rather than
3980 // leaving them judging a heartbeat that arrives on a different beat.
3981 // Jittered on the same shape as adoptLoop's delay: every box on the same
3982 // cadence would all hit the lease and the bucket in the same second
3983 // otherwise.
3984 delay := time.Minute
3985 for {
3986 select {
3987 case <-time.After(delay):
3988 case <-statusNudge:
3989 // A rollout transition (flag on only): publish within seconds,
3990 // at most every two, so a busy rollout cannot hammer the bucket.
3991 time.Sleep(2 * time.Second)
3992 }
3993 delay = status.SnapshotPeriod + rand.N(time.Minute)
3994
3995 // Routes are listed here (not left to statusTick's own reg.List())
3996 // because the Listening probe needs to know which sockets are
3997 // systemd-socket-activated BEFORE it's built — see
3998 // wrapListeningProbe's doc comment. statusTick lists again for
3999 // itself since it stays store/lease-agnostic and self-contained;
4000 // one extra List() per 5-minute tick is not worth threading routes
4001 // through the pure function's signature for.
4002 routes, err := reg.List()
4003 if err != nil {
4004 log.Printf("agent: status tick: list routes: %v", err)
4005 continue
4006 }
4007
4008 memTotal, memAvail := readMemBytes()
4009 p := status.Probes{
4010 MachineID: readMachineID(),
4011 InstanceID: cachedInstanceID(),
4012 Region: boxRegion,
4013 MemFrac: readMemFrac(),
4014 DiskFrac: readDiskFrac(),
4015 Listening: wrapListeningProbe(unitBackedSockets(routes), statusSocketProbe(routes, svc)),
4016 // One systemctl call per 5-minute tick, asking the only question
4017 // that catches an enabled-but-never-armed adoption timer — the
4018 // flagship's silent no-adopt state, which nothing on any page
4019 // could see before this.
4020 AdoptRunning: adoptTimerActive(),
4021 // A second systemctl call per tick, asking the same question of
4022 // the config-sync trigger: an enabled-but-unstarted
4023 // carlos-adopt.path is a box whose unit-backed config silently
4024 // waits for the hourly timer (issue #44).
4025 ConfigPathRunning: configPathActive(),
4026 // A third, for the restart trigger: an enabled-but-unstarted
4027 // carlos-restart.path is a box where `carlos restart` answers
4028 // "requested" and nothing cycles until the hourly pass (issue #113).
4029 RestartPathRunning: restartPathActive(),
4030 // Whether this box's own alert rules have anywhere to deliver,
4031 // read from the env the metrics loop's sinks are built from
4032 // (discussion #192: an unset webhook was silent for weeks).
4033 Alerts: alertDeliveryProbe(os.Getenv, nil),
4034 // ExpiredCanaries answers spec §5 for this box's own routes; see
4035 // countExpiredCanaries and the canaryStore/canaryPub comment above.
4036 ExpiredCanaries: countExpiredCanaries(ctx, routes, canaryStore, canaryPub, time.Now()),
4037 // LagSecs stays nil: litestream's per-DB replication lag isn't
4038 // surfaced up to the agent process anywhere yet — the
4039 // replication machinery (internal/lease's Replicator) lives
4040 // inside the activator, one level down, and doesn't currently
4041 // export a "seconds since last successful sync" reading this
4042 // loop could poll. Publishing a guess (0? the box's own clock
4043 // skew?) would be worse than the honest null CollectBox already
4044 // knows how to carry — wiring this up is a seam for later, not
4045 // a gap to paper over now.
4046 MemTotalBytes: memTotal,
4047 MemAvailBytes: memAvail,
4048 Load1: readLoad1(),
4049 DiskFreeBytes: readDiskFreeBytes(),
4050 Release: version.Build,
4051 Platform: runtime.GOOS + "-" + runtime.GOARCH,
4052 // The invalid-value line is newSystemLeg's to say, once at
4053 // arming; this tick publishes the channel the box actually
4054 // follows, five-minutely, and has nowhere to say it again.
4055 SystemChannel: boxSystemChannel(os.Getenv),
4056 System: systemStateProbe(release.OptDir()),
4057 Tunnel: tunnelState(),
4058 GeoIP: geoIPState(),
4059 // The edge's probe table, read WHOLE every tick (never merged
4060 // with the last one), so a route this box has stopped probing
4061 // leaves no ghost behind in the next snapshot. nil until the
4062 // edge-ready callback publishes the source, and on any box that
4063 // probes nothing — both publish no key at all.
4064 Upstreams: upstreamStates(),
4065 // One stat (two with a WAL) per database route per tick, which
4066 // is what `carlos instances list` and `carlos routes` show as DB.
4067 DBBytes: dbFileBytes,
4068 // Layout is the armed pass's classification of every
4069 // instance-kind row, or, with no activator on this box, the
4070 // rows judged by path shape under Mode "unmanaged" — see
4071 // layoutStatusReport's own doc comment for why a box with no
4072 // pass still says something rather than nothing.
4073 Layout: layoutStatusReport(routes),
4074 }
4075 // The hosts this box is refusing because it cannot provide their
4076 // config (amadan #213): private Metrics only, nothing public.
4077 claimReport(&p)
4078 p.ConfigRefusals, p.ConfigRefusalsTotal = configRefusalProbe()
4079 p.ConfigGuard = configGuardProbe()
4080 // Builds the root pass rolled back (D1), as the metrics loop's
4081 // refusal watch last read them. Private Metrics only: hosts.
4082 p.ReleaseRefusals = releaseRefusalProbe()
4083 // The rollout scheduler's report (zero-downtime spec §8.4): nil
4084 // with CARLOS_SWAP off. Private Metrics only.
4085 p.Rollout = rolloutStatusProbe()
4086 // Hosts held for a dev proof (D3): private Metrics only.
4087 p.ProofWaits = proofWaitProbe()
4088 // The build-state builds of the instances this box hosts (settings
4089 // spec rev 11 §4.4), for the housekeeper's retention. A failed
4090 // collection is reported as such: the housekeeper prunes nothing
4091 // while it is fresh. Private Metrics only: hosts.
4092 if pins, err := buildStates(); err != nil {
4093 log.Printf("agent: status tick: build states: %v", err)
4094 p.BuildStatesErr = err.Error()
4095 } else {
4096 p.BuildStates = pins
4097 }
4098 // The update-settings protocol this box enforces, and root's last
4099 // decision per instance (settings spec §4.3). Private Metrics only.
4100 fillUpdateProbes(&p, reg, release.OptDir())
4101
4102 // A fleet box (CARLOS_FLEET set) never gets here as a contender: it
4103 // self-reports only, never attempting the election at all — no
4104 // lease.Acquire call, no BuildFleet, no PublishFleet. See
4105 // shouldSkipPublisherElection and the package doc comment's
4106 // CARLOS_FLEET entry.
4107 var held *lease.Held
4108 publish := false
4109 if !skipElection {
4110 // path.Join (an S3 key, not a filesystem path — matches
4111 // activator.go's Config.hibernatePrefix convention) rather than a
4112 // bare "+" concat: guards against a trailing-slash
4113 // CARLOS_STATUS_PREFIX producing a doubled "//status-lease" key.
4114 var acqErr error
4115 held, acqErr = lease.Acquire(ctx, bucket, path.Join(prefix, "status-lease"), region)
4116 publish = acqErr == nil
4117 if acqErr != nil {
4118 var heldErr *lease.HeldError
4119 if !errors.As(acqErr, &heldErr) {
4120 log.Printf("agent: status tick: lease acquire: %v", acqErr)
4121 }
4122 }
4123 }
4124
4125 // groupStatusTick is statusTick for every box but one its region's
4126 // group launched, which publishes only under its group reference
4127 // (census ruling region-ha-2b, boxreference.go).
4128 if err := groupStatusTick(ctx, store, reg, p, publish, fleetLabel, os.Getenv); err != nil {
4129 log.Printf("agent: status tick: %v", err)
4130 }
4131
4132 if publish {
4133 if err := held.Release(ctx); err != nil {
4134 log.Printf("agent: status tick: lease release: %v", err)
4135 }
4136 }
4137 }
4138}
4139
4140// statusTick is the fleet status page's pure per-tick work, extracted out of
4141// statusLoop so it needs no systemd, S3, or timers to test: given an
4142// already-built Store (DirStore in tests, the real S3Store in production)
4143// and an already-decided publish flag (statusLoop's lease result; a fixed
4144// true/false in tests), it does exactly what one tick does and nothing else.
4145// The box's own snapshot is written unconditionally — every box self-reports
4146// every tick regardless of who's publishing — and BuildFleet+PublishFleet
4147// only run when publish is true, which for a DirStore-backed dev/test caller
4148// with no lease in play at all just means "always true, no election needed"
4149// (a lone box is its own only publisher).
4150//
4151// labelOverride, when non-empty, replaces the box's hashed-pseudonym label
4152// (status.CollectBox derives one from p.MachineID by default) with the given
4153// value before WriteBox — statusLoop passes a fleet box's channel label
4154// (resolveFleetIdentity) here so its heartbeat lands at
4155// fleets/<fleet>/status/boxes/<label>.json under the SAME label the edge
4156// dials, not an opaque pseudonym only the public status page ever wanted.
4157// "" (every platform box) leaves CollectBox's own label untouched.
4158func statusTick(ctx context.Context, s release.Store, reg *registry.Registry, p status.Probes, publish bool, labelOverride string) error {
4159 routes, err := reg.List()
4160 if err != nil {
4161 return err
4162 }
4163 now := time.Now()
4164
4165 // Aliases carries route_aliases' DNS/cert stamps (Task 2/5) up to the box
4166 // snapshot (issue #151) — read fresh alongside routes above rather than
4167 // folded into the caller's Probes in statusLoop, since reg is already in
4168 // scope here and a failed read must not sink the whole snapshot: logged
4169 // and left nil (CollectBox already treats nil the same as "no alias rows
4170 // exist"), same best-effort shape as every other probe in this loop.
4171 if aliases, err := reg.ListAliases(); err != nil {
4172 log.Printf("agent: status tick: list aliases: %v", err)
4173 } else {
4174 p.Aliases = aliases
4175 }
4176
4177 box := status.CollectBox(routes, p, now)
4178 if labelOverride != "" {
4179 box.Label = labelOverride
4180 }
4181 if err := status.WriteBox(ctx, s, box); err != nil {
4182 return err
4183 }
4184 if !publish {
4185 return nil
4186 }
4187
4188 // status.SnapshotPeriod is this loop's own tick cadence, named once in the
4189 // status package so the console's private host table and `carlos status
4190 // -live` measure staleness against the same period this publish does.
4191 fleet, err := status.BuildFleet(ctx, s, routes, now, status.SnapshotPeriod)
4192 if err != nil {
4193 return err
4194 }
4195 return status.PublishFleet(ctx, s, fleet)
4196}
4197
4198// makeRestartRoute builds the adopt pass's restart callback, dispatched on
4199// what owns the route's process. THREE arms, in this order:
4200//
4201// 1. a route that NAMES its own unit (registry.Route.Unit — an operator-run
4202// bespoke service, console.service) is try-restarted by that name, and
4203// returns. This arm comes first because it is the only one that can be
4204// right when it applies: such a route is addr-backed, so the activator
4205// owns no child to Bounce and there is no carlos-app@<host> unit to
4206// try-restart — both of the arms below would run and cycle nothing,
4207// which is issue #110 (adoption stamped Version, and therefore
4208// X-Carlos-Version, behind a console still serving the old build).
4209// 2. a carlos-app@<host> unit tenant is try-restarted by host (a no-op when
4210// the unit isn't running — hibernators are never woken by this).
4211// 3. an exec child is Bounced through the activator, the same graceful
4212// cycle config sync uses. systemctl cannot reach a child of THIS
4213// process, which is why exec-backed warm instances used to keep serving
4214// the old binary until they happened to hibernate: the symlink was
4215// repointed under them and nothing cycled the process.
4216//
4217// Hosts the registry cannot answer for take arm 2: try-restart of a
4218// nonexistent unit is a no-op, whereas a bounce of a process we might not own
4219// is not.
4220//
4221// A bounce that cycled NOTHING falls through to try-restart rather than
4222// returning: an exec-backed route with an ACTIVE carlos-app@<host> unit is a
4223// real state — an unstamped unit tenant, the one adopt_advice.go warns about
4224// (issue #14) — and try-restart is what has always cycled it. The same
4225// fallthrough covers a box with no activator armed at all. When the route
4226// really is a sleeping exec child the cost is one try-restart of a stopped or
4227// absent unit, which starts nothing and at worst logs once: this callback
4228// only fires when the version actually moves (adoptHost early-returns on
4229// cur == ver), so it is one line per version change per host, never per tick.
4230//
4231// The named-unit arm takes NO part in that fallthrough, in either direction.
4232// It does not fall through (a route that names its unit has said where its
4233// process is; trying the other two afterwards could only cycle something
4234// else), and nothing falls through INTO it (the other arms are reached only
4235// when Unit is empty, i.e. when there is no name to try).
4236//
4237// No path can WAKE anything, which is what lets release.Pass hand this
4238// callback hibernating routes (RestartHibernating): Bounce returns false
4239// without starting anything when the host is not awake, and try-restart is
4240// defined as a no-op on a unit that is not running — including the named one,
4241// which is why it is try-restart and not restart (a console an operator has
4242// deliberately stopped stays stopped).
4243//
4244// ERRORS split by arm, and the split is the point:
4245//
4246// Arms 2 and 3 log and return nil, unchanged: a failed try-restart of a
4247// carlos-app@<host> unit must not poison the adopted version for the other
4248// hosts in the pass (release/adopt.go's rule), and it cannot be told apart
4249// from the ordinary case anyway — a sleeping exec child reaches arm 2 under
4250// RestartHibernating precisely because there is nothing there to cycle.
4251// Fire-and-forget is what makes that safe.
4252//
4253// Arm 1 RETURNS the failure, because for a named unit the ordinary case is
4254// not indistinguishable: try-restart of a unit that is loaded and stopped
4255// exits 0, so a non-zero exit means the name is wrong (exit 5, "not loaded")
4256// or the service would not come back. Both are "the new build is NOT
4257// running", and swallowing them would hand adoptHost a success it did not
4258// have: it would stamp Version, the edge would serve the new
4259// X-Carlos-Version, and `carlos deploy` would report live over the old
4260// process — issue #110 again, now with the stamp actively lying rather than
4261// merely unattended. Returned, adoptHost rolls the symlink back to the
4262// version that IS running, skips SetVersions, prints the failure, and the
4263// next pass retries; the header stays honest on the old version throughout.
4264func makeRestartRoute(reg *registry.Registry, tryRestart func(host string), tryRestartUnit func(rt registry.Route) error, bounce func(host string) bool) func(host string) error {
4265 return func(host string) error {
4266 rt, err := reg.Get(host)
4267 if err != nil {
4268 log.Printf("adopt: registry lookup %s: %v (restarting as a unit)", host, err)
4269 }
4270 if err == nil && rt != nil && rt.Unit != "" {
4271 return tryRestartUnit(*rt)
4272 }
4273 if err == nil && rt != nil && rt.Backing != registry.BackingUnit {
4274 if bounce != nil && bounce(host) {
4275 log.Printf("adopt: bounced exec instance %s onto its new version", host)
4276 return nil
4277 }
4278 }
4279 tryRestart(host)
4280 return nil
4281 }
4282}
4283
4284// makeStartRoute builds the adopt pass's rollback-start callback
4285// (release.WithStartUnit): after a restart that FAILED has had its symlink
4286// rolled back, start the unit the route NAMES on the restored version —
4287// `systemctl start`, not try-restart, because the premise of that path is
4288// that the unit is down (a try-restart there is a no-op and the box stays
4289// serving nothing).
4290//
4291// ONE arm, deliberately, where makeRestartRoute has three: only a route that
4292// names its own unit has anything for this callback to start. The other two
4293// process owners never route here at all — their restart arms swallow their
4294// failures and return nil, so the rollback path this callback belongs to is
4295// unreachable for them — and a start of carlos-app@<host> or an activator
4296// child from here could WAKE a sleeper, which no adopt path may do. A route
4297// with an empty Unit, an unknown host, or a registry that will not answer
4298// therefore starts NOTHING and returns nil: for those shapes "nothing was
4299// stopped by the failed restart" is the only answer this callback can stand
4300// over. (The converged-daemon rollback is caps.Restore's business, not
4301// this callback's — see release.WithStartUnit.)
4302//
4303// This cannot wake anything, twice over. By ARGUMENT: it fires only after a
4304// try-restart of the named unit returned non-zero, and try-restart exits 0
4305// on a loaded-but-stopped unit — so a deliberately stopped service never
4306// produces the error that reaches this, and a unit whose name resolves to
4307// nothing gives systemctl start nothing to bring up either. And by
4308// STRUCTURE: a Hibernate-flagged route starts nothing here, full stop. That
4309// guard costs nothing real — provision refuses the addr-with-hibernate
4310// shape, so no legitimate named-unit route hibernates, and the only
4311// hibernating+Unit row is the misconfiguration restart.go already documents
4312// — and it turns "an adopt path cannot wake a sleeper" back into a property
4313// of the code's shape rather than of an argument about systemd exit codes.
4314func makeStartRoute(reg *registry.Registry, startUnit func(unit string) error) func(host string) error {
4315 return func(host string) error {
4316 rt, err := reg.Get(host)
4317 if err != nil || rt == nil || rt.Hibernate || rt.Unit == "" {
4318 return nil
4319 }
4320 return startUnit(rt.Unit)
4321 }
4322}
4323
4324// systemctlTryRestart is makeRestartRoute's production unit path: cycle the
4325// route's carlos-app@ unit iff it's running, best-effort. A missing or
4326// stopped unit is not this pass's problem — try-restart is by definition a
4327// no-op on one, which is what makes makeRestartRoute's fallthrough safe:
4328// an exec route whose child is asleep returns false from bounce (nothing to
4329// cycle) and lands here under RestartHibernating, where try-restart finds no
4330// carlos-app@ unit for it and does nothing. Sleeping hosts DO reach this
4331// line; they simply have nothing here to restart, and pick the new version
4332// up on their next wake.
4333func systemctlTryRestart(host string) {
4334 out, err := exec.Command("systemctl", "try-restart", activator.UnitName(host)).CombinedOutput()
4335 if err != nil {
4336 log.Printf("adopt: try-restart carlos-app@%s: %v: %s (continuing)", host, err, strings.TrimSpace(string(out)))
4337 }
4338}
4339
4340// systemctlTryRestartUnit is makeRestartRoute's NAMED-unit path: cycle the
4341// unit the route itself names, iff it's running. Same try-restart-not-restart
4342// reasoning as systemctlTryRestart above — a service an operator has stopped
4343// stays stopped, and try-restart on a loaded-but-stopped unit exits 0.
4344//
4345// It LOGS AND RETURNS, where its carlos-app@ sibling only logs, and the
4346// asymmetry is deliberate: this name came from an operator rather than from
4347// activator.UnitName, so a non-zero exit here means the stamp is wrong or the
4348// service will not come back — never the routine "nothing was running"
4349// systemctlTryRestart has to tolerate. The journal line is kept as well as
4350// the error because the error's own destination is a pass that prints one
4351// summary line per host; the exit status and systemctl's own words belong in
4352// the journal where an operator greps for the unit name.
4353//
4354// The unit reaches exec.Command as ONE argv element, so there is no shell for
4355// a name to escape into; the shape rail lives at the write (registry's
4356// ValidUnitName), where a typo can still be reported to the person who typed
4357// it.
4358func systemctlTryRestartUnit(unit string) error {
4359 out, err := exec.Command("systemctl", "try-restart", unit).CombinedOutput()
4360 if err != nil {
4361 log.Printf("adopt: try-restart %s: %v: %s", unit, err, strings.TrimSpace(string(out)))
4362 return fmt.Errorf("try-restart %s: %w: %s", unit, err, strings.TrimSpace(string(out)))
4363 }
4364 return nil
4365}
4366
4367// unitOnly adapts a unit-name restart to makeRestartRoute's route-aware arm.
4368func unitOnly(f func(unit string) error) func(rt registry.Route) error {
4369 return func(rt registry.Route) error { return f(rt.Unit) }
4370}
4371
4372// checkedUnitRestart is the named-unit arm with D1's settle check (spec
4373// §2.2-§2.3). The binding is verified first: the check judges a restart only
4374// when the unit's ExecStart runs THIS route's live link. Anything it cannot
4375// establish (show denied, wrong binding, unit stopped) falls back to today's
4376// plain try-restart and returns that command's own error.
4377//
4378// settled runs only when the settle check passed (never for a unit stopped
4379// during it, a failed settle or an unchecked fallback). The agent loops pass
4380// nil: the sandboxed agent cannot hash another unit's executable, and spec
4381// §4.5 has root record the healthy history.
4382func agentCheckedUnitRestart(sys unitSystem, optDir string, getenv func(string) string, settled func(rt registry.Route) error) func(rt registry.Route) error {
4383 return func(rt registry.Route) error {
4384 before, err := sys.Show(rt.Unit)
4385 if err != nil {
4386 log.Printf("adopt: %s: cannot read %s (%v); restart not checked", rt.Host, rt.Unit, err)
4387 return sys.TryRestart(rt.Unit)
4388 }
4389 link := filepath.Join(optDir, "live", rt.Host)
4390 if before.ExecPath != link {
4391 log.Printf("adopt: binding: %s runs %s, but the unit stamp is on %s; restart not checked", rt.Unit, before.ExecPath, rt.Host)
4392 return sys.TryRestart(rt.Unit)
4393 }
4394 if before.ActiveState != "active" {
4395 return sys.TryRestart(rt.Unit) // stopped by an operator: stays a no-op
4396 }
4397 want, err := filepath.EvalSymlinks(link)
4398 if err != nil {
4399 return sys.TryRestart(rt.Unit)
4400 }
4401 if err := sys.TryRestart(rt.Unit); err != nil {
4402 return err
4403 }
4404 timeout := unitSettleTimeout(getenv)
4405 err = settleUnit(sys, rt.Unit, before.InvocationID, settleOpts{
4406 Timeout: timeout, Poll: 500 * time.Millisecond, Hold: 5 * time.Second,
4407 WantDir: filepath.Dir(want), Ready: routeReady(rt, 2*time.Second),
4408 Now: time.Now, Sleep: time.Sleep})
4409 if errors.Is(err, errUnitStopped) {
4410 // An operator stopped it between the read above and now. Not a
4411 // verdict on the build, and a rollback restart would undo their
4412 // stop: same as a unit that was stopped before the restart.
4413 log.Printf("adopt: %s: %s was stopped during settle; not judged, not restarted", rt.Host, rt.Unit)
4414 return nil
4415 }
4416 if err == nil {
4417 log.Printf("adopt: %s: %s settled on %s", rt.Host, rt.Unit, filepath.Dir(want))
4418 // D3 (proofobserve.go): a build that passed the settle check is a
4419 // build this box has run well.
4420 if settled != nil {
4421 err = settled(rt)
4422 }
4423 }
4424 return err
4425 }
4426}
4427
4428// rollbackUnitRestart is D1's rollback restart (release.WithRollbackRestart):
4429// the live link already names the restored build; replace whatever is
4430// running (a deadline failure can leave the rejected build up) and check
4431// the restored build comes up too.
4432func agentRollbackUnitRestart(reg *registry.Registry, sys unitSystem, optDir string, getenv func(string) string) func(host string) error {
4433 return func(host string) error {
4434 rt, err := reg.Get(host)
4435 if err != nil || rt == nil || rt.Unit == "" {
4436 return nil
4437 }
4438 before, _ := sys.Show(rt.Unit)
4439 if err := sys.ResetFailed(rt.Unit); err != nil {
4440 log.Printf("adopt: %s: reset-failed %s: %v (continuing)", host, rt.Unit, err)
4441 }
4442 if err := sys.Restart(rt.Unit); err != nil {
4443 return err
4444 }
4445 want, err := filepath.EvalSymlinks(filepath.Join(optDir, "live", host))
4446 if err != nil {
4447 return err
4448 }
4449 err = settleUnit(sys, rt.Unit, before.InvocationID, settleOpts{
4450 Timeout: unitSettleTimeout(getenv), Poll: 500 * time.Millisecond, Hold: 5 * time.Second,
4451 WantDir: filepath.Dir(want), Ready: routeReady(*rt, 2*time.Second),
4452 Now: time.Now, Sleep: time.Sleep})
4453 if errors.Is(err, errUnitStopped) {
4454 // The restored build is what the link names; a stop an operator
4455 // made while it settled is theirs to keep, not a failed recovery.
4456 log.Printf("adopt: %s: %s was stopped during the rollback's settle; not judged, not restarted", host, rt.Unit)
4457 return nil
4458 }
4459 if err != nil {
4460 return fmt.Errorf("rollback did not come up either: %w", err)
4461 }
4462 return nil
4463 }
4464}
4465
4466// systemctlStartUnit is makeStartRoute's production path: bring the named
4467// unit up after a rolled-back restart left it down. START, not try-restart —
4468// try-restart is a no-op on a stopped unit, which on this path would leave
4469// the box with a correct symlink and no process, and hand the NEXT pass's
4470// try-restart an exit 0 to stamp a never-run build with. Idempotent where it
4471// matters: `systemctl start` of an already-running unit exits 0 and does
4472// nothing, so a service something else restored first is not cycled twice.
4473//
4474// LOGS AND RETURNS like systemctlTryRestartUnit, and for the same reason:
4475// this name came from an operator, its failure means the box is serving
4476// nothing, and the caller's one summary line per host must say so while the
4477// exit status and systemd's own words stay in the journal under the unit
4478// name an operator would grep for. Same no-shell shape too: the unit reaches
4479// exec.Command as ONE argv element.
4480func systemctlStartUnit(unit string) error {
4481 out, err := exec.Command("systemctl", "start", unit).CombinedOutput()
4482 if err != nil {
4483 log.Printf("adopt: start %s: %v: %s", unit, err, strings.TrimSpace(string(out)))
4484 return fmt.Errorf("start %s: %w: %s", unit, err, strings.TrimSpace(string(out)))
4485 }
4486 return nil
4487}
4488
4489// rootRestartRoute is the root adopt pass's restart callback (RunAdopt, out of
4490// carlos-adopt.service): no bounce. That process has no activator and no
4491// children of its own — the exec instances belong to the agent, and another
4492// process's children are not its to stop. With bounce nil the dispatch's arm
4493// 3 is unreachable and every route it used to try-restart by host still is,
4494// byte-identically; what it gains is arm 1.
4495//
4496// Arm 1 has to be HERE, not only on the agent's in-process loops, and the
4497// reason is the sandbox. carlos-edge.service runs as User=carlos with
4498// ProtectSystem=strict and ReadWritePaths=/data /run/carlos, so the agent
4499// cannot write /opt/carlos: on a real box its pass reports the stage as
4500// blocked and relays to this verb (carlos-stage.path), which does the
4501// staging, the repoint, the restart and the version stamp. A restart arm
4502// wired only into the agent would therefore be a restart arm the consoles
4503// never reach.
4504//
4505// settled is checkedUnitRestart's hook for a named unit whose settle check
4506// passed (D3: recordSettled, which adds the build to the healthy history);
4507// nil records nothing.
4508// bounceFn adapts an activator to makeRestartRoute's bounce parameter,
4509// nil-safe: a box with no activator armed gets a nil callback, which
4510// makeRestartRoute reads as "nothing running to cycle" rather than falling
4511// back to a systemctl path that could not work for an exec route anyway.
4512func bounceFn(svc *activator.Service) func(host string) bool {
4513 if svc == nil {
4514 return nil
4515 }
4516 return svc.Bounce
4517}
4518
4519// adoptBounceFn is bounceFn for an ADOPT pass's restart dispatch
4520// (makeRestartRoute on adoptLoop and fastConvergeLoop): the same bounce,
4521// except that a child already exec'd from the build live/<host> names is left
4522// running — see activator.BounceAdopted for the interleaving that produces
4523// one. `carlos restart` keeps bounceFn: a restart request cycles whatever is
4524// running.
4525func adoptBounceFn(svc *activator.Service) func(host string) bool {
4526 if svc == nil {
4527 return nil
4528 }
4529 if m := armedRollout(); m != nil {
4530 return m.Adopted
4531 }
4532 if rolloutUnavailable() != "" {
4533 return nil // no adopt bounce while managed starts are refused
4534 }
4535 if h := heldHolds(); h != nil {
4536 // Declined with failure records: a host a record holds keeps
4537 // running rather than stop into a refused wake (final review F6).
4538 return h.skipHost("adopt bounce", svc.BounceAdopted)
4539 }
4540 return svc.BounceAdopted
4541}
4542
4543// wakeFn adapts an activator to cycleWorkers' wake parameter, nil-safe like
4544// bounceFn: a box with no activator armed gets a nil callback, which
4545// cycleWorkers reads as "nothing to wake" rather than panicking on a nil
4546// activator.
4547func wakeFn(svc *activator.Service) func(registry.Route) error {
4548 if svc == nil {
4549 return nil
4550 }
4551 return svc.WakeWorker
4552}
4553
4554// openDeployStore opens the release store the box adopts from:
4555// CARLOS_DEPLOYMENT_BUCKET (S3), CARLOS_DEPLOYMENT_DIR (a directory — tests
4556// and single-box no-AWS deployments), or — on a fleet box with neither set —
4557// the edge's store door over the tunnel env (spec F5, openFleetDoorStore).
4558// Mirrors cli.go's openStore, but returns the error instead of dying: the
4559// agent degrades, the CLI exits.
4560func openDeployStore(ctx context.Context) (release.Store, error) {
4561 return openDeployStoreEnv(ctx, os.Getenv)
4562}
4563
4564// OpenDeployStore is openDeployStore for callers outside this package. The
4565// relay reads its authority record and writes its heartbeat and metering to
4566// the same store the agent adopts from, chosen by the same rule, so a box
4567// that has moved onto the fleet door does not leave the relay behind on a
4568// bucket it can no longer reach.
4569func OpenDeployStore(ctx context.Context) (release.Store, error) { return openDeployStore(ctx) }
4570
4571// openDeployStoreEnv is openDeployStore with the environment injected — the
4572// same getenv seam resolveStatusBucket already uses in this file, so the
4573// selection rule can be tested without mutating process-wide env.
4574//
4575// Order is the whole contract (spec F5): a bucket or dir still set WINS,
4576// before fleet identity is even consulted, so every platform box and every
4577// grandfathered fleet box that still has bucket credentials keeps exactly
4578// today's store, byte for byte. Only a box that resolves a fleet identity
4579// AND has no bucket/dir at all reaches the door — that is the cutover
4580// switch, per box, no flag day: drop the bucket var and the box is on the
4581// API; put it back and it is off.
4582func openDeployStoreEnv(ctx context.Context, getenv func(string) string) (release.Store, error) {
4583 if b := getenv("CARLOS_DEPLOYMENT_BUCKET"); b != "" {
4584 return release.NewDeploymentS3Store(ctx, b)
4585 }
4586 if d := getenv("CARLOS_DEPLOYMENT_DIR"); d != "" {
4587 ds := release.NewDirStore(d)
4588 ds.EnableChangeIndex()
4589 return ds, nil
4590 }
4591 identity, err := resolveFleetIdentity(getenv)
4592 switch {
4593 case err == nil:
4594 return openFleetDoorStore(getenv, identity)
4595 case !errors.Is(err, errNotFleetBox):
4596 // CARLOS_FLEET is set but the identity is unusable, and there is no
4597 // bucket to fall back to: this box has no store at all. Return the
4598 // identity error itself rather than the generic "no bucket" line —
4599 // the actual fix is the label, and the generic message would send an
4600 // operator hunting for a bucket var that is deliberately absent.
4601 return nil, err
4602 }
4603 return nil, errNoDeployStore
4604}
4605
4606// openFleetDoorStore builds the fleetapi client a bucketless fleet box adopts
4607// through: the edge origin it already tunnels to (CARLOS_TUNNEL_URL), its
4608// channel identity ("<fleet>/<label>" — the same string tunnel_client.go
4609// registers with and edge.fleetChannelFor dials), and the tunnel bearer token.
4610//
4611// Missing tunnel env here is an ERROR, never a silent no-store: the caller
4612// established that this box is a fleet box with no bucket, so there is no
4613// other store it could possibly be meant to use, and a quiet nil would leave
4614// every loop in this file reporting "not configured" about a box that is in
4615// fact configured — just with one variable missing.
4616func openFleetDoorStore(getenv func(string) string, identity FleetIdentity) (release.Store, error) {
4617 base := strings.TrimSpace(getenv("CARLOS_TUNNEL_URL"))
4618 token := strings.TrimSpace(getenv("CARLOS_TUNNEL_TOKEN"))
4619 name := identity.Fleet + "/" + identity.Label
4620 if base == "" || token == "" {
4621 return nil, fmt.Errorf("fleet box %s has no deployment bucket and cannot reach the store door: both CARLOS_TUNNEL_URL and CARLOS_TUNNEL_TOKEN must be set", name)
4622 }
4623 return fleetapi.New(base, name, token)
4624}
4625
4626// resolveStatusPrefix returns the prefix every key statusLoop and metricsLoop
4627// touch is rebased under (release.NewPrefixStore at both call sites).
4628//
4629// An explicit CARLOS_STATUS_PREFIX always wins, unchanged, on every box. What
4630// this adds is the DOOR-branch default: a fleet box publishing through the
4631// store door with an EMPTY prefix would otherwise write status/boxes/... and
4632// metrics/boxes/... at the shared platform root, and the door's write ACL
4633// admits only fleets/<own fleet>/** — every heartbeat and every flush would
4634// 404, forever, on a box that looks correctly configured. So on that branch
4635// alone the prefix defaults to "fleets/<fleet>/", which is exactly the layout
4636// the console's fleet page already reads.
4637//
4638// S3-path semantics are deliberately untouched: an empty prefix on a
4639// bucket-backed box is the platform box's shared root, which is correct
4640// there, and a bucket-backed fleet box keeps whatever the operator set (the
4641// customer-fleets rollout sets CARLOS_STATUS_PREFIX=fleets/<name>/ by hand).
4642func resolveStatusPrefix(getenv func(string) string) string {
4643 if p := getenv("CARLOS_STATUS_PREFIX"); p != "" {
4644 return p
4645 }
4646 if resolveStatusBucket(getenv) != "" {
4647 return ""
4648 }
4649 identity, err := resolveFleetIdentity(getenv)
4650 if err != nil {
4651 return ""
4652 }
4653 return "fleets/" + identity.Fleet + "/"
4654}
4655
4656// errNoStatusStore is openStatusStore's "this deployment simply hasn't
4657// configured one" sentinel — distinct from a real construction failure, so
4658// each loop can keep its own long-standing disabled-log wording for the
4659// unconfigured case while sharing one selection rule.
4660var errNoStatusStore = errors.New("no status store configured")
4661
4662// openStatusStore resolves the store statusLoop and metricsLoop publish
4663// through, plus a human phrase naming where that is for their armed log
4664// lines. Same F5 rule as openDeployStoreEnv, read through
4665// resolveStatusBucket: an explicit CARLOS_STATUS_BUCKET (else the
4666// CARLOS_DEPLOYMENT_BUCKET fallback) WINS and keeps direct S3; only a fleet
4667// box with neither set speaks the door.
4668//
4669// LEDGER NOTE (spec F5): the door serves the DEPLOYMENT store. A fleet box
4670// that sets CARLOS_STATUS_BUCKET to some OTHER bucket therefore keeps direct
4671// S3 for status/metrics while its releases may come through the door — that
4672// is the bucket-set-wins rule doing its job, not a bug: an operator who named
4673// a bucket explicitly gets that bucket.
4674//
4675// The CARLOS_STATUS_PREFIX rebase both loops apply (release.NewPrefixStore)
4676// is deliberately NOT applied here — it stays at each call site, unchanged,
4677// and wraps whichever store this returns.
4678func openStatusStore(ctx context.Context, getenv func(string) string) (release.Store, string, error) {
4679 if bucket := resolveStatusBucket(getenv); bucket != "" {
4680 // The fallback is the deployment bucket, so open it as one; status/ and
4681 // metrics/ are not change scopes and bump nothing.
4682 s, err := release.NewDeploymentS3Store(ctx, bucket)
4683 if err != nil {
4684 return nil, "", err
4685 }
4686 return s, "bucket " + bucket, nil
4687 }
4688 identity, err := resolveFleetIdentity(getenv)
4689 switch {
4690 case err == nil:
4691 s, derr := openFleetDoorStore(getenv, identity)
4692 if derr != nil {
4693 return nil, "", derr
4694 }
4695 return s, "the fleet store door as " + identity.Fleet + "/" + identity.Label, nil
4696 case !errors.Is(err, errNotFleetBox):
4697 return nil, "", err
4698 }
4699 return nil, "", errNoStatusStore
4700}
4701
4702// RunAdopt is the `carlos adopt` verb: one immediate adoption pass — the
4703// carlos-adopt.timer unit's ExecStart, and an operator's "converge now".
4704// Runs as root (see carlos-adopt.service): this is the ONLY place the
4705// UNIT-BACKED half of config sync (appconfig.Sync) may run, because it
4706// writes /etc/carlos/apps and restarts units — the same filesystem-write
4707// shape release staging needs root for, and exactly what the sandboxed
4708// carlos-edge.service's in-process adopt loop (ProtectSystem=strict,
4709// ReadWritePaths=/data /run/carlos only) cannot do. The EXEC-BACKED half
4710// (appconfig.SyncExec) runs elsewhere, in the agent's in-process
4711// configSyncLoop: it needs neither root nor /etc access, because it only
4712// ever writes under /data and bounces its own children.
4713//
4714// -config-only runs the UNIT-BACKED CONFIG half ALONE: appconfig.Sync plus
4715// its restarts, no release.Pass, no canary nag, no backing advice. That is
4716// what carlos-config-sync.service execs, and it exists because the trigger
4717// behind it is a per-config-write path unit, not the hourly timer. Letting
4718// that trigger run a full adoption pass would mean one tenant's `carlos env
4719// set` could stage and cut over a DIFFERENT app's pending release within
4720// seconds of an unrelated write, and would fire the canary-TTL nag per
4721// config write instead of hourly — the "hourly root pass alone" split
4722// release/adopt.go and the nag comment below both depend on.
4723//
4724// -restart-only runs the RESTART-REQUEST half alone (issue #113): compare each
4725// app's restart generation against this box's marks and cycle the ones that
4726// moved, nothing else. That is what carlos-restart.service execs, and it is
4727// separated from -config-only for the same reason -config-only is separated
4728// from the full pass: the trigger behind it is a per-request path unit, and a
4729// request to cycle one app's processes must not stage a release, run the
4730// capability converge, or nag about canaries. It deliberately needs NO release
4731// key and NO secrets key — restarting verifies no signature and renders no
4732// config — so a box missing either still honours it.
4733func RunAdopt(args []string) { runAdopt(args, productionClaimFactory) }
4734
4735func runAdopt(args []string, buildClaims ClaimFactory) {
4736 runAdoptWithEnsureSystemUnits(args, buildClaims, func() {
4737 ensureSystemUnits("/etc", filepath.Join(systemRootDir(release.OptDir()), systemLockFile), productionSysOps().systemctl, systemctlProbe, log.Printf)
4738 })
4739}
4740
4741// runAdoptWithEnsureSystemUnits keeps the root-only unit effect injectable;
4742// the rest of adopt startup, including its fatal registry diagnostics, is real.
4743func runAdoptWithEnsureSystemUnits(args []string, buildClaims ClaimFactory, ensureUnits func(), sources ...adoptRootSource) {
4744 open := func() (rootstate.Store, error) { return rootstate.Open(rootReleasePrivate, true, false) }
4745 probe := rootparticipation.OpenForUpdate
4746 if len(sources) == 1 {
4747 open = sources[0].OpenRelease
4748 probe = sources[0].Census
4749 }
4750 err := withClaimRootRelease(context.Background(), hostwrite.Enabled(os.Getenv), open, func() error {
4751 return runAdoptUnderRootRelease(args, buildClaims, ensureUnits, probe, productionRootUpdates(dataDir(), probe))
4752 }, probe)
4753 if err != nil {
4754 log.Fatalf("adopt: %v", err)
4755 }
4756}
4757
4758func runAdoptUnderRootRelease(args []string, buildClaims ClaimFactory, ensureUnits func(), census func() (rootparticipation.Ledger, bool, error), updateProviders ...release.RootUpdateOpener) error {
4759 reapStagingWrites(release.OptDir(), dataDir(), "/etc", log.Printf)
4760 if os.Geteuid() == 0 {
4761 if err := repairOffcloudCredentials("/etc/carlos/credentials"); err != nil {
4762 log.Printf("ALERT: adopt: offcloud credentials: %v", err)
4763 }
4764 }
4765 claimCtx, err := claimEntryContext(context.Background(), os.Getenv, openDeployStore)
4766 if err != nil {
4767 log.Fatalf("adopt: host claims: %v", err)
4768 }
4769 claimsEnabled := hostwrite.Enabled(claimEntryEnv(claimCtx))
4770 var claims *ClaimWiring
4771 if !claimsEnabled {
4772 // Keep the flag-off startup effects in their old order:
4773 // disabled wiring, flags, pass lock, unit repair, registry open.
4774 claims, err = newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
4775 if err != nil {
4776 log.Fatalf("adopt: host claims: %v", err)
4777 }
4778 }
4779 fs := flag.NewFlagSet("adopt", flag.ExitOnError)
4780 regPath := fs.String("registry", filepath.Join(dataDir(), "registry.db"), "registry database")
4781 configOnly := fs.Bool("config-only", false, "sync unit-backed config and restart changed routes only: no release adoption, no canary nag")
4782 stageOnly := fs.Bool("stage-only", false, "adopt releases without the canary nag: the agent-requested pass (carlos-stage.service)")
4783 restartOnly := fs.Bool("restart-only", false, "cycle the routes of apps whose restart generation moved, and nothing else: the agent-requested pass (carlos-restart.service)")
4784 fs.Usage = cliflag.Usage(fs, "")
4785 _ = fs.Parse(args)
4786 var reg *registry.Registry
4787 if claimsEnabled {
4788 reg, err = openServiceRegistry(*regPath)
4789 if err != nil {
4790 if hint := registryOpenHint(*regPath, err); hint != "" {
4791 log.Fatalf("adopt: open registry %s: %v\nadopt: %s", *regPath, err, hint)
4792 }
4793 log.Fatalf("adopt: open registry %s: %v", *regPath, err)
4794 }
4795 defer reg.Close()
4796 claimCtx = withClaimRuntimeResources(claimCtx, ClaimRuntimeResources{Mode: "adopt", Registry: reg, DataDir: dataDir(), OptDir: release.OptDir(), UnitsDir: "/etc/systemd/system"})
4797 claims, err = newClaimWiring(claimCtx, claimEntryEnv(claimCtx), buildClaims)
4798 if err != nil {
4799 log.Fatalf("adopt: host claims: %v", err)
4800 }
4801 }
4802 defer func() {
4803 ctx, cancel := context.WithTimeout(context.Background(), hostbox.RequestLimit)
4804 defer cancel()
4805 if err := claims.close(ctx); err != nil {
4806 log.Printf("adopt: host claims shutdown: %v", err)
4807 }
4808 }()
4809
4810 // One exclusive lock for the whole pass, in every mode, because the four
4811 // units that exec this verb can start in the same second and the sweep
4812 // below is the first thing a pass does that a concurrent pass cannot
4813 // safely repeat (lockAdoptPass).
4814 //
4815 // A lock that cannot be taken is NOT fatal: it drops the sweep and runs
4816 // the pass exactly as this verb ran before the sweep existed. Every other
4817 // half is idempotent against a concurrent pass — that is why they were
4818 // unserialized for a year — so the honest response to "I cannot serialize"
4819 // is to skip the one half that needs it, not to stop converging the box.
4820 unlockPass, passLockErr := lockAdoptPass(dataDir())
4821 if passLockErr != nil {
4822 log.Printf("adopt: pass lock %s unavailable (%v) — continuing unserialized, staged-version sweep skipped", adoptPassLockPath(dataDir()), passLockErr)
4823 } else {
4824 defer unlockPass()
4825 }
4826 // Spec §7 belt 2: arm the self-update relay if the last hand roll did
4827 // not. Before the fatal gates below, and never fatal itself.
4828 if !*restartOnly {
4829 ensureUnits()
4830 }
4831 // Root keeps SQLite in the service-user child for every adoption mode,
4832 // including the early return for restart-only.
4833 if !claimsEnabled {
4834 reg, err = openServiceRegistry(*regPath)
4835 if err != nil {
4836 if hint := registryOpenHint(*regPath, err); hint != "" {
4837 log.Fatalf("adopt: open registry %s: %v\nadopt: %s", *regPath, err, hint)
4838 }
4839 log.Fatalf("adopt: open registry %s: %v", *regPath, err)
4840 }
4841 defer reg.Close()
4842 }
4843 // The moves protocol floor (moves spec §11): re-judge carlos.prev
4844 // against the armed floor and the held rows on every root pass, in every
4845 // mode, so a raised floor or a new hold reaches the units' choice within
4846 // the hour, or within seconds of a SetHold (its caller touches
4847 // stage-requested). Here, before the fatal gates below, so a pass that
4848 // dies later still judged. Never fatal; one stat on a box with no floor.
4849 restartChecks := productionRootRestartChecks(systemdUnits{}, release.OptDir())
4850 restartChecks.legacy = func() (bool, error) {
4851 if census == nil {
4852 return false, os.ErrPermission
4853 }
4854 ledger, required, err := census()
4855 if ledger.State != nil {
4856 ledger.State.Close()
4857 }
4858 return !required && err == nil, err
4859 }
4860 if err := recoverRootRestarts(claimCtx, systemdUnits{}, release.OptDir(), os.Getenv, restartChecks, claims.Writes); err != nil {
4861 log.Printf("adopt: %v", err)
4862 }
4863 if err := rootJudgePrev(reg); err != nil {
4864 log.Printf("adopt: moves protocol floor: %v", err)
4865 }
4866
4867 // Step 0 (discussion #111): evict the staged versions nothing on this box
4868 // still names, BEFORE anything that can fail because the disk is full.
4869 //
4870 // The placement is the whole point of the fix, not a detail. Below this
4871 // line are three fatal gates — openDeployStore, the release pubkey parse,
4872 // and release.Pass, whose bake clock is written with CreateTemp under
4873 // /data and which log.Fatalf's the pass on error. On the box this is for,
4874 // with 1.2 MB free, a sweep placed after any of them never runs, and the
4875 // box has no way to free itself. It needs only the registry and /opt, so
4876 // it goes as early as having a registry allows. restartPass sits below
4877 // release.Pass's fatality for exactly this reason and says so.
4878 //
4879 // Hourly only: -stage-only is triggered per relayed release, up to once a
4880 // minute per loop, and walking the version store that often to free a
4881 // day's worth of shipping is work for no gain. -config-only and
4882 // -restart-only are narrow triggers that must do their one job.
4883 generationPins := productionRootGenerationPins()
4884 if passLockErr == nil && !*stageOnly && !*configOnly && !*restartOnly {
4885 sweepStagedVersionsPass(reg, generationPins)
4886 }
4887
4888 ctx := context.Background()
4889 store, err := openDeployStore(ctx)
4890 if err != nil {
4891 log.Fatalf("adopt: %v", err)
4892 }
4893 // One callback for both halves below, built once here now that it needs
4894 // the registry — it reads each route's `unit` column to decide what to
4895 // cycle. See rootRestartRoute for why the named-unit arm has to exist on
4896 // THIS pass.
4897 //
4898 // Only the RELEASE pass can actually reach that arm. appconfig.Sync skips
4899 // every route whose Backing is not BackingUnit before it restarts anything
4900 // (an exec-backed route's config is delivered by the agent's in-process
4901 // loop instead), and a unit-backed route is exactly what `carlos route
4902 // -unit` refuses to stamp — so config sync sees no unit-named route by
4903 // construction. It is handed the same callback anyway because there should
4904 // be ONE restart dispatch on this pass, not two that could drift.
4905 //
4906 // settled: D1's settle passing records the build in the healthy history
4907 // (D3, proofobserve.go). Root is the process that can hash the unit's
4908 // executable.
4909 observer := productionProofObserver(release.OptDir())
4910 observer.resolve = restartChecks.resolve
4911 settled := observer.recordSettled(release.RootHealthyDir)
4912 restart := rootRestartRouteWith(reg, systemdUnits{}, release.OptDir(), os.Getenv, settled, restartChecks)
4913 // The restart-request half, and the ONE place it dispatches from —
4914 // including for exec-backed routes, whose children belong to the agent:
4915 // rootRestartRoute's try-restart arm reaches a carlos-app@ unit, and an
4916 // exec child that this pass cannot cycle is cycled by nothing else, which
4917 // is a real limit of the -restart-only path recorded in restartPass.
4918 //
4919 // -restart-only returns here: no config sync, no capability converge, no
4920 // release.Pass, no nag, and — above all — no release public key, which is
4921 // parsed below and is fatal when unusable. A restart request must not fail
4922 // on a key it never needs.
4923 if *restartOnly {
4924 logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {
4925 routes, err := reg.List()
4926 if err != nil {
4927 log.Fatalf("adopt: list routes: %v", err)
4928 }
4929 restartPass(ctx, routes, store, restartMarksPath(dataDir()), restart)
4930 return nil
4931 }))
4932 return nil
4933 }
4934 // Parsed here, before config sync, exactly as the full pass always has
4935 // (a box with an unusable release key must die before it converges
4936 // anything), but SKIPPED entirely for -config-only: that mode never
4937 // verifies a signature, and making a config-sync trigger fatal on a
4938 // release-key problem would fail every config write on a box whose
4939 // releases were already going nowhere.
4940 var pub ed25519.PublicKey
4941 if !*configOnly {
4942 if pub, err = release.ParsePublicKey(os.Getenv("CARLOS_RELEASE_PUBKEY"), "CARLOS_RELEASE_PUBKEY"); err != nil {
4943 log.Fatalf("adopt: %v", err)
4944 }
4945 }
4946 // Config sync is optional — inert until CARLOS_SECRETS_KEY is set, the
4947 // same convention every other optional loop in this package follows. It
4948 // runs before release adoption so a newly adopted route's very first
4949 // start already sees its env file, and a pass carrying both a release
4950 // and a config change never starts the new binary against stale config.
4951 if raw := os.Getenv("CARLOS_SECRETS_KEY"); raw != "" {
4952 secretsPriv, err := appconfig.ParsePrivateKey(raw, "CARLOS_SECRETS_KEY")
4953 if err != nil {
4954 log.Printf("adopt: config sync disabled: %v", err)
4955 } else if err := appconfig.Sync(ctx, reg, store, secretsPriv, appconfig.DefaultEnvDir, restart, configAckFunc(ctx, store, "adopt"), claims.Writes); err != nil {
4956 log.Printf("adopt: config sync: %v", err)
4957 }
4958 }
4959 if *configOnly {
4960 // Everything below is release adoption and its reporting — the half
4961 // carlos-config-sync.service must never reach, however often a
4962 // config write triggers it. The restart half is below for the same
4963 // reason: a config write is not a restart request, and running the
4964 // comparison here would only move it earlier by minutes at the cost of
4965 // firing on a trigger that has nothing to do with it.
4966 return nil
4967 }
4968 // The restart-request half again (issue #113), here as the FLOOR under
4969 // carlos-restart.path: that relay can drop an event that lands while the
4970 // service is mid-run, exactly as the config one can, and a dropped restart
4971 // with no floor would be a request that never happens at all.
4972 //
4973 // BEFORE release.Pass deliberately. That call is fatal on error, so a box
4974 // with an unstageable release would otherwise never reach this line — and
4975 // a wedged instance is most worth cycling precisely on a box where
4976 // something else is already wrong.
4977 logRestartRequestError(productionRestartRequestPass(ctx, store, claims.Writes, func() error {
4978 if routes, err := reg.List(); err != nil {
4979 log.Printf("adopt: list routes for the restart pass: %v", err)
4980 } else {
4981 restartPass(ctx, routes, store, restartMarksPath(dataDir()), restart)
4982 }
4983 return nil
4984 }))
4985
4986 // Step 2 (spec §4d): the grant-drift capability converge, BETWEEN config
4987 // sync and release adoption. Before release.Pass because it repairs the
4988 // shape of what is ALREADY live — a revoked grant, a deleted drop-in, a
4989 // socket unit nobody enabled — and a route that release.Pass then moves
4990 // to a new version converges again inside adoptHost, which is a no-op
4991 // when nothing about the shape changed: same bytes, no write, no reload,
4992 // no restart.
4993 //
4994 // The zero caps.Ops is the production one (real /etc/systemd/system, real
4995 // systemctl, real useradd), for the same reason WithCapabilities below is
4996 // only ever set here: this verb is root out of carlos-adopt.service, and
4997 // the in-process loops in this file cannot write unit configuration at
4998 // all.
4999 //
5000 // Never fatal. A failure here is the registry not listing; adoption of
5001 // releases is a separate job and must still run.
5002 if err := capabilityPass(reg, release.OptDir(), productionClaimCapsOps(caps.Ops{}, restartChecks.allowLegacy)); err != nil {
5003 log.Printf("adopt: capability pass: %v", err)
5004 }
5005
5006 // NagExpiredCanaries(os.Stdout) is set HERE ONLY (2026-08-03 review) — the
5007 // canary-TTL nag (spec §5) belongs to the hourly root pass
5008 // (carlos-adopt.service, this verb's ExecStart), the same split PR #9's
5009 // config sync above uses. adoptLoop and fastConvergeLoop in this same
5010 // package call release.Pass/PassFilter without the nag option, so an
5011 // expired canary nags once an hour from here, never
5012 // every 2-3 seconds from the in-process fast loop.
5013 //
5014 // WithCapabilities(caps.Ops{}) is likewise set HERE ONLY, and the zero Ops
5015 // is the production one (real /etc/systemd/system, real systemctl, real
5016 // useradd). This verb runs as root out of carlos-adopt.service; the
5017 // in-process loops in this same file run inside carlos-edge.service's
5018 // ProtectSystem=strict sandbox and cannot write unit configuration at all,
5019 // which is why adoptLoop and fastPassOpts each carry a comment saying
5020 // never to add this one.
5021 //
5022 // -stage-only drops the nag and keeps everything else. It is what
5023 // carlos-stage.service runs when the agent relays a version it could not
5024 // stage: that request arrives at most once per stageRequestEvery per loop, and the nag
5025 // is an HOURLY visibility feature — firing it per request would flood the
5026 // journal with the same expired-canary lines, which is precisely what the
5027 // option's own doc comment says never to do. The capability converge and
5028 // config sync stay: both are jobs only this root pass can do, and a
5029 // newly staged version wants its unit shape and env file converged with
5030 // it, not an hour later.
5031 //
5032 // WithStartUnit is wired on THIS pass above all: on a hardened box the
5033 // sandboxed agent's stage is blocked and relayed here, so this is the
5034 // pass that actually try-restarts a console's named unit — and therefore
5035 // the pass whose failed restart, rolled back, must start that unit again
5036 // on the restored version rather than leave a correct symlink over no
5037 // process (release.WithStartUnit; without it the NEXT pass's try-restart
5038 // no-ops on the stopped unit, exits 0, and stamps a build that has never
5039 // run). -stage-only keeps it for the same reason it keeps the converge:
5040 // a relayed stage is exactly the request that performs such a restart.
5041 pr := proofRoot{
5042 obs: observer, reg: reg, store: store, bakeDir: dataDir(),
5043 healthyDir: release.RootHealthyDir, policyPath: release.ProofPolicyPath,
5044 readKey: readProvingKey, hourly: !*stageOnly, logf: log.Printf,
5045 }
5046 history, err := configureRootProof(&pr, restartChecks, func() (*release.RootHistoryPass, error) {
5047 return openRootHistory(store, release.OptDir())
5048 })
5049 if err != nil {
5050 log.Printf("adopt: %v", err)
5051 } else if history != nil {
5052 defer history.History.Store.Close()
5053 linkRootHistory(history, restartChecks, release.OptDir(), os.Getenv)
5054 }
5055 recovery := productionNamedUnitRecovery(store, pub, release.OptDir(), claims.Writes)
5056 passOpts := []release.PassOption{release.WithRootGenerationPins(generationPins), release.WithCapabilities(productionClaimCapsOps(caps.Ops{}, restartChecks.allowLegacy)), release.BakeDir(dataDir()), release.WithRestartAttemptSettled(restartChecks.finish),
5057 release.WithRunningVersion(restartChecks.runningVersion),
5058 release.WithSelectedRestarts(
5059 rootSelectedRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, settled, restartChecks, false),
5060 rootSelectedRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, nil, restartChecks, true)),
5061 release.WithStartUnit(rootStartRoute(reg, release.OptDir(), restartChecks)), release.WithHostWrites(claims.Writes),
5062 release.WithDigestRatchet(pointerRatchet(true, release.OptDir(), dataDir(), os.Getenv)),
5063 release.WithRollbackRestart(rollbackUnitRestart(reg, systemdUnits{}, release.OptDir(), os.Getenv, restartChecks)),
5064 // D3's gate: a fresh gate per root pass, so the policy file is read
5065 // at the start of this pass (BeginPass) like every other. Healthy
5066 // history is root-owned; diagnostic refusals still use BakeDir.
5067 release.WithProofGate(release.NewRootProofGate(release.ProofPolicyPath, dataDir())),
5068 // Settings spec §4.3: root alone resolves install requests and
5069 // writes /opt/carlos/consumed/<host>, before it stages.
5070 updatesPassOption(dataDir(), true, updateProviders...)}
5071 passOpts, err = rootAdoptionOptions(passOpts, restartChecks, history, func() release.PassOption { return recovery.passOption(reg, restartChecks, os.Getenv) })
5072 if err != nil {
5073 return err
5074 }
5075 if !*stageOnly {
5076 passOpts = append(passOpts, release.NagExpiredCanaries(os.Stdout))
5077 }
5078 // D3 (proofroot.go) wraps the pass. Before it, record every named-unit
5079 // route's live build that has run well as healthy, so the build live when
5080 // a requirement is switched on is in the history before this pass can
5081 // adopt a proven head over it (final review H4). After it, mint on a
5082 // proving box and, on the hourly pass, say when a requirement is off;
5083 // a route the pass changed is looked at again first.
5084 passErr := pr.around(ctx, func() error {
5085 return release.Pass(ctx, reg, store, pub, release.OptDir(), restart, passOpts...)
5086 })
5087 if passErr != nil {
5088 log.Printf("adopt: %v", passErr)
5089 }
5090
5091 // A move's held stage (moves design §7.3, T7): the sandboxed agent
5092 // cannot write /opt, so a move's target leaves a request under
5093 // /data/moves/held-stage and touches stage-requested. Served here, after
5094 // release adoption, on -stage-only and on the hourly pass (its floor).
5095 // Reads nothing and writes nothing unless a request exists, and none
5096 // exist while CARLOS_MOVES_PHASE1 is off.
5097 heldStagePass(ctx, dataDir(), release.OptDir(), pub, pointerRatchet(true, release.OptDir(), dataDir(), os.Getenv), reg, store, BoxLabel())
5098
5099 // Step 4: the advisories. The two capability advisories that fall out of
5100 // READING manifests (declared-but-ungranted, and capabilities on an
5101 // exec-backed route) are printed by capabilityPass above, where the
5102 // manifests are already in hand; this one needs the box's firewall and
5103 // its passwd database instead, so it lives here with the other
5104 // systemd-probing advisory.
5105 if routes, err := reg.List(); err == nil {
5106 adviseBackingMismatches(routes)
5107 adviseHostRoleGateMissing(routes, lookupUID, nftGateUIDs)
5108 }
5109 return passErr
5110}
5111