| index de4a994f0..2c4d8730c 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -309,18 +309,27 @@ type actEntry struct { |
| // proc, not a passenger on someone else's: the co-spawn fields above stay |
| // zero for it, and it never becomes a co-spawn parent either. |
| // |
| - // worker, workerIdle, workerMaxRun and workerToken are set once, at entry |
| + // worker, workerIdle, workerMaxRun and readyc are set once, at entry |
| // construction, BEFORE the wake goroutine is launched, and never written |
| // again — so wake/doWake read them off-lock safely (the `go` statement is |
| - // the happens-before edge). spawnedAt and workerVersion are written under |
| - // a.mu like every other post-wake field. |
| + // the happens-before edge). Everything else below is written under a.mu, |
| + // like every other post-wake field: workerToken and spawnedAt at the awake |
| + // transition and on a supervised restart, the rest by the agent socket. |
| worker bool // this entry is a sidecar: no socket, no requests |
| workerIdle time.Duration // 0 = fall back to cfg.Idle |
| workerMaxRun time.Duration // 0 = no ceiling |
| - workerToken string // the per-wake agent-socket token |
| + readyc chan struct{} // closed when the child reports ready (ends the settle window) |
| + readyNoted bool // guards readyc's one close |
| |
| - spawnedAt time.Time // when this wake's child spawned — the max_run clock |
| + workerToken string // the per-wake agent-socket token |
| + spawnedAt time.Time // when this child spawned — the max_run and healthy-run clock |
| workerVersion string // the build the child reported over the agent socket |
| + |
| + // Crash-restart pacing for an always-on worker, the co-spawn supervisor's |
| + // rule applied to a standalone member: doubling, capped, and reset by a |
| + // run that lasted sidecarHealthyRun. |
| + workerBackoff time.Duration |
| + workerRetryAt time.Time |
| } |
| |
| // Sidecar restart pacing — Restart=always semantics without a supervisor: the |
| @@ -338,6 +347,30 @@ const ( |
| // fallback behind this timeout is insurance. |
| const sidecarStopTimeout = 5 * time.Second |
| |
| +// sidecarSettle is the standalone worker's readiness window — what stands in |
| +// for the health probe a child that binds nothing can never answer. A worker |
| +// is up once it either reports ready over the agent socket (NoteWorkerReady) |
| +// or survives this long without exiting; a child that exits INSIDE the window |
| +// is a failed wake, released and surfaced exactly like a failed instance |
| +// start. A var, not a const, so tests can shrink it — spawnHealthTimeout's |
| +// precedent in this package (backend_exec.go). |
| +var sidecarSettle = 10 * time.Second |
| + |
| +// settlePoll derives the settle window's liveness poll: Proc reports death |
| +// through Exited(), not a channel, so the window polls. Scaled off the window |
| +// so a shrunken test window still gets several looks, and capped so the |
| +// production window doesn't spin. |
| +func settlePoll(window time.Duration) time.Duration { |
| + poll := window / 8 |
| + if poll > 250*time.Millisecond { |
| + poll = 250 * time.Millisecond |
| + } |
| + if poll < time.Millisecond { |
| + poll = time.Millisecond |
| + } |
| + return poll |
| +} |
| + |
| // Build convergence: how long an awake child may serve a symlink-stale binary |
| // before the sweep gracefully cycles it, and how many to cycle per tick. |
| // - ConvergeGrace comfortably exceeds one sweep interval (≤30s) and a |
| @@ -682,6 +715,7 @@ func (a *Service) wake(e *actEntry) { |
| // (it costs a field write) but read only for workers: an instance's |
| // run length is bounded by its idle window and its members. |
| e.spawnedAt = a.now() |
| + e.workerToken = w.workerToken |
| // The build this child actually execed, captured at Start() inside the |
| // backend (not re-read here — a ship during the wake window would make a |
| // re-read name the new target while the process runs the old binary, and |
| @@ -713,6 +747,10 @@ type wokenInstance struct { |
| sidecar Proc |
| sidecarWanted bool |
| spawnVersion string // live/<host> target read at exec — for build convergence |
| + // workerToken is this wake's agent-socket credential, minted inside the |
| + // wake and carried out here so it is published onto the entry under a.mu, |
| + // at the awake transition, and never for a wake that failed. |
| + workerToken string |
| } |
| |
| // doWake runs the side effects of one wake. It takes the ENTRY, not the bare |
| @@ -777,9 +815,20 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| var ( |
| proc Proc |
| spawnVer string |
| + token string |
| ) |
| if e.worker { |
| - proc, spawnVer, err = a.backend.StartWorker(ctx, rt, a.workerEnv(e)) |
| + // Mint this wake's token here — under the lease, after the restore, |
| + // immediately before the spawn (design §7.2's order). A mint failure |
| + // is a wake failure like any other: release and report. It stays a |
| + // local until the wake succeeds, so nothing publishes a credential for |
| + // a child that never came up. |
| + token, err = newWorkerToken() |
| + if err != nil { |
| + _ = lease.Release(ctx) |
| + return nil, fmt.Errorf("start %s: %w", host, err) |
| + } |
| + proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, token)) |
| } else { |
| proc, spawnVer, err = a.backend.Start(ctx, rt) |
| } |
| @@ -788,6 +837,21 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| return nil, fmt.Errorf("start %s: %w", host, err) |
| } |
| |
| + // 4b. The settle window, workers only: the readiness check for a child |
| + // with nothing to probe. A child that dies inside it never becomes an |
| + // awake entry — the lease goes back and the wake reports, exactly as a |
| + // failed Start does, so a sidecar that crashes on boot doesn't sit in the |
| + // map holding a lease until someone notices. |
| + if e.worker { |
| + if serr := a.awaitWorkerSettle(ctx, e, proc); serr != nil { |
| + stopCtx, cancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + _ = proc.Stop(stopCtx) |
| + cancel() |
| + _ = lease.Release(ctx) |
| + return nil, fmt.Errorf("start %s: %w", host, serr) |
| + } |
| + } |
| + |
| // 5. Trade the held lease for running replication, at the route's |
| // replication-class sync interval (the per-plan durability ladder). |
| // The DB path rides the route; when empty the backend defaults it to |
| @@ -807,7 +871,7 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // marked requests pass through the edge while we're awake). A spawn |
| // failure is loud but never fails the wake: members come before bots, and |
| // the sweeper retries with backoff. |
| - w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer} |
| + w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer, workerToken: token} |
| // A worker is nobody's parent: it IS a sidecar, so it never co-spawns one. |
| // Skipped outright rather than left to HasSidecar's answer — the staging |
| // file is keyed by host, and a host that happened to have one staged must |
| @@ -898,24 +962,69 @@ func (a *Service) newWorkerEntry(rt registry.Route) *actEntry { |
| worker: true, |
| workerIdle: time.Duration(rt.IdleSeconds) * time.Second, |
| workerMaxRun: time.Duration(rt.MaxRunSeconds) * time.Second, |
| - workerToken: newWorkerToken(), |
| + readyc: make(chan struct{}), |
| } |
| } |
| |
| +// randRead is crypto/rand's entropy, behind a seam so a test can make minting |
| +// fail. Nothing else in this package should read entropy. |
| +var randRead = rand.Read |
| + |
| // newWorkerToken mints the per-wake agent-socket credential. A fresh one every |
| // wake is the point: a token from a previous incarnation of the child never |
| // matches, so an orphan that survived a stop cannot keep talking to the agent |
| -// as though it were the live worker. A rand failure is fatal to the wake, not |
| -// silently downgraded — an empty token would authorize nothing (WorkerToken's |
| -// "" is the refusal), but a PREDICTABLE one would authorize everyone. |
| -func newWorkerToken() string { |
| +// as though it were the live worker. |
| +// |
| +// It returns an error rather than panicking. The agent supervises every route |
| +// on the box — one host that cannot mint a token must fail ITS wake (the lease |
| +// released, like any other wake failure), never take the process, and every |
| +// other tenant on the box, down with it. Deliberately no fallback source: an |
| +// empty token would authorize nothing, but a predictable one would authorize |
| +// everyone. |
| +func newWorkerToken() (string, error) { |
| var b [32]byte |
| - if _, err := rand.Read(b[:]); err != nil { |
| - // crypto/rand.Read never returns an error on any platform we ship to; |
| - // if it somehow did, an unguessable-token-less wake must not proceed. |
| - panic("activator: crypto/rand failed: " + err.Error()) |
| + if _, err := randRead(b[:]); err != nil { |
| + return "", fmt.Errorf("mint worker token: %w", err) |
| + } |
| + return hex.EncodeToString(b[:]), nil |
| +} |
| + |
| +// errWorkerExitedEarly: the child died inside its settle window. That is a |
| +// failed wake — a crash on boot (bad argv, missing config, panic at start) is |
| +// exactly what the instance path's health probe would have caught. |
| +var errWorkerExitedEarly = errors.New("worker exited during its settle window") |
| + |
| +// awaitWorkerSettle is the socketless child's health probe: it is up once it |
| +// says so over the agent socket, or once it has survived sidecarSettle without |
| +// exiting. Real time, not a.now — like waitHealthy, this bounds a real |
| +// process's first seconds, and the test clock exists to move idle windows, not |
| +// to model a boot. |
| +func (a *Service) awaitWorkerSettle(ctx context.Context, e *actEntry, proc Proc) error { |
| + window := sidecarSettle |
| + if window <= 0 { |
| + return nil |
| + } |
| + deadline := time.NewTimer(window) |
| + defer deadline.Stop() |
| + tick := time.NewTicker(settlePoll(window)) |
| + defer tick.Stop() |
| + for { |
| + select { |
| + case <-e.readyc: |
| + return nil // the child reported ready: up, early |
| + case <-deadline.C: |
| + if proc.Exited() { |
| + return errWorkerExitedEarly |
| + } |
| + return nil |
| + case <-tick.C: |
| + if proc.Exited() { |
| + return errWorkerExitedEarly |
| + } |
| + case <-ctx.Done(): |
| + return ctx.Err() |
| + } |
| } |
| - return hex.EncodeToString(b[:]) |
| } |
| |
| // WakeWorker brings a sidecar up, or does nothing if it is already awake. |
| @@ -984,23 +1093,86 @@ func (a *Service) WakeWorker(rt registry.Route) error { |
| } |
| } |
| |
| -// workerEnv is what the wake hands the child: where to call back, who it is, |
| -// and this wake's credential. The backend applies it OVER the tenant's own env |
| -// file so a tenant cannot forge its own token (StartWorker's contract). |
| -func (a *Service) workerEnv(e *actEntry) map[string]string { |
| - sock := a.cfg.AgentSocket |
| - if sock == "" { |
| - sock = DefaultAgentSocket |
| +// agentSocket is where a worker child dials back. |
| +func (a *Service) agentSocket() string { |
| + if a.cfg.AgentSocket == "" { |
| + return DefaultAgentSocket |
| } |
| - name, _, _ := strings.Cut(e.host, ".") // the sidecar's declared name |
| + return a.cfg.AgentSocket |
| +} |
| + |
| +// workerEnv is what a spawn hands the child: where to call back, who it is, |
| +// and this incarnation's credential. The backend applies it OVER the tenant's |
| +// own env file so a tenant cannot forge its own token (StartWorker's |
| +// contract). A free function of plain strings, not a method on the entry, so |
| +// the two callers — the wake (off-lock, with a token still in a local) and the |
| +// crash restart (under a.mu) — both compose it without touching entry state |
| +// they don't own. |
| +func workerEnv(socket, host, token string) map[string]string { |
| + name, _, _ := strings.Cut(host, ".") // the sidecar's declared name |
| return map[string]string{ |
| - "CARLOS_AGENT_SOCKET": sock, |
| - "CARLOS_SIDECAR_TOKEN": e.workerToken, |
| - "CARLOS_SIDECAR_HOST": e.host, |
| + "CARLOS_AGENT_SOCKET": socket, |
| + "CARLOS_SIDECAR_TOKEN": token, |
| + "CARLOS_SIDECAR_HOST": host, |
| "CARLOS_SIDECAR_NAME": name, |
| } |
| } |
| |
| +// restartWorkerLocked respawns an ALWAYS-ON worker whose child died, in place |
| +// — the caller holds a.mu and has already established that e is an awake |
| +// worker with `hibernate: false` and an exited proc. |
| +// |
| +// Restart=always without a supervisor, the co-spawn path's rule applied to a |
| +// standalone member: doubling backoff from sidecarRestartBase to |
| +// sidecarRestartMax, and a run that lasted sidecarHealthyRun earns a fresh |
| +// one, so a crash a day never escalates to the cap. The first crash respawns |
| +// immediately and ARMS the delay for the next one — a crash loop therefore |
| +// paces itself instead of hot-looping the box. |
| +// |
| +// In place, not through a stop and re-wake: the lease and the replication are |
| +// still ours (only the child died), and design §7.2 says an always-on worker |
| +// runs with no lease churn. Spawned under a.mu on purpose, exactly as the |
| +// co-spawn restart above is: an exec is quick, and the lock is what makes a |
| +// restart impossible to race with a stop. |
| +// |
| +// A new token every incarnation, and only ON SUCCESS: a failed respawn must |
| +// not invalidate the credential of a child that might still be limping along, |
| +// and the dead child's token must not survive into the next one. |
| +func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) { |
| + if !e.spawnedAt.IsZero() && now.Sub(e.spawnedAt) >= sidecarHealthyRun { |
| + e.workerBackoff = 0 |
| + } |
| + if now.Before(e.workerRetryAt) { |
| + return // still backing off; the child stays dead until it elapses |
| + } |
| + if e.workerBackoff == 0 { |
| + e.workerBackoff = sidecarRestartBase |
| + } else { |
| + e.workerBackoff = min(e.workerBackoff*2, sidecarRestartMax) |
| + } |
| + e.workerRetryAt = now.Add(e.workerBackoff) |
| + |
| + tok, err := newWorkerToken() |
| + if err != nil { |
| + log.Printf("activator: WORKER %s restart: %v (next attempt in %s)", e.host, err, e.workerBackoff) |
| + return |
| + } |
| + proc, spawnVer, err := a.backend.StartWorker(context.Background(), e.route, workerEnv(a.agentSocket(), e.host, tok)) |
| + if err != nil { |
| + log.Printf("activator: WORKER %s exited and failed to restart: %v (next attempt in %s)", e.host, err, e.workerBackoff) |
| + emitEvent(a.Logs, e.route, "error", "sidecar exited and could not be restarted — retrying") |
| + return |
| + } |
| + log.Printf("activator: worker %s exited — restarted (backoff now %s)", e.host, e.workerBackoff) |
| + emitEvent(a.Logs, e.route, "error", "sidecar exited — restarted") |
| + e.proc = proc |
| + e.workerToken = tok |
| + e.spawnVersion = spawnVer |
| + e.staleSince = time.Time{} // fresh exec: it can't be stale yet |
| + e.spawnedAt = now |
| + e.lastActive = now |
| +} |
| + |
| // NoteWorkerContact stamps last-activity from an agent-socket call. It is a |
| // worker's only sign of life: nothing routes to a sidecar, so no request will |
| // ever stamp one, and without contact the idle clock runs to its window and |
| @@ -1013,17 +1185,28 @@ func (a *Service) NoteWorkerContact(host string) { |
| } |
| } |
| |
| -// NoteWorkerReady records a worker's own "I am up" — last-activity plus the |
| -// build it reports running. Deliberately the child's claim, not ours: §9.2's |
| -// honesty rule is that a sidecar which never reports shows unknown, never |
| -// "current". (Task 8 additionally persists the version through the registry's |
| -// version setter; this is the in-memory half the agent socket needs today.) |
| +// NoteWorkerReady records a worker's own "I am up" — last-activity, the build |
| +// it reports running, and the end of its settle window. Deliberately the |
| +// child's claim, not ours: §9.2's honesty rule is that a sidecar which never |
| +// reports shows unknown, never "current". (Task 8 additionally persists the |
| +// version through the registry's version setter; this is the in-memory half |
| +// the agent socket needs today.) |
| +// |
| +// It acts on a WAKING entry as well as an awake one — that is the whole point |
| +// of the settle window: a child that boots fast says so and its wake resolves |
| +// immediately instead of waiting out sidecarSettle. |
| func (a *Service) NoteWorkerReady(host, version string) { |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| - if e := a.entries[host]; e != nil && e.worker { |
| - e.lastActive = a.now() |
| - e.workerVersion = version |
| + e := a.entries[host] |
| + if e == nil || !e.worker { |
| + return |
| + } |
| + e.lastActive = a.now() |
| + e.workerVersion = version |
| + if !e.readyNoted { |
| + e.readyNoted = true |
| + close(e.readyc) |
| } |
| } |
| |
| @@ -1204,6 +1387,17 @@ func (a *Service) sweepOnce() { |
| continue |
| } |
| if e.proc.Exited() { |
| + // An ALWAYS-ON worker's exit is a crash: `hibernate: false` means |
| + // the record exists ⇒ the process runs, so it is restarted in |
| + // place — same supervision as the co-spawn path, and no lease |
| + // churn, because only the child died (the lease and replication |
| + // are still ours). A HIBERNATING worker's exit means "I finished |
| + // my work", and the stand-down below is exactly right for it — |
| + // as it is for an instance. |
| + if e.worker && !e.route.Hibernate { |
| + a.restartWorkerLocked(e, now) |
| + continue |
| + } |
| log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| a.beginStopLocked(e) |
| continue |
| @@ -1217,20 +1411,28 @@ func (a *Service) sweepOnce() { |
| // lastActive is stamped by agent-socket contact instead of by a |
| // released request, so the same expression means the right thing for |
| // both. |
| + // |
| + // An ALWAYS-ON worker has no idle clock at all (design §7.2: |
| + // "hibernate: false … no idle clock, no lease churn"), and no run |
| + // ceiling either. Both would strand it: a stopped always-on worker has |
| + // no wake source — no request can reach it, and it has no schedule — |
| + // so it would stay down until the agent restarted. An instance is |
| + // never in this arm: unflagged instances don't enter the activator. |
| + hibernates := !e.worker || e.route.Hibernate |
| idleWindow := a.cfg.Idle |
| if e.workerIdle > 0 { |
| idleWindow = e.workerIdle |
| } |
| - if e.inflight == 0 && now.Sub(e.lastActive) >= idleWindow { |
| + if hibernates && e.inflight == 0 && now.Sub(e.lastActive) >= idleWindow { |
| log.Printf("activator: %s idle %s — hibernating", e.host, now.Sub(e.lastActive).Truncate(time.Second)) |
| a.beginStopLocked(e) |
| continue |
| } |
| // A worker that never reports idle still has a ceiling: a wedged |
| // reconnect loop must not hold a lease and a box forever. Zero means |
| - // no ceiling — an always-on sidecar is a legitimate shape, so this |
| - // must never invent one nobody declared. |
| - if e.worker && e.workerMaxRun > 0 && now.Sub(e.spawnedAt) >= e.workerMaxRun { |
| + // no ceiling — a sidecar that declared none is a legitimate shape, so |
| + // this must never invent one nobody asked for. |
| + if e.worker && hibernates && e.workerMaxRun > 0 && now.Sub(e.spawnedAt) >= e.workerMaxRun { |
| log.Printf("activator: %s hit its max_run of %s — stopping", e.host, e.workerMaxRun) |
| a.beginStopLocked(e) |
| continue |
| @@ -1253,6 +1455,16 @@ func (a *Service) sweepOnce() { |
| cur, readErr := a.liveVer(e.host) |
| cycle, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, now, ConvergeGrace) |
| e.staleSince = next |
| + // An always-on worker is NOT cycled here: the cycle's second half is |
| + // "something wakes it back up", and for a worker with no request path |
| + // and no schedule there is nothing to. It would convergence-cycle |
| + // itself off the box. It converges instead at the agent's next restart |
| + // or through `carlos restart` (Task 11) — a real gap, deliberately |
| + // left visible rather than closed with a stop that strands a member's |
| + // bot. |
| + if cycle && e.worker && !e.route.Hibernate { |
| + cycle = false |
| + } |
| if cycle && cycled < maxConvergeCyclesPerSweep { |
| log.Printf("activator: %s serving %s but live symlink is now %s — cycling onto the new build", e.host, e.spawnVersion, cur) |
| a.beginStopLocked(e) |
amadan