| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "time" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | // ReplaceWorker runs one socketless attempt (zero-downtime spec §6.5, §7.9) |
| 14 | // on an awake worker or native sidecar member: stop-then-start on the SAME |
| 15 | // entry, through respawnWorkerLocked, so the lease and the replicator stay |
| 16 | // put. The new child must still be running firstWakeSettle (10 s) after it |
| 17 | // started; a clean exit inside that window is a one-shot's success. On a |
| 18 | // failure: |
| 19 | // - restart or config: the build it replaced is respawned (spec §6.5); |
| 20 | // - release, install or retry in hold mode: nothing falls back, and the |
| 21 | // entry stops, so the instance stays down (spec §6.4, §8.3); |
| 22 | // - release, install or retry in overlap mode: fallback, the host's good |
| 23 | // build, is respawned in its place; with none, the entry stops, as a |
| 24 | // failed wake does today (spec §8.2). |
| 25 | // |
| 26 | // Every respawn here asks plan 3's ownership check at its exec, and keeps |
| 27 | // its termination proof (respawnWorkerLocked). The attempt's own respawn |
| 28 | // also asks the attempt's §7.4 checks there, at all three of its points |
| 29 | // (before the preparation, at the exec, before the install). A refusal |
| 30 | // that is a supersession brings back the build the attempt replaced, unless |
| 31 | // the attempt is a hold whose build was exec'd before the refusal (its |
| 32 | // install check shed it): that build may have migrated the database, so |
| 33 | // nothing is restored and the entry stops (final review F2; spec §6.4, |
| 34 | // §8.3). Any other refusal (no ownership) spawns nothing at all (spec |
| 35 | // §7.4). |
| 36 | // |
| 37 | // The scheduler runs the cohort-failure transition from the result, then |
| 38 | // calls PublishHeldServing for a fallback's identity. newest is called with |
| 39 | // a.mu held, as Swap's is. |
| 40 | func (a *Service) ReplaceWorker(ctx context.Context, at Attempt, fallback *release.StagedIdentity, newest func(host string) uint64) (out SwapResult) { |
| 41 | if a.cfg.HostWrites.Enabled { |
| 42 | return abandon(ErrClaimIsolationUnavailable) |
| 43 | } |
| 44 | // The execution evidence (SwapResult.Executed, final review F8): the |
| 45 | // attempt's respawn started a process of its build (attemptExec), |
| 46 | // installed or shed. Read once the respawn is over; false before it |
| 47 | // starts. A caller that stops waiting still gets it (final review wave |
| 48 | // 2, N4): the respawn is joined, or reported executed-or-unknown. |
| 49 | var executed bool |
| 50 | defer func() { out.Executed = executed }() |
| 51 | if newest == nil { |
| 52 | return abandon(errSwapSuperseded) |
| 53 | } |
| 54 | if _, ok := a.backend.(stagedWorkerStarter); !ok { |
| 55 | // A backend that cannot exec a named build would run whatever the |
| 56 | // live link names: nothing of the attempt's choosing (as a planned |
| 57 | // worker wake is refused, spawnPlannedWorker). |
| 58 | return abandon(errSwapNoGens) |
| 59 | } |
| 60 | onto := release.StagedIdentity{Version: at.Build, StagedPath: at.StagedPath, Digest: at.Digest, Demand: at.Demand} |
| 61 | a.mu.Lock() |
| 62 | e := a.entries[at.Host] |
| 63 | if err := a.swapChecksLocked(e, at, newest); err != nil { |
| 64 | a.mu.Unlock() |
| 65 | return abandon(err) |
| 66 | } |
| 67 | if !e.worker { |
| 68 | a.mu.Unlock() |
| 69 | return abandon(errSwapKind) |
| 70 | } |
| 71 | if e.workerRespawning || e.firstWatch != nil { |
| 72 | a.mu.Unlock() |
| 73 | return abandon(errSwapBusy) |
| 74 | } |
| 75 | prev := e.workerServing |
| 76 | // The attempt's build and its fallback, for the generations file (Task |
| 77 | // 12). startWorker's pending generation already names the build about |
| 78 | // to exec at each spawn; listing the attempt's own here too is harmless. |
| 79 | e.attemptBuilds = []string{onto.Version} |
| 80 | if fallback != nil { |
| 81 | e.attemptBuilds = append(e.attemptBuilds, fallback.Version) |
| 82 | } |
| 83 | // The same checks, carried into the respawn: it runs them again before |
| 84 | // its preparation, at the exec and before the install, after the old |
| 85 | // child has retired (spec §6.3, §7.4, §8.1). |
| 86 | e.attemptGate = func() error { return a.swapChecksLocked(e, at, newest) } |
| 87 | e.attemptVeto, e.attemptExec = nil, false |
| 88 | // Spec §9.4: the bounce event names the attempt's cause. |
| 89 | emitEvent(a.Logs, e.route, "info", bounceText(true, at.Cause)) |
| 90 | // The incarnation before the respawn: the attempt's child counts only if |
| 91 | // the respawn installed a NEW one (P5 review finding 5). A failed exec |
| 92 | // keeps the old e.proc and its identity, which for a restart or a config |
| 93 | // attempt is the same artifact, so the artifact alone passed it. |
| 94 | inc0 := e.workerInc |
| 95 | a.respawnWorkerLocked(e, e.proc, "attempt ("+string(at.Cause)+")", &onto) |
| 96 | done := e.respawnDone |
| 97 | a.mu.Unlock() |
| 98 | defer func() { |
| 99 | a.mu.Lock() |
| 100 | e.attemptBuilds = nil |
| 101 | a.mu.Unlock() |
| 102 | // The attempt's builds are no longer pinned on its account; what |
| 103 | // the worker runs now is its captured build, published at install. |
| 104 | _ = a.persistGenerations() |
| 105 | }() |
| 106 | |
| 107 | if !a.awaitRespawn(ctx, done) { |
| 108 | executed = a.joinAttemptRespawn(e, done) |
| 109 | return abandon(errSwapStopped) |
| 110 | } |
| 111 | a.mu.Lock() |
| 112 | veto := e.attemptVeto |
| 113 | executed = e.attemptExec |
| 114 | e.attemptVeto, e.attemptExec = nil, false |
| 115 | if a.entries[at.Host] != e || e.state != actAwake { |
| 116 | a.mu.Unlock() |
| 117 | return abandon(errSwapStopped) |
| 118 | } |
| 119 | if veto != nil { |
| 120 | // The respawn's re-check refused the attempt's build: before its |
| 121 | // exec nothing of it ran; at the install check its child was shed. |
| 122 | // Only a supersession brings back the build it replaced (index P5 |
| 123 | // #1); without ownership (a lost lease, a fence, shutdown) nothing |
| 124 | // spawns, a restore included, and the entry stops through the |
| 125 | // sweep's lease-loss path. |
| 126 | switch { |
| 127 | case errors.Is(veto, errBuildStateUnwritten) && !executed: |
| 128 | // The attempt's build state could not be written, so its |
| 129 | // exec was refused (settings spec §4.2): nothing of it ran, |
| 130 | // and ownership is intact. verdictOf reads it as NoOwnership |
| 131 | // (so a plain respawn logs a refusal), but here the replaced |
| 132 | // build comes back at once, as for an unpublished build (the |
| 133 | // incarnation-unchanged arm below). Its restore records its |
| 134 | // own build first, and a refusal there leaves the dead child |
| 135 | // to the sweep's crash restart. |
| 136 | a.restoreLocked(e, prev) |
| 137 | case verdictOf(veto) != gateSuperseded: |
| 138 | case executed && at.Mode == ModeHold: |
| 139 | // A hold build that was exec'd may have migrated the database, |
| 140 | // and nothing falls back from a hold (spec §6.4, §8.3; final |
| 141 | // review F2): no restore of the replaced build. The entry stops |
| 142 | // and stays down until its next attempt or wake. Abandoned, not |
| 143 | // a failure of the build: it was superseded, not seen to fail. |
| 144 | log.Printf("activator: worker %s: hold attempt onto %s was superseded after its exec; the build it replaced (%s) is not restored, and the worker stops", at.Host, onto.Version, prev.Version) |
| 145 | a.beginStopLocked(e, StopRelease) |
| 146 | default: |
| 147 | a.restoreLocked(e, prev) |
| 148 | } |
| 149 | a.mu.Unlock() |
| 150 | return abandon(fmt.Errorf("re-checked at the spawn: %w", veto)) |
| 151 | } |
| 152 | if e.workerInc == inc0 || !sameArtifact(e.workerServing, onto) { |
| 153 | // Nothing of the attempt's build ran: no new incarnation was |
| 154 | // installed (the exec failed, its build could not be published |
| 155 | // first, the old child is not proven gone), or the backend exec'd |
| 156 | // other bytes. Put the build it replaced back, as a failed restart |
| 157 | // does (spec §6.5), through a respawn that asks ownership and the |
| 158 | // termination proof at its own exec. This is a staging refusal, |
| 159 | // never a failure of the build, and never a success: a restart or |
| 160 | // config attempt whose child never ran acknowledges nothing (spec |
| 161 | // §8.1). |
| 162 | a.restoreLocked(e, prev) |
| 163 | a.mu.Unlock() |
| 164 | return abandon(fmt.Errorf("staging: worker not respawned onto %s", onto.Version)) |
| 165 | } |
| 166 | // The observation. firstWatch hides the worker from Current and lets an |
| 167 | // intentional stop (beginStopLocked) cancel it. unexpectedExit is |
| 168 | // cleared here, for this child: the arms that set it never clear it, and |
| 169 | // a mark left by an earlier crash would make beginStopLocked leave an |
| 170 | // intentional stop to be read as this build's exit. |
| 171 | obsCtx, cancel := context.WithCancel(ctx) |
| 172 | e.firstWatch, e.unexpectedExit = cancel, false |
| 173 | proc, drain, inc1 := e.proc, e.retireTook, e.workerInc |
| 174 | a.mu.Unlock() |
| 175 | |
| 176 | r := observeSettle(obsCtx, proc, firstWakeSettle) |
| 177 | cancel() |
| 178 | |
| 179 | a.mu.Lock() |
| 180 | e.firstWatch = nil |
| 181 | unexpected := e.unexpectedExit |
| 182 | // mine: the entry, awake, still running the attempt's child. With the |
| 183 | // scheduler's CoalesceRecycle nothing recycles it meanwhile; without it |
| 184 | // (or after a stop), the attempt does not touch what replaced it. |
| 185 | mine := a.entries[at.Host] == e && e.state == actAwake && e.workerInc == inc1 && !e.workerRespawning |
| 186 | if r == settleOK || (r == settleExited && !procCrashed(proc)) { |
| 187 | // The checks once more before the result counts (spec §6.3, §8.1): |
| 188 | // a newer release or a halted cohort since the spawn means this |
| 189 | // build must not be recorded good. Its child stays until the newer |
| 190 | // attempt replaces it. A one-shot that finished may have been parked |
| 191 | // meanwhile: no longer awake, but still not superseded. Only the |
| 192 | // sweep's clean park of this very child counts (markCleanParkLocked, |
| 193 | // past the sweep's own fence and lease checks), and only while no |
| 194 | // fence or shutdown stands now. Any other stop of the exited child |
| 195 | // (a lost lease, a fence, the agent's shutdown) leaves it not awake |
| 196 | // too, with no ownership behind the exit, and is abandoned. |
| 197 | err := a.swapChecksLocked(e, at, newest) |
| 198 | if r == settleExited && errors.Is(err, errSwapNotAwake) && |
| 199 | e.parkedInc == inc1 && a.fences[at.Host] == 0 && !a.shuttingDown { |
| 200 | err = nil |
| 201 | if newest(at.Host) != at.SchedRev { |
| 202 | err = errSwapSuperseded |
| 203 | } |
| 204 | } |
| 205 | if err != nil { |
| 206 | a.mu.Unlock() |
| 207 | return abandon(err) |
| 208 | } |
| 209 | g := a.workerResultGen(e, onto) |
| 210 | a.mu.Unlock() |
| 211 | // A clean exit inside the window is a one-shot's success (spec |
| 212 | // §6.5): nothing keeps running to be acknowledged later, so the |
| 213 | // result says so and the scheduler records the build good from it. |
| 214 | return SwapResult{Outcome: SwapDone, Gen: g, Drain: drain, CleanExit: r == settleExited} |
| 215 | } |
| 216 | if r == settleCancelled && !unexpected { |
| 217 | // An intentional stop cancels the observation and records nothing |
| 218 | // (spec §6.4). |
| 219 | a.mu.Unlock() |
| 220 | return abandon(errSwapStopped) |
| 221 | } |
| 222 | reason := "exited within " + firstWakeSettle.String() + " of starting" |
| 223 | adoption := at.Cause == CauseRelease || at.Cause == CauseInstall || at.Cause == CauseRetry |
| 224 | switch { |
| 225 | case !adoption: |
| 226 | if mine { |
| 227 | a.restoreLocked(e, prev) |
| 228 | } |
| 229 | a.mu.Unlock() |
| 230 | return SwapResult{Outcome: SwapFailed, Reason: reason} |
| 231 | case at.Mode == ModeHold: |
| 232 | if mine { |
| 233 | a.beginStopLocked(e, StopRelease) |
| 234 | } |
| 235 | a.mu.Unlock() |
| 236 | return SwapResult{Outcome: SwapDown, Reason: reason} |
| 237 | case !mine: |
| 238 | // The entry stopped, or another child replaced B, while it was |
| 239 | // observed: there is nothing here to fall back on. |
| 240 | a.mu.Unlock() |
| 241 | return SwapResult{Outcome: SwapFailed, Reason: reason} |
| 242 | case fallback == nil: |
| 243 | a.beginStopLocked(e, StopRelease) |
| 244 | a.mu.Unlock() |
| 245 | return SwapResult{Outcome: SwapFailed, Reason: reason + "; no good build to fall back to"} |
| 246 | } |
| 247 | // The fallback's publication waits for the failure transition (spec |
| 248 | // §8.1; P5 review finding 7): the respawn stores it (holdServing), and |
| 249 | // the scheduler delivers it with PublishHeldServing once done() has |
| 250 | // recorded B's failure. A synchronous publication here would sit in |
| 251 | // front of that record, and a blocked callback would leave B unheld. |
| 252 | // Like every respawn, the fallback asks ownership and the termination |
| 253 | // proof at its exec: a lost lease spawns nothing (index P6 new#1). |
| 254 | e.holdServing = true |
| 255 | inc2 := e.workerInc |
| 256 | a.respawnWorkerLocked(e, e.proc, "falling back to "+fallback.Version, fallback) |
| 257 | done = e.respawnDone |
| 258 | a.mu.Unlock() |
| 259 | if !a.awaitRespawn(ctx, done) { |
| 260 | return SwapResult{Outcome: SwapFailed, Reason: reason} |
| 261 | } |
| 262 | a.mu.Lock() |
| 263 | defer a.mu.Unlock() |
| 264 | if a.entries[at.Host] == e && e.workerInc != inc2 && sameArtifact(e.workerServing, *fallback) { |
| 265 | return SwapResult{Outcome: SwapFailed, Gen: a.workerResultGen(e, *fallback), Reason: reason} |
| 266 | } |
| 267 | return SwapResult{Outcome: SwapFailed, Reason: reason + "; the good build did not start either"} |
| 268 | } |
| 269 | |
| 270 | // restoreLocked respawns prev, the build an attempt replaced, on e (spec |
| 271 | // §6.5), retiring whatever child e holds. Like every respawn it asks |
| 272 | // ownership and the termination proof at its exec, so a host no longer |
| 273 | // owned gets nothing. Nothing is done when another respawn is in flight, |
| 274 | // or when prev was never captured (the sweep's crash restart then |
| 275 | // respawns from the live link, as a recycle with no captured build does). |
| 276 | // The caller holds a.mu. |
| 277 | func (a *Service) restoreLocked(e *actEntry, prev release.StagedIdentity) { |
| 278 | if e.workerRespawning || prev.StagedPath == "" { |
| 279 | return |
| 280 | } |
| 281 | a.respawnWorkerLocked(e, e.proc, "restoring "+prev.Version, &prev) |
| 282 | } |
| 283 | |
| 284 | // workerResultGen is the generation a socketless attempt reports: id's |
| 285 | // artifact, marked Worker so the scheduler compares it by its artifact and |
| 286 | // leaves its serving version to publishWorkerServing, with the entry's |
| 287 | // generation number. The caller holds a.mu. |
| 288 | func (a *Service) workerResultGen(e *actEntry, id release.StagedIdentity) Gen { |
| 289 | g := workerGenOf(id) |
| 290 | g.N = e.gen.N |
| 291 | return g |
| 292 | } |
| 293 | |
| 294 | // PublishHeldServing delivers a worker's publication that a fallback held |
| 295 | // back (holdServing) until the caller had recorded the failure it falls |
| 296 | // back from (spec §8.1). Incarnation-ordered like any other: a newer install |
| 297 | // since makes it a no-op. Called with no lock held. |
| 298 | func (a *Service) PublishHeldServing(host string) { |
| 299 | a.mu.Lock() |
| 300 | e := a.entries[host] |
| 301 | var held *heldServing |
| 302 | if e != nil { |
| 303 | held, e.heldServing = e.heldServing, nil |
| 304 | } |
| 305 | a.mu.Unlock() |
| 306 | if held != nil { |
| 307 | a.publishWorkerServing(e, held.id, held.inc) |
| 308 | } |
| 309 | } |
| 310 | |
| 311 | // heldServing is a worker publication held for PublishHeldServing. |
| 312 | type heldServing struct { |
| 313 | id release.StagedIdentity |
| 314 | inc uint64 |
| 315 | } |
| 316 | |
| 317 | // publishWorkerServing delivers a worker's captured identity for |
| 318 | // incarnation inc of entry e (zero-downtime spec §7.9). It is delivered only |
| 319 | // while e is the host's entry and inc its current incarnation, checked and |
| 320 | // delivered under servingMu, so an older publication can never land after a |
| 321 | // newer one. Called with no lock held. |
| 322 | // |
| 323 | // An empty version is published too (P5 review finding 8): a child exec'd |
| 324 | // from an external live-link target has no version to capture, and the |
| 325 | // row's previous ServingVersion must be cleared, not left naming a build |
| 326 | // that no longer runs. |
| 327 | func (a *Service) publishWorkerServing(e *actEntry, id release.StagedIdentity, inc uint64) { |
| 328 | if a.OnWorkerServing == nil { |
| 329 | return |
| 330 | } |
| 331 | a.servingMu.Lock() |
| 332 | defer a.servingMu.Unlock() |
| 333 | a.mu.Lock() |
| 334 | current := a.entries[e.host] == e && e.workerInc == inc |
| 335 | a.mu.Unlock() |
| 336 | if !current { |
| 337 | return |
| 338 | } |
| 339 | a.OnWorkerServing(e.host, id) |
| 340 | } |
| 341 | |
| 342 | // sameArtifact reports whether the backend exec'd exactly the chosen bytes: |
| 343 | // the staged path and its digest (the release demand is not the backend's |
| 344 | // to report). |
| 345 | func sameArtifact(got, want release.StagedIdentity) bool { |
| 346 | return got.StagedPath == want.StagedPath && got.Digest == want.Digest |
| 347 | } |
| 348 | |
| 349 | // joinAttemptRespawn is ReplaceWorker's answer for SwapResult.Executed when |
| 350 | // its caller stopped waiting before the attempt's respawn finished (final |
| 351 | // review wave 2, N4). The respawn runs on regardless: it is the entry's, on |
| 352 | // the service lifetime, and may still exec the attempt's build. So it is |
| 353 | // joined, bounded by its own worst case end to end (respawnJoinBound, the |
| 354 | // bound a terminal stop's join uses), and its evidence read as on the |
| 355 | // normal path. One still running past the bound is reported executed or |
| 356 | // unknown (true): only proof that nothing ran may count as unexecuted. |
| 357 | func (a *Service) joinAttemptRespawn(e *actEntry, done chan struct{}) bool { |
| 358 | bound := time.NewTimer(respawnJoinBound) |
| 359 | defer bound.Stop() |
| 360 | select { |
| 361 | case <-done: |
| 362 | case <-bound.C: |
| 363 | log.Printf("activator: worker %s: the attempt's respawn did not finish within %v of its caller leaving; reporting its build as possibly executed", e.host, respawnJoinBound) |
| 364 | return true |
| 365 | } |
| 366 | a.mu.Lock() |
| 367 | defer a.mu.Unlock() |
| 368 | executed := e.attemptExec |
| 369 | e.attemptVeto, e.attemptExec = nil, false |
| 370 | return executed |
| 371 | } |
| 372 | |
| 373 | // awaitRespawn waits for a respawn's goroutine to finish with every child it |
| 374 | // made (respawnDone closes last). False when ctx ended first. |
| 375 | func (a *Service) awaitRespawn(ctx context.Context, done chan struct{}) bool { |
| 376 | select { |
| 377 | case <-done: |
| 378 | return true |
| 379 | case <-ctx.Done(): |
| 380 | return false |
| 381 | } |
| 382 | } |
| 383 | |
amadan