| index 4b8534902..a650a4c7b 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -330,11 +330,14 @@ type actEntry struct { |
| // run that lasted sidecarHealthyRun. |
| workerBackoff time.Duration |
| workerRetryAt time.Time |
| - // workerRespawning is set while convergeWorkerLocked's goroutine owns this |
| + // workerRespawning is set while respawnWorkerLocked's goroutine owns this |
| // entry's child: the sweeper skips the entry entirely so the moment |
| // between the old child's stop and the new one's spawn — when the proc |
| - // reads as Exited — is not mistaken for a crash. |
| + // reads as Exited — is not mistaken for a crash. respawnDone is closed |
| + // when that goroutine finishes, and is what stop() waits on: a stop must |
| + // not release the lease while a spawn against the same DB is in flight. |
| workerRespawning bool |
| + respawnDone chan struct{} |
| } |
| |
| // Sidecar restart pacing — Restart=always semantics without a supervisor: the |
| @@ -461,6 +464,10 @@ type Service struct { |
| |
| mu sync.Mutex |
| entries map[string]*actEntry |
| + // shuttingDown is set by Shutdown before it stops anything, and read by |
| + // the supervised-respawn goroutines: the agent is going away, so nothing |
| + // new may be spawned against a DB whose lease is about to be released. |
| + shuttingDown bool |
| |
| sweepStop chan struct{} |
| sweepDone chan struct{} |
| @@ -720,7 +727,6 @@ 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 |
| @@ -752,10 +758,6 @@ 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 |
| @@ -825,14 +827,21 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| if e.worker { |
| // 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. |
| + // is a wake failure like any other: release and report. |
| token, err = newWorkerToken() |
| if err != nil { |
| _ = lease.Release(ctx) |
| return nil, fmt.Errorf("start %s: %w", host, err) |
| } |
| + // Published onto the entry NOW, not at the awake transition: the |
| + // child's first act is to call /v1/ready over the agent socket, which |
| + // authorizes against WorkerToken — and that call is what ENDS the |
| + // settle window. A token that only appeared once the entry was awake |
| + // would make the early-ready path unreachable: the socket would |
| + // refuse the very call it is waiting for, and every wake would pay the |
| + // full window. Publishing early gives nothing away: a wake that fails |
| + // deletes its entry, so the token still dies with it. |
| + a.setWorkerToken(e, token) |
| proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, token)) |
| } else { |
| proc, spawnVer, err = a.backend.Start(ctx, rt) |
| @@ -876,7 +885,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, workerToken: token} |
| + w := &wokenInstance{rep: rep, proc: proc, spawnVersion: spawnVer} |
| // 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 |
| @@ -1069,6 +1078,24 @@ func (a *Service) WakeWorker(rt registry.Route) error { |
| |
| switch e.state { |
| case actAwake: |
| + // Same stand-down beginRequest makes: a lost lease or a dead child |
| + // found at wake time turns into a stop, and the loop then re-wakes |
| + // it. Without this, a boot pass or an operator command would report |
| + // success for a worker whose lease is gone — the one answer that |
| + // makes a broken sidecar look healthy. A respawn in flight is not a |
| + // dead child: its proc is mid-replacement, so leave it be. |
| + if !e.workerRespawning { |
| + if repErr := e.rep.Err(); repErr != nil || e.proc.Exited() { |
| + if repErr != nil { |
| + log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| + } else { |
| + log.Printf("activator: worker %s exited — recycling", e.host) |
| + } |
| + a.beginStopLocked(e) |
| + a.mu.Unlock() |
| + continue |
| + } |
| + } |
| a.mu.Unlock() |
| return nil // already up: a no-op, not a second child |
| |
| @@ -1123,26 +1150,21 @@ func workerEnv(socket, host, token string) map[string]string { |
| } |
| } |
| |
| -// 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. |
| +// restartWorkerLocked replaces an ALWAYS-ON worker's dead child — 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. |
| +// paces itself instead of hot-looping the box. The pacing is decided here, |
| +// under the lock; the exec itself is not (see respawnWorkerLocked). |
| // |
| // 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. |
| +// runs with no lease churn. |
| func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) { |
| if !e.spawnedAt.IsZero() && now.Sub(e.spawnedAt) >= sidecarHealthyRun { |
| e.workerBackoff = 0 |
| @@ -1156,96 +1178,123 @@ func (a *Service) restartWorkerLocked(e *actEntry, now time.Time) { |
| 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 |
| + log.Printf("activator: worker %s exited — restarting (backoff now %s)", e.host, e.workerBackoff) |
| + // nil: this child is already dead, so there is nothing to retire first. |
| + a.respawnWorkerLocked(e, nil, "exited") |
| } |
| |
| // convergeWorkerLocked replaces an always-on worker's child with one exec'd |
| -// from the build live/<host> now points at, on the SAME entry — the caller |
| -// holds a.mu and has already established the drift persisted past |
| -// ConvergeGrace and that the per-sweep budget allows it. |
| +// from the build live/<host> now points at — the caller holds a.mu and has |
| +// already established the drift persisted past ConvergeGrace and that the |
| +// per-sweep budget allows it. The old child is retired first: it is alive, and |
| +// two incarnations of an outbound worker would double its work. |
| +func (a *Service) convergeWorkerLocked(e *actEntry) { |
| + a.respawnWorkerLocked(e, e.proc, "converging onto a new build") |
| +} |
| + |
| +// workerSpawnTimeout bounds a supervised respawn's exec. It is generous |
| +// because ExecBackend.StartWorker may materialize the build from the bucket |
| +// first (a dangling live/<host> after a retention prune), but it is BOUNDED: |
| +// an unbounded spawn that never returns would leave workerRespawning set, and |
| +// the sweeper skips a respawning entry — no stand-down, no restart, no idle, |
| +// and nothing able to wake it. A wedged fetch must cost one entry a couple of |
| +// minutes, not the entry forever. |
| +const workerSpawnTimeout = 2 * time.Minute |
| + |
| +// respawnWorkerLocked replaces a worker's child on the SAME entry, keeping its |
| +// lease and replication: only the process is new. The caller holds a.mu. |
| +// stopOld is the child to retire first (nil when it is already dead). |
| // |
| -// Stop THEN spawn, and both off the lock, on their own goroutine: the old |
| -// child's SIGTERM drain is bounded but not instant, and nothing may hold a.mu |
| -// across it — every request wake on the box queues behind that lock. Stopping |
| -// first (rather than spawning first and stopping after) is deliberate: a |
| -// sidecar is an outbound worker, so two live incarnations would double its |
| -// work — a bot posting twice — where a brief gap costs only latency. |
| +// Everything expensive happens OFF a.mu, on its own goroutine. That is not a |
| +// stylistic preference: ExecBackend.StartWorker can materialize a build from |
| +// the bucket, and the SIGTERM drain of the outgoing child is bounded but not |
| +// instant — either one held across the global mutex would stall every request |
| +// wake, Owns and Shutdown on the box. (The co-spawn supervisor above DOES exec |
| +// under the lock; that is StartSidecar, which never fetches anything. The two |
| +// are not interchangeable.) |
| // |
| -// e.workerRespawning is the interlock. While it is set the sweeper skips the |
| -// entry entirely, so the mid-respawn moment when the old proc reads as Exited |
| -// is not mistaken for a crash (which would spawn a SECOND child). If the |
| -// entry is stopped or replaced while the respawn is in flight, the new child |
| -// is stopped instead of installed — a stop always wins. |
| +// Three orderings carry the safety: |
| +// - Retire, THEN spawn. A sidecar is an outbound worker: two live |
| +// incarnations double its work, where a gap costs only latency. |
| +// - Ask again, under the lock, immediately BEFORE spawning. A stop that |
| +// landed while the old child was draining has already released the lease, |
| +// and a child spawned after that would be writing the tenant's DB with no |
| +// single-writer claim behind it. Shedding it afterwards is not enough — |
| +// the agent may be gone by then. |
| +// - Check once more before INSTALLING, and shed the child if the entry was |
| +// stopped or replaced meanwhile. A stop always wins. |
| +// |
| +// e.workerRespawning is the sweeper's interlock (mid-respawn the entry's proc |
| +// reads as exited, which read as a crash would spawn a second child), and |
| +// e.respawnDone is what stop() and Shutdown wait on so a stop never completes |
| +// — nor the process exit — while a spawn is still in flight. |
| // |
| // A failed spawn deliberately leaves the dead child in place: the next sweep |
| -// sees Exited and the crash-restart path takes over, with its backoff. The |
| -// convergence attempt degrades into supervision rather than into a hole. |
| -func (a *Service) convergeWorkerLocked(e *actEntry) { |
| +// sees Exited and the crash-restart path takes over with its backoff. A failed |
| +// converge degrades into supervision rather than into a hole. |
| +func (a *Service) respawnWorkerLocked(e *actEntry, stopOld Proc, why string) { |
| e.workerRespawning = true |
| - old, rt, host := e.proc, e.route, e.host |
| + e.respawnDone = make(chan struct{}) |
| + done, rt, host := e.respawnDone, e.route, e.host |
| go func() { |
| - stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| - if err := old.Stop(stopCtx); err != nil { |
| - log.Printf("activator: worker %s converge: stopping the old child: %v", host, err) |
| - } |
| - cancel() |
| - |
| var ( |
| proc Proc |
| spawnVer string |
| + tok string |
| + err error |
| ) |
| - tok, err := newWorkerToken() |
| - if err == nil { |
| - proc, spawnVer, err = a.backend.StartWorker(context.Background(), rt, workerEnv(a.agentSocket(), host, tok)) |
| + if stopOld != nil { |
| + stopCtx, cancel := context.WithTimeout(context.Background(), actStopTimeout) |
| + if serr := stopOld.Stop(stopCtx); serr != nil { |
| + log.Printf("activator: worker %s respawn: retiring the old child: %v", host, serr) |
| + } |
| + cancel() |
| + } |
| + |
| + // Still wanted? Asked HERE, before anything is spawned: past this |
| + // point a stop can only be raced, never prevented. |
| + a.mu.Lock() |
| + wanted := a.entries[host] == e && e.state == actAwake && !a.shuttingDown |
| + a.mu.Unlock() |
| + |
| + if wanted { |
| + ctx, cancel := context.WithTimeout(context.Background(), workerSpawnTimeout) |
| + if tok, err = newWorkerToken(); err == nil { |
| + proc, spawnVer, err = a.backend.StartWorker(ctx, rt, workerEnv(a.agentSocket(), host, tok)) |
| + } |
| + cancel() |
| } |
| |
| a.mu.Lock() |
| e.workerRespawning = false |
| - live := a.entries[host] == e && e.state == actAwake |
| - if err != nil { |
| + close(done) |
| + live := a.entries[host] == e && e.state == actAwake && !a.shuttingDown |
| + switch { |
| + case !wanted || !live: |
| a.mu.Unlock() |
| - log.Printf("activator: WORKER %s failed to respawn onto the new build: %v (the sweep's crash restart takes it from here)", host, err) |
| - emitEvent(a.Logs, rt, "error", "sidecar could not restart onto the new build — retrying") |
| - return |
| - } |
| - if !live { |
| - // Stopped or replaced under us — a stop always wins. Shed the |
| - // child we just made rather than installing it over the top. |
| + log.Printf("activator: worker %s stopped mid-respawn (%s) — abandoning it", host, why) |
| + if proc != nil { |
| + // Spawned into a window that closed under us: shed it rather |
| + // than leave a child nothing owns. |
| + shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| + _ = proc.Stop(shedCtx) |
| + shedCancel() |
| + } |
| + case err != nil: |
| a.mu.Unlock() |
| - log.Printf("activator: worker %s stopped mid-converge — discarding the respawned child", host) |
| - shedCtx, shedCancel := context.WithTimeout(context.Background(), sidecarStopTimeout) |
| - _ = proc.Stop(shedCtx) |
| - shedCancel() |
| - return |
| + log.Printf("activator: WORKER %s failed to respawn (%s): %v — the sweep's crash restart takes it from here", host, why, err) |
| + emitEvent(a.Logs, rt, "error", "sidecar could not be restarted — retrying") |
| + default: |
| + e.proc = proc |
| + e.workerToken = tok |
| + e.spawnVersion = spawnVer |
| + e.staleSince = time.Time{} // fresh exec: it cannot be stale yet |
| + e.spawnedAt = a.now() |
| + e.lastActive = a.now() |
| + a.mu.Unlock() |
| + log.Printf("activator: worker %s respawned (%s), now running %s", host, why, spawnVer) |
| + emitEvent(a.Logs, rt, "info", "sidecar restarted") |
| } |
| - e.proc = proc |
| - e.workerToken = tok |
| - e.spawnVersion = spawnVer |
| - e.staleSince = time.Time{} // fresh exec: it cannot be stale yet |
| - e.spawnedAt = a.now() |
| - e.lastActive = a.now() |
| - a.mu.Unlock() |
| - log.Printf("activator: worker %s respawned onto %s", host, spawnVer) |
| - emitEvent(a.Logs, rt, "info", "sidecar restarted onto a new build") |
| }() |
| } |
| |
| @@ -1321,19 +1370,35 @@ func (a *Service) SleepWorker(host string) { |
| a.beginStopLocked(e) |
| } |
| |
| -// WorkerToken reports the live per-wake token for host, or "" — the agent |
| -// socket's authorization check. A token from a previous wake never matches, |
| -// and a host that is asleep, waking, stopping or not a worker at all has no |
| -// token, so every such call is refused. |
| +// WorkerToken reports the live per-incarnation token for host, or "" — the |
| +// agent socket's authorization check. A token from a previous wake, or from a |
| +// previous incarnation of a restarted child, never matches. |
| +// |
| +// It answers for a WAKING entry as well as an awake one, because the child's |
| +// first call — /v1/ready, which ends the settle window — necessarily arrives |
| +// while the wake is still resolving. Refusing then would make the early-ready |
| +// path unreachable and cost every wake the full window. Nothing else is |
| +// loosened: a stopping entry, an asleep host, an instance and an unknown host |
| +// all still get "", and a failed wake takes its token with it when the entry |
| +// is deleted. |
| func (a *Service) WorkerToken(host string) string { |
| a.mu.Lock() |
| defer a.mu.Unlock() |
| - if e := a.entries[host]; e != nil && e.worker && e.state == actAwake { |
| + if e := a.entries[host]; e != nil && e.worker && (e.state == actAwake || e.state == actWaking) { |
| return e.workerToken |
| } |
| return "" |
| } |
| |
| +// setWorkerToken publishes a wake's freshly minted credential onto the entry. |
| +// Its own critical section because doWake runs off a.mu — every other write to |
| +// an entry's fields is likewise made under the lock. |
| +func (a *Service) setWorkerToken(e *actEntry, tok string) { |
| + a.mu.Lock() |
| + e.workerToken = tok |
| + a.mu.Unlock() |
| +} |
| + |
| // stop drives awake→asleep: stop the child (its SIGTERM drain is the final |
| // write), checkpoint what's on disk to S3 and release the lease, then forget |
| // the entry. The local DB stays in place — it's the warm cache for the next |
| @@ -1349,6 +1414,26 @@ func (a *Service) stop(e *actEntry) { |
| } |
| emitEvent(a.Logs, e.route, "info", stopping) |
| |
| + // Wait out a supervised respawn first. Everything below — the final |
| + // checkpoint and the LEASE RELEASE — assumes no other process is being |
| + // pointed at this DB; a respawn that finished after the release would run |
| + // a child with no single-writer claim behind it, on the one path whose |
| + // whole purpose is single-writer. The respawn's own pre-spawn check makes |
| + // this wait short (it aborts as soon as it sees this state flip); the |
| + // bound is insurance against a wedged exec, and instances never wait at |
| + // all (respawnDone is nil for them). |
| + a.mu.Lock() |
| + waitRespawn := e.respawnDone |
| + respawning := e.workerRespawning |
| + a.mu.Unlock() |
| + if respawning && waitRespawn != nil { |
| + select { |
| + case <-waitRespawn: |
| + case <-time.After(workerSpawnTimeout + actStopTimeout): |
| + log.Printf("activator: %s respawn did not finish before its stop — proceeding without it", e.host) |
| + } |
| + } |
| + |
| // Co-stop, sidecar FIRST — before the next-due probe and the instance's |
| // own SIGTERM. Two reasons: the sidecar must never outlive its instance |
| // (an orphan's reconnect loop would be a standing wake source — the |
| @@ -1729,6 +1814,14 @@ func (a *Service) Shutdown(ctx context.Context) { |
| // below, so peers stop proxying here immediately and their wakes succeed |
| // the moment each lease releases. Deploys, box failure drills, and |
| // scale-down all ride this same path. (`a.mesh.stop(ctx)` went here.) |
| + // Announce the shutdown BEFORE anything stops: a supervised respawn in |
| + // flight checks this flag immediately before it spawns, so it abandons |
| + // rather than starting a child this shutdown would then have to chase. |
| + // (stop() waits out any respawn that got past that check, so the waits |
| + // below cover the rest.) |
| + a.mu.Lock() |
| + a.shuttingDown = true |
| + a.mu.Unlock() |
| if a.sweepStop != nil { |
| close(a.sweepStop) |
| <-a.sweepDone |
amadan