| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "os" |
| 9 | "os/exec" |
| 10 | "sync" |
| 11 | "syscall" |
| 12 | "time" |
| 13 | |
| 14 | "github.com/carlosframework/platform/internal/cgroup" |
| 15 | "github.com/carlosframework/platform/internal/hostbox" |
| 16 | "github.com/carlosframework/platform/internal/hostclaim" |
| 17 | "github.com/carlosframework/platform/internal/hostwrite" |
| 18 | "github.com/carlosframework/platform/internal/moves" |
| 19 | "github.com/carlosframework/platform/internal/provision" |
| 20 | "github.com/carlosframework/platform/internal/registry" |
| 21 | "github.com/carlosframework/platform/internal/release" |
| 22 | ) |
| 23 | |
| 24 | var ErrAttemptOver = moves.ErrAttemptOver |
| 25 | |
| 26 | type StartPolicy struct { |
| 27 | // RecoveryFence is the owning move cut generation, supplied by its adapter. |
| 28 | RecoveryFence uint64 |
| 29 | MaxStarts int |
| 30 | Window time.Duration |
| 31 | Deadline time.Time |
| 32 | Probe bool |
| 33 | Prepared *moves.InstallManifest |
| 34 | Inputs *provision.MoveInputs |
| 35 | // Pin is a hold swap's chosen build (zero-downtime spec §7.8). Prepare |
| 36 | // must build a start that execs exactly Pin.StagedPath after checking |
| 37 | // Pin.Digest. startAdmitted refuses a ClaimStart whose Version is not |
| 38 | // Pin.Build. |
| 39 | Pin *Gen |
| 40 | // Gen is the generation the activator chose for this start: its number |
| 41 | // and the sockets the provider must bind (zero-downtime plan 3, Task |
| 42 | // 10). Zero is today's start on the canonical socket. startAdmitted |
| 43 | // refuses a ClaimStart that bound anything else (genMatches). |
| 44 | Gen Gen |
| 45 | // Fallback is a first wake's good build (zero-downtime spec §8.2), set |
| 46 | // only beside Pin, by plan 5's planner. It is honoured once the isolated |
| 47 | // launch opens; on this base the launch refuses before anything runs |
| 48 | // (stepStale), so there is nothing to fall back from. Any fallback arm |
| 49 | // added later must record its build first (recordBuild, settings spec |
| 50 | // §4.2; ruling R-EXEC-CLAIMFALLBACK). |
| 51 | Fallback *Gen |
| 52 | // HealthBy, when set, caps the pinned start's Ready window: a first |
| 53 | // wake's health check ends at the earlier of Window and HealthBy (spec |
| 54 | // §8.2: the waiters' deadline minus 25 s). |
| 55 | HealthBy time.Time |
| 56 | // Gate, when set, is the attempt gate (taking a.mu itself): a claims-on |
| 57 | // hold's pinCheck, or a planned wake's ownership check (plan 5). |
| 58 | // startAdmitted runs it in the launch's before hook, immediately before |
| 59 | // every exec, after the admission and Restore (spec §6.3, §7.4). |
| 60 | Gate func() error |
| 61 | // PinCheck, when set, is the planner's selection gate (plan 5's |
| 62 | // WakePlan.Check) for the pin about to exec. startAdmitted runs it |
| 63 | // beside Gate, at every exec. |
| 64 | PinCheck func(pin Gen) error |
| 65 | // Held is a held instance's target (settings spec §4.3 rule 3), "" when |
| 66 | // the wake is not held. startAdmitted's before hook refuses a start |
| 67 | // whose hashed build is neither the target nor a planned wake's |
| 68 | // recovery boot (heldIdentityOK), before the record and the exec. The |
| 69 | // activator sets it; providers ignore it. |
| 70 | Held string |
| 71 | } |
| 72 | type startPolicy = StartPolicy |
| 73 | |
| 74 | // ClaimRuntime is assembled by the host owner before Service.Start. |
| 75 | // Admit reserves the first serving attempt and binds its resulting cold session. |
| 76 | // Prepared probes instead consume the moves owner's single-use admission. |
| 77 | type ClaimRuntime struct { |
| 78 | // ResumedAdmission verifies a journalled recovery that persisted prelaunch |
| 79 | // under the existing session during this admission call. |
| 80 | ResumedAdmission func(context.Context, hostclaim.Observed, hostclaim.Observed) error |
| 81 | RecoveryAttemptsReady func() bool |
| 82 | FailedAdmission func(context.Context, hostclaim.Observed, error) error |
| 83 | Box string |
| 84 | // Bucket and LeaseRegion are the claim actor's bucket and the region its |
| 85 | // leases sign for: a claimed wake leases Holder.Location there, and the |
| 86 | // instance's build-state.json sits beside that lease (settings spec |
| 87 | // §4.2, ruling R-CLAIMPREFIX). |
| 88 | Bucket, LeaseRegion string |
| 89 | Read func(context.Context, string) (hostclaim.Observed, error) |
| 90 | StopForeign func(context.Context, string) error |
| 91 | Prerequisites func(context.Context, registry.Route, hostclaim.Observed, startPolicy) error |
| 92 | Admit func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (hostclaim.Observed, error) |
| 93 | Prepare func(context.Context, registry.Route, hostclaim.Observed, startPolicy) (ClaimStart, error) |
| 94 | // Pins reports that Prepare honours StartPolicy.Pin. A hold swap is |
| 95 | // refused on a claims-on box whose runtime does not, before anything |
| 96 | // stops. |
| 97 | Pins bool |
| 98 | // Generations reports that Prepare honours StartPolicy.Gen (binds its |
| 99 | // sockets and reports them in ClaimStart.Gen). Without it a claims-on |
| 100 | // box keeps one generation per entry, on the canonical socket, and |
| 101 | // every swap is refused before anything is raised, stopped or spawned |
| 102 | // (errClaimsNoGenerations). |
| 103 | Generations bool |
| 104 | } |
| 105 | |
| 106 | // ClaimStart is one admitted session. Its closures retain the admission tuple. |
| 107 | // Only Restore may purge data. Process retries reuse it and the live replication. |
| 108 | type ClaimStart struct { |
| 109 | Live context.Context |
| 110 | PrepareLaunch func(context.Context) (*RoleLaunch, error) |
| 111 | CurrentAttempt func() hostclaim.WakeRetryAttempt |
| 112 | RoleLauncher *RoleLaunch |
| 113 | // OpenProbe journals the first app-start time before execution and returns |
| 114 | // that same time on retries, including after controller restart. Its |
| 115 | // provider binds the moves attempt and admission, never resets Opened. |
| 116 | OpenProbe func(context.Context, time.Time) (time.Time, error) |
| 117 | Work hostwrite.Work |
| 118 | Identity hostwrite.Identity |
| 119 | Manager *cgroup.Manager |
| 120 | Host string |
| 121 | Limit cgroup.Limit |
| 122 | Restore func(context.Context) error |
| 123 | Materialize func(context.Context) error |
| 124 | RetryClaimWrite func(hostclaim.Observed, hostclaim.Observed) error |
| 125 | Build func() *exec.Cmd |
| 126 | Ready func(context.Context, *exec.Cmd) error |
| 127 | // Reserve(0) validates the already reserved first attempt. Later calls CAS. |
| 128 | Reserve func(context.Context, int) error |
| 129 | Failed func(context.Context, int) error |
| 130 | Healthy func(context.Context, int, *exec.Cmd) error |
| 131 | Replicate func(context.Context, string) (Rep, error) |
| 132 | Publish func(context.Context, hostclaim.Observed) (hostclaim.Observed, error) |
| 133 | // Retain owns uncertain process or replica cleanup until it can be proved. |
| 134 | Retain func(Proc, Rep, error) |
| 135 | Version string |
| 136 | // Staged is the file the launch hashed, from one resolution of the |
| 137 | // host's live link (SnapshotClaimLaunch): the version it is staged |
| 138 | // under, its path and its digest (ExecutableSHA256). Version is a |
| 139 | // separate read of the link, so a repoint between the two can make them |
| 140 | // disagree; startAdmitted records Staged.Version, and refuses a start |
| 141 | // whose Staged.Version is not Version (settings spec §4.2, ruling |
| 142 | // R-EXEC-CLAIMVERSION). |
| 143 | Staged release.StagedIdentity |
| 144 | // Gen is what the provider bound. startAdmitted refuses a start whose |
| 145 | // number, socket or control socket is not StartPolicy.Gen's. |
| 146 | Gen Gen |
| 147 | } |
| 148 | |
| 149 | // genMatches reports whether a provider bound exactly the generation the |
| 150 | // activator chose: its number, its socket and its control socket. A handle |
| 151 | // that names a socket nobody serves would route requests nowhere. |
| 152 | func genMatches(want, got Gen) bool { |
| 153 | return want.N == got.N && want.Socket == got.Socket && want.Control == got.Control |
| 154 | } |
| 155 | |
| 156 | func (s ClaimStart) valid() bool { |
| 157 | return s.Work != nil && s.Identity.Config != "" && s.Identity.Release != "" && s.Host != "" && |
| 158 | s.Restore != nil && s.Materialize != nil && s.Build != nil && s.Ready != nil && s.Reserve != nil && |
| 159 | (s.RoleLauncher != nil || s.PrepareLaunch != nil) && s.CurrentAttempt != nil && s.Failed != nil && s.Healthy != nil && s.Replicate != nil && s.Publish != nil && s.Retain != nil |
| 160 | } |
| 161 | |
| 162 | func (a *Service) doClaimWake(ctx context.Context, e *actEntry) (*wokenInstance, error) { |
| 163 | r := a.cfg.Claims |
| 164 | if r == nil || r.Box == "" || r.Read == nil || r.StopForeign == nil { |
| 165 | return nil, hostwrite.ErrUnavailable |
| 166 | } |
| 167 | // A failed wake whose generation is not proven stopped still holds the |
| 168 | // host, and its process may still run against the database |
| 169 | // (zero-downtime spec §7.4). As doWake does, no new wake starts until |
| 170 | // the sweep proves it, and a stray legacy sidecar refuses it the same |
| 171 | // way. |
| 172 | a.mu.Lock() |
| 173 | _, unproven := a.unprovenWakes[e.host] |
| 174 | unproven = unproven || len(a.straySidecars[e.host]) > 0 |
| 175 | a.mu.Unlock() |
| 176 | if unproven { |
| 177 | return nil, fmt.Errorf("%s: refusing to wake: %w", e.host, errWakeUnproven) |
| 178 | } |
| 179 | observed, err := r.Read(ctx, e.host) |
| 180 | if err != nil { |
| 181 | return nil, err |
| 182 | } |
| 183 | // K6 precedes registry reads, lease acquisition, materialization and policy checks. |
| 184 | executor := "" |
| 185 | if observed.Claim != nil { |
| 186 | executor = observed.Claim.Op.Exec |
| 187 | if observed.Claim.Op.Kind == hostclaim.KindServing && observed.Claim.Holder != nil { |
| 188 | executor = observed.Claim.Holder.WriterBox |
| 189 | } |
| 190 | } |
| 191 | if observed.Claim == nil || executor != r.Box { |
| 192 | stopErr := r.StopForeign(ctx, e.host) |
| 193 | return nil, probeRefused(e, errors.Join(hostbox.ErrChanged, stopErr)) |
| 194 | } |
| 195 | if e.route.WakeIncarnation != "" && (observed.Claim == nil || observed.Claim.Holder == nil || observed.Claim.Holder.Incarnation != e.route.WakeIncarnation) { |
| 196 | return nil, probeRefused(e, hostbox.ErrChanged) |
| 197 | } |
| 198 | // A probe (ProbeWake, moves Plan 3 Task 5a) is admitted only at its |
| 199 | // op's probing phase (K6's probe row) and runs under its own policy. |
| 200 | probe := e.probe |
| 201 | if err := hostbox.WakePhase(observed.Claim, r.Box, probe != nil || e.recoveryPolicy != nil); err != nil { |
| 202 | return nil, probeRefused(e, err) |
| 203 | } |
| 204 | if probe != nil && observed.Claim.Op.ID != probe.op { |
| 205 | return nil, probeRefused(e, errProbeOp) |
| 206 | } |
| 207 | if e.recoveryPolicy != nil && (!e.recoveryPolicy.Probe || !recoveryProbeKind(observed.Claim.Op.Kind)) { |
| 208 | return nil, hostbox.ErrChanged |
| 209 | } |
| 210 | if r.Prerequisites == nil || r.Admit == nil || r.Prepare == nil || r.FailedAdmission == nil { |
| 211 | return nil, hostwrite.ErrUnavailable |
| 212 | } |
| 213 | rt := e.route |
| 214 | if a.Routes != nil { |
| 215 | fresh, err := a.Routes.Get(e.host) |
| 216 | if err != nil { |
| 217 | return nil, err |
| 218 | } |
| 219 | if fresh == nil { |
| 220 | return nil, hostbox.ErrChanged |
| 221 | } |
| 222 | rt = *fresh |
| 223 | } |
| 224 | // The held wake check (settings spec §4.3 rule 3), as doWake's step |
| 225 | // 0.65, before planning and before the admission. Never for a hold's |
| 226 | // pinned wake (e.pinCheck: the swap chose its build; planned claimed |
| 227 | // wakes set e.pin too, below, so e.pin alone does not say "hold"), and |
| 228 | // never for a moves probe, which runs the landing's prepared build under |
| 229 | // its own policy. The launcher execs the claim inputs' file, so a |
| 230 | // retained copy cannot be injected here (direct false): a target not |
| 231 | // installed is asked of root and waited for, and the row re-read. The |
| 232 | // launch's before hook refuses a hashed build that is not the target |
| 233 | // (StartPolicy.Held). |
| 234 | var held string |
| 235 | if e.pinCheck == nil && probe == nil && e.recoveryPolicy == nil { |
| 236 | held, _, err = a.wakeCheck(ctx, rt.Host, false) |
| 237 | if err != nil { |
| 238 | return nil, fmt.Errorf("%s: %w", rt.Host, err) |
| 239 | } |
| 240 | if held != "" && a.Routes != nil { |
| 241 | fresh, err := a.Routes.Get(e.host) |
| 242 | if err != nil { |
| 243 | return nil, err |
| 244 | } |
| 245 | if fresh == nil { |
| 246 | return nil, hostbox.ErrChanged |
| 247 | } |
| 248 | rt = *fresh |
| 249 | } |
| 250 | } |
| 251 | // Claimed wakes run plan 3's pinned start (StartPolicy.Pin) and get the |
| 252 | // same first-wake budget and fallback as an unclaimed wake (zero-downtime |
| 253 | // spec §8.2): firstWake extends the waiters' deadline in begin, and the |
| 254 | // policy below carries the health cap and the good build. The planner |
| 255 | // is called off a.mu. On this base every claims-on launch refuses |
| 256 | // before its exec hook (RoleLaunch.Start), so none of it runs a build. |
| 257 | // A moves probe (ProbeWake) is never planned: it runs the landing's |
| 258 | // prepared build under its own policy, and a plan's outcome must not be |
| 259 | // recorded against a build it did not boot. |
| 260 | if a.Planner != nil && e.pin == nil && probe == nil { |
| 261 | // As doWake: an earlier planned wake's outcome is delivered before |
| 262 | // this one plans (final review F3), and a wait that ends first |
| 263 | // refuses the wake (final review wave 2, N1). |
| 264 | if err := a.awaitOutcomes(ctx, rt.Host); err != nil { |
| 265 | return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err) |
| 266 | } |
| 267 | plan := a.Planner(ctx, rt) |
| 268 | if plan.Refuse != "" { |
| 269 | return nil, fmt.Errorf("%s: refusing to wake: %s", rt.Host, plan.Refuse) |
| 270 | } |
| 271 | if plan.planned() { |
| 272 | pin := genOf(plan.Boot) |
| 273 | a.mu.Lock() |
| 274 | e.pin, e.plan, e.firstWake = &pin, plan, plan.FirstWake |
| 275 | a.mu.Unlock() |
| 276 | } |
| 277 | } |
| 278 | // A pinned or planned build is published before the admission, as |
| 279 | // doWake does before its lease (spec §4.2, §8.5): claimed wakes never |
| 280 | // reach doWake's publication. A failed write refuses the wake with |
| 281 | // nothing admitted and no build blamed. |
| 282 | a.mu.Lock() |
| 283 | staged := e.pin != nil || e.plan.planned() |
| 284 | a.mu.Unlock() |
| 285 | if staged { |
| 286 | if err := a.publishBeforeExec(); err != nil { |
| 287 | return nil, fmt.Errorf("%s: refusing to wake: %w", rt.Host, err) |
| 288 | } |
| 289 | } |
| 290 | policy := startPolicy{MaxStarts: 5, Window: 30 * time.Second, Pin: e.pin, Held: held} |
| 291 | if e.recoveryPolicy != nil { |
| 292 | policy = *e.recoveryPolicy |
| 293 | } else if probe != nil { |
| 294 | // K1: the row is still held by the probe's hold op. |
| 295 | if !probeHeld(rt, probe.holdOp) { |
| 296 | return nil, probeRefused(e, errProbeRow) |
| 297 | } |
| 298 | policy = probe.policy |
| 299 | } else if e.pin != nil && e.pinCheck != nil { |
| 300 | // A claims-on hold's pinned wake: the attempt gate before anything |
| 301 | // is admitted, and again in the launch's before hook at every exec, |
| 302 | // after the admission and Restore (spec §6.3, §7.4), as |
| 303 | // startPinnedChecked does for a claims-off one. |
| 304 | gate := func() error { |
| 305 | a.mu.Lock() |
| 306 | defer a.mu.Unlock() |
| 307 | return e.pinCheck() |
| 308 | } |
| 309 | if err := gate(); err != nil { |
| 310 | return nil, err |
| 311 | } |
| 312 | policy.Gate = gate |
| 313 | } |
| 314 | if a.gens && r.Generations { |
| 315 | policy.Gen = a.claimGenAt(rt, nextGen()) |
| 316 | } |
| 317 | if p := e.plan; p.planned() && p.FirstWake && p.Mode == ModeOverlap && p.Fallback != nil { |
| 318 | // The same budget as spawnPlanned: B's health until the waiters' |
| 319 | // deadline minus 25 s, then the good build in its place. Fallback is |
| 320 | // declared on StartPolicy but honoured only once the isolated launch |
| 321 | // opens (startAdmitted does not run it on this base). |
| 322 | fb := genOf(*p.Fallback) |
| 323 | policy.Fallback = &fb |
| 324 | policy.HealthBy = e.wakeStart.Add(firstWakeBudget - firstWakeKillReserve) |
| 325 | } |
| 326 | if p := e.plan; p.planned() { |
| 327 | // At every claimed planned exec, B and its fallback alike, inside |
| 328 | // the work after the admission and Restore (startAdmitted's |
| 329 | // beforeStart): the ownership check (spec §7.4), and the planner's |
| 330 | // selection gate for the pin about to exec (§8.1–8.4; P5 review |
| 331 | // finding 3). A refusal there records nothing and starts nothing; |
| 332 | // the wake fails and the next one is planned afresh. Both run off |
| 333 | // a.mu (the gate takes it itself). |
| 334 | if policy.Gate == nil { |
| 335 | policy.Gate = func() error { |
| 336 | a.mu.Lock() |
| 337 | defer a.mu.Unlock() |
| 338 | return a.ownedLocked(e, true) |
| 339 | } |
| 340 | } |
| 341 | if p.Check != nil { |
| 342 | policy.PinCheck = func(pin Gen) error { |
| 343 | if err := p.Check(pin.Build); err != nil { |
| 344 | return fmt.Errorf("%w: %v", errPlanStale, err) |
| 345 | } |
| 346 | return nil |
| 347 | } |
| 348 | } |
| 349 | } |
| 350 | if err := r.Prerequisites(ctx, rt, observed, policy); err != nil { |
| 351 | return nil, err |
| 352 | } |
| 353 | admitted, err := r.Admit(ctx, rt, observed, policy) |
| 354 | if err != nil { |
| 355 | return nil, err |
| 356 | } |
| 357 | if admitted.Claim == nil || admitted.Claim.Holder == nil || observed.Claim.Holder == nil || admitted.Claim.Holder.Incarnation != observed.Claim.Holder.Incarnation { |
| 358 | return nil, hostbox.ErrChanged |
| 359 | } |
| 360 | if admitted.Claim.Holder.Session <= observed.Claim.Holder.Session || admitted.Claim.Holder.Generation == observed.Claim.Holder.Generation { |
| 361 | if !recoveryProbeKind(observed.Claim.Op.Kind) || !hostclaim.SameWriterIdentity(observed, admitted) || r.ResumedAdmission == nil { |
| 362 | return nil, hostbox.ErrChanged |
| 363 | } |
| 364 | if err := r.ResumedAdmission(ctx, observed, admitted); err != nil { |
| 365 | return nil, err |
| 366 | } |
| 367 | } |
| 368 | e.route = rt |
| 369 | if e.recoveryPolicy != nil { |
| 370 | e.probeAdmitted = admitted |
| 371 | } |
| 372 | if err := a.startAdmitted(ctx, e, admitted, policy); err != nil { |
| 373 | recoveryCtx, cancelRecovery := context.WithTimeout(context.Background(), 10*time.Second) |
| 374 | defer cancelRecovery() |
| 375 | err = errors.Join(err, r.FailedAdmission(recoveryCtx, admitted, err)) |
| 376 | if probe != nil { |
| 377 | err = errors.Join(ErrProbeFailed, err) |
| 378 | } |
| 379 | // claimedPlanResult carries a planned wake's outcome to afterWake: a |
| 380 | // non-nil result with an error carries no stopOutcome, so wake() |
| 381 | // records the uncertain stop it records today. A probe is never |
| 382 | // planned, so its result stays nil as on main. |
| 383 | return a.claimedPlanResult(e, nil), err |
| 384 | } |
| 385 | if probe != nil { |
| 386 | probe.admitted = &admitted |
| 387 | } |
| 388 | return a.claimedPlanResult(e, e.claimResult), nil |
| 389 | } |
| 390 | |
| 391 | // claimStartBackoff is startAdmitted's wait before its second start; the |
| 392 | // wait doubles before each start after that (1, 2, 4, 8 s). A var so tests |
| 393 | // can shrink it. |
| 394 | var claimStartBackoff = time.Second |
| 395 | |
| 396 | func (a *Service) startAdmitted(ctx context.Context, e *actEntry, admitted hostclaim.Observed, policy startPolicy) (result error) { |
| 397 | r := a.cfg.Claims |
| 398 | if !a.cfg.HostWrites.Enabled || r == nil || r.Read == nil || r.Prepare == nil || e == nil || admitted.Host != e.host || policy.MaxStarts < 1 || policy.MaxStarts > 5 || policy.Window <= 0 || policy.Window > 30*time.Second { |
| 399 | return hostwrite.ErrUnavailable |
| 400 | } |
| 401 | budget := &probeWindow{policy: policy, now: time.Now} |
| 402 | defer func() { result = budget.result(result) }() |
| 403 | if !policy.Deadline.IsZero() { |
| 404 | bounded, cancel := context.WithDeadline(ctx, policy.Deadline) |
| 405 | defer cancel() |
| 406 | ctx = bounded |
| 407 | } |
| 408 | if err := ctx.Err(); err != nil { |
| 409 | return err |
| 410 | } |
| 411 | if err := hostbox.WakePhase(admitted.Claim, r.Box, policy.Probe); err != nil { |
| 412 | return err |
| 413 | } |
| 414 | fresh, err := r.Read(ctx, e.host) |
| 415 | if err != nil { |
| 416 | return err |
| 417 | } |
| 418 | if fresh.ETag != admitted.ETag && !(recoveryProbeKind(admitted.Claim.Op.Kind) && hostclaim.SameWriterIdentity(fresh, admitted)) { |
| 419 | return hostbox.ErrChanged |
| 420 | } |
| 421 | s, err := r.Prepare(ctx, e.route, admitted, policy) |
| 422 | if err != nil { |
| 423 | return err |
| 424 | } |
| 425 | if !s.valid() || s.Host != e.host || (policy.Probe && s.OpenProbe == nil) { |
| 426 | return hostwrite.ErrUnavailable |
| 427 | } |
| 428 | if policy.Pin != nil && s.Version != policy.Pin.Build { |
| 429 | // A hold swap's wake must run the build its attempt chose |
| 430 | // (zero-downtime spec §7.8). A start prepared for anything else is |
| 431 | // refused before Restore touches the database. |
| 432 | return hostwrite.ErrUnavailable |
| 433 | } |
| 434 | if err := a.claimedBuild(e.host, s); err != nil { |
| 435 | // The bytes the snapshot hashed are not the build the start names: |
| 436 | // nothing could record the build that would run. Refused before |
| 437 | // Restore touches the database, and no failed attempt of the build. |
| 438 | return err |
| 439 | } |
| 440 | if policy.Gen.N != 0 && !genMatches(policy.Gen, s.Gen) { |
| 441 | // The provider bound something other than the generation the |
| 442 | // activator chose: its handle would name a socket nobody serves. |
| 443 | // Refused before Restore touches the database. |
| 444 | return hostwrite.ErrUnavailable |
| 445 | } |
| 446 | if s.Live != nil { |
| 447 | if err := s.Live.Err(); err != nil { |
| 448 | return err |
| 449 | } |
| 450 | joined, cancel := context.WithCancel(ctx) |
| 451 | defer cancel() |
| 452 | stop := context.AfterFunc(s.Live, cancel) |
| 453 | defer stop() |
| 454 | ctx = joined |
| 455 | } |
| 456 | prefix, err := hostbox.GenerationPrefix(*admitted.Claim.Holder) |
| 457 | if err != nil { |
| 458 | return err |
| 459 | } |
| 460 | if err := s.Restore(ctx); err != nil { |
| 461 | return err |
| 462 | } |
| 463 | var rep Rep |
| 464 | var retained bool |
| 465 | defer func() { |
| 466 | if result == nil || rep == nil || retained { |
| 467 | return |
| 468 | } |
| 469 | cleanup, cancel := context.WithTimeout(context.Background(), 10*time.Second) |
| 470 | defer cancel() |
| 471 | if err := rep.Close(cleanup); err != nil { |
| 472 | s.Retain(nil, rep, err) |
| 473 | result = errors.Join(result, err) |
| 474 | } |
| 475 | }() |
| 476 | var last error |
| 477 | for attempt := 0; attempt < policy.MaxStarts; attempt++ { |
| 478 | if err := budget.result(ctx.Err()); err != nil { |
| 479 | return err |
| 480 | } |
| 481 | if err := s.Reserve(ctx, attempt); err != nil { |
| 482 | return err |
| 483 | } |
| 484 | if err := s.Materialize(ctx); err != nil { |
| 485 | return errors.Join(err, s.Failed(ctx, attempt)) |
| 486 | } |
| 487 | if s.RoleLauncher == nil { |
| 488 | s.RoleLauncher, err = s.PrepareLaunch(ctx) |
| 489 | if err != nil { |
| 490 | return errors.Join(err, s.Failed(ctx, attempt)) |
| 491 | } |
| 492 | if s.RoleLauncher == nil { |
| 493 | return errors.Join(hostwrite.ErrUnavailable, s.Failed(ctx, attempt)) |
| 494 | } |
| 495 | } |
| 496 | // This hook runs after strict preparation and immediately before Start. |
| 497 | // Restore and materialization consume only the separate cut deadline. |
| 498 | reached := false // the exec hook ran: an error before it is the admission's (claimAdmission) |
| 499 | unrecorded := false |
| 500 | beforeStart := func() error { |
| 501 | reached = true |
| 502 | if policy.Gate != nil { |
| 503 | // The attempt gate at the exec: a claims-on hold's pinCheck, |
| 504 | // or a planned wake's ownership check (plan 5). |
| 505 | if err := policy.Gate(); err != nil { |
| 506 | return err |
| 507 | } |
| 508 | } |
| 509 | if policy.PinCheck != nil && policy.Pin != nil { |
| 510 | // The planner's selection gate for the build about to exec |
| 511 | // (plan 5): a failure recorded since planning holds it. |
| 512 | if err := policy.PinCheck(*policy.Pin); err != nil { |
| 513 | return err |
| 514 | } |
| 515 | } |
| 516 | // The build state names the build before it runs against the |
| 517 | // database (settings spec §4.2), beside the claimed lease's |
| 518 | // lock.json: the claim actor's bucket at Holder.Location, never |
| 519 | // the session's g-<gen>/ prefix (ruling R-CLAIMPREFIX). The file |
| 520 | // is already open and verified against its digest here |
| 521 | // (RoleLaunch.Start). Every exec records, a retry's included; |
| 522 | // the same build makes the write idempotent. Moves probes come |
| 523 | // through here too (ruling R-EXEC-PROBES). |
| 524 | // |
| 525 | // A held wake first checks the build the snapshot hashed |
| 526 | // (s.Staged, the version of the bytes opened; Task 5d) against |
| 527 | // its target (settings spec §4.3 rule 3, Astra I2 #4): root may |
| 528 | // have repointed between the wait and the snapshot. A refusal |
| 529 | // records nothing, and like a refused record it is terminal for |
| 530 | // this loop: no failed attempt, no retry (unrecorded). |
| 531 | if policy.Held != "" { |
| 532 | a.mu.Lock() |
| 533 | plan, route := e.plan, e.route |
| 534 | a.mu.Unlock() |
| 535 | if !heldIdentityOK(policy.Held, s.Staged, plan, route) { |
| 536 | unrecorded = true |
| 537 | log.Printf("activator: %s not started: it is held on %s, and the launch hashed %q (the live link moved after the check); the next wake checks again", e.host, policy.Held, s.Staged.Version) |
| 538 | return fmt.Errorf("%w: %s would run %q, wants %s", errStaleLink, e.host, s.Staged.Version, policy.Held) |
| 539 | } |
| 540 | } |
| 541 | at := WithBuildStateLocation(ctx, BuildStateLocation{Bucket: r.Bucket, Prefix: admitted.Claim.Holder.Location}) |
| 542 | // A planned recovery boot records with the planner's selection |
| 543 | // (ruling R-PLANNED-RECOVERY), as doWake's step 2b does. |
| 544 | a.mu.Lock() |
| 545 | plan, route := e.plan, e.route |
| 546 | a.mu.Unlock() |
| 547 | at = withRecoverySelection(at, plan, route, s.Staged.Version) |
| 548 | if err := a.recordBuild(at, e.host, s.Staged.Version); err != nil { |
| 549 | unrecorded = true |
| 550 | return err |
| 551 | } |
| 552 | if policy.Gate != nil { |
| 553 | // Asked again after the record, as startSidecar does: a |
| 554 | // fence raised or a lease lost while the build state was |
| 555 | // written (the PUT) execs nothing (Task 12a). |
| 556 | if err := policy.Gate(); err != nil { |
| 557 | return err |
| 558 | } |
| 559 | } |
| 560 | if err := budget.start(ctx, s.OpenProbe); err != nil { |
| 561 | return err |
| 562 | } |
| 563 | if e.pin != nil { |
| 564 | // A hold's pinned wake has reached its build's exec: from |
| 565 | // here a failure is the build's (Task 6, markPinExec). |
| 566 | a.markPinExec(e) |
| 567 | } |
| 568 | return nil |
| 569 | } |
| 570 | cmd, prepared, settled, producer, startErr := s.RoleLauncher.Start(ctx, "main", s.CurrentAttempt(), s.Build, beforeStart) |
| 571 | // A refusal that came back before the exec hook (the closed isolated |
| 572 | // launch, a claim read, an absent claim, a changed stamp) ran nothing |
| 573 | // of the build: it is the admission's, never the build's failure. |
| 574 | startErr = claimAdmission(startErr, reached) |
| 575 | |
| 576 | p := newClaimProcessFor(cmd, prepared, s.RoleLauncher.Root != nil) |
| 577 | p.settle = settled |
| 578 | p.bindProducer(producer) |
| 579 | if unrecorded { |
| 580 | // A refused record, or a held wake's refused identity, ran |
| 581 | // nothing of the build (ruling R-EXEC-CLAIMRETRY): no failed |
| 582 | // attempt is written (s.Failed), and the loop does not retry it. |
| 583 | // The wake fails as fenced (memberSafeWakeReason). |
| 584 | cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second) |
| 585 | cleanErr := p.Stop(cleanup) |
| 586 | cancelCleanup() |
| 587 | if cleanErr != nil { |
| 588 | s.Retain(p, rep, errors.Join(startErr, cleanErr)) |
| 589 | retained = true |
| 590 | return errors.Join(startErr, cleanErr) |
| 591 | } |
| 592 | return startErr |
| 593 | } |
| 594 | if startErr == nil { |
| 595 | deadline := time.Now().Add(policy.Window) |
| 596 | if policy.Probe { |
| 597 | deadline = budget.opened.Add(policy.Window) |
| 598 | } |
| 599 | if !policy.HealthBy.IsZero() && policy.HealthBy.Before(deadline) { |
| 600 | // A first wake's pinned build: health until the waiters' |
| 601 | // deadline minus 25 s (zero-downtime spec §8.2). |
| 602 | deadline = policy.HealthBy |
| 603 | } |
| 604 | readyCtx, cancelReady := context.WithDeadline(ctx, deadline) |
| 605 | startErr = s.Ready(readyCtx, cmd) |
| 606 | cancelReady() |
| 607 | startErr = budget.result(startErr) |
| 608 | } |
| 609 | ready := startErr == nil |
| 610 | budget.ready = ready |
| 611 | if ready { |
| 612 | rep, startErr = s.Replicate(ctx, prefix) |
| 613 | if startErr == nil && rep == nil { |
| 614 | startErr = hostwrite.ErrUnavailable |
| 615 | } |
| 616 | } |
| 617 | if startErr == nil { |
| 618 | current, err := r.Read(ctx, e.host) |
| 619 | if err != nil { |
| 620 | startErr = err |
| 621 | } else if !sameAdmission(current, admitted) { |
| 622 | startErr = hostbox.ErrChanged |
| 623 | } else { |
| 624 | _, startErr = s.Publish(ctx, current) |
| 625 | } |
| 626 | } |
| 627 | if startErr == nil { |
| 628 | startErr = s.Healthy(ctx, attempt, cmd) |
| 629 | } |
| 630 | if startErr == nil { |
| 631 | a.mu.Lock() |
| 632 | if e.pinFailed != nil && policy.Pin != nil && e.pinFailed.Build == policy.Pin.Build { |
| 633 | e.pinFailed, e.pinFailReason = nil, "" // a retry of the same pin served |
| 634 | } |
| 635 | a.mu.Unlock() |
| 636 | // The generation's handle: the sockets the provider bound |
| 637 | // (genMatches, above). Zero with generations off, and wake() |
| 638 | // falls back to the canonical handle as before. |
| 639 | g := policy.Gen |
| 640 | if g.N != 0 { |
| 641 | g.Build = s.Version |
| 642 | if policy.Pin != nil { |
| 643 | g.StagedPath, g.Digest, g.Demand = policy.Pin.StagedPath, policy.Pin.Digest, policy.Pin.Demand |
| 644 | } |
| 645 | } |
| 646 | if (policy.Pin == nil || g.N == 0) && s.Staged.StagedPath != "" && s.Staged.Digest != "" && s.Staged.Version == s.Version { |
| 647 | // The file this start ran (the hashed resolution of the |
| 648 | // live link), so a legacy sidecar restart runs that file |
| 649 | // by path, never whatever the link names by then (Task |
| 650 | // 5d; ruling R-EXEC-SIDECAR). With no number, wake() |
| 651 | // keeps it on the canonical handle it gives this start. |
| 652 | // A pinned start with no number takes the same file: its |
| 653 | // version was checked against Pin.Build above, and |
| 654 | // s.Staged is what ran; the demand is the pin's (Task 12a). |
| 655 | g.Build, g.StagedPath, g.Digest = s.Version, s.Staged.StagedPath, s.Staged.Digest |
| 656 | if policy.Pin != nil { |
| 657 | g.Demand = policy.Pin.Demand |
| 658 | } |
| 659 | } |
| 660 | e.claimResult = &wokenInstance{proc: p, rep: rep, spawnVersion: s.Version, gen: g, stopOutcome: &stopObservation{}} |
| 661 | return nil |
| 662 | } |
| 663 | cleanup, cancelCleanup := context.WithTimeout(context.Background(), 10*time.Second) |
| 664 | cleanErr := p.Stop(cleanup) |
| 665 | cancelCleanup() |
| 666 | if cleanErr != nil { |
| 667 | s.Retain(p, rep, errors.Join(startErr, cleanErr)) |
| 668 | retained = true |
| 669 | return errors.Join(startErr, cleanErr) |
| 670 | } |
| 671 | if err := s.Failed(ctx, attempt); err != nil { |
| 672 | return errors.Join(startErr, err) |
| 673 | } |
| 674 | last = startErr |
| 675 | // THE outcome classifier (Task 4): a start the wake's lifecycle ended |
| 676 | // (maintenance, Shutdown, a stop), that the attempt gate refused, or |
| 677 | // that the launch refused before its exec hook is no failure of the |
| 678 | // build and records nothing (spec §6.4, §7.4). Only a pinned build |
| 679 | // that ran (cmd) and failed before it was ready is recorded, for plan |
| 680 | // 5's events (a failed first wake, spec §8.2). |
| 681 | if policy.Pin != nil && !ready && cmd != nil && classifyStep(ctx, startErr) == stepFailed { |
| 682 | a.mu.Lock() |
| 683 | if e.pinFailed == nil { |
| 684 | failed := *policy.Pin |
| 685 | e.pinFailed, e.pinFailReason = &failed, startErr.Error() |
| 686 | } |
| 687 | a.mu.Unlock() |
| 688 | } |
| 689 | // A replication/publication failure has an uncertain remote outcome. |
| 690 | // The next process start cannot reuse that opening as another attempt. |
| 691 | if ready { |
| 692 | return last |
| 693 | } |
| 694 | if errors.Is(startErr, ErrAttemptOver) { |
| 695 | return startErr |
| 696 | } |
| 697 | if attempt+1 < policy.MaxStarts { |
| 698 | waitCtx := ctx |
| 699 | cancelWait := func() {} |
| 700 | if policy.Probe && !budget.opened.IsZero() { |
| 701 | waitCtx, cancelWait = context.WithDeadline(ctx, budget.opened.Add(policy.Window)) |
| 702 | } |
| 703 | timer := time.NewTimer(claimStartBackoff << attempt) |
| 704 | select { |
| 705 | case <-waitCtx.Done(): |
| 706 | timer.Stop() |
| 707 | cancelWait() |
| 708 | return budget.result(waitCtx.Err()) |
| 709 | case <-timer.C: |
| 710 | cancelWait() |
| 711 | } |
| 712 | } |
| 713 | } |
| 714 | return last |
| 715 | } |
| 716 | |
| 717 | type claimContainment interface { |
| 718 | KillAndEmpty(context.Context) error |
| 719 | Close() error |
| 720 | } |
| 721 | |
| 722 | type claimProcess struct { |
| 723 | cmd *exec.Cmd |
| 724 | prepared claimContainment |
| 725 | done chan struct{} |
| 726 | mu sync.Mutex |
| 727 | drained, closed bool |
| 728 | // root marks a role root launched in an isolated unit: the agent holds |
| 729 | // no containment, settle is root's stop of the reservation, and done |
| 730 | // follows the unit's main process through a pidfd. |
| 731 | root bool |
| 732 | settle func(context.Context) error |
| 733 | producerFinish func(error) |
| 734 | } |
| 735 | |
| 736 | func newClaimProcess(cmd *exec.Cmd, p *cgroup.Prepared) *claimProcess { |
| 737 | c := &claimProcess{cmd: cmd, done: make(chan struct{})} |
| 738 | if p != nil { |
| 739 | c.prepared = p |
| 740 | } |
| 741 | if cmd == nil || cmd.Process == nil { |
| 742 | close(c.done) |
| 743 | } else { |
| 744 | go func() { _ = cmd.Wait(); close(c.done) }() |
| 745 | } |
| 746 | return c |
| 747 | } |
| 748 | func (p *claimProcess) Exited() bool { |
| 749 | select { |
| 750 | case <-p.done: |
| 751 | return true |
| 752 | default: |
| 753 | return false |
| 754 | } |
| 755 | } |
| 756 | func (p *claimProcess) Stop(ctx context.Context) error { |
| 757 | p.mu.Lock() |
| 758 | defer p.mu.Unlock() |
| 759 | if p.closed { |
| 760 | return nil |
| 761 | } |
| 762 | if p.root { |
| 763 | return p.stopRoot(ctx) |
| 764 | } |
| 765 | if p.prepared == nil { |
| 766 | if p.cmd != nil && p.cmd.Process != nil { |
| 767 | return hostwrite.ErrUnavailable |
| 768 | } |
| 769 | p.closed = true |
| 770 | if p.producerFinish != nil { |
| 771 | p.producerFinish(nil) |
| 772 | } |
| 773 | return nil |
| 774 | } |
| 775 | if !p.drained { |
| 776 | if err := p.prepared.KillAndEmpty(ctx); err != nil { |
| 777 | return err |
| 778 | } |
| 779 | p.drained = true |
| 780 | } |
| 781 | select { |
| 782 | case <-ctx.Done(): |
| 783 | return ctx.Err() |
| 784 | case <-p.done: |
| 785 | } |
| 786 | if p.settle != nil { |
| 787 | if err := p.settle(ctx); err != nil { |
| 788 | return err |
| 789 | } |
| 790 | } |
| 791 | if err := p.prepared.Close(); err != nil { |
| 792 | return err |
| 793 | } |
| 794 | p.closed = true |
| 795 | if p.producerFinish != nil { |
| 796 | p.producerFinish(nil) |
| 797 | } |
| 798 | return nil |
| 799 | } |
| 800 | |
| 801 | // Retire is a serving claims-on generation's retirement (zero-downtime spec |
| 802 | // §7.5 step 6, index P6 #4): SIGTERM to its leader, the app's own drain for |
| 803 | // up to grace, then Stop, whose KillAndEmpty on the generation's own child |
| 804 | // and the emptiness proof are unchanged. A candidate that never became |
| 805 | // ready is not retired: its owner Stops it, which kills at once. |
| 806 | func (p *claimProcess) Retire(ctx context.Context, grace time.Duration) error { |
| 807 | if p.cmd != nil && p.cmd.Process != nil && !p.Exited() { |
| 808 | err := p.cmd.Process.Signal(syscall.SIGTERM) |
| 809 | switch { |
| 810 | case err == nil, errors.Is(err, os.ErrProcessDone): |
| 811 | t := time.NewTimer(grace) |
| 812 | select { |
| 813 | case <-p.done: |
| 814 | case <-t.C: |
| 815 | case <-ctx.Done(): |
| 816 | } |
| 817 | t.Stop() |
| 818 | default: |
| 819 | // No drain is possible; Stop still kills and proves it. |
| 820 | log.Printf("activator: SIGTERM to a retiring claims-on generation (pid %d): %v; stopping it without a drain", p.cmd.Process.Pid, err) |
| 821 | } |
| 822 | } |
| 823 | return p.Stop(ctx) |
| 824 | } |
| 825 | |
| 826 | // provenStopped is a claims-on generation's termination proof (zero-downtime |
| 827 | // spec §7.4): its Stop has completed, which takes KillAndEmpty on its own |
| 828 | // child, the exit, the settle and the close. Stop is idempotent once |
| 829 | // closed, so this retries a Stop that failed before. |
| 830 | func (p *claimProcess) provenStopped(ctx context.Context) bool { |
| 831 | if err := p.Stop(ctx); err != nil { |
| 832 | return false |
| 833 | } |
| 834 | p.mu.Lock() |
| 835 | defer p.mu.Unlock() |
| 836 | return p.closed |
| 837 | } |
| 838 | |
| 839 | func sameAdmission(a, b hostclaim.Observed) bool { |
| 840 | if a.Absent || b.Absent || a.Host != b.Host || a.Claim == nil || b.Claim == nil || a.Claim.Holder == nil || b.Claim.Holder == nil { |
| 841 | return false |
| 842 | } |
| 843 | x, y := a.Claim, b.Claim |
| 844 | return x.Holder.Incarnation == y.Holder.Incarnation && x.Holder.Stamp == y.Holder.Stamp && x.Holder.Session == y.Holder.Session && x.Holder.Generation == y.Holder.Generation && x.Op.ID == y.Op.ID && x.Op.Kind == y.Op.Kind && x.Op.Phase == y.Op.Phase && x.Op.Exec == y.Op.Exec |
| 845 | } |
| 846 | |
| 847 | func (p *claimProcess) bindProducer(producer RoleProducer) { |
| 848 | if producer.Context == nil || producer.Finish == nil { |
| 849 | return |
| 850 | } |
| 851 | p.producerFinish = producer.Finish |
| 852 | go func() { |
| 853 | select { |
| 854 | case <-producer.Context.Done(): |
| 855 | case <-p.done: |
| 856 | } |
| 857 | cleanup, cancel := context.WithTimeout(context.Background(), 30*time.Second) |
| 858 | defer cancel() |
| 859 | for { |
| 860 | if err := p.Stop(cleanup); err == nil { |
| 861 | return |
| 862 | } |
| 863 | select { |
| 864 | case <-cleanup.Done(): |
| 865 | return |
| 866 | case <-time.After(100 * time.Millisecond): |
| 867 | } |
| 868 | } |
| 869 | }() |
| 870 | } |
| 871 | |
amadan