| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "log" |
| 5 | "path/filepath" |
| 6 | "sort" |
| 7 | |
| 8 | "github.com/carlosframework/platform/internal/release" |
| 9 | ) |
| 10 | |
| 11 | // Worker pin publication (zero-downtime deploys spec §4.2, §8.5; rulings |
| 12 | // R17, R19). |
| 13 | // |
| 14 | // A worker recycles onto the build it was exec'd from, its captured serving |
| 15 | // identity, long after live/<host> and the registry row have moved on. Root's |
| 16 | // staging and root's version sweep must never overwrite or remove that staged |
| 17 | // copy, and root is a separate process that cannot see this one's memory, so |
| 18 | // the activator publishes it: the generations file (release.WriteGenerations, |
| 19 | // read by root through release.GenerationPins and release.GenerationVersions) |
| 20 | // lists, for every worker entry it holds with a captured serving build and |
| 21 | // for every build a spawn is about to exec, that build under the worker's |
| 22 | // host, by version (when it is a valid version name) and by its staged path |
| 23 | // (when it has one). It does so whatever CARLOS_SWAP says; this |
| 24 | // flag-independent writer is the one plan 5 extends. |
| 25 | // |
| 26 | // Plan 5 (Task 12) adds every live or pending INSTANCE generation (spec |
| 27 | // §8.5: current, successor, retiring, spawning): the current generation, a |
| 28 | // hold's pinned one, a planned wake's build and its fallback, every |
| 29 | // operation-set member, a failed wake retained until its end is proven, and |
| 30 | // a socketless attempt's builds. An instance generation is listed with |
| 31 | // generations on, or whenever it was exec'd from a staged path |
| 32 | // (genIdentity). Since settings Task 5c every exec-backed wake runs its |
| 33 | // recorded file by path, flag off included, so a flag-off file names those |
| 34 | // instances' files too; only a generation started from the live link (a |
| 35 | // backend that cannot pin a file) stays unlisted, the link being root's |
| 36 | // pin already. |
| 37 | // |
| 38 | // A cold wake (no plan, no pin) captures the live link once after its |
| 39 | // restore and publishes it as a pending build before it records it |
| 40 | // (settleWakeIdentity, ruling R-COLDPUBLISH); wake() drops that pending |
| 41 | // build once the entry's generation names the file. |
| 42 | // |
| 43 | // Every staged instance spawn (a planned wake and its fallback, a hold's |
| 44 | // pinned wake, an overlap successor) publishes the entry state that names |
| 45 | // its build BEFORE the exec, and does not exec when that write fails |
| 46 | // (errStagedBuildNotPublished): doWake and doClaimWake before the lease, checkPlan |
| 47 | // for a replanned build, swapOverlap before its Step 3a. The entry state is |
| 48 | // set first (e.plan, e.pin, the operation set) and then written, rather than |
| 49 | // recorded as a pending build: that state already lives as long as the |
| 50 | // spawn's process may, so nothing has to be dropped when the spawn installs. |
| 51 | // |
| 52 | // A hold is the exception (zdd-readiness Task 13, Astra R1 T12, R3 §5 item |
| 53 | // 9). It stops its incumbent before its pinned entry exists, and from that |
| 54 | // stop until the pinned wake publishes nothing serves and nothing names the |
| 55 | // hold's build, so root could sweep or restage it under the wake about to |
| 56 | // run it. swapHold records the build as a pending generation and writes the |
| 57 | // file BEFORE the stop, and drops the record when it returns: by then the |
| 58 | // pinned entry, or a retained failed wake, names the build for as long as a |
| 59 | // process of it may run. |
| 60 | // |
| 61 | // The file is written at Start, and before the first worker spawn (so it exists, even |
| 62 | // empty, before any worker runs), before every staged spawn's exec, naming |
| 63 | // the build about to run (startWorker's before-exec hook, ruling R19 item 3), |
| 64 | // after every spawn or respawn installs its captured identity in place of |
| 65 | // that pending one, and after every worker stop removes one. A write that |
| 66 | // fails is logged and marks the publication dirty, and the sweep retries it. |
| 67 | // A failed write before an exec refuses that spawn; elsewhere it blocks |
| 68 | // nothing, because root reads a publication that did not finish as an |
| 69 | // incomplete pin set (release.ReadGenerations) and then neither restages nor |
| 70 | // sweeps. |
| 71 | |
| 72 | // pendingGeneration names one build a spawn is about to exec, recorded by |
| 73 | // startWorker's before-exec hook until the spawn installs or fails. 0 is |
| 74 | // none. |
| 75 | type pendingGeneration uint64 |
| 76 | |
| 77 | type pendingBuild struct { |
| 78 | host string |
| 79 | id release.StagedIdentity |
| 80 | } |
| 81 | |
| 82 | // addPendingGeneration records id as a build host's spawn is about to exec, |
| 83 | // and returns its name. Never call it with a.mu held. |
| 84 | func (a *Service) addPendingGeneration(host string, id release.StagedIdentity) pendingGeneration { |
| 85 | a.mu.Lock() |
| 86 | defer a.mu.Unlock() |
| 87 | return a.addPendingGenerationLocked(host, id) |
| 88 | } |
| 89 | |
| 90 | // addPendingGenerationLocked is addPendingGeneration for a caller that holds |
| 91 | // a.mu: a hold registers its build in the same critical section as its |
| 92 | // admission (swapHold). |
| 93 | func (a *Service) addPendingGenerationLocked(host string, id release.StagedIdentity) pendingGeneration { |
| 94 | if a.genPending == nil { |
| 95 | a.genPending = map[pendingGeneration]pendingBuild{} |
| 96 | } |
| 97 | a.genPendingSeq++ |
| 98 | k := pendingGeneration(a.genPendingSeq) |
| 99 | a.genPending[k] = pendingBuild{host: host, id: id} |
| 100 | return k |
| 101 | } |
| 102 | |
| 103 | // dropPendingGenerationLocked forgets a pending build: its spawn failed, its |
| 104 | // child was shed, or its captured identity has been installed on the entry. |
| 105 | // 0 is a no-op. The caller holds a.mu. |
| 106 | func (a *Service) dropPendingGenerationLocked(k pendingGeneration) { |
| 107 | if k != 0 { |
| 108 | delete(a.genPending, k) |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | // generationsSnapshotLocked is what the generations file should say now: for |
| 113 | // every entry (waking, awake, mid-swap, mid-respawn or stopping, since its |
| 114 | // processes may still run) what entryGenerationsLocked names, the build of |
| 115 | // every retained failed wake (unprovenWakes), and every pending build, each |
| 116 | // listed under its host by version when the version is a valid version name, |
| 117 | // and by staged path (made absolute) when it has one. A build with neither |
| 118 | // names nothing root could restage or sweep. Each host's lists are sorted and |
| 119 | // hold each entry once. The caller holds a.mu. |
| 120 | func (a *Service) generationsSnapshotLocked() release.Generations { |
| 121 | hosts, paths := map[string]map[string]bool{}, map[string]map[string]bool{} |
| 122 | add := func(host string, id release.StagedIdentity) { |
| 123 | if v := id.Version; v != "" && release.ValidVersion(v) { |
| 124 | if hosts[host] == nil { |
| 125 | hosts[host] = map[string]bool{} |
| 126 | } |
| 127 | hosts[host][v] = true |
| 128 | } |
| 129 | if p := id.StagedPath; p != "" { |
| 130 | // A relative path is published as the exec resolved it, from |
| 131 | // this process's working directory; if even that fails, the |
| 132 | // relative path stays and WriteGenerations refuses it, so the |
| 133 | // spawn does not happen. |
| 134 | if abs, err := filepath.Abs(p); err == nil { |
| 135 | p = abs |
| 136 | } |
| 137 | if paths[host] == nil { |
| 138 | paths[host] = map[string]bool{} |
| 139 | } |
| 140 | paths[host][p] = true |
| 141 | } |
| 142 | } |
| 143 | for host, e := range a.entries { |
| 144 | a.entryGenerationsLocked(e, func(id release.StagedIdentity) { add(host, id) }) |
| 145 | } |
| 146 | // A failed wake retained until its termination is proven: its entry is |
| 147 | // gone, but its process may still run the build (F12; spec §4.2, §8.5). |
| 148 | // A worker's is its captured identity, an instance's its generation. |
| 149 | for host, u := range a.unprovenWakes { |
| 150 | add(host, u.build) |
| 151 | if id, ok := a.genIdentity(u.gen); ok { |
| 152 | add(host, id) |
| 153 | } |
| 154 | } |
| 155 | for _, pb := range a.genPending { |
| 156 | add(pb.host, pb.id) |
| 157 | } |
| 158 | return release.Generations{Hosts: sortedLists(hosts), Paths: sortedLists(paths)} |
| 159 | } |
| 160 | |
| 161 | // entryGenerationsLocked calls add with every build e's processes run or may |
| 162 | // be about to run (spec §8.5): |
| 163 | // - a worker's captured serving build; |
| 164 | // - the current generation, and every operation-set member's (successor, |
| 165 | // retiring, spawning), and the build of a shed worker child the set (or, |
| 166 | // with generations off, the entry) retains (final review F12, R2-1); |
| 167 | // - a hold's pinned generation, from the moment its waking entry exists; |
| 168 | // - a planned wake's build and its fallback, from the moment the plan is |
| 169 | // stored, so both are on disk before either execs; |
| 170 | // - a socketless attempt's build and fallback (Task 11); |
| 171 | // - the generation every run not yet proven settled was admitted |
| 172 | // against (merge ruling M2). |
| 173 | // |
| 174 | // The plan's builds stay listed while the entry lives, as e.plan does |
| 175 | // (afterWake reads it after the wake): a few more pins than the processes |
| 176 | // strictly need, never fewer. The caller holds a.mu. |
| 177 | func (a *Service) entryGenerationsLocked(e *actEntry, add func(release.StagedIdentity)) { |
| 178 | gen := func(g Gen) { |
| 179 | if id, ok := a.genIdentity(g); ok { |
| 180 | add(id) |
| 181 | } |
| 182 | } |
| 183 | if e.worker { |
| 184 | add(e.workerServing) |
| 185 | } |
| 186 | gen(e.gen) |
| 187 | if e.pin != nil { |
| 188 | gen(*e.pin) |
| 189 | } |
| 190 | if e.plan.planned() { |
| 191 | add(e.plan.Boot) |
| 192 | if e.plan.Fallback != nil { |
| 193 | add(*e.plan.Fallback) |
| 194 | } |
| 195 | } |
| 196 | for _, m := range e.ops.members { |
| 197 | add(m.build) |
| 198 | gen(m.gen) |
| 199 | } |
| 200 | if e.shed != nil { |
| 201 | add(e.shed.build) |
| 202 | } |
| 203 | for _, v := range e.attemptBuilds { |
| 204 | add(release.StagedIdentity{Version: v}) |
| 205 | } |
| 206 | // A run's generation, until the run is proven settled (merge ruling |
| 207 | // M2): its command execs that build, and may outlive the generation's |
| 208 | // own process once a swap retires it. |
| 209 | for _, h := range cmdHoldsLocked(e) { |
| 210 | gen(h.gen) |
| 211 | } |
| 212 | } |
| 213 | |
| 214 | // genIdentity is g as the generations file names it, and whether it is |
| 215 | // listed at all. A generation exec'd from a staged path is listed whatever |
| 216 | // the switch says. Otherwise it is listed only with generations on: with |
| 217 | // them off an instance's generation is the live link's build, which root |
| 218 | // pins as the row's Version already, and plan 4's flag-off file never |
| 219 | // named it. |
| 220 | func (a *Service) genIdentity(g Gen) (release.StagedIdentity, bool) { |
| 221 | if g.StagedPath == "" && (!a.gens || g.Build == "") { |
| 222 | return release.StagedIdentity{}, false |
| 223 | } |
| 224 | return release.StagedIdentity{Version: g.Build, StagedPath: g.StagedPath}, true |
| 225 | } |
| 226 | |
| 227 | // publishesLocked reports whether e names anything in the generations file, |
| 228 | // so a change that removes e must rewrite it. The caller holds a.mu. |
| 229 | func (a *Service) publishesLocked(e *actEntry) bool { |
| 230 | named := false |
| 231 | a.entryGenerationsLocked(e, func(id release.StagedIdentity) { |
| 232 | if id.StagedPath != "" || release.ValidVersion(id.Version) { |
| 233 | named = true |
| 234 | } |
| 235 | }) |
| 236 | return named |
| 237 | } |
| 238 | |
| 239 | // publishBeforeExec writes the generations file before a staged instance |
| 240 | // spawn's exec (spec §4.2, §8.5; the index's generations-file contract). |
| 241 | // The caller has already put the build about to run on its entry (e.plan, |
| 242 | // e.pin, the operation set), so the snapshot names it. When the write fails |
| 243 | // the caller does not exec: root reads only what the file names, and could |
| 244 | // sweep or restage a build it does not name under the process. The refusal |
| 245 | // is errStagedBuildNotPublished, which classifyStep reads as abandonment, |
| 246 | // never a failure of the build. The write error itself goes to the operator |
| 247 | // log (persistGenerations), not into the returned error, which may reach a |
| 248 | // member's wake event (as the fence's wording, memberSafeWakeReason). Never |
| 249 | // call it with a.mu held. |
| 250 | func (a *Service) publishBeforeExec() error { |
| 251 | if err := a.persistGenerations(); err != nil { |
| 252 | return errStagedBuildNotPublished |
| 253 | } |
| 254 | return nil |
| 255 | } |
| 256 | |
| 257 | func sortedLists(m map[string]map[string]bool) map[string][]string { |
| 258 | out := make(map[string][]string, len(m)) |
| 259 | for k, set := range m { |
| 260 | l := make([]string, 0, len(set)) |
| 261 | for v := range set { |
| 262 | l = append(l, v) |
| 263 | } |
| 264 | sort.Strings(l) |
| 265 | out[k] = l |
| 266 | } |
| 267 | return out |
| 268 | } |
| 269 | |
| 270 | // persistGenerations writes the generations file from a snapshot of the |
| 271 | // entries. Never call it with a.mu held: it takes genMu first, then a.mu only |
| 272 | // for the snapshot, and does the file I/O off a.mu. genMu serialises the |
| 273 | // writers, and each one snapshots after taking it, so an older snapshot can |
| 274 | // never land over a newer one. A no-op when GenerationsFile is empty. |
| 275 | func (a *Service) persistGenerations() error { |
| 276 | if a.GenerationsFile == "" { |
| 277 | return nil |
| 278 | } |
| 279 | a.genMu.Lock() |
| 280 | defer a.genMu.Unlock() |
| 281 | return a.persistGenerationsGenLocked() |
| 282 | } |
| 283 | |
| 284 | // persistGenerationsGenLocked is persistGenerations' body. The caller holds |
| 285 | // genMu and not a.mu. |
| 286 | func (a *Service) persistGenerationsGenLocked() error { |
| 287 | a.mu.Lock() |
| 288 | g := a.generationsSnapshotLocked() |
| 289 | a.mu.Unlock() |
| 290 | if err := release.WriteGenerations(a.GenerationsFile, g); err != nil { |
| 291 | a.genDirty = true |
| 292 | log.Printf("activator: publish generations to %s: %v; the sweep retries", a.GenerationsFile, err) |
| 293 | return err |
| 294 | } |
| 295 | a.genWritten = true |
| 296 | a.genDirty = false |
| 297 | return nil |
| 298 | } |
| 299 | |
| 300 | // publishGenerationsBeforeSpawn makes sure the generations file has been |
| 301 | // written once before a worker is spawned. A failure is logged and left to |
| 302 | // the sweep; this write does not decide the spawn, the staged spawn's own |
| 303 | // before-exec publication does (startWorker). Never call it with a.mu held. |
| 304 | func (a *Service) publishGenerationsBeforeSpawn() { |
| 305 | if a.GenerationsFile == "" { |
| 306 | return |
| 307 | } |
| 308 | a.genMu.Lock() |
| 309 | defer a.genMu.Unlock() |
| 310 | if a.genWritten { |
| 311 | return |
| 312 | } |
| 313 | _ = a.persistGenerationsGenLocked() |
| 314 | } |
| 315 | |
| 316 | // retryGenerations rewrites the generations file if the last write failed. |
| 317 | // The sweep calls it after releasing a.mu. |
| 318 | func (a *Service) retryGenerations() { |
| 319 | if a.GenerationsFile == "" { |
| 320 | return |
| 321 | } |
| 322 | a.genMu.Lock() |
| 323 | defer a.genMu.Unlock() |
| 324 | if !a.genDirty { |
| 325 | return |
| 326 | } |
| 327 | _ = a.persistGenerationsGenLocked() |
| 328 | } |
| 329 | |
amadan