| 1 | package activator |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "log" |
| 7 | "maps" |
| 8 | "reflect" |
| 9 | "time" |
| 10 | |
| 11 | "github.com/carlosframework/platform/internal/hostbox" |
| 12 | "github.com/carlosframework/platform/internal/hostclaim" |
| 13 | "github.com/carlosframework/platform/internal/hostwrite" |
| 14 | "github.com/carlosframework/platform/internal/registry" |
| 15 | ) |
| 16 | |
| 17 | // The source's recovery probe entry (claim Plan 10 J2; moves §7.8 source |
| 18 | // probing). The host's recovery probe adapter calls RecoveryProbe once the |
| 19 | // record and the claim say probing with this box executing. |
| 20 | // |
| 21 | // One call is one claim wake of rt under the caller's probe policy, through |
| 22 | // the same path as every claim wake: K6 (a foreign executor is stopped and |
| 23 | // refused before anything else), the queued incarnation, WakePhase with |
| 24 | // probe set, then ClaimRuntime's Prerequisites, Admit and Prepare and |
| 25 | // startAdmitted's start loop. The policy, its Probe and its exact Inputs, |
| 26 | // reaches Prerequisites, Admit and Prepare unchanged: the host's |
| 27 | // admitsOperation, the moves Inputs owner and the attempt journal decide |
| 28 | // what is admitted, rendered and counted. Only a reverting or restoring op |
| 29 | // at probing is probed here; a landing probe is not this entry's. |
| 30 | // |
| 31 | // Lifecycle. The probe owns the host's entry exactly as a wake does: one per |
| 32 | // host. An identical call joins the probe in flight (or the healthy one it |
| 33 | // left) and shares its outcome; any other entry, or a different policy, is |
| 34 | // refused with hostbox.ErrChanged. Requests and sidecar requests never join |
| 35 | // a probe and are refused with ErrFenced, as under a fence. |
| 36 | // |
| 37 | // Fences. The rollback holds its cut fence on this host through the whole |
| 38 | // recovery, and the claim's exact probing executor, not the fence, is the |
| 39 | // authority this entry starts under. The adapter supplies its own cut fence |
| 40 | // generation in RecoveryFence; an unrelated installed fence refuses the wake. |
| 41 | // The probe runs with its own fence installed |
| 42 | // and the fence keeps everything else out. Installing or replacing a fence |
| 43 | // stops the probe. If the moves worker already lifted its own fence, the |
| 44 | // claim must still confirm the exact opened session before the entry serves. |
| 45 | // |
| 46 | // A healthy start (OperationAttempts.Healthy accepted it) is retained: the |
| 47 | // entry is awake with the started process and its replication, held. While |
| 48 | // held it serves nothing, has no idle clock or convergence and ignores |
| 49 | // Bounce; a lost lease or an exit still stands it down. The sweep reads the |
| 50 | // claim for it (reviewRecoveryHold): once the source opened on the same |
| 51 | // session (or the op released to serving keeping it) the hold ends and the |
| 52 | // owning cut fence is released and the entry serves as any awake entry; |
| 53 | // once the claim names another session or |
| 54 | // executor, the entry is stopped. |
| 55 | func (a *Service) RecoveryProbe(ctx context.Context, rt registry.Route, p StartPolicy) error { |
| 56 | if a == nil || ctx == nil { |
| 57 | return hostwrite.ErrUnavailable |
| 58 | } |
| 59 | // Every dependency of the run is checked before the first read, stop or |
| 60 | // write. |
| 61 | if !a.RecoveryProbeReady() { |
| 62 | return hostwrite.ErrUnavailable |
| 63 | } |
| 64 | if rt.Host == "" || !recoveryProbePolicy(p) { |
| 65 | return errRecoveryProbePolicy |
| 66 | } |
| 67 | a.mu.Lock() |
| 68 | if a.shuttingDown { |
| 69 | a.mu.Unlock() |
| 70 | return ErrFenced |
| 71 | } |
| 72 | if fence := a.fences[rt.Host]; fence != 0 && fence != p.RecoveryFence { |
| 73 | a.mu.Unlock() |
| 74 | return ErrFenced |
| 75 | } |
| 76 | if e := a.entries[rt.Host]; e != nil { |
| 77 | a.mu.Unlock() |
| 78 | return a.joinRecoveryProbe(ctx, e, p) |
| 79 | } |
| 80 | ctx, cancel := context.WithTimeout(ctx, recoveryProbeLimit) |
| 81 | defer cancel() |
| 82 | policy := p |
| 83 | inputs := *p.Inputs |
| 84 | inputs.BundleETags = maps.Clone(p.Inputs.BundleETags) |
| 85 | policy.Inputs = &inputs |
| 86 | e := &actEntry{host: rt.Host, route: rt, state: actWaking, ready: make(chan struct{}), recoveryPolicy: &policy, probeHeld: true, wakeCancel: cancel} |
| 87 | if a.entries == nil { |
| 88 | a.entries = make(map[string]*actEntry) |
| 89 | } |
| 90 | a.entries[rt.Host] = e |
| 91 | fence := a.fences[rt.Host] |
| 92 | a.mu.Unlock() |
| 93 | if a.lifeCtx != nil { |
| 94 | stop := context.AfterFunc(a.lifeCtx, cancel) |
| 95 | defer stop() |
| 96 | } |
| 97 | w, err := a.doClaimWake(ctx, e) |
| 98 | return a.finishRecoveryProbe(e, fence, w, err) |
| 99 | } |
| 100 | |
| 101 | // RecoveryProbeReady checks wiring only. The host calls it before recording |
| 102 | // inputs; admission still validates the exact claim and session on every run. |
| 103 | func (a *Service) RecoveryProbeReady() bool { |
| 104 | if a == nil || !a.cfg.HostWrites.Enabled { |
| 105 | return false |
| 106 | } |
| 107 | r := a.cfg.Claims |
| 108 | return r != nil && r.RecoveryAttemptsReady != nil && r.RecoveryAttemptsReady() && r.Box != "" && r.Read != nil && r.StopForeign != nil && r.Prerequisites != nil && r.Admit != nil && r.Prepare != nil && r.FailedAdmission != nil |
| 109 | } |
| 110 | |
| 111 | // recoveryProbeLimit bounds one probe call, as wake bounds a wake. |
| 112 | const recoveryProbeLimit = 5 * time.Minute |
| 113 | |
| 114 | // errRecoveryProbePolicy is a call that is not a recovery probe: no host, an |
| 115 | // ordinary or prepared policy, no inputs, or starts and window outside the |
| 116 | // start loop's bounds. A caller bug, refused before anything runs. |
| 117 | var errRecoveryProbePolicy = fmt.Errorf("%w: not a recovery probe policy", hostbox.ErrUnsupported) |
| 118 | |
| 119 | func recoveryProbePolicy(p StartPolicy) bool { |
| 120 | return p.Probe && p.Inputs != nil && p.Prepared == nil && p.MaxStarts >= 1 && p.MaxStarts <= 5 && p.Window > 0 && p.Window <= 30*time.Second |
| 121 | } |
| 122 | |
| 123 | // recoveryProbeKind reports whether kind is the source's recovery, the only |
| 124 | // op this entry probes. |
| 125 | func recoveryProbeKind(kind string) bool { |
| 126 | return kind == hostclaim.KindRestoring || kind == hostclaim.KindReverting |
| 127 | } |
| 128 | |
| 129 | func sameRecoveryProbe(x, y StartPolicy) bool { |
| 130 | return x.RecoveryFence == y.RecoveryFence && x.MaxStarts == y.MaxStarts && x.Window == y.Window && x.Deadline.Equal(y.Deadline) && x.Probe == y.Probe && |
| 131 | x.Prepared == nil && y.Prepared == nil && x.Inputs != nil && y.Inputs != nil && reflect.DeepEqual(*x.Inputs, *y.Inputs) |
| 132 | } |
| 133 | |
| 134 | // joinRecoveryProbe answers a call that found host's entry already present. |
| 135 | // Only a probe entry under the same policy is joined: a probe in flight |
| 136 | // shares its outcome, and a retained healthy probe answers once its process, |
| 137 | // its replication and the claim's session still hold. A failed or superseded |
| 138 | // held probe is retired so the next pass can spend its remaining attempts. |
| 139 | func (a *Service) joinRecoveryProbe(ctx context.Context, e *actEntry, p StartPolicy) error { |
| 140 | for { |
| 141 | a.mu.Lock() |
| 142 | if a.entries[e.host] != e || e.recoveryPolicy == nil { |
| 143 | a.mu.Unlock() |
| 144 | return hostbox.ErrChanged |
| 145 | } |
| 146 | if !sameRecoveryProbe(*e.recoveryPolicy, p) { |
| 147 | // A new input selection must not strand an older held probe. |
| 148 | // Cancel a pending start, including a late healthy result, or |
| 149 | // retire the held process before the next pass can start anew. |
| 150 | if e.probeHeld { |
| 151 | e.probeObsolete = true |
| 152 | if e.state == actWaking && e.wakeCancel != nil { |
| 153 | e.wakeCancel() |
| 154 | } |
| 155 | if e.state == actAwake { |
| 156 | a.beginStopLocked(e, StopFenced) |
| 157 | } |
| 158 | } |
| 159 | a.mu.Unlock() |
| 160 | return hostbox.ErrChanged |
| 161 | } |
| 162 | switch e.state { |
| 163 | case actWaking: |
| 164 | ready := e.ready |
| 165 | a.mu.Unlock() |
| 166 | select { |
| 167 | case <-ready: |
| 168 | if e.err != nil { |
| 169 | return e.err |
| 170 | } |
| 171 | continue |
| 172 | case <-ctx.Done(): |
| 173 | return ctx.Err() |
| 174 | } |
| 175 | case actAwake: |
| 176 | if a.retireRecoveryProbeLocked(e) { |
| 177 | a.mu.Unlock() |
| 178 | return hostbox.ErrChanged |
| 179 | } |
| 180 | admitted := e.probeAdmitted |
| 181 | a.mu.Unlock() |
| 182 | r := a.cfg.Claims |
| 183 | fresh, err := r.Read(ctx, e.host) |
| 184 | if err != nil { |
| 185 | return err |
| 186 | } |
| 187 | a.mu.Lock() |
| 188 | defer a.mu.Unlock() |
| 189 | if a.entries[e.host] != e || e.state != actAwake || a.retireRecoveryProbeLocked(e) { |
| 190 | return hostbox.ErrChanged |
| 191 | } |
| 192 | _, err = recoveryHoldVerdict(admitted, fresh, r.Box) |
| 193 | if err != nil && e.probeHeld { |
| 194 | a.beginStopLocked(e, StopFenced) |
| 195 | } |
| 196 | return err |
| 197 | default: |
| 198 | a.mu.Unlock() |
| 199 | return hostbox.ErrChanged |
| 200 | } |
| 201 | } |
| 202 | } |
| 203 | |
| 204 | // finishRecoveryProbe resolves the probe entry: a failed probe is forgotten |
| 205 | // as a failed wake is; a healthy one is retained, held, unless the fence it |
| 206 | // started under was replaced, when it stops and answers ErrFenced. An owner |
| 207 | // lift is checked against the exact claim session by the hold review. |
| 208 | func (a *Service) finishRecoveryProbe(e *actEntry, fence uint64, w *wokenInstance, err error) error { |
| 209 | a.mu.Lock() |
| 210 | if err == nil && (w == nil || w.proc == nil || w.rep == nil) { |
| 211 | err = hostwrite.ErrUnavailable |
| 212 | } |
| 213 | if err != nil { |
| 214 | e.err = err |
| 215 | outcome := &stopObservation{uncertain: true} |
| 216 | if w != nil && w.stopOutcome != nil { |
| 217 | outcome = w.stopOutcome |
| 218 | } |
| 219 | a.rememberStopLocked(e.host, outcome) |
| 220 | if a.entries[e.host] == e { |
| 221 | delete(a.entries, e.host) |
| 222 | } |
| 223 | a.mu.Unlock() |
| 224 | close(e.ready) |
| 225 | return err |
| 226 | } |
| 227 | now := a.now() |
| 228 | e.state = actAwake |
| 229 | e.probeFence = fence |
| 230 | e.rep, e.proc = w.rep, w.proc |
| 231 | e.spawnVersion = w.spawnVersion |
| 232 | // Retain the launcher's socket; legacy claim launchers use the same |
| 233 | // canonical fallback as an ordinary wake. Publish only when opened. |
| 234 | e.gen = w.gen |
| 235 | if e.gen.N == 0 && !e.worker { |
| 236 | e.gen = canonicalGen(e.route, nextGen(), w.spawnVersion) |
| 237 | } |
| 238 | e.lastActive, e.spawnedAt = now, now |
| 239 | e.bounceAfterWake = false |
| 240 | if e.probeObsolete { |
| 241 | e.err = hostbox.ErrChanged |
| 242 | a.beginStopLocked(e, StopFenced) |
| 243 | } else if a.shuttingDown || a.recoveryFenceChangedLocked(e) { |
| 244 | log.Printf("activator: %s recovery probe healthy, but its fence changed; stopping it", e.host) |
| 245 | e.err = ErrFenced |
| 246 | why := StopFenced |
| 247 | if a.shuttingDown { |
| 248 | why = StopShutdown |
| 249 | } |
| 250 | a.beginStopLocked(e, why) |
| 251 | } else { |
| 252 | log.Printf("activator: %s recovery probe healthy; held until the source opens", e.host) |
| 253 | } |
| 254 | err = e.err |
| 255 | a.mu.Unlock() |
| 256 | close(e.ready) |
| 257 | return err |
| 258 | } |
| 259 | |
| 260 | // recoveryHoldVerdict compares the claim read now with the admission a held |
| 261 | // probe started under. The same writer (incarnation, stamp, session, |
| 262 | // generation) on the same op and executor at probing or probed keeps the |
| 263 | // hold; at opened, or released to serving with this box still the writer, |
| 264 | // the hold ends (open). Anything else is hostbox.ErrChanged: the probe's |
| 265 | // session is over. |
| 266 | func recoveryHoldVerdict(admitted, fresh hostclaim.Observed, box string) (open bool, err error) { |
| 267 | x, y := admitted.Claim, fresh.Claim |
| 268 | if fresh.Absent || fresh.Host != admitted.Host || box == "" || x == nil || y == nil || x.Holder == nil || y.Holder == nil { |
| 269 | return false, hostbox.ErrChanged |
| 270 | } |
| 271 | for _, lost := range y.Lost { |
| 272 | if lost == box { |
| 273 | return false, hostbox.ErrChanged |
| 274 | } |
| 275 | } |
| 276 | h, g := x.Holder, y.Holder |
| 277 | if h.Incarnation != g.Incarnation || h.Stamp != g.Stamp || h.Session != g.Session || h.Generation != g.Generation { |
| 278 | return false, hostbox.ErrChanged |
| 279 | } |
| 280 | if y.Op.Kind == hostclaim.KindServing && y.Op.ID == "" && y.Op.Phase == "" && g.WriterBox == box && (y.Op.Exec == "" || y.Op.Exec == box) { |
| 281 | return true, nil |
| 282 | } |
| 283 | if y.Op.ID != x.Op.ID || y.Op.Kind != x.Op.Kind || x.Op.Exec != box || y.Op.Exec != box || !hostclaim.SessionReaches(y.Op.Kind, x.Op.Phase, y.Op.Phase) { |
| 284 | return false, hostbox.ErrChanged |
| 285 | } |
| 286 | return y.Op.Phase == hostclaim.PhaseOpened, nil |
| 287 | } |
| 288 | |
| 289 | // reviewRecoveryHold reads the claim for a held probe entry (off a.mu) and |
| 290 | // opens or stops it per recoveryHoldVerdict. An unreadable claim changes |
| 291 | // nothing; the next sweep reads again. |
| 292 | func (a *Service) reviewRecoveryHold(e *actEntry) { |
| 293 | r := a.cfg.Claims |
| 294 | a.mu.Lock() |
| 295 | admitted := e.probeAdmitted |
| 296 | a.mu.Unlock() |
| 297 | var fresh hostclaim.Observed |
| 298 | err := hostwrite.ErrUnavailable |
| 299 | if r != nil && r.Read != nil { |
| 300 | parent := a.lifeCtx |
| 301 | if parent == nil { |
| 302 | parent = context.Background() |
| 303 | } |
| 304 | ctx, cancel := context.WithTimeout(parent, 10*time.Second) |
| 305 | fresh, err = r.Read(ctx, e.host) |
| 306 | cancel() |
| 307 | } |
| 308 | a.mu.Lock() |
| 309 | defer a.mu.Unlock() |
| 310 | e.probeReviewing = false |
| 311 | if a.entries[e.host] != e || e.state != actAwake || !e.probeHeld { |
| 312 | return |
| 313 | } |
| 314 | // Recheck after the off-lock read: a failed child, lost lease or a |
| 315 | // replacement fence must not be opened by a late claim answer. |
| 316 | if a.retireRecoveryProbeLocked(e) || err != nil { |
| 317 | return |
| 318 | } |
| 319 | open, err := recoveryHoldVerdict(admitted, fresh, r.Box) |
| 320 | switch { |
| 321 | case err != nil: |
| 322 | log.Printf("activator: %s recovery probe's session is over (%v); stopping it", e.host, err) |
| 323 | a.beginStopLocked(e, StopFenced) |
| 324 | case open: |
| 325 | // The entry retains both the observed fence and the adapter's |
| 326 | // owning generation. Check and remove under the same lock as the |
| 327 | // hold transition; never lift a replacement operation's fence. |
| 328 | // A lift clears an owned fence's owner with it (FenceAndStopAs), |
| 329 | // so no owner outlives its fence. |
| 330 | if e.recoveryPolicy.RecoveryFence != 0 && a.fences[e.host] == e.recoveryPolicy.RecoveryFence { |
| 331 | delete(a.fences, e.host) |
| 332 | delete(a.fenceOwners, e.host) |
| 333 | } |
| 334 | log.Printf("activator: %s source opened on the recovery probe's session; serving", e.host) |
| 335 | if e.gen.N != 0 { |
| 336 | a.noteCurrent(e.host, e.gen.N) |
| 337 | } |
| 338 | if a.gens && linksCanonical(e.route, e.gen) { |
| 339 | if err := swapCanonicalLink(e.route.Socket, e.gen.Socket); err != nil { |
| 340 | log.Printf("activator: %s: link %s to %s: %v (processes dialling the canonical path will not reach it)", e.host, e.route.Socket, e.gen.Socket, err) |
| 341 | } |
| 342 | } |
| 343 | e.probeHeld = false |
| 344 | e.lastActive = a.now() |
| 345 | } |
| 346 | } |
| 347 | |
| 348 | // retireRecoveryProbeLocked is shared by the real sweep and callers joining |
| 349 | // a retained probe. Retirement leaves the cut fence in place for the retry. |
| 350 | // The caller holds a.mu and has verified that e is the current awake entry. |
| 351 | func (a *Service) retireRecoveryProbeLocked(e *actEntry) bool { |
| 352 | fenceChanged := e.probeHeld && a.recoveryFenceChangedLocked(e) |
| 353 | if e.probeObsolete || fenceChanged || e.proc == nil || e.proc.Exited() || e.rep == nil || e.rep.Err() != nil { |
| 354 | why := StopFenced |
| 355 | switch { |
| 356 | case e.probeObsolete || fenceChanged: |
| 357 | case e.rep == nil || e.rep.Err() != nil: |
| 358 | why = StopLeaseLost |
| 359 | default: |
| 360 | why = StopExited |
| 361 | } |
| 362 | a.beginStopLocked(e, why) |
| 363 | return true |
| 364 | } |
| 365 | return false |
| 366 | } |
| 367 | |
| 368 | // A moves worker may have closed the rollback and lifted its own fence |
| 369 | // before the activator reads opened/serving. Defer that case to the claim |
| 370 | // verdict, keeping probeHeld in force while the read is outstanding. |
| 371 | func (a *Service) recoveryFenceChangedLocked(e *actEntry) bool { |
| 372 | fence := a.fences[e.host] |
| 373 | if fence == 0 && e.recoveryPolicy != nil && e.probeFence == e.recoveryPolicy.RecoveryFence { |
| 374 | return false |
| 375 | } |
| 376 | return fence != e.probeFence || (fence != 0 && (e.recoveryPolicy == nil || fence != e.recoveryPolicy.RecoveryFence)) |
| 377 | } |
| 378 | |
amadan