| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "log" |
| 7 | "time" |
| 8 | |
| 9 | "github.com/carlosframework/platform/internal/hostclaim" |
| 10 | "github.com/carlosframework/platform/internal/hostwrite" |
| 11 | "github.com/carlosframework/platform/internal/moves" |
| 12 | "github.com/carlosframework/platform/internal/provision" |
| 13 | "github.com/carlosframework/platform/internal/registry" |
| 14 | ) |
| 15 | |
| 16 | // The landing's probe wake (moves Phase 1 Plan 3 Task 5a; spec §7.7, §7.14; |
| 17 | // ruling H2). ProbeWake is glue over the claim's probe admission path (claim |
| 18 | // Plan 9 Group D): the same entry, wake and doClaimWake a request's wake |
| 19 | // runs, with the probe's start policy and the probe's two refusals (the |
| 20 | // registry row held by HoldOp, the claim this box's landing op at probing). |
| 21 | // The claim runtime does everything else (Prerequisites, Admit, Prepare and |
| 22 | // startAdmitted): the moves owners it dispatches to choose and evidence the |
| 23 | // cold admission into the prepared generation, render and materialize the |
| 24 | // held inputs, run the start check and publish; the moves attempts journal |
| 25 | // the window and the starts; the claim's probe readiness checks the peer. |
| 26 | // ProbeWake contains no admission, restore or readiness logic of its own. |
| 27 | // |
| 28 | // Production: unarmed. Nothing calls ProbeWake until the landing's probing |
| 29 | // phase (Plan 3 Task 6) is armed with Plan 5 Task 9's parks, and on main |
| 30 | // every claim start is refused (ClaimRoleAssembly.Preflight and |
| 31 | // RoleLaunch.Start answer ErrClaimIsolationUnavailable) until the claim's |
| 32 | // isolated root launcher lands. |
| 33 | |
| 34 | var ( |
| 35 | // ErrProbeRefused is a probe refused before any admission: the row is not |
| 36 | // held by the probe's hold op, the claim does not name this box the |
| 37 | // executor of the probe's op at probing, or the host already has a live |
| 38 | // entry. Nothing was admitted and nothing ran. |
| 39 | ErrProbeRefused = errors.New("activator: the probe was refused before admission") |
| 40 | // ErrProbeFailed is an admitted probe attempt that did not pass: the |
| 41 | // claim settled the admitted tree (FailedAdmission) and the attempt is |
| 42 | // over. The cause is joined to it (ErrAttemptOver, a refusal from the |
| 43 | // moves owners, a readiness failure). |
| 44 | ErrProbeFailed = errors.New("activator: the probe attempt failed") |
| 45 | ) |
| 46 | |
| 47 | // The probe's start policy (§7.7): at most five starts, the 30 s window |
| 48 | // counted from the first app start. |
| 49 | const ( |
| 50 | probeMaxStarts = 5 |
| 51 | probeWindow30 = time.Duration(moves.ProbeSeconds * float64(time.Second)) |
| 52 | ) |
| 53 | |
| 54 | // ProbeConfig is one probe attempt. |
| 55 | type ProbeConfig struct { |
| 56 | Host string |
| 57 | // Op is the claim's op id: the move id. |
| 58 | Op string |
| 59 | // HoldOp is the registry row's hold_op: the move id for a landing. |
| 60 | HoldOp string |
| 61 | // Deadline bounds the whole attempt: the landing's deadline, the earlier |
| 62 | // of cut_deadline and landing_deadline (ruling H8). |
| 63 | Deadline time.Time |
| 64 | // Prepared and Inputs are the build's install manifest and the move's |
| 65 | // pins; the moves owners check them against the journal and the record. |
| 66 | Prepared *moves.InstallManifest |
| 67 | Inputs *provision.MoveInputs |
| 68 | } |
| 69 | |
| 70 | // ProbeResult names the admitted session that passed: the process the entry |
| 71 | // now holds, which serves after the open on the same session (H3). |
| 72 | type ProbeResult struct { |
| 73 | Session uint64 |
| 74 | Stamp uint64 |
| 75 | Generation string |
| 76 | } |
| 77 | |
| 78 | // LiveProbe reports the awake entry ProbeWake left for host: the probe's op |
| 79 | // and its admitted session. ok is false for no entry, a waking or stopping |
| 80 | // one, or an entry that is not a passed probe. |
| 81 | func (a *Service) LiveProbe(host string) (op string, res ProbeResult, ok bool) { |
| 82 | a.mu.Lock() |
| 83 | defer a.mu.Unlock() |
| 84 | e := a.entries[host] |
| 85 | if e == nil || e.state != actAwake || e.probe == nil || e.probe.admitted == nil || e.probe.admitted.Claim == nil || e.probe.admitted.Claim.Holder == nil { |
| 86 | return "", ProbeResult{}, false |
| 87 | } |
| 88 | return e.probe.op, probeResultOf(e.probe.admitted), true |
| 89 | } |
| 90 | |
| 91 | func probeResultOf(o *hostclaim.Observed) ProbeResult { |
| 92 | h := o.Claim.Holder |
| 93 | return ProbeResult{Session: h.Session, Stamp: h.Stamp, Generation: h.Generation} |
| 94 | } |
| 95 | |
| 96 | // Test seams: retireBetweenLocks runs in RetireExcept between the probe |
| 97 | // snapshot and the retirement pass; probeWakePublished runs in ProbeWake |
| 98 | // after the entry is published and before its wake starts. |
| 99 | var retireBetweenLocks, probeWakePublished, retireBeforeRegistryRead func() |
| 100 | |
| 101 | // probeEntry marks an actEntry ProbeWake made. |
| 102 | type probeEntry struct { |
| 103 | policy startPolicy |
| 104 | op, holdOp string |
| 105 | admitted *hostclaim.Observed |
| 106 | // parent is ProbeWake's context: the wake's own context derives from |
| 107 | // it, so a cancel is in force before the entry is published or its |
| 108 | // goroutine starts (fix round 1, finding 7). |
| 109 | parent context.Context |
| 110 | // phase is the probe's open (final review, ruling 3): held, opening |
| 111 | // once the target's open marks it (OpenProbeEntry, before it releases |
| 112 | // the hold), opened once the release is confirmed (ConfirmProbeOpened). |
| 113 | // Retirement protection holds from opening on; idle protection ends only |
| 114 | // at opened. |
| 115 | phase probePhase |
| 116 | // managed is set when the edge manages the opened probe's ready row: the |
| 117 | // entry is an ordinary serving one from then on. The probe metadata is |
| 118 | // kept, never cleared, so an older overlapping retirement pass still |
| 119 | // validates against it (ruling 2). |
| 120 | managed bool |
| 121 | // seq counts every change of phase or managed. A retirement pass keeps |
| 122 | // an entry whose seq moved since its snapshot: its registry read is |
| 123 | // older than the change. |
| 124 | seq uint64 |
| 125 | } |
| 126 | |
| 127 | // probePhase is a probe entry's open state. |
| 128 | type probePhase int |
| 129 | |
| 130 | const ( |
| 131 | probeHeldPhase probePhase = iota |
| 132 | probeOpening |
| 133 | probeOpened |
| 134 | ) |
| 135 | |
| 136 | // probeSnap is the immutable probe metadata a retirement pass snapshots |
| 137 | // under a.mu and reads outside it (ruling 2). |
| 138 | type probeSnap struct { |
| 139 | holdOp string |
| 140 | phase probePhase |
| 141 | seq uint64 |
| 142 | } |
| 143 | |
| 144 | // ConfirmProbeOpened records that the target's open released host's hold |
| 145 | // for op (final review, ruling 3): the entry, opening, is opened, and its |
| 146 | // idle protection ends. false when host has no opening probe entry of op. |
| 147 | func (a *Service) ConfirmProbeOpened(host, op string) bool { |
| 148 | a.mu.Lock() |
| 149 | defer a.mu.Unlock() |
| 150 | e := a.entries[host] |
| 151 | if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.managed || e.probe.phase < probeOpening { |
| 152 | return false |
| 153 | } |
| 154 | if e.probe.phase != probeOpened { |
| 155 | e.probe.phase = probeOpened |
| 156 | e.probe.seq++ |
| 157 | } |
| 158 | return true |
| 159 | } |
| 160 | |
| 161 | // OpenProbeEntry marks host's awake probe entry for op opening, before the |
| 162 | // target's open releases the row's hold (Plan 3 Task 7; final review, |
| 163 | // ruling 3). RetireExcept keeps it while its row is held or, once released, |
| 164 | // ready, until the edge manages it (the open race); it stays exempt from |
| 165 | // idle stop until ConfirmProbeOpened. A retried open marks it again. false |
| 166 | // when host has no awake probe entry of op: the next request then wakes the |
| 167 | // host under the commit. |
| 168 | func (a *Service) OpenProbeEntry(host, op string) bool { |
| 169 | a.mu.Lock() |
| 170 | defer a.mu.Unlock() |
| 171 | e := a.entries[host] |
| 172 | if e == nil || e.state != actAwake || e.probe == nil || e.probe.op != op || e.probe.admitted == nil || e.probe.managed { |
| 173 | return false |
| 174 | } |
| 175 | if e.probe.phase < probeOpening { |
| 176 | e.probe.phase = probeOpening |
| 177 | e.probe.seq++ |
| 178 | } |
| 179 | return true |
| 180 | } |
| 181 | |
| 182 | // ProbeWake runs one probe attempt for cfg.Host and waits for it. On |
| 183 | // success the host's entry is awake, holding the admitted session's process |
| 184 | // and replication, and stays held while the row is (RetireExcept). |
| 185 | func (a *Service) ProbeWake(ctx context.Context, cfg ProbeConfig) (ProbeResult, error) { |
| 186 | if ctx == nil || cfg.Host == "" || cfg.Op == "" || cfg.HoldOp == "" || cfg.Deadline.IsZero() { |
| 187 | return ProbeResult{}, hostwrite.ErrUnavailable |
| 188 | } |
| 189 | if !a.cfg.HostWrites.Enabled || a.cfg.Claims == nil || a.Routes == nil { |
| 190 | return ProbeResult{}, hostwrite.ErrUnavailable |
| 191 | } |
| 192 | // The cut deadline alone, before any admission: the window opens only |
| 193 | // at the first app start (brief step 4). |
| 194 | if !a.now().Before(cfg.Deadline) { |
| 195 | return ProbeResult{}, errors.Join(ErrProbeRefused, ErrAttemptOver) |
| 196 | } |
| 197 | rt, err := a.Routes.Get(cfg.Host) |
| 198 | if err != nil { |
| 199 | return ProbeResult{}, err |
| 200 | } |
| 201 | if rt == nil || !probeHeld(*rt, cfg.HoldOp) { |
| 202 | return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeRow) |
| 203 | } |
| 204 | e := &actEntry{host: cfg.Host, route: *rt, state: actWaking, ready: make(chan struct{}), |
| 205 | probe: &probeEntry{op: cfg.Op, holdOp: cfg.HoldOp, parent: ctx, |
| 206 | policy: startPolicy{MaxStarts: probeMaxStarts, Window: probeWindow30, Deadline: cfg.Deadline, Probe: true, Prepared: cfg.Prepared, Inputs: cfg.Inputs}}} |
| 207 | a.mu.Lock() |
| 208 | switch { |
| 209 | case a.fences[cfg.Host] != 0: |
| 210 | a.mu.Unlock() |
| 211 | return ProbeResult{}, errors.Join(ErrProbeRefused, ErrFenced) |
| 212 | case a.entries[cfg.Host] != nil: |
| 213 | a.mu.Unlock() |
| 214 | return ProbeResult{}, errors.Join(ErrProbeRefused, errProbeLive) |
| 215 | } |
| 216 | a.entries[cfg.Host] = e |
| 217 | a.mu.Unlock() |
| 218 | if probeWakePublished != nil { |
| 219 | probeWakePublished() |
| 220 | } |
| 221 | go a.wake(e) |
| 222 | <-e.ready |
| 223 | // e.err and e.probe are written before e.ready closes. |
| 224 | if e.err != nil { |
| 225 | return ProbeResult{}, e.err |
| 226 | } |
| 227 | return probeResultOf(e.probe.admitted), nil |
| 228 | } |
| 229 | |
| 230 | var ( |
| 231 | errProbeRow = errors.New("activator: the registry row is not held by the probe's hold op") |
| 232 | errProbeOp = errors.New("activator: the claim is not the probe's op") |
| 233 | errProbeLive = errors.New("activator: the host already has a live entry") |
| 234 | ) |
| 235 | |
| 236 | func probeHeld(rt registry.Route, holdOp string) bool { |
| 237 | return registry.IsHold(rt.Status) && rt.HoldOp == holdOp |
| 238 | } |
| 239 | |
| 240 | // probeRefused marks a probe's refusal before admission; other wakes' |
| 241 | // errors are unchanged. |
| 242 | func probeRefused(e *actEntry, err error) error { |
| 243 | if e.probe == nil || err == nil { |
| 244 | return err |
| 245 | } |
| 246 | return errors.Join(ErrProbeRefused, err) |
| 247 | } |
| 248 | |
| 249 | // heldProbes snapshots, under a.mu, the immutable metadata of every probe |
| 250 | // entry RetireExcept may retire (seen), then reads their registry rows |
| 251 | // outside the lock from the snapshot alone, never dereferencing e.probe |
| 252 | // there (final review, ruling 2). held is the entries it keeps beyond keep: |
| 253 | // those whose row is still held by the probe's hold op, and opening or |
| 254 | // opened ones whose row the open made ready, which the edge's next reload |
| 255 | // manages (the open race). The edge manages only ready rows (edge.Reload), |
| 256 | // so a held probe is never in keep. A row that cannot be read keeps its |
| 257 | // entry for this reload. RetireExcept re-validates each entry's seq under |
| 258 | // its final lock: one that moved since the snapshot is kept. |
| 259 | func (a *Service) heldProbes(keep map[string]bool) (seen map[*actEntry]probeSnap, held map[*actEntry]bool) { |
| 260 | seen, held = map[*actEntry]probeSnap{}, map[*actEntry]bool{} |
| 261 | a.mu.Lock() |
| 262 | for _, e := range a.entries { |
| 263 | if e.probe != nil && !e.probe.managed && !keep[e.host] { |
| 264 | seen[e] = probeSnap{holdOp: e.probe.holdOp, phase: e.probe.phase, seq: e.probe.seq} |
| 265 | } |
| 266 | } |
| 267 | a.mu.Unlock() |
| 268 | if retireBeforeRegistryRead != nil { |
| 269 | retireBeforeRegistryRead() |
| 270 | } |
| 271 | if a.Routes == nil { |
| 272 | return seen, held |
| 273 | } |
| 274 | for e, snap := range seen { |
| 275 | rt, err := a.Routes.Get(e.host) |
| 276 | if err != nil { |
| 277 | log.Printf("activator: %s: the probe's row could not be read (%v); kept for this reload", e.host, err) |
| 278 | held[e] = true |
| 279 | continue |
| 280 | } |
| 281 | switch { |
| 282 | case rt == nil: |
| 283 | case probeHeld(*rt, snap.holdOp): |
| 284 | held[e] = true |
| 285 | case snap.phase >= probeOpening && rt.Status == "ready" && rt.HoldOp == "": |
| 286 | held[e] = true |
| 287 | } |
| 288 | } |
| 289 | return seen, held |
| 290 | } |
| 291 | |
amadan