amadan
zdd-5 task 9: planned wakes, the first-wake budget and fallback, the 10 s observation
WakePlan, GenEvent, Current, BounceFor and the Planner/OnGen/OnDrift hooks. Every planned exec, B's and its fallback's, instance and worker, goes through startPinnedAt or startWorker with planGate: the wake's own lease, ownership (ownedLocked waking) and the plan's selection check. startPinnedAt takes the first-wake health deadline; a worker spawn refused for publication (ErrBuildNotPublished) is stepStale in classifyStep. A planned generation carries its booted staged identity, so the legacy sidecar binds to it. Claims-on planned wakes set the start policy and refuse at the closed launch, recording nothing.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
7 files changed,
+1873
−16
internal/activator/activator.go+134 −6internal/activator/claimwake.go+56 −2internal/activator/swap.go+7 −0internal/activator/swaphold.go+17 −4internal/activator/wakeplan.go+558 −0internal/activator/wakeplan_test.go+1097 −0internal/activator/zdd5_surface_test.go+4 −4
diff --git a/internal/activator/activator.go b/internal/activator/activator.go| index 5046de730..e10f87baf 100644 |
| --- a/internal/activator/activator.go |
| +++ b/internal/activator/activator.go |
| @@ -461,7 +461,9 @@ type actEntry struct { |
| // that follows the hold's stop execs exactly this staged artifact rather |
| // than the live link. It is set at construction (holdWakeLocked), before |
| // the wake goroutine starts, and never written again, like the worker |
| - // fields. |
| + // fields. The one exception is a claims-on planned wake (plan 5, |
| + // doClaimWake), whose own wake goroutine sets it under a.mu before its |
| + // start policy is built; every reader is that goroutine or holds a.mu. |
| pin *Gen |
| // pinCheck is a hold's attempt checks, re-run under a.mu at the pinned |
| // exec. Set at construction (holdWakeLocked), like pin. |
| @@ -493,6 +495,24 @@ type actEntry struct { |
| // under a.mu. |
| genReaping bool |
| + // The rollout scheduler's view of this entry (zero-downtime plan 5). |
| + // plan is set by the wake goroutine before the spawn (doWake, |
| + // doClaimWake, checkPlan's replan) under a.mu, and read without it only |
| + // by that goroutine. firstWake is begin()'s cue to wait out the |
| + // first-wake budget; wakeStart (real time, set at the wake's start) |
| + // anchors B's health window. unexpectedExit marks a stop that began |
| + // because the child died on its own. firstWatch cancels a first wake's |
| + // observation when an intentional stop begins, and firstWatchCtx is |
| + // that observation's context, both registered in the awake transition |
| + // (armFirstWatchLocked) and read by watchFirstWake's goroutine. All but |
| + // plan are written under a.mu. |
| + plan WakePlan |
| + firstWake bool |
| + wakeStart time.Time |
| + unexpectedExit bool |
| + firstWatch context.CancelFunc |
| + firstWatchCtx context.Context |
| + |
| // Standalone sidecar ("worker") entries. A worker is this entry's own |
| // 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. |
| @@ -927,6 +947,19 @@ type Service struct { |
| // and fail OPEN, since a read error must not stop legitimate bindings. |
| OnWake func(rt registry.Route) |
| + // Planner, when set, says which build each wake boots (zero-downtime |
| + // spec §8.2). The rollout manager sets it with CARLOS_SWAP=overlap; nil |
| + // keeps today's wake from the live link. It is called without a.mu held. |
| + Planner func(ctx context.Context, rt registry.Route) WakePlan |
| + // OnGen, when set, hears every planned wake's outcome. It is never |
| + // called with a.mu held: it reaches the scheduler's goodMu, which calls |
| + // back into Current (emitGen). |
| + OnGen func(GenEvent) |
| + // OnDrift, when set, replaces the sweep's convergence stop: with the flag |
| + // on, the scheduler owns every move onto a new build (spec §10). It runs |
| + // on its own goroutine, never under a.mu. |
| + OnDrift func(host string) |
| + |
| // CoalesceRecycle, when set, is asked before a worker is recycled for a |
| // crash or max_run (spec §7.9). True means an attempt for that host (a |
| // queued release attempt admissible now, or one already running) will |
| @@ -1023,6 +1056,8 @@ func New(cfg Config, backend Backend, due DueStore) *Service { |
| func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (release func(), err error) { |
| deadline := time.NewTimer(wakeWaitBound) |
| defer deadline.Stop() |
| + arrival := time.Now() |
| + extended := false |
| for { |
| a.mu.Lock() |
| if a.fences[rt.Host] != 0 { |
| @@ -1059,6 +1094,9 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| + // Not an intentional stop: a first wake's observation |
| + // reads it as the build's failure (spec §6.4). |
| + e.unexpectedExit = true |
| } |
| a.beginStopLocked(e) |
| a.mu.Unlock() |
| @@ -1083,6 +1121,18 @@ func (a *Service) begin(ctx context.Context, rt registry.Route, stamp bool) (rel |
| case <-ctx.Done(): |
| return nil, ctx.Err() |
| case <-deadline.C: |
| + // A first wake onto a new build gives its waiters arrival + |
| + // 55 s (zero-downtime spec §8.2): enough for B's health |
| + // window, B's kill and the good build's boot. Only that kind |
| + // of wake; firstWake is set only by a planner (flag on). |
| + a.mu.Lock() |
| + fw := e.firstWake |
| + a.mu.Unlock() |
| + if fw && !extended { |
| + extended = true |
| + deadline.Reset(time.Until(arrival.Add(firstWakeBudget))) |
| + continue |
| + } |
| return nil, errWakeTimeout |
| } |
| @@ -1144,6 +1194,7 @@ func (a *Service) BeginSidecarRequest(ctx context.Context, rt registry.Route) er |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| log.Printf("activator: instance %s exited unexpectedly — recycling", e.host) |
| + e.unexpectedExit = true // the build's failure inside a first wake's observation (spec §6.4) |
| } |
| a.beginStopLocked(e) |
| a.mu.Unlock() |
| @@ -1230,6 +1281,9 @@ func (a *Service) wake(e *actEntry) { |
| defer cancel() |
| a.mu.Lock() |
| e.wakeCancel = cancel |
| + // Real time, not a.now: it anchors a first wake's real health deadline |
| + // (spawnPlanned, doClaimWake). |
| + e.wakeStart = time.Now() |
| if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| cancel() |
| } |
| @@ -1333,6 +1387,12 @@ func (a *Service) wake(e *actEntry) { |
| 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) |
| } |
| } |
| + // A first wake's observation is registered here, after e.gen is the |
| + // new generation and before e.ready closes, in this critical |
| + // section (zero-downtime spec §6.4), not once the generation is |
| + // already reachable. The fence and bounce checks below come after |
| + // it, so a stop they begin cancels it. |
| + a.armFirstWatchLocked(e, w) |
| if a.fences[e.host] != 0 || (a.gens && a.shuttingDown) { |
| a.beginStopLocked(e) |
| } else if e.bounceAfterWake { |
| @@ -1356,6 +1416,9 @@ func (a *Service) wake(e *actEntry) { |
| } |
| a.mu.Unlock() |
| close(e.ready) |
| + // A planned wake's outcome, for the scheduler: off a.mu, since OnGen |
| + // reaches its goodMu, which calls back into Current. |
| + a.afterWake(e, w, err) |
| if err != nil && e.worker { |
| // The failed wake's entry is gone, and with it any build its spawn |
| // published (generations.go). |
| @@ -1439,6 +1502,14 @@ type wokenInstance struct { |
| // `carlos logs` already carries the line. |
| spawnTook time.Duration |
| wakeTook time.Duration |
| + |
| + // planned is the generation a planned wake left serving; failed and |
| + // failReason name a first wake's build that failed; fellBack says the |
| + // good build replaced it (zero-downtime spec §8.2). |
| + planned Gen |
| + failed *release.StagedIdentity |
| + failReason string |
| + fellBack bool |
| } |
| // doWake runs the side effects of one wake. It takes the ENTRY, not the bare |
| @@ -1512,6 +1583,20 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| return w, fmt.Errorf("%s: refusing to wake: %w", host, errWakeUnproven) |
| } |
| + // 0.7. Which build this wake boots (zero-downtime spec §8.2). A hold |
| + // swap's pinned wake (e.pin) already knows; everything else asks the |
| + // planner, off a.mu, before the lease: a refused plan takes no lease. |
| + var plan WakePlan |
| + if a.Planner != nil && e.pin == nil { |
| + plan = a.Planner(ctx, rt) |
| + if plan.Refuse != "" { |
| + return w, fmt.Errorf("%s: refusing to wake: %s", host, plan.Refuse) |
| + } |
| + a.mu.Lock() |
| + e.plan, e.firstWake = plan, plan.FirstWake |
| + a.mu.Unlock() |
| + } |
| + |
| // 1. The lease first: everything after this line is done as the prefix's |
| // single writer. Contention is not retried here — on one box it means a |
| // previous incarnation hasn't released; tell the client to come back after |
| @@ -1638,11 +1723,23 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| ident release.StagedIdentity |
| pend pendingGeneration |
| ) |
| - // startWorker publishes the build before its exec (ruling R19 |
| - // item 3), and refuses the spawn when it cannot. |
| - proc, ident, pend, err = a.startWorker(ctx, rt, workerEnv(a.agentSocket(), host, token), nil, nil) |
| + env := workerEnv(a.agentSocket(), host, token) |
| + if plan.planned() { |
| + // A planned worker (zero-downtime spec §6.5): the named build, |
| + // behind planGate at each exec, with startWorker's publication |
| + // and the installation of the captured identity done inside |
| + // (PF8), and its readiness settled there too. |
| + proc, ident, err = a.spawnPlannedWorker(ctx, e, rt, plan, env, w, lease) |
| + if err == nil { |
| + w.planned = workerGenOf(ident) |
| + } |
| + } else { |
| + // startWorker publishes the build before its exec (ruling R19 |
| + // item 3), and refuses the spawn when it cannot. |
| + proc, ident, pend, err = a.startWorker(ctx, rt, env, nil, nil) |
| + } |
| spawnVer, w.spawnIdentity = ident.Version, ident |
| - if err == nil { |
| + if err == nil && !plan.planned() { |
| // Installed while the entry is still waking, in place of the |
| // pending build, in one critical section, so no snapshot sees |
| // neither: the child is running these bytes from this moment, |
| @@ -1663,6 +1760,14 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // goes back (Worker, genChildEmpty). |
| w.gen = Gen{N: nextGen(), Build: spawnVer, Worker: true} |
| } |
| + } else if plan.planned() { |
| + // A planned instance wake (zero-downtime spec §8.2): the plan's |
| + // staged build, every exec behind planGate, with the first-wake |
| + // budget and fallback. w.gen carries the booted staged identity, so |
| + // the legacy sidecar below binds to it (startSidecar). |
| + var g Gen |
| + proc, spawnVer, g, err = a.spawnPlanned(ctx, e, rt, plan, w, lease) |
| + w.gen, w.planned = g, g |
| } else if e.pin != nil { // swapsim:M3 |
| var g Gen |
| proc, spawnVer, g, err = a.startPinnedChecked(ctx, e, rt, *e.pin, lease) |
| @@ -1697,7 +1802,9 @@ func (a *Service) doWake(ctx context.Context, e *actEntry) (*wokenInstance, erro |
| // helper is what records the stop's OUTCOME on w.stopOutcome, and a wake |
| // that tore its own child down without recording it would leave the |
| // stopHistory blind to exactly the case it exists for. |
| - if e.worker { |
| + // A planned worker's readiness already ran inside spawnPlannedWorker |
| + // (plannedWorkerReady, spec §6.5). |
| + if e.worker && !plan.planned() { |
| if serr := a.awaitWorkerSettle(ctx, e, proc); serr != nil { |
| a.cleanupFailedWake(ctx, rt, w, lease) |
| return w, fmt.Errorf("start %s: %w", host, serr) |
| @@ -1773,6 +1880,13 @@ func (a *Service) beginStopLocked(e *actEntry) { |
| if e.settleCancel != nil && (e.proc == nil || !e.proc.Exited()) { |
| e.settleCancel() |
| } |
| + // The same for a planned first wake's observation (spec §6.4): an |
| + // intentional stop ends it and records nothing. A stop that began |
| + // because the child died on its own (unexpectedExit), or of a process |
| + // that has already exited, leaves it to read the exit as the build's. |
| + if e.firstWatch != nil && !e.unexpectedExit && (e.proc == nil || !e.proc.Exited()) { |
| + e.firstWatch() |
| + } |
| e.state = actStopping |
| e.stopped = make(chan struct{}) |
| go a.stop(e) |
| @@ -2240,6 +2354,7 @@ func (a *Service) WakeWorker(rt registry.Route) error { |
| log.Printf("activator: LEASE LOST for %s (%v) — standing down", e.host, repErr) |
| } else { |
| log.Printf("activator: worker %s exited — recycling", e.host) |
| + e.unexpectedExit = true // not an intentional stop (spec §6.4) |
| } |
| a.beginStopLocked(e) |
| a.mu.Unlock() |
| @@ -3304,6 +3419,9 @@ func (a *Service) sweepOnce() { |
| continue |
| } |
| if e.proc.Exited() { |
| + // Whatever follows, the child died on its own: a first wake's |
| + // observation reads it as the build's failure (spec §6.4). |
| + e.unexpectedExit = true |
| // A worker's exit is read by STATUS, which is the only thing that |
| // can tell "my work is done" from "I fell over": |
| // |
| @@ -3430,6 +3548,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 |
| + if cycle && a.OnDrift != nil { |
| + // CARLOS_SWAP=overlap: the scheduler owns every move onto a new |
| + // build, paced (spec §10). The sweep only tells it to look, on |
| + // its own goroutine and so off a.mu (the scheduler calls back |
| + // into Current), and carries on with this entry's sidecar |
| + // supervision below. No converge budget is spent. |
| + host := e.host |
| + go a.OnDrift(host) |
| + cycle = false |
| + } |
| if cycle && a.convergeBudget > 0 { |
| // An always-on worker converges as one MOVE, not as a cycle: the |
| // cycle's second half is "something wakes it back up", and a |
diff --git a/internal/activator/claimwake.go b/internal/activator/claimwake.go| index a2f1dece7..b0188ba60 100644 |
| --- a/internal/activator/claimwake.go |
| +++ b/internal/activator/claimwake.go |
| @@ -183,6 +183,24 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, |
| } |
| rt = *fresh |
| } |
| + // Claimed wakes run plan 3's pinned start (StartPolicy.Pin) and get the |
| + // same first-wake budget and fallback as an unclaimed wake (zero-downtime |
| + // spec §8.2): firstWake extends the waiters' deadline in begin, and the |
| + // policy below carries the health cap and the good build. The planner |
| + // is called off a.mu. On this base every claims-on launch refuses |
| + // before its exec hook (RoleLaunch.Start), so none of it runs a build. |
| + if a.Planner != nil && e.pin == nil { |
| + plan := a.Planner(ctx, rt) |
| + if plan.Refuse != "" { |
| + return nil, fmt.Errorf("%s: refusing to wake: %s", rt.Host, plan.Refuse) |
| + } |
| + if plan.planned() { |
| + pin := genOf(plan.Boot) |
| + a.mu.Lock() |
| + e.pin, e.plan, e.firstWake = &pin, plan, plan.FirstWake |
| + a.mu.Unlock() |
| + } |
| + } |
| policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin} |
| if e.pin != nil && e.pinCheck != nil { |
| // A claims-on hold's pinned wake: the attempt gate before anything |
| @@ -202,6 +220,39 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, |
| if a.gens && r.Generations { |
| policy.Gen = a.claimGenAt(rt, nextGen()) |
| } |
| + if p := e.plan; p.planned() && p.FirstWake && p.Mode == ModeOverlap && p.Fallback != nil { |
| + // The same budget as spawnPlanned: B's health until the waiters' |
| + // deadline minus 25 s, then the good build in its place. Fallback is |
| + // declared on StartPolicy but honoured only once the isolated launch |
| + // opens (startAdmitted does not run it on this base). |
| + fb := genOf(*p.Fallback) |
| + policy.Fallback = &fb |
| + policy.HealthBy = e.wakeStart.Add(firstWakeBudget - firstWakeKillReserve) |
| + } |
| + if p := e.plan; p.planned() { |
| + // At every claimed planned exec, B and its fallback alike, inside |
| + // the work after the admission and Restore (startAdmitted's |
| + // beforeStart): the ownership check (spec §7.4), and the planner's |
| + // selection gate for the pin about to exec (§8.1–8.4; P5 review |
| + // finding 3). A refusal there records nothing and starts nothing; |
| + // the wake fails and the next one is planned afresh. Both run off |
| + // a.mu (the gate takes it itself). |
| + if policy.Gate == nil { |
| + policy.Gate = func() error { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return a.ownedLocked(e, true) |
| + } |
| + } |
| + if p.Check != nil { |
| + policy.PinCheck = func(pin Gen) error { |
| + if err := p.Check(pin.Build); err != nil { |
| + return fmt.Errorf("%w: %v", errPlanStale, err) |
| + } |
| + return nil |
| + } |
| + } |
| + } |
| if err := r.Prerequisites(ctx, rt, observed, policy); err != nil { |
| return nil, err |
| } |
| @@ -216,9 +267,12 @@ func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, |
| if err := a.startAdmitted(ctx, e, admitted, policy); err != nil { |
| recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), 10*time.Second) |
| defer cancelRecovery() |
| - return nil, errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err)) |
| + // claimedPlanResult carries a planned wake's outcome to afterWake: a |
| + // non-nil result with an error carries no stopOutcome, so wake() |
| + // records the uncertain stop it records today. |
| + return a.claimedPlanResult(e, nil), errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err)) |
| } |
| - return e.claimResult, nil |
| + return a.claimedPlanResult(e, e.claimResult), nil |
| } |
| func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostclaim.Observed, policy startPolicy) (result error) { |
diff --git a/internal/activator/swap.go b/internal/activator/swap.go| index 5c01e0ae5..d4b302af5 100644 |
| --- a/internal/activator/swap.go |
| +++ b/internal/activator/swap.go |
| @@ -121,6 +121,13 @@ func classifyStep(lifecycle context.Context, err error) stepOutcome { |
| // or as a claims validation inside the work at the exec. Nothing of |
| // the build ran because of it (spec §7.4). |
| return stepStale |
| + case errors.Is(err, ErrBuildNotPublished): |
| + // A worker spawn refused because its build could not be published |
| + // before the exec (ruling R19 item 3): nothing of the build ran, so |
| + // it is abandonment, never a failure of the build. Classified here |
| + // and not in verdictOf, which main's flag-off respawn reads; every |
| + // caller of classifyStep is flag-on (swaps, holds, planned wakes). |
| + return stepStale |
| } |
| return stepFailed |
| } |
diff --git a/internal/activator/swaphold.go b/internal/activator/swaphold.go| index 46b94c060..d29880ea4 100644 |
| --- a/internal/activator/swaphold.go |
| +++ b/internal/activator/swaphold.go |
| @@ -280,7 +280,7 @@ func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt regist |
| if err := gate(); err != nil { |
| return nil, "", pin, err |
| } |
| - proc, ver, g, err := a.startPinnedAt(ctx, rt, pin, gate) |
| + proc, ver, g, err := a.startPinnedAt(ctx, rt, pin, gate, time.Time{}) |
| if proc != nil || genStartUnproven(err) { |
| a.markPinExec(e) |
| } |
| @@ -331,13 +331,20 @@ func (a *Service) wakeFailureEvent(e *actEntry, err error) (string, bool) { |
| // records the outcome, and a caller that reads a nil proc as "nothing ran" |
| // is right to. |
| func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) { |
| - return a.startPinnedAt(ctx, rt, pin, nil) |
| + return a.startPinnedAt(ctx, rt, pin, nil, time.Time{}) |
| } |
| // startPinnedAt is startPinned with gate, when set, asked immediately |
| // before StartStaged: a refusal there returns no process, with nothing of |
| // the build run. |
| -func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, gate func() error) (Proc, string, Gen, error) { |
| +// |
| +// healthBy, when set, caps the health check (zero-downtime spec §8.2): a |
| +// first wake's B is checked until the earlier of healthBy (the waiters' |
| +// deadline minus 25 s) and spawnHealthTimeout, which genHealthy applies |
| +// itself. StartStaged returns the process at once, so this is the only |
| +// health bound, and a B that misses it comes back with its process for the |
| +// caller to record and kill. Zero leaves the check as it was. |
| +func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, gate func() error, healthBy time.Time) (Proc, string, Gen, error) { |
| gb, ok := a.backend.(SwapBackend) |
| if !ok { |
| return nil, "", pin, errSwapNoGens |
| @@ -361,7 +368,13 @@ func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, |
| if g.N != pin.N || g.Build != pin.Build || g.StagedPath != pin.StagedPath { |
| return proc, "", pin, fmt.Errorf("%w: generation %d of %s from %q; the hold chose %s from %q", errPinMismatch, g.N, g.Build, g.StagedPath, pin.Build, pin.StagedPath) |
| } |
| - if err := a.genHealthy(ctx, g); err != nil { |
| + hctx := ctx |
| + if !healthBy.IsZero() { |
| + var cancel context.CancelFunc |
| + hctx, cancel = context.WithDeadline(ctx, healthBy) |
| + defer cancel() |
| + } |
| + if err := a.genHealthy(hctx, g); err != nil { |
| return proc, "", g, err |
| } |
| gb.GenReady(rt, g, proc) |
diff --git a/internal/activator/wakeplan.go b/internal/activator/wakeplan.go| new file mode 100644 |
| index 000000000..5b5fc943b |
| --- /dev/null |
| +++ b/internal/activator/wakeplan.go |
| @@ -0,0 +1,558 @@ |
| +// wakeplan.go is the activator's half of the rollout scheduler (zero-downtime |
| +// deploys spec §6.4, §8.2): which build a wake boots, the first-wake budget, |
| +// the fallback to the good build, and the 10 s observation. All of it is |
| +// reached through Service.Planner, which only CARLOS_SWAP=overlap sets. With |
| +// Planner nil (the flag off) nothing here runs. |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "errors" |
| + "fmt" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// WakePlan is the scheduler's answer to "which build does this wake boot" |
| +// (spec §8.2). The zero plan keeps today's wake. |
| +type WakePlan struct { |
| + // Boot is the artifact to exec, by its immutable staged path. |
| + Boot release.StagedIdentity |
| + Mode SwapMode |
| + // FirstWake: Boot has never served on this instance. The wake is a canary |
| + // for it, with the 55 s budget and the 10 s observation. |
| + FirstWake bool |
| + // Fallback is the good build, booted when a first wake in overlap mode |
| + // fails. Nil when there is none. |
| + Fallback *release.StagedIdentity |
| + // Refuse, when set, fails the wake before the lease with this reason. |
| + Refuse string |
| + // FailedUnder is the PromotedAt of the pointer Boot was promoted under, |
| + // read by the planner before the wake runs, so a failed first wake is |
| + // published with no blocking read (spec §8.1), as dispatch.FailedUnder |
| + // is for an attempt. Zero: the failure's own time stands in. |
| + FailedUnder time.Time |
| + // Check is the planner's selection gate (spec §7.4, §8.1–8.4): nil when |
| + // build may still boot. Planning runs before the lease and the restore, |
| + // so a sibling's failure recorded since can hold the build it picked. |
| + // The activator asks it at every planned exec, after the restore; the |
| + // good build always passes. Nil: nothing to re-check. It is called |
| + // without a.mu held. |
| + Check func(build string) error |
| +} |
| + |
| +func (p WakePlan) planned() bool { return p.Boot.StagedPath != "" } |
| + |
| +// errPlanStale: the planned build is no longer the one to boot. A |
| +// supersession (gateSuperseded), so it records nothing and falls back to |
| +// nothing (classifyStep's stepStale). |
| +var errPlanStale = fmt.Errorf("the planned build is no longer the one to boot: %w", errSwapSuperseded) |
| + |
| +// checkPlan is a planned wake's selection gate before its spawn (spec §7.4, |
| +// §8.1–8.4), after the lease and the restore: a refused plan is planned |
| +// again once, on the state already restored, and a second refusal abandons |
| +// the wake (errPlanStale). It returns the plan to boot. The exec itself asks |
| +// the returned plan's Check once more (planGate), with no second replan. |
| +// The planner and Check are called without a.mu held. |
| +func (a *Service) checkPlan(ctx context.Context, e *actEntry, rt registry.Route, plan WakePlan) (WakePlan, error) { |
| + if plan.Check == nil { |
| + return plan, nil |
| + } |
| + err := plan.Check(plan.Boot.Version) |
| + if err == nil { |
| + return plan, nil |
| + } |
| + if a.Planner == nil { |
| + return plan, fmt.Errorf("%w: %v", errPlanStale, err) |
| + } |
| + next := a.Planner(ctx, rt) |
| + if next.Refuse != "" || !next.planned() { |
| + return plan, fmt.Errorf("%w: %v; replanned: %s", errPlanStale, err, next.Refuse) |
| + } |
| + if next.Check != nil { |
| + if nerr := next.Check(next.Boot.Version); nerr != nil { |
| + return plan, fmt.Errorf("%w: %v", errPlanStale, nerr) |
| + } |
| + } |
| + a.mu.Lock() |
| + e.plan, e.firstWake = next, next.FirstWake |
| + a.mu.Unlock() |
| + return next, nil |
| +} |
| + |
| +// wakeOwned is the ownership check at a wake's exec (spec §7.4): the wake's |
| +// lifecycle has not ended, and plan 3's ownedLocked holds for a waking |
| +// entry (the entry is the host's, no fence, no shutdown; its lease is the |
| +// wake's own, which planGate checks beside it). |
| +func (a *Service) wakeOwned(ctx context.Context, e *actEntry) error { |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + return a.ownedLocked(e, true) |
| +} |
| + |
| +// planGate is the gate every planned exec passes, B's and its fallback's |
| +// alike, asked immediately before the exec (startPinnedAt's gate, or |
| +// startWorker's before-exec hook): the wake's locally held lease |
| +// (leaseLost; the entry has no replica yet, so ownedLocked cannot see it), |
| +// the ownership check (wakeOwned), and the plan's selection gate for the |
| +// build about to exec. Every refusal is abandonment, never a failure of the |
| +// build: the lease and ownership refusals are gateNoOwnership, the |
| +// selection gate's is errPlanStale (gateSuperseded), a cancelled wake is |
| +// the lifecycle's. Check runs after a.mu is released. |
| +func (a *Service) planGate(ctx context.Context, e *actEntry, lease Lease, plan WakePlan, build string) func() error { |
| + return func() error { |
| + if err := leaseLost(lease); err != nil { |
| + return err |
| + } |
| + if err := a.wakeOwned(ctx, e); err != nil { |
| + return err |
| + } |
| + if plan.Check != nil { |
| + if err := plan.Check(build); err != nil { |
| + return fmt.Errorf("%w: %v", errPlanStale, err) |
| + } |
| + } |
| + return nil |
| + } |
| +} |
| + |
| +func genOf(id release.StagedIdentity) Gen { |
| + return Gen{Build: id.Version, StagedPath: id.StagedPath, Digest: id.Digest, Demand: id.Demand} |
| +} |
| + |
| +// workerGenOf is a worker's generation as the scheduler sees it: its |
| +// captured staged artifact, marked Worker, with no number. The scheduler |
| +// compares a worker's generations by their artifact (rollout's |
| +// acknowledged), because a respawn on the same entry changes the build |
| +// without a number of its own. |
| +func workerGenOf(id release.StagedIdentity) Gen { |
| + g := genOf(id) |
| + g.Worker = true |
| + return g |
| +} |
| + |
| +// withIdentity is g, the generation a planned exec started, carrying the |
| +// staged identity it was booted from (PF10, spec §12 "Legacy sidecar |
| +// binding"): the build, its staged path, digest and demand. The legacy |
| +// sidecar co-spawned with it execs that staged path (startSidecar), and the |
| +// scheduler records it as good by its digest. |
| +func withIdentity(g Gen, id release.StagedIdentity) Gen { |
| + g.Build, g.StagedPath, g.Digest, g.Demand = id.Version, id.StagedPath, id.Digest, id.Demand |
| + return g |
| +} |
| + |
| +// GenEventKind names what happened to a planned wake. |
| +type GenEventKind string |
| + |
| +const ( |
| + GenHealthy GenEventKind = "healthy" // the planned build is serving |
| + GenSettled GenEventKind = "settled" // a first wake survived its 10 s |
| + GenFailed GenEventKind = "failed" // the planned build failed, nothing fell back |
| + GenFellBack GenEventKind = "fellback" // the planned build failed, the good build serves |
| + // GenCancelled: a first wake's observation ended without a verdict (an |
| + // intentional stop, spec §6.4). It records nothing about the build; it |
| + // only closes the observation GenHealthy opened. |
| + GenCancelled GenEventKind = "cancelled" |
| +) |
| + |
| +// GenEvent is a planned wake's outcome, for the scheduler. |
| +type GenEvent struct { |
| + Kind GenEventKind |
| + Route registry.Route |
| + Gen Gen // the generation serving (healthy, settled, fellback) |
| + Failed release.StagedIdentity // the build that failed (failed, fellback) |
| + Mode SwapMode |
| + FirstWake bool |
| + Reason string |
| + At time.Time |
| + // FailedUnder is the plan's FailedUnder (failed, fellback): the |
| + // failing pointer's PromotedAt, read before the wake ran. |
| + FailedUnder time.Time |
| +} |
| + |
| +// The first-wake budget (spec §8.2). Vars so tests can shrink them. |
| +var ( |
| + firstWakeBudget = 55 * time.Second |
| + firstWakeKillReserve = 25 * time.Second |
| + firstWakeKillBound = 2 * time.Second |
| + firstWakeSettle = 10 * time.Second |
| +) |
| + |
| +// genKiller (plan 3 Task 2, opset.go) is plan 2's generation kill |
| +// (ExecBackend.KillGen and its SplitBackend forward): cgroup.kill on |
| +// generation n's child cgroup, then wait until it is empty. This file uses |
| +// it and declares none. |
| + |
| +// Current is the generation an awake instance serves, for the scheduler: a |
| +// worker's is its captured spawn build (workerGenOf). A first wake under its |
| +// 10 s observation reads as not awake, so the scheduler leaves it alone |
| +// until the observation ends. |
| +func (a *Service) Current(host string) (Gen, bool) { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[host] |
| + if e == nil || e.state != actAwake || e.firstWatch != nil { |
| + return Gen{}, false |
| + } |
| + if e.worker { |
| + return workerGenOf(e.workerServing), true |
| + } |
| + return e.gen, true |
| +} |
| + |
| +// BounceFor is Bounce for a scheduled attempt: the event names the attempt's |
| +// cause rather than "carlos restart" (spec §9.4). |
| +func (a *Service) BounceFor(host string, cause Cause) bool { |
| + a.mu.Lock() |
| + defer a.mu.Unlock() |
| + e := a.entries[host] |
| + if e == nil || e.state != actAwake { |
| + return false |
| + } |
| + emitEvent(a.Logs, e.route, "info", bounceText(e.worker, cause)) |
| + a.beginStopLocked(e) |
| + return true |
| +} |
| + |
| +// bounceText is the one place BounceFor words its event. Draft wording: |
| +// plan 6 rewrites it through copy review. The flag-off literal in |
| +// bounceLocked is never edited. |
| +func bounceText(worker bool, cause Cause) string { |
| + noun := "instance" |
| + if worker { |
| + noun = "sidecar" |
| + } |
| + return noun + " restarting (" + string(cause) + ")" |
| +} |
| + |
| +// killFailed ends a failed first-wake build outright (spec §8.2 step 1), |
| +// bounded by firstWakeKillBound, and reports whether its end is proven. An |
| +// instance generation is killed with cgroup.kill on its own child and waited |
| +// empty (genKiller). Any error, errNoGenContainment included, is unproven: |
| +// without a generation child there is no kill that reaches what B forked, |
| +// and the good build must never start beside it. A worker is not a |
| +// generation child (plan 2 D4): its proven stop, bounded the same, stands in |
| +// for the kill, and THE termination proof (genTerminated on a Worker |
| +// generation) empties and checks its host cgroup. Never called with a.mu |
| +// held. |
| +func (a *Service) killFailed(rt registry.Route, g Gen, p Proc, worker bool) error { |
| + ctx, cancel := context.WithTimeout(context.Background(), firstWakeKillBound) |
| + defer cancel() |
| + if !worker { |
| + k, ok := a.backend.(genKiller) |
| + if !ok { |
| + return fmt.Errorf("%s: this backend cannot kill a generation", rt.Host) |
| + } |
| + if err := k.KillGen(ctx, rt, g.N); err != nil { |
| + return err |
| + } |
| + // The kill proved the end. The handle is still released through |
| + // its Stop, the only close a flat child's control listener gets |
| + // (backend_exec.go): left bound, it is the stale-socket trap |
| + // (#129). Its own bound, since the kill may have spent this one. |
| + sctx, scancel := context.WithTimeout(context.Background(), firstWakeKillBound) |
| + defer scancel() |
| + _ = p.Stop(sctx) |
| + return nil |
| + } |
| + if err := p.Stop(ctx); err != nil { |
| + return err |
| + } |
| + if !a.genTerminated(rt, Gen{N: g.N, Build: g.Build, Worker: true}, p) { |
| + return ErrStillRunning |
| + } |
| + return nil |
| +} |
| + |
| +// spawnPlanned is a planned wake's instance spawn (spec §8.2). The work |
| +// before the exec (lease, restore, seed) has already run once and is shared. |
| +// |
| +// Every exec, B's and its fallback's, goes through startPinnedAt with |
| +// planGate (PF7): the lease, the ownership and the plan's selection gate, |
| +// asked at the exec. A refusal there runs nothing and is abandonment. |
| +// |
| +// Execution evidence comes from the spawn's result, as startPinnedChecked |
| +// reads it: a returned process, or an unproven-start sentinel. Without it |
| +// nothing of the build ran and nothing is recorded. With it, only what |
| +// classifyStep calls the build's failure is recorded (a cancelled wake, a |
| +// stale attempt and a mismatched exec are not), and only for a first wake. |
| +// |
| +// For a first wake in overlap mode with a good build, B's health window |
| +// ends at the wake's start plus 55 s minus 25 s, a failed B is killed |
| +// outright, and the good build gets whatever time remains. When B's end is |
| +// not proven, nothing more starts. |
| +func (a *Service) spawnPlanned(ctx context.Context, e *actEntry, rt registry.Route, plan WakePlan, w *wokenInstance, lease Lease) (Proc, string, Gen, error) { |
| + plan, err := a.checkPlan(ctx, e, rt, plan) |
| + if err != nil { |
| + return nil, "", Gen{}, err |
| + } |
| + canFall := plan.FirstWake && plan.Fallback != nil && plan.Mode == ModeOverlap |
| + var healthBy time.Time |
| + if canFall { |
| + healthBy = e.wakeStart.Add(firstWakeBudget - firstWakeKillReserve) |
| + } |
| + boot := plan.Boot |
| + proc, ver, g, err := a.startPinnedAt(ctx, rt, genOf(boot), a.planGate(ctx, e, lease, plan, boot.Version), healthBy) |
| + if err == nil { |
| + return proc, ver, withIdentity(g, boot), nil |
| + } |
| + executed := proc != nil || genStartUnproven(err) |
| + if !executed || classifyStep(ctx, err) != stepFailed || errors.Is(err, errPinMismatch) { |
| + // Nothing of the build ran, or the wake's lifecycle ended it |
| + // (maintenance, Shutdown, a stop), or its gate refused, or the |
| + // backend ran something else: no failure of the build (spec §6.4, |
| + // §7.4). Nothing is recorded and nothing falls back; doWake's |
| + // cleanup stops a returned process and proves it gone. |
| + return proc, ver, g, err |
| + } |
| + if plan.FirstWake { |
| + w.failed, w.failReason = &boot, err.Error() |
| + } |
| + if !canFall || proc == nil { |
| + return proc, ver, g, err |
| + } |
| + if kerr := a.killFailed(rt, g, proc, false); kerr != nil { |
| + return proc, ver, g, fmt.Errorf("%w; its end was not proven, so the good build was not started: %v", err, kerr) |
| + } |
| + good := *plan.Fallback |
| + fproc, fver, fg, ferr := a.startPinnedAt(ctx, rt, genOf(good), a.planGate(ctx, e, lease, plan, good.Version), time.Time{}) |
| + if ferr != nil { |
| + if fproc == nil && !genStartUnproven(ferr) { |
| + // Nothing of the good build ran (its gate refused: a fence, a |
| + // shutdown, a lost lease or entry since B's exec). B's failure is |
| + // still recorded (w.failed); its process, proven gone, is what |
| + // doWake's cleanup proves once more. |
| + return proc, ver, g, fmt.Errorf("%w; the good build was not started: %v", err, ferr) |
| + } |
| + return fproc, fver, fg, fmt.Errorf("%w; the good build %s failed too: %w", err, good.Version, ferr) |
| + } |
| + w.fellBack = true |
| + return fproc, fver, withIdentity(fg, good), nil |
| +} |
| + |
| +// spawnPlannedWorker is spawnPlanned for a socketless child (spec §6.5). |
| +// Each spawn uses startWorker's contract exactly as doWake's worker branch |
| +// does (PF8): the build is published before its exec, planGate is asked at |
| +// the exec (after the publication), and on success the captured identity is |
| +// installed and the pending publication dropped in one critical section, |
| +// then published again. A backend without the staged capability cannot |
| +// exec a named build, so a planned worker is refused there before anything |
| +// runs (errSwapNoGens). plannedWorkerReady stands in for the health check, |
| +// and a failed release-cause first wake in overlap mode is replaced by the |
| +// good build once the failed child's host cgroup is proven empty. |
| +func (a *Service) spawnPlannedWorker(ctx context.Context, e *actEntry, rt registry.Route, plan WakePlan, env map[string]string, w *wokenInstance, lease Lease) (Proc, release.StagedIdentity, error) { |
| + if _, ok := a.backend.(stagedWorkerStarter); !ok { |
| + return nil, release.StagedIdentity{}, errSwapNoGens |
| + } |
| + plan, err := a.checkPlan(ctx, e, rt, plan) |
| + if err != nil { |
| + return nil, release.StagedIdentity{}, err |
| + } |
| + start := func(id release.StagedIdentity) (Proc, release.StagedIdentity, error) { |
| + p, got, pend, err := a.startWorker(ctx, rt, env, &id, a.planGate(ctx, e, lease, plan, id.Version)) |
| + if err != nil { |
| + return p, got, err |
| + } |
| + a.mu.Lock() |
| + e.workerServing = got |
| + a.dropPendingGenerationLocked(pend) |
| + a.mu.Unlock() |
| + _ = a.persistGenerations() |
| + if got.StagedPath != id.StagedPath { |
| + return p, got, fmt.Errorf("%w: worker build %s from %q; the plan chose %s from %q", errPinMismatch, got.Version, got.StagedPath, id.Version, id.StagedPath) |
| + } |
| + return p, got, plannedWorkerReady(ctx, p) |
| + } |
| + canFall := plan.FirstWake && plan.Fallback != nil && plan.Mode == ModeOverlap |
| + boot := plan.Boot |
| + proc, id, err := start(boot) |
| + if err == nil || proc == nil { |
| + // Ready, or nothing of the build ran: a refusal at the gate, an |
| + // unpublished build (ErrBuildNotPublished), a spawn error with no |
| + // child. |
| + return proc, id, err |
| + } |
| + if classifyStep(ctx, err) != stepFailed || errors.Is(err, errPinMismatch) { |
| + // A cancelled readiness window is the wake's lifecycle, not the |
| + // build (spec §6.4, §6.5): nothing recorded, nothing falls back. |
| + return proc, id, err |
| + } |
| + if plan.FirstWake { |
| + w.failed, w.failReason = &boot, err.Error() |
| + } |
| + if !canFall { |
| + return proc, id, err |
| + } |
| + if kerr := a.killFailed(rt, Gen{Build: id.Version}, proc, true); kerr != nil { |
| + return proc, id, fmt.Errorf("%w; its end was not proven, so the good build was not started: %v", err, kerr) |
| + } |
| + fproc, fid, ferr := start(*plan.Fallback) |
| + if ferr != nil { |
| + if fproc == nil { |
| + // Nothing of the good build ran: B's failure is still recorded, |
| + // and B's child, proven gone, is what doWake's cleanup proves. |
| + return proc, id, fmt.Errorf("%w; the good build was not started: %v", err, ferr) |
| + } |
| + return fproc, fid, fmt.Errorf("%w; the good build %s failed too: %w", err, plan.Fallback.Version, ferr) |
| + } |
| + w.fellBack = true |
| + return fproc, fid, nil |
| +} |
| + |
| +// plannedWorkerReady is spec §6.5's readiness for a socketless planned |
| +// wake, the rule ReplaceWorker applies too: the child is ready once it is |
| +// still running firstWakeSettle (10 s) after it started, and a clean exit |
| +// inside that window is a one-shot's success. Any other exit fails. An early |
| +// ready report does not shorten the window. The flag-off wake keeps |
| +// awaitWorkerSettle (ready report, sidecarSettle, every early exit a |
| +// failure), unchanged. |
| +func plannedWorkerReady(ctx context.Context, p Proc) error { |
| + switch observeSettle(ctx, p, firstWakeSettle) { |
| + case settleOK: |
| + return nil |
| + case settleExited: |
| + if !procCrashed(p) { |
| + return nil |
| + } |
| + return errWorkerExitedEarly |
| + } |
| + if err := ctx.Err(); err != nil { |
| + return err |
| + } |
| + return context.Canceled |
| +} |
| + |
| +// emitGen hands ev to OnGen. It is never called with a.mu held: OnGen |
| +// reaches the scheduler's served, publishServing and writeGood, which take |
| +// its goodMu and call back into Current. |
| +func (a *Service) emitGen(ev GenEvent) { |
| + if a.OnGen != nil { |
| + a.OnGen(ev) |
| + } |
| +} |
| + |
| +// afterWake reports a planned wake's outcome. wake() calls it after |
| +// e.ready has closed, without a.mu (emitGen's rule). |
| +func (a *Service) afterWake(e *actEntry, w *wokenInstance, err error) { |
| + plan := e.plan // written only by this wake's own goroutine |
| + if a.OnGen == nil || w == nil || !plan.planned() { |
| + return |
| + } |
| + now := time.Now() |
| + switch { |
| + case err != nil: |
| + if w.failed != nil { |
| + a.emitGen(GenEvent{Kind: GenFailed, Route: e.route, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder}) |
| + } |
| + case w.fellBack: |
| + a.emitGen(GenEvent{Kind: GenFellBack, Route: e.route, Gen: w.planned, Failed: *w.failed, Mode: plan.Mode, FirstWake: true, Reason: w.failReason, At: now, FailedUnder: plan.FailedUnder}) |
| + default: |
| + a.emitGen(GenEvent{Kind: GenHealthy, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: plan.FirstWake, At: now}) |
| + switch { |
| + case plan.FirstWake && e.worker: |
| + // A socketless first wake's readiness window (plannedWorkerReady, |
| + // spec §6.5) was its settle: it is settled now. A second window |
| + // would fail a build that was already ready, where §8.6 calls for |
| + // ordinary crash recovery. |
| + a.emitGen(GenEvent{Kind: GenSettled, Route: e.route, Gen: w.planned, Mode: plan.Mode, FirstWake: true, At: now}) |
| + case plan.FirstWake: |
| + a.watchFirstWake(e, w.planned, plan) |
| + } |
| + } |
| +} |
| + |
| +// armFirstWatchLocked registers a first wake's observation in the awake |
| +// transition, before e.ready closes (spec §6.4): Current hides the |
| +// generation from the instant it is reachable, and an exit or a stop |
| +// before the observer goroutine runs is classified by it, never lost. |
| +// wake() calls it after e.gen is set, and before the fence and bounce |
| +// checks there, so a stop they begin cancels the observation. The caller |
| +// holds a.mu. |
| +func (a *Service) armFirstWatchLocked(e *actEntry, w *wokenInstance) { |
| + if a.Planner == nil || a.OnGen == nil || e.worker || !e.plan.planned() || !e.plan.FirstWake || w.fellBack || w.failed != nil { |
| + return |
| + } |
| + ctx, cancel := context.WithCancel(context.Background()) |
| + e.firstWatch, e.firstWatchCtx = cancel, ctx |
| +} |
| + |
| +// watchFirstWake is a first wake's 10 s observation (spec §6.4), on plan 3's |
| +// observeSettle. The waiting requests do not sit through it. An intentional |
| +// stop cancels it (beginStopLocked) and records nothing; an unexpected exit |
| +// fails the build. |
| +func (a *Service) watchFirstWake(e *actEntry, g Gen, plan WakePlan) { |
| + cancelled := func(rt registry.Route) { |
| + // Every observation GenHealthy opened is closed by an event; this |
| + // one says "no verdict", so the scheduler stops showing the host |
| + // as settling and records nothing (spec §6.4, §9.1). |
| + a.emitGen(GenEvent{Kind: GenCancelled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()}) |
| + } |
| + // The observation was registered in the awake transition |
| + // (armFirstWatchLocked), so a stop that began meanwhile has already |
| + // cancelled it, or, for a crash, marked unexpectedExit: observeSettle and |
| + // the classification below decide either way. Nothing registered means |
| + // no first-wake observation for this wake. |
| + a.mu.Lock() |
| + ctx, rt, proc := e.firstWatchCtx, e.route, e.proc |
| + a.mu.Unlock() |
| + if ctx == nil { |
| + cancelled(rt) |
| + return |
| + } |
| + go func() { |
| + r := observeSettle(ctx, proc, firstWakeSettle) |
| + a.mu.Lock() |
| + cancel := e.firstWatch |
| + e.firstWatch, e.firstWatchCtx = nil, nil |
| + rt, unexpected := e.route, e.unexpectedExit |
| + a.mu.Unlock() |
| + if cancel != nil { |
| + cancel() |
| + } |
| + exited := r == settleExited || (r == settleCancelled && unexpected) |
| + switch { |
| + case r == settleOK: |
| + a.emitGen(GenEvent{Kind: GenSettled, Route: rt, Gen: g, Mode: plan.Mode, FirstWake: true, At: time.Now()}) |
| + case exited: |
| + a.emitGen(GenEvent{Kind: GenFailed, Route: rt, Failed: plan.Boot, Mode: plan.Mode, FirstWake: true, |
| + Reason: "exited within " + firstWakeSettle.String() + " of becoming healthy", At: time.Now(), FailedUnder: plan.FailedUnder}) |
| + default: // an intentional stop cancelled it |
| + cancelled(rt) |
| + } |
| + }() |
| +} |
| + |
| +// claimedPlanResult carries a claims-on planned wake's outcome to afterWake. |
| +// Plan 3's startAdmitted records the pinned build it saw fail |
| +// (e.pinFailed) and, once the isolated launch opens, starts the good build |
| +// in its place (StartPolicy.Fallback). w is the claim's result, nil when |
| +// the wake failed. The returned value is what doClaimWake hands back to |
| +// wake(). On this base every claims-on launch refuses before its exec hook, |
| +// so pinFailed stays nil and a planned claimed wake reports nothing. |
| +func (a *Service) claimedPlanResult(e *actEntry, w *wokenInstance) *wokenInstance { |
| + a.mu.Lock() |
| + failed, why, plan := e.pinFailed, e.pinFailReason, e.plan |
| + a.mu.Unlock() |
| + if !plan.planned() { |
| + return w |
| + } |
| + if w == nil { |
| + if failed == nil || !plan.FirstWake { |
| + return nil // nothing of the planned build ran: no failure of the build |
| + } |
| + return &wokenInstance{failed: &plan.Boot, failReason: why} |
| + } |
| + served := plan.Boot |
| + if failed != nil && plan.FirstWake && plan.Fallback != nil && w.spawnVersion == plan.Fallback.Version { |
| + served = *plan.Fallback |
| + w.failed, w.failReason, w.fellBack = &plan.Boot, why, true |
| + } |
| + w.planned = genOf(served) |
| + w.planned.N, w.planned.Socket, w.planned.Control = w.gen.N, w.gen.Socket, w.gen.Control |
| + return w |
| +} |
diff --git a/internal/activator/wakeplan_test.go b/internal/activator/wakeplan_test.go| new file mode 100644 |
| index 000000000..495514f98 |
| --- /dev/null |
| +++ b/internal/activator/wakeplan_test.go |
| @@ -0,0 +1,1097 @@ |
| +package activator |
| + |
| +import ( |
| + "context" |
| + "crypto/sha256" |
| + "encoding/hex" |
| + "errors" |
| + "fmt" |
| + "maps" |
| + "os" |
| + "path/filepath" |
| + "slices" |
| + "strconv" |
| + "strings" |
| + "sync" |
| + "sync/atomic" |
| + "testing" |
| + "time" |
| + |
| + "github.com/carlosframework/platform/internal/cgroup" |
| + "github.com/carlosframework/platform/internal/hostclaim" |
| + "github.com/carlosframework/platform/internal/hostwrite" |
| + "github.com/carlosframework/platform/internal/registry" |
| + "github.com/carlosframework/platform/internal/release" |
| +) |
| + |
| +// planGenBackend is plan 3's SwapBackend (startPinnedAt's staged spawn) plus |
| +// plan 2's KillGen over the package's fakeBackend. Its generations are |
| +// fakeProcs a test can crash. It also starts a legacy sidecar on a |
| +// generation's staged build (SidecarGenStarter), recording the generation |
| +// it was handed. |
| +type planGenBackend struct { |
| + *fakeBackend |
| + gmu sync.Mutex |
| + gens []Gen |
| + procs map[uint64]*fakeProc |
| + killed []uint64 |
| + killErr error |
| + gate chan struct{} |
| + sidecarGens []Gen |
| +} |
| + |
| +func newPlanGenBackend() *planGenBackend { |
| + return &planGenBackend{fakeBackend: &fakeBackend{}, procs: map[uint64]*fakeProc{}} |
| +} |
| + |
| +func (b *planGenBackend) StartStaged(_ context.Context, _ registry.Route, g Gen) (Gen, Proc, error) { |
| + b.gmu.Lock() |
| + gate := b.gate |
| + b.gmu.Unlock() |
| + if gate != nil { |
| + <-gate |
| + } |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + g.Socket = "/run/carlos/g/test." + strconv.FormatUint(g.N, 10) + ".sock" |
| + g.Control = g.Socket + ".control" |
| + p := &fakeProc{label: "gen:" + g.Build, rec: b.fakeBackend} |
| + b.gens = append(b.gens, g) |
| + b.procs[g.N] = p |
| + return g, p, nil |
| +} |
| + |
| +func (b *planGenBackend) GenTier(registry.Route, string) cgroup.Limit { return cgroup.TierDefault } |
| +func (b *planGenBackend) GenReady(registry.Route, Gen, Proc) {} |
| +func (b *planGenBackend) RaiseGenParent(context.Context, registry.Route, cgroup.Limit, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *planGenBackend) RestoreGenParent(context.Context, registry.Route, cgroup.Limit) error { |
| + return nil |
| +} |
| +func (b *planGenBackend) GenSwitched(registry.Route, Gen, Proc) error { return nil } |
| + |
| +func (b *planGenBackend) KillGen(_ context.Context, _ registry.Route, n uint64) error { |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + b.killed = append(b.killed, n) |
| + if b.killErr != nil { |
| + return b.killErr |
| + } |
| + if p := b.procs[n]; p != nil { |
| + p.exit() |
| + } |
| + return nil |
| +} |
| + |
| +func (b *planGenBackend) StartSidecarGen(_ context.Context, _ registry.Route, g Gen) (Proc, error) { |
| + b.gmu.Lock() |
| + b.sidecarGens = append(b.sidecarGens, g) |
| + b.gmu.Unlock() |
| + return &fakeProc{label: "sidecar", rec: b.fakeBackend}, nil |
| +} |
| + |
| +var ( |
| + _ SwapBackend = (*planGenBackend)(nil) |
| + _ genKiller = (*planGenBackend)(nil) |
| + _ SidecarGenStarter = (*planGenBackend)(nil) |
| +) |
| + |
| +func (b *planGenBackend) started() []Gen { |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + return append([]Gen(nil), b.gens...) |
| +} |
| + |
| +func (b *planGenBackend) kills() []uint64 { |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + return append([]uint64(nil), b.killed...) |
| +} |
| + |
| +func (b *planGenBackend) procOf(n uint64) *fakeProc { |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + return b.procs[n] |
| +} |
| + |
| +func (b *planGenBackend) sidecarsOn() []Gen { |
| + b.gmu.Lock() |
| + defer b.gmu.Unlock() |
| + return append([]Gen(nil), b.sidecarGens...) |
| +} |
| + |
| +// stagedID writes a staged build and returns its identity: plan 3's |
| +// startPinnedAt verifies the bytes before anything runs. |
| +func stagedID(t *testing.T, dir, ver string) release.StagedIdentity { |
| + t.Helper() |
| + body := []byte("build " + ver) |
| + path := filepath.Join(dir, ver) |
| + if err := os.WriteFile(path, body, 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + sum := sha256.Sum256(body) |
| + return release.StagedIdentity{Version: ver, StagedPath: path, Digest: hex.EncodeToString(sum[:]), Demand: "hold-" + ver} |
| +} |
| + |
| +type genLog struct { |
| + mu sync.Mutex |
| + evs []GenEvent |
| +} |
| + |
| +func (l *genLog) add(ev GenEvent) { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + l.evs = append(l.evs, ev) |
| +} |
| + |
| +func (l *genLog) kinds() []GenEventKind { |
| + l.mu.Lock() |
| + defer l.mu.Unlock() |
| + out := make([]GenEventKind, 0, len(l.evs)) |
| + for _, ev := range l.evs { |
| + out = append(out, ev.Kind) |
| + } |
| + return out |
| +} |
| + |
| +func (l *genLog) wait(t *testing.T, kind GenEventKind) GenEvent { |
| + t.Helper() |
| + deadline := time.Now().Add(5 * time.Second) |
| + for time.Now().Before(deadline) { |
| + l.mu.Lock() |
| + for _, ev := range l.evs { |
| + if ev.Kind == kind { |
| + l.mu.Unlock() |
| + return ev |
| + } |
| + } |
| + l.mu.Unlock() |
| + time.Sleep(5 * time.Millisecond) |
| + } |
| + t.Fatalf("no %s event; saw %v", kind, l.kinds()) |
| + return GenEvent{} |
| +} |
| + |
| +func planActivator(t *testing.T, plan WakePlan) (*Service, *planGenBackend, *genLog) { |
| + t.Helper() |
| + b := newPlanGenBackend() |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg) |
| + a.gens = true // plan 1's switch, as CARLOS_SWAP=overlap sets it before New |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + a.Planner = func(context.Context, registry.Route) WakePlan { return plan } |
| + a.genHealth = func(context.Context, Gen) error { return nil } |
| + return a, b, evs |
| +} |
| + |
| +func shrinkFirstWake(t *testing.T, budget, reserve, settle time.Duration) { |
| + t.Helper() |
| + b, r, s, k := firstWakeBudget, firstWakeKillReserve, firstWakeSettle, firstWakeKillBound |
| + firstWakeBudget, firstWakeKillReserve, firstWakeSettle = budget, reserve, settle |
| + t.Cleanup(func() { firstWakeBudget, firstWakeKillReserve, firstWakeSettle, firstWakeKillBound = b, r, s, k }) |
| +} |
| + |
| +func TestFirstWakeFallsBackToTheGoodBuildWithinTheBudget(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + var mu sync.Mutex |
| + deadlines := map[string]time.Time{} |
| + a.genHealth = func(ctx context.Context, g Gen) error { |
| + dl, _ := ctx.Deadline() |
| + mu.Lock() |
| + deadlines[g.Build] = dl |
| + mu.Unlock() |
| + if g.Build == "v2" { |
| + return errors.New("GET / answered 500") |
| + } |
| + return nil |
| + } |
| + start := time.Now() |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatalf("the waiting request was not served by the good build: %v", err) |
| + } |
| + done() |
| + gens := b.started() |
| + if len(gens) != 2 || gens[0].Build != "v2" || gens[1].Build != "v1" { |
| + t.Fatalf("spawned %+v, want v2 then the good v1", gens) |
| + } |
| + mu.Lock() |
| + window := deadlines["v2"].Sub(start) |
| + mu.Unlock() |
| + if window < 29*time.Second || window > 31*time.Second { |
| + t.Fatalf("B's health window ends %v after the wake began, want 30s: the 55s deadline minus 25s", window) |
| + } |
| + if k := b.kills(); len(k) != 1 || k[0] != gens[0].N { |
| + t.Fatalf("killed %v, want B's generation %d alone, by cgroup.kill", k, gens[0].N) |
| + } |
| + ev := evs.wait(t, GenFellBack) |
| + if ev.Failed.Version != "v2" || ev.Gen.Build != "v1" { |
| + t.Fatalf("fallback event = %+v", ev) |
| + } |
| + if g, ok := a.Current("one.example.com"); !ok || g.Build != "v1" { |
| + t.Fatalf("Current = %+v %v, want the good build serving", g, ok) |
| + } |
| +} |
| + |
| +// Spec §6.4: once a hold build has executed, nothing falls back. |
| +func TestAHoldFirstWakeNeverFallsBack(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeHold, FirstWake: true, Fallback: &good}) |
| + a.genHealth = func(context.Context, Gen) error { return errors.New("GET / answered 500") } |
| + if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil { |
| + t.Fatal("a failed hold first wake was served") |
| + } |
| + if gens := b.started(); len(gens) != 1 { |
| + t.Fatalf("spawned %+v, want B alone", gens) |
| + } |
| + if ev := evs.wait(t, GenFailed); ev.Mode != ModeHold || ev.Failed.Version != "v2" { |
| + t.Fatalf("failure event = %+v", ev) |
| + } |
| +} |
| + |
| +// Spec §7.4, §8.1–8.4 (P5 review finding 3): a planned wake re-checks its |
| +// build at the exec, after the lease and the restore. Here a sibling's |
| +// failure holds v2 between the planning and the exec: the wake is planned |
| +// again on the restored state and boots the good build, and v2 never |
| +// execs. Without the re-check v2 ran after its hold. |
| +func TestAPlannedWakeRechecksItsBuildAtTheExec(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{}) |
| + var held atomic.Bool |
| + check := func(build string) error { |
| + if build == "v2" && held.Load() { |
| + return errors.New("v2 failed on a sibling") |
| + } |
| + return nil |
| + } |
| + var plans atomic.Int32 |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + if plans.Add(1) == 1 { |
| + held.Store(true) // the sibling's failure lands after this plan is made |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: check} |
| + } |
| + return WakePlan{Boot: good, Mode: ModeOverlap, Check: check} |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatalf("the wake: %v", err) |
| + } |
| + done() |
| + if gens := b.started(); len(gens) != 1 || gens[0].Build != "v1" { |
| + t.Fatalf("spawned %+v, want the good v1 alone", gens) |
| + } |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a build that never ran was recorded", k) |
| + } |
| +} |
| + |
| +// PF7: the selection gate is asked again AT the exec, inside startPinnedAt's |
| +// gate (after the staged bytes are verified and the tier read), not only |
| +// before the spawn. A failure recorded in that moment holds the build: it |
| +// never execs, the wake is abandoned (errPlanStale), and nothing is |
| +// recorded against the build. |
| +func TestThePlanCheckAtTheExecRefusesWithoutRecording(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{}) |
| + var calls atomic.Int32 |
| + check := func(build string) error { |
| + if build == "v2" && calls.Add(1) >= 2 { |
| + return errors.New("v2 failed on a sibling") |
| + } |
| + return nil |
| + } |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: check} |
| + } |
| + _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if !errors.Is(err, errPlanStale) { |
| + t.Fatalf("err = %v, want errPlanStale from the exec's gate", err) |
| + } |
| + if got := classifyStep(nil, err); got != stepStale { |
| + t.Fatalf("classifyStep = %v, want stepStale: a refusal at the exec is abandonment", got) |
| + } |
| + if gens := b.started(); len(gens) != 0 { |
| + t.Fatalf("spawned %+v: a held build execed", gens) |
| + } |
| + if calls.Load() < 2 { |
| + t.Fatalf("Check asked %d times, want the second at the exec", calls.Load()) |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a refusal before the exec was recorded", k) |
| + } |
| +} |
| + |
| +// PF7: the wake's own locally held lease is part of every planned exec's |
| +// gate. A renewal that fails after the restore execs nothing: no |
| +// generation starts, nothing is recorded, nothing falls back. |
| +func TestAPlannedWakeWhoseLeaseIsLostExecsNothing(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{}) |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, |
| + Check: func(string) error { |
| + // Asked after the lease and the restore: lose the lease here. |
| + b.lastRep().setErr(errors.New("lease renewal failed")) |
| + return nil |
| + }} |
| + } |
| + _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if !errors.Is(err, errSwapLeaseLost) { |
| + t.Fatalf("err = %v, want the lost lease (errSwapLeaseLost)", err) |
| + } |
| + if gens := b.started(); len(gens) != 0 { |
| + t.Fatalf("spawned %+v on a lost lease", gens) |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a refusal before the exec was recorded", k) |
| + } |
| +} |
| + |
| +// Spec §7.4 (P5 review rule a): a wake's fallback passes the ownership |
| +// check at its own exec. Here the host is fenced while B fails its health: |
| +// the good build is not started. B's failure is still reported. Checked |
| +// only before B's exec, the fallback ran on a fenced host. |
| +func TestAFallbackOnAFencedHostIsNotStarted(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + const host = "one.example.com" |
| + a.genHealth = func(_ context.Context, g Gen) error { |
| + if g.Build != "v2" { |
| + return nil |
| + } |
| + a.mu.Lock() |
| + a.fences = map[string]uint64{host: 1} // maintenance fences the host meanwhile |
| + a.mu.Unlock() |
| + return errors.New("GET / answered 500") |
| + } |
| + if _, err := a.beginRequest(context.Background(), testRoute(host)); err == nil { |
| + t.Fatal("a fenced host's fallback served") |
| + } |
| + if gens := b.started(); len(gens) != 1 || gens[0].Build != "v2" { |
| + t.Fatalf("spawned %+v, want B alone", gens) |
| + } |
| + evs.wait(t, GenFailed) |
| +} |
| + |
| +func TestAnUnprovenKillStartsNoSecondGeneration(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + a.genHealth = func(_ context.Context, g Gen) error { return errors.New("GET / answered 500") } |
| + b.killErr = ErrStillRunning |
| + if _, err := a.beginRequest(context.Background(), testRoute("one.example.com")); err == nil { |
| + t.Fatal("served while B's end was unproven") |
| + } |
| + if gens := b.started(); len(gens) != 1 { |
| + t.Fatalf("spawned %+v: two generations against one database while B may still run", gens) |
| + } |
| + evs.wait(t, GenFailed) |
| +} |
| + |
| +func TestAnExitInsideTheFirstTenSecondsFailsTheBuild(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 400*time.Millisecond) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + evs.wait(t, GenHealthy) |
| + if _, ok := a.Current("one.example.com"); ok { |
| + t.Fatal("Current must hide a first wake under observation, so nothing swaps it meanwhile") |
| + } |
| + b.procOf(b.started()[0].N).crash() |
| + if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" { |
| + t.Fatalf("failure event = %+v", ev) |
| + } |
| +} |
| + |
| +func TestAnIntentionalStopInsideTheWindowRecordsNothing(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 400*time.Millisecond) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + evs.wait(t, GenHealthy) |
| + a.BounceFor("one.example.com", CauseConfig) |
| + time.Sleep(800 * time.Millisecond) |
| + // Spec §6.4: nothing about the build is recorded. Spec §9.1: the |
| + // observation is still closed, by GenCancelled, or the scheduler shows |
| + // the host settling for ever and the deploy watch never completes. |
| + if k := evs.kinds(); !slices.Equal(k, []GenEventKind{GenHealthy, GenCancelled}) { |
| + t.Fatalf("events %v: want healthy, then cancelled, and nothing recorded against the build", k) |
| + } |
| +} |
| + |
| +func TestAFirstWakeThatSurvivesSettles(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, _, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + if ev := evs.wait(t, GenSettled); ev.Gen.Build != "v2" { |
| + t.Fatalf("settled event = %+v", ev) |
| + } |
| + g, ok := a.Current("one.example.com") |
| + if !ok || g.Build != "v2" { |
| + t.Fatalf("Current = %+v %v after the settle", g, ok) |
| + } |
| + // PF10: the planned generation carries its staged identity, so the |
| + // scheduler can record it as good (it needs the digest) and the legacy |
| + // sidecar binds to it. |
| + if g.StagedPath != next.StagedPath || g.Digest != next.Digest || g.Demand != next.Demand || g.N == 0 { |
| + t.Fatalf("Current = %+v, want generation of %+v with its staged identity", g, next) |
| + } |
| +} |
| + |
| +func TestFirstWakeWaitersGetTheLongerBudget(t *testing.T) { |
| + old := wakeWaitBound |
| + wakeWaitBound = 100 * time.Millisecond |
| + t.Cleanup(func() { wakeWaitBound = old }) |
| + shrinkFirstWake(t, 600*time.Millisecond, 200*time.Millisecond, time.Millisecond) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + |
| + first, b, _ := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + b.gate = make(chan struct{}) |
| + time.AfterFunc(250*time.Millisecond, func() { close(b.gate) }) |
| + done, err := first.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatalf("a first wake's waiter gave up at the ordinary bound: %v", err) |
| + } |
| + done() |
| + |
| + plain, b2, _ := planActivator(t, WakePlan{Boot: good, Mode: ModeOverlap}) |
| + b2.gate = make(chan struct{}) |
| + time.AfterFunc(250*time.Millisecond, func() { close(b2.gate) }) |
| + if _, err := plain.beginRequest(context.Background(), testRoute("two.example.com")); !errors.Is(err, errWakeTimeout) { |
| + t.Fatalf("an ordinary wake's waiter = %v, want errWakeTimeout at wakeWaitBound", err) |
| + } |
| +} |
| + |
| +func TestARefusedPlanTakesNoLease(t *testing.T) { |
| + a, b, _ := planActivator(t, WakePlan{Refuse: "held down after build v2 failed in hold mode"}) |
| + _, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err == nil || !strings.Contains(err.Error(), "refusing to wake") { |
| + t.Fatalf("err = %v", err) |
| + } |
| + if acquires, _, _ := b.counts(); acquires != 0 { |
| + t.Fatal("a refused wake took the lease") |
| + } |
| +} |
| + |
| +// Claims on (plan 3 Task 10 runs the fallback inside startAdmitted): a |
| +// claimed first wake that fell back reports as one, a failed one names the |
| +// build that failed, and one where nothing of the build ran reports nothing. |
| +func TestAClaimedFirstWakeReportsLikeAnUnclaimedOne(t *testing.T) { |
| + good := release.StagedIdentity{Version: "v1", StagedPath: "/opt/v1", Digest: "d1"} |
| + next := release.StagedIdentity{Version: "v2", StagedPath: "/opt/v2", Digest: "d2"} |
| + a := &Service{} |
| + e := &actEntry{host: "one.example.com", plan: WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}, |
| + pinFailed: &Gen{Build: "v2"}, pinFailReason: "GET / answered 500"} |
| + w := a.claimedPlanResult(e, &wokenInstance{spawnVersion: "v1", gen: Gen{N: 9}}) |
| + if !w.fellBack || w.failed == nil || w.failed.Version != "v2" || w.planned.Build != "v1" || w.planned.N != 9 { |
| + t.Fatalf("fell back: %+v", w) |
| + } |
| + if w := a.claimedPlanResult(e, nil); w == nil || w.failed == nil || w.failed.Version != "v2" { |
| + t.Fatalf("failed: %+v", w) |
| + } |
| + e.pinFailed = nil |
| + if w := a.claimedPlanResult(e, nil); w != nil { |
| + t.Fatalf("nothing of the build ran, yet %+v", w) |
| + } |
| +} |
| + |
| +// PF17: every claims-on launch refuses on this base (RoleLaunch.Start |
| +// answers ErrClaimIsolationUnavailable before its exec hook). A planned |
| +// claimed wake carries the plan into the start policy (the pin, the good |
| +// build, the health cap, the ownership gate and the selection gate), meets |
| +// the closed launch, and records nothing: no failure event, no fallback. |
| +func TestAPlannedClaimedWakeRefusesAtTheLaunchRecordingNothing(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + const host = "x.example" |
| + obs := claimedObs(host) |
| + obs.Claim.Op.Exec = "box" |
| + obs.Claim.Holder.WriterBox = "box" |
| + admitted := obs |
| + claim := *obs.Claim |
| + holder := *claim.Holder |
| + holder.Session, holder.Generation = 2, "two" |
| + claim.Holder = &holder |
| + admitted.Claim = &claim |
| + work, _ := (hostwrite.Config{}).Select(context.Background(), host, "spawn") |
| + var ( |
| + pmu sync.Mutex |
| + policies []startPolicy |
| + reads atomic.Int32 |
| + ) |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := New(Config{Bucket: "test-bucket", Idle: time.Hour, HostWrites: hostwrite.Config{Enabled: true}, Claims: &ClaimRuntime{Box: "box", |
| + Read: func(context.Context, string) (hostclaim.Observed, error) { |
| + if reads.Add(1) == 1 { |
| + return obs, nil |
| + } |
| + return admitted, nil |
| + }, |
| + StopForeign: func(context.Context, string) error { return nil }, |
| + FailedAdmission: func(context.Context, hostclaim.Observed, error) error { return nil }, |
| + Prerequisites: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error { return nil }, |
| + Admit: func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) { |
| + return admitted, nil |
| + }, |
| + Prepare: func(_ context.Context, _ registry.Route, _ hostclaim.Observed, p startPolicy) (ClaimStart, error) { |
| + pmu.Lock() |
| + policies = append(policies, p) |
| + pmu.Unlock() |
| + return claimedStartFor(admitted, work, p.Pin.Build), nil |
| + }}}, &fakeBackend{}, reg) |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good, Check: func(string) error { return nil }} |
| + } |
| + _, err = a.beginRequest(context.Background(), registry.Route{Host: host, Kind: "instance", Hibernate: true}) |
| + if !errors.Is(err, ErrClaimIsolationUnavailable) { |
| + t.Fatalf("err = %v, want the closed launch", err) |
| + } |
| + pmu.Lock() |
| + defer pmu.Unlock() |
| + if len(policies) != 1 { |
| + t.Fatalf("prepared %d times, want once", len(policies)) |
| + } |
| + p := policies[0] |
| + if p.Pin == nil || p.Pin.Build != "v2" || p.Pin.StagedPath != next.StagedPath || p.Pin.Digest != next.Digest { |
| + t.Fatalf("policy pin = %+v, want the planned build", p.Pin) |
| + } |
| + if p.Fallback == nil || p.Fallback.Build != "v1" || p.HealthBy.IsZero() || p.Gate == nil || p.PinCheck == nil { |
| + t.Fatalf("policy = %+v: want the good build, the health cap, the ownership and selection gates", p) |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a launch refused before its exec recorded something", k) |
| + } |
| +} |
| + |
| +// execStagedBackend runs planned wakes through the REAL exec backend's |
| +// StartStaged (real processes, plan 3's real flat and foreign spawn paths) |
| +// over fakeBackend's lease and restore. Its KillGen stops the generation's |
| +// process, because a test fixture is not cgroup v2. |
| +type execStagedBackend struct { |
| + *fakeBackend |
| + eb *ExecBackend |
| + smu sync.Mutex |
| + procs map[uint64]Proc |
| +} |
| + |
| +func (b *execStagedBackend) StartStaged(ctx context.Context, rt registry.Route, g Gen) (Gen, Proc, error) { |
| + got, p, err := b.eb.StartStaged(ctx, rt, g) |
| + if p != nil { |
| + b.smu.Lock() |
| + b.procs[g.N] = p |
| + b.smu.Unlock() |
| + } |
| + return got, p, err |
| +} |
| +func (b *execStagedBackend) GenTier(rt registry.Route, build string) cgroup.Limit { |
| + return b.eb.GenTier(rt, build) |
| +} |
| +func (b *execStagedBackend) GenReady(rt registry.Route, g Gen, p Proc) { b.eb.GenReady(rt, g, p) } |
| +func (b *execStagedBackend) RaiseGenParent(ctx context.Context, rt registry.Route, inc, succ cgroup.Limit) error { |
| + return b.eb.RaiseGenParent(ctx, rt, inc, succ) |
| +} |
| +func (b *execStagedBackend) RestoreGenParent(ctx context.Context, rt registry.Route, tier cgroup.Limit) error { |
| + return b.eb.RestoreGenParent(ctx, rt, tier) |
| +} |
| +func (b *execStagedBackend) GenSwitched(rt registry.Route, g Gen, p Proc) error { |
| + return b.eb.GenSwitched(rt, g, p) |
| +} |
| +func (b *execStagedBackend) Layout(rt registry.Route) (Layout, string) { return b.eb.Layout(rt) } |
| +func (b *execStagedBackend) KillGen(ctx context.Context, _ registry.Route, n uint64) error { |
| + b.smu.Lock() |
| + p := b.procs[n] |
| + b.smu.Unlock() |
| + if p == nil { |
| + return nil |
| + } |
| + if err := p.Stop(ctx); err != nil { |
| + return err |
| + } |
| + if !p.Exited() { |
| + return ErrStillRunning |
| + } |
| + return nil |
| +} |
| + |
| +// stagedScript writes an executable staged build and returns its identity. |
| +func stagedScript(t *testing.T, dir, ver, body string) release.StagedIdentity { |
| + t.Helper() |
| + path := filepath.Join(dir, ver) |
| + if err := os.WriteFile(path, []byte(body), 0o755); err != nil { |
| + t.Fatal(err) |
| + } |
| + sum := sha256.Sum256([]byte(body)) |
| + return release.StagedIdentity{Version: ver, StagedPath: path, Digest: hex.EncodeToString(sum[:])} |
| +} |
| + |
| +// Spec §8.2 on the real exec backend: B never answers, so its failure must |
| +// be seen inside the wake's own budget, recorded, and the good build booted |
| +// in its place. StartStaged hands back the running process at once (plan 3 |
| +// Task 1), so startPinnedAt's health deadline (55 s − 25 s, shrunk here) is |
| +// the only one, and the kill reaches a real process. A StartStaged that |
| +// waited out the backend's own 30 s spawnHealthTimeout and returned a nil |
| +// handle made spawnPlanned read "nothing executed": no failure recorded, no |
| +// fallback, and the waiter timed out. The foreign row runs the same way on |
| +// its canonical socket (plan 1's reading): refusing it failed every planned |
| +// wake of a foreign row. |
| +func TestAPlannedFirstWakeOnTheRealExecBackendFallsBackInTime(t *testing.T) { |
| + for _, layout := range []string{"flat", "foreign"} { |
| + t.Run(layout, func(t *testing.T) { |
| + oldBound := wakeWaitBound |
| + wakeWaitBound = time.Second |
| + t.Cleanup(func() { wakeWaitBound = oldBound }) |
| + shrinkFirstWake(t, 4*time.Second, 2*time.Second, time.Millisecond) // B's health window: 2 s |
| + eb := testExecBackend(t) |
| + eb.RunDirOverride = t.TempDir() |
| + host := "real.example.com" |
| + stageLive(t, eb, host, "v0") |
| + rt := registry.Route{Host: host, Kind: "instance", Hibernate: true, DBPath: filepath.Join(eb.DataDir, host+".db"), |
| + Socket: filepath.Join(eb.RunDirOverride, host+".sock")} |
| + want := LayoutFlat |
| + if layout == "foreign" { |
| + rt.Socket, want = filepath.Join(t.TempDir(), "hand-set.sock"), LayoutForeign |
| + } |
| + if got, _ := eb.Layout(rt); got != want { |
| + t.Fatalf("fixture layout %v, want %v", got, want) |
| + } |
| + dir := t.TempDir() |
| + next := stagedScript(t, dir, "v2", "#!/bin/sh\nexec sleep 30\n") // never binds |
| + good := stagedScript(t, dir, "v1", "#!/bin/sh\nexec sleep 30\n") |
| + b := &execStagedBackend{fakeBackend: &fakeBackend{}, eb: eb, procs: map[uint64]Proc{}} |
| + reg, err := registry.Open(filepath.Join(t.TempDir(), "registry.db")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + a := New(Config{Bucket: "test-bucket", Idle: time.Hour}, b, reg) |
| + a.gens = true |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + // B gets the real check: GET / on its real socket, which nothing |
| + // binds, bounded only by the caller's deadline. The good build is |
| + // let through, because a shell script cannot answer HTTP. |
| + a.genHealth = func(ctx context.Context, g Gen) error { |
| + if g.Build == "v1" { |
| + return nil |
| + } |
| + return waitHealthyCtx(ctx, g.Socket, spawnHealthTimeout) |
| + } |
| + t.Cleanup(func() { a.Shutdown(context.Background()) }) |
| + start := time.Now() |
| + done, err := a.beginRequest(context.Background(), rt) |
| + if err != nil { |
| + t.Fatalf("the waiter was not served by the good build: %v", err) |
| + } |
| + done() |
| + if took := time.Since(start); took > 4*time.Second { |
| + t.Fatalf("served after %v, past the first-wake budget", took) |
| + } |
| + ev := evs.wait(t, GenFellBack) |
| + if ev.Failed.Version != "v2" || ev.Gen.Build != "v1" { |
| + t.Fatalf("fallback event = %+v", ev) |
| + } |
| + b.smu.Lock() |
| + procs := maps.Clone(b.procs) |
| + b.smu.Unlock() |
| + if len(procs) != 2 { |
| + t.Fatalf("%d generations spawned, want B and the good build", len(procs)) |
| + } |
| + for n, p := range procs { |
| + if n != ev.Gen.N && !p.Exited() { |
| + t.Fatalf("B's process (generation %d) outlived its kill", n) |
| + } |
| + } |
| + }) |
| + } |
| +} |
| + |
| +// PF10 (spec §12 "Legacy sidecar binding"): the co-spawned legacy sidecar |
| +// of a planned wake runs the build the wake left serving. After a first |
| +// wake fell back, that is the good build, by its staged path, never B's |
| +// and never the live link. |
| +func TestAFallenBackFirstWakeBindsItsSidecarToTheGoodBuild(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + b.sidecarStaged = true |
| + a.genHealth = func(_ context.Context, g Gen) error { |
| + if g.Build == "v2" { |
| + return errors.New("GET / answered 500") |
| + } |
| + return nil |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + ev := evs.wait(t, GenFellBack) |
| + sc := b.sidecarsOn() |
| + if len(sc) != 1 || sc[0].StagedPath != good.StagedPath || sc[0].Digest != good.Digest || sc[0].N != ev.Gen.N { |
| + t.Fatalf("sidecar started on %+v, want the good build's generation %d at %s", sc, ev.Gen.N, good.StagedPath) |
| + } |
| + if n := b.sidecarStartCount(); n != 0 { |
| + t.Fatalf("%d live-link sidecar starts, want none", n) |
| + } |
| + if ev.Gen.StagedPath != good.StagedPath || ev.Gen.Digest != good.Digest || ev.Gen.Demand != good.Demand { |
| + t.Fatalf("fallback event's generation %+v lacks the good build's staged identity", ev.Gen) |
| + } |
| +} |
| + |
| +// Spec §6.5: a socketless planned wake is ready once it is still running |
| +// 10 s after it started, and a clean exit inside that window is a |
| +// one-shot's success: the same rule ReplaceWorker applies. The flag-off |
| +// awaitWorkerSettle fails every early exit, exit 0 included, so a one-shot's |
| +// first wake was recorded as a failed build (and could halt its cohort). |
| +func TestAPlannedOneShotWorkerThatExitsCleanlyIsReady(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + b.workerDeadOnStart = true // the child exits at once, cleanly |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatalf("WakeWorker: %v (a one-shot's clean exit is its success)", err) |
| + } |
| + ev := evs.wait(t, GenSettled) |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a clean exit was recorded against the build", k) |
| + } |
| + if calls := b.calls(); len(calls) != 1 { |
| + t.Fatalf("spawns %v: a clean exit fell back to the good build", calls) |
| + } |
| + // A worker's generation is compared by its staged artifact (the |
| + // scheduler's acknowledged), so the event names it, and it is a worker's. |
| + if !ev.Gen.Worker || ev.Gen.StagedPath != next.StagedPath || ev.Gen.Digest != next.Digest { |
| + t.Fatalf("settled event's generation = %+v, want the worker's v2 artifact", ev.Gen) |
| + } |
| +} |
| + |
| +// plannedHostKill is plan 4's stagedFake plus plan 3's KillHost, failing |
| +// while bad: a worker host cgroup that still holds a process. (The name |
| +// stagedHostKill is worker_retained_pin_test.go's.) |
| +type plannedHostKill struct { |
| + *stagedFake |
| + bad atomic.Bool |
| +} |
| + |
| +func (s *plannedHostKill) KillHost(context.Context, registry.Route) error { |
| + if s.bad.Load() { |
| + return errors.New("the host cgroup still holds a process") |
| + } |
| + return nil |
| +} |
| + |
| +// Spec §7.4, §7.9: a failed socketless first wake falls back to the good |
| +// build only once the failed worker's host cgroup is proven empty (plan 3's |
| +// genTerminated on a Worker generation). Its leader's exit alone let the |
| +// good build start beside a descendant of the failed one. |
| +func TestAFailedWorkerFirstWakeFallsBackOnlyOnceItsCgroupIsEmpty(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + hk := &plannedHostKill{stagedFake: b} |
| + hk.bad.Store(true) |
| + a.backend = hk |
| + a.OnGen = (&genLog{}).add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + go func() { |
| + waitFor(t, "the v2 child", func() bool { return len(b.calls()) == 1 }) |
| + time.Sleep(50 * time.Millisecond) |
| + b.lastWorker().crash() |
| + }() |
| + _ = a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) |
| + if calls := b.calls(); len(calls) != 1 { |
| + t.Fatalf("spawns %v: the good build started beside the failed worker's unproven host cgroup", calls) |
| + } |
| +} |
| + |
| +// The same, with the host cgroup proven empty: the good build replaces the |
| +// failed worker, through startWorker's gate, and the wake reports the |
| +// fallback with the good build's artifact. |
| +func TestAFailedWorkerFirstWakeFallsBackToTheGoodBuild(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 300*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + hk := &plannedHostKill{stagedFake: b} |
| + a.backend = hk |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + go func() { |
| + waitFor(t, "the v2 child", func() bool { return len(b.calls()) == 1 }) |
| + time.Sleep(50 * time.Millisecond) |
| + b.lastWorker().crash() |
| + }() |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatalf("WakeWorker: %v, want the good build serving", err) |
| + } |
| + calls := b.calls() |
| + if len(calls) != 2 || calls[0] == nil || calls[0].Version != "v2" || calls[1] == nil || calls[1].Version != "v1" { |
| + t.Fatalf("spawns %v, want v2 then the good v1, each by name", calls) |
| + } |
| + ev := evs.wait(t, GenFellBack) |
| + if ev.Failed.Version != "v2" || ev.Gen.StagedPath != good.StagedPath || !ev.Gen.Worker { |
| + t.Fatalf("fallback event = %+v", ev) |
| + } |
| + if got := a.entryServing("bot.x.internal"); got.StagedPath != good.StagedPath { |
| + t.Fatalf("serving %+v, want the good build", got) |
| + } |
| +} |
| + |
| +// PF9: a planned worker spawn refused because its build could not be |
| +// published first (ErrBuildNotPublished) ran nothing of the build. It is |
| +// abandonment: no failure recorded, nothing falls back. |
| +func TestAPlannedWorkerRefusedForPublicationRecordsNothing(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + blocker := filepath.Join(t.TempDir(), "not-a-dir") |
| + if err := os.WriteFile(blocker, nil, 0o644); err != nil { |
| + t.Fatal(err) |
| + } |
| + a.GenerationsFile = filepath.Join(blocker, "generations.json") // unwritable |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")) |
| + if !errors.Is(err, ErrBuildNotPublished) { |
| + t.Fatalf("WakeWorker = %v, want ErrBuildNotPublished", err) |
| + } |
| + if n := b.workerStarts(); n != 0 { |
| + t.Fatalf("%d worker children started for an unpublished build", n) |
| + } |
| + if calls := b.calls(); len(calls) != 1 { |
| + t.Fatalf("spawn attempts %v: a refusal before the exec fell back", calls) |
| + } |
| + time.Sleep(50 * time.Millisecond) |
| + if k := evs.kinds(); len(k) != 0 { |
| + t.Fatalf("events %v: a refusal before the exec was recorded", k) |
| + } |
| +} |
| + |
| +// PF9: the one publication sentinel is abandonment wherever it surfaces. |
| +func TestErrBuildNotPublishedIsAbandonment(t *testing.T) { |
| + err := fmt.Errorf("start bot.x.internal: %w", fmt.Errorf("%w: disk full", ErrBuildNotPublished)) |
| + if got := classifyStep(nil, err); got != stepStale { |
| + t.Fatalf("classifyStep = %v, want stepStale", got) |
| + } |
| + if got := verdictOf(err); got != gateProceed { |
| + t.Fatalf("verdictOf = %v: main's flag-off respawn reads verdictOf, which must stay as it was", got) |
| + } |
| +} |
| + |
| +// Spec §6.4, §7.4: a wake cancelled by its lifecycle (here Shutdown, as |
| +// maintenance or a stop would) is no failure of the build: no failure |
| +// event, no fallback, so no halted cohort and no hold-down. spawnPlanned |
| +// that marked every error with a process as a failure recorded one. |
| +func TestACancelledFirstWakeRecordsNothing(t *testing.T) { |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + entered := make(chan struct{}) |
| + var once sync.Once |
| + a.genHealth = func(ctx context.Context, _ Gen) error { |
| + once.Do(func() { close(entered) }) |
| + <-ctx.Done() |
| + return ctx.Err() |
| + } |
| + go a.beginRequest(context.Background(), testRoute("one.example.com")) //nolint:errcheck |
| + <-entered |
| + a.Shutdown(context.Background()) |
| + time.Sleep(100 * time.Millisecond) |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a cancelled wake was recorded against the build", k) |
| + } |
| + if gens := b.started(); len(gens) != 1 { |
| + t.Fatalf("spawned %+v: a cancelled wake fell back", gens) |
| + } |
| +} |
| + |
| +// Spec §6.4, §8.1: a first wake's observation is registered in the awake |
| +// transition, before the generation is reachable. Here the build crashes, |
| +// and the sweep starts stopping it, in the moment between the wake becoming |
| +// ready and its observer running. Registered late (in afterWake), Current |
| +// exposed the build as stable meanwhile, and the observer, finding the |
| +// entry stopping, reported GenCancelled: a real failure of the build lost. |
| +func TestAFirstWakeIsObservedFromTheMomentItIsReady(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 2*time.Second) |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a, b, evs := planActivator(t, WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good}) |
| + var visible atomic.Bool |
| + a.OnGen = func(ev GenEvent) { |
| + if ev.Kind == GenHealthy { |
| + if _, ok := a.Current("one.example.com"); ok { |
| + visible.Store(true) |
| + } |
| + b.procOf(ev.Gen.N).crash() |
| + a.sweepOnce() // the sweep sees the exit before the observer runs |
| + } |
| + evs.add(ev) |
| + } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + if ev := evs.wait(t, GenFailed); ev.Failed.Version != "v2" { |
| + t.Fatalf("failure event = %+v", ev) |
| + } |
| + if slices.Contains(evs.kinds(), GenCancelled) { |
| + t.Fatalf("events %v: a crash was reported as a cancelled observation", evs.kinds()) |
| + } |
| + if visible.Load() { |
| + t.Fatal("Current exposed the first wake as stable before its observation was registered") |
| + } |
| +} |
| + |
| +// Spec §6.5, §8.6: a socketless first wake is settled once its readiness |
| +// window is over. A crash after it is ordinary crash recovery, not a failed |
| +// release. A second, HTTP-style observation after the readiness window |
| +// recorded a worker that crashed at 10 s plus a moment as a failed build. |
| +func TestAPlannedWorkerFirstWakeSettlesAtItsReadiness(t *testing.T) { |
| + shrinkFirstWake(t, 55*time.Second, 25*time.Second, 200*time.Millisecond) |
| + a, b, _, _, _ := stagedActivator(t) |
| + a.gens = true |
| + evs := &genLog{} |
| + a.OnGen = evs.add |
| + dir := t.TempDir() |
| + good, next := stagedID(t, dir, "v1"), stagedID(t, dir, "v2") |
| + a.Planner = func(context.Context, registry.Route) WakePlan { |
| + return WakePlan{Boot: next, Mode: ModeOverlap, FirstWake: true, Fallback: &good} |
| + } |
| + if err := a.WakeWorker(alwaysOnWorkerRoute("bot.x.internal")); err != nil { |
| + t.Fatal(err) |
| + } |
| + evs.wait(t, GenSettled) |
| + b.lastWorker().crash() // after its readiness window |
| + time.Sleep(400 * time.Millisecond) |
| + if k := evs.kinds(); slices.Contains(k, GenFailed) || slices.Contains(k, GenFellBack) { |
| + t.Fatalf("events %v: a crash after readiness was recorded against the build", k) |
| + } |
| +} |
| + |
| +// The flag off: no Planner, today's wake, no events. |
| +func TestWithoutAPlannerTheWakeIsTodays(t *testing.T) { |
| + a, b, _, _ := testActivator(t, time.Hour) |
| + var fired bool |
| + a.OnGen = func(GenEvent) { fired = true } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + if b.startCount() != 1 || fired { |
| + t.Fatalf("starts %d, events %v: want today's Start and nothing else", b.startCount(), fired) |
| + } |
| +} |
| + |
| +func TestDriftFeedsTheSchedulerInsteadOfStopping(t *testing.T) { |
| + a, b, clk, _ := testActivator(t, time.Hour) |
| + a.liveVer = func(string) (string, error) { return "v2", nil } |
| + b.execVerFn = func() string { return "v1" } |
| + var drifted []string |
| + var mu sync.Mutex |
| + a.OnDrift = func(h string) { mu.Lock(); drifted = append(drifted, h); mu.Unlock() } |
| + done, err := a.beginRequest(context.Background(), testRoute("one.example.com")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + done() |
| + a.sweepOnce() |
| + clk.advance(ConvergeGrace + time.Second) |
| + a.sweepOnce() |
| + time.Sleep(50 * time.Millisecond) |
| + if b.lastProc().isStopped() { |
| + t.Fatal("the sweep stopped a drifted instance; with the flag on the scheduler owns the move (spec §10)") |
| + } |
| + mu.Lock() |
| + defer mu.Unlock() |
| + if len(drifted) == 0 || drifted[0] != "one.example.com" { |
| + t.Fatalf("OnDrift saw %v", drifted) |
| + } |
| +} |
| + |
| +type causeSink struct { |
| + mu sync.Mutex |
| + msgs []string |
| +} |
| + |
| +func (s *causeSink) App(string, string, string, string) {} |
| +func (s *causeSink) Event(_, _, _, _, msg string) { |
| + s.mu.Lock() |
| + defer s.mu.Unlock() |
| + s.msgs = append(s.msgs, msg) |
| +} |
| + |
| +// Spec §9.4: the bounce event names the attempt's cause. |
| +func TestBounceForNamesTheCause(t *testing.T) { |
| + a := newTestServiceAwake(t, "one.example.com") |
| + sink := &causeSink{} |
| + a.Logs = sink |
| + if !a.BounceFor("one.example.com", CauseConfig) { |
| + t.Fatal("BounceFor stopped nothing") |
| + } |
| + sink.mu.Lock() |
| + defer sink.mu.Unlock() |
| + if len(sink.msgs) != 1 || sink.msgs[0] != "instance restarting (config)" { |
| + t.Fatalf("events = %v", sink.msgs) |
| + } |
| +} |
diff --git a/internal/activator/zdd5_surface_test.go b/internal/activator/zdd5_surface_test.go| index c086635bb..357464ccf 100644 |
| --- a/internal/activator/zdd5_surface_test.go |
| +++ b/internal/activator/zdd5_surface_test.go |
| @@ -42,10 +42,10 @@ var ( |
| _ func(*Service, *actEntry, Proc, string, *release.StagedIdentity) = (*Service).respawnWorkerLocked // plan 4 |
| // Added by the controller: later tasks build on these merged names. |
| - _ func(*Service, context.Context, registry.Route, Gen, func() error) (Proc, string, Gen, error) = (*Service).startPinnedAt // swaphold.go |
| - _ func(*Service, context.Context, *actEntry, registry.Route, Gen, Lease) (Proc, string, Gen, error) = (*Service).startPinnedChecked // swaphold.go |
| - _ error = ErrBuildNotPublished // workerbuild.go |
| - _ func(*Service) error = (*Service).persistGenerations // generations_file.go |
| + _ func(*Service, context.Context, registry.Route, Gen, func() error, time.Time) (Proc, string, Gen, error) = (*Service).startPinnedAt // swaphold.go; plan 5 Task 9 added the health deadline |
| + _ func(*Service, context.Context, *actEntry, registry.Route, Gen, Lease) (Proc, string, Gen, error) = (*Service).startPinnedChecked // swaphold.go |
| + _ error = ErrBuildNotPublished // workerbuild.go |
| + _ func(*Service) error = (*Service).persistGenerations // generations_file.go |
| ) |
| // Fields plan 5 reads or sets. |