carlos / platform Public

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

Plain git — no account needed to clone.

Download

Download this file

1// Package activator is the hibernation state machine: instances hibernate.
2// When the agent is given an S3 bucket (CARLOS_HIBERNATE_BUCKET) it stops
3// relying on systemd to keep instance processes alive and instead runs them
4// as its own children, on demand — "units are for the edge, instances are its
5// children". A request for a host with no live process claims the per-host S3
6// single-writer lease, restores the DB from S3 if the local copy is missing,
7// spawns the instance, and proxies; after an idle window the instance is
8// checkpointed back to S3, the lease released, and the process stopped. State
9// parks in S3; processes exist only while someone's talking.
10//
11// Ported from lchat activator.go (Eleven's activator, issue #550 Part 4 v2),
12// with the mesh (multi-box placement: gossip/HRW, peer proxying) dropped for
13// single-box v1 — `// mesh seam:` comments mark every point where it slots
14// back in. Membership of one: hash picks self, leases are uncontested except
15// against a previous incarnation of ourselves.
16//
17// Sidecar co-spawn (LEGACY, but LIVE — Eleven runs on it today): a host with
18// a sidecar staged (/data/<host>-sidecar.env) gets its `sidecar run` child
19// spawned right after the instance turns healthy, and SIGTERMed first at
20// sleep — the sidecar exists exactly while its instance is awake, so its
21// lifecycle solves itself. Its own connections flow back through the edge
22// (the box pins its hosted names to loopback), so the sidecar marks them
23// (SidecarHeader) and the activator excludes them from the idle clock —
24// otherwise the pair could never sleep, and an orphaned sidecar's reconnect
25// loop would wake the instance right back (see ServeHTTP).
26//
27// Standalone sidecar members (NATIVE): a route with registry.Route.IsSidecar
28// true is its own record, its own registry row, with no parent instance to
29// hang its lifecycle from — WakeWorker is its request-less entry point, and
30// ExecBackend.StartWorker (not StartSidecar) is what spawns it. This is the
31// package's other sidecar mechanism, not a replacement for the co-spawn
32// above: a host may carry one or the other but never both (the provisioning
33// reconciler's mint-time check in internal/provision/sidecar_reconcile.go
34// refuses to mint a native row for a host that already has a co-spawn env
35// file staged). Retiring the co-spawn path is a follow-up, gated on Eleven's
36// own migration onto native sidecar records.
37//
38// Opt-in is PER HOST (routes.hibernate): with the bucket env set, only
39// flagged instances are activator-managed; the rest stay on their systemd
40// units and are proxied straight. That's what makes a production canary safe —
41// the env ships fleet-wide as a no-op, then flipping one host's flag migrates
42// exactly one instance: its unit stops, the flag flips, and the first request
43// wakes it here.
44//
45// The due-time index: scheduled server-side work fires on time even while its
46// instance sleeps. At sleep the stop path asks the instance for its earliest
47// pending delivery (GET /api/next-due over the socket, admin-gated) and
48// records it in the registry (routes.next_due — persisted so it survives
49// agent restarts); the sweeper wakes any sleeping host whose time has
50// arrived, with no client request. Once up, the instance's own scheduler
51// delivers, and the normal idle window puts it back to sleep — which
52// re-records the NEXT due time, so the cycle maintains itself. Worst-case
53// lateness ≈ one activator sweep interval (≤30s) + spawn + one scheduler tick.
54package activator
55
56import (
57 "context"
58 "crypto/rand"
59 "encoding/hex"
60 "errors"
61 "fmt"
62 "log"
63 "net/http"
64 "os"
65 "path"
66 "slices"
67 "strconv"
68 "strings"
69 "sync"
70 "sync/atomic"
71 "time"
72
73 "github.com/carlosframework/platform/internal/cgroup"
74 "github.com/carlosframework/platform/internal/edge"
75 "github.com/carlosframework/platform/internal/hostclaim"
76 "github.com/carlosframework/platform/internal/hostwrite"
77 "github.com/carlosframework/platform/internal/instancectl"
78 "github.com/carlosframework/platform/internal/lease"
79 "github.com/carlosframework/platform/internal/registry"
80 "github.com/carlosframework/platform/internal/release"
81)
82
83// SidecarHeader marks every request a co-spawned sidecar makes to its own
84// instance through the edge. Marked traffic is activity-invisible: served
85// while the instance is awake, refused (503) while it isn't, never a wake.
86// The marker is deliberately spoofable-but-harmless: it only DOWNGRADES the
87// sender's own requests (no activity credit; 503 while the instance sleeps).
88// It cannot suppress other clients' activity stamps and cannot wake anything,
89// so an outsider sending it gains nothing.
90const SidecarHeader = "X-Carlos-Sidecar"
91
92// Config is the agent-process opt-in, read from env. Nil (bucket unset) keeps
93// proxy-only behavior byte-identical: instances run under systemd, the edge
94// only proxies.
95type Config struct {
96 HostWrites hostwrite.Config
97 Claims *ClaimRuntime
98 Bucket string // CARLOS_HIBERNATE_BUCKET (required to enable)
99 Region string // CARLOS_HIBERNATE_REGION (optional; AWS default chain resolves "")
100 Prefix string // CARLOS_HIBERNATE_PREFIX (optional key prefix inside the bucket)
101 Idle time.Duration // CARLOS_HIBERNATE_IDLE (default 15m; seconds-granular, e.g. "90s")
102 // AgentSocket is the unix socket a worker child dials to reach back —
103 // the direction that makes a sidecar possible at all: the agent listens,
104 // children call in; the platform never dials a sidecar. Handed to every
105 // worker as CARLOS_AGENT_SOCKET. Empty falls back to DefaultAgentSocket.
106 AgentSocket string
107}
108
109// DefaultAgentSocket is where the agent listens for its worker children.
110const DefaultAgentSocket = "/run/carlos/agent.sock"
111
112// ConfigFromEnv reads the CARLOS_HIBERNATE_* env vars. A nil, nil return
113// means the bucket is unset: hibernation disabled, don't construct a Service.
114func ConfigFromEnv() (*Config, error) {
115 bucket := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_BUCKET"))
116 if bucket == "" {
117 return nil, nil
118 }
119 cfg := &Config{
120 Bucket: bucket,
121 Region: strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_REGION")),
122 Prefix: strings.Trim(strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_PREFIX")), "/"),
123 Idle: 15 * time.Minute,
124 AgentSocket: DefaultAgentSocket,
125 }
126 if v := strings.TrimSpace(os.Getenv("CARLOS_HIBERNATE_IDLE")); v != "" {
127 d, err := time.ParseDuration(v)
128 if err != nil || d <= 0 {
129 return nil, fmt.Errorf("CARLOS_HIBERNATE_IDLE %q: want a positive duration like 15m or 90s", v)
130 }
131 cfg.Idle = d
132 }
133 return cfg, nil
134}
135
136// hibernatePrefix is the per-instance S3 key prefix: <Prefix>/<host>/ (the
137// lease's lock.json is colocated under it by the lease package).
138func (c Config) hibernatePrefix(host string) string {
139 return HibernatePrefix(c.Prefix, host)
140}
141
142// HibernatePrefix is that layout as a free function, exported for the
143// verification pass (internal/ops) that restore-drills these replicas. It is
144// a function rather than a comment because a drill that guessed the prefix
145// wrong would report "no replica" for a perfectly good backup — coverage
146// theatre of exactly the kind the drill exists to end.
147func HibernatePrefix(prefix, host string) string {
148 return path.Join(prefix, host)
149}
150
151// mesh seam: Eleven also derived meshPrefix() here — <Prefix>/mesh/members,
152// the membership heartbeats' home beside the per-instance prefixes (the
153// bucket is the mesh's only shared substrate; instance prefixes are public
154// FQDNs and the bare label "mesh" never is one, so they can't collide).
155
156// --- backend seam ------------------------------------------------------------
157//
158// The state machine's side effects (S3 lease, restore, process spawn) sit
159// behind these interfaces so the machine itself is unit-testable with fakes —
160// wake coalescing, idle→sleep, and lease-lost stand-down need no AWS.
161
162// Backend is the side-effect seam (Eleven's actBackend). ExecBackend is the
163// production implementation for exec-child instances; UnitBackend drives
164// carlos-app@ unit tenants; SplitBackend composes the two by the route's
165// Backing. Every method takes the full route (Eleven passed bare host/dbPath
166// strings) precisely so that composition can dispatch without a registry
167// read of its own.
168type Backend interface {
169 // Acquire claims the per-host single-writer lease. Contention surfaces as
170 // errLeaseHeld so the edge can answer 503 + Retry-After.
171 Acquire(ctx context.Context, rt registry.Route) (Lease, error)
172 // Restore materializes the local DB (rt.DBPath; the backend defaults it)
173 // from S3 when it's absent (a no-op when the warm-cache copy is still on
174 // disk). fresh reports that the local DB was absent AND no replica
175 // existed — a first boot, never a restore. The activator runs the route's
176 // Seed exactly then.
177 Restore(ctx context.Context, rt registry.Route) (fresh bool, err error)
178 // Seed runs the route's declared first-boot initialization (rt.Seed, a
179 // JSON argv array) against the instance's own database, before anything
180 // has spawned. Called by doWake only on a fresh boot with a seed
181 // declared, so it must be safe to assume an empty database — and its
182 // failure fails the wake, because a half-seeded instance must not serve.
183 Seed(ctx context.Context, rt registry.Route) error
184 // Start spawns the instance process and waits until its socket accepts. It
185 // also returns the build it execed (the live/<host> symlink target read at
186 // spawn), so the sweep can later spot a repoint under a still-running child
187 // and converge it. Empty when unknown — the sweep treats that as "don't
188 // cycle" (fail-safe).
189 Start(ctx context.Context, rt registry.Route) (Proc, string, error)
190 // NextDue asks a running instance for its earliest pending scheduled
191 // delivery (unix seconds; 0 = nothing scheduled). Queried at sleep time,
192 // while the socket still answers, to feed the due-time index.
193 NextDue(ctx context.Context, rt registry.Route) (int64, error)
194 // HasSidecar reports whether the route's host has a sidecar staged on
195 // this box.
196 HasSidecar(rt registry.Route) bool
197 // StartSidecar spawns the host's sidecar. Fire-and-forget: a sidecar is a
198 // pure outbound client with no socket, so there's nothing to health-check —
199 // its own connect/backoff loop recovers from a slow start.
200 StartSidecar(ctx context.Context, rt registry.Route) (Proc, error)
201 // StartWorker spawns a child that binds NOTHING: no socket, no port. No
202 // health probe is possible or wanted — readiness is the caller's settle
203 // window, or the child's own call over the agent socket. rt.Args (a JSON
204 // argv array) names what to run; empty or unparseable is refused rather
205 // than exec'd bare. env is merged over the route's own env layers and
206 // wins conflicts — it carries the per-wake sidecar token, which the
207 // tenant's own env file must never be able to override.
208 StartWorker(ctx context.Context, rt registry.Route, env map[string]string) (Proc, string, error)
209}
210
211// DueStore persists the due-time index (Eleven's actDueStore) — the registry
212// in production (routes.next_due), so a sleeping host's timer survives agent
213// restarts instead of dying with the process. *registry.Registry satisfies it
214// directly.
215type DueStore interface {
216 SetNextDue(host string, due int64) error
217 // ClearNextDueIf consumes a FIRED due time — see the registry's own doc
218 // comment. Conditional on the row still holding the value that fired, so
219 // a child that posted its next due time during this very wake keeps it.
220 ClearNextDueIf(host string, due int64) error
221 DueInstances(now int64) ([]registry.Route, error)
222 // AlwaysOnWorkers lists the non-hibernating sidecar rows this box carries,
223 // so the sweep can start one that appeared after the agent's boot pass —
224 // which runs once, at arming, and never again.
225 AlwaysOnWorkers() ([]registry.Route, error)
226 // EarliestTimers is the hibernation guard's input: host -> soonest armed
227 // timer. Read with DueInstances, before the sweep takes a.mu.
228 EarliestTimers() (map[string]int64, error)
229}
230
231// RouteLookup is the registry read doWake performs before Acquire (design
232// §6.8): the edge hands a wake the route it had cached at request time, which
233// can be stale by the time the wake actually runs (a config edit landed in
234// between); doWake re-reads the committed row so it spawns against
235// whatever's really in the registry, not a snapshot. *registry.Registry
236// satisfies this directly with its own Get.
237type RouteLookup interface {
238 Get(host string) (*registry.Route, error)
239}
240
241// leaseReleaseBudget bounds a wake bail-out's lease release: one small S3
242// DELETE, the same five seconds lease.Replicator.Close gives its own.
243const leaseReleaseBudget = 5 * time.Second
244
245// releaseLease gives a failed wake's lease back under a budget of its own,
246// never the wake's context. A fence or Shutdown cancels that context, and
247// the real release (an S3 DeleteObject) fails at once on a dead one while
248// spending the handle anyway, which leaves the lease held for its whole TTL
249// and the host refusing wakes until then. WithoutCancel keeps the
250// context's values and drops only its cancellation and deadline.
251func releaseLease(ctx context.Context, lease Lease) error {
252 rctx, cancel := context.WithTimeout(context.WithoutCancel(ctx), leaseReleaseBudget)
253 defer cancel()
254 return lease.Release(rctx)
255}
256
257// releaseRetrier is the OPTIONAL half of Lease that a failed Release
258// leaves something to retry with (#292, Astra P3): lease.Held.Release
259// spends its handle even when the DELETE fails, so the same conditional
260// delete is re-issued through RetryRelease. execLease has it.
261type releaseRetrier interface {
262 RetryRelease(ctx context.Context) error
263}
264
265// owedRelease is a lease whose release failed: kept, and retried by
266// finishPendingStops (the sweep, and Shutdown's polling) until it lands or
267// Shutdown's budget ends. busy marks a retry in flight; lastLog paces the
268// failure lines. Both under a.mu.
269type owedRelease struct {
270 host string
271 lease releaseRetrier
272 busy bool
273 lastLog time.Time
274}
275
276// giveBack is releaseLease for a lease the agent holds for host, with a
277// failure kept as an obligation instead of forgotten (#292, Astra P3).
278// It reports whether the release landed now.
279func (a *Service) giveBack(ctx context.Context, host string, lease Lease) bool {
280 err := releaseLease(ctx, lease)
281 if err != nil {
282 a.oweRelease(host, lease, err)
283 }
284 return err == nil
285}
286
287// oweRelease records a failed release of host's lease for retry. A lease
288// that cannot retry is left to its TTL, as before. Never called with a.mu
289// held.
290func (a *Service) oweRelease(host string, lease Lease, err error) {
291 r, ok := lease.(releaseRetrier)
292 if !ok {
293 log.Printf("activator: %s: giving the lease back failed: %v (the lease's TTL frees it)", host, err)
294 return
295 }
296 log.Printf("activator: %s: giving the lease back failed: %v; retrying it", host, err)
297 a.mu.Lock()
298 if a.owedReleases == nil {
299 a.owedReleases = make(map[*owedRelease]struct{})
300 }
301 a.owedReleases[&owedRelease{host: host, lease: r, lastLog: time.Now()}] = struct{}{}
302 a.mu.Unlock()
303}
304
305// retryOwedReleasesLocked starts one retry of every owed release that has
306// none in flight. The caller holds a.mu.
307func (a *Service) retryOwedReleasesLocked() {
308 for o := range a.owedReleases {
309 if o.busy {
310 continue
311 }
312 o.busy = true
313 go a.retryOwedRelease(o)
314 }
315}
316
317// retryOwedRelease is one conditional retry, under the release budget of
318// its own. Landed, or no longer ours to release, the obligation goes.
319func (a *Service) retryOwedRelease(o *owedRelease) {
320 ctx, cancel := context.WithTimeout(context.Background(), leaseReleaseBudget)
321 err := o.lease.RetryRelease(ctx)
322 cancel()
323 // A record past its own expiry is retired even though the delete still
324 // fails (lease.ErrReleaseLapsed): the lease is free to any acquirer by
325 // then, and a persistent refusal must not keep obligations forever.
326 lapsed := errors.Is(err, lease.ErrReleaseLapsed)
327 a.mu.Lock()
328 o.busy = false
329 logIt := err == nil || lapsed || time.Since(o.lastLog) >= 30*time.Second
330 if err == nil || lapsed {
331 delete(a.owedReleases, o)
332 } else if logIt {
333 o.lastLog = time.Now()
334 }
335 a.mu.Unlock()
336 switch {
337 case err == nil:
338 log.Printf("activator: %s: lease given back on retry", o.host)
339 case lapsed:
340 log.Printf("activator: %s: lease release retired, never deleted: %v", o.host, err)
341 case logIt:
342 log.Printf("activator: %s: retrying the lease's release: %v", o.host, err)
343 }
344}
345
346// leaseHealth is the OPTIONAL half of Lease: a held lease that can say
347// whether it is still held (no failed renewal, not spent). execLease has it
348// (lease.Held.Err). A hold's pinned wake asks it at its exec, before the
349// lease has become a replica (final review F7).
350type leaseHealth interface {
351 Err() error
352}
353
354// leaseLost is l's liveness as an ownership refusal: errSwapLeaseLost
355// (gateNoOwnership) when l says it is no longer held, nil when it is held
356// or cannot say.
357func leaseLost(l Lease) error {
358 if h, ok := l.(leaseHealth); ok {
359 if err := h.Err(); err != nil {
360 return fmt.Errorf("%w: %v", errSwapLeaseLost, err)
361 }
362 }
363 return nil
364}
365
366// Lease is a held-but-not-yet-replicating lease (wake in progress; Eleven's
367// actLease).
368type Lease interface {
369 // Replicate trades the lease for running replication of the (now present)
370 // local DB; the lease handle is spent on success. syncInterval is the
371 // route's replication-class tick (replicationSyncInterval; 0 = stream).
372 Replicate(ctx context.Context, dbPath string, syncInterval time.Duration) (Rep, error)
373 // Release gives the lease up — the bail-out when a wake fails partway.
374 Release(ctx context.Context) error
375}
376
377// Rep is running replication coupled to the lease (Eleven's actRep).
378type Rep interface {
379 // Err reports lost leadership (a lease renewal failed).
380 Err() error
381 // Close does the final checkpoint/sync and releases the lease.
382 Close(ctx context.Context) error
383}
384
385// Proc is a running instance child process (Eleven's actProc).
386type Proc interface {
387 // Stop terminates gracefully (SIGTERM, bounded wait, then SIGKILL).
388 Stop(ctx context.Context) error
389 // Exited reports whether the process has died on its own.
390 Exited() bool
391}
392
393// procCrasher is the OPTIONAL half of Proc: a child that can say how it
394// exited. A hibernating worker's exit is ambiguous without it — design §7.2
395// says a crash restarts, unqualified, while a clean exit from a bot that has
396// finished its work is exactly what parking is for — and exit status is the
397// one fact that tells them apart.
398//
399// Optional rather than a widening of Proc because it is only ever consulted
400// for a worker, and ExecBackend (the only backend that spawns one;
401// UnitBackend refuses StartWorker outright) is the only implementation that
402// can answer. Kept additive so no other Proc — production or test — has to
403// grow a method it cannot answer honestly.
404type procCrasher interface {
405 // Crashed reports a DEFINITELY non-zero exit. Only ever true for a child
406 // that has actually exited.
407 Crashed() bool
408}
409
410// procCrashed asks a Proc whether it crashed, defaulting to NO for a child
411// that cannot say. That default is the safe direction: a crash misread as a
412// finish parks a worker its schedule or an operator can still wake, while a
413// finish misread as a crash restarts a bot that had nothing left to do —
414// forever.
415func procCrashed(p Proc) bool {
416 c, ok := p.(procCrasher)
417 return ok && c.Crashed()
418}
419
420// Replication classes — the registry's routes.replication vocabulary
421// ("" | stream | batched | daily), mapped to litestream sync intervals by
422// replicationSyncInterval.
423//
424// Aliases, not copies: internal/registry owns the vocabulary now that
425// `carlos route --replication` writes the column against it (discussion
426// #10), and two independent spellings of the same three strings is exactly
427// how a setter and a reader drift into disagreeing about what `daily` means.
428const (
429 replicationStream = registry.ReplicationStream // change-driven ~1s sync — loss window seconds
430 replicationBatched = registry.ReplicationBatched // sync at most every 5m — loss window minutes
431 replicationDaily = registry.ReplicationDaily // sync at wake + every 24h — loss window up to a day
432)
433
434// replicationSyncInterval maps a class to the litestream replica sync
435// interval it buys. 0 means "keep litestream's default" (1s, change-driven —
436// idle ticks make zero S3 calls), which is what stream and the unset ""
437// deliberately share so pre-ladder registries keep today's behavior exactly.
438func replicationSyncInterval(class string) time.Duration {
439 switch class {
440 case replicationBatched:
441 return 5 * time.Minute
442 case replicationDaily:
443 return 24 * time.Hour
444 }
445 return 0
446}
447
448// --- state machine -----------------------------------------------------------
449
450// Per-host lifecycle. An absent entry is the fourth state: asleep.
451//
452// asleep --request--------------------> waking (one wake; concurrent
453// waking --wake ok--------------------> awake requests coalesce on it)
454// waking --wake failed----------------> asleep (error served; next request retries)
455// awake --idle timeout / lease lost /
456// child died----------------> stopping (final checkpoint, release lease)
457// stopping --stop done----------------> asleep (local DB left as a warm cache)
458//
459// A request that lands on waking waits (bounded) for the wake; one that lands
460// on stopping waits for the stop, then re-enters as a fresh wake.
461type actState int
462
463const (
464 actWaking actState = iota
465 actAwake
466 actStopping
467)
468
469type actEntry struct {
470 claimResult *wokenInstance
471 // probe is set for an entry ProbeWake made (moves Plan 3 Task 5): the
472 // claim path runs with the probe's policy, and the entry is held for
473 // the move while its registry row is.
474 probe *probeEntry
475 // RecoveryProbe retains its separate recovery policy. recoveryPolicy is the
476 // caller's probe policy, set at construction and never written again;
477 // nil for every other entry, whose wake is main's. probeAdmitted is the
478 // admission the probe started under, written by the wake before the
479 // awake transition. probeHeld keeps requests, bounces and the idle clock
480 // off the entry until the claim says the source opened on that session.
481 recoveryPolicy *StartPolicy
482 probeAdmitted hostclaim.Observed
483 probeHeld bool
484 probeReviewing bool
485 probeObsolete bool
486 probeFence uint64
487 // bounceAfterWake is a Bounce that arrived while this INSTANCE was still
488 // waking. The wake had already read its config and execed its build, so
489 // the bounce's reason has not reached the child; wake() spends it the
490 // moment the wake finishes, before any waiting request is admitted.
491 bounceAfterWake bool
492 host string
493 route registry.Route
494 state actState
495
496 ready chan struct{} // waking: closed when the wake resolves; then check err
497 err error // waking: the wake's outcome
498 stopped chan struct{} // stopping: closed when the stop completes
499 wakeCancel context.CancelFunc
500 stopOutcome *stopObservation
501
502 // stopWhy is why this entry is stopping (beginStopLocked), written under
503 // a.mu; finishStop hands it to a run it interrupts (run spec §5.3).
504 stopWhy string
505
506 // cmd is the run attached to this instance (run spec §5.3), written
507 // under a.mu. While it is set the idle sweep never stops the entry;
508 // every other stop stops and settles its unit in finishStop, before
509 // replication closes, and a unit not proven settled keeps the stop
510 // pending with replication and the lease held (merge ruling M1).
511 cmd *cmdHold
512 // cmdLeft holds runs released before their unit was proven settled
513 // (cmdRun.Release): their Stop failed or never came. finishStop settles
514 // them as it does cmd, before replication closes. Written under a.mu.
515 cmdLeft []*cmdHold
516
517 sidecarUncertain bool
518
519 // Activity tracking: an instance is idle only when nothing is in flight
520 // AND the idle window has passed since the last request finished. In-flight
521 // counting (not just timestamps) is what keeps an instance with one open
522 // WebSocket awake — the proxy's ServeHTTP blocks for the tunnel's life.
523 // (Sidecar-marked requests register in neither — see BeginSidecarRequest.)
524 inflight int
525 lastActive time.Time
526
527 rep Rep
528 proc Proc
529
530 // Convergence: the `live/<host>` symlink target this child was spawned
531 // against. A running process never re-reads the symlink, so once a ship
532 // repoints it, an awake child keeps serving the old build until it's
533 // cycled. spawnVersion lets the sweep notice the drift; staleSince is when
534 // it first did, so a brief transient (a symlink moved then rolled straight
535 // back) doesn't trigger a needless cycle. Both empty/zero until the wake
536 // resolves.
537 spawnVersion string
538 staleSince time.Time
539
540 // gen is the generation this entry's process is (zero-downtime spec
541 // §7.1): its number, socket, control socket and build. Set under a.mu at
542 // the awake transition and never changed after it in this plan (plan 3's
543 // switch replaces it). Zero for a worker and before the wake resolves.
544 gen Gen
545
546 // Sidecar co-spawn: the sidecar exists exactly while its instance is
547 // awake. All sidecar fields are written under a.mu in the awake state only
548 // (wake's attach, the sweeper's restart); a stop flips the state under the
549 // same lock first, so stop() owns them uncontended from then on.
550 sidecar Proc // the co-spawned `sidecar run` child; nil = none running
551 sidecarWanted bool // the host had a sidecar staged at wake time
552 sidecarStarted time.Time // when the current sidecar spawned (backoff reset)
553 sidecarBackoff time.Duration // last restart delay (doubles, capped)
554 sidecarRetryAt time.Time // earliest next restart attempt
555 sidecarStarting bool // a restart's backend.StartSidecar call is in flight, off a.mu (design §6.7)
556 // sidecarStartDone is closed by the in-flight restart goroutine when it
557 // finishes (installed or discarded), non-nil exactly while
558 // sidecarStarting is true. stop() waits on it, bounded by
559 // sidecarStopTimeout, instead of polling sidecarStarting.
560 sidecarStartDone chan struct{}
561 // sidecarGen is the generation whose build the legacy sidecar runs;
562 // written under a.mu.
563 sidecarGen uint64
564
565 // pendingStop marks a stop whose proc.Stop returned ErrStillRunning
566 // (design §6.3): termination is unproven, so stop() returned without
567 // releasing the lease or forgetting the entry, and the entry stays in
568 // stopping. sweepOnce (and, during Shutdown, finishPendingStops) polls
569 // proc.Exited() for every pendingStop entry and finishes the stop
570 // (finishPendingStop) the moment it's proven.
571 pendingStop bool
572 // stopRetrying is true while a retried proc.Stop call for a still-running
573 // pendingStop entry is in flight, so finishPendingStops does not launch a
574 // second one on top of it. stopRetriedAt is when the last retry was
575 // issued; finishPendingStops waits actStopTimeout between retries, the
576 // same cadence proc.Stop itself is bounded by.
577 stopRetrying bool
578 stopRetriedAt time.Time
579
580 // ops is this entry's operation set (zero-downtime spec §7.4): every
581 // generation other than the current one that is spawning,
582 // health-checking or draining. Written under a.mu. Empty unless a swap
583 // has run on this entry, and swaps need plan 1's generations switch.
584 ops opSet
585 // pin is a hold swap's chosen build (zero-downtime spec §7.8). The wake
586 // that follows the hold's stop execs exactly this staged artifact rather
587 // than the live link. It is set at construction (holdWakeLocked), before
588 // the wake goroutine starts, and never written again, like the worker
589 // fields. The one exception is a claims-on planned wake (plan 5,
590 // doClaimWake), whose own wake goroutine sets it under a.mu before its
591 // start policy is built; every reader is that goroutine or holds a.mu.
592 pin *Gen
593 // pinCheck is a hold's attempt checks, re-run under a.mu at the pinned
594 // exec. Set at construction (holdWakeLocked), like pin.
595 pinCheck func() error
596 // pinExec is the pinned wake's execution evidence, set (under a.mu,
597 // markPinExec) from its spawn's RESULT: a returned process, or an
598 // unproven-start sentinel (final review F10). It is set before any
599 // cleanup of the failed wake. Unset, nothing of the build ran, so the
600 // hold is abandoned, never down (spec §6.4), and its refusal is no
601 // member's event.
602 pinExec bool
603 // holdNext is a hold swap waiting for this entry's stop to finish.
604 // finishStop installs its pinned wake. Written under a.mu.
605 holdNext *holdSpec
606 // pinFailed is the pinned build a claims-on start ran and saw fail
607 // before it was ready, and pinFailReason why (zero-downtime spec §8.2).
608 // startAdmitted writes them under a.mu, only for a failure classifyStep
609 // calls the build's; plan 5 reports them. Nil when the pinned build
610 // served, or never ran.
611 pinFailed *Gen
612 pinFailReason string
613 // settleCancel, on a hold canary's pinned entry, ends its settle
614 // observation. beginStopLocked calls it while the process has not
615 // exited, so an intentional stop is never read as a crash (spec §6.4).
616 // Written under a.mu.
617 settleCancel context.CancelFunc
618 // genReaping is true while reapCurrentGen is running for this pending
619 // stop, so finishPendingStops does not start a second one. Written
620 // under a.mu.
621 genReaping bool
622 // sidecarJoin is a sidecar start a flag-off terminal stop stopped
623 // waiting for (sidecarStartJoinWait) while it was still in flight: the
624 // stop stays pending, with the lease, until it closes (Task 8, F6 plan
625 // 5), and only then is the host cgroup proven empty. Written under a.mu.
626 sidecarJoin chan struct{}
627 // unprovenSidecars are sidecars of this stopping entry, generations off,
628 // whose stop did not prove their exit: the co-stopped sidecar, or one a
629 // restart brought back after the stop began (fix wave 1, item 1). The
630 // lease is not released while one of them has not exited, whatever the
631 // layout: a nested row has no host-cgroup proof to reach them. The sweep
632 // re-stops them (reapCurrentGen). Written under a.mu.
633 unprovenSidecars []Proc
634 // launchLayout is the layout the wake launched this entry's child on
635 // (wokenInstance.launchLayout), set at install. A flag-off terminal stop
636 // decides from it whether the host cgroup must be proven empty
637 // (stopHostProven, hostCgroupIsTheWakes). "" requires the proof. Written
638 // under a.mu.
639 launchLayout Layout
640
641 // The rollout scheduler's view of this entry (zero-downtime plan 5).
642 // plan is set by the wake goroutine before the spawn (doWake,
643 // doClaimWake, checkPlan's replan) under a.mu, and read without it only
644 // by that goroutine. firstWake is begin()'s cue to wait out the
645 // first-wake budget; wakeStart (real time, set at the wake's start)
646 // anchors B's health window. unexpectedExit marks a stop that began
647 // because the child died on its own. firstWatch cancels a first wake's
648 // observation when an intentional stop begins, and firstWatchCtx is
649 // that observation's context, both registered in the awake transition
650 // (armFirstWatchLocked) and read by watchFirstWake's goroutine. All but
651 // plan are written under a.mu.
652 plan WakePlan
653 firstWake bool
654 wakeStart time.Time
655 unexpectedExit bool
656 firstWatch context.CancelFunc
657 firstWatchCtx context.Context
658 // firstWatchRelease releases the host's outcome fence the observation
659 // took when it was armed (final review F3), once its terminal event has
660 // been delivered. Under a.mu.
661 firstWatchRelease func()
662
663 // Standalone sidecar ("worker") entries. A worker is this entry's own
664 // proc, not a passenger on someone else's: the co-spawn fields above stay
665 // zero for it, and it never becomes a co-spawn parent either.
666 //
667 // worker, workerIdle, workerMaxRun and readyc are set once, at entry
668 // construction, BEFORE the wake goroutine is launched, and never written
669 // again — so wake/doWake read them off-lock safely (the `go` statement is
670 // the happens-before edge). Everything else below is written under a.mu,
671 // like every other post-wake field: workerToken and spawnedAt at the awake
672 // transition and on a supervised restart, the rest by the agent socket.
673 worker bool // this entry is a sidecar: no socket, no requests
674 workerIdle time.Duration // 0 = fall back to cfg.Idle
675 workerMaxRun time.Duration // 0 = no ceiling
676 readyc chan struct{} // closed when the child reports ready (ends the settle window)
677 readyNoted bool // guards readyc's one close
678
679 // dueFired is the next_due value that TRIGGERED this wake, and only for a
680 // worker woken by the due-time sweep (0 everywhere else). A fired timer is
681 // one-shot: the wake consumes the schedule it acted on (consumeDueTime),
682 // which is what stops a hibernating sidecar waking, idling, parking and
683 // re-waking on every sweep forever. Set before the wake goroutine starts
684 // and never written again, like the other construction-time worker fields.
685 dueFired int64
686
687 workerToken string // the per-wake agent-socket token
688 spawnedAt time.Time // when this child spawned — the max_run and healthy-run clock
689 // workerVersion is what the child last reported over the agent socket:
690 // DISPLAY metadata only (spec §7.9). It can be empty or an app-chosen
691 // string, so it never identifies the executable and is never written to
692 // the registry, whose Version is root's install stamp.
693 workerVersion string
694 // workerServing is the build this worker's current child was exec'd
695 // from, captured by the backend at the exec (spec §7.9): staged path and
696 // sha256. Crash and max_run recycles respawn exactly this, never what
697 // live/<host> names by then. Zero when the backend could not report it.
698 // Written under a.mu at the awake transition and on every installed
699 // respawn.
700 workerServing release.StagedIdentity
701
702 // Crash-restart pacing for an always-on worker, the co-spawn supervisor's
703 // rule applied to a standalone member: doubling, capped, and reset by a
704 // run that lasted sidecarHealthyRun.
705 workerBackoff time.Duration
706 workerRetryAt time.Time
707 // workerRespawning is set while respawnWorkerLocked's goroutine owns this
708 // entry's child: the sweeper skips the entry entirely so the moment
709 // between the old child's stop and the new one's spawn — when the proc
710 // reads as Exited — is not mistaken for a crash. respawnDone is closed
711 // when that goroutine finishes, and is what stop() waits on: a stop must
712 // not release the lease while a spawn against the same DB is in flight.
713 workerRespawning bool
714 respawnDone chan struct{}
715
716 // The socketless attempt's state (zero-downtime plan 5 Task 11,
717 // ReplaceWorker). attemptBuilds names the builds an attempt in flight may
718 // exec (its own and its fallback), so the generations file pins them
719 // before anything execs them (Task 12); startWorker's pending generation
720 // already publishes the exact build about to exec, so this is for the
721 // fallback above all. retireTook is how long the last respawn took to
722 // retire its old child: the "swapped" event's drain time for a worker
723 // (spec §9.4).
724 //
725 // attemptGate is an attempt's §7.4 checks (swapChecksLocked with the
726 // attempt's revision), and holdServing marks a ReplaceWorker fallback
727 // whose publication waits for the scheduler's failure record (spec
728 // §8.1). Both arm the NEXT respawnWorkerLocked only: it takes them when
729 // it is called, so a later respawn (a restore, a fallback, a crash
730 // restart) never inherits them, and a ReplaceWorker that stops waiting
731 // cannot disarm a respawn still running. attemptVeto is that respawn's
732 // answer when the gate refused, read and cleared by ReplaceWorker, and
733 // attemptExec its execution evidence (final review F2): the respawn
734 // started a process of the attempt's build, whether or not it was then
735 // installed or shed. A supersession restores the replaced build only
736 // when it is false.
737 // heldServing is the publication holdServing kept back, delivered by
738 // PublishHeldServing. workerInc is the worker child's incarnation,
739 // bumped at every install (a wake, a respawn): an attempt counts only if
740 // it moved, and a publication is delivered only for the latest.
741 // parkedInc is the incarnation a clean-exit park stopped
742 // (markCleanParkLocked), 0 for none: the only stop after which
743 // ReplaceWorker counts an exited one-shot as a good build. All written
744 // under a.mu.
745 attemptBuilds []string
746 retireTook time.Duration
747 attemptGate func() error
748 attemptVeto error
749 attemptExec bool
750 holdServing bool
751 heldServing *heldServing
752 workerInc uint64
753 parkedInc uint64
754
755 // shed is a respawn child, refused at its installation with generations
756 // off, whose exit is not proven yet (final review R2-1). With
757 // generations on the operation set holds such a child as a member. The
758 // entry keeps it, and its build pinned, until the sweep proves it
759 // exited (reapShedLocked): a terminal stop keeps the lease until then,
760 // and no respawn execs beside it.
761 shed *shedChild
762}
763
764// shedChild is a respawn child that was exec'd and then refused at its
765// installation, kept until its exit is proven (final review R2-1). Every
766// field is written under a.mu. reaping marks a re-stop or proof in flight.
767type shedChild struct {
768 proc Proc
769 build release.StagedIdentity
770 reaping bool
771 retriedAt time.Time
772}
773
774// Sidecar restart pacing — Restart=always semantics without a supervisor: the
775// sweeper redoes a dead sidecar with exponential backoff, capped, and a run
776// that survived sidecarHealthyRun earns a fresh backoff (so one crash a day
777// never escalates to the cap).
778const (
779 sidecarRestartBase = 2 * time.Second
780 sidecarRestartMax = time.Minute
781 sidecarHealthyRun = time.Minute
782)
783
784// sidecarStopTimeout serves two related bounds. First, its name: the
785// sidecar's own stop at sleep time. `sidecar run` has no drain handler, so
786// the SIGTERM is effectively immediate, and the SIGKILL fallback behind this
787// timeout is insurance. Second (design §6.7), as sidecarStartJoinWait's
788// default: how long a flag-off stop() waits on sidecarStartDone for an
789// in-flight sidecar restart to land before carrying on with whatever
790// e.sidecar currently is.
791const sidecarStopTimeout = 5 * time.Second
792
793// sidecarStartJoinWait is how long a flag-off terminal stop waits for an
794// in-flight sidecar start before it carries on with the rest of the stop.
795// It is a promptness bound only: a start still in flight then keeps the
796// stop pending, with the lease, until it returns and has stopped what it
797// started, and the host cgroup is proven empty after that (Task 8, F6
798// plan 5). A var so tests can shrink it.
799var sidecarStartJoinWait = sidecarStopTimeout
800
801// sidecarSettle is the standalone worker's readiness window — what stands in
802// for the health probe a child that binds nothing can never answer. A worker
803// is up once it either reports ready over the agent socket (NoteWorkerReady)
804// or survives this long without exiting; a child that exits INSIDE the window
805// is a failed wake, released and surfaced exactly like a failed instance
806// start. A var, not a const, so tests can shrink it — spawnHealthTimeout's
807// precedent in this package (backend_exec.go).
808var sidecarSettle = 10 * time.Second
809
810// settlePoll derives the settle window's liveness poll: Proc reports death
811// through Exited(), not a channel, so the window polls. Scaled off the window
812// so a shrunken test window still gets several looks, and capped so the
813// production window doesn't spin.
814func settlePoll(window time.Duration) time.Duration {
815 poll := window / 8
816 if poll > 250*time.Millisecond {
817 poll = 250 * time.Millisecond
818 }
819 if poll < time.Millisecond {
820 poll = time.Millisecond
821 }
822 return poll
823}
824
825// maxWorkerStartsPerSweep bounds the sweep's always-on arm — how many
826// stranded always-on sidecars it starts per tick. The same trade host's
827// workerBootConcurrency makes: bounded fan-out, not one-at-a-time and not
828// the whole box at once, because each wake claims a lease and restores a
829// database. The remainder is started on the next sweep.
830const maxWorkerStartsPerSweep = 8
831
832// stopDiscardedSidecar stops a sidecar process a restart attempt returned
833// but could not install. The wake it belonged to moved on (stopped, or
834// replaced by a fresh wake of the same host) while the call was in flight.
835// Bounded by sidecarStopTimeout so a wedged Stop can't leak the goroutine
836// forever; fire-and-forget, since nothing is waiting on this beyond the
837// process actually dying.
838func stopDiscardedSidecar(sc Proc) {
839 if sc == nil {
840 return
841 }
842 go func() {
843 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
844 defer cancel()
845 sc.Stop(ctx) //nolint:errcheck
846 }()
847}
848
849// stopReturnedSidecar stops a sidecar a flag-off restart brought back for
850// an entry that is stopping (Task 8, F6 plan 5): synchronously, bounded by
851// sidecarStopTimeout, so the start's join closes only after it. A sidecar
852// whose stop did not prove its exit (sidecarStopUnproven) is not forgotten
853// (fix wave 1, item 1): it is kept on the entry, and the stop keeps its
854// lease until it has exited (sidecarsExited). The stop's host proof
855// follows, and with a delegated cgroup that proof also reaches whatever
856// this stop left. Never called with a.mu held.
857func (a *Service) stopReturnedSidecar(e *actEntry, sc Proc) {
858 if sc == nil {
859 return
860 }
861 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
862 err := sc.Stop(ctx)
863 cancel()
864 if !sidecarStopUnproven(sc, err) {
865 return
866 }
867 log.Printf("activator: %s: the sidecar a restart brought back for a stopping entry is not proven gone (%v); keeping the lease until it has exited", e.host, err)
868 a.mu.Lock()
869 e.unprovenSidecars = append(e.unprovenSidecars, sc)
870 a.mu.Unlock()
871}
872
873// sidecarStopUnproven reports whether a flag-off sidecar stop left its
874// termination unproven: the stop failed (a nested unit's ErrStillRunning,
875// a signal that could not be sent) and the process has not exited. A stop
876// that returned nil is the stop's own proof, as it is for the main
877// process's flag-off stop: every real Proc returns nil only once its
878// process is gone.
879func sidecarStopUnproven(sc Proc, err error) bool {
880 return err != nil && !sc.Exited()
881}
882
883// sidecarsExited reports whether every sidecar a flag-off stop kept
884// unproven (actEntry.unprovenSidecars) has exited: the sidecar's own stop
885// proof, which is the only one a nested row has. With restop it first
886// stops again, bounded, each one still running (the sweep's retry,
887// reapCurrentGen). Proven ones are forgotten. Never called with a.mu held.
888func (a *Service) sidecarsExited(e *actEntry, restop bool) bool {
889 a.mu.Lock()
890 kept := slices.Clone(e.unprovenSidecars)
891 a.mu.Unlock()
892 if len(kept) == 0 {
893 return true
894 }
895 var gone []Proc
896 for _, sc := range kept {
897 if restop && !sc.Exited() {
898 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
899 if err := sc.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
900 log.Printf("activator: %s: re-stop a sidecar not proven gone: %v", e.host, err)
901 }
902 cancel()
903 }
904 if sc.Exited() {
905 gone = append(gone, sc)
906 }
907 }
908 a.mu.Lock()
909 e.unprovenSidecars = slices.DeleteFunc(e.unprovenSidecars, func(p Proc) bool { return slices.Contains(gone, p) })
910 left := len(e.unprovenSidecars)
911 a.mu.Unlock()
912 if left > 0 {
913 log.Printf("activator: stop %s: %d sidecar(s) it stopped have not exited; keeping the lease until they have", e.host, left)
914 }
915 return left == 0
916}
917
918// discardSidecar is how a sidecar start that lost its entry is shed. With
919// generations off it is today's stopDiscardedSidecar, fire-and-forget. With
920// them on it is synchronous and bounded, and the caller runs it before it
921// closes the join (sidecarStartDone) a terminal stop waits on (spec §7.4).
922// A sidecar whose termination it cannot prove stays tracked
923// (trackStraySidecarLocked).
924func (a *Service) discardSidecar(rt registry.Route, sc Proc) {
925 if sc == nil {
926 return
927 }
928 if !a.gens {
929 stopDiscardedSidecar(sc)
930 return
931 }
932 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
933 err := sc.Stop(ctx)
934 cancel()
935 if !a.sidecarTerminated(rt, sc) {
936 log.Printf("activator: %s: a discarded SIDECAR is not proven gone (%v); keeping it tracked", rt.Host, err)
937 a.mu.Lock()
938 a.trackStraySidecarLocked(rt, sc)
939 a.mu.Unlock()
940 }
941}
942
943var errSidecarUnproven = errors.New("a legacy sidecar of this host is not proven gone")
944
945// sidecarKiller is the backend's legacy-sidecar kill: whatever is left in
946// the host's sidecar cgroup is killed, and it returns once that cgroup is
947// empty.
948type sidecarKiller interface {
949 KillSidecar(ctx context.Context, rt registry.Route) error
950}
951
952// sidecarTerminated is THE termination proof for a legacy sidecar (spec
953// §7.4, §7.5), as genTerminated is for a generation: its leader has exited
954// AND its sidecar cgroup is proven empty. childProc.Stop signals only the
955// leader, so its exit says nothing of a process it forked. A backend with
956// no sidecar kill (a test double), or one that answers errNoGenContainment
957// (the sidecar shares the host cgroup, flag off), has nothing more to
958// prove. A nil process never ran. Never called with a.mu held. The cgroup
959// is this host's legacy sidecar's alone: one entry per host, and the
960// interlock and the join mean no second sidecar of the host runs while one
961// is being proven.
962func (a *Service) sidecarTerminated(rt registry.Route, sc Proc) bool {
963 if sc == nil {
964 return true
965 }
966 if !sc.Exited() {
967 return false
968 }
969 k, ok := a.backend.(sidecarKiller)
970 if !ok {
971 return true
972 }
973 ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
974 defer cancel()
975 if err := k.KillSidecar(ctx, rt); err != nil && !errors.Is(err, errNoGenContainment) {
976 log.Printf("activator: %s: the sidecar's leader exited but its cgroup is not proven empty: %v", rt.Host, err)
977 return false
978 }
979 return true
980}
981
982// trackStraySidecarLocked keeps a legacy sidecar whose termination is
983// unproven: it is never forgotten (spec §7.4). While one is tracked the host
984// takes no new wake (errWakeUnproven), starts no sidecar (startSidecar), and
985// no stop of it releases its lease (releasable); the sweep
986// (reapStraySidecarsLocked) re-stops and re-proves it. The caller holds a.mu.
987func (a *Service) trackStraySidecarLocked(rt registry.Route, sc Proc) {
988 if a.straySidecars == nil {
989 a.straySidecars = make(map[string][]*straySidecar)
990 }
991 a.straySidecars[rt.Host] = append(a.straySidecars[rt.Host], &straySidecar{proc: sc, rt: rt})
992}
993
994// reapStraySidecarsLocked re-stops each stray sidecar and re-proves it off
995// the lock (the proof is a kernel or root call), forgetting it only once
996// sidecarTerminated holds. The caller holds a.mu.
997func (a *Service) reapStraySidecarsLocked() {
998 for _, strays := range a.straySidecars {
999 for _, s := range strays {
1000 if s.reaping {
1001 continue
1002 }
1003 s.reaping = true
1004 go func(s *straySidecar) {
1005 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
1006 if err := s.proc.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
1007 log.Printf("activator: re-stop stray sidecar %s: %v", s.rt.Host, err)
1008 }
1009 cancel()
1010 proven := a.sidecarTerminated(s.rt, s.proc)
1011 a.mu.Lock()
1012 defer a.mu.Unlock()
1013 s.reaping = false
1014 if !proven {
1015 return
1016 }
1017 host := s.rt.Host
1018 keep := slices.DeleteFunc(a.straySidecars[host], func(x *straySidecar) bool { return x == s })
1019 if len(keep) == 0 {
1020 delete(a.straySidecars, host)
1021 } else {
1022 a.straySidecars[host] = keep
1023 }
1024 }(s)
1025 }
1026 }
1027}
1028
1029// Build convergence: how long an awake child may serve a symlink-stale binary
1030// before the sweep gracefully cycles it, and how many to cycle per tick.
1031// - ConvergeGrace comfortably exceeds one sweep interval (≤30s) and a
1032// sidecar's ~32s reconnect, so a genuinely-persistent drift is required —
1033// a symlink that moved then rolled straight back never trips it, and a
1034// briefly-busy host gets a chance to idle-hibernate (which converges for
1035// free) first.
1036// - one per tick staggers a fleet-wide ship across sweeps instead of blipping
1037// every host's connections at once.
1038//
1039// Convergence is a latency/observability fix, not urgent (the supply-chain bake
1040// window already holds releases), so a leisurely, gentle cadence is right.
1041const (
1042 ConvergeGrace = 90 * time.Second
1043 maxConvergeCyclesPerSweep = 1
1044)
1045
1046// ConvergeDecision is the pure state machine behind build convergence: given
1047// the symlink target read this tick (cur/readErr), the build the child execed
1048// (spawnVer), and when the drift was first seen (staleSince), it says whether
1049// to cycle now and returns the staleSince to carry forward. Split out so the
1050// grace/reset logic is unit-testable without a live activator. Three cases:
1051// - read error: unknown — neither cycle nor clear the grace; retry next tick.
1052// - converged / rolled-back / nothing to compare (cur=="" or spawnVer=="" or
1053// cur==spawnVer): clear the grace clock, don't cycle.
1054// - stale (symlink moved since spawn): start the grace clock on first sight,
1055// cycle once the drift has persisted past grace.
1056func ConvergeDecision(cur string, readErr error, spawnVer string, staleSince, now time.Time, grace time.Duration) (cycle bool, nextStale time.Time) {
1057 switch {
1058 case readErr != nil:
1059 return false, staleSince
1060 case cur == "" || spawnVer == "" || cur == spawnVer:
1061 return false, time.Time{}
1062 default:
1063 if staleSince.IsZero() {
1064 staleSince = now
1065 }
1066 return now.Sub(staleSince) >= grace, staleSince
1067 }
1068}
1069
1070// wakeWaitBound caps how long a request queues on someone else's wake or stop
1071// before giving up with 503 — a wedged wake must not pile up connections
1072// forever. A var so tests can shrink it.
1073var wakeWaitBound = 30 * time.Second
1074
1075// wakeBoundExtended is a test seam, nil in production: awaitWake calls it
1076// right after a first wake's waiter extends its bound (spec §8.2), so a test
1077// can fail the wake inside the extension window.
1078var wakeBoundExtended func()
1079
1080// errLeaseHeld: the per-host lease is held elsewhere. On one box this means a
1081// previous incarnation hasn't released (crashed agent; lease expires within
1082// ~lease.LeaseTTL) — the edge answers 503 with Retry-After.
1083type errLeaseHeld struct {
1084 holder string
1085 retryAfter time.Duration
1086}
1087
1088func (e *errLeaseHeld) Error() string {
1089 return fmt.Sprintf("instance lease held by %q (retry in %s)", e.holder, e.retryAfter)
1090}
1091
1092// errConfigRefused: this box cannot provide the instance's sealed config
1093// (no key, the wrong key, or it could not tell), so it will not take the
1094// lease (amadan #213). The condition outlives one request and a keyed
1095// sibling may take the lease in the meantime, so the edge answers 503 with a
1096// longer Retry-After than a lease contention gets.
1097type errConfigRefused struct{ err error }
1098
1099func (e *errConfigRefused) Error() string {
1100 return "instance config unavailable on this box: " + e.err.Error()
1101}
1102func (e *errConfigRefused) Unwrap() error { return e.err }
1103
1104// configAdmitter is the optional Backend capability behind the config gate
1105// (amadan #213): AdmitConfig answers whether this box can provide rt's config.
1106// Acquire asks it before taking a lease; paths that keep a lease and start a
1107// child anyway (a worker's in-place respawn) ask it directly. A backend
1108// without it is not gated.
1109type configAdmitter interface {
1110 AdmitConfig(ctx context.Context, rt registry.Route) error
1111}
1112
1113// seedPlacer is the optional half of a Backend that records a seed placed
1114// after the first upload (seeded-instances spec §3.4).
1115type seedPlacer interface {
1116 SeedPlaced(ctx context.Context, rt registry.Route, rep Rep) error
1117}
1118
1119// configRefusedRetryAfter is the Retry-After (seconds) a refused wake sends.
1120const configRefusedRetryAfter = 30
1121
1122var errWakeTimeout = errors.New("timed out waiting for instance wake")
1123var errStartupStopped = errors.New("activator stopped during startup")
1124
1125// errShuttingDown refuses a request once Shutdown has begun (#292): nothing
1126// wakes, and nothing is recycled, after the agent has started going away.
1127var errShuttingDown = errors.New("activator: the agent is shutting down")
1128
1129// Service is the activator (Eleven's activatorSvc). It structurally satisfies
1130// the edge package's Activator interface — the edge never imports this
1131// package: the dependency points activator→edge only (backend_exec.go takes
1132// edge.Cache and edge.ListenControl; serve calls edge.MarkAnswered), and the
1133// seam back is the interface the edge declares for itself.
1134type Service struct {
1135 // Startup gates admission while the agent recovers installations. Set before publication.
1136 Startup <-chan struct{}
1137 // startupSeenAt is when serve first found Startup closed (unix nanos,
1138 // 0 until then): requests in the window after it may carry a proxy the
1139 // edge chose before recovery moved their upstream.
1140 startupSeenAt atomic.Int64
1141
1142 // idleFor is the plans idle hook (plans 2), set through SetIdleFor: given an
1143 // entry's route it names the plan's idle window, or 0 for no opinion. It
1144 // can only shorten a window. nil means today's behaviour.
1145 idleFor atomic.Pointer[func(registry.Route) time.Duration]
1146
1147 cfg Config
1148 backend Backend
1149 due DueStore
1150 now func() time.Time // a seam for the unit tests' clock
1151 liveVer func(string) (string, error) // live/<host> symlink target; a seam for tests
1152
1153 // Routes, when non-nil, is read by doWake before Acquire (design §6.8)
1154 // to refresh the route against the registry's committed row rather than
1155 // spawning against the copy the edge cached at request time. Nil (every
1156 // test that doesn't wire it, and any caller that doesn't set it) keeps
1157 // today's behaviour: the wake runs on the route it was handed.
1158 Routes RouteLookup
1159
1160 // Logs, when non-nil, receives member-visible platform events (wake
1161 // outcomes, restart bounces) for `carlos logs` — set before Start, like
1162 // the backend's own Logs field. Nil (every test, a box without the
1163 // shipping loop) emits nothing.
1164 Logs EventSink
1165
1166 // mesh seam: Eleven kept the multi-box placement layer here (mesh *meshSvc):
1167 // requests for hosts whose HRW owner is another live box were proxied
1168 // there, and the due-time sweeper only woke hosts this box owned. Nil
1169 // meant a single box — hashing would pick self everywhere, so no mesh
1170 // codepath was touched at all. Single-box v1 drops the field; it re-enters
1171 // with the mesh port.
1172
1173 // lifeCtx is cancelled by Shutdown: the seam that lets an in-flight worker
1174 // spawn be TORN DOWN rather than waited out. See Shutdown for why waiting
1175 // is the wrong trade there.
1176 lifeCtx context.Context
1177 lifeStop context.CancelFunc
1178
1179 mu sync.Mutex
1180 entries map[string]*actEntry
1181 // shuttingDown is set by Shutdown before it stops anything, and read by
1182 // the supervised-respawn goroutines: the agent is going away, so nothing
1183 // new may be spawned against a DB whose lease is about to be released.
1184 // begin, WakeWorker, wake and the sweep read it too, with swaps on or
1185 // off (#292): once it is set nothing wakes, and an exited child is
1186 // Shutdown's to stop, never recycled as an unexpected exit.
1187 shuttingDown bool
1188 // shutdownCtx is Shutdown's budget, set with shuttingDown: a legacy
1189 // proof runs under the sooner of its own budget and this (#292).
1190 shutdownCtx context.Context
1191 // proofCancels are the legacy proofs running now (legacyRunGone), so
1192 // Shutdown can end them with its budget. Under a.mu.
1193 proofCancels map[*byte]context.CancelFunc
1194
1195 // convergeBudget is how many build-convergence cycles this sweep may
1196 // still start: reset to maxConvergeCyclesPerSweep at the top of every
1197 // sweepOnce, and spent by the sweep's drift check and by a recycle that
1198 // converges (recycleWorkerLocked), so the two share one pace. Under a.mu.
1199 convergeBudget int
1200
1201 // uncapturedSaid is the hosts whose uncaptured worker a recycle has
1202 // declined to respawn from the live link (recycleWorkerLocked), so the
1203 // agent says so once per host and not at every backoff or sweep. Under
1204 // a.mu.
1205 uncapturedSaid map[string]bool
1206
1207 // liveExecGate is the backend's LiveExecGate as SetLiveExecGate set it:
1208 // nil on every box but one that declined CARLOS_SWAP=overlap and honours
1209 // failure records or hold-ran markers. The worker converge and recycle
1210 // ask it under a.mu before they stop anything (convergeWorkerLocked,
1211 // recycleWorkerLocked). liveGateSaid is the (host, build) a refusal there
1212 // was last logged for, so a drift the sweep sees every pass is said once.
1213 // Under a.mu.
1214 liveExecGate func(rt registry.Route, build string) error
1215 liveGateSaid map[string]string
1216
1217 fences map[string]uint64
1218 seedExclusions map[string]*SeedExclusion
1219 seedDiagnostics map[string]*seedDiagnostic
1220 // fenceOwners names the owner of an owned fence (FenceAndStopAs), by
1221 // host. A fence installed any other way has no entry. Under a.mu.
1222 fenceOwners map[string]string
1223 tokenPreparations map[string]chan struct{}
1224 stopHistory map[string]*stopObservation
1225 // straySidecars are legacy sidecars whose termination is unproven, per
1226 // host (zero-downtime spec §7.4). Task 7 tracks into it; a stop keeps
1227 // its lease while its host has one (releasable). Under a.mu.
1228 straySidecars map[string][]*straySidecar
1229 // unprovenWakes are failed wakes whose generation is not proven stopped
1230 // (spec §7.4), per host, made lazily. Each keeps its lease and refuses
1231 // the host's next wake until the sweep proves it
1232 // (reapUnprovenWakesLocked). Under a.mu.
1233 unprovenWakes map[string]*unprovenWake
1234 // owedReleases are lease releases that failed and are retried until
1235 // they land (giveBack, #292). Under a.mu.
1236 owedReleases map[*owedRelease]struct{}
1237 // outcomeFences are planned-wake outcomes taken but not yet delivered
1238 // to OnGen, per host, made lazily (final review F3): a failed or
1239 // fallen-back first wake's, from before its entry goes until afterWake's
1240 // OnGen has returned, and a first wake's observation's, from its arming
1241 // until its terminal event has been delivered. A new wake of the host
1242 // waits for them (bounded) before it plans. Under a.mu.
1243 outcomeFences map[string]*outcomeFence
1244
1245 sweepStop chan struct{}
1246 sweepDone chan struct{}
1247 // lastSweep is when the sweeper last finished a sweep, in unix
1248 // nanoseconds (0: never): the hosting loop's own beat, which the load
1249 // balancer's health answer reads (region HA spec 2 §6, LastSweep).
1250 lastSweep atomic.Int64
1251
1252 // gens is the generations switch (SetGenerations) as read at New. Off:
1253 // every spawn binds the canonical socket and no retire hook fires.
1254 gens bool
1255 // cacheGens is host -> *atomic.Uint64, the newest generation that has
1256 // been current for that host (CacheGen). A sync.Map, never under a.mu:
1257 // the edge cache reads it while holding the cache's own lock.
1258 cacheGens sync.Map
1259 // retiredMu guards retiredHooks (OnGenRetired). A leaf lock: nothing is
1260 // called while it is held.
1261 retiredMu sync.Mutex
1262 retiredHooks []func(host, socket string)
1263
1264 // OnWake, when set, is called with the ROUTE after EVERY successful wake
1265 // — fleet boxes use it to publish a binding object to the deployment
1266 // bucket (customer-fleets plan, Task 6) naming this box as the current
1267 // holder of host's lease; nil (every non-fleet deployment) is a pure
1268 // no-op. Deliberately a callback rather than a fleet-aware field on
1269 // Config: the activator has no notion of "fleet" anywhere else in this
1270 // package, and threading FleetName/BoxLabel through Config would leak a
1271 // customer-fleets concept into a package that otherwise only knows
1272 // hosts, leases, and processes. wake() invokes it in its own goroutine
1273 // (see below) so a slow or failing publish can never delay a request
1274 // that's waiting on this same wake.
1275 //
1276 // The whole route rather than the bare host because the hook's own
1277 // refusal is kind-shaped: a binding must never be published for a
1278 // sidecar, and the wake path already HAS the route. Passing the host
1279 // alone forced the hook to re-read the registry to learn the kind, which
1280 // made a security-relevant refusal depend on an I/O call that can fail —
1281 // and fail OPEN, since a read error must not stop legitimate bindings.
1282 OnWake func(rt registry.Route)
1283
1284 // Planner, when set, says which build each wake boots (zero-downtime
1285 // spec §8.2). The rollout manager sets it with CARLOS_SWAP=overlap; nil
1286 // keeps today's wake from the live link. It is called without a.mu held.
1287 Planner func(ctx context.Context, rt registry.Route) WakePlan
1288 // OnGen, when set, hears every planned wake's outcome. It is never
1289 // called with a.mu held: it reaches the scheduler's goodMu, which calls
1290 // back into Current (emitGen).
1291 OnGen func(GenEvent)
1292 // OnDrift, when set, replaces the sweep's convergence stop: with the flag
1293 // on, the scheduler owns every move onto a new build (spec §10). It runs
1294 // on its own goroutine, never under a.mu.
1295 OnDrift func(host string)
1296
1297 // CoalesceRecycle, when set, is asked before a worker is recycled for a
1298 // crash or max_run (spec §7.9). True means an attempt for that host (a
1299 // queued release attempt admissible now, or one already running) will
1300 // make the one spawn itself, so the recycle is skipped. The sweep asks
1301 // again next time if the child is still due. Nil, the default and the
1302 // whole of plan 4, recycles every time. It is called with a.mu held: it
1303 // must answer from its own state, never block, and never call back into
1304 // Service. Set before Start, like OnWake. Plan 5's scheduler fills it.
1305 CoalesceRecycle func(host string) bool
1306
1307 // BeforeExec records, under the instance's lease, that build is about
1308 // to run against host's database (settings spec §4.2), before every
1309 // spawn (recordBuild). The context names the lease's location
1310 // (BuildStateLocationFrom). A non-nil error refuses the spawn, and is
1311 // never a failure of the build. nil records nothing: a box with no
1312 // replica bucket. Set before Start, like OnWake.
1313 BeforeExec func(ctx context.Context, host, build string) error
1314
1315 // HeldTarget, when set, is the agent's answer for a held instance's wake
1316 // (settings spec §4.3 rule 3): the build it must run, in root's rule
1317 // order, or Held false for an instance root follows the channel for. The
1318 // wake asks it before planning and before the lease (wakeCheck), never
1319 // for a hold swap's pinned wake or a moves probe. It runs off a.mu and
1320 // may read the bucket. nil checks nothing: today's wake. Set before
1321 // Start, like OnWake.
1322 HeldTarget func(host string) HeldWake
1323 // RequestStage, when set, asks root to install build for host (the
1324 // stage relay), for a held wake whose target is not installed. It must
1325 // not block: the wake waits for the install itself. Set before Start.
1326 RequestStage func(host, build string)
1327
1328 // OnWorkerServing, when set, hears a worker's captured serving identity
1329 // each time a wake or a respawn installs a child (zero-downtime spec
1330 // §7.9), an empty version included. It runs with no lock held, through
1331 // publishWorkerServing, which delivers only the entry's latest
1332 // incarnation. The agent sets it whatever CARLOS_SWAP says, so a
1333 // flag-off box keeps `carlos sidecars ls`'s running version right
1334 // (registry ServingVersion). Set before Start, like OnWake.
1335 OnWorkerServing func(host string, id release.StagedIdentity)
1336 // servingMu orders OnWorkerServing deliveries: the incarnation check and
1337 // the delivery happen under it, so an older publication can never land
1338 // after a newer one. Taken BEFORE a.mu, never while holding it.
1339 servingMu sync.Mutex
1340
1341 // GenerationsFile, when set, is where the activator publishes every
1342 // worker's captured serving build and every live or pending instance
1343 // generation's build for root's pin set (release.GenerationsPath of the
1344 // data dir root's passes use as BakeDir; see generations_file.go). A
1345 // staged spawn whose build cannot be written there first is refused.
1346 // Empty publishes nothing. Set before Start and before any wake, like
1347 // OnWake.
1348 GenerationsFile string
1349
1350 // schedMu guards schedFences, each host's schedule write fence
1351 // (schedfence.go). It is never held with a.mu or a fence's own lock.
1352 schedMu sync.Mutex
1353 schedFences map[string]*schedFence
1354 // genMu serialises the generations file's writers (persistGenerations).
1355 // It is taken BEFORE a.mu, never while holding it. genWritten (a write
1356 // has succeeded) and genDirty (the last write failed; the sweep retries)
1357 // are read and written under genMu only.
1358 genMu sync.Mutex
1359 genWritten bool
1360 genDirty bool
1361 // genPending holds the builds spawns are about to exec, published
1362 // before each exec (generations.go, ruling R19 item 3); genPendingSeq
1363 // names them. Both under a.mu.
1364 genPending map[pendingGeneration]pendingBuild
1365 genPendingSeq uint64
1366 // joinWaiting, when set, is called by a terminal stop's joins as each
1367 // is about to wait: what is "operation set" (the barrier, joinOps, once
1368 // it has cancelled the attempts, on a non-empty set) or "sidecar start"
1369 // (stop() joining an in-flight supervised sidecar start). It is a test
1370 // seam: the positive signal that a stop has reached a join. Set before
1371 // the stop it observes; nil in production.
1372 joinWaiting func(host, what string)
1373 // wakeEnded, when set, is called as a wake goroutine (wake) returns,
1374 // after its last write (the generations file's republish). It is a test
1375 // seam: a test that removes the generations file's directory waits on
1376 // it first (fix wave 1, item 6). Set before the wake; nil in production.
1377 wakeEnded func(host string)
1378 // genHealth, when set, replaces a generation's health check. It is a
1379 // test seam. Nil is waitHealthyCtx on the generation's own socket (spec
1380 // §7.1: health uses the handle, never AppSocket).
1381 genHealth func(ctx context.Context, g Gen) error
1382 // claimSpawn, when set, replaces a claims-on successor's spawn. It is a
1383 // test seam: a claims-on spawn needs a delegated cgroup tree and an open
1384 // isolated launch. Nil in production (no production file assigns it,
1385 // TestClaimSpawnIsATestSeam), where every claims-on launch is closed on
1386 // this base and claimsSwapRefusal refuses the swap before its raise.
1387 claimSpawn func(ctx context.Context, e *actEntry, rt registry.Route, work hostwrite.Work, g Gen) (Gen, Proc, error)
1388}
1389
1390// New wires the state machine to its side-effect backend and due-time store.
1391// In production due is the *registry.Registry itself.
1392func New(cfg Config, backend Backend, due DueStore) *Service {
1393 a := &Service{
1394 cfg: cfg,
1395 backend: backend,
1396 due: due,
1397 now: time.Now,
1398 entries: make(map[string]*actEntry),
1399 gens: GenerationsEnabled(),
1400 }
1401 // The service lifetime, cancelled by Shutdown. Only work that a shutdown
1402 // should ABANDON hangs off it — today that is exactly the supervised
1403 // worker respawn's exec. Everything a shutdown must still FINISH (the
1404 // final checkpoint, the lease release, the SIGTERM drain) keeps its own
1405 // Background context with its own timeout, so a shutdown can never cancel
1406 // the durability it exists to guarantee.
1407 a.lifeCtx, a.lifeStop = context.WithCancel(context.Background())
1408 // The convergence sweep's live/<host> reader. The exec backend knows the
1409 // symlink layout (OptDir is its field, not a package global — Eleven used a
1410 // global here); any other backend — the unit tests' fakes — leaves
1411 // convergence off ("" = unknown → never cycle) unless the test wires
1412 // a.liveVer itself.
1413 if lv, ok := backend.(interface {
1414 LiveVersion(host string) (string, error)
1415 }); ok {
1416 a.liveVer = lv.LiveVersion
1417 } else {
1418 a.liveVer = func(string) (string, error) { return "", nil }
1419 }
1420 return a
1421}
1422
1423// startupFreshWindow bounds how long after startup completes a request's
1424// route is re-checked against the registry (serve). A proxy the edge chose
1425// before Startup closed reaches serve within milliseconds; the window is
1426// generous, and costs one registry read per hibernating request inside it.
1427const startupFreshWindow = 2 * time.Minute
1428
1429// nearStartup reports whether this request is within startupFreshWindow of
1430// the first time serve found Startup closed. The first observation is at or
1431// after the real close, so a request racing the close is always inside it.
1432func (a *Service) nearStartup() bool {
1433 now := time.Now().UnixNano()
1434 at := a.startupSeenAt.Load()
1435 if at == 0 {
1436 if a.startupSeenAt.CompareAndSwap(0, now) {
1437 return true
1438 }
1439 at = a.startupSeenAt.Load()
1440 }
1441 return now-at < int64(startupFreshWindow)
1442}
1443
1444func (a *Service) awaitStartup(ctx context.Context, deadline <-chan time.Time) error {
1445 if a.Startup == nil {
1446 return nil
1447 }
1448 select {
1449 case <-a.Startup:
1450 if a.lifeCtx.Err() != nil {
1451 return errStartupStopped
1452 }
1453 return ctx.Err()
1454 case <-ctx.Done():
1455 return ctx.Err()
1456 case <-a.lifeCtx.Done():
1457 return errStartupStopped
1458 case <-deadline:
1459 return errWakeTimeout
1460 }
1461}
1462
1463// begin registers a request against rt's instance, waking it if asleep.
1464//
1465// stamp is what separates a visitor from the platform's own clock. Both take
1466// the in-flight count — a delivery in progress must never be cut off by a
1467// concurrent sweep — but only a visitor's release moves lastActive. A
1468// scheduled tick that stamped would reset the idle window on every fire, so
1469// any schedule tighter than that window would pin its instance awake forever
1470// for however little work the handler does.
1471func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (release func(), err error) {
1472 deadline := time.NewTimer(wakeWaitBound)
1473 defer deadline.Stop()
1474 arrival := time.Now()
1475 if err := a.awaitStartup(ctx, deadline.C); err != nil {
1476 return nil, err
1477 }
1478 extended := false
1479 for {
1480 a.mu.Lock()
1481 if a.fences[rt.Host] != 0 {
1482 a.mu.Unlock()
1483 return nil, ErrFenced
1484 }
1485 // Once Shutdown has begun, a request wakes nothing and recycles
1486 // nothing (#292), WakeWorker's rule. Checked on every pass of the
1487 // loop, not only at entry: a request queued on a stop or a wake
1488 // before the shutdown began would otherwise come round and wake the
1489 // host afresh, taking a lease Shutdown never sees. An exited child
1490 // found here is left for Shutdown's own stop, not read as the
1491 // build's failure.
1492 if a.shuttingDown {
1493 a.mu.Unlock()
1494 return nil, errShuttingDown
1495 }
1496 e := a.entries[rt.Host]
1497 if e == nil {
1498 // Asleep → kick off ONE wake; this request and any latecomers all
1499 // coalesce on e.ready rather than racing a second lease claim
1500 // (singleflight). The wake runs detached so even its initiator is
1501 // held to the same bounded wait as everyone queued behind it.
1502 //
1503 // FALL THROUGH to the switch holding the lock — the initiator
1504 // waits on the entry IT created. The old shape unlocked and
1505 // re-read the map, and a wake that FAILED inside that one-
1506 // preemption window had already deleted its entry: the initiator
1507 // found nothing and started a second wake — two lease acquires
1508 // and two restore attempts per request (#184, seen only as a
1509 // loaded-suite test flake because the window is that narrow).
1510 e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
1511 a.entries[rt.Host] = e
1512 go a.wake(e)
1513 }
1514 // A held recovery probe serves nothing until the source opens.
1515 if e.probeHeld {
1516 a.mu.Unlock()
1517 return nil, ErrFenced
1518 }
1519
1520 switch e.state {
1521 case actAwake:
1522 // Stand down synchronously-detected failure: a dead child or a
1523 // lost lease found at request time turns into a stop, and the
1524 // request waits for it then re-wakes — recovery is the routine
1525 // wake path.
1526 if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
1527 why := StopExited
1528 if repErr != nil {
1529 log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
1530 why = StopLeaseLost
1531 } else {
1532 log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
1533 // Not an intentional stop: a first wake's observation
1534 // reads it as the build's failure (spec §6.4).
1535 e.unexpectedExit = true
1536 }
1537 a.beginStopLocked(e, why)
1538 a.mu.Unlock()
1539 continue
1540 }
1541 e.inflight++
1542 a.mu.Unlock()
1543 return a.releaseFunc(e, stamp), nil
1544
1545 case actWaking:
1546 ready := e.ready
1547 a.mu.Unlock()
1548 if err := a.awaitWake(ctx, e, ready, deadline, arrival, &extended); err != nil {
1549 return nil, err
1550 }
1551 // A failed wake reports to every coalesced waiter (retrying
1552 // here would turn one bad wake into a storm); a successful one
1553 // loops to register against the now-awake entry.
1554 if e.err != nil {
1555 return nil, e.err
1556 }
1557 continue
1558
1559 case actStopping:
1560 stopped := e.stopped
1561 a.mu.Unlock()
1562 select {
1563 case <-stopped:
1564 continue // entry is gone; loop wakes it fresh
1565 case <-ctx.Done():
1566 return nil, ctx.Err()
1567 case <-deadline.C:
1568 return nil, errWakeTimeout
1569 }
1570 }
1571 }
1572}
1573
1574// beginRequest is begin for member traffic: activity counts.
1575func (a *Service) beginRequest(ctx context.Context, rt registry.Route) (release func(), err error) {
1576 return a.begin(ctx, rt, true)
1577}
1578
1579// awaitWake is a waiter's wait on one wake: begin's and WakeWorker's. It
1580// returns nil once ready closes (the caller then reads e.err), ctx.Err(), or
1581// errWakeTimeout at the deadline. A first wake onto a new build gives its
1582// waiters arrival + firstWakeBudget (zero-downtime spec §8.2): enough for
1583// B's health window, B's kill and the good build's boot. Only that kind of
1584// wake (firstWake is set only by a planner, flag on), and once per caller:
1585// *extended carries across the caller's loop.
1586//
1587// After the extension it waits again on the SAME ready and never goes back
1588// to the entries map (round 2 Task 6). A wake that fails inside the
1589// extension closes ready and deletes its entry; a waiter that re-read the
1590// map found nothing and started a second wake, two lease acquires and two
1591// spawns for one caller (#184's shape). Called without a.mu.
1592func (a *Service) awaitWake(ctx context.Context, e *actEntry, ready <-chan struct{}, deadline *time.Timer, arrival time.Time, extended *bool) error {
1593 for {
1594 select {
1595 case <-ready:
1596 return nil
1597 case <-ctx.Done():
1598 return ctx.Err()
1599 case <-deadline.C:
1600 a.mu.Lock()
1601 fw := e.firstWake
1602 a.mu.Unlock()
1603 if !fw || *extended {
1604 return errWakeTimeout
1605 }
1606 *extended = true
1607 deadline.Reset(time.Until(arrival.Add(firstWakeBudget)))
1608 if wakeBoundExtended != nil {
1609 wakeBoundExtended()
1610 }
1611 }
1612 }
1613}
1614
1615// errSidecarAsleep: a sidecar-marked request arrived for a host with no awake
1616// instance. Sidecar traffic never wakes anything — the activator itself spawns
1617// the sidecar as part of a wake, so a marked request for a sleeping host is
1618// either a stray reconnect racing a stop or an outsider spoofing the marker;
1619// both get a 503 and change nothing.
1620var errSidecarAsleep = errors.New("instance asleep; sidecar requests do not wake it")
1621
1622// BeginSidecarRequest is beginRequest for sidecar-marked traffic: pass it
1623// through to an awake instance WITHOUT registering activity — no in-flight
1624// count, no lastActive stamp (the sidecar's own long-lived WebSocket must not
1625// hold its instance awake; that non-count is the whole reason the pair can
1626// sleep) — and refuse, rather than wake, when the instance is asleep or
1627// stopping.
1628func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) error {
1629 deadline := time.NewTimer(wakeWaitBound)
1630 defer deadline.Stop()
1631 if err := a.awaitStartup(ctx, deadline.C); err != nil {
1632 return err
1633 }
1634 for {
1635 a.mu.Lock()
1636 if a.fences[rt.Host] != 0 {
1637 a.mu.Unlock()
1638 return ErrFenced
1639 }
1640 e := a.entries[rt.Host]
1641 if e == nil || a.shuttingDown {
1642 // Shutting down: the stop is Shutdown's, and an exited child is
1643 // not recycled here (#292).
1644 a.mu.Unlock()
1645 return errSidecarAsleep
1646 }
1647 if e.probeHeld {
1648 a.mu.Unlock()
1649 return ErrFenced
1650 }
1651 switch e.state {
1652 case actAwake:
1653 // Same dead-child / lost-lease detection as a member request, but
1654 // the recovery stops at the stop: the loop then finds a stopping
1655 // (or absent) entry and turns this request away instead of
1656 // chaining into a re-wake.
1657 if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
1658 why := StopExited
1659 if repErr != nil {
1660 log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
1661 why = StopLeaseLost
1662 } else {
1663 log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
1664 e.unexpectedExit = true // the build's failure inside a first wake's observation (spec §6.4)
1665 }
1666 a.beginStopLocked(e, why)
1667 a.mu.Unlock()
1668 continue
1669 }
1670 a.mu.Unlock()
1671 return nil
1672
1673 case actWaking:
1674 // Joining a wake someone else started is fine — the co-spawned
1675 // sidecar boots moments before its entry flips awake, so its first
1676 // dial can land in that sliver; joining can't initiate anything.
1677 // A failed wake just reads as asleep (the member path already
1678 // logged it).
1679 ready := e.ready
1680 a.mu.Unlock()
1681 select {
1682 case <-ready:
1683 if e.err != nil {
1684 return errSidecarAsleep
1685 }
1686 continue
1687 case <-ctx.Done():
1688 return ctx.Err()
1689 case <-deadline.C:
1690 return errWakeTimeout
1691 }
1692
1693 default: // actStopping — the stop exists to shed exactly this connection
1694 a.mu.Unlock()
1695 return errSidecarAsleep
1696 }
1697 }
1698}
1699
1700// mesh seam: Eleven exposed hasLiveEntry(host) here — the mesh's "already
1701// living here" check (waking and stopping both counted: a waking host is
1702// committed to this box, and a stopping one still holds the lease, so
1703// proxying it away would only trade a short wait for a loud lease race) —
1704// and awakeCount() for the fleet status page.
1705
1706func (a *Service) releaseFunc(e *actEntry, stamp bool) func() {
1707 var once sync.Once
1708 return func() {
1709 once.Do(func() {
1710 a.mu.Lock()
1711 e.inflight--
1712 if stamp {
1713 e.lastActive = a.now()
1714 }
1715 a.mu.Unlock()
1716 })
1717 }
1718}
1719
1720// consumeDueTime clears the due time that fired THIS wake, so a fired timer
1721// wakes its worker exactly once. A no-op for every other wake — an operator
1722// command, the boot pass, an instance's request wake — which is what leaves a
1723// schedule that has not yet come due completely untouched.
1724//
1725// Conditional on the row still holding the fired value (ClearNextDueIf): the
1726// child spawned by this very wake may already have called /v1/next-due during
1727// its settle window, and that newer schedule must survive the consumption of
1728// the older one.
1729func (a *Service) consumeDueTime(e *actEntry) {
1730 if !e.worker || e.dueFired <= 0 {
1731 return
1732 }
1733 if err := a.due.ClearNextDueIf(e.host, e.dueFired); err != nil {
1734 // Loud, but not a failed wake: the child is up and doing its work. The
1735 // cost of the miss is one extra wake on a later sweep, which is
1736 // idempotent — the opposite trade to failing a wake that succeeded.
1737 log.Printf("activator: consume fired due time for %s: %v — it may wake once more", e.host, err)
1738 }
1739}
1740
1741// wake drives asleep→awake: lease, restore-if-missing, spawn, replicate,
1742// co-spawn the sidecar. It runs on the first requester's goroutine but under
1743// its own generous timeout, detached from that request's context — queued
1744// requests still need the wake to finish even if the requester who started it
1745// hangs up.
1746func (a *Service) wake(e *actEntry) {
1747 if a.wakeEnded != nil {
1748 defer a.wakeEnded(e.host)
1749 }
1750 parent := context.Background()
1751 if e.probe != nil && e.probe.parent != nil {
1752 // A probe's wake ends with its caller (ProbeWake).
1753 parent = e.probe.parent
1754 }
1755 ctx, cancel := context.WithTimeout(parent, 5*time.Minute)
1756 defer cancel()
1757 a.mu.Lock()
1758 e.wakeCancel = cancel
1759 // Real time, not a.now: it anchors a first wake's real health deadline
1760 // (spawnPlanned, doClaimWake).
1761 e.wakeStart = time.Now()
1762 // A wake that starts after Shutdown began is cancelled before its
1763 // lease, whatever CARLOS_SWAP says (#292): with swaps off it used to run
1764 // on, and take a lease the exiting agent never gave back.
1765 if a.fences[e.host] != 0 || a.shuttingDown {
1766 cancel()
1767 }
1768 a.mu.Unlock()
1769 wakeStart := a.now()
1770 w, err := a.doWake(ctx, e)
1771 if w != nil {
1772 w.wakeTook = a.now().Sub(wakeStart)
1773 }
1774
1775 // A due-fired wake CONSUMES the schedule it acted on — here, on success,
1776 // and never at stop. The distinction is the whole fix: a worker that goes
1777 // to sleep BEFORE its timer fires must keep its schedule (that row is its
1778 // only way back up), while one whose timer has already fired must not be
1779 // handed the same fired time again on the next sweep. Ordered before the
1780 // awake transition below so nothing can observe this entry awake with the
1781 // spent schedule still standing. A FAILED wake deliberately consumes
1782 // nothing: the index is then the retry, exactly as the due sweep's own
1783 // comment describes.
1784 if err == nil {
1785 a.consumeDueTime(e)
1786 }
1787
1788 // The member-facing record of this wake (spec §5): composed here, where
1789 // the route and the whole error are in hand — the journal line below the
1790 // backends already write stays exactly as it was.
1791 if err != nil {
1792 if genStartUnproven(err) {
1793 // The member's event reads as the fence (memberSafeWakeReason);
1794 // the generation detail is the operator's.
1795 log.Printf("activator: %s: wake failed and its generation is not proven stopped: %v", e.host, err)
1796 }
1797 if msg, ok := a.wakeFailureEvent(e, err); ok {
1798 emitEvent(a.Logs, e.route, "error", msg)
1799 }
1800 } else {
1801 emitEvent(a.Logs, e.route, "info", wakeEventMessage(w, e.worker))
1802 if steps := w.steps.String(); steps != "" {
1803 log.Printf("activator: %s: woke in %s: %s", e.host, w.wakeTook.Round(time.Millisecond), steps)
1804 }
1805 }
1806
1807 var (
1808 servingID release.StagedIdentity
1809 servingInc uint64
1810 publishServing bool
1811 republish bool
1812 releaseOutcome func()
1813 )
1814 a.mu.Lock()
1815 if a.outcomeOwedLocked(e, w, err) {
1816 // A failed or fallen-back first wake (final review F3; spec §8.1,
1817 // O7), or a hold first wake that ran and was ended with no verdict
1818 // (final review wave 2, N2): its failure or hold-ran marker is
1819 // recorded only by afterWake's OnGen, after this
1820 // critical section and after e.ready closes. The host's outcome
1821 // fence is taken here, before the entry goes and before anyone
1822 // waiting is released, so a new wake of the host does not plan
1823 // until that record exists. Released once OnGen has returned.
1824 releaseOutcome = a.holdOutcomeLocked(e.host)
1825 }
1826 if err != nil {
1827 // The failed wake's entry goes, and with it everything it published
1828 // (a worker's captured build, a plan's builds, a pinned build); a
1829 // generation retained as an unproven wake is listed from there.
1830 republish = e.worker || a.publishesLocked(e)
1831 if u := a.unprovenWakes[e.host]; u != nil {
1832 _, listed := a.genIdentity(u.gen)
1833 republish = republish || listed
1834 }
1835 e.err = err
1836 outcome := &stopObservation{uncertain: true}
1837 if w != nil && w.stopOutcome != nil {
1838 outcome = w.stopOutcome
1839 }
1840 a.rememberStopLocked(e.host, outcome)
1841 delete(a.entries, e.host) // next request retries from scratch
1842 } else {
1843 e.state = actAwake
1844 e.rep, e.proc = w.rep, w.proc
1845 e.sidecar, e.sidecarWanted = w.sidecar, w.sidecarWanted
1846 e.sidecarUncertain = w.sidecarUncertain
1847 e.sidecarGen = w.sidecarGen
1848 e.sidecarStarted = a.now()
1849 e.sidecarBackoff = 0
1850 e.sidecarRetryAt = time.Time{} // a wake-time spawn failure retries on the next sweep
1851 e.lastActive = a.now()
1852 // The max_run clock starts when the child does. Set for every entry
1853 // (it costs a field write) but read only for workers: an instance's
1854 // run length is bounded by its idle window and its members.
1855 e.spawnedAt = a.now()
1856 // The build this child actually execed, captured at Start() inside the
1857 // backend (not re-read here — a ship during the wake window would make a
1858 // re-read name the new target while the process runs the old binary, and
1859 // the sweep would never converge it). Empty = unknown → the sweep won't
1860 // cycle. Fresh wake, so it can't be stale yet.
1861 e.spawnVersion = w.spawnVersion
1862 e.workerServing = w.spawnIdentity
1863 if e.worker {
1864 // A new worker incarnation (spec §7.9): its identity is published
1865 // once the wake's outcome has been reported (below).
1866 e.workerInc++
1867 servingID, servingInc, publishServing = e.workerServing, e.workerInc, true
1868 }
1869 e.staleSince = time.Time{}
1870 // The generation is current from here (spec §7.1). A wake the
1871 // claim path ran (doClaimWake) spawned through its provider, not
1872 // startGen. With generations on and a runtime that honours
1873 // StartPolicy.Gen it carries the handle the provider bound (Task
1874 // 10); otherwise it gets the canonical handle and a number of its
1875 // own.
1876 e.gen = w.gen
1877 e.launchLayout = w.launchLayout
1878 if e.gen.N == 0 && !e.worker {
1879 ran := e.gen
1880 e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion)
1881 if ran.StagedPath != "" && ran.Build == w.spawnVersion {
1882 // A claimed start with no number names the file it ran
1883 // (startAdmitted): the handle keeps it, so a legacy
1884 // sidecar restart runs that file too (Task 5d).
1885 e.gen.StagedPath, e.gen.Digest, e.gen.Demand = ran.StagedPath, ran.Digest, ran.Demand
1886 }
1887 }
1888 if a.gens && e.worker && e.gen.N == 0 {
1889 // A worker woken some way that carried no number (doWake gives
1890 // every worker spawn its own): numbered here, with the build it
1891 // runs, so the barrier and the generations file see it. Worker:
1892 // its termination is proven on its host cgroup (Task 11).
1893 e.gen = Gen{N: nextGen(), Build: w.spawnVersion, Worker: true}
1894 }
1895 if e.gen.N != 0 {
1896 a.noteCurrent(e.host, e.gen.N)
1897 }
1898 // An instance's current generation is in the generations file from
1899 // here (spec §8.5). A worker's build was published at its install.
1900 republish = !e.worker && a.publishesLocked(e)
1901 if e.sidecarGen == 0 {
1902 // A wake that carried no generation number (the claim path) ran
1903 // its sidecar on the live link, which is that wake's own build.
1904 e.sidecarGen = e.gen.N
1905 }
1906 // Outside consumers of the canonical path reach the generation
1907 // that just became current (spec §7.1). A failed link costs them
1908 // the canonical path, not the wake: internal traffic never uses it.
1909 if a.gens && linksCanonical(e.route, e.gen) {
1910 if err := swapCanonicalLink(e.route.Socket, e.gen.Socket); err != nil {
1911 log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err)
1912 }
1913 }
1914 // A first wake's observation is registered here, after e.gen is the
1915 // new generation and before e.ready closes, in this critical
1916 // section (zero-downtime spec §6.4), not once the generation is
1917 // already reachable. The fence and bounce checks below come after
1918 // it, so a stop they begin cancels it.
1919 a.armFirstWatchLocked(e, w)
1920 if a.fences[e.host] != 0 || a.shuttingDown {
1921 why := StopFenced
1922 if a.fences[e.host] == 0 {
1923 why = StopShutdown
1924 }
1925 a.beginStopLocked(e, why)
1926 } else if e.bounceAfterWake {
1927 // A config delivery or `carlos restart` asked for while this wake
1928 // was under way (Bounce). Same spot and same reason as the fence
1929 // just above: the state flips to stopping before e.ready closes,
1930 // so a request waiting on the wake is never proxied to this child
1931 // — it waits out the stop and wakes a fresh one.
1932 e.bounceAfterWake = false
1933 a.bounceLocked(e)
1934 }
1935 if a.OnWake != nil && e.state == actAwake {
1936 // Fire-and-forget, off this goroutine: e.route is read once here,
1937 // before this goroutine can be reused for anything else — the
1938 // hook must never share e.host's own goroutine's fate (a wedged
1939 // bucket Put must not hold e.ready closed any longer than the
1940 // wake itself already takes).
1941 rt := e.route
1942 go a.OnWake(rt)
1943 }
1944 }
1945 if w != nil && w.pend != 0 {
1946 // The cold wake's pending build (step 2b): the generation installed
1947 // above, or the failed wake retained as unproven, now names it
1948 // itself, or nothing of it ran. Dropped in the same critical
1949 // section, so no snapshot sees neither.
1950 a.dropPendingGenerationLocked(w.pend)
1951 republish = true
1952 }
1953 a.mu.Unlock()
1954 close(e.ready)
1955 // A planned wake's outcome, for the scheduler: off a.mu, since OnGen
1956 // reaches its goodMu, which calls back into Current.
1957 a.afterWake(e, w, err)
1958 if releaseOutcome != nil {
1959 releaseOutcome()
1960 }
1961 if publishServing {
1962 // A worker's identity, after afterWake and never before it (spec
1963 // §8.1; P5 review finding 7): for a fallback wake, afterWake's
1964 // GenFellBack has the scheduler record B's failure synchronously,
1965 // so B is held before the fallback's identity is published, and a
1966 // publication that blocks cannot hold that record up.
1967 a.publishWorkerServing(e, servingID, servingInc)
1968 }
1969 if republish {
1970 // After a change, the write only logs: the file it failed to
1971 // replace pins at least as much, and the sweep retries it.
1972 _ = a.persistGenerations()
1973 }
1974}
1975
1976// wakeEventMessage is the member-facing wake line, split out so the one thing
1977// an app can actually scrape has a test of its own.
1978//
1979// It carries the wake's timing because an instance cannot time its own wake:
1980// the activator execs the binary, the process starts, opens SQLite, runs its
1981// migrations and parses its templates, and only THEN can the app's own code
1982// observe anything. So a cold-start budget's most important number is the one
1983// the app structurally cannot produce, and every app on the platform meets
1984// the same wall (discussion #6).
1985//
1986// Two figures, because one would not settle the question the budget is
1987// really asking. `spawn` is Backend.Start alone — exec to the socket
1988// accepting, which is the app's OWN boot and none of the platform's work.
1989// The total around it is the lease, the restore, the seed and replication.
1990// An app reading "woke in 1.4s (spawn 1.2s)" knows its boot is the problem;
1991// reading "woke in 1.4s (spawn 200ms)" knows it is not, which is the
1992// distinction #6 says cannot be made at all today.
1993//
1994// Rounded to a millisecond: this is a log line a person reads and a script
1995// greps, and microseconds of a multi-hundred-millisecond wake are noise that
1996// makes both harder.
1997func wakeEventMessage(w *wokenInstance, worker bool) string {
1998 // A worker is not "woken" in the member's vocabulary — nothing dialed it,
1999 // and there was no request waiting on the other end of this. The timings
2000 // below still mean exactly what they mean for an instance, which is why
2001 // only the noun changes.
2002 msg := "instance woke"
2003 if worker {
2004 msg = "sidecar started"
2005 }
2006 if w == nil {
2007 return msg
2008 }
2009 if w.wakeTook > 0 {
2010 msg += " in " + w.wakeTook.Round(time.Millisecond).String()
2011 if w.spawnTook > 0 {
2012 msg += " (spawn " + w.spawnTook.Round(time.Millisecond).String() + ")"
2013 }
2014 }
2015 if w.spawnVersion != "" {
2016 msg += " (version " + w.spawnVersion + ")"
2017 }
2018 return msg
2019}
2020
2021// wokenInstance is a successful wake's yield. sidecar is nil when the host has
2022// none staged — or when the spawn failed, which is deliberately not an error:
2023// members > bots, so a host whose sidecar won't start still serves (the
2024// sweeper retries via sidecarWanted).
2025type wokenInstance struct {
2026 stopOutcome *stopObservation
2027 sidecarUncertain bool
2028 rep Rep
2029 proc Proc
2030 sidecar Proc
2031 sidecarWanted bool
2032 sidecarGen uint64 // the generation whose build the legacy sidecar runs
2033 spawnVersion string // live/<host> target read at exec — for build convergence
2034 spawnIdentity release.StagedIdentity // the build exec'd, as the backend captured it (workers; spec §7.9)
2035 gen Gen // the spawned generation's handle (instances only; zero for a worker)
2036 // genUnproven: the spawn failed and its error says the generation's
2037 // kill was not proven (genStartUnproven). cleanupFailedWake keeps the
2038 // lease until the sweep proves it, whatever a.gens says.
2039 genUnproven bool
2040 // legacyUncertain: the spawn was a start of the host's original run
2041 // unit that root may still carry out (legacyStartUncertain, #292).
2042 // cleanupFailedWake keeps the lease until the unit is proven gone.
2043 legacyUncertain bool
2044 // pend is the build a cold wake captured and published as pending at
2045 // step 2b (settleWakeIdentity; ruling R-COLDPUBLISH). wake() drops it in
2046 // the critical section that ends the wake: by then the entry's
2047 // generation (or a retained unproven wake's) names the build, or
2048 // nothing of it ran.
2049 pend pendingGeneration
2050 // launchLayout is the layout the row read immediately before this
2051 // wake's launch (final review round 1, finding 6, Astra): the failed
2052 // wake's cleanup decides from it, not from the layout at cleanup time,
2053 // whether the host cgroup must be proven empty (hostCgroupIsTheWakes).
2054 // "" when the backend cannot say (no Classifier), which requires the
2055 // proof.
2056 launchLayout Layout
2057 // spawnTook is how long Backend.Start took: exec to the socket accepting,
2058 // which is the instance's OWN boot and nothing of the platform's. wakeTook
2059 // is the whole wake around it, filled in by wake() rather than here
2060 // because the caller owns the clock that bounds it.
2061 //
2062 // They exist because an instance cannot time its own wake — almost all of
2063 // it happens before the app's code can observe anything — so the figure
2064 // an app most wants for a cold-start budget is the one it structurally
2065 // cannot produce (discussion #6). Reported on the wake event, where
2066 // `carlos logs` already carries the line.
2067 spawnTook time.Duration
2068 wakeTook time.Duration
2069 // steps is the operator's breakdown of wakeTook, journal only.
2070 steps wakeSteps
2071
2072 // planned is the generation a planned wake left serving; failed and
2073 // failReason name a first wake's build that failed; fellBack says the
2074 // good build replaced it (zero-downtime spec §8.2). untried says a
2075 // first wake's B was never exec'd because too little of its health
2076 // window was left (firstWakeMinVerdict), so the good build was booted
2077 // in its place as an ordinary wake.
2078 planned Gen
2079 // bootRan: the plan's own build (Boot) ran, or may have (a process came
2080 // back, or startMayHaveRun says its start is unproven or uncertain).
2081 // afterWake reads it for a hold first wake that ended with no verdict
2082 // (holdRanUnjudged; final review wave 2, N2).
2083 bootRan bool
2084 failed *release.StagedIdentity
2085 failReason string
2086 fellBack bool
2087 untried bool
2088}
2089
2090// doWake runs the side effects of one wake. It takes the ENTRY, not the bare
2091// route, because two of its steps depend on what kind of child this is — the
2092// spawn call and the co-spawn — and those facts (e.worker, e.workerToken) are
2093// fixed at entry construction, before the wake goroutine starts, so reading
2094// them here without a.mu is safe by construction.
2095func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, error) {
2096 if a.cfg.HostWrites.Enabled {
2097 return a.doClaimWake(ctx, e)
2098 }
2099 rt := e.route
2100 host := rt.Host
2101 // Until the lease (step 1) this wake starts nothing, so a refusal before
2102 // it is known-no-start evidence, which leaves the host's stop history as
2103 // it was (seeded pre-land re-review, P2; census ruling 292fix).
2104 w := &wokenInstance{stopOutcome: &stopObservation{}}
2105 w.stopOutcome.notStarted()
2106 w.steps.begin(a.now())
2107 if ctx.Err() != nil || !a.wakeAllowed(host) {
2108 return w, ErrFenced
2109 }
2110
2111 // 0. Refresh against the registry's committed row (design §6.8): the
2112 // edge cached rt at request time, and a config edit (provision, a path
2113 // change, a residence move) can land between that cache and this wake
2114 // actually running. Spawning against a stale DBPath/Socket would boot
2115 // the instance onto the wrong place on disk or bind the wrong socket, so
2116 // the wake re-reads the row it's actually about to spawn and carries
2117 // that forward. Every step below reads rt, not the edge's copy.
2118 if a.Routes != nil {
2119 fresh, err := a.Routes.Get(rt.Host)
2120 if err != nil {
2121 return w, fmt.Errorf("%s: route lookup: %w", rt.Host, err)
2122 }
2123 if fresh == nil {
2124 return w, fmt.Errorf("%s: no route", rt.Host)
2125 }
2126 if fresh.DBPath != rt.DBPath || fresh.Socket != rt.Socket {
2127 log.Printf("activator: %s: route refreshed: db %q -> %q socket %q -> %q", rt.Host, rt.DBPath, fresh.DBPath, rt.Socket, fresh.Socket)
2128 }
2129 // The refreshed row must still be the route this wake began for,
2130 // and the activator's to wake (final review F4, wakeidentity.go).
2131 if err := wakeRouteChanged(rt, *fresh, e.worker); err != nil {
2132 return w, err
2133 }
2134 rt = *fresh
2135 a.mu.Lock()
2136 if e := a.entries[rt.Host]; e != nil {
2137 e.route = rt
2138 }
2139 a.mu.Unlock()
2140 }
2141
2142 // 0.5. Refuse a fenced or conflicting row before it ever reaches Acquire
2143 // or Restore (design §6.8). The in-Start/in-Seed refusal below still
2144 // exists as the second line of defense, but by the time either of those
2145 // runs the lease is already held and Restore has already written to
2146 // disk. For a fenced row that means restoring into a database a
2147 // migration journal says is being vacated, and for a conflict row it
2148 // means writing one host's database into another host's instance
2149 // directory. Checking here, on the row this wake just refreshed against,
2150 // closes both before either write happens.
2151 if c, ok := a.backend.(Classifier); ok {
2152 if layout, reason := c.Layout(rt); layout == LayoutFenced || layout == LayoutConflict {
2153 return w, fmt.Errorf("%s: refusing to wake: %s", rt.Host, reason)
2154 }
2155 }
2156
2157 // 0.6. A failed wake whose generation is not proven stopped still holds
2158 // the host's lease, and its process may still run against the database
2159 // (zero-downtime spec §7.4). No new wake starts until the sweep proves
2160 // it (reapUnprovenWakesLocked). A legacy sidecar of the host whose
2161 // termination is unproven refuses it the same way
2162 // (reapStraySidecarsLocked).
2163 a.mu.Lock()
2164 _, unproven := a.unprovenWakes[host]
2165 unproven = unproven || len(a.straySidecars[host]) > 0
2166 a.mu.Unlock()
2167 if unproven {
2168 return w, fmt.Errorf("%s: refusing to wake: %w", host, errWakeUnproven)
2169 }
2170
2171 // 0.65. The held wake check (settings spec §4.3 rule 3): a held
2172 // instance never wakes on an installed build other than its target
2173 // (wakeCheck). Before planning and before the lease, so a refusal or a
2174 // wait for root holds nothing. Not for a hold swap's pinned wake: the
2175 // swap chose e.pin from root's install. A route whose backend can exec a
2176 // named file (identityFor) may run a retained copy of the target
2177 // directly; any other waits for root. After a wait the row is re-read
2178 // as step 0 reads it, so the planner sees the build root just installed.
2179 var (
2180 held string
2181 heldID *release.StagedIdentity
2182 )
2183 if e.pin == nil {
2184 var err error
2185 held, heldID, err = a.wakeCheck(ctx, host, identityFor(a.backend, rt) != nil)
2186 if err != nil {
2187 return w, fmt.Errorf("%s: %w", host, err)
2188 }
2189 if held != "" && heldID == nil && a.Routes != nil {
2190 fresh, err := a.Routes.Get(host)
2191 if err != nil {
2192 return w, fmt.Errorf("%s: route lookup: %w", host, err)
2193 }
2194 if fresh == nil {
2195 return w, fmt.Errorf("%s: no route", host)
2196 }
2197 // The wait for root can be long: the row read after it must
2198 // still be the route this wake began for, as at step 0 (final
2199 // review F4, wakeidentity.go).
2200 if err := wakeRouteChanged(rt, *fresh, e.worker); err != nil {
2201 return w, err
2202 }
2203 rt = *fresh
2204 a.mu.Lock()
2205 if e := a.entries[rt.Host]; e != nil {
2206 e.route = rt
2207 }
2208 a.mu.Unlock()
2209 }
2210 }
2211
2212 // 0.7. Which build this wake boots (zero-downtime spec §8.2). A hold
2213 // swap's pinned wake (e.pin) already knows; everything else asks the
2214 // planner, off a.mu, before the lease: a refused plan takes no lease.
2215 var plan WakePlan
2216 if a.Planner != nil && e.pin == nil {
2217 // An earlier planned wake's outcome still on its way to the
2218 // scheduler is waited for, bounded (final review F3): planning
2219 // before its failure record exists would pick the failed build. A
2220 // wait that ends first refuses the wake before the planner and the
2221 // lease (final review wave 2, N1): nothing is planned or launched.
2222 if err := a.awaitOutcomes(ctx, host); err != nil {
2223 return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
2224 }
2225 plan = a.Planner(ctx, rt)
2226 if plan.Refuse != "" {
2227 return w, fmt.Errorf("%s: refusing to wake: %s", host, plan.Refuse)
2228 }
2229 a.mu.Lock()
2230 e.plan, e.firstWake = plan, plan.FirstWake
2231 a.mu.Unlock()
2232 }
2233 // 0.8. A staged build, planned or pinned, is published before anything
2234 // execs it (spec §4.2, §8.5): the entry now names it (and a plan's
2235 // fallback), so the write puts both in root's pin set. A failed write
2236 // refuses the wake here, before the lease: nothing is taken, nothing
2237 // runs, and no build is blamed (errStagedBuildNotPublished).
2238 if plan.planned() || e.pin != nil {
2239 if err := a.publishBeforeExec(); err != nil {
2240 return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
2241 }
2242 }
2243
2244 // 1. The lease first: everything after this line is done as the prefix's
2245 // single writer. Contention is not retried here — on one box it means a
2246 // previous incarnation hasn't released; tell the client to come back after
2247 // the TTL instead of stalling its request.
2248 w.steps.mark(a.now(), "checks")
2249 *w.stopOutcome = stopObservation{}
2250 lease, err := a.backend.Acquire(ctx, rt)
2251 if err != nil {
2252 if lease != nil {
2253 w.stopOutcome.uncertain = true
2254 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2255 }
2256 var held *errLeaseHeld
2257 if errors.As(err, &held) {
2258 log.Printf("activator: LEASE HELD for %s by %q — a previous incarnation hasn't released; refusing to wake (retry after %s)",
2259 host, held.holder, held.retryAfter)
2260 }
2261 return w, err
2262 }
2263 w.steps.mark(a.now(), "lease")
2264
2265 if ctx.Err() != nil || !a.wakeAllowed(host) {
2266 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2267 return w, ErrFenced
2268 }
2269 // 1.5. Flag off, the host cgroup is proven empty under the lease before
2270 // anything is restored or started (Task 8 fix round 1, finding 4). The
2271 // stop that released the lease proved it, but an agent restart forgets
2272 // a pending stop and its kill marks, so after the TTL a leftover can
2273 // still populate it. Unproven, the wake is refused and the lease is left
2274 // to its TTL, as an unproven seed leaves it: the leftover may still
2275 // write the database.
2276 if !a.gens {
2277 if err := a.wakeHostProven(ctx, e, rt); err != nil {
2278 log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
2279 w.stopOutcome.uncertain = true
2280 return w, fmt.Errorf("%s: refusing to wake: %w", host, err)
2281 }
2282 }
2283 // 2. Restore the DB from S3 when the local copy is gone — the restore must
2284 // happen *under* the lock (restoring without it could later shadow newer
2285 // S3 state written by the true holder). When it's still on disk it is the
2286 // warm cache: on one box, with the lease held, the local copy can only be
2287 // what we ourselves checkpointed at the last sleep. That holds only while
2288 // a region has one box. Claim ruling R6 ends it (region HA spec 1 H13):
2289 // once wakes are admitted by hostclaim's AdmitWriter, every admitted
2290 // writer purges its local copy and restores from the certified remote,
2291 // and no marker is ever read to skip that.
2292 fresh, err := a.backend.Restore(ctx, rt)
2293 if err != nil {
2294 if errors.Is(err, ErrSeedUnproven) {
2295 // An earlier seed's processes are still not proven gone (its
2296 // <db>.seed-unproven obligation, Task 8): they may still write
2297 // the database, so the lease is left to its TTL, as the seed's
2298 // own ErrSeedUnproven leaves it (step 3).
2299 log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
2300 w.stopOutcome.uncertain = true
2301 } else {
2302 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2303 }
2304 return w, fmt.Errorf("restore %s: %w", host, err)
2305 }
2306 w.steps.mark(a.now(), "restore")
2307 if ctx.Err() != nil || !a.wakeAllowed(host) {
2308 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2309 return w, ErrFenced
2310 }
2311
2312 // 2b. One executable identity for this wake (settings spec §4.2, Astra
2313 // I1 #5): chosen now, recorded now, under the lease taken in step 1 and
2314 // before the seed and the spawn, and the only file this wake runs, its
2315 // seed included (settleWakeIdentity). A route whose backend cannot pin
2316 // an executable (identityFor nil: a unit-backed row, which systemd execs
2317 // from the live link itself) keeps today's calls and records nothing.
2318 // Any refusal here has run nothing: the lease goes back and the wake
2319 // fails, never blaming the build.
2320 //
2321 // A planned wake asks its selection gate first (checkPlan, ruling
2322 // R-REPLAN), so a replan lands before the identity is chosen and the
2323 // seed and the boot run one build. spawnPlanned asks again at the exec,
2324 // and startPinnedAt records again if that later replan changes the
2325 // build.
2326 if plan.planned() {
2327 next, err := a.checkPlan(ctx, e, rt, plan)
2328 if err != nil {
2329 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2330 return w, fmt.Errorf("start %s: %w", host, err)
2331 }
2332 plan = next
2333 }
2334 ib := identityFor(a.backend, rt)
2335 var id release.StagedIdentity
2336 if ib == nil && held != "" && !plan.planned() {
2337 // A held route whose backend cannot pin an executable (a unit row:
2338 // systemd execs the live link itself): nothing here can capture the
2339 // file, so the link is read once more just before the start. It
2340 // narrows a repoint after the wait; it cannot close it.
2341 if v := a.liveVersion(host); v != held {
2342 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2343 return w, fmt.Errorf("start %s: %w: %s is on %q, wants %s", host, errStaleLink, host, v, held)
2344 }
2345 }
2346 if ib != nil {
2347 var (
2348 idErr error
2349 wctx context.Context
2350 )
2351 id, w.pend, wctx, idErr = a.settleWakeIdentity(ctx, ib, e, rt, held, heldID)
2352 if idErr != nil {
2353 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2354 return w, fmt.Errorf("start %s: %w", host, idErr)
2355 }
2356 if id.StagedPath == "" {
2357 // A live link outside the version store with no hook (ruling
2358 // R-EXEC-NOVERSION-2): today's live-link wake, nothing recorded.
2359 ib = nil
2360 } else {
2361 // The record is pending on wctx, written at the first exec
2362 // (amadan #272).
2363 ctx = wctx
2364 }
2365 }
2366 w.steps.mark(a.now(), "identity")
2367
2368 // 3. First-boot seed, strictly between the restore and the spawn — so it
2369 // runs before Start's first health probe, and a brand-new instance is born
2370 // seeded rather than briefly serving an empty database. fresh is the
2371 // backend's one fact worth keeping: the local DB was absent AND no replica
2372 // existed, so a restore can never be mistaken for a first boot and a seed
2373 // can never run over recovered data. A seed that fails fails the WHOLE
2374 // wake (the Start-failure precedent: release the lease, surface the
2375 // error) — a half-seeded instance must not serve.
2376 //
2377 // The seed runs the wake's recorded identity (SeedExe) on every wake
2378 // kind, planned and pinned included.
2379 if fresh && rt.Seed != "" {
2380 w.stopOutcome.uncertain = true
2381 seed := func() error { return a.backend.Seed(ctx, rt) }
2382 if ib != nil {
2383 seed = func() error { return ib.SeedExe(ctx, rt, id) }
2384 }
2385 if err := seed(); err != nil {
2386 if errors.Is(err, ErrSeedUnproven) {
2387 // A SIGKILL survivor of the seed may still be writing the
2388 // database: releasing the lease here would be exactly the
2389 // write-after-release hazard the lease exists to prevent
2390 // (design §6.2), so it is left for its own TTL to expire,
2391 // the same way stop() leaves ErrStillRunning.
2392 log.Printf("activator: %s: %v; leaving the lease to its TTL", host, err)
2393 } else {
2394 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2395 }
2396 return w, fmt.Errorf("seed %s: %w", host, err)
2397 }
2398 w.stopOutcome.uncertain = false
2399 w.steps.mark(a.now(), "seed")
2400 }
2401
2402 // 4. Spawn before replication starts: a fresh instance creates its own DB
2403 // on boot, so waiting for the socket guarantees the file exists for
2404 // litestream to open. Writes that land in the gap are picked up by the
2405 // replica's first sync.
2406 //
2407 // Both doors are TIMED, because this is the half an app CANNOT see.
2408 // Backend.Start execs the binary and waits until its socket accepts, so it
2409 // contains the whole of the instance's own boot — opening SQLite, running
2410 // migrations, parsing templates — and none of the platform's lease,
2411 // restore or replication work. An app measuring its own TTFB starts the
2412 // clock after all of it (discussion #6). For a worker the figure means
2413 // less (StartWorker returns at exec, not at readiness — see below), but
2414 // it is recorded on the same terms rather than left absent, so the two
2415 // paths report through one field.
2416 //
2417 // A worker takes the other door: it binds nothing, so Start's health
2418 // probe has no socket to dial and would spend its whole timeout failing
2419 // the wake. StartWorker spawns and returns — readiness is the child's own
2420 // call over the agent socket, which is why the wake env below carries the
2421 // socket path and this wake's token.
2422 var (
2423 proc Proc
2424 spawnVer string
2425 token string
2426 )
2427 if ctx.Err() != nil || !a.wakeAllowed(host) {
2428 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2429 return w, ErrFenced
2430 }
2431 w.stopOutcome.uncertain = true
2432 spawnStart := a.now()
2433 // The layout this launch uses, read on the row it launches, right
2434 // before it (finding 6): a layout that moves before a failed wake's
2435 // cleanup must not move where its descendants are looked for.
2436 if c, ok := a.backend.(Classifier); ok {
2437 w.launchLayout, _ = c.Layout(rt)
2438 }
2439 if e.worker {
2440 // Mint this wake's token here — under the lease, after the restore,
2441 // immediately before the spawn (design §7.2's order). A mint failure
2442 // is a wake failure like any other: release and report.
2443 token, err = newWorkerToken()
2444 if err != nil {
2445 // w goes back with the error, so wake() drops step 2b's
2446 // pending build (w.pend) and records this stop's outcome.
2447 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2448 return w, fmt.Errorf("start %s: %w", host, err)
2449 }
2450 // Published onto the entry NOW, not at the awake transition: the
2451 // child's first act is to call /v1/ready over the agent socket, which
2452 // authorizes against WorkerToken — and that call is what ENDS the
2453 // settle window. A token that only appeared once the entry was awake
2454 // would make the early-ready path unreachable: the socket would
2455 // refuse the very call it is waiting for, and every wake would pay the
2456 // full window. Publishing early gives nothing away: a wake that fails
2457 // deletes its entry, so the token still dies with it.
2458 a.setWorkerToken(e, token)
2459 // No registry write (spec §7.9): the row's Version is root's install
2460 // stamp, and the build this child runs is the identity the backend
2461 // captures at the exec, not anything the child later says.
2462 //
2463 // The generations file exists before any worker runs (ruling R17).
2464 a.publishGenerationsBeforeSpawn()
2465 var (
2466 ident release.StagedIdentity
2467 pend pendingGeneration
2468 )
2469 env := workerEnv(a.agentSocket(), host, token)
2470 if plan.planned() {
2471 // A planned worker (zero-downtime spec §6.5): the named build,
2472 // behind planGate at each exec, with startWorker's publication
2473 // and the installation of the captured identity done inside
2474 // (PF8), and its readiness settled there too.
2475 proc, ident, err = a.spawnPlannedWorker(ctx, e, rt, plan, env, w, lease)
2476 if err == nil {
2477 w.planned = workerGenOf(ident)
2478 }
2479 } else {
2480 // startWorker publishes the build before its exec (ruling R19
2481 // item 3), and refuses the spawn when it cannot. With an
2482 // identity settled at step 2b it execs exactly that file
2483 // (onto): its hook publishes and gates it again, and skips the
2484 // record the wake already wrote. The second pending record of
2485 // the same build is harmless (the file lists each build once)
2486 // and is dropped as the wake's own is.
2487 var onto *release.StagedIdentity
2488 if ib != nil {
2489 onto = &id
2490 }
2491 proc, ident, pend, err = a.startWorker(ctx, rt, env, onto, nil)
2492 }
2493 spawnVer, w.spawnIdentity = ident.Version, ident
2494 if err == nil && !plan.planned() {
2495 // Installed while the entry is still waking, in place of the
2496 // pending build, in one critical section, so no snapshot sees
2497 // neither: the child is running these bytes from this moment,
2498 // settle window or not, so root must keep counting them as
2499 // referenced. The awake transition stores the same identity
2500 // again.
2501 a.mu.Lock()
2502 e.workerServing = ident
2503 a.dropPendingGenerationLocked(pend)
2504 a.mu.Unlock()
2505 _ = a.persistGenerations()
2506 }
2507 if a.gens {
2508 // The wake's generation, whether or not the worker becomes
2509 // ready (zero-downtime spec §7.4): a worker binds nothing, but
2510 // it is numbered and carries the build it execed, and
2511 // cleanupFailedWake proves its host cgroup before the lease
2512 // goes back (Worker, genChildEmpty).
2513 w.gen = Gen{N: nextGen(), Build: spawnVer, Worker: true}
2514 }
2515 } else if plan.planned() {
2516 // A planned instance wake (zero-downtime spec §8.2): the plan's
2517 // staged build, every exec behind planGate, with the first-wake
2518 // budget and fallback. w.gen carries the booted staged identity, so
2519 // the legacy sidecar below binds to it (startSidecar).
2520 var g Gen
2521 proc, spawnVer, g, err = a.spawnPlanned(ctx, e, rt, plan, w, lease)
2522 w.gen, w.planned = g, g
2523 } else if e.pin != nil { // swapsim:M3
2524 var g Gen
2525 proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin, lease)
2526 w.gen = g
2527 } else if ib != nil {
2528 // The identity step 2b recorded, as the next generation, exactly
2529 // where startGen would put it: the canonical socket (and, nested,
2530 // the original run unit) with the flag off, generation n's own
2531 // socket and unit with it on (startGenExe).
2532 var g Gen
2533 proc, g, err = a.startGenExe(ctx, ib, rt, id)
2534 spawnVer, w.gen = g.Build, g
2535 } else {
2536 var g Gen
2537 proc, g, err = a.startGen(ctx, rt)
2538 spawnVer, w.gen = g.Build, g
2539 }
2540 spawnTook := a.now().Sub(spawnStart)
2541 w.steps.mark(a.now(), "spawn")
2542 w.proc = proc
2543 if err != nil {
2544 // A start whose kill was unproven may have left its generation
2545 // running with no process handle (plan 2's startUnitAt). Its lease
2546 // is kept, process or not (P2-13).
2547 w.genUnproven = genStartUnproven(err) // swapsim:M9
2548 // A legacy run unit's start that root may still carry out (a start
2549 // job systemd queued, which cancelling the client does not cancel)
2550 // is stopped and proven gone before its lease goes (#292, P1).
2551 w.legacyUncertain = legacyStartMayRun(err)
2552 switch {
2553 case proc != nil || w.genUnproven || w.legacyUncertain:
2554 a.cleanupFailedWake(ctx, rt, w, lease)
2555 case launchNotStarted(err) && wakeStartedNothing(w):
2556 // A start its pre-exec gate refused ran nothing, and this wake
2557 // has started nothing else: its seed (step 3) has finished, and
2558 // its replica (step 5) and sidecar (step 6) come after this
2559 // spawn. Its stop is known, not uncertain (seeded pre-land
2560 // review, P2; census ruling 292fix).
2561 w.stopOutcome.notStarted()
2562 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2563 case launchNotStarted(err):
2564 // Something of this wake is running after all: it goes through
2565 // the failed wake's cleanup and its proofs, not the shortcut.
2566 a.cleanupFailedWake(ctx, rt, w, lease)
2567 default:
2568 w.stopOutcome.leaseReleased = a.giveBack(ctx, host, lease)
2569 }
2570 return w, fmt.Errorf("start %s: %w", host, err)
2571 }
2572
2573 // 4b. The settle window, workers only: the readiness check for a child
2574 // with nothing to probe. A child that dies inside it never becomes an
2575 // awake entry — the lease goes back and the wake reports, exactly as a
2576 // failed Start does, so a sidecar that crashes on boot doesn't sit in the
2577 // map holding a lease until someone notices.
2578 //
2579 // Through cleanupFailedWake, not a hand-rolled stop-and-release: that
2580 // helper is what records the stop's OUTCOME on w.stopOutcome, and a wake
2581 // that tore its own child down without recording it would leave the
2582 // stopHistory blind to exactly the case it exists for.
2583 // A planned worker's readiness already ran inside spawnPlannedWorker
2584 // (plannedWorkerReady, spec §6.5).
2585 if e.worker && !plan.planned() {
2586 if serr := a.awaitWorkerSettle(ctx, e, proc); serr != nil {
2587 a.cleanupFailedWake(ctx, rt, w, lease)
2588 return w, fmt.Errorf("start %s: %w", host, serr)
2589 }
2590 w.steps.mark(a.now(), "settle")
2591 }
2592
2593 if ctx.Err() != nil || !a.wakeAllowed(host) {
2594 a.cleanupFailedWake(ctx, rt, w, lease)
2595 return w, ErrFenced
2596 }
2597 // 5. Trade the held lease for running replication, at the route's
2598 // replication-class sync interval (the per-plan durability ladder).
2599 // The DB path rides the route; when empty the backend defaults it to
2600 // <DataDir>/<host>.db internally (Eleven defaulted here via a package
2601 // global — base dirs are backend fields now).
2602 rep, err := lease.Replicate(ctx, rt.DBPath, replicationSyncInterval(rt.Replication))
2603 w.rep = rep
2604 if err != nil {
2605 a.cleanupFailedWake(ctx, rt, w, lease)
2606 return w, fmt.Errorf("replicate %s: %w", host, err)
2607 }
2608
2609 // 5b. A seeded row's first upload, then its record (spec §3.4 step 5).
2610 // A failure here does not fail the wake: the instance is serving from
2611 // the placed file, and the next wake records it (finishPlacement).
2612 if p, ok := a.backend.(seedPlacer); ok {
2613 if err := p.SeedPlaced(ctx, rt, rep); err != nil {
2614 log.Printf("activator: %s: seed placement not recorded yet: %v", host, err)
2615 }
2616 }
2617 w.steps.mark(a.now(), "replicate")
2618
2619 // 6. Co-spawn the sidecar, last — only once the instance is healthy and
2620 // replicating, so the sidecar always boots against a live instance (its
2621 // marked requests pass through the edge while we're awake). A spawn
2622 // failure is loud but never fails the wake: members come before bots, and
2623 // the sweeper retries with backoff.
2624 w.rep, w.spawnVersion, w.spawnTook = rep, spawnVer, spawnTook
2625 w.stopOutcome = nil
2626 // A worker is nobody's parent: it IS a sidecar, so it never co-spawns one.
2627 // Skipped outright rather than left to HasSidecar's answer — the staging
2628 // file is keyed by host, and a host that happened to have one staged must
2629 // not turn a standalone sidecar into a supervisor.
2630 if e.worker {
2631 return w, nil
2632 }
2633 if ctx.Err() != nil || !a.wakeAllowed(host) {
2634 return w, nil
2635 }
2636 if a.backend.HasSidecar(rt) {
2637 w.sidecarWanted = true
2638 w.sidecarGen = w.gen.N
2639 if sc, err := a.startSidecar(ctx, e, rt, w.gen, true); err != nil {
2640 w.sidecarUncertain = true
2641 w.sidecar = sc
2642 if sc == nil {
2643 log.Printf("activator: SIDECAR for %s failed to start: %v — serving without it (the sweep retries)", host, err)
2644 } else {
2645 log.Printf("activator: SIDECAR for %s failed to start: %v — retaining returned process", host, err)
2646 }
2647 } else {
2648 w.sidecar = sc
2649 }
2650 w.steps.mark(a.now(), "sidecar")
2651 }
2652 return w, nil
2653}
2654
2655// actStopTimeout bounds a graceful instance stop (the instance's own SIGTERM
2656// drain is expected ~10s; leave headroom before the SIGKILL).
2657const actStopTimeout = 20 * time.Second
2658
2659// beginStopLocked flips an awake entry to stopping and kicks the stop off; the
2660// caller holds a.mu. In-flight requests are not waited for (the caller's
2661// SIGTERM drain inside actStopTimeout is what covers them). Callers: the
2662// idle sweep (instance has been idle past cfg.Idle), a dead child, a lost
2663// lease (open requests on either of the last two are already doomed), and
2664// Bounce (config changed — the instance may be mid-request, but the drain
2665// still gives it actStopTimeout to finish before SIGKILL). Every caller says
2666// why (the Stop* reasons in cmd.go): a run the stop interrupts tells the
2667// member (run spec §5.3).
2668func (a *Service) beginStopLocked(e *actEntry, why string) {
2669 // A hold canary's settle ends with an intentional stop, before anything
2670 // is stopped (zero-downtime spec §6.4). A process that already exited
2671 // died of its build, and its stop must not hide that.
2672 if e.settleCancel != nil && (e.proc == nil || !e.proc.Exited()) {
2673 e.settleCancel()
2674 }
2675 // The same for a planned first wake's observation (spec §6.4): an
2676 // intentional stop ends it and records nothing. A stop that began
2677 // because the child died on its own (unexpectedExit), or of a process
2678 // that has already exited, leaves it to read the exit as the build's.
2679 if e.firstWatch != nil && !e.unexpectedExit && (e.proc == nil || !e.proc.Exited()) {
2680 e.firstWatch()
2681 }
2682 e.state = actStopping
2683 e.stopWhy = why
2684 if e.cmd != nil {
2685 e.cmd.cancelPrepare() // a stop interrupts a run's download
2686 }
2687 e.stopped = make(chan struct{})
2688 go a.stop(e)
2689}
2690
2691// Owns reports whether host has a live entry in ANY state — waking,
2692// awake, or stopping. This is the external-socket advisory's predicate
2693// (issue #17): while an entry exists, whatever is serving the host's
2694// socket is this activator's own child — possibly still binding its
2695// listener, or draining SIGTERM — so any-state presence, not awake-only,
2696// is what keeps the advisory from calling the activator's own children
2697// strays.
2698func (a *Service) Owns(host string) bool {
2699 a.mu.Lock()
2700 defer a.mu.Unlock()
2701 return a.entries[host] != nil
2702}
2703
2704// IdleWindow is cfg.Idle, for the box-side reporters that must tell a member
2705// whether their schedule is tight enough to hold the instance resident. Read
2706// only; the window is fixed at construction from CARLOS_HIBERNATE_IDLE.
2707func (a *Service) IdleWindow() time.Duration { return a.cfg.Idle }
2708
2709// Bounce initiates a graceful stop of host's RUNNING instance so that
2710// its next wake — the next request, or its schedule — respawns it with
2711// a freshly composed environment. This is config delivery's restart
2712// half: the analogue of the unit path's try-restart, with the same
2713// caveat that in-flight requests are not waited for. It never starts
2714// anything: not-awake (waking, stopping, asleep, unknown) is a no-op,
2715// and the instance picks the new config up at its natural next wake.
2716//
2717// NOT kind-gated (Task 11 item 4, carried from Task 6's review) — e's
2718// entry has no branch on e.worker, so this stops an awake WORKER exactly
2719// as it stops an instance, and stop() itself is worker-aware (it skips
2720// the socket-only next-due probe, co-stops any attached sidecar first,
2721// etc.), so nothing here needs a kind check to behave correctly. The
2722// EXPLICIT decision this makes is what it deliberately does NOT do: start
2723// anything back up. For an instance that is exactly right — the next
2724// HTTP request is the wake half of the cycle, and Bounce owns only the
2725// stop half. A worker has no such request path; nothing dials one. A
2726// caller that wants a WORKER actually cycled — not just stopped — must
2727// pair this call with its own WakeWorker once Bounce returns true (see
2728// host/restart.go's cycleWorkers, the one caller that does): an
2729// always-on (Hibernate: false) worker left unpaired is stranded until the
2730// agent restarts (the boot pass that starts one runs once, at arming,
2731// never again), and a hibernating one stays silent until its schedule
2732// next comes due — neither of which "carlos restart" asked for. This is
2733// why SyncExec (appconfig/syncexec.go) never calls Bounce for a sidecar
2734// route at all: an automatic, unpaired bounce on every ordinary
2735// config-sync pass would carry exactly that risk with no operator command
2736// behind it.
2737func (a *Service) Bounce(host string) bool {
2738 a.mu.Lock()
2739 defer a.mu.Unlock()
2740 e := a.entries[host]
2741 if e != nil && e.probeHeld {
2742 // A held recovery probe runs the moves-held inputs, not delivered
2743 // config: the recovery owns its process until the source opens.
2744 return false
2745 }
2746 if e != nil && e.state == actWaking && !e.worker {
2747 // Not dropped: held for the end of the wake (bounceAfterWake). Still
2748 // reported as no stop, because none has happened — the callers that
2749 // count stops, or fall through to systemctl on false, see what they
2750 // always saw. Workers keep the old rule: their restart pairs a stop
2751 // with its own WakeWorker only on true (host/restart.go cycleWorkers),
2752 // and a stop held for later would park one with nothing to start it.
2753 e.bounceAfterWake = true
2754 return false
2755 }
2756 if e == nil || e.state != actAwake {
2757 return false
2758 }
2759 return a.bounceLocked(e)
2760}
2761
2762// bounceLocked is Bounce's stop for an entry the caller has found awake,
2763// under a.mu.
2764func (a *Service) bounceLocked(e *actEntry) bool {
2765 // Both Bounce and BounceAdopted must preserve the admitted recovery.
2766 if e.probeHeld {
2767 return false
2768 }
2769 msg := "instance restarting (carlos restart)"
2770 if e.worker {
2771 msg = "sidecar restarting (carlos restart)"
2772 }
2773 emitEvent(a.Logs, e.route, "info", msg)
2774 a.beginStopLocked(e, StopRestart)
2775 return true
2776}
2777
2778// startSidecar is every legacy co-spawned sidecar spawn: the wake tail,
2779// late attachment and the supervisor (sweepOnce), and a swap's replacement
2780// (zero-downtime spec §7.5). It runs its generation's recorded file by
2781// path whatever CARLOS_SWAP says (ruling R-EXEC-SIDECAR): a flag-off
2782// sidecar started from the live link could run a different build than its
2783// instance, and a nested one would rebind the host's shared app mount to
2784// the link. A generation that names no staged file (an instance woken
2785// before this binary, or the recovery probe's) has the live link resolved
2786// once, and that file run by path. On a declined box whose gate refuses
2787// that build, the instance's own build runs instead when the gate passes
2788// it (sidecarOwnBuild); otherwise the refusal stands. Each exec is
2789// recorded first (settings spec §4.2): the wake's co-spawn skips the build its wake already recorded, while a restart or
2790// a replacement writes it. A refused record starts nothing and is
2791// returned like any start failure, which never fails a wake. Only a route
2792// no backend can pin a file for (a unit row) keeps today's StartSidecar,
2793// live link and all, recording nothing.
2794//
2795// Immediately before the spawn it asks the gate's ownership check (spec
2796// §7.4): e must still be the host's, unfenced, with no shutdown, and (awake)
2797// its lease held; waking is the wake tail, whose lease is the wake's own.
2798// The check runs whatever the switch says (an accepted dormancy exception,
2799// recorded with the controller's rulings for Task 7): a host that does not
2800// own itself gets no co-spawned sidecar. While a stray sidecar of the host
2801// is tracked (its termination unproven) nothing starts beside it. The
2802// check is asked again after the record and before the exec, as
2803// startPinnedAt's gate is: a lease lost or a fence raised while the build
2804// state was written execs nothing.
2805//
2806// A resolved live link outside the version store with no BeforeExec set
2807// keeps today's live-link start, as the wake does (ruling
2808// R-EXEC-NOVERSION-2); with one set the record refuses it.
2809func (a *Service) startSidecar(ctx context.Context, e *actEntry, rt registry.Route, g Gen, waking bool) (Proc, error) {
2810 owned := func() error {
2811 a.mu.Lock()
2812 defer a.mu.Unlock()
2813 err := a.ownedLocked(e, waking)
2814 if err == nil && len(a.straySidecars[rt.Host]) > 0 {
2815 err = errSidecarUnproven
2816 }
2817 return err
2818 }
2819 if err := owned(); err != nil {
2820 return nil, err
2821 }
2822 s, ok := a.backend.(SidecarGenStarter)
2823 if !ok {
2824 return a.backend.StartSidecar(ctx, rt)
2825 }
2826 if g.StagedPath == "" {
2827 ib := identityFor(a.backend, rt)
2828 if ib == nil {
2829 return a.backend.StartSidecar(ctx, rt)
2830 }
2831 id, err := ib.LiveIdentity(ctx, rt)
2832 if err != nil {
2833 return nil, err
2834 }
2835 if id.Version == "" && a.BeforeExec == nil {
2836 return a.backend.StartSidecar(ctx, rt)
2837 }
2838 // The resolved file runs by path, past the backend's live-link
2839 // check (startSidecarChild), so a declined box's live exec gate is
2840 // asked here, before the record (askLiveExecGate). A refusal starts
2841 // nothing, as the backend's would, and the supervisor backs off.
2842 // A refused link whose build is not the instance's own falls back
2843 // to the instance's build (sidecarOwnBuild), so a sidecar is not
2844 // stranded beside an instance the gate already let run.
2845 if id.Version != "" {
2846 if err := a.askLiveExecGate(rt, id.Version); err != nil {
2847 own, ok := a.sidecarOwnBuild(rt, g.Build, id.Version)
2848 if !ok {
2849 return nil, err
2850 }
2851 id = own
2852 }
2853 }
2854 g.Build, g.StagedPath, g.Digest, g.Demand = id.Version, id.StagedPath, id.Digest, id.Demand
2855 // The resolved build's own tier (ruling R-TIER), never the live
2856 // tier read later.
2857 if gt, ok := a.backend.(interface {
2858 GenTier(rt registry.Route, build string) cgroup.Limit
2859 }); ok {
2860 g.Tier = gt.GenTier(rt, id.Version)
2861 }
2862 }
2863 if err := a.recordBuild(ctx, rt.Host, g.Build); err != nil {
2864 return nil, err
2865 }
2866 if err := owned(); err != nil {
2867 return nil, err
2868 }
2869 return s.StartSidecarGen(ctx, rt, g)
2870}
2871
2872// sidecarOwnBuild is an unstaged sidecar's recovery when the declined gate
2873// refuses the build its live link names (readiness 2 Task 5): root
2874// repointed the link at a held build while the instance kept serving its
2875// own, and the supervisor would otherwise retry the held build forever. own
2876// is the generation's build (the instance's), refused the link's. When own
2877// is a different, named build, its staged copy is named and hashed through
2878// the backend (buildIdentifier) and the gate is asked about it too; both
2879// passing returns that identity, which startSidecar then records and runs
2880// by path as it would the link's. Anything else reports false, and the
2881// caller keeps the original refusal. A held build is never run: the gate
2882// has passed whatever this returns. The hash runs outside a.mu, as the
2883// link's own capture does.
2884func (a *Service) sidecarOwnBuild(rt registry.Route, own, refused string) (release.StagedIdentity, bool) {
2885 if own == "" || own == refused {
2886 return release.StagedIdentity{}, false
2887 }
2888 bi, ok := a.backend.(buildIdentifier)
2889 if !ok {
2890 return release.StagedIdentity{}, false
2891 }
2892 id, err := bi.BuildIdentity(rt, own)
2893 if err != nil {
2894 log.Printf("activator: %s: sidecar not started: this box will not start %s, the build its live link names, and the instance's own build %s cannot be named: %v", rt.Host, refused, own, err)
2895 return release.StagedIdentity{}, false
2896 }
2897 if id.Version != own {
2898 return release.StagedIdentity{}, false
2899 }
2900 if err := a.askLiveExecGate(rt, id.Version); err != nil {
2901 log.Printf("activator: %s: sidecar not started: this box will not start %s, the build its live link names, nor the instance's own build %s: %v", rt.Host, refused, own, err)
2902 return release.StagedIdentity{}, false
2903 }
2904 log.Printf("activator: %s: sidecar starting on the instance's own build %s (%s): this box will not start %s, the build its live link names", rt.Host, id.Version, id.StagedPath, refused)
2905 return id, true
2906}
2907
2908// replaceSidecar is §7.5 step 8: a legacy co-spawned sidecar is restarted,
2909// stop-then-start, onto the generation just switched to. It claims the
2910// supervisor's interlock (sidecarStarting, sidecarStartDone) for the whole
2911// replacement. A supervised restart cannot race it, and stop() waits on it
2912// exactly as it waits on a supervised restart.
2913//
2914// A supervised start already in flight captured the generation it read
2915// when it began, which may be the one this swap just retired. Skipping the
2916// replacement because of it would leave the old build's sidecar beside the
2917// new server with nothing pending, so replaceSidecar joins it (its
2918// sidecarStartDone, bounded by the supervisor's DeadlineControl) and then
2919// replaces whatever it installed.
2920func (a *Service) replaceSidecar(e *actEntry) {
2921 a.mu.Lock()
2922 for e.sidecarStarting && a.entries[e.host] == e && e.state == actAwake {
2923 wait := e.sidecarStartDone
2924 a.mu.Unlock()
2925 if wait != nil {
2926 <-wait
2927 }
2928 a.mu.Lock()
2929 }
2930 if a.entries[e.host] != e || e.state != actAwake || !e.sidecarWanted || len(a.straySidecars[e.host]) > 0 {
2931 // A stray still unproven: the reaper proves it first, and the
2932 // supervisor then starts the current generation's sidecar.
2933 a.mu.Unlock()
2934 return
2935 }
2936 if e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen == e.gen.N {
2937 // Already on the current generation (the sweep's retry got here
2938 // first): nothing to replace.
2939 a.mu.Unlock()
2940 return
2941 }
2942 e.sidecarStarting = true
2943 done := make(chan struct{})
2944 e.sidecarStartDone = done
2945 old, rt, g := e.sidecar, e.route, e.gen
2946 a.mu.Unlock()
2947 defer close(done) // only after the old sidecar is proven gone, or retained
2948 if old != nil {
2949 ctx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
2950 if err := old.Stop(ctx); err != nil && !errors.Is(err, ErrStillRunning) {
2951 log.Printf("activator: stop sidecar %s for its replacement: %v", e.host, err)
2952 }
2953 cancel()
2954 if !a.sidecarTerminated(rt, old) {
2955 // Unproven (spec §7.4, §7.5): the leader still runs, or something
2956 // it forked is left in its cgroup. It becomes a tracked stray, in
2957 // the same critical section that takes it off the entry, so no
2958 // supervisor sees the entry without a sidecar and nothing to wait
2959 // for. Nothing starts beside it, a stop keeps the lease
2960 // (releasable), and the reaper proves it.
2961 a.mu.Lock()
2962 if a.entries[e.host] == e {
2963 e.sidecar = nil
2964 e.sidecarStarting, e.sidecarStartDone, e.sidecarUncertain = false, nil, true
2965 }
2966 a.trackStraySidecarLocked(rt, old)
2967 a.mu.Unlock()
2968 log.Printf("activator: SIDECAR for %s is not proven gone for its replacement; tracking it (the sweep re-proves it)", e.host)
2969 return
2970 }
2971 }
2972 ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
2973 sc, err := a.startSidecar(ctx, e, rt, g, false)
2974 cancel()
2975 var discard Proc
2976 defer func() { a.discardSidecar(rt, discard) }() // after the unlock, before done closes
2977 a.mu.Lock()
2978 defer a.mu.Unlock()
2979 if a.entries[e.host] != e {
2980 discard = sc
2981 return
2982 }
2983 e.sidecarStarting = false
2984 e.sidecarStartDone = nil
2985 if e.state != actAwake {
2986 // A terminal stop is joined on this replacement (done). It stops
2987 // e.sidecar after the join and proves it: install sc for it.
2988 e.sidecar, e.sidecarGen = sc, g.N
2989 return
2990 }
2991 e.sidecar, e.sidecarStarted, e.sidecarBackoff, e.sidecarGen = sc, a.now(), 0, g.N
2992 if err != nil {
2993 e.sidecarUncertain = true
2994 log.Printf("activator: SIDECAR for %s failed to start on generation %d: %v (the sweep retries)", e.host, g.N, err)
2995 }
2996}
2997
2998// BounceAdopted is Bounce for an ADOPT pass's restart callback, with one
2999// thing Bounce cannot know taken into account: the callback fires after the
3000// pass has repointed live/<host>, so a child already exec'd from the build the
3001// symlink names has nothing to pick up, and stopping it buys a drain and a
3002// wake for no change of binary.
3003//
3004// That child exists whenever something else finished the adoption first. A
3005// root pass adopts, ConvergeAdopted cycles the instance, a request wakes it
3006// onto the new build — and a slower pass that read the old symlink before all
3007// of that reaches its own restart afterwards. Bounce would stop the
3008// replacement; this leaves it alone.
3009//
3010// Everything else is Bounce: same no-op on a host that is not awake, same
3011// event, same stop, and it applies to a worker exactly as Bounce does. An
3012// unknown spawn version or an unreadable symlink bounces — with nothing to
3013// compare, the safe reading of "adopted" is "the child may be stale".
3014//
3015// `carlos restart` and config delivery keep the unconditional Bounce: there
3016// the request IS the reason, whatever build is running.
3017func (a *Service) BounceAdopted(host string) bool {
3018 a.mu.Lock()
3019 defer a.mu.Unlock()
3020 e := a.entries[host]
3021 if e == nil || e.state != actAwake {
3022 return false
3023 }
3024 if e.spawnVersion != "" {
3025 if cur, err := a.liveVer(host); err == nil && cur == e.spawnVersion {
3026 return false
3027 }
3028 }
3029 return a.bounceLocked(e)
3030}
3031
3032// Drifted lists the awake INSTANCES whose child was exec'd from a build
3033// live/<host> no longer names, each with the version the symlink names now.
3034// It is the cheap half of ConvergeAdopted below: a caller polling at seconds
3035// cadence asks this first and, on the ordinary tick where nothing has shipped,
3036// gets nothing back and reads no registry row at all. One readlink per awake
3037// instance, the same read the sweep makes.
3038//
3039// A read error, an unknown spawn version and an unknown symlink target are all
3040// "not drifted" — the sweep's own rule (ConvergeDecision): with nothing to
3041// compare there is nothing to act on. Workers are never listed; see
3042// ConvergeAdopted for why they stay with the sweep.
3043func (a *Service) Drifted() map[string]string {
3044 a.mu.Lock()
3045 defer a.mu.Unlock()
3046 var out map[string]string
3047 for _, e := range a.entries {
3048 if e.worker || e.state != actAwake || e.spawnVersion == "" || a.fences[e.host] != 0 {
3049 continue
3050 }
3051 cur, err := a.liveVer(e.host)
3052 if err != nil || cur == "" || cur == e.spawnVersion {
3053 continue
3054 }
3055 if out == nil {
3056 out = map[string]string{}
3057 }
3058 out[e.host] = cur
3059 }
3060 return out
3061}
3062
3063// ConvergeAdopted cycles host's awake instance onto ver, a version an adopt
3064// pass has FINISHED adopting for it, without the sweep's ConvergeGrace. It
3065// reports whether it started a cycle.
3066//
3067// It exists because the adoption a real box performs is delegated. The agent
3068// runs sandboxed and cannot write the version store, so the root pass stages,
3069// repoints live/<host> and stamps the route — and cycles no exec child,
3070// because the children belong to this process. The agent's own pass then
3071// finds the host up to date and never reaches its restart callback, which
3072// left the sweep's grace as the only thing standing between a deploy and the
3073// build it shipped (discussion #231). This is the restart half of that
3074// adoption, performed by the one process that can.
3075//
3076// The grace is safe to skip HERE and nowhere else, because of what the caller
3077// vouches for: ver is the version the route is STAMPED with, and the stamp is
3078// written only after the adopting pass is done with the host — past its own
3079// restart and any rollback. The sweep cannot know that; it sees a symlink
3080// move and waits to learn whether it holds. ver must also still be what the
3081// symlink names, checked here under the lock: a stamp the symlink has moved
3082// on from is a newer adoption in flight, and its own stamp will bring the
3083// caller back.
3084//
3085// Like Bounce it never starts anything — an instance's next request is the
3086// wake half — and it is the same stop the sweep's convergence performs:
3087// SIGTERM drain, final checkpoint, lease release.
3088//
3089// Instances only. A worker's registry version is what its child REPORTS
3090// (cleared on every respawn), not what adoption stamped, so the agreement
3091// this rests on does not exist for one; the sweep's convergeWorkerLocked,
3092// behind the grace, remains a worker's path onto a new build.
3093func (a *Service) ConvergeAdopted(host, ver string) bool {
3094 a.mu.Lock()
3095 defer a.mu.Unlock()
3096 e := a.entries[host]
3097 if e == nil || e.worker || e.probeHeld || e.state != actAwake || a.fences[host] != 0 {
3098 return false
3099 }
3100 if ver == "" || e.spawnVersion == "" || e.spawnVersion == ver {
3101 return false
3102 }
3103 if cur, err := a.liveVer(host); err != nil || cur != ver {
3104 return false
3105 }
3106 log.Printf("activator: %s serving %s but %s is adopted — cycling onto the new build", host, e.spawnVersion, ver)
3107 a.beginStopLocked(e, StopRelease)
3108 return true
3109}
3110
3111// --- standalone sidecars (workers) -------------------------------------------
3112//
3113// A worker is a member of this same state machine with the request path
3114// removed. Nothing routes to a sidecar — the edge refuses the row on kind, and
3115// ServeHTTP gates on `Kind == "instance"` — so no request can ever wake one:
3116// every wake arrives from the due-time sweep, an operator command, or the
3117// agent's boot pass, and its only sign of life is contact over the agent
3118// socket. beginRequest, BeginSidecarRequest and ServeHTTP are untouched.
3119
3120// newWorkerEntry builds the waking entry for a standalone sidecar: the route's
3121// own ceilings (0 = platform default / no ceiling) and the channel its settle
3122// window waits on. The wake's token is NOT minted here — it is minted inside
3123// doWake, under the lease and immediately before the spawn, and published onto
3124// the entry there. Every field this does set is written BEFORE the entry is
3125// published to a.entries and the wake goroutine starts, so the wake path may
3126// read them without a.mu; nothing rewrites them afterwards.
3127func (a *Service) newWorkerEntry(rt registry.Route) *actEntry {
3128 return &actEntry{
3129 host: rt.Host,
3130 route: rt,
3131 state: actWaking,
3132 ready: make(chan struct{}),
3133 worker: true,
3134 workerIdle: time.Duration(rt.IdleSeconds) * time.Second,
3135 workerMaxRun: time.Duration(rt.MaxRunSeconds) * time.Second,
3136 readyc: make(chan struct{}),
3137 }
3138}
3139
3140// randRead is crypto/rand's entropy, behind a seam so a test can make minting
3141// fail. Nothing else in this package should read entropy.
3142var randRead = rand.Read
3143
3144// newWorkerToken mints the per-wake agent-socket credential. A fresh one every
3145// wake is the point: a token from a previous incarnation of the child never
3146// matches, so an orphan that survived a stop cannot keep talking to the agent
3147// as though it were the live worker.
3148//
3149// It returns an error rather than panicking. The agent supervises every route
3150// on the box — one host that cannot mint a token must fail ITS wake (the lease
3151// released, like any other wake failure), never take the process, and every
3152// other tenant on the box, down with it. Deliberately no fallback source: an
3153// empty token would authorize nothing, but a predictable one would authorize
3154// everyone.
3155func newWorkerToken() (string, error) {
3156 var b [32]byte
3157 if _, err := randRead(b[:]); err != nil {
3158 return "", fmt.Errorf("mint worker token: %w", err)
3159 }
3160 return hex.EncodeToString(b[:]), nil
3161}
3162
3163// errWorkerExitedEarly: the child died inside its settle window. That is a
3164// failed wake — a crash on boot (bad argv, missing config, panic at start) is
3165// exactly what the instance path's health probe would have caught.
3166var errWorkerExitedEarly = errors.New("worker exited during its settle window")
3167
3168// awaitWorkerSettle is the socketless child's health probe: it is up once it
3169// says so over the agent socket, or once it has survived sidecarSettle without
3170// exiting. Real time, not a.now — like waitHealthy, this bounds a real
3171// process's first seconds, and the test clock exists to move idle windows, not
3172// to model a boot.
3173func (a *Service) awaitWorkerSettle(ctx context.Context, e *actEntry, proc Proc) error {
3174 window := sidecarSettle
3175 if window <= 0 {
3176 return nil
3177 }
3178 deadline := time.NewTimer(window)
3179 defer deadline.Stop()
3180 tick := time.NewTicker(settlePoll(window))
3181 defer tick.Stop()
3182 for {
3183 select {
3184 case <-e.readyc:
3185 return nil // the child reported ready: up, early
3186 case <-deadline.C:
3187 if proc.Exited() {
3188 return errWorkerExitedEarly
3189 }
3190 return nil
3191 case <-tick.C:
3192 if proc.Exited() {
3193 return errWorkerExitedEarly
3194 }
3195 case <-ctx.Done():
3196 return ctx.Err()
3197 }
3198 }
3199}
3200
3201// WakeWorker brings a sidecar up, or does nothing if it is already awake.
3202// It is the request-less twin of beginRequest: a sidecar has no socket, so no
3203// request can ever wake one — every wake arrives from the due-time sweep, an
3204// operator command, or the agent's boot pass.
3205//
3206// The route arrives whole and is the only input: the idle window and the run
3207// ceiling ride on the row, so no caller (and no sweep tick) needs a bucket
3208// read to decide a worker has outrun its ceiling.
3209//
3210// Idempotent: concurrent and repeated calls coalesce on the ONE wake, exactly
3211// as concurrent requests do — three calls produce one child.
3212func (a *Service) WakeWorker(rt registry.Route) error {
3213 deadline := time.NewTimer(wakeWaitBound)
3214 defer deadline.Stop()
3215 if err := a.awaitStartup(a.lifeCtx, deadline.C); err != nil {
3216 return err
3217 }
3218 arrival := time.Now()
3219 extended := false
3220 for {
3221 a.mu.Lock()
3222 // Refused once a shutdown has announced itself. Shutdown sets the flag
3223 // BEFORE it snapshots the entry map, so a wake dispatched after that
3224 // point — a sidecar-wake leg tick, a boot pass, an operator command
3225 // landing mid-drain — would insert a fresh entry that snapshot never
3226 // sees: a child holding the tenant's lease past the agent's own exit,
3227 // with nothing left to stop it. The respawn path checks the same flag
3228 // for the same reason.
3229 if a.shuttingDown {
3230 a.mu.Unlock()
3231 return fmt.Errorf("wake worker %s: the agent is shutting down", rt.Host)
3232 }
3233 e := a.entries[rt.Host]
3234 if e == nil {
3235 // Asleep → kick off ONE wake and FALL THROUGH to the switch still
3236 // holding the lock, waiting on the entry we just created. Unlocking
3237 // and re-reading a.entries[host] here is the #184 bug: a wake that
3238 // fails deletes its entry, so the initiator would find nothing and
3239 // start a second one.
3240 e = a.newWorkerEntry(rt)
3241 a.entries[rt.Host] = e
3242 go a.wake(e)
3243 } else if !e.worker {
3244 // A live entry for this host that is NOT a worker means the route
3245 // changed kind under a running instance (or two rows collided).
3246 // Refuse rather than adopt someone else's child: its proc, lease
3247 // and idle discipline are the instance path's.
3248 a.mu.Unlock()
3249 return fmt.Errorf("wake worker %s: an instance entry already holds this host", rt.Host)
3250 }
3251
3252 switch e.state {
3253 case actAwake:
3254 // Same stand-down beginRequest makes: a lost lease or a dead child
3255 // found at wake time turns into a stop, and the loop then re-wakes
3256 // it. Without this, a boot pass or an operator command would report
3257 // success for a worker whose lease is gone — the one answer that
3258 // makes a broken sidecar look healthy. A respawn in flight is not a
3259 // dead child: its proc is mid-replacement, so leave it be.
3260 if !e.workerRespawning {
3261 if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() {
3262 why := StopExited
3263 if repErr != nil {
3264 log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr)
3265 why = StopLeaseLost
3266 } else {
3267 log.Printf("activator: worker %s exited — recycling", e.host)
3268 e.unexpectedExit = true // not an intentional stop (spec §6.4)
3269 }
3270 a.beginStopLocked(e, why)
3271 a.mu.Unlock()
3272 continue
3273 }
3274 }
3275 a.mu.Unlock()
3276 return nil // already up: a no-op, not a second child
3277
3278 case actWaking:
3279 ready := e.ready
3280 a.mu.Unlock()
3281 // A worker's caller gets a first wake's extension as much as a
3282 // request does (Astra R3 flag triage T10): without it a
3283 // fallback that succeeds after the ordinary bound was reported
3284 // as a timeout.
3285 if err := a.awaitWake(context.Background(), e, ready, deadline, arrival, &extended); err != nil {
3286 return err
3287 }
3288 // A failed wake reports to every coalesced caller; retrying
3289 // here would turn one bad wake into a storm (beginRequest's
3290 // rule, for the same reason).
3291 return e.err
3292
3293 case actStopping:
3294 stopped := e.stopped
3295 a.mu.Unlock()
3296 select {
3297 case <-stopped:
3298 continue // entry is gone; loop wakes it fresh
3299 case <-deadline.C:
3300 return errWakeTimeout
3301 }
3302 }
3303 }
3304}
3305
3306// agentSocket is where a worker child dials back.
3307func (a *Service) agentSocket() string {
3308 if a.cfg.AgentSocket == "" {
3309 return DefaultAgentSocket
3310 }
3311 return a.cfg.AgentSocket
3312}
3313
3314// workerEnv is what a spawn hands the child: where to call back, who it is,
3315// and this incarnation's credential. The backend applies it OVER the tenant's
3316// own env file so a tenant cannot forge its own token (StartWorker's
3317// contract). A free function of plain strings, not a method on the entry, so
3318// the two callers — the wake (off-lock, with a token still in a local) and the
3319// crash restart (under a.mu) — both compose it without touching entry state
3320// they don't own.
3321func workerEnv(socket, host, token string) map[string]string {
3322 name, _, _ := strings.Cut(host, ".") // the sidecar's declared name
3323 return map[string]string{
3324 "CARLOS_AGENT_SOCKET": socket,
3325 "CARLOS_SIDECAR_TOKEN": token,
3326 "CARLOS_SIDECAR_HOST": host,
3327 "CARLOS_SIDECAR_NAME": name,
3328 }
3329}
3330
3331// restartWorkerLocked replaces a worker's dead child — the caller holds a.mu
3332// and has already established that e is an awake worker whose proc has
3333// exited, and that the exit is one that must be restarted: ANY exit for an
3334// always-on worker (`hibernate: false` — the record exists ⇒ the process
3335// runs), or a non-zero exit for a hibernating one (a crash; a clean exit is
3336// that bot finishing its work, and parks instead).
3337//
3338// Restart=always without a supervisor, the co-spawn path's rule applied to a
3339// standalone member: doubling backoff from sidecarRestartBase to
3340// sidecarRestartMax, and a run that lasted sidecarHealthyRun earns a fresh
3341// one, so a crash a day never escalates to the cap. The first crash respawns
3342// immediately and ARMS the delay for the next one — a crash loop therefore
3343// paces itself instead of hot-looping the box. The pacing is decided here,
3344// under the lock; the exec itself is not (see respawnWorkerLocked).
3345//
3346// In place, not through a stop and re-wake: the lease and the replication are
3347// still ours (only the child died), and design §7.2 says an always-on worker
3348// runs with no lease churn. A hibernating worker restarted here keeps its
3349// lease for the same reason — with the consequence, recorded at the sweep's
3350// exited arm, that a crash LOOP holds that lease until an operator or the
3351// record intervenes.
3352func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) {
3353 if !e.spawnedAt.IsZero() && now.Sub(e.spawnedAt) >= sidecarHealthyRun {
3354 e.workerBackoff = 0
3355 }
3356 if now.Before(e.workerRetryAt) {
3357 return // still backing off; the child stays dead until it elapses
3358 }
3359 if e.workerBackoff == 0 {
3360 e.workerBackoff = sidecarRestartBase
3361 } else {
3362 e.workerBackoff = min(e.workerBackoff*2, sidecarRestartMax)
3363 }
3364 e.workerRetryAt = now.Add(e.workerBackoff)
3365 log.Printf("activator: worker %s exited — restarting (backoff now %s)", e.host, e.workerBackoff)
3366 // nil: this child is already dead, so there is nothing to retire first.
3367 a.recycleWorkerLocked(e, nil, "exited")
3368}
3369
3370// convergeWorkerLocked replaces an always-on worker's child with one exec'd
3371// from the build live/<host> now points at — the caller holds a.mu and has
3372// already established the drift persisted past ConvergeGrace and that the
3373// per-sweep budget allows it. The old child is retired first: it is alive, and
3374// two incarnations of an outbound worker would double its work.
3375//
3376// nil: a converge execs what live/<host> names now. Until plan 5 queues it as
3377// a paced release attempt, this is the one path that moves a worker to a
3378// different installed build.
3379//
3380// build is the build live/<host> names, as the sweep just read it. On a box
3381// that declined CARLOS_SWAP=overlap and honours failure records or hold-ran
3382// markers, the live exec gate is asked about it first (Task 9): the old
3383// child is retired before the spawn, so a refusal at the spawn would leave
3384// the worker down. Refused, nothing is stopped, the worker keeps running its
3385// current build, and it reports false: no converge happened, and none of the
3386// sweep's budget is spent.
3387func (a *Service) convergeWorkerLocked(e *actEntry, build string) bool {
3388 if a.liveGateRefusesLocked(e, build, "it keeps running "+e.spawnVersion) {
3389 return false
3390 }
3391 log.Printf("activator: worker %s serving %s but live symlink is now %s — respawning onto the new build", e.host, e.spawnVersion, build)
3392 a.respawnWorkerLocked(e, e.proc, "converging onto a new build", nil)
3393 return true
3394}
3395
3396// liveGateRefusesLocked asks the live exec gate (SetLiveExecGate) whether
3397// e may move onto build (a worker's converge or recycle, or the sweep's
3398// cycle of an instance or hibernating worker), the build live/<host> names,
3399// and reports a refusal, which it logs once per host and build. No gate
3400// (every box but a declined one that honours something) refuses nothing.
3401// The gate reads only memory and the row. The caller holds a.mu.
3402func (a *Service) liveGateRefusesLocked(e *actEntry, build, left string) bool {
3403 if a.liveExecGate == nil {
3404 return false
3405 }
3406 err := a.liveExecGate(e.route, build)
3407 if err == nil {
3408 delete(a.liveGateSaid, e.host)
3409 return false
3410 }
3411 if a.liveGateSaid[e.host] != build {
3412 if a.liveGateSaid == nil {
3413 a.liveGateSaid = map[string]string{}
3414 }
3415 a.liveGateSaid[e.host] = build
3416 log.Printf("activator: %s: live/%s names %s, which this box will not start (%v), so it does not move onto it: %s", e.host, e.host, build, err, left)
3417 }
3418 return true
3419}
3420
3421// workerSpawnTimeout bounds a supervised respawn's exec. It is generous
3422// because ExecBackend.StartWorker may materialize the build from the bucket
3423// first (a dangling live/<host> after a retention prune), but it is BOUNDED:
3424// an unbounded spawn that never returns would leave workerRespawning set, and
3425// the sweeper skips a respawning entry — no stand-down, no restart, no idle,
3426// and nothing able to wake it. A wedged fetch must cost one entry a couple of
3427// minutes, not the entry forever.
3428const workerSpawnTimeout = 2 * time.Minute
3429
3430// respawnJoinBound bounds a terminal stop's wait on a supervised respawn: the
3431// respawn goroutine's own worst case end to end (retiring the old child, the
3432// exec, and shedding a raced one). A var so tests can shrink it.
3433var respawnJoinBound = actStopTimeout + workerSpawnTimeout + sidecarStopTimeout
3434
3435// respawnWorkerLocked replaces a worker's child on the SAME entry, keeping its
3436// lease and replication: only the process is new. The caller holds a.mu.
3437// stopOld is the child to retire first (nil when it is already dead).
3438//
3439// onto is the build to exec: the captured serving build for a recycle
3440// (recycleWorkerLocked), or nil for a converge, which execs what live/<host>
3441// names now. A recycle onto a staged copy whose bytes have changed is refused
3442// by the backend (release.ErrStagedCopyChanged) and lands in the failed-spawn
3443// arm below, which keeps the captured build: the worker stays down, and each
3444// crash restart asks for that build again and is refused again while its
3445// bytes differ (spec §4.2, ruling R17).
3446//
3447// Everything expensive happens OFF a.mu, on its own goroutine. That is not a
3448// stylistic preference: ExecBackend.StartWorker can materialize a build from
3449// the bucket, and the SIGTERM drain of the outgoing child is bounded but not
3450// instant — either one held across the global mutex would stall every request
3451// wake, Owns and Shutdown on the box. (The co-spawn supervisor above DOES exec
3452// under the lock; that is StartSidecar, which never fetches anything. The two
3453// are not interchangeable.)
3454//
3455// Three orderings carry the safety:
3456// - Retire, THEN spawn. A sidecar is an outbound worker: two live
3457// incarnations double its work, where a gap costs only latency.
3458// - Ask again, under the lock, immediately BEFORE spawning. A stop that
3459// landed while the old child was draining has already released the lease,
3460// and a child spawned after that would be writing the tenant's DB with no
3461// single-writer claim behind it. Shedding it afterwards is not enough —
3462// the agent may be gone by then.
3463// - Check once more before INSTALLING, and shed the child if the entry was
3464// stopped or replaced meanwhile. A stop always wins.
3465//
3466// e.workerRespawning is the sweeper's interlock (mid-respawn the entry's proc
3467// reads as exited, which read as a crash would spawn a second child), and
3468// e.respawnDone is what stop() and Shutdown wait on so a stop never completes
3469// — nor the process exit — while a spawn is still in flight.
3470//
3471// A failed spawn deliberately leaves the dead child in place: the next sweep
3472// sees Exited and the crash-restart path takes over with its backoff. A failed
3473// converge degrades into supervision rather than into a hole.
3474func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string, onto *release.StagedIdentity) {
3475 if a.cfg.HostWrites.Enabled {
3476 e.attemptVeto = ErrClaimIsolationUnavailable
3477 e.respawnDone = make(chan struct{})
3478 close(e.respawnDone)
3479 return
3480 }
3481 e.workerRespawning = true
3482 e.respawnDone = make(chan struct{})
3483 done, rt, host := e.respawnDone, e.route, e.host
3484 // A socketless attempt's checks and a fallback's held publication arm
3485 // this respawn only (ReplaceWorker sets them just before the call): taken
3486 // here, under the caller's a.mu, so no later respawn inherits them.
3487 attemptGate, hold := e.attemptGate, e.holdServing
3488 e.attemptGate, e.holdServing = nil, false
3489 // check is the ownership check every exec asks, whatever the flag (spec
3490 // §7.4), ANDed with an attempt's own checks when this respawn execs an
3491 // attempt's build (spec §6.3, §8.1): a newer release, or a halted
3492 // cohort, whose halt bumps the revision, that lands while the old child
3493 // drains or the build is prepared means the attempt's build must not
3494 // run. The caller holds a.mu.
3495 check := func() error {
3496 if err := a.ownedLocked(e, false); err != nil {
3497 return err
3498 }
3499 if attemptGate != nil {
3500 return attemptGate()
3501 }
3502 return nil
3503 }
3504 // With generations on, the respawn is a member of the operation set
3505 // until it has finished with every child it made: a stop's barrier joins
3506 // it, and cancelling it tears down the exec (zero-downtime plan 3,
3507 // Task 11). Off, m stays nil and spawnParent is just lifeCtx's child.
3508 spawnParent, cancelSpawn := context.WithCancel(a.lifeCtx)
3509 var m *genMember
3510 oldGen := e.gen // the retiring child's generation, for its proof
3511 if a.gens {
3512 m = &genMember{gen: Gen{N: nextGen(), Worker: true}, role: roleSpawning, cancel: cancelSpawn, done: make(chan struct{})}
3513 a.addMemberLocked(e, m)
3514 }
3515 go func() {
3516 // respawnDone closes LAST, once this goroutine is finished with every
3517 // child it created — including one it had to SHED. That ordering is
3518 // the invariant stop() leans on: when respawnDone is observed closed,
3519 // no child spawned by this respawn is still running, so the final
3520 // checkpoint and lease release below it cannot land while a live
3521 // process is still pointed at the same database. Closing it before the
3522 // shed (as this first did) only narrowed that race; it did not close
3523 // it.
3524 defer close(done)
3525 defer cancelSpawn()
3526 shed := false // a shed child went to dropMember, which leaves the set itself
3527 if m != nil {
3528 defer func() { // runs before close(done): defers are LIFO
3529 if !shed {
3530 a.leaveSet(e, m)
3531 }
3532 }()
3533 }
3534 var (
3535 proc Proc
3536 ident release.StagedIdentity
3537 pend pendingGeneration // the build published before the exec
3538 tok string
3539 err error
3540 // refused is the config gate's answer for this respawn (amadan
3541 // #213). The lease is still held, so Acquire's gate is not in
3542 // the way, and a respawn onto config this box cannot provide
3543 // would start a worker without its secrets while keeping the
3544 // lease from a sibling that could run it.
3545 refused error
3546 )
3547 // The old child's stop, timed on the wall clock: a socketless
3548 // attempt reports it as its drain (spec §9.4).
3549 var retireTook time.Duration
3550 if stopOld != nil {
3551 retireStart := time.Now()
3552 stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
3553 if serr := stopOld.Stop(stopCtx); serr != nil {
3554 log.Printf("activator: worker %s respawn: retiring the old child: %v", host, serr)
3555 }
3556 cancel()
3557 retireTook = time.Since(retireStart)
3558 }
3559
3560 // The retired child must be proven gone, its host cgroup included,
3561 // before anything new execs beside its descendants (zero-downtime
3562 // spec §7.4, §7.9). Unproven: nothing spawns; the entry keeps its
3563 // lease, and the sweep's crash path retires it again through
3564 // genTerminated. stopOld may be nil (a crash restart, a restore):
3565 // the old child is gone, but what it forked lives in the host
3566 // cgroup, and genTerminated proves a Worker generation there with
3567 // or without a handle.
3568 retired := true
3569 if a.gens {
3570 oldGen.Worker = true
3571 retired = a.genTerminated(rt, oldGen, stopOld)
3572 } else {
3573 // Generations off, an earlier respawn's shed child that is not
3574 // proven gone may still run beside this exec (final review
3575 // R2-1): nothing execs until the sweep has proven it exited.
3576 a.mu.Lock()
3577 retired = e.shed == nil
3578 a.mu.Unlock()
3579 }
3580
3581 // Still wanted? Asked HERE, before anything is spawned: past this
3582 // point a stop can only be raced, never prevented. The gate's
3583 // ownership check (spec §7.4): every exec asks it, whatever the
3584 // switch says. It covers the entry still being the host's, awake
3585 // and not shutting down, and adds the lease still held and no
3586 // fence. An attempt's own checks are ANDed in (check), and plan 3's
3587 // termination proof (retired) is kept beside both: nothing execs
3588 // beside an old child that is not proven gone.
3589 a.mu.Lock()
3590 ownErr := check()
3591 a.mu.Unlock()
3592 owned := ownErr == nil
3593 wanted := owned && retired
3594
3595 if wanted {
3596 // Off the service lifetime, so Shutdown cancels this exec instead
3597 // of blocking on it. The timeout is the slow-materialize bound;
3598 // the parent is the "are we still here at all" one.
3599 ctx, cancel := context.WithTimeout(spawnParent, workerSpawnTimeout)
3600 // The ownership check again at the exec itself (final review
3601 // F1): the config admission, the materialization and the
3602 // build's publication all come between the check above and the
3603 // exec, and the lease can be lost, a fence set or the entry
3604 // stopped in that time. startWorker hands it to the backend as
3605 // the exec gate, asked after the publication and the cgroup's
3606 // preparation and before each start attempt (Astra R3 §5 item
3607 // 4). An attempt's checks ride on it (check): a supersession in
3608 // that time refuses the exec too.
3609 gate := func() error {
3610 if err := ctx.Err(); err != nil {
3611 return err
3612 }
3613 a.mu.Lock()
3614 defer a.mu.Unlock()
3615 return check()
3616 }
3617 if g, ok := a.backend.(configAdmitter); ok {
3618 refused = g.AdmitConfig(ctx, rt)
3619 }
3620 if refused == nil {
3621 // No registry write (spec §7.9): see doWake.
3622 if tok, err = newWorkerToken(); err == nil {
3623 proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok), onto, gate)
3624 }
3625 }
3626 cancel()
3627 }
3628
3629 a.mu.Lock()
3630 e.workerRespawning = false
3631 // And once more before the child is installed: ownership lost after
3632 // the exec sheds the child (final review F1). ownedLocked covers what
3633 // the old liveness check did (the entry is the host's, awake, not
3634 // shutting down) and adds the lease and the fences; an attempt's
3635 // checks ride on it (check), so an attempt superseded during its exec
3636 // sheds its child rather than install it.
3637 if owned {
3638 ownErr = check()
3639 }
3640 live := ownErr == nil
3641 if attemptGate != nil {
3642 // The attempt's answer for ReplaceWorker: which gate refused, at
3643 // the first check, at the exec, or before the install. Nil when
3644 // nothing refused (an unproven old child, a failed exec, a
3645 // config refusal or an install), which ReplaceWorker reads from
3646 // the incarnation instead.
3647 switch {
3648 case ownErr != nil:
3649 e.attemptVeto = ownErr
3650 case refused == nil && verdictOf(err) != gateProceed:
3651 e.attemptVeto = err
3652 }
3653 // The execution evidence (final review F2): a process of the
3654 // attempt's build was started, installed below or shed, or the
3655 // start's error says it may have been (startMayHaveRun; final
3656 // review wave 2, N4). Without it, nothing of the build ran.
3657 e.attemptExec = startMayHaveRun(proc, err)
3658 }
3659 switch {
3660 case owned && !retired && live:
3661 a.mu.Unlock()
3662 log.Printf("activator: worker %s NOT respawned (%s): an earlier child's termination is not proven (its host cgroup or a shed respawn child may still hold a process, or the proof could not run), so nothing execs beside it; the sweep retries", host, why)
3663 case !wanted || !live:
3664 a.mu.Unlock()
3665 reason := ownErr
3666 if reason == nil {
3667 reason = errors.New("the old generation's termination is not proven")
3668 }
3669 log.Printf("activator: worker %s NOT respawned (%s): %v; abandoning it", host, why, reason)
3670 if proc != nil {
3671 // Spawned into a window that closed under us: shed it rather
3672 // than leave a child nothing owns. With generations on it is
3673 // the member's process: dropMember stops it and proves its
3674 // host cgroup empty, or keeps it in the set for the reaper.
3675 if m != nil {
3676 // The child's build moves to the member before its
3677 // pending record is dropped below: if dropMember cannot
3678 // prove the child gone, the member is retained, and the
3679 // generations file keeps naming its build until the
3680 // reaper proves it (final review F12).
3681 a.mu.Lock()
3682 m.proc, m.build = proc, ident
3683 a.mu.Unlock()
3684 shed = true
3685 a.dropMember(e, m)
3686 } else {
3687 // Generations off (final review R2-1): the entry takes
3688 // the child, and its build moves from the pending record
3689 // to it in the same critical section, BEFORE the Stop.
3690 // A terminal stop that gives up waiting on this respawn
3691 // still finds it, and keeps the lease; the sweep reaps it
3692 // once its exit is proven (reapShedLocked).
3693 s := &shedChild{proc: proc, build: ident}
3694 a.mu.Lock()
3695 e.shed, s.retriedAt = s, a.now()
3696 a.dropPendingGenerationLocked(pend)
3697 a.mu.Unlock()
3698 pend = 0
3699 shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
3700 serr := proc.Stop(shedCtx)
3701 shedCancel()
3702 a.mu.Lock()
3703 gone := proc.Exited()
3704 if gone && e.shed == s {
3705 e.shed = nil
3706 }
3707 a.mu.Unlock()
3708 if !gone {
3709 log.Printf("activator: worker %s: the shed respawn child's exit is not proven (%v); the entry keeps it and its build until the sweep proves it", host, serr)
3710 }
3711 _ = a.persistGenerations()
3712 }
3713 }
3714 if pend != 0 {
3715 // Its published build stays listed until the child is shed.
3716 a.mu.Lock()
3717 a.dropPendingGenerationLocked(pend)
3718 a.mu.Unlock()
3719 _ = a.persistGenerations()
3720 }
3721 case refused != nil:
3722 // Stand down: the old child is already gone, and nothing may
3723 // start in its place. A normal stop checkpoints and releases the
3724 // lease, so a keyed sibling can take it. Nothing re-wakes it here
3725 // until the box can provide the config (the config gate refuses
3726 // every wake before then).
3727 a.beginStopLocked(e, StopRestart)
3728 a.mu.Unlock()
3729 log.Printf("activator: worker %s NOT respawned (%s): %v — standing down and releasing its lease", host, why, refused)
3730 case verdictOf(err) != gateProceed:
3731 // The exec's own ownership check refused (startWorker's gate),
3732 // and ownership came back before the install check (a fence
3733 // lifted), or an attempt's checks refused there: nothing exec'd,
3734 // and this is no failure of the build. The dead child stays, and
3735 // the sweep's crash restart asks again (an attempt's restore is
3736 // ReplaceWorker's, from attemptVeto).
3737 a.mu.Unlock()
3738 log.Printf("activator: worker %s NOT respawned (%s): refused at its exec: %v", host, why, err)
3739 case err != nil:
3740 // The captured build is kept whatever the failure, a refused
3741 // recycle (release.ErrStagedCopyChanged) included: spec §4.2
3742 // refuses a recovery onto changed bytes, and that refusal must
3743 // outlast the retry (ruling R17), so the next crash restart asks
3744 // for the same captured build again. Only a converge, admitted
3745 // on its own terms, moves the worker onto another build.
3746 a.mu.Unlock()
3747 log.Printf("activator: WORKER %s failed to respawn (%s): %v — the sweep's crash restart takes it from here", host, why, err)
3748 emitEvent(a.Logs, rt, "error", "sidecar could not be restarted. retrying")
3749 default:
3750 e.proc = proc
3751 e.workerToken = tok
3752 e.spawnVersion = ident.Version
3753 e.workerServing = ident
3754 e.workerVersion = "" // the new child has not reported yet
3755 e.retireTook = retireTook
3756 // A new incarnation, published below (spec §7.9), unless this is
3757 // a ReplaceWorker fallback: its publication is held until the
3758 // scheduler has recorded the failure it falls back from (spec
3759 // §8.1), and PublishHeldServing delivers it.
3760 e.workerInc++
3761 inc := e.workerInc
3762 if hold {
3763 e.heldServing = &heldServing{id: ident, inc: inc}
3764 }
3765 if m != nil {
3766 e.gen = Gen{N: m.gen.N, Build: ident.Version, Worker: true}
3767 }
3768 // The captured build replaces the one published before the exec.
3769 a.dropPendingGenerationLocked(pend)
3770 // Only an exec from the live link starts a fresh drift clock. A
3771 // recycle onto the same captured build serves exactly what it
3772 // served before, so a drift it was serving keeps its age, and a
3773 // max_run shorter than ConvergeGrace cannot starve the converge
3774 // (ruling R13).
3775 if onto == nil {
3776 e.staleSince = time.Time{}
3777 }
3778 e.spawnedAt = a.now()
3779 e.lastActive = a.now()
3780 a.mu.Unlock()
3781 // The captured build may have changed (a converge): publish it.
3782 _ = a.persistGenerations()
3783 if !hold {
3784 a.publishWorkerServing(e, ident, inc)
3785 }
3786 log.Printf("activator: worker %s respawned (%s), now running %s", host, why, ident.Version)
3787 emitEvent(a.Logs, rt, "info", "sidecar restarted")
3788 }
3789 }()
3790}
3791
3792// NoteWorkerContact stamps last-activity from an agent-socket call. It is a
3793// worker's only sign of life: nothing routes to a sidecar, so no request will
3794// ever stamp one, and without contact the idle clock runs to its window and
3795// the sweeper sleeps it.
3796func (a *Service) NoteWorkerContact(host string) {
3797 a.mu.Lock()
3798 defer a.mu.Unlock()
3799 if e := a.entries[host]; e != nil && e.worker {
3800 e.lastActive = a.now()
3801 }
3802}
3803
3804// NoteWorkerReady records a worker's own "I am up": last-activity, the end
3805// of its settle window, and the build it SAYS it runs. That claim is display
3806// metadata only (zero-downtime deploys spec §7.9): it can be empty or an
3807// app-chosen string, so it never identifies the executable, never overwrites
3808// registry Version (root's install stamp) and never decides a recycle. The
3809// serving build is workerServing, captured at the exec.
3810//
3811// It acts on a WAKING entry as well as an awake one, which is the whole point
3812// of the settle window: a child that boots fast says so and its wake resolves
3813// immediately instead of waiting out sidecarSettle.
3814func (a *Service) NoteWorkerReady(host, version string) {
3815 a.mu.Lock()
3816 defer a.mu.Unlock()
3817 e := a.entries[host]
3818 if e == nil || !e.worker {
3819 return
3820 }
3821 e.lastActive = a.now()
3822 e.workerVersion = version
3823 if !e.readyNoted {
3824 e.readyNoted = true
3825 close(e.readyc)
3826 }
3827}
3828
3829// SetWorkerNextDue records a sidecar's next scheduled work, straight to the
3830// due-time index — while the child is ALIVE and can say so. That is why the
3831// stop path never probes a worker: the answer is already recorded, and a probe
3832// of a process we are about to kill could only be worse.
3833func (a *Service) SetWorkerNextDue(host string, at int64) error {
3834 return a.due.SetNextDue(host, at)
3835}
3836
3837// SleepWorker stops an awake HIBERNATING worker now — the /v1/idle call, "I
3838// have nothing to do". It is a stop REQUESTED immediately rather than an idle
3839// timestamp nudged backwards, so the ordinary stop path (park the DB, release
3840// the lease) does the work. Anything other than an awake worker is a no-op:
3841// this must never start something, and never touch an instance.
3842//
3843// An ALWAYS-ON worker's request is logged and REFUSED. Its contract is "the
3844// record exists, so the process runs" — and obeying would strand it: the stop
3845// deletes the entry, so not even the crash-restart sweep has anything left to
3846// see, and it stays down until the agent restarts. A sidecar that genuinely
3847// wants to stop exits its own process, which the sweeper then restarts (or
3848// its operator deletes the record, which retires it for good).
3849func (a *Service) SleepWorker(host string) {
3850 a.mu.Lock()
3851 defer a.mu.Unlock()
3852 e := a.entries[host]
3853 if e == nil || !e.worker || e.state != actAwake {
3854 return
3855 }
3856 if !e.route.Hibernate {
3857 log.Printf("activator: worker %s reported idle but is always-on (hibernate:false) — ignoring", e.host)
3858 return
3859 }
3860 emitEvent(a.Logs, e.route, "info", "sidecar reported idle — stopping")
3861 a.beginStopLocked(e, StopIdle)
3862}
3863
3864// WorkerToken reports the live per-incarnation token for host, or "" — the
3865// agent socket's authorization check. A token from a previous wake, or from a
3866// previous incarnation of a restarted child, never matches.
3867//
3868// It answers for a WAKING entry as well as an awake one, because the child's
3869// first call — /v1/ready, which ends the settle window — necessarily arrives
3870// while the wake is still resolving. Refusing then would make the early-ready
3871// path unreachable and cost every wake the full window. Nothing else is
3872// loosened: a stopping entry, an asleep host, an instance and an unknown host
3873// all still get "", and a failed wake takes its token with it when the entry
3874// is deleted.
3875func (a *Service) WorkerToken(host string) string {
3876 a.mu.Lock()
3877 defer a.mu.Unlock()
3878 if e := a.entries[host]; e != nil && e.worker && (e.state == actAwake || e.state == actWaking) {
3879 return e.workerToken
3880 }
3881 return ""
3882}
3883
3884// setWorkerToken publishes a wake's freshly minted credential onto the entry.
3885// Its own critical section because doWake runs off a.mu — every other write to
3886// an entry's fields is likewise made under the lock.
3887func (a *Service) setWorkerToken(e *actEntry, tok string) {
3888 a.mu.Lock()
3889 e.workerToken = tok
3890 a.mu.Unlock()
3891}
3892
3893// stop drives awake→asleep: stop the child (its SIGTERM drain is the final
3894// write), checkpoint what's on disk to S3 and release the lease, then forget
3895// the entry. The local DB stays in place — it's the warm cache for the next
3896// wake; S3 is the durable truth under the lease.
3897func (a *Service) stop(e *actEntry) {
3898 outcome := &stopObservation{sidecarStop: e.sidecar == nil, sidecarExited: e.sidecar == nil, uncertain: e.sidecarUncertain}
3899 // The deliberate-stop marker for the member's timeline: hibernation,
3900 // convergence cycling, bounce, shutdown all pass through here, so an
3901 // "exited" line WITHOUT this line before it reads as what it is — a
3902 // crash.
3903 stopping := "instance stopping"
3904 if e.worker {
3905 stopping = "sidecar stopping"
3906 }
3907 emitEvent(a.Logs, e.route, "info", stopping)
3908
3909 // Wait out a supervised respawn first. Everything below — the final
3910 // checkpoint and the LEASE RELEASE — assumes no other process is being
3911 // pointed at this DB; a respawn that finished after the release would run
3912 // a child with no single-writer claim behind it, on the one path whose
3913 // whole purpose is single-writer. The respawn's own pre-spawn check makes
3914 // this wait short (it aborts as soon as it sees this state flip); the
3915 // bound is insurance against a wedged exec, and instances never wait at
3916 // all (respawnDone is nil for them).
3917 //
3918 // Gated on the CHANNEL alone, deliberately, and never on
3919 // e.workerRespawning: that flag is cleared in the respawn's final critical
3920 // section, BEFORE it sheds a child it raced into existence, so a stop
3921 // goroutine scheduled after that section would read false and skip the
3922 // wait entirely — with up to sidecarStopTimeout of draining child left to
3923 // go. beginStopLocked flips the state under a.mu and then launches this
3924 // goroutine, so the two genuinely race for the lock with nothing ordering
3925 // them. The channel does not have that gap: it is created before the
3926 // respawn starts and closed only when the respawn is finished with every
3927 // child it made. A respawn that ended long ago costs a receive on an
3928 // already-closed channel.
3929 a.mu.Lock()
3930 waitRespawn := e.respawnDone
3931 a.mu.Unlock()
3932 if waitRespawn != nil {
3933 // The bound covers the respawn goroutine's OWN worst case end to end —
3934 // retiring the old child, the exec, and shedding a raced one — so this
3935 // wait can never give up on a respawn that is still making progress.
3936 bound := time.NewTimer(respawnJoinBound)
3937 select {
3938 case <-waitRespawn:
3939 case <-bound.C:
3940 log.Printf("activator: %s respawn did not finish before its stop; stopping the rest, the lease is kept until it has", e.host)
3941 }
3942 bound.Stop() // the common case closes in milliseconds; don't leave a 2m timer standing
3943 }
3944
3945 // A sidecar restart (sweepOnce) runs its backend.StartSidecar call off
3946 // a.mu (design §6.7) and may still be in flight when a stop begins. Wait,
3947 // bounded by sidecarStopTimeout, for it to land: either it installs a
3948 // fresh e.sidecar in time and the co-stop below stops that one directly,
3949 // or it doesn't and the restart goroutine's own entry-identity check (it
3950 // re-reads a.entries[host] on the far side of the call) finds this entry
3951 // gone or stopping and stops whatever it started instead. Either way
3952 // nothing is left orphaned; this wait only decides which of the two
3953 // stops it. sidecarStartDone is closed by the restart goroutine when it
3954 // finishes, however it finishes, so this waits on that signal directly
3955 // rather than polling sidecarStarting on a timer.
3956 //
3957 // With generations off the wait is bounded (sidecarStartJoinWait), but
3958 // a start still in flight past it is not left behind (Task 8, F6 plan
3959 // 5): the rest of the stop runs, and then the stop stays pending, with
3960 // the lease, until the start returns and has stopped what it started
3961 // (the restart goroutine stops it before it closes sidecarStartDone).
3962 // Only then is the host cgroup proven empty and the lease released
3963 // (reapCurrentGen).
3964 a.mu.Lock()
3965 startDone, hook := e.sidecarStartDone, a.joinWaiting
3966 a.mu.Unlock()
3967 var sidecarJoin chan struct{}
3968 if startDone != nil {
3969 if hook != nil {
3970 hook(e.host, "sidecar start")
3971 }
3972 if a.gens {
3973 <-startDone // the start, and every process it left, is accounted for
3974 } else {
3975 bound := time.NewTimer(sidecarStartJoinWait)
3976 select {
3977 case <-startDone:
3978 case <-bound.C:
3979 // Recorded now, so every pending path below (a leader still
3980 // running, an unproven generation) waits for it too.
3981 sidecarJoin = startDone
3982 a.mu.Lock()
3983 e.sidecarJoin = startDone
3984 a.mu.Unlock()
3985 }
3986 bound.Stop()
3987 }
3988 }
3989
3990 // The barrier (zero-downtime spec §7.4). Every generation in this
3991 // entry's operation set is cancelled and joined before the final
3992 // checkpoint and the lease release below: a spawning or health-checking
3993 // successor, and a draining retiring generation. With no swap ever run
3994 // on this entry the set is empty and this returns at once.
3995 joined := a.joinOps(e) // swapsim:M1
3996
3997 // Co-stop, sidecar FIRST — before the next-due probe and the instance's
3998 // own SIGTERM. Two reasons: the sidecar must never outlive its instance
3999 // (an orphan's reconnect loop would be a standing wake source — the
4000 // ping-pong this feature exists to prevent — and its scheduled work would
4001 // keep firing into a host we're about to stop), and with the sidecar gone
4002 // first, the schedule the next-due probe reads can't shift under it.
4003 if e.sidecar != nil {
4004 scCtx, scCancel := context.WithTimeout(context.Background(), sidecarStopTimeout)
4005 err := e.sidecar.Stop(scCtx)
4006 outcome.sidecarStop, outcome.sidecarExited = err == nil, e.sidecar.Exited()
4007 if err != nil {
4008 log.Printf("activator: stop sidecar %s: %v", e.host, err)
4009 }
4010 scCancel()
4011 if !a.gens && sidecarStopUnproven(e.sidecar, err) {
4012 // Flag off, its exit is kept to be proven before the lease goes
4013 // (fix wave 1, item 1): a nested row has no host-cgroup proof to
4014 // reach it.
4015 a.mu.Lock()
4016 e.unprovenSidecars = append(e.unprovenSidecars, e.sidecar)
4017 a.mu.Unlock()
4018 }
4019 if a.gens {
4020 // A sidecar is gone only when its leader exited AND its cgroup is
4021 // empty (spec §7.4, §7.5). Unproven, it is tracked: the lease is
4022 // kept (releasable) and the sweep re-proves it.
4023 a.mu.Lock()
4024 rt := e.route
4025 a.mu.Unlock()
4026 if !a.sidecarTerminated(rt, e.sidecar) {
4027 a.mu.Lock()
4028 a.trackStraySidecarLocked(rt, e.sidecar)
4029 a.mu.Unlock()
4030 }
4031 }
4032 }
4033
4034 // The due-time index: ask the instance for its earliest pending scheduled
4035 // delivery BEFORE stopping it — the socket dies with the process. (When the
4036 // stop was triggered by a dead child, the query just fails fast.) Any
4037 // failure records 0, i.e. "sleeps until a request": a missed probe must
4038 // degrade to pre-index behavior, never block or crash the sleep. A due time
4039 // already in the past is recorded as-is and the sweeper wakes the host on
4040 // its next tick — there is deliberately no minimum-sleep guard beyond
4041 // that. The idle sweep alone would make thrash impossible (an awake
4042 // instance only sleeps after a full idle window), but Bounce breaks that
4043 // bound on purpose: it can stop an instance shortly after it woke, so
4044 // consecutive wake/sleep cycles are bounded by config-change frequency,
4045 // not cfg.Idle. That's accepted — a bounce is rare relative to request
4046 // traffic, and each cycle is still a bounded SIGTERM drain, not a busy
4047 // spin.
4048 //
4049 // A WORKER skips this step entirely — both the probe and the write. It has
4050 // no socket to answer on, and its next-due already arrived over the agent
4051 // socket while it was alive (SetWorkerNextDue wrote it straight to the
4052 // registry), which is strictly better than asking a process we are about
4053 // to kill. Writing here would be actively wrong: the probe would fail, and
4054 // the failure path records 0 — erasing the schedule that is a sleeping
4055 // sidecar's ONLY way back up, since nothing can dial one.
4056 if !e.worker {
4057 dueCtx, cancelDue := context.WithTimeout(context.Background(), 5*time.Second)
4058 due, err := a.nextDue(dueCtx, e)
4059 cancelDue()
4060 if err != nil {
4061 log.Printf("activator: next-due probe %s: %v — recording none (it sleeps until a request)", e.host, err)
4062 due = 0
4063 }
4064 if err := a.due.SetNextDue(e.host, due); err != nil {
4065 log.Printf("activator: record next-due %s: %v", e.host, err)
4066 } else if due > 0 {
4067 log.Printf("activator: %s next scheduled delivery at %s — will wake for it", e.host, time.Unix(due, 0).UTC().Format(time.RFC3339))
4068 }
4069 }
4070
4071 // Separate budgets: a SIGTERM drain that runs long must not eat the final
4072 // checkpoint's time — the S3 sync is the part that makes sleep durable.
4073 procCtx, cancelProc := context.WithTimeout(context.Background(), actStopTimeout)
4074 defer cancelProc()
4075 // Declared rather than inferred: the next-due probe above used to leave an
4076 // `err` in this scope, but it is inside the `if !e.worker` block now, so
4077 // this one is its own.
4078 var err error
4079 if e.proc != nil {
4080 err = e.proc.Stop(procCtx)
4081 if errors.Is(err, ErrStillRunning) {
4082 // The lease is never released over an unproven termination
4083 // (design §6.3, codex R2-5): stop() bails out here rather than
4084 // falling through to the checkpoint and lease release below,
4085 // leaving the entry in stopping with pendingStop set.
4086 // sweepOnce polls proc.Exited() for it and calls finishStop,
4087 // the rest of what this function would have done, the moment
4088 // termination is actually proven.
4089 log.Printf("activator: stop %s: %v; keeping the lease until termination is proven", e.host, err)
4090 // The call just above is this entry's first stop attempt, and it
4091 // was already bounded by actStopTimeout. keepPendingLocked
4092 // records it as stopRetriedAt, which is what makes
4093 // finishPendingStops' own cadence a true actStopTimeout apart
4094 // from THIS attempt, rather than firing a redundant retry the
4095 // moment pendingStop becomes visible to it.
4096 a.mu.Lock()
4097 a.keepPendingLocked(e, outcome)
4098 a.mu.Unlock()
4099 return
4100 }
4101 outcome.mainStop, outcome.mainExited = err == nil, e.proc.Exited()
4102 } else {
4103 err = ErrStopUnproven
4104 }
4105 if err != nil {
4106 log.Printf("activator: stop %s: %v", e.host, err)
4107 }
4108 if !joined {
4109 // A generation's termination is unproven, so the lease stays held,
4110 // as for ErrStillRunning above. finishPendingStops reaps the member
4111 // and then finishes this stop.
4112 log.Printf("activator: stop %s: a generation's termination is unproven; keeping the lease until it is", e.host)
4113 a.mu.Lock()
4114 a.keepPendingLocked(e, outcome)
4115 a.mu.Unlock()
4116 return
4117 }
4118 a.mu.Lock()
4119 settled := a.respawnSettledLocked(e)
4120 if !settled {
4121 // A respawn still running past the bounded wait above, or the
4122 // child it shed and could not prove gone (final review R2-1): the
4123 // lease stays held, as for ErrStillRunning, and finishPendingStops
4124 // finishes this stop once the respawn is done and its child proven
4125 // exited.
4126 a.keepPendingLocked(e, outcome)
4127 }
4128 a.mu.Unlock()
4129 if !settled {
4130 log.Printf("activator: stop %s: a respawn or the child it shed is not proven finished; keeping the lease until it is", e.host)
4131 return
4132 }
4133 if !a.gens && sidecarJoin != nil {
4134 // A sidecar start is still in flight past the bounded wait above
4135 // (Task 8, F6 plan 5): the stop stays pending with the lease until
4136 // it returns, and finishPendingStops proves the host cgroup then.
4137 log.Printf("activator: stop %s: a sidecar start is still in flight; keeping the lease until it returns", e.host)
4138 a.mu.Lock()
4139 a.keepPendingLocked(e, outcome)
4140 a.mu.Unlock()
4141 return
4142 }
4143 if !a.gens && !a.sidecarsExited(e, false) {
4144 // A sidecar this stop stopped, or one a restart brought back, has
4145 // not exited (fix wave 1, item 1): the stop stays pending with the
4146 // lease, and the sweep re-stops it (reapCurrentGen).
4147 a.mu.Lock()
4148 a.keepPendingLocked(e, outcome)
4149 a.mu.Unlock()
4150 return
4151 }
4152 if !a.gens && !a.stopHostProven(e) {
4153 // Flag off, the leader's exit says nothing of what it forked (Task
4154 // 8, F6): the host cgroup is proven empty first. Unproven, the stop
4155 // stays pending with the lease, as for ErrStillRunning, and the
4156 // sweep retries the proof (reapCurrentGen).
4157 log.Printf("activator: stop %s: its host cgroup is not proven empty; keeping the lease until it is", e.host)
4158 a.mu.Lock()
4159 a.keepPendingLocked(e, outcome)
4160 a.mu.Unlock()
4161 return
4162 }
4163 if a.gens && !a.releasable(e) { // swapsim:M8
4164 // THE retain-until-proven rule (spec §7.4): unproven, the stop stays
4165 // pending with the lease held, exactly as for ErrStillRunning, and
4166 // the sweep retries (reapCurrentGen). That covers a leader still
4167 // alive after a failed kill (a claims-on KillAndEmpty error is not
4168 // ErrStillRunning, so it reaches here), a child not proven empty, a
4169 // worker's host cgroup, and a stray sidecar.
4170 log.Printf("activator: stop %s: not every process of the entry is proven gone; keeping the lease until it is", e.host)
4171 a.mu.Lock()
4172 a.keepPendingLocked(e, outcome)
4173 a.mu.Unlock()
4174 return
4175 }
4176 a.finishStop(e, outcome)
4177}
4178
4179// finishPendingStop finishes a stop that was left pending because proc.Stop
4180// returned ErrStillRunning (design §6.3): now that proc.Exited() has proven
4181// termination, spec §6.3 calls for retrying proc.Stop once more before
4182// declaring the main process closed. Spec §6.3 says "sweepOnce retries
4183// proc.Stop on its cadence until done closes", so outcome.mainStop reflects
4184// an actual observation rather than an assumption baked in at the
4185// ErrStillRunning return. A unitProc that has already exited takes its
4186// done branch and returns nil immediately; a childProc's Stop is a no-op
4187// on an already-exited process either way, so retrying costs nothing once
4188// the process is well and truly gone. mainExited is unconditionally true:
4189// the exit that made this entry eligible for this pass IS that proof,
4190// whatever this retried Stop call itself reports.
4191func (a *Service) finishPendingStop(e *actEntry, outcome *stopObservation) {
4192 ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
4193 err := e.proc.Stop(ctx)
4194 cancel()
4195 outcome.mainStop, outcome.mainExited = err == nil, true
4196 if err != nil {
4197 log.Printf("activator: stop %s: %v", e.host, err)
4198 }
4199 a.finishStop(e, outcome)
4200}
4201
4202// finishPendingStops scans for entries left pending because proc.Stop
4203// returned ErrStillRunning (design §6.3, codex R2-5) and finishes every one
4204// whose proc has since exited, via finishPendingStop. Called from
4205// sweepOnce's own tick and, once more, from Shutdown. The sweeper is
4206// stopped by the time Shutdown runs, so without this second call a process
4207// that exits during shutdown would never get its final checkpoint. The
4208// caller holds a.mu.
4209func (a *Service) finishPendingStops() {
4210 // Failed wakes whose generation is not proven stopped hold their lease
4211 // until it is (zero-downtime spec §7.4).
4212 a.reapUnprovenWakesLocked()
4213 // Lease releases that failed are retried on the same tick (#292).
4214 a.retryOwedReleasesLocked()
4215 a.reapStraySidecarsLocked()
4216 for _, e := range a.entries {
4217 // Members first (zero-downtime spec §7.4): a generation whose stop
4218 // could not be proven is finished here once it exits.
4219 a.reapMembersLocked(e)
4220 // And a shed respawn child kept with generations off (final review
4221 // R2-1).
4222 a.reapShedLocked(e)
4223 // A parent restore that failed when the last member left keeps its
4224 // obligation and is retried here (final review F11).
4225 a.retryParentRestoreLocked(e)
4226 // A run whose stop was asked for and is not proven, pending stop
4227 // or not (re-review 6).
4228 a.retryCmdSettlesLocked(e)
4229 if !e.pendingStop || len(e.ops.members) > 0 || !a.respawnSettledLocked(e) {
4230 continue
4231 }
4232 if e.proc == nil || e.proc.Exited() {
4233 // Members gone and the main process (if any) exited: the stop
4234 // finishes once its termination proof holds (reapCurrentGen):
4235 // flag on, the retain-until-proven rule (releasable); flag off,
4236 // once a sidecar start it left in flight has returned, the host
4237 // cgroup proven empty (Task 8, F6). A process still alive falls
4238 // through to main's re-stop cadence below.
4239 if !e.genReaping && (a.gens || sidecarStartJoined(e.sidecarJoin)) {
4240 e.genReaping = true
4241 go a.reapCurrentGen(e)
4242 }
4243 continue
4244 }
4245 // Still running: spec §6.3 says sweepOnce retries proc.Stop on its
4246 // own cadence until done closes, and that applies here too, not only
4247 // once termination has already been proven some other way. Without
4248 // this, a stop whose first Control.Stop was itself lost (the socket
4249 // refused, a 502 mid-roll) never asks again, and the entry sits in
4250 // stopping holding the lease until the agent restarts.
4251 if e.stopRetrying || a.now().Sub(e.stopRetriedAt) < actStopTimeout {
4252 continue
4253 }
4254 e.stopRetrying = true
4255 e.stopRetriedAt = a.now()
4256 go a.retryPendingStop(e)
4257 }
4258}
4259
4260// sidecarStartJoined reports whether a sidecar start a flag-off stop left
4261// in flight (actEntry.sidecarJoin) has returned: none was, or its done
4262// channel is closed.
4263func sidecarStartJoined(join chan struct{}) bool {
4264 if join == nil {
4265 return true
4266 }
4267 select {
4268 case <-join:
4269 return true
4270 default:
4271 return false
4272 }
4273}
4274
4275// respawnDoneLocked reports whether e has no supervised respawn in flight.
4276// The caller holds a.mu.
4277func respawnDoneLocked(e *actEntry) bool {
4278 if e.respawnDone == nil {
4279 return true
4280 }
4281 select {
4282 case <-e.respawnDone:
4283 return true
4284 default:
4285 return false
4286 }
4287}
4288
4289// respawnSettledLocked reports whether e's supervised respawns are done
4290// with every child they made: none is in flight, and no child one shed is
4291// kept unproven (final review R2-1). A terminal stop finishes, and gives
4292// up the lease, only then. The caller holds a.mu.
4293func (a *Service) respawnSettledLocked(e *actEntry) bool {
4294 return e.shed == nil && respawnDoneLocked(e)
4295}
4296
4297// reapShedLocked finishes a shed respawn child kept with generations off
4298// (final review R2-1), once the respawn that shed it is done. An exited
4299// child is forgotten and its build unpinned. One still running is stopped
4300// again, bounded, every actStopTimeout, the cadence finishPendingStops
4301// keeps for the main process. The caller holds a.mu.
4302func (a *Service) reapShedLocked(e *actEntry) {
4303 s := e.shed
4304 if s == nil || s.reaping || !respawnDoneLocked(e) {
4305 return
4306 }
4307 if !s.proc.Exited() && a.now().Sub(s.retriedAt) < actStopTimeout {
4308 return
4309 }
4310 s.reaping, s.retriedAt = true, a.now()
4311 go func() {
4312 if !s.proc.Exited() {
4313 ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
4314 err := s.proc.Stop(ctx)
4315 cancel()
4316 if err != nil && !errors.Is(err, ErrStillRunning) {
4317 log.Printf("activator: %s: retry stop of the shed respawn child: %v", e.host, err)
4318 }
4319 }
4320 a.mu.Lock()
4321 s.reaping = false
4322 gone := s.proc.Exited()
4323 if gone && e.shed == s {
4324 e.shed = nil
4325 }
4326 a.mu.Unlock()
4327 if gone {
4328 log.Printf("activator: %s: the shed respawn child has exited", e.host)
4329 _ = a.persistGenerations()
4330 }
4331 }()
4332}
4333
4334// retryPendingStop re-issues proc.Stop for a pendingStop entry that is
4335// still running, off a.mu like finishPendingStop's own call. Unlike that
4336// one, a return here proves nothing by itself. Done closing is still what
4337// proves termination, and the next finishPendingStops pass is what notices
4338// it and calls finishPendingStop. This only clears the in-flight guard so a
4339// later pass can try again.
4340func (a *Service) retryPendingStop(e *actEntry) {
4341 ctx, cancel := context.WithTimeout(context.Background(), actStopTimeout)
4342 err := e.proc.Stop(ctx)
4343 cancel()
4344 if err != nil && !errors.Is(err, ErrStillRunning) {
4345 log.Printf("activator: retry stop %s: %v", e.host, err)
4346 }
4347 a.mu.Lock()
4348 e.stopRetrying = false
4349 a.mu.Unlock()
4350}
4351
4352// finishStop is stop()'s (and finishPendingStop's) tail: the final
4353// checkpoint, lease release, and forgetting the entry. Split out (design
4354// §6.3, codex R2-5) so a pendingStop entry's eventual finish shares it with
4355// stop()'s own ordinary path. outcome carries forward whatever stop() had
4356// already observed (the sidecar co-stop, the due-time probe) before it hit
4357// ErrStillRunning and returned, plus whatever finishPendingStop's own
4358// retried proc.Stop just added.
4359func (a *Service) finishStop(e *actEntry, outcome *stopObservation) {
4360 // A run attached to this instance is stopped and settled before
4361 // replication closes (run spec §5.3, codex 1): everything it committed
4362 // is in the final checkpoint, and nothing writes the database after.
4363 // A run whose stop is not proven keeps the entry, replication, the
4364 // lease and its handle (merge ruling M1): the stop stays pending, as
4365 // for any process whose termination is unproven, and the sweep tries
4366 // the run's settle again (finishPendingStops). The check lives here, so
4367 // no stop path can skip it.
4368 if !a.settleCmds(e) {
4369 log.Printf("activator: stop %s: a run is not proven stopped; keeping replication and the lease until it is", e.host)
4370 a.mu.Lock()
4371 a.keepPendingLocked(e, outcome)
4372 a.mu.Unlock()
4373 return
4374 }
4375 // Close = final sync + lease release. After a lost lease the release part
4376 // fails harmlessly (someone else owns the lock); the sync still preserves
4377 // a local checkpoint.
4378 ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
4379 defer cancel()
4380 repErr := ErrStopUnproven
4381 if e.rep != nil {
4382 repErr = e.rep.Close(ctx)
4383 }
4384 outcome.replicaClosed = repErr == nil
4385 if repErr != nil {
4386 log.Printf("activator: final checkpoint %s: %v", e.host, repErr)
4387 }
4388 a.mu.Lock()
4389 outcome.uncertain = !outcome.complete()
4390 e.stopOutcome = outcome
4391 a.rememberStopLocked(e.host, outcome)
4392 republish := e.worker || a.publishesLocked(e)
4393 delete(a.entries, e.host)
4394 retired := e.gen
4395 // The canonical link goes with the last generation (spec §7.1). Under
4396 // a.mu, the lock the awake transition creates links under, so a new
4397 // entry's link cannot be removed by this one's stop.
4398 if a.gens && linksCanonical(e.route, retired) {
4399 if err := removeCanonicalLink(e.route.Socket, retired.Socket); err != nil {
4400 log.Printf("activator: %s: remove link %s: %v", e.host, e.route.Socket, err)
4401 }
4402 }
4403 // A hold swap (zero-downtime spec §7.8): its pinned wake is installed in
4404 // the same critical section, before e.stopped closes, so a request
4405 // waiting on this stop finds the pinned entry, never an empty slot it
4406 // would fill with a wake of the live link.
4407 var next *actEntry
4408 if h := e.holdNext; h != nil {
4409 e.holdNext = nil
4410 var herr error
4411 next, herr = a.holdWakeLocked(e, h)
4412 h.woke <- holdWoke{next: next, err: herr}
4413 }
4414 a.mu.Unlock()
4415 if republish {
4416 // Its processes are gone, so its builds are no longer referenced by
4417 // them.
4418 _ = a.persistGenerations()
4419 }
4420 // The generation stopped being current when its entry went (spec §7.2:
4421 // its idle connections close now, not at the transport's idle timeout).
4422 a.genRetired(e.host, retired)
4423 close(e.stopped)
4424 if next != nil {
4425 go a.wake(next)
4426 }
4427 log.Printf("activator: %s hibernated", e.host)
4428}
4429
4430// Start launches the background sweeper that puts idle instances to sleep and
4431// stands down lost leases/dead children between requests.
4432func (a *Service) Start() {
4433 // The generations file is written before anything spawns, whatever
4434 // CARLOS_SWAP says (the index contract): a file left by a previous
4435 // incarnation is stale, and one a crash cut short reads to root as an
4436 // incomplete pin set, so root would neither restage nor sweep until this
4437 // process changes it. The write is off a.mu, and a failure is logged and
4438 // marks the publication dirty for the sweep to retry; it never fails or
4439 // blocks Start.
4440 _ = a.persistGenerations()
4441 a.mu.Lock()
4442 defer a.mu.Unlock()
4443 if a.shuttingDown {
4444 return
4445 }
4446 a.sweepStop = make(chan struct{})
4447 a.sweepDone = make(chan struct{})
4448 // Sweep well inside the idle window so a sleep happens promptly after it
4449 // elapses; clamp so second-granular test idles still sweep sanely.
4450 interval := a.cfg.Idle / 4
4451 if interval < 100*time.Millisecond {
4452 interval = 100 * time.Millisecond
4453 }
4454 if interval > 30*time.Second {
4455 interval = 30 * time.Second
4456 }
4457 go func() {
4458 defer close(a.sweepDone)
4459 if a.Startup != nil {
4460 select {
4461 case <-a.Startup:
4462 case <-a.sweepStop:
4463 return
4464 }
4465 }
4466 t := time.NewTicker(interval)
4467 defer t.Stop()
4468 for {
4469 select {
4470 case <-a.sweepStop:
4471 return
4472 case <-t.C:
4473 a.sweepOnce()
4474 a.lastSweep.Store(time.Now().UnixNano())
4475 }
4476 }
4477 }()
4478}
4479
4480// LastSweep is when the sweeper last finished a sweep, or the zero time
4481// before its first. The sweep runs at least every 30 seconds, so a box
4482// whose last sweep is older than a minute has a hosting loop that is stuck
4483// (region HA spec 2 §6). A nil service has never swept.
4484func (a *Service) LastSweep() time.Time {
4485 if a == nil {
4486 return time.Time{}
4487 }
4488 n := a.lastSweep.Load()
4489 if n == 0 {
4490 return time.Time{}
4491 }
4492 return time.Unix(0, n)
4493}
4494
4495// markCleanParkLocked records that a clean-exit park is stopping e's current
4496// worker incarnation (round 2 Task 6). ReplaceWorker's observation forgives
4497// an exited one-shot that is no longer awake (spec §6.5) only when this mark
4498// names the incarnation it watched: the park ran with no fence, the lease
4499// still held and a clean exit. A stop for a lost lease, a fence or a
4500// shutdown never marks, so the build is never recorded good on it. The
4501// conditions are checked here again so the mark never depends on the
4502// caller's order. The caller holds a.mu.
4503func (a *Service) markCleanParkLocked(e *actEntry) {
4504 if a.fences[e.host] != 0 || e.rep == nil || e.rep.Err() != nil || procCrashed(e.proc) {
4505 return
4506 }
4507 e.parkedInc = e.workerInc
4508}
4509
4510// sweepOnce examines every awake instance once: idle ones hibernate; a lost
4511// lease or dead child stands down immediately regardless of activity. It then
4512// checks the due-time index and wakes any sleeping host whose scheduled
4513// delivery has come due.
4514func (a *Service) sweepOnce() {
4515 now := a.now()
4516 // The due query runs before the lock — it's a registry read, and nothing
4517 // under a.mu should wait on SQLite.
4518 dueRoutes, err := a.due.DueInstances(now.Unix())
4519 if err != nil {
4520 log.Printf("activator: due-time sweep: %v", err)
4521 }
4522 // The always-on arm's input, read on the same off-lock terms and for the
4523 // same reason: nothing under a.mu should wait on SQLite.
4524 alwaysOn, err := a.due.AlwaysOnWorkers()
4525 if err != nil {
4526 log.Printf("activator: always-on sidecar sweep: %v", err)
4527 }
4528 // The guard's input, read here for the same reason the due query is:
4529 // nothing under a.mu may wait on SQLite. A failure does NOT degrade to
4530 // today's behaviour: earliest ends up nil, so no host is held, and a
4531 // host with a schedule tighter than the idle window flips from one
4532 // resident process into a wake/sleep pair every tick — a 5m schedule
4533 // becomes ~288 checkpoint-and-restore traversals a day, precisely the
4534 // trade the guard exists to refuse. Fail open anyway: refusing to
4535 // hibernate anything until the query recovers would hold every
4536 // instance on the box awake, which is worse than losing the guard for
4537 // the instances that have one.
4538 earliest, err := a.due.EarliestTimers()
4539 if err != nil {
4540 log.Printf("activator: earliest-timer sweep: %v — hibernating without the schedule guard", err)
4541 earliest = nil
4542 }
4543 a.mu.Lock()
4544 if a.shuttingDown {
4545 // Shutdown has begun and is about to stop this sweeper: from here it
4546 // owns every entry (#292). A child found exited now is its stop's,
4547 // not an unexpected exit to recycle, and nothing is due-woken or
4548 // started. Shutdown runs finishPendingStops itself.
4549 a.mu.Unlock()
4550 return
4551 }
4552 // Finish stops that were left pending because proc.Stop returned
4553 // ErrStillRunning (design §6.3, codex R2-5): termination wasn't proven
4554 // at the time, so stop() kept the lease and returned without forgetting
4555 // the entry. Checked before the idle pass below for the same reason it's
4556 // a separate call at all: a pendingStop entry is in actStopping, not
4557 // actAwake, so the idle pass would never reach it.
4558 a.finishPendingStops()
4559 // Build-convergence cycles this tick may start (rate-limited), shared
4560 // with the recycles below that converge instead (recycleWorkerLocked).
4561 a.convergeBudget = maxConvergeCyclesPerSweep
4562 for _, e := range a.entries {
4563 // Recovery retains its cut fence. Supervise its process and lease
4564 // before the ordinary fenced-host skip; only the claim can open it.
4565 if e.probeHeld && e.state == actAwake {
4566 if a.retireRecoveryProbeLocked(e) {
4567 continue
4568 }
4569 if !e.probeReviewing {
4570 e.probeReviewing = true
4571 go a.reviewRecoveryHold(e)
4572 }
4573 continue
4574 }
4575 if a.fences[e.host] != 0 {
4576 continue
4577 }
4578 if e.state != actAwake {
4579 continue
4580 }
4581 // A converge respawn owns this entry's child until it finishes: its
4582 // old proc is stopped and its new one not yet installed, which every
4583 // check below would misread. One tick of patience, not a policy.
4584 if e.workerRespawning {
4585 continue
4586 }
4587 if err := e.rep.Err(); err != nil {
4588 log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, err)
4589 a.beginStopLocked(e, StopLeaseLost)
4590 continue
4591 }
4592 if e.proc.Exited() {
4593 // Whatever follows, the child died on its own: a first wake's
4594 // observation reads it as the build's failure (spec §6.4).
4595 e.unexpectedExit = true
4596 // A worker's exit is read by STATUS, which is the only thing that
4597 // can tell "my work is done" from "I fell over":
4598 //
4599 // - always-on (`hibernate: false`): the record exists ⇒ the
4600 // process runs, so ANY exit is restarted in place — same
4601 // supervision as the co-spawn path, and no lease churn,
4602 // because only the child died (the lease and replication are
4603 // still ours).
4604 // - hibernating, clean exit (status 0): the bot finished. Park
4605 // it — the stand-down below, exactly as for an instance.
4606 // - hibernating, NON-ZERO exit: a crash, and design §7.2 says a
4607 // crash restarts, unqualified. Without this arm a hibernating
4608 // worker that fell over before setting a next-due slept
4609 // forever, and the log called it "exited" as though it had
4610 // finished.
4611 //
4612 // The restart carries the same doubling backoff, so a crash loop
4613 // PACES itself — it is not terminated. Nothing below this arm ever
4614 // runs for a crashing child: this branch continues unconditionally
4615 // while the proc reads as exited, and a successful respawn resets
4616 // both spawnedAt and lastActive, so neither max_run nor the idle
4617 // clock can fire between crashes. A hibernating sidecar that
4618 // crash-loops therefore keeps its entry and its lease at a 2s→1m
4619 // cadence until an operator parks it (a wake stands a broken
4620 // worker down and recycles it) or its record goes away. A
4621 // crash-COUNT ceiling that parks after N failures is the right
4622 // bound and is filed as a follow-up: it is a design decision about
4623 // when the platform gives up on a member's bot, not a detail of
4624 // this arm.
4625 if e.worker && (!e.route.Hibernate || procCrashed(e.proc)) {
4626 a.restartWorkerLocked(e, now)
4627 continue
4628 }
4629 if e.worker {
4630 log.Printf("activator: sidecar %s exited cleanly — its work is done; parking it", e.host)
4631 // Past this sweep's fence skip and lease check: the one
4632 // stop of an exited one-shot that a ReplaceWorker
4633 // observation may still count as a good build.
4634 a.markCleanParkLocked(e)
4635 } else {
4636 log.Printf("activator: instance %s exited unexpectedly — recycling", e.host)
4637 }
4638 a.beginStopLocked(e, StopExited)
4639 continue
4640 }
4641 // The idle predicate is shared verbatim; only its WINDOW can differ.
4642 // A worker declares its own on its row (Task 1) so this sweep never
4643 // needs a bucket read to decide one has sat idle long enough; zero —
4644 // every instance, and any sidecar that declared nothing — falls back
4645 // to the platform default, so an instance's behaviour is unchanged.
4646 // inflight is always 0 for a worker (nothing routes to it) and
4647 // lastActive is stamped by agent-socket contact instead of by a
4648 // released request, so the same expression means the right thing for
4649 // both.
4650 //
4651 // An ALWAYS-ON worker has no idle clock at all (design §7.2:
4652 // "hibernate: false … no idle clock, no lease churn"): idleness is
4653 // not a reason to stop something whose record says it must be
4654 // running. It does still have a run CEILING, but that ceiling
4655 // recycles its child rather than stopping the entry (see max_run
4656 // below) — a stopped always-on worker has no wake source of its own.
4657 // An instance is never in this arm: unflagged instances don't enter
4658 // the activator.
4659 hibernates := !e.worker || e.route.Hibernate
4660 // A held probe entry (ProbeWake) gets no traffic: the edge routes no
4661 // held row. It is not idle until the target's open marks it open
4662 // (OpenProbeEntry, Plan 3 Task 7).
4663 if e.probe != nil && !e.probe.managed && e.probe.phase < probeOpened {
4664 hibernates = false
4665 }
4666 idleWindow := a.cfg.Idle
4667 if e.workerIdle > 0 {
4668 idleWindow = e.workerIdle
4669 }
4670 if f := a.idleFor.Load(); f != nil {
4671 if d := (*f)(e.route); d > 0 && d < idleWindow {
4672 idleWindow = d
4673 }
4674 }
4675 if hibernates && e.inflight == 0 && e.cmd == nil && now.Sub(e.lastActive) >= idleWindow {
4676 // Don't sleep into an imminent tick. A wake/sleep pair is a full
4677 // checkpoint-and-restore round trip; spending one to be absent
4678 // for less time than we would have idled anyway is a straight
4679 // loss. The threshold is THIS entry's idle window rather than a
4680 // constant of its own: it is already the stated opinion about how
4681 // long residency is worth holding — the deployment's for an
4682 // instance, and the app's own manifest value for a worker, which
4683 // is exactly the figure a worker's schedule should be weighed
4684 // against.
4685 next, ok := earliest[e.host]
4686 heldForTick := ok && time.Unix(next, 0).Sub(now) <= idleWindow
4687 if !heldForTick {
4688 log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second))
4689 a.beginStopLocked(e, StopIdle)
4690 continue
4691 }
4692 // Fall through to the convergence check below: a host held awake
4693 // for its schedule never reaches a "next wake" to converge on, so
4694 // it is now one of the always-awake hosts that arm was written
4695 // for — the same class as a sidecar host that never sleeps.
4696 }
4697 // A worker that never reports idle still has a ceiling: a wedged
4698 // reconnect loop must not hold a lease and a box forever. Zero means
4699 // no ceiling — a sidecar that declared none is a legitimate shape, so
4700 // this must never invent one nobody asked for.
4701 //
4702 // It applies to an ALWAYS-ON worker too — the wedged reconnect loop
4703 // this guard exists for is exactly the shape a long-lived worker gets
4704 // into — but it is spent differently there: the child is RECYCLED on
4705 // the same entry, keeping the lease and replication, because stopping
4706 // an always-on worker leaves nothing to start it again. That is the
4707 // same move build convergence makes for one (convergeWorkerLocked),
4708 // and for the same reason.
4709 if e.worker && e.workerMaxRun > 0 && now.Sub(e.spawnedAt) >= e.workerMaxRun {
4710 if !e.route.Hibernate {
4711 log.Printf("activator: always-on sidecar %s hit its max_run of %s — recycling its child", e.host, e.workerMaxRun)
4712 a.recycleWorkerLocked(e, e.proc, "hit its max_run")
4713 continue
4714 }
4715 log.Printf("activator: %s hit its max_run of %s — stopping", e.host, e.workerMaxRun)
4716 a.beginStopLocked(e, StopIdle)
4717 continue
4718 }
4719 // Converge onto a shipped build. A running child never re-reads
4720 // live/<host>, so once a ship repoints the symlink this awake child is
4721 // serving a stale binary. Idle hosts converge for free on their next
4722 // wake; this cycle is for the always-awake ones — notably every
4723 // sidecar host, which never sleeps — that otherwise serve the old
4724 // build indefinitely. Gracefully cycle: beginStopLocked drains via
4725 // SIGTERM, takes the final checkpoint, releases the lease, and leaves
4726 // the host asleep; a member request (or the due-time index) then wakes
4727 // it onto the new binary — the same clean cycle `systemctl restart`
4728 // gives a unit-managed host. Unlike idle hibernation this can fire
4729 // with a request in flight; the stop still drains HTTP gracefully and
4730 // reconnecting clients coalesce onto the fresh wake, so a long upload
4731 // past the drain is cut exactly as a restart would cut it. A pinned
4732 // canary never trips this: the updater leaves its symlink alone, so
4733 // spawnVersion still matches.
4734 cur, readErr := a.liveVer(e.host)
4735 cycle, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, ConvergeGrace)
4736 e.staleSince = next
4737 if cycle && a.OnDrift != nil {
4738 // CARLOS_SWAP=overlap: the scheduler owns every move onto a new
4739 // build, paced (spec §10). The sweep only tells it to look, on
4740 // its own goroutine and so off a.mu (the scheduler calls back
4741 // into Current), and carries on with this entry's sidecar
4742 // supervision below. No converge budget is spent.
4743 host := e.host
4744 go a.OnDrift(host)
4745 cycle = false
4746 }
4747 if cycle && a.convergeBudget > 0 {
4748 // An always-on worker converges as one MOVE, not as a cycle: the
4749 // cycle's second half is "something wakes it back up", and a
4750 // worker with no request path and no schedule has nothing that
4751 // would. Stop-then-respawn on the same entry is that second half,
4752 // performed here. Both existing guards still apply — this only
4753 // runs once the drift has persisted past ConvergeGrace, and it
4754 // consumes the same one-per-sweep budget, so a bad build cannot
4755 // respawn the fleet at once.
4756 //
4757 // A converge the live exec gate refuses (a declined box that
4758 // honours a record or marker holding cur, Task 9) stops nothing
4759 // and spends no budget: the worker carries on to supervision
4760 // below, running its current build.
4761 //
4762 // So is the cycle of an instance or a hibernating worker (fix
4763 // round 1, ruling 2): its next wake would be refused, so stopping
4764 // it now turns a held build into an outage. A refusal keeps it
4765 // running on its current build.
4766 if e.worker && !e.route.Hibernate {
4767 if a.convergeWorkerLocked(e, cur) {
4768 a.convergeBudget--
4769 continue
4770 }
4771 } else if !a.liveGateRefusesLocked(e, cur, "it keeps running "+e.spawnVersion) {
4772 log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur)
4773 a.beginStopLocked(e, StopRelease)
4774 a.convergeBudget--
4775 continue
4776 }
4777 }
4778 // Sidecar supervision — Restart=always, sweep-driven, no supervisor
4779 // goroutine: a sidecar that died (or failed to spawn at wake) while
4780 // its instance stays awake is respawned with doubling, capped backoff;
4781 // a run past sidecarHealthyRun resets it. Spawned under a.mu
4782 // deliberately: an exec is quick, and the lock is what makes a restart
4783 // impossible to race with a stop (sidecar fields only change in the
4784 // awake state, and every stop flips the state under this same lock
4785 // first).
4786 //
4787 // A sidecar env staged MID-FLIGHT (attached to an awake child — the
4788 // opt-in path) is ADOPTED here rather than waiting for the next wake,
4789 // so opting in takes effect in seconds. Safe against double-running:
4790 // only hibernating hosts are activator children and their systemd
4791 // sidecar units stay disabled — the staged sidecar has exactly one
4792 // possible runner.
4793 //
4794 // Never for a worker: it is a sidecar itself, not a host that has one
4795 // (the same refusal doWake makes at wake time — restated here because
4796 // this adoption arm can fire on a long-awake entry that never went
4797 // through step 6 again).
4798 if e.worker {
4799 continue
4800 }
4801 if a.gens && e.state == actAwake && e.sidecarWanted && !e.sidecarStarting &&
4802 e.sidecar != nil && !e.sidecar.Exited() && e.sidecarGen != e.gen.N {
4803 // A legacy sidecar still on a retired generation's build: a
4804 // replacement whose stop of it was unproven (spec §7.5).
4805 go a.replaceSidecar(e)
4806 }
4807 if !e.sidecarWanted && a.backend.HasSidecar(e.route) {
4808 e.sidecarWanted = true
4809 log.Printf("activator: sidecar staged for awake %s — adopting", e.host)
4810 }
4811 if e.sidecarWanted && (e.sidecar == nil || e.sidecar.Exited()) && !now.Before(e.sidecarRetryAt) {
4812 if e.sidecar != nil && now.Sub(e.sidecarStarted) >= sidecarHealthyRun {
4813 e.sidecarBackoff = 0
4814 }
4815 if e.sidecarBackoff == 0 {
4816 e.sidecarBackoff = sidecarRestartBase
4817 } else {
4818 e.sidecarBackoff = min(e.sidecarBackoff*2, sidecarRestartMax)
4819 }
4820 e.sidecarRetryAt = now.Add(e.sidecarBackoff)
4821 // The restart call itself runs off a.mu (design §6.7, codex
4822 // R2-6): held under the lock, a stalled StartSidecar (a hung
4823 // exec, a wedged control socket) would block every other host's
4824 // wake and sweep behind it. sidecarStarting keeps a second
4825 // attempt from launching while one is already in flight.
4826 //
4827 // The far side re-checks by ENTRY IDENTITY (e0, the *actEntry
4828 // itself), not a generation counter: a per-entry counter that
4829 // starts at 0 and is bumped once when an entry becomes awake
4830 // reads 1 for every entry's first wake, so it can never tell
4831 // "this entry" apart from "a completely different entry, for
4832 // the same host, after a stop and rewake, that also just had
4833 // its first wake". A pointer
4834 // comparison against a.entries[host] has no such collision: a
4835 // fresh wake always allocates a brand new *actEntry (begin() and
4836 // the due-wake path both do `&actEntry{...}`), so identity alone
4837 // says whether this attempt's wake is still the current one.
4838 if !e.sidecarStarting {
4839 e.sidecarStarting = true
4840 done := make(chan struct{})
4841 e.sidecarStartDone = done
4842 e0 := e
4843 host := e.host
4844 route := e.route
4845 gen := e.gen
4846 prior := e.sidecar // exited, or nil
4847 backoff := e.sidecarBackoff
4848 go func() {
4849 defer close(done)
4850 if a.gens && prior != nil && !a.sidecarTerminated(route, prior) {
4851 // The sidecar being restarted exited, but something it
4852 // forked is left in its cgroup (spec §7.4, §7.5). It is
4853 // not forgotten: it becomes a tracked stray, and the
4854 // start below refuses beside it.
4855 a.mu.Lock()
4856 a.trackStraySidecarLocked(route, prior)
4857 a.mu.Unlock()
4858 }
4859 ctx, cancel := context.WithTimeout(context.Background(), instancectl.DeadlineControl)
4860 sc, err := a.startSidecar(ctx, e0, route, gen, false)
4861 cancel()
4862 var discard, stopNow Proc
4863 defer func() { // after the unlock, before done closes
4864 a.discardSidecar(route, discard)
4865 a.stopReturnedSidecar(e0, stopNow)
4866 // The join is given up only now, with the cleanup
4867 // finished (fix wave 1, item 1): a stop that reads
4868 // it while the sidecar this start brought back is
4869 // still being stopped must wait for that stop.
4870 a.mu.Lock()
4871 if e0.sidecarStartDone == done {
4872 e0.sidecarStarting, e0.sidecarStartDone = false, nil
4873 }
4874 a.mu.Unlock()
4875 }()
4876 a.mu.Lock()
4877 defer a.mu.Unlock()
4878 if a.entries[host] != e0 {
4879 // This host's map slot no longer holds e0 at all:
4880 // forgotten (stopped and deleted) or replaced by an
4881 // entirely different entry from a fresh wake. e0 is
4882 // unreachable from here on, so there's nothing left
4883 // on it to update; just stop whatever came back so
4884 // it doesn't outlive the wake it doesn't belong to,
4885 // or clobber a fresh wake's own sidecar.
4886 discard = sc
4887 return
4888 }
4889 // Still e0: this attempt owns its sidecarStarting flag
4890 // and join, and the deferred cleanup above clears them
4891 // once it has stopped whatever must be stopped,
4892 // regardless of what state e0 is in now: a stop may
4893 // have begun (flipping e0 to actStopping) while the
4894 // call was in flight. Cleared here, before the cleanup
4895 // ran, a stop reading the join in between would miss
4896 // the sidecar still being stopped (fix wave 1, item 1).
4897 if e0.state != actAwake {
4898 // Stopped mid-call, not yet forgotten: don't install
4899 // onto a wake that's on its way out.
4900 if a.gens {
4901 // stop() is joined on this goroutine (done). It
4902 // stops e.sidecar after that join, synchronously,
4903 // and records whether it exited: install sc so it
4904 // is that stop's to prove.
4905 e0.sidecar, e0.sidecarGen = sc, gen.N
4906 return
4907 }
4908 // Flag off, the stop either waited this start out or
4909 // is pending on it (e0.sidecarJoin, Task 8): sc is
4910 // stopped here, synchronously, before done closes,
4911 // so the stop's host proof runs after it.
4912 stopNow = sc
4913 return
4914 }
4915 if a.gens && e0.gen.N != gen.N {
4916 // A swap switched generations while this start was in
4917 // flight, so sc runs the retired generation's build
4918 // (spec §7.5). It is NOT discarded asynchronously
4919 // (stopDiscardedSidecar): the swap's replaceSidecar,
4920 // joined on this goroutine, would then start the new
4921 // sidecar while this one still ran, and a terminal
4922 // stop could pass the join over it (§7.4). It is
4923 // installed, so it stays tracked: replaceSidecar stops
4924 // it synchronously and starts the current generation's
4925 // only once it is proven gone, and a terminal stop
4926 // stops it like any sidecar.
4927 e0.sidecar, e0.sidecarStarted, e0.sidecarGen = sc, a.now(), gen.N
4928 if err != nil {
4929 e0.sidecarUncertain = true
4930 }
4931 return
4932 }
4933 if err != nil {
4934 e0.sidecarUncertain = true
4935 if sc == nil {
4936 log.Printf("activator: SIDECAR restart for %s failed: %v (next attempt in %s)", host, err, backoff)
4937 } else {
4938 log.Printf("activator: SIDECAR restart for %s failed: %v — retaining returned process", host, err)
4939 }
4940 e0.sidecar, e0.sidecarGen = sc, gen.N
4941 return
4942 }
4943 log.Printf("activator: sidecar for %s was down — restarted (backoff now %s)", host, backoff)
4944 e0.sidecar, e0.sidecarGen = sc, gen.N
4945 e0.sidecarStarted = a.now()
4946 }()
4947 }
4948 }
4949 }
4950 // Due-time wakes: a request-less wake through the same singleflight state
4951 // machine, so a client request arriving mid-wake just coalesces. Only hosts
4952 // with NO live entry are woken — an awake host is already serving its
4953 // schedule; a waking one is on its way; a stopping one will re-record its
4954 // next_due when the stop completes and be picked up on a later tick. The
4955 // registry's stale past value while the host is up is harmless for the same
4956 // reason. A failed due wake (e.g. a lease still held by a crashed
4957 // incarnation) removes the entry, and the index — only rewritten at the
4958 // next successful sleep — naturally retries it every tick until it works.
4959 for _, rt := range dueRoutes {
4960 // Only hibernating, ready hosts: the index exists to wake sleeping
4961 // instances, and a systemd-managed host is always on — spawning a
4962 // child for it would fight its unit over the socket. This also makes
4963 // a leftover next_due inert on a host flipped back to systemd (the
4964 // flip-off's release re-records one after the flag is already clear).
4965 // The registry's DueInstances is deliberately broader (no hibernate
4966 // filter), so the gate lives here, with the machine it protects.
4967 if !rt.Hibernate || rt.Status != "ready" {
4968 continue
4969 }
4970 if a.fences[rt.Host] != 0 {
4971 continue
4972 }
4973 if _, live := a.entries[rt.Host]; live {
4974 continue
4975 }
4976 // mesh seam: in the multi-box mesh only the HRW owner due-wakes a host —
4977 // otherwise every box sweeping a due entry would race a wake per tick
4978 // (the lease would let exactly one win, but loudly, forever). Eleven:
4979 // `if a.mesh != nil && !a.mesh.owns(rt.Host) { continue }`.
4980 if a.allowFilesDiagnosticLocked(a.seedDiagnostics[rt.Host], "due") {
4981 log.Printf("activator: %s has a scheduled delivery due — waking it", rt.Host)
4982 }
4983 // A sidecar row comes up as a WORKER (registry.DueInstances returns
4984 // both kinds, because the due-time index is a sidecar's only timer
4985 // wake source — nothing can dial one). Woken through the instance path
4986 // it would spawn via Start and spend the whole health-probe timeout
4987 // dialling a socket it never binds.
4988 var e *actEntry
4989 if rt.IsSidecar() {
4990 e = a.newWorkerEntry(rt)
4991 // The value that fired, carried into the wake so a SUCCESSFUL one
4992 // can consume exactly it (consumeDueTime) — a worker's stop writes
4993 // no next_due at all, so without this the row keeps its passed
4994 // value and this same sweep re-wakes it every tick.
4995 e.dueFired = rt.NextDue
4996 } else {
4997 e = &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{})}
4998 }
4999 a.entries[rt.Host] = e
5000 go a.wake(e)
5001 }
5002
5003 // The always-on arm: design §7.2 says the record exists ⇒ the process
5004 // runs, and until this existed that held only across agent RESTARTS. The
5005 // boot pass (host's runWorkerBootPass) runs once, at arming; the due loop
5006 // above gates on Hibernate; so an always-on sidecar created while the
5007 // agent was running had nothing that would ever start it — no request can
5008 // reach one, and it has no schedule. It also covers every other way an
5009 // always-on worker ends up with no live entry: a wake that failed, or a
5010 // stop with no re-wake behind it.
5011 //
5012 // Bounded per sweep for the reason host's workerBootConcurrency is
5013 // bounded: each wake claims an S3 lease and restores a database, so a box
5014 // just handed a fleet of sidecars starts them in batches. The remainder
5015 // comes up on the next sweep.
5016 started := 0
5017 for _, rt := range alwaysOn {
5018 if a.shuttingDown {
5019 break // the agent is going away; nothing new may claim a lease
5020 }
5021 if started >= maxWorkerStartsPerSweep {
5022 break
5023 }
5024 if _, live := a.entries[rt.Host]; live {
5025 continue // awake, waking, or draining a stop — none of those is stranded
5026 }
5027 log.Printf("activator: always-on sidecar %s is not running — starting it", rt.Host)
5028 e := a.newWorkerEntry(rt)
5029 a.entries[rt.Host] = e
5030 go a.wake(e)
5031 started++
5032 }
5033 a.mu.Unlock()
5034 // A generations file write that failed is retried here, off a.mu.
5035 a.retryGenerations()
5036}
5037
5038// RetireExcept stops (final checkpoint + lease release) every live entry whose
5039// host is not in keep — the reload-time reconcile behind flipping a host's
5040// hibernate flag off (and, as a bonus, route removal while awake). The flip
5041// clears the route's flag and reloads the edge; the map swap stops routing
5042// requests through the activator, and this puts the child down and gives the
5043// S3 lease back so the host's systemd unit can bind the socket. It reuses the
5044// ordinary stop path — the narrowest hook that exists — so the release still
5045// records next_due and checkpoints to S3 like any sleep. In-flight requests on
5046// a retired child are cut off as a deploy restart would cut them — the flip is
5047// a deliberate operator action.
5048func (a *Service) RetireExcept(keep map[string]bool) {
5049 // A probe's entry (ProbeWake) is held for its move while the registry
5050 // row is: the edge manages ready rows only, so it is never in keep.
5051 seen, held := a.heldProbes(keep)
5052 if retireBetweenLocks != nil {
5053 retireBetweenLocks()
5054 }
5055 a.mu.Lock()
5056 var waking []*actEntry
5057 for _, e := range a.entries {
5058 if e.probe != nil {
5059 snap, covered := seen[e]
5060 switch {
5061 case covered && e.probe.seq != snap.seq:
5062 // The probe changed since this pass's snapshot (an open, or a
5063 // newer pass took it over): the registry read is older than
5064 // the change and cannot judge it. Kept for this reload.
5065 continue
5066 case !e.probe.managed && keep[e.host] && e.probe.phase >= probeOpening:
5067 // The edge manages the opened probe's ready row: from here it
5068 // is an ordinary serving entry (Plan 3 Task 7). The metadata
5069 // stays (ruling 2).
5070 e.probe.managed = true
5071 e.probe.seq++
5072 continue
5073 case !e.probe.managed && (!covered || held[e]):
5074 continue
5075 }
5076 }
5077 if keep[e.host] {
5078 continue
5079 }
5080 // Workers are outside this reconcile's world entirely. `keep` is the
5081 // EDGE's managed set, built by edge.Reload — which refuses a sidecar
5082 // row on kind as the first act of its loop, so a worker's host can
5083 // never appear in it. Without this guard every edge reload (any route
5084 // change at all) would stop every running standalone sidecar. A
5085 // sidecar's removal authority belongs to the sidecar reconcile pass,
5086 // not to the edge's route map.
5087 // A held recovery belongs to its claim operation, not the edge's
5088 // serving set. Explicit fences and claim/session review still stop it.
5089 if e.worker || e.probeHeld {
5090 continue
5091 }
5092 switch e.state {
5093 case actAwake:
5094 log.Printf("activator: %s is no longer hibernate-managed — releasing", e.host)
5095 a.beginStopLocked(e, StopRetired)
5096 case actWaking:
5097 // A wake can't be stopped mid-flight; finish the retire once it
5098 // resolves (below, off-lock).
5099 waking = append(waking, e)
5100 case actStopping:
5101 // Already on its way out.
5102 }
5103 }
5104 a.mu.Unlock()
5105 for _, e := range waking {
5106 go func(e *actEntry) {
5107 <-e.ready
5108 a.mu.Lock()
5109 defer a.mu.Unlock()
5110 // Only if THIS entry is still the live one and its wake succeeded —
5111 // a failed wake already removed itself, and a fresh entry for the
5112 // same host means the route was re-flagged in the meantime.
5113 if a.entries[e.host] == e && e.state == actAwake {
5114 log.Printf("activator: %s is no longer hibernate-managed — releasing (wake finished first)", e.host)
5115 a.beginStopLocked(e, StopRetired)
5116 }
5117 }(e)
5118 }
5119}
5120
5121// Quiesced answers whether a host currently has NO live activator entry —
5122// no child running, none waking, none mid-stop. The residence-move cut
5123// (internal/host's move worker) gates its final mirror round on this: the
5124// route's Status change retires the entry through RetireExcept, and only a
5125// quiet host's replica prefix is byte-final.
5126func (a *Service) Quiesced(host string) bool {
5127 a.mu.Lock()
5128 defer a.mu.Unlock()
5129 _, live := a.entries[host]
5130 return !live
5131}
5132
5133// Shutdown checkpoints and stops every child — the agent going down takes its
5134// instances with it (they're its children; leaving them orphaned would strand
5135// the leases until TTL expiry and leave the sockets bound against the next
5136// incarnation's spawns). Blocks until all stops complete or ctx expires.
5137func (a *Service) Shutdown(ctx context.Context) {
5138 // mesh seam — drain step 1, order-pinned: Eleven left the mesh membership
5139 // FIRST — the heartbeat is deleted before any checkpoint or lease release
5140 // below, so peers stop proxying here immediately and their wakes succeed
5141 // the moment each lease releases. Deploys, box failure drills, and
5142 // scale-down all ride this same path. (`a.mesh.stop(ctx)` went here.)
5143 // Announce the shutdown BEFORE anything stops: a supervised respawn in
5144 // flight checks this flag immediately before it spawns, so it abandons
5145 // rather than starting a child this shutdown would then have to chase.
5146 // (stop() waits out any respawn that got past that check, so the waits
5147 // below cover the rest.)
5148 a.mu.Lock()
5149 a.shuttingDown = true
5150 // Proofs that start or run from here end with this budget too
5151 // (legacyRunGone, #292): none may outlive Shutdown.
5152 a.shutdownCtx = ctx
5153 sweepStop, sweepDone := a.sweepStop, a.sweepDone
5154 a.mu.Unlock()
5155 // …and those already running end with it too (Astra, round 3).
5156 context.AfterFunc(ctx, a.cancelProofs)
5157 // …and cancel the service lifetime, which TEARS DOWN any worker spawn
5158 // already past that check rather than letting the stop below wait it out.
5159 // The wait is right for a Bounce or an idle sleep — no deadline, and
5160 // letting the exec finish is tidier — but a shutdown has a deadline it
5161 // does not control: systemd's TimeoutStopSec is 90s by default, and
5162 // workerSpawnTimeout alone is 2 minutes. An agent that blocks on one
5163 // host's bucket materialize gets SIGKILLed mid-shutdown, which strands
5164 // leases and sockets for EVERY host on the box — far worse than
5165 // abandoning one respawn. The cancelled respawn aborts and closes its
5166 // respawnDone, so stop()'s wait falls through immediately.
5167 if a.lifeStop != nil {
5168 a.lifeStop()
5169 }
5170 if sweepStop != nil {
5171 close(sweepStop)
5172 <-sweepDone
5173 }
5174 a.mu.Lock()
5175 // The sweeper is stopped as of the line above, so nothing else will ever
5176 // poll a pendingStop entry (design §6.3) again on its own. Run the same
5177 // pass once here, and again on the ticker below while waiting, so a
5178 // process that finally exits during shutdown still gets its final
5179 // checkpoint instead of being abandoned mid-drain.
5180 a.finishPendingStops()
5181 var waits []chan struct{}
5182 var waking []*actEntry
5183 for _, e := range a.entries {
5184 switch e.state {
5185 case actAwake:
5186 a.beginStopLocked(e, StopShutdown)
5187 waits = append(waits, e.stopped)
5188 case actStopping:
5189 waits = append(waits, e.stopped)
5190 case actWaking:
5191 // Shutdown covers waking entries too (zero-downtime spec §7.4),
5192 // whatever CARLOS_SWAP says. The wake is cancelled, and once it
5193 // resolves, wake() itself stops the entry (through the barrier
5194 // with swaps on), or the failed wake has given its lease back
5195 // under a budget of its own (releaseLease).
5196 //
5197 // Swaps off used to leave a waking entry to resolve on its own,
5198 // with the lease's TTL to bound the damage. That is #292: an
5199 // edge restart stops every carlos-instance@ unit before the
5200 // agent's SIGTERM, the agent recycles one and a request wakes it
5201 // again, and that wake's start is a systemctl start job queued
5202 // behind the edge restart. The agent exited with the wake still
5203 // holding its lease, and the next agent refused the host for the
5204 // lease's whole 15 minutes.
5205 if e.wakeCancel != nil {
5206 e.wakeCancel()
5207 }
5208 waking = append(waking, e) // swapsim:M6
5209 }
5210 }
5211 a.mu.Unlock()
5212 // One ticker for every wait below, the wake joins included (#292,
5213 // Astra P2): a wake that ignores its cancellation can hold a join until
5214 // the deadline, and a stop left pending meanwhile must still be
5215 // finished, and its lease given back, as soon as its process is proven
5216 // gone, not once every wake has joined.
5217 poll := time.NewTicker(100 * time.Millisecond)
5218 defer poll.Stop()
5219 pollPending := func() {
5220 a.mu.Lock()
5221 a.finishPendingStops()
5222 a.mu.Unlock()
5223 }
5224 for _, e := range waking {
5225 for joined := false; !joined; {
5226 select {
5227 case <-e.ready:
5228 joined = true
5229 case <-poll.C:
5230 pollPending()
5231 case <-ctx.Done():
5232 a.logUnresolvedAtExit()
5233 return
5234 }
5235 }
5236 a.mu.Lock()
5237 if a.entries[e.host] == e && e.state == actStopping {
5238 waits = append(waits, e.stopped)
5239 }
5240 a.mu.Unlock()
5241 }
5242 for _, w := range waits {
5243 for done := false; !done; {
5244 select {
5245 case <-w:
5246 done = true
5247 case <-poll.C:
5248 pollPending()
5249 case <-ctx.Done():
5250 a.logUnresolvedAtExit()
5251 return
5252 }
5253 }
5254 }
5255 // Retained failed wakes (zero-downtime spec §7.4, P2-13) have no entry
5256 // for the loop above to wait on, so they get the same polling here: the
5257 // reaper (reapUnprovenWakesLocked, inside finishPendingStops) retries
5258 // each one's stop and proof and gives its lease back once it holds. A
5259 // wake still unproven when the budget runs out keeps its lease; the
5260 // lease's TTL frees it.
5261 //
5262 // With generations off the map fills too, so flag-off Shutdown waits
5263 // here as well, bounded by ctx: from GenContainment's sentinels
5264 // (genStartUnproven), and, on any box, from a failed wake whose leader
5265 // survived its stop (cleanupFailedWake's leaderLives, the fail-closed
5266 // exception to dormancy from the task 2 fix). That flag-off retention
5267 // proves only the leader's exit, not its descendants (final review F6,
5268 // deferred). With neither, the map is empty and this returns at once.
5269 //
5270 // Lease releases that failed (#292, Astra P3) are retried here the same
5271 // way, inside the same budget (retryOwedReleasesLocked, also inside
5272 // finishPendingStops), and named at exit if they are still owed.
5273 for {
5274 a.mu.Lock()
5275 left := len(a.unprovenWakes) + len(a.owedReleases)
5276 if left > 0 {
5277 a.finishPendingStops()
5278 }
5279 a.mu.Unlock()
5280 if left == 0 {
5281 return
5282 }
5283 select {
5284 case <-poll.C:
5285 case <-ctx.Done():
5286 a.logUnresolvedAtExit()
5287 return
5288 }
5289 }
5290}
5291
5292// logUnresolvedAtExit names what Shutdown's budget ran out on: failed
5293// wakes not proven stopped, whose leases are kept, and lease releases that
5294// failed and were not resolved by a retry. The TTL frees either.
5295func (a *Service) logUnresolvedAtExit() {
5296 a.mu.Lock()
5297 var held, owed []string
5298 for host, u := range a.unprovenWakes {
5299 held = append(held, fmt.Sprintf("%s (generation %d)", host, u.gen.N))
5300 }
5301 for o := range a.owedReleases {
5302 owed = append(owed, o.host)
5303 }
5304 a.mu.Unlock()
5305 if len(held) > 0 {
5306 slices.Sort(held)
5307 log.Printf("activator: shutdown: out of time with %d failed wake(s) not proven stopped, their leases still held: %s", len(held), strings.Join(held, ", "))
5308 }
5309 if len(owed) > 0 {
5310 slices.Sort(owed)
5311 log.Printf("activator: shutdown: out of time with %d lease release(s) unresolved, held until their TTL: %s", len(owed), strings.Join(owed, ", "))
5312 }
5313}
5314
5315// ServeHTTP answers one proxied request for a hibernating instance: wake (or
5316// join a wake), proxy, stamp activity. Lease contention and wake timeouts are
5317// 503s — "sleeping hosts are slow to wake", never a 5xx cascade of retries.
5318// It reports whether it handled the request (wrote the response or proxied
5319// it); false means "not activator-managed — proxy straight", so the edge can
5320// call it unconditionally.
5321func (a *Service) ServeHTTP(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
5322 return a.serve(w, r, rt, proxy, true)
5323}
5324
5325// ServeScheduled is ServeHTTP for a platform-delivered scheduled tick: same
5326// wake, same in-flight hold, same error mapping — no activity credit.
5327//
5328// A dedicated method rather than a wire marker like SidecarHeader, and the
5329// difference is the delivery path, not taste. The sidecar dials its
5330// instance's public URL, so its traffic arrives through the edge and the
5331// request is the only place a signal could ride. A tick never leaves the
5332// process (schedule_wiring.go: "the tick never leaves the box"), so the call
5333// site IS the signal, and there is no header for an outsider to send.
5334//
5335// A tick never wakes a managed route this box has stopped serving (a status
5336// other than ready: the residence-move cut's "moving", instance-moves design
5337// G12). The due-wake sweep holds the same gate. The edge never hands the
5338// activator such a row, but the runner fires whatever row the registry
5339// returns, and a tick that woke a host mid-cut would restore it and let it
5340// write behind the move's final mirror round. The move worker's fence closes
5341// the race in-process; this gate is the half that survives an agent restart.
5342func (a *Service) ServeScheduled(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler) bool {
5343 if rt.Kind == "instance" && rt.Hibernate && rt.Status != "" && rt.Status != "ready" {
5344 maintenanceResponse(w, r)
5345 return true
5346 }
5347 return a.serve(w, r, rt, proxy, false)
5348}
5349
5350func (a *Service) serve(w http.ResponseWriter, r *http.Request, rt registry.Route, proxy http.Handler, stamp bool) bool {
5351 if rt.Kind == "instance" && rt.Hibernate && a.Startup != nil {
5352 deadline := time.NewTimer(wakeWaitBound)
5353 err := a.awaitStartup(r.Context(), deadline.C)
5354 deadline.Stop()
5355 if err != nil {
5356 return a.serveWakeError(w, r, rt, err)
5357 }
5358 // The edge may have chosen rt and its proxy before startup recovery
5359 // ran, whether this request waited here or arrived just as Startup
5360 // closed. If recovery moved this host's upstream (a layout journal
5361 // finished), that proxy points at the old socket: answer as a wake
5362 // still in progress, and the retry is routed afresh. Only requests
5363 // near startup are checked; after the window the edge's own
5364 // selection is current.
5365 if a.nearStartup() && a.Routes != nil {
5366 if fresh, ferr := a.Routes.Get(rt.Host); ferr == nil && fresh != nil &&
5367 (fresh.Socket != rt.Socket || fresh.Addr != rt.Addr) {
5368 return a.serveWakeError(w, r, rt, errWakeTimeout)
5369 }
5370 }
5371 }
5372
5373 a.mu.Lock()
5374 fenced := a.fences[rt.Host] != 0
5375 a.mu.Unlock()
5376 if fenced {
5377 maintenanceResponse(w, r)
5378 return true
5379 }
5380 // Per-host gating: only a hibernate-flagged instance is activator-managed;
5381 // unflagged instances and sibling services are the edge's to proxy
5382 // straight — the deploy-safety property that lets the bucket env ship
5383 // fleet-wide as a no-op before any host is flagged.
5384 if rt.Kind != "instance" || !rt.Hibernate {
5385 return false
5386 }
5387
5388 // Sidecar-marked traffic (SidecarHeader — the sidecar stamps every REST
5389 // call and its WebSocket dial) is activity-invisible: proxied while the
5390 // instance is awake, refused while it isn't, never a wake. The co-spawned
5391 // sidecar dials its instance's public URL through this very edge (the box
5392 // pins its hosted names to loopback), so without this its keepalives
5393 // would read as member activity — the instance could never idle — and
5394 // after a sleep its reconnect loop would be the wake source: ping-pong
5395 // forever. Spoofing the marker is harmless by construction: it only
5396 // DOWNGRADES the sender's own requests (no activity credit; 503 while
5397 // asleep). It cannot suppress other clients' activity stamps and cannot
5398 // wake a sleeping host, so an outsider sending it gains nothing.
5399 if r.Header.Get(SidecarHeader) != "" {
5400 // The marker is an edge-side signal only (activity exclusion, above);
5401 // strip it so nothing downstream can ever grow a meaning an outsider
5402 // could spoof.
5403 r.Header.Del(SidecarHeader)
5404 if err := a.BeginSidecarRequest(r.Context(), rt); err != nil {
5405 if errors.Is(err, ErrFenced) {
5406 maintenanceResponse(w, r)
5407 return true
5408 }
5409 if errors.Is(err, context.Canceled) {
5410 return true // the sidecar hung up while queued on a wake
5411 }
5412 w.Header().Set("Retry-After", "5")
5413 // Ours, not the app's: the edge must not file this 503 as an
5414 // edge error (edge/answered.go; spec 2026-09-07 §3.2).
5415 edge.MarkAnswered(r)
5416 http.Error(w, "instance is asleep; sidecar requests do not wake it", http.StatusServiceUnavailable)
5417 return true
5418 }
5419 proxy.ServeHTTP(w, r)
5420 return true
5421 }
5422
5423 // mesh seam: Eleven proxied a member request for a host with no live
5424 // local entry and an HRW owner on another live box to that peer here (one
5425 // hop max, guarded by a forwarded marker); a host with ANY live local
5426 // entry stayed local — placement converges at the next sleep boundary,
5427 // since its state is already in S3. Single-box v1 always wakes locally.
5428
5429 release, err := a.begin(r.Context(), rt, stamp)
5430 if err != nil {
5431 return a.serveWakeError(w, r, rt, err)
5432 }
5433 defer release()
5434 proxy.ServeHTTP(w, r)
5435 return true
5436}
5437
5438func (a *Service) serveWakeError(w http.ResponseWriter, r *http.Request, rt registry.Route, err error) bool {
5439 var held *errLeaseHeld
5440 var refused *errConfigRefused
5441 switch {
5442 case errors.Is(err, ErrFenced):
5443 maintenanceResponse(w, r)
5444 case errors.As(err, &refused):
5445 // Quiet by design: the config guard logs once per host and
5446 // publishes the refusal in the box's status snapshot. A line per
5447 // request here would bury that one.
5448 w.Header().Set("Retry-After", strconv.Itoa(configRefusedRetryAfter))
5449 edge.MarkAnswered(r)
5450 http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
5451 case errors.As(err, &held):
5452 retry := int(held.retryAfter/time.Second) + 1
5453 w.Header().Set("Retry-After", strconv.Itoa(retry))
5454 edge.MarkAnswered(r)
5455 http.Error(w, "instance is waking elsewhere; retry shortly", http.StatusServiceUnavailable)
5456 case errors.Is(err, errWakeTimeout):
5457 w.Header().Set("Retry-After", "5")
5458 edge.MarkAnswered(r)
5459 http.Error(w, "instance is waking; retry shortly", http.StatusServiceUnavailable)
5460 case errors.Is(err, context.Canceled):
5461 // The client hung up while queued; nothing to write.
5462 default:
5463 // The wake failure is already the member's platform log error
5464 // (this log line, shipped by logsink.go); marking it keeps the
5465 // edge from filing the same failure a second time.
5466 if a.allowFilesDiagnostic(rt.Host, err, "http") {
5467 log.Printf("activator: wake %s: %v", rt.Host, err)
5468 }
5469 edge.MarkAnswered(r)
5470 http.Error(w, "instance failed to wake", http.StatusServiceUnavailable)
5471 }
5472 return true
5473}
5474
5475// SetIdleFor installs the plans idle hook (plans 2). A plan can only shorten
5476// an entry's idle window: the sweep takes the smaller of the entry's own
5477// window and the hook's answer, and ignores 0. Always-on workers have no idle
5478// clock, so the hook never reaches them. nil removes the hook.
5479func (a *Service) SetIdleFor(f func(registry.Route) time.Duration) {
5480 if f == nil {
5481 a.idleFor.Store(nil)
5482 return
5483 }
5484 a.idleFor.Store(&f)
5485}
5486