| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/cgroup" |
| 11 | "github.com/carlosframework/platform/internal/hostwrite" |
| 12 | "github.com/carlosframework/platform/internal/registry" |
| 13 | ) |
| 14 | |
| 15 | var ( |
| 16 | // errHoldStopSlow: the old generation's stop outlasted wakeWaitBound, the |
| 17 | // bound a hold holds requests for (spec §7.8). The hold is withdrawn |
| 18 | // before its stop finishes, so nothing of the attempt's build runs. |
| 19 | errHoldStopSlow = errors.New("swap: the old generation's stop outlasted the hold's bound; the hold is withdrawn") |
| 20 | // errPinMismatch: the backend ran something other than the hold's |
| 21 | // choice. A process ran, so the hold is down, but the text is the |
| 22 | // attempt's diagnostic and never a member's event. |
| 23 | errPinMismatch = errors.New("swap: the backend ran a different generation from the hold's choice") |
| 24 | ) |
| 25 | |
| 26 | // holdSpec is a hold swap waiting for its entry's stop (spec §7.8). |
| 27 | // finishStop answers on woke exactly once, unless swapHold withdraws the |
| 28 | // hold first (under a.mu, by clearing the entry's holdNext). |
| 29 | type holdSpec struct { |
| 30 | gen Gen // the attempt's build, staged path, digest and demand; N is drawn at spawn |
| 31 | at Attempt |
| 32 | newest func(string) uint64 |
| 33 | woke chan holdWoke // buffered 1 |
| 34 | // settleCancel ends a canary's settle observation. holdWakeLocked puts |
| 35 | // it on the pinned entry, so an intentional stop of that entry cancels |
| 36 | // the observation before anything is stopped (spec §6.4). |
| 37 | settleCancel context.CancelFunc |
| 38 | } |
| 39 | |
| 40 | // holdWoke is finishStop's answer: the pinned wake it installed, or why it |
| 41 | // installed none. |
| 42 | type holdWoke struct { |
| 43 | next *actEntry |
| 44 | err error |
| 45 | } |
| 46 | |
| 47 | // lifeDone is the agent lifetime's Done channel. It is nil (blocks forever |
| 48 | // in a select) for a Service built without New. |
| 49 | func (a *Service) lifeDone() <-chan struct{} { |
| 50 | if a.lifeCtx == nil { |
| 51 | return nil |
| 52 | } |
| 53 | return a.lifeCtx.Done() |
| 54 | } |
| 55 | |
| 56 | // swapHold is §7.8: stop-then-start on the existing path, through the |
| 57 | // barrier. The old generation stops while begin() holds requests, up to |
| 58 | // wakeWaitBound. When the stop finishes, finishStop installs a waking |
| 59 | // entry pinned to the attempt's staged build before it releases anyone |
| 60 | // waiting on the stop. Held requests are served once that generation is |
| 61 | // healthy. |
| 62 | // |
| 63 | // A row that cannot run the staged build at all is refused before |
| 64 | // anything stops: a stop-then-start whose start cannot happen would only |
| 65 | // cost the instance its running process. So is a hold whose build cannot be |
| 66 | // published in the generations file first (zdd-readiness Task 13): the |
| 67 | // admission registers the build as a pending generation, the file is written |
| 68 | // off a.mu, and the admission runs again before the stop begins. |
| 69 | // |
| 70 | // The caller waits for the stop for at most wakeWaitBound, and no longer |
| 71 | // than its own context or the agent's life. Any of the three withdraws the |
| 72 | // hold under a.mu while the stop is still under way, and the attempt is |
| 73 | // abandoned: nothing of its build has run. Once finishStop has taken the |
| 74 | // hold, the pinned wake may exec the build against the database, so the |
| 75 | // result follows that wake to its end (bounded by the wake's own deadlines) |
| 76 | // and is never abandoned on the caller's account. |
| 77 | func (a *Service) swapHold(ctx context.Context, at Attempt, newest func(string) uint64) SwapResult { |
| 78 | if err := verifyStaged(at.StagedPath, at.Digest); err != nil { |
| 79 | return abandon(fmt.Errorf("staging: %w", err)) |
| 80 | } |
| 81 | work, err := a.selectWork(ctx, at.Host) |
| 82 | if err != nil { |
| 83 | return abandon(fmt.Errorf("admission: %w", err)) |
| 84 | } |
| 85 | if !hostwrite.Legacy(work) { |
| 86 | // Claims on: refused before anything stops, whatever the runtime's |
| 87 | // capabilities (claimsSwapRefusal). The pinned wake would meet the |
| 88 | // closed claims launch, and a stop whose start cannot happen would |
| 89 | // only cost the instance its running process. |
| 90 | if err := a.claimsSwapRefusal(false); err != nil { |
| 91 | return abandon(err) |
| 92 | } |
| 93 | } |
| 94 | sctx, scancel := context.WithCancel(ctx) |
| 95 | defer scancel() |
| 96 | h := &holdSpec{ |
| 97 | gen: Gen{Build: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand}, |
| 98 | at: at, |
| 99 | newest: newest, |
| 100 | woke: make(chan holdWoke, 1), |
| 101 | } |
| 102 | if at.Canary { |
| 103 | h.settleCancel = scancel |
| 104 | } |
| 105 | var ( |
| 106 | held *actEntry // the entry whose stop the hold waits on |
| 107 | stopAt time.Time // wall clock: the old generation's stop, for the drain time (spec §9.4) |
| 108 | admitted *actEntry // the entry the first admission registered the build against |
| 109 | pend pendingGeneration |
| 110 | ) |
| 111 | // The hold's build is published BEFORE the incumbent stops |
| 112 | // (zdd-readiness Task 13, Astra R1 T12, R3 §5 item 9). From the stop |
| 113 | // until the pinned wake's own publication nothing serves and, without |
| 114 | // this, nothing names the build, so root could sweep or restage it under |
| 115 | // the wake about to run it. The record is dropped when swapHold returns, |
| 116 | // whatever the outcome: a withdrawn hold ran nothing, and a taken one is |
| 117 | // named by its pinned entry (or a retained failed wake) from then on. |
| 118 | defer func() { |
| 119 | if pend == 0 { |
| 120 | return |
| 121 | } |
| 122 | a.mu.Lock() |
| 123 | a.dropPendingGenerationLocked(pend) |
| 124 | a.mu.Unlock() |
| 125 | _ = a.persistGenerations() |
| 126 | }() |
| 127 | // Admission, then the registration in the same critical section. |
| 128 | err = a.admit(ctx, work, func() error { |
| 129 | a.mu.Lock() |
| 130 | defer a.mu.Unlock() |
| 131 | e := a.entries[at.Host] |
| 132 | if err := a.holdAdmitLocked(e, at, newest); err != nil { |
| 133 | return err |
| 134 | } |
| 135 | admitted = e |
| 136 | if id, ok := a.genIdentity(h.gen); ok { |
| 137 | pend = a.addPendingGenerationLocked(at.Host, id) |
| 138 | } |
| 139 | return nil |
| 140 | }) |
| 141 | if err != nil { |
| 142 | return abandon(err) |
| 143 | } |
| 144 | // persistGenerations takes genMu and then a.mu, so it runs here, between |
| 145 | // the two critical sections, never inside the admission. A write that |
| 146 | // fails stops nothing: the incumbent keeps serving and the attempt is |
| 147 | // abandoned with nothing of its build run (errStagedBuildNotPublished, |
| 148 | // which is never a failure of the build). |
| 149 | if err := a.publishBeforeExec(); err != nil { |
| 150 | return abandon(err) |
| 151 | } |
| 152 | // The admission once more, with a.mu re-taken: anything the first one |
| 153 | // checked may have changed while the file was written (plan review #10). |
| 154 | // The same entry, still awake and owned, the attempt still the newest, |
| 155 | // and every gate of the first admission. A refusal withdraws the pending |
| 156 | // build (the deferred drop) and stops nothing. |
| 157 | err = a.admit(ctx, work, func() error { |
| 158 | a.mu.Lock() |
| 159 | defer a.mu.Unlock() |
| 160 | e := a.entries[at.Host] |
| 161 | if err := a.holdAdmitLocked(e, at, newest); err != nil { |
| 162 | return err |
| 163 | } |
| 164 | if e != admitted { |
| 165 | // The entry the build was admitted against has gone (a sleep and |
| 166 | // a fresh wake in the window). The checks pass on its successor, |
| 167 | // but this attempt was never admitted there. |
| 168 | return errSwapNotAwake |
| 169 | } |
| 170 | e.holdNext = h |
| 171 | held, stopAt = e, time.Now() |
| 172 | // A run the stop interrupts reads the attempt's cause: a restart or |
| 173 | // a config change is a restart, not a new release (stopWhyFor). |
| 174 | a.beginStopLocked(e, stopWhyFor(at.Cause)) |
| 175 | return nil |
| 176 | }) |
| 177 | if err != nil { |
| 178 | if held != nil { |
| 179 | // The work refused after its body began the stop. The stop |
| 180 | // stands; the hold does not. |
| 181 | if !a.withdrawHold(held, h) { |
| 182 | return a.holdResult(sctx, at, newest, <-h.woke, time.Since(stopAt)) |
| 183 | } |
| 184 | } |
| 185 | return abandon(err) |
| 186 | } |
| 187 | bound := time.NewTimer(wakeWaitBound) |
| 188 | defer bound.Stop() |
| 189 | var hw holdWoke |
| 190 | select { |
| 191 | case hw = <-h.woke: |
| 192 | case <-bound.C: |
| 193 | hw, err = a.withdrawOrTake(held, h, errHoldStopSlow) |
| 194 | case <-ctx.Done(): |
| 195 | hw, err = a.withdrawOrTake(held, h, errSwapStopped) |
| 196 | case <-a.lifeDone(): |
| 197 | hw, err = a.withdrawOrTake(held, h, errSwapShutdown) |
| 198 | } |
| 199 | if err != nil { |
| 200 | return abandon(err) |
| 201 | } |
| 202 | return a.holdResult(sctx, at, newest, hw, time.Since(stopAt)) // finishStop answers once the old generation is proven gone |
| 203 | } |
| 204 | |
| 205 | // holdAdmitLocked is a hold's admission (spec §7.4, §7.8): the attempt's |
| 206 | // checks on e, then the gates that refuse a row before anything stops (the |
| 207 | // kind, the layout, an operation already under way). swapHold asks it twice, |
| 208 | // before the build's publication and again after it, with a.mu re-taken. |
| 209 | // The caller holds a.mu. |
| 210 | func (a *Service) holdAdmitLocked(e *actEntry, at Attempt, newest func(string) uint64) error { |
| 211 | if err := a.swapChecksLocked(e, at, newest); err != nil { |
| 212 | return err |
| 213 | } |
| 214 | if e.worker || e.route.Kind != "instance" || e.route.Backing == registry.BackingUnit { |
| 215 | return errSwapKind |
| 216 | } |
| 217 | // The layouts that can run a staged generation: flat, foreign on its |
| 218 | // canonical socket, and nested (root mounts the staged build, Task 9). |
| 219 | if l, ok := a.swapLayout(e.route); ok && l != LayoutFlat && l != LayoutForeign && l != LayoutNested { |
| 220 | return errSwapLayout |
| 221 | } |
| 222 | if len(e.ops.members) > 0 { |
| 223 | return errSwapBusy |
| 224 | } |
| 225 | return nil |
| 226 | } |
| 227 | |
| 228 | // withdrawHold clears h from e while the stop has not taken it, and reports |
| 229 | // whether it did. False means finishStop already took the hold and answered |
| 230 | // on h.woke in the same critical section. |
| 231 | func (a *Service) withdrawHold(e *actEntry, h *holdSpec) bool { |
| 232 | a.mu.Lock() |
| 233 | defer a.mu.Unlock() |
| 234 | if e.holdNext != h { |
| 235 | return false |
| 236 | } |
| 237 | e.holdNext = nil |
| 238 | return true |
| 239 | } |
| 240 | |
| 241 | // withdrawOrTake withdraws the hold for why, or, when finishStop has taken |
| 242 | // it already, returns its answer. |
| 243 | func (a *Service) withdrawOrTake(e *actEntry, h *holdSpec, why error) (holdWoke, error) { |
| 244 | if a.withdrawHold(e, h) { |
| 245 | return holdWoke{}, why |
| 246 | } |
| 247 | return <-h.woke, nil |
| 248 | } |
| 249 | |
| 250 | // holdResult follows a hold's pinned wake to its outcome. Its execution |
| 251 | // evidence is the pinned wake's (pinExec, set from the spawn's result), and |
| 252 | // a wake that came up healthy ran the build: SwapResult.Executed (final |
| 253 | // review F8). A pinned start handed to root whose outcome is uncertain |
| 254 | // (startUncertain) is executed-or-unknown too (final review wave 2, N4), |
| 255 | // and it is the build's failure: the hold is down (Astra R3 ruling 206). |
| 256 | func (a *Service) holdResult(sctx context.Context, at Attempt, newest func(string) uint64, hw holdWoke, drain time.Duration) (out SwapResult) { |
| 257 | if hw.err != nil { |
| 258 | return abandon(hw.err) |
| 259 | } |
| 260 | <-hw.next.ready |
| 261 | a.mu.Lock() |
| 262 | g, werr, proc, execed := hw.next.gen, hw.next.err, hw.next.proc, hw.next.pinExec |
| 263 | a.mu.Unlock() |
| 264 | defer func() { out.Executed = execed || werr == nil || startMayHaveRun(nil, werr) }() |
| 265 | switch { |
| 266 | case werr != nil && proc == nil && startMayHaveRun(nil, werr): |
| 267 | // A nested start root accepted whose outcome is uncertain (it |
| 268 | // exited before it was observed, root's start call failed, or its |
| 269 | // kill is unproven) may have run the build against the database |
| 270 | // (Astra R3 ruling 206, zdd-readiness Task 6). It is a failure of |
| 271 | // the build whatever ended the wake: down, so done() holds the build |
| 272 | // down and halts the cohort. An abandonment with Executed would be |
| 273 | // deferred and could run the build again. |
| 274 | return SwapResult{Outcome: SwapDown, Gen: g, Reason: werr.Error(), Drain: drain} |
| 275 | case werr != nil && !execed: |
| 276 | // The pinned wake failed before its build's exec: the lease, the |
| 277 | // restore, the row, the staged bytes, the claims admission. Nothing |
| 278 | // of the build ran, so this is no failure of it and nothing is held |
| 279 | // down (spec §6.4, §8.3; index P5/P6). The attempt is abandoned. |
| 280 | return abandon(werr) |
| 281 | case werr == nil: |
| 282 | if at.Canary { |
| 283 | // A canary must survive the settle before it counts (spec §6.2, |
| 284 | // §8.5). Plan 5 does not observe a pinned wake itself. |
| 285 | r := observeSettle(sctx, proc, swapSettle) |
| 286 | a.mu.Lock() |
| 287 | hw.next.settleCancel = nil |
| 288 | a.mu.Unlock() |
| 289 | switch r { |
| 290 | case settleExited: |
| 291 | // Nothing falls back from a hold (spec §8.3): the instance is |
| 292 | // down until the sweep notices the exit and a later wake runs. |
| 293 | return SwapResult{Outcome: SwapDown, Gen: g, Reason: errSettleExit.Error(), Drain: drain} |
| 294 | case settleCancelled: |
| 295 | return abandon(errSwapStopped) |
| 296 | } |
| 297 | } |
| 298 | // The checks once more before the result counts (spec §6.3, §8.1): |
| 299 | // a newer attempt or a halted cohort since the wake began means this |
| 300 | // build is not recorded good. It keeps serving until the newer |
| 301 | // attempt replaces it. The same gate as every other exec point. |
| 302 | a.mu.Lock() |
| 303 | cerr := a.attemptCurrentLocked(hw.next, at, newest, false) |
| 304 | a.mu.Unlock() |
| 305 | if cerr != nil { |
| 306 | return abandon(cerr) |
| 307 | } |
| 308 | return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain} |
| 309 | case classifyStep(nil, werr) == stepStale: |
| 310 | // The pinned wake's own gate refused the exec (pinCheck): nothing |
| 311 | // of the build ran, and this is no failure of it. |
| 312 | return abandon(werr) |
| 313 | case classifyStep(nil, werr) == stepCancelled: |
| 314 | // An intentional stop ended the pinned wake (spec §6.4). |
| 315 | return abandon(errSwapStopped) |
| 316 | } |
| 317 | return SwapResult{Outcome: SwapDown, Gen: g, Reason: werr.Error(), Drain: drain} |
| 318 | } |
| 319 | |
| 320 | // holdWakeLocked is a hold's re-check before the spawn, made when its stop |
| 321 | // has finished and under the same lock. If the check passes, it installs |
| 322 | // the pinned waking entry. The caller holds a.mu and starts the wake. |
| 323 | func (a *Service) holdWakeLocked(e *actEntry, h *holdSpec) (*actEntry, error) { |
| 324 | pin := h.gen |
| 325 | next := &actEntry{host: e.host, route: e.route, state: actWaking, ready: make(chan struct{}), pin: &pin, |
| 326 | settleCancel: h.settleCancel} |
| 327 | // The same gate here, at the stop's end, and again at the pinned wake's |
| 328 | // exec (spec §6.3, §7.4): the lease and the restore come between. |
| 329 | next.pinCheck = func() error { return a.attemptCurrentLocked(next, h.at, h.newest, true) } |
| 330 | a.entries[e.host] = next |
| 331 | if err := next.pinCheck(); err != nil { |
| 332 | delete(a.entries, e.host) |
| 333 | return nil, err |
| 334 | } |
| 335 | return next, nil |
| 336 | } |
| 337 | |
| 338 | // startPinnedChecked is a hold's pinned spawn behind its ownership and its |
| 339 | // attempt's checks. The gate is the attempt's checks (pinCheck, set by |
| 340 | // holdWakeLocked: the entry, the fences, the shutdown, the revision) plus |
| 341 | // the liveness of the lease this wake actually holds (leaseLost; the entry |
| 342 | // has no replica yet, so ownedLocked's e.rep check cannot see it). It runs |
| 343 | // under a.mu before the preparation, and again at the exec itself, |
| 344 | // immediately before StartStaged (final review F7): a renewal that failed |
| 345 | // or a fence set during Restore, the stage check or the tier read execs |
| 346 | // nothing. Its refusals are NoOwnership. |
| 347 | // |
| 348 | // The execution evidence comes from the spawn's result (final review F10): |
| 349 | // a returned process, or an unproven-start sentinel, marks the wake as one |
| 350 | // whose build may have run, before doWake cleans the failed wake up. |
| 351 | // Anything else ran nothing of the build. |
| 352 | func (a *Service) startPinnedChecked(ctx context.Context, e *actEntry, rt registry.Route, pin Gen, lease Lease) (Proc, string, Gen, error) { |
| 353 | gate := func() error { |
| 354 | a.mu.Lock() |
| 355 | defer a.mu.Unlock() |
| 356 | if e.pinCheck != nil { |
| 357 | if err := e.pinCheck(); err != nil { |
| 358 | return err |
| 359 | } |
| 360 | } |
| 361 | return leaseLost(lease) |
| 362 | } |
| 363 | if err := gate(); err != nil { |
| 364 | return nil, "", pin, err |
| 365 | } |
| 366 | proc, ver, g, err := a.startPinnedAt(ctx, rt, pin, gate, time.Time{}) |
| 367 | if proc != nil || genStartUnproven(err) { |
| 368 | a.markPinExec(e) |
| 369 | } |
| 370 | return proc, ver, g, err |
| 371 | } |
| 372 | |
| 373 | // markPinExec records e's pinned wake's execution evidence (pinExec). |
| 374 | // startPinnedChecked sets it from the spawn's result. A claims-on pinned |
| 375 | // spawn (Task 10) sets it from its exec hook; every claims-backed hold is |
| 376 | // refused before its stop on this base (claimsSwapRefusal), so that path |
| 377 | // is never a hold's here. |
| 378 | func (a *Service) markPinExec(e *actEntry) { |
| 379 | a.mu.Lock() |
| 380 | e.pinExec = true |
| 381 | a.mu.Unlock() |
| 382 | } |
| 383 | |
| 384 | // pinExecReached reports whether e's pinned wake has execution evidence. |
| 385 | func (a *Service) pinExecReached(e *actEntry) bool { |
| 386 | a.mu.Lock() |
| 387 | defer a.mu.Unlock() |
| 388 | return e.pinExec |
| 389 | } |
| 390 | |
| 391 | // wakeFailureEvent is the member's event for a failed wake, and whether |
| 392 | // there is one. An unpinned wake's is what it always was. A hold's pinned |
| 393 | // wake is the attempt's business (plan 5's swap events carry its member |
| 394 | // line): a refusal before the exec, a stale or cancelled one, or a spawn |
| 395 | // that ran the wrong build is logged only, so no swap text reaches the |
| 396 | // member's timeline. A pinned build that ran and failed gets the wording |
| 397 | // every failed wake gets. |
| 398 | func (a *Service) wakeFailureEvent(e *actEntry, err error) (string, bool) { |
| 399 | if !a.allowFilesDiagnostic(e.host, err, "event") { |
| 400 | return "", false |
| 401 | } |
| 402 | if e.pin == nil { |
| 403 | return memberSafeWakeReason(err), true |
| 404 | } |
| 405 | if !a.pinExecReached(e) || classifyStep(nil, err) != stepFailed || errors.Is(err, errPinMismatch) { |
| 406 | log.Printf("activator: %s: hold's pinned wake refused or abandoned: %v", e.host, err) |
| 407 | return "", false |
| 408 | } |
| 409 | return memberSafeWakeReason(err), true |
| 410 | } |
| 411 | |
| 412 | // startPinned is a hold's spawn: the attempt's staged build, verified and |
| 413 | // exec'd by path at its own tier, then the health check through the |
| 414 | // generation handle. StartStaged returns as soon as the process runs (Task |
| 415 | // 1), so this health check, and its deadline, is the only one. The proc |
| 416 | // comes back even on failure, so doWake's cleanupFailedWake stops it and |
| 417 | // records the outcome, and a caller that reads a nil proc as "nothing ran" |
| 418 | // is right to. |
| 419 | func (a *Service) startPinned(ctx context.Context, rt registry.Route, pin Gen) (Proc, string, Gen, error) { |
| 420 | return a.startPinnedAt(ctx, rt, pin, nil, time.Time{}) |
| 421 | } |
| 422 | |
| 423 | // startPinnedAt is startPinned with gate, when set, asked immediately |
| 424 | // before StartStaged: a refusal there returns no process, with nothing of |
| 425 | // the build run. The same gate rides into StartStaged on its context |
| 426 | // (withExecGate; final review F1) and is asked again at the backend's final |
| 427 | // launch point, after the backend's own preparation (the config read, the |
| 428 | // sockets, a nested row's launch file). A refusal there is the same: no |
| 429 | // process, so no execution evidence (startPinnedChecked, spawnPlanned). |
| 430 | // |
| 431 | // healthBy, when set, caps the health check (zero-downtime spec §8.2): a |
| 432 | // first wake's B is checked until the earlier of healthBy (the waiters' |
| 433 | // deadline minus 25 s) and spawnHealthTimeout, which genHealthy applies |
| 434 | // itself. StartStaged returns the process at once, so this is the only |
| 435 | // health bound, and a B that misses it comes back with its process for the |
| 436 | // caller to record and kill. Zero leaves the check as it was. |
| 437 | func (a *Service) startPinnedAt(ctx context.Context, rt registry.Route, pin Gen, gate func() error, healthBy time.Time) (Proc, string, Gen, error) { |
| 438 | gb, ok := a.backend.(SwapBackend) |
| 439 | if !ok { |
| 440 | return nil, "", pin, errSwapNoGens |
| 441 | } |
| 442 | if err := verifyStaged(pin.StagedPath, pin.Digest); err != nil { |
| 443 | return nil, "", pin, fmt.Errorf("staging: %w", err) |
| 444 | } |
| 445 | pin.N = nextGen() |
| 446 | if pin.Tier == (cgroup.Limit{}) { |
| 447 | pin.Tier = gb.GenTier(rt, pin.Build) // the pinned artifact's own tier (spec §7.3) |
| 448 | } |
| 449 | if gate != nil { |
| 450 | if err := gate(); err != nil { |
| 451 | return nil, "", pin, err |
| 452 | } |
| 453 | } |
| 454 | // Settings spec §4.2: the build state names pin.Build before it runs |
| 455 | // against the data. Every staged instance exec funnels through here: a |
| 456 | // hold's pinned wake, a planned boot, the good build booted untried and |
| 457 | // a planned fallback (a different build touching the data, so it |
| 458 | // records again). A refusal returns no process: nothing ran. If the |
| 459 | // backend's own exec gate then refuses, the state names a build about to |
| 460 | // run that did not, which the spec allows ("or one about to"). |
| 461 | // Timed for the journal (wake latency, pass 2). |
| 462 | began := time.Now() |
| 463 | if err := a.recordBuild(ctx, rt.Host, pin.Build); err != nil { |
| 464 | return nil, "", pin, err |
| 465 | } |
| 466 | recorded := time.Now() |
| 467 | g, proc, err := gb.StartStaged(withExecGate(ctx, gate), rt, pin) |
| 468 | if err != nil { |
| 469 | return proc, "", pin, err |
| 470 | } |
| 471 | started := time.Now() |
| 472 | if g.N != pin.N || g.Build != pin.Build || g.StagedPath != pin.StagedPath { |
| 473 | 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) |
| 474 | } |
| 475 | hctx := ctx |
| 476 | if !healthBy.IsZero() { |
| 477 | var cancel context.CancelFunc |
| 478 | hctx, cancel = context.WithDeadline(ctx, healthBy) |
| 479 | defer cancel() |
| 480 | } |
| 481 | if err := a.genHealthy(hctx, g); err != nil { |
| 482 | return proc, "", g, err |
| 483 | } |
| 484 | log.Printf("activator: %s: generation %d of %s: record %s, start %s, healthy %s later", rt.Host, g.N, g.Build, |
| 485 | recorded.Sub(began).Round(time.Millisecond), started.Sub(recorded).Round(time.Millisecond), time.Since(started).Round(time.Millisecond)) |
| 486 | gb.GenReady(rt, g, proc) |
| 487 | return proc, g.Build, g, nil |
| 488 | } |
| 489 | |
| 490 | // foreignRow reports whether host's row has the foreign layout (a |
| 491 | // hand-set socket path), which keeps its canonical socket. A flat row whose |
| 492 | // generation path would not fit a unix socket path reads foreign here too |
| 493 | // (swapLayout, rev 19 rule 4). |
| 494 | func (a *Service) foreignRow(host string) bool { |
| 495 | a.mu.Lock() |
| 496 | e := a.entries[host] |
| 497 | var rt registry.Route |
| 498 | if e != nil { |
| 499 | rt = e.route |
| 500 | } |
| 501 | a.mu.Unlock() |
| 502 | if e == nil { |
| 503 | return false |
| 504 | } |
| 505 | l, ok := a.swapLayout(rt) |
| 506 | return ok && l == LayoutForeign |
| 507 | } |
| 508 | |
| 509 | // swapLayout is rt's layout for the swap's decisions: the backend's |
| 510 | // GenLayout where it has one (a flat row whose generation path would not |
| 511 | // fit reads foreign, rev 19 rule 4), else its Classifier layout. ok is false |
| 512 | // for a backend that can say neither, whose rows are never refused on |
| 513 | // layout here. |
| 514 | func (a *Service) swapLayout(rt registry.Route) (Layout, bool) { |
| 515 | if g, ok := a.backend.(genLayouter); ok { |
| 516 | return g.GenLayout(rt), true |
| 517 | } |
| 518 | if c, ok := a.backend.(Classifier); ok { |
| 519 | l, _ := c.Layout(rt) |
| 520 | return l, true |
| 521 | } |
| 522 | return "", false |
| 523 | } |
| 524 | |
amadan