| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "log" |
| 8 | "sync" |
| 9 | |
| 10 | "github.com/carlosframework/platform/internal/registry" |
| 11 | "github.com/carlosframework/platform/internal/release" |
| 12 | ) |
| 13 | |
| 14 | // errBuildStateUnwritten refuses a spawn whose build could not be recorded |
| 15 | // as the one about to touch the instance's database (settings spec §4.2). |
| 16 | // Nothing of the build ran, so it is never a failure of the build: the |
| 17 | // classifiers read it as a refusal (classifyStep, verdictOf), and a member |
| 18 | // sees the maintenance fence's wording (memberSafeWakeReason). |
| 19 | var errBuildStateUnwritten = errors.New("the build state could not be written, so this build was not started") |
| 20 | |
| 21 | // recordBuild runs Service.BeforeExec, if set, for a spawn about to run |
| 22 | // build against host's database (settings spec §4.2): the build state names |
| 23 | // the build before the process opens the data, and a write that fails |
| 24 | // refuses the spawn with errBuildStateUnwritten. Callers hold the |
| 25 | // instance's lease and call it after their own gate, immediately before the |
| 26 | // exec. A nil BeforeExec is today's behaviour. |
| 27 | // |
| 28 | // Not to be confused with withBeforeExec / runBeforeExec (workerbuild.go): |
| 29 | // those carry startWorker's exec-time hook on a worker spawn's context, and |
| 30 | // that hook is one of recordBuild's callers. |
| 31 | // |
| 32 | // A cold wake's own record is pending until its first exec |
| 33 | // (settleWakeIdentity, amadan #272): a call for that build writes nothing |
| 34 | // here, and the exec seam writes it (runPendingRecord), after every refusal |
| 35 | // the backend makes. A call for another build writes now and supersedes |
| 36 | // it, so the exec seam never writes the wake's build over this one. A build |
| 37 | // with no name (an executable outside the version store) is refused while |
| 38 | // the hook is set: no build state could name it (ruling R-EXEC-NOVERSION). |
| 39 | // |
| 40 | // With host claims off the lease is the hibernate lease, so a context that |
| 41 | // names no location gets that one (Config.Bucket, HibernatePrefix), and the |
| 42 | // agent writes beside its lock.json. A claimed start attaches its own |
| 43 | // (startAdmitted); with claims on and none attached (a claimed instance's |
| 44 | // legacy sidecar restart), the agent locates the host through its claim. |
| 45 | func (a *Service) recordBuild(ctx context.Context, host, build string) error { |
| 46 | if a.BeforeExec == nil { |
| 47 | return nil |
| 48 | } |
| 49 | if p := pendingRecordFrom(ctx); p != nil && p.defers(build) { |
| 50 | return nil |
| 51 | } |
| 52 | if build == "" { |
| 53 | log.Printf("activator: %s not started: the executable it would run is outside the version store, so no build state can name it (settings spec 4.2); move the instance onto a promoted build", host) |
| 54 | return fmt.Errorf("%w: the executable is outside the version store", errBuildStateUnwritten) |
| 55 | } |
| 56 | if _, ok := BuildStateLocationFrom(ctx); !ok && !a.cfg.HostWrites.Enabled { |
| 57 | ctx = WithBuildStateLocation(ctx, BuildStateLocation{Bucket: a.cfg.Bucket, Prefix: a.cfg.hibernatePrefix(host)}) |
| 58 | } |
| 59 | if err := a.BeforeExec(ctx, host, build); err != nil { |
| 60 | log.Printf("activator: %s: %s not started: its build state could not be written: %v", host, build, err) |
| 61 | return fmt.Errorf("%w: %v", errBuildStateUnwritten, err) |
| 62 | } |
| 63 | return nil |
| 64 | } |
| 65 | |
| 66 | // askLiveExecGate asks the live exec gate (SetLiveExecGate) whether rt may |
| 67 | // start build, a build this branch captured from the live link and will run |
| 68 | // by path: a cold wake's step 2b capture (settleWakeIdentity) and a sidecar |
| 69 | // on an unstaged generation (startSidecar). The backend asks the gate only |
| 70 | // on its live-link arms (exe == "", onto == nil), which those starts never |
| 71 | // take, so without this a declined box's exec-time refusal (zdd-readiness |
| 72 | // Task 9) would not be asked at all. It is asked before the publication and |
| 73 | // the record, so a build the box will not start is never named as the one |
| 74 | // about to run (settings spec §4.2) and completes no request. No gate (every |
| 75 | // box but a declined one that honours records or markers) refuses nothing. |
| 76 | // The gate reads only memory, the row and the link. |
| 77 | func (a *Service) askLiveExecGate(rt registry.Route, build string) error { |
| 78 | a.mu.Lock() |
| 79 | gate := a.liveExecGate |
| 80 | a.mu.Unlock() |
| 81 | if gate == nil { |
| 82 | return nil |
| 83 | } |
| 84 | return gate(rt, build) |
| 85 | } |
| 86 | |
| 87 | // claimedBuild is the claimed path's version check before its record |
| 88 | // (ruling R-EXEC-CLAIMVERSION). The record must name the bytes that run: |
| 89 | // s.Staged is the file the snapshot hashed, and the launch verifies the |
| 90 | // file it opens against that digest. s.Version is a separate read of the |
| 91 | // live link, so a repoint between the hash and that read leaves the start |
| 92 | // naming a build other than the one it would run, and it is refused. With |
| 93 | // no hook nothing is recorded, and the start is today's. |
| 94 | func (a *Service) claimedBuild(host string, s ClaimStart) error { |
| 95 | if a.BeforeExec == nil || s.Staged.Version == s.Version { |
| 96 | return nil |
| 97 | } |
| 98 | log.Printf("activator: %s not started: the executable it hashed (%s) is build %q, but the start names build %q (the live link moved between the two reads); the next wake reads it again", host, s.Staged.StagedPath, s.Staged.Version, s.Version) |
| 99 | return fmt.Errorf("%w: the executable hashed for this start is build %q, but the start names build %q", errBuildStateUnwritten, s.Staged.Version, s.Version) |
| 100 | } |
| 101 | |
| 102 | // settleWakeIdentity is doWake's step 2b (settings spec §4.2, Astra I1 |
| 103 | // #5): the one file this wake runs, its seed included, chosen under the |
| 104 | // wake's lease after the restore and before anything execs. Its record is |
| 105 | // made pending on the context it returns, and written at the first exec. |
| 106 | // |
| 107 | // - A planned wake runs its plan's boot build, and a pinned wake (a hold |
| 108 | // swap's) its pin. Step 0.8 published both before the lease. |
| 109 | // - A held wake with a retained, verified copy of its target (heldID; |
| 110 | // the held wake check, settings spec §4.3 rule 3, with no planner) runs |
| 111 | // that copy. |
| 112 | // - Any other wake (flag off; flag on with no plan; a worker) captures the |
| 113 | // live link once (LiveIdentity, which materializes a missing link |
| 114 | // first). That capture, and a retained copy, is published as a pending |
| 115 | // generation before it is recorded (ruling R-COLDPUBLISH; anchors D18: |
| 116 | // no plan or pin published a retained copy): the generations file names |
| 117 | // a build before its build state does, and root's sweep keeps the file |
| 118 | // until the process that runs it is gone. A write that fails refuses the |
| 119 | // wake as a worker spawn's refuses its exec. The pending record is |
| 120 | // returned so the wake drops it once the entry names the build itself |
| 121 | // (wake), or the wake failed. Before that publication, a declined box's |
| 122 | // live exec gate is asked about the captured build (askLiveExecGate), |
| 123 | // so a build the box will not start is neither published nor recorded. |
| 124 | // - A held wake (held, its target) refuses an identity that is neither |
| 125 | // its target nor the planner's recovery boot (heldIdentityOK; Astra I2 |
| 126 | // #4): root repointed between the wait and the capture. That refusal |
| 127 | // comes before the publication and the record, so nothing names it. An |
| 128 | // identity with no version is never the target, so a held wake never |
| 129 | // takes the unversioned live-link fallback below (ruling |
| 130 | // R-EXEC-NOVERSION-2). |
| 131 | // - A live link outside the version store (Version "") names no build a |
| 132 | // build state could hold, and root refuses a staged executable with no |
| 133 | // version (instancectl.validateStagedExe). With BeforeExec set it is |
| 134 | // refused here (recordBuild, ruling R-EXEC-NOVERSION). With no hook (no |
| 135 | // replica bucket) the wake is today's (ruling R-EXEC-NOVERSION-2): an |
| 136 | // identity with no file comes back, nothing is published or recorded, |
| 137 | // and the caller takes the live-link calls in every layout. |
| 138 | // |
| 139 | // The record is not written here (amadan #272). The context returned |
| 140 | // carries it pending (withPendingRecord), and the seed's or the spawn's |
| 141 | // exec seam writes it (runPendingRecord): after the backend's own refusals |
| 142 | // (the replica-key gate, and the cgroup's: a killed cgroup kept, a |
| 143 | // populated legacy parent, a limit not applied, a placement that failed), |
| 144 | // immediately before the process execs. A launch refused before its exec |
| 145 | // leaves no record naming a build that never ran, and completes no request |
| 146 | // (settings spec §4.2: the record is what completes a request, at |
| 147 | // execution). A recovery plan's boot (the good build in place of the |
| 148 | // installed one the failure book blocks) is recorded with the planner's |
| 149 | // selection (ruling R-PLANNED-RECOVERY), so the agent's revalidation admits |
| 150 | // it before root's consumed file names the installed build failed. With no |
| 151 | // hook nothing is pending, as nothing is recorded. Any error means nothing |
| 152 | // was seeded or spawned; the caller releases the lease. |
| 153 | func (a *Service) settleWakeIdentity(ctx context.Context, ib identityBackend, e *actEntry, rt registry.Route, held string, heldID *release.StagedIdentity) (release.StagedIdentity, pendingGeneration, context.Context, error) { |
| 154 | a.mu.Lock() |
| 155 | pin, plan := e.pin, e.plan |
| 156 | a.mu.Unlock() |
| 157 | var ( |
| 158 | id release.StagedIdentity |
| 159 | pend pendingGeneration |
| 160 | captured bool // neither planned nor pinned: published here, below |
| 161 | ) |
| 162 | switch { |
| 163 | case plan.planned(): |
| 164 | id = plan.Boot |
| 165 | case pin != nil: |
| 166 | id = release.StagedIdentity{Version: pin.Build, StagedPath: pin.StagedPath, Digest: pin.Digest, Demand: pin.Demand} |
| 167 | case heldID != nil: |
| 168 | id, captured = *heldID, true |
| 169 | default: |
| 170 | var err error |
| 171 | if id, err = ib.LiveIdentity(ctx, rt); err != nil { |
| 172 | return release.StagedIdentity{}, 0, ctx, err |
| 173 | } |
| 174 | captured = true |
| 175 | } |
| 176 | if !heldIdentityOK(held, id, plan, rt) { |
| 177 | log.Printf("activator: %s not woken: it is held on %s, and this wake would run %q (the live link moved after the check); the next wake checks again", rt.Host, held, id.Version) |
| 178 | return release.StagedIdentity{}, 0, ctx, fmt.Errorf("%w: %s would run %q, wants %s", errStaleLink, rt.Host, id.Version, held) |
| 179 | } |
| 180 | if captured { |
| 181 | if id.Version == "" { |
| 182 | return release.StagedIdentity{}, 0, ctx, a.recordBuild(ctx, rt.Host, "") |
| 183 | } |
| 184 | // A declined box's exec-time gate, asked about the file this wake |
| 185 | // will run, before it is published or recorded (askLiveExecGate): |
| 186 | // the spawn below runs it by path, past the backend's live-link |
| 187 | // check. Its refusal comes back as it is, as the backend's would. |
| 188 | if err := a.askLiveExecGate(rt, id.Version); err != nil { |
| 189 | return release.StagedIdentity{}, 0, ctx, err |
| 190 | } |
| 191 | pend = a.addPendingGeneration(rt.Host, id) |
| 192 | if err := a.persistGenerations(); err != nil { |
| 193 | log.Printf("activator: %s not woken: the build it would run (%s %s) could not be published first: %v", rt.Host, id.Version, id.StagedPath, err) |
| 194 | if e.worker { |
| 195 | // A worker's wake reads as its exec-time refusal always has |
| 196 | // (startWorker's hook, final review F9). |
| 197 | return id, pend, ctx, fmt.Errorf("%w: %v", ErrBuildNotPublished, err) |
| 198 | } |
| 199 | return id, pend, ctx, errStagedBuildNotPublished |
| 200 | } |
| 201 | } |
| 202 | if a.BeforeExec == nil { |
| 203 | return id, pend, ctx, nil |
| 204 | } |
| 205 | rctx := ctx |
| 206 | if plan.planned() { |
| 207 | rctx = withRecoverySelection(ctx, plan, rt, id.Version) |
| 208 | } |
| 209 | build := id.Version |
| 210 | write := func() error { return a.recordBuild(rctx, rt.Host, build) } |
| 211 | // The wake's own admission, asked again once the write is done (Astra |
| 212 | // review of zdd-owed-items, P1): the bucket write takes time, and a |
| 213 | // fence raised or the wake's context ended meanwhile must start |
| 214 | // nothing, as doWake's own check before the spawn refuses it. |
| 215 | recheck := func() error { |
| 216 | if ctx.Err() != nil || !a.wakeAllowed(rt.Host) { |
| 217 | return ErrFenced |
| 218 | } |
| 219 | return nil |
| 220 | } |
| 221 | return id, pend, withPendingRecord(ctx, build, write, recheck), nil |
| 222 | } |
| 223 | |
| 224 | // pendingRecord is one wake's record of build, written at most once, at |
| 225 | // the first exec that asks for it (runPendingRecord). |
| 226 | // |
| 227 | // The write is a bucket PUT, and the lease can be lost, a fence raised or |
| 228 | // an attempt superseded while it runs. So every exec that consults the |
| 229 | // record asks recheck (the wake's own admission: its context and the |
| 230 | // maintenance fence) after it, and an exec seam that has an exec gate asks |
| 231 | // that gate again once the write has run (runExecGate, launchGateWith). |
| 232 | // Either refusal starts nothing. |
| 233 | // |
| 234 | // A refusal there leaves the record naming a build that did not run. It is |
| 235 | // not undone. An undo would be a second bucket write, made after the lease |
| 236 | // may already be lost, so it could land under another holder's lease; the |
| 237 | // object's CAS would stop it overwriting that holder's record, but writing |
| 238 | // outside the lease is exactly what the build state's single-writer rule |
| 239 | // (ruling R-BS-UPDATE) forbids. The window is the one main had before |
| 240 | // amadan #272, where step 2b wrote the record and doWake then checked the |
| 241 | // fence: the record names a build that passed every refusal of its own and |
| 242 | // was about to run, the next wake records again, and a request it completed |
| 243 | // was the request for that build. |
| 244 | type pendingRecord struct { |
| 245 | build string |
| 246 | write func() error |
| 247 | recheck func() error |
| 248 | |
| 249 | mu sync.Mutex |
| 250 | // done is set once the write ran; err is its answer, which every later |
| 251 | // exec of the wake gets too. |
| 252 | done bool |
| 253 | err error |
| 254 | // superseded is set once a record of another build was written in this |
| 255 | // wake (recordBuild): the pending one would name the wrong build. |
| 256 | superseded bool |
| 257 | } |
| 258 | |
| 259 | type pendingWakeKey struct{} |
| 260 | |
| 261 | // withPendingRecord returns ctx carrying build's pending record, written by |
| 262 | // write and followed by recheck (nil asks nothing) at every exec. |
| 263 | func withPendingRecord(ctx context.Context, build string, write, recheck func() error) context.Context { |
| 264 | return context.WithValue(ctx, pendingWakeKey{}, &pendingRecord{build: build, write: write, recheck: recheck}) |
| 265 | } |
| 266 | |
| 267 | func pendingRecordFrom(ctx context.Context) *pendingRecord { |
| 268 | p, _ := ctx.Value(pendingWakeKey{}).(*pendingRecord) |
| 269 | return p |
| 270 | } |
| 271 | |
| 272 | // defers reports whether a record of build is the pending one's to write |
| 273 | // (recordBuild then writes nothing). A record of another build supersedes |
| 274 | // it, and once it is superseded nothing defers to it. |
| 275 | func (p *pendingRecord) defers(build string) bool { |
| 276 | if build == "" { |
| 277 | return false |
| 278 | } |
| 279 | p.mu.Lock() |
| 280 | defer p.mu.Unlock() |
| 281 | if build != p.build { |
| 282 | p.superseded = true |
| 283 | return false |
| 284 | } |
| 285 | return !p.superseded |
| 286 | } |
| 287 | |
| 288 | // run writes the record once, then asks recheck. wrote reports that this |
| 289 | // call made the write. A superseded record writes and asks nothing. |
| 290 | func (p *pendingRecord) run() (wrote bool, err error) { |
| 291 | p.mu.Lock() |
| 292 | if p.superseded { |
| 293 | p.mu.Unlock() |
| 294 | return false, nil |
| 295 | } |
| 296 | if !p.done { |
| 297 | p.err, p.done, wrote = p.write(), true, true |
| 298 | } |
| 299 | err = p.err |
| 300 | p.mu.Unlock() |
| 301 | if err != nil { |
| 302 | return wrote, err |
| 303 | } |
| 304 | if p.recheck != nil { |
| 305 | if err := p.recheck(); err != nil { |
| 306 | return wrote, err |
| 307 | } |
| 308 | } |
| 309 | return wrote, nil |
| 310 | } |
| 311 | |
| 312 | // runPendingRecord writes ctx's pending wake record, if it carries one that |
| 313 | // is still due. Every exec seam a cold wake reaches asks it immediately |
| 314 | // before its exec: the flat seed's and spawn's cgroup before-start hook, |
| 315 | // a worker's, and, nested, right before root is asked to run the seed or |
| 316 | // the run unit. An error refuses that exec, with nothing run. wrote |
| 317 | // reports that this call made the write, so a seam with an exec gate asks |
| 318 | // it again. |
| 319 | func runPendingRecord(ctx context.Context) (wrote bool, err error) { |
| 320 | if p := pendingRecordFrom(ctx); p != nil { |
| 321 | return p.run() |
| 322 | } |
| 323 | return false, nil |
| 324 | } |
| 325 | |
| 326 | // pendingRecordHook is runPendingRecord as a cgroup before-start hook for |
| 327 | // a spawn that has no other: nil when ctx carries no pending record, so |
| 328 | // that spawn is exactly as it was. |
| 329 | func pendingRecordHook(ctx context.Context) func() error { |
| 330 | if pendingRecordFrom(ctx) == nil { |
| 331 | return nil |
| 332 | } |
| 333 | return func() error { |
| 334 | _, err := runPendingRecord(ctx) |
| 335 | return err |
| 336 | } |
| 337 | } |
| 338 | |
amadan