| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/registry" |
| 10 | "github.com/carlosframework/platform/internal/release" |
| 11 | ) |
| 12 | |
| 13 | // stagedWorkerStarter is the optional Backend capability behind spec §7.9: |
| 14 | // spawn a worker from a NAMED staged build (onto != nil, a recycle) or from |
| 15 | // whatever live/<host> names now (onto == nil, a wake or a converge), and say |
| 16 | // exactly which bytes were exec'd. ExecBackend and SplitBackend have it. A |
| 17 | // backend without it keeps today's StartWorker, and its workers have no |
| 18 | // captured identity to recycle onto. An implementation runs ctx's |
| 19 | // before-exec hook (runBeforeExec) once it has resolved the identity and |
| 20 | // before it execs, and does not exec when the hook fails (ruling R19 item 3). |
| 21 | // It asks ctx's exec gate (withExecGate) as StartStaged does: after all of |
| 22 | // its preparation, the cgroup's included, immediately before each start |
| 23 | // attempt, and when the gate refuses starts nothing and returns its error |
| 24 | // (Astra R3 §5 item 4). |
| 25 | type stagedWorkerStarter interface { |
| 26 | StartWorkerStaged(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity) (Proc, release.StagedIdentity, error) |
| 27 | } |
| 28 | |
| 29 | var ( |
| 30 | _ stagedWorkerStarter = (*ExecBackend)(nil) |
| 31 | _ stagedWorkerStarter = (*SplitBackend)(nil) |
| 32 | ) |
| 33 | |
| 34 | // ErrBuildNotPublished is a worker spawn refused because the build it was |
| 35 | // about to exec could not be written to the generations file first (ruling |
| 36 | // R19 item 3): root could not see that build as referenced, so it is not |
| 37 | // run. The failed-spawn handling applies, and the sweep retries the write. |
| 38 | // It is never a failure of the build (classifyStep). Its text is main's |
| 39 | // (final review F9): it reaches a member through a worker wake's failure |
| 40 | // event with the flag off, which must read as it always did. Plan 5's |
| 41 | // staged publication has its own sentinel, errStagedBuildNotPublished. |
| 42 | var ErrBuildNotPublished = errors.New("worker build not published before its exec") |
| 43 | |
| 44 | // errStagedBuildNotPublished is ErrBuildNotPublished for plan 5's staged |
| 45 | // publication (Task 12, publishBeforeExec): the instance spawns (a planned |
| 46 | // or pinned wake, a replanned build, an overlap successor) and a planned |
| 47 | // wake's step 0.8, which a planned worker wake passes too before its lease. |
| 48 | // A worker's own exec-time publication (startWorker's hook) keeps |
| 49 | // ErrBuildNotPublished. Abandonment as well (classifyStep), and only |
| 50 | // reachable with the flag on. A member's wake event never shows its text: |
| 51 | // memberSafeWakeReason maps it to the maintenance fence's wording (final |
| 52 | // review F9). |
| 53 | var errStagedBuildNotPublished = errors.New("staged build not published before its exec") |
| 54 | |
| 55 | // beforeExecKey carries the activator's before-exec hook on a spawn's |
| 56 | // context (withBeforeExec). |
| 57 | type beforeExecKey struct{} |
| 58 | |
| 59 | // withBeforeExec returns ctx carrying f, the hook a staged worker spawn runs |
| 60 | // between resolving the identity it will exec and the exec itself (ruling R19 |
| 61 | // item 3). It rides on the context so StartWorkerStaged's signature, and the |
| 62 | // stagedWorkerStarter interface, stay as they are. |
| 63 | // |
| 64 | // The hook is the build's publication, then the build-state record (settings |
| 65 | // spec §4.2) behind one ask of the caller's gate, as an instance's record |
| 66 | // sits behind its gate in startPinnedAt. That ask is not the exec-time |
| 67 | // check: the gate also rides the context as an instance spawn's does |
| 68 | // (withExecGate), because the hook runs before the cgroup's preparation and |
| 69 | // once, where ownership has to be asked after that preparation and before |
| 70 | // every start attempt (Astra R3 §5 item 4, ruling 206). |
| 71 | func withBeforeExec(ctx context.Context, f func(release.StagedIdentity) error) context.Context { |
| 72 | return context.WithValue(ctx, beforeExecKey{}, f) |
| 73 | } |
| 74 | |
| 75 | // runBeforeExec runs ctx's before-exec hook, if any, with the identity about |
| 76 | // to be exec'd. A backend that implements stagedWorkerStarter calls it after |
| 77 | // resolving that identity and before its spawn's preparation, and does not |
| 78 | // exec when it returns an error. |
| 79 | func runBeforeExec(ctx context.Context, id release.StagedIdentity) error { |
| 80 | if f, ok := ctx.Value(beforeExecKey{}).(func(release.StagedIdentity) error); ok && f != nil { |
| 81 | return f(id) |
| 82 | } |
| 83 | return nil |
| 84 | } |
| 85 | |
| 86 | // startWorker spawns through the staged capability when the backend has it, |
| 87 | // and through StartWorker otherwise. Never call it with a.mu held. |
| 88 | // |
| 89 | // Through the staged capability, the build about to be exec'd is published |
| 90 | // BEFORE the exec (ruling R19 item 3): the backend's before-exec hook records |
| 91 | // it as a pending generation and writes the generations file. A write that |
| 92 | // fails refuses the spawn with ErrBuildNotPublished. The returned |
| 93 | // pendingGeneration names that record. On an error startWorker has already |
| 94 | // dropped it; on success the caller drops it (dropPendingGenerationLocked) |
| 95 | // under a.mu in the same critical section that installs the captured |
| 96 | // identity, or after shedding the child, and then publishes again. |
| 97 | // |
| 98 | // gate, when set, is the caller's ownership check at the exec (final review |
| 99 | // F1). Through the staged capability it rides the context as the exec gate |
| 100 | // (withExecGate), and the backend asks it as an instance spawn's: after the |
| 101 | // materialization, the publication, the cgroup's preparation and the |
| 102 | // process's placement, and before EACH exec, the PID-move fallback after a |
| 103 | // failed flat placement included (Astra R3 §5 item 4, ruling 206). In the |
| 104 | // before-exec hook, as it |
| 105 | // was, it was asked once and before the cgroup was prepared, so ownership |
| 106 | // lost during that preparation, or between a failed start and its retry, |
| 107 | // still started the worker. On a backend without the staged capability it |
| 108 | // is asked immediately before StartWorker. Its refusal execs nothing and is |
| 109 | // returned as is. |
| 110 | func (a *Service) startWorker(ctx context.Context, rt registry.Route, env map[string]string, onto *release.StagedIdentity, gate func() error) (Proc, release.StagedIdentity, pendingGeneration, error) { |
| 111 | if a.cfg.HostWrites.Enabled { |
| 112 | return nil, release.StagedIdentity{}, 0, ErrClaimIsolationUnavailable |
| 113 | } |
| 114 | s, ok := a.backend.(stagedWorkerStarter) |
| 115 | if !ok { |
| 116 | // No staged capability: no identity is captured, so no build state |
| 117 | // is recorded. Production's SplitBackend always has the capability; |
| 118 | // it sends a unit-backed row (which UnitBackend refuses as a worker, |
| 119 | // and which holds no real lease) to StartWorker without the hook. |
| 120 | // So this arm, and that one, record nothing: settings spec §4.2 |
| 121 | // covers the rows that run under a lease in the replica bucket. |
| 122 | if gate != nil { |
| 123 | if err := gate(); err != nil { |
| 124 | return nil, release.StagedIdentity{}, 0, err |
| 125 | } |
| 126 | } |
| 127 | p, ver, err := a.backend.StartWorker(ctx, rt, env) |
| 128 | return p, release.StagedIdentity{Version: ver}, 0, err |
| 129 | } |
| 130 | var pend pendingGeneration |
| 131 | hook := func(id release.StagedIdentity) error { |
| 132 | if pend != 0 { |
| 133 | a.mu.Lock() |
| 134 | a.dropPendingGenerationLocked(pend) |
| 135 | a.mu.Unlock() |
| 136 | } |
| 137 | pend = a.addPendingGeneration(rt.Host, id) |
| 138 | if err := a.persistGenerations(); err != nil { |
| 139 | log.Printf("activator: worker %s not spawned: the build it would exec (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err) |
| 140 | return fmt.Errorf("%w: %v", ErrBuildNotPublished, err) |
| 141 | } |
| 142 | // Settings spec §4.2: the build state names the exact identity |
| 143 | // about to be exec'd, under the worker's lease, after its |
| 144 | // publication. Every worker spawn passes here: a wake (planned or |
| 145 | // not, boot and fallback), a crash restart, a max_run recycle, a |
| 146 | // converge and ReplaceWorker. The caller's gate is asked first, as |
| 147 | // startPinnedAt and spawnSuccessor ask theirs before their record, |
| 148 | // so a worker whose ownership is already gone records nothing. The |
| 149 | // backend asks it again on the context (withExecGate) after the |
| 150 | // cgroup's preparation and before each start attempt; a refusal |
| 151 | // there leaves the state naming a build about to run that did not, |
| 152 | // which the spec allows ("or one about to"). |
| 153 | if gate != nil { |
| 154 | if err := gate(); err != nil { |
| 155 | return err |
| 156 | } |
| 157 | } |
| 158 | return a.recordBuild(ctx, rt.Host, id.Version) |
| 159 | } |
| 160 | p, id, err := s.StartWorkerStaged(withBeforeExec(withExecGate(ctx, gate), hook), rt, env, onto) |
| 161 | if err != nil && pend != 0 { |
| 162 | a.mu.Lock() |
| 163 | a.dropPendingGenerationLocked(pend) |
| 164 | a.mu.Unlock() |
| 165 | _ = a.persistGenerations() |
| 166 | pend = 0 |
| 167 | } |
| 168 | return p, id, pend, err |
| 169 | } |
| 170 | |
| 171 | // recycleWorkerLocked replaces a worker's child for a reason that is not a |
| 172 | // build change: a crash, or an always-on worker's max_run. Automatic |
| 173 | // recycling never changes a worker's build (spec §7.9): the child is |
| 174 | // respawned from its SERVING build, the staged path and digest captured at |
| 175 | // its exec, whatever live/<host> names by now. The caller holds a.mu. |
| 176 | // |
| 177 | // CoalesceRecycle is asked first. When it says a queued release attempt is |
| 178 | // admissible now, that attempt makes the one spawn and the recycle is |
| 179 | // skipped. A worker spawned by a backend without the staged capability has |
| 180 | // no captured build. With no scheduler (CoalesceRecycle nil) it keeps |
| 181 | // today's respawn from the live link. With one, armed or refusing, it never |
| 182 | // does (Astra R3 §5 item 4): the live link names whatever root installed |
| 183 | // last, a build no admission chose, and on a refusing box possibly one the |
| 184 | // unreadable failure book holds down. So a live child (stopOld, a max_run |
| 185 | // recycle) is left running, and a dead one stays down until a release |
| 186 | // attempt or a wake starts it. The agent says so once per host. |
| 187 | // |
| 188 | // Then the drift (ruling R13). Until plan 5 queues it as a release attempt, |
| 189 | // the converge is the one path that moves a worker onto an installed build, |
| 190 | // so a recycle must not starve it: an exited worker never reaches the |
| 191 | // sweep's drift check, and a max_run shorter than ConvergeGrace would recycle |
| 192 | // before the check fired. So the recycle reads live/<host> as the sweep does |
| 193 | // (ConvergeDecision over a.liveVer and spawnVersion), starting the drift |
| 194 | // clock if this is the first sight of it, and once the drift has lasted |
| 195 | // ConvergeGrace it converges instead (onto nil). |
| 196 | // |
| 197 | // That converge is paced like the sweep's own (ruling R17): it spends the |
| 198 | // sweep's converge budget (a.convergeBudget, reset each sweep), so a box |
| 199 | // whose workers' recycles fall due together still moves at most |
| 200 | // maxConvergeCyclesPerSweep of them per sweep. And it happens only while no |
| 201 | // scheduler is present: when CoalesceRecycle is set (plan 5), that |
| 202 | // scheduler's admission owns every build change, so a recycle it did not |
| 203 | // coalesce never converges. Either way, a converge that is not admitted |
| 204 | // recycles the captured build, and the drift keeps its age for a later |
| 205 | // sweep. |
| 206 | // |
| 207 | // On a box that declined CARLOS_SWAP=overlap and honours failure records or |
| 208 | // hold-ran markers, a move onto the live link is also asked of the live |
| 209 | // exec gate first (SetLiveExecGate, Task 9), before anything is stopped: a |
| 210 | // converge it refuses recycles the captured build instead, and a worker |
| 211 | // with no captured build is not respawned from the link, so a live child |
| 212 | // keeps running and a dead one stays down until a wake starts it. |
| 213 | func (a *Service) recycleWorkerLocked(e *actEntry, stopOld Proc, why string) { |
| 214 | if hook := a.CoalesceRecycle; hook != nil && hook(e.host) { |
| 215 | log.Printf("activator: worker %s %s; a queued release attempt makes the spawn instead of a recycle", e.host, why) |
| 216 | return |
| 217 | } |
| 218 | var onto *release.StagedIdentity |
| 219 | if e.workerServing.StagedPath != "" { |
| 220 | cur, readErr := a.liveVer(e.host) |
| 221 | converge, next := ConvergeDecision(cur, readErr, e.spawnVersion, e.staleSince, a.now(), ConvergeGrace) |
| 222 | e.staleSince = next |
| 223 | switch { |
| 224 | case !converge: |
| 225 | case a.CoalesceRecycle != nil: |
| 226 | log.Printf("activator: worker %s %s; live/%s names %s, but the release scheduler owns build changes, so it recycles its captured build %s", e.host, why, e.host, cur, e.spawnVersion) |
| 227 | case a.convergeBudget <= 0: |
| 228 | log.Printf("activator: worker %s %s; live/%s names %s, but this sweep's converge budget is spent, so it recycles its captured build %s and converges on a later sweep", e.host, why, e.host, cur, e.spawnVersion) |
| 229 | case a.liveGateRefusesLocked(e, cur, "it recycles its captured build "+e.spawnVersion): |
| 230 | // A declined box that honours a record or marker holding cur |
| 231 | // (Task 9): the captured build is recycled, as for a converge |
| 232 | // not admitted, and the drift keeps its age. |
| 233 | default: |
| 234 | a.convergeBudget-- |
| 235 | log.Printf("activator: worker %s %s; it serves %s but live/%s has named %s for at least %s, so it converges onto the new build instead of recycling", e.host, why, e.spawnVersion, e.host, cur, ConvergeGrace) |
| 236 | a.respawnWorkerLocked(e, stopOld, "converging onto a new build", nil) |
| 237 | return |
| 238 | } |
| 239 | id := e.workerServing |
| 240 | onto = &id |
| 241 | } else if a.CoalesceRecycle != nil { |
| 242 | if !a.uncapturedSaid[e.host] { |
| 243 | if a.uncapturedSaid == nil { |
| 244 | a.uncapturedSaid = map[string]bool{} |
| 245 | } |
| 246 | a.uncapturedSaid[e.host] = true |
| 247 | left := uncapturedLeft(stopOld) |
| 248 | log.Printf("activator: worker %s %s; it has no captured build, and the release scheduler owns its recycles, so it is not respawned from live/%s: %s", e.host, why, e.host, left) |
| 249 | } |
| 250 | return |
| 251 | } else if a.liveLinkRefusedLocked(e, uncapturedLeft(stopOld)) { |
| 252 | // No captured build, and the build the live link names is one this |
| 253 | // declined box will not start (Task 9): asked before anything is |
| 254 | // stopped, so a live child keeps running and a dead one stays down |
| 255 | // until a wake starts it, as on an armed box. |
| 256 | return |
| 257 | } else { |
| 258 | log.Printf("activator: worker %s %s; no captured build to recycle onto, so it respawns from live/%s", e.host, why, e.host) |
| 259 | } |
| 260 | a.respawnWorkerLocked(e, stopOld, why, onto) |
| 261 | } |
| 262 | |
| 263 | // liveLinkRefusedLocked is liveGateRefusesLocked for the build live/<host> |
| 264 | // names now. With no live exec gate (every box but a declined one that |
| 265 | // honours something) it reads nothing and refuses nothing, so the recycle is |
| 266 | // today's. A link that cannot be read is left to the spawn's own gate |
| 267 | // (StartWorkerStaged). The caller holds a.mu. |
| 268 | func (a *Service) liveLinkRefusedLocked(e *actEntry, left string) bool { |
| 269 | if a.liveExecGate == nil { |
| 270 | return false |
| 271 | } |
| 272 | cur, err := a.liveVer(e.host) |
| 273 | return err == nil && a.liveGateRefusesLocked(e, cur, left) |
| 274 | } |
| 275 | |
| 276 | // uncapturedLeft is what becomes of a worker with no captured build whose |
| 277 | // recycle is not made: a live child (stopOld) keeps running, a dead one |
| 278 | // stays down. |
| 279 | func uncapturedLeft(stopOld Proc) string { |
| 280 | if stopOld == nil { |
| 281 | return "it stays down until a release attempt or a wake starts it" |
| 282 | } |
| 283 | return "the running child keeps running" |
| 284 | } |
| 285 | |
| 286 | // WorkerReportedVersion is what host's worker last said it runs, over the |
| 287 | // agent socket: display metadata only (spec §7.9). It may be empty or an |
| 288 | // app-chosen string, and it never identifies the executable. "" when no |
| 289 | // worker entry exists. |
| 290 | func (a *Service) WorkerReportedVersion(host string) string { |
| 291 | a.mu.Lock() |
| 292 | defer a.mu.Unlock() |
| 293 | if e := a.entries[host]; e != nil && e.worker { |
| 294 | return e.workerVersion |
| 295 | } |
| 296 | return "" |
| 297 | } |
| 298 | |
amadan