| 1 | package host |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "encoding/json" |
| 6 | "errors" |
| 7 | "fmt" |
| 8 | "log" |
| 9 | "strconv" |
| 10 | "strings" |
| 11 | "time" |
| 12 | |
| 13 | "github.com/carlosframework/platform/internal/activator" |
| 14 | "github.com/carlosframework/platform/internal/hostclaim" |
| 15 | "github.com/carlosframework/platform/internal/moves" |
| 16 | "github.com/carlosframework/platform/internal/provision" |
| 17 | ) |
| 18 | |
| 19 | // The landing's probing phase (moves spec §7.6, §7.7, §7.13; Plan 3 Task 6). |
| 20 | // |
| 21 | // restored to probing is the landing's step (movesland.go). At probing the |
| 22 | // pass makes the probe attempts T11 admitted (H10: the record's |
| 23 | // CutAttempts, written by MoveAdmit), at most moves.AttemptsFor("landing", |
| 24 | // n). Each attempt is counted on the record (provision.MoveProbeAttempt) |
| 25 | // and journalled (Journal.Attempt) before its call, then made through the |
| 26 | // probe wake (Task 5's activator.Service.ProbeWake behind moveCfg.Probe): |
| 27 | // one cold admission into the prepared generation (or a fresh one for a |
| 28 | // later attempt), restored, rendered, published and started by the claim |
| 29 | // (Group D), up to five starts inside the attempt's 30 s window, bounded by |
| 30 | // the landing's deadline (H8). The attempt's own journal steps |
| 31 | // (probe_opened_<k>, probe_start_<k>) are Task 5's movesProbeAttempts'; |
| 32 | // probe_failed_<k> is this file's, written when the attempt is over. |
| 33 | // |
| 34 | // Outcomes. A pass: probing to probed, record then claim (the claim's step |
| 35 | // keeps the admitted writer session, H3); the commit is Task 7's. Task 5's |
| 36 | // ProbeInputsChanged: its recorded sentence, the configuration rollback. An |
| 37 | // attempt that is over (ErrProbeFailed, ErrAttemptOver): the next one, if |
| 38 | // any; none left: the attempts-exhausted rollback; the deadline passed: the |
| 39 | // deadline's. Rollbacks are marks (MoveMarkRollback); the one conversion to |
| 40 | // MoveRollBack is the pass's rollback decision (movesrollback.go). Any other |
| 41 | // error ends the pass and the same attempt is made again on the next. |
| 42 | // |
| 43 | // A crash restarts only the probe (§7.13): local and restored are never run |
| 44 | // again; a fresh worker finds probing and the counted attempt. One cold |
| 45 | // admission per budgeted attempt (§7.7, fix round 1, ruling 2): an attempt |
| 46 | // is admitted once Task 5's journalled session binding (probe_session_<s> = |
| 47 | // k) exists for it, and an admitted attempt is consumed on restart, failed |
| 48 | // unless it passed. Only a counted attempt with no binding is made again. |
| 49 | // Fix round 2: the claim is the source of truth for the admission itself. |
| 50 | // probe_admitting_<k> records the claim's writer before the call; a claim |
| 51 | // whose writer moved on since admitted under k though Reserve's binding |
| 52 | // never became durable. Before an admitted attempt with no probe_passed_<k> |
| 53 | // is consumed, the activator's live entry is asked: this move's, awake, |
| 54 | // with the claim's writer, it passed, and its pass is journalled then. |
| 55 | // |
| 56 | // A pass is journalled (probe_passed_<k>: the admitted session, stamp and |
| 57 | // generation) before the probing to probed record write (ruling 1). A pass |
| 58 | // that finds it does not probe again: it checks the claim's holder still |
| 59 | // carries that writer and the activator's live probe entry is this move's, |
| 60 | // then retries only the phase writes, record then claim. A writer that is |
| 61 | // gone consumes the attempt as failed. |
| 62 | // |
| 63 | // Unarmed in production: moveCfg.Probe is nil until claim Plan 11's launcher |
| 64 | // (R10) and Plan 5 Task 9's arming, so the landing waits at restored. |
| 65 | |
| 66 | // probeStepFailed marks attempt k over in the op journal; probeStepPassed |
| 67 | // names attempt k's passing writer (probePassed). |
| 68 | const ( |
| 69 | probeStepFailed = "probe_failed_" |
| 70 | probeStepPassed = "probe_passed_" |
| 71 | probeStepAdmitting = "probe_admitting_" |
| 72 | ) |
| 73 | |
| 74 | // probeAdmitting is a probe_admitting_<k> step's value, written before the |
| 75 | // call: the claim's writer session and admission count the attempt started |
| 76 | // from. The claim is the source of truth for whether the attempt admitted |
| 77 | // (fix round 2, ruling 1). |
| 78 | type probeAdmitting struct { |
| 79 | Session uint64 `json:"session"` |
| 80 | Admissions int `json:"admissions"` |
| 81 | } |
| 82 | |
| 83 | // probePassed is a probe_passed_<k> step's value: the admitted writer the |
| 84 | // passing attempt left serving. |
| 85 | type probePassed struct { |
| 86 | Session uint64 `json:"session"` |
| 87 | Stamp uint64 `json:"stamp"` |
| 88 | Generation string `json:"generation"` |
| 89 | } |
| 90 | |
| 91 | // probeWindowLen is the attempt's window, counted from its first start. |
| 92 | var probeWindowLen = time.Duration(moves.ProbeSeconds * float64(time.Second)) |
| 93 | |
| 94 | // MovesRecordAdmittedAttempts is Task 5's Admitted hook (H10): the probe |
| 95 | // attempts the record's admission priced, refused when it recorded none. |
| 96 | func MovesRecordAdmittedAttempts(_ context.Context, rec *provision.Record) (int, error) { |
| 97 | if rec == nil || rec.Move == nil || rec.Move.CutAttempts < 1 { |
| 98 | return 0, errors.New("moves: " + provision.RollbackProbeUnpriced) |
| 99 | } |
| 100 | return rec.Move.CutAttempts, nil |
| 101 | } |
| 102 | |
| 103 | // attemptSpent reports whether counted attempt k is over: its failure is |
| 104 | // journalled, it was admitted (attemptAdmittedBy), or its window opened and |
| 105 | // has passed. |
| 106 | func (l *landingRun) attemptSpent(j *moves.Journal, k int, h *hostclaim.Holder) bool { |
| 107 | if j.Done(probeStepFailed+strconv.Itoa(k)) || l.attemptAdmittedBy(j, k, h) { |
| 108 | return true |
| 109 | } |
| 110 | raw, ok := j.Steps[probeStepOpened+strconv.Itoa(k)] |
| 111 | if !ok { |
| 112 | return false |
| 113 | } |
| 114 | opened, err := time.Parse(time.RFC3339Nano, raw) |
| 115 | return err == nil && !l.cfg().Now().Before(opened.Add(probeWindowLen)) |
| 116 | } |
| 117 | |
| 118 | // attemptAdmitted reports whether a session is journalled as bound to |
| 119 | // record attempt k (movesProbeAttempts.bindSession). |
| 120 | func attemptAdmitted(j *moves.Journal, k int) bool { |
| 121 | want := strconv.Itoa(k) |
| 122 | for step, v := range j.Steps { |
| 123 | if strings.HasPrefix(step, probeStepSession) && v == want { |
| 124 | return true |
| 125 | } |
| 126 | } |
| 127 | return false |
| 128 | } |
| 129 | |
| 130 | // attemptAdmittedBy reports whether attempt k admitted a writer: Task 5's |
| 131 | // binding exists for it, or probe_admitting_<k> was written and the claim |
| 132 | // has since moved on (another writer session, or a newer cold admission on |
| 133 | // this box). probe_admitting_<k> with the claim unchanged is an attempt that |
| 134 | // never admitted, and only that may reuse its index. |
| 135 | func (l *landingRun) attemptAdmittedBy(j *moves.Journal, k int, h *hostclaim.Holder) bool { |
| 136 | if attemptAdmitted(j, k) { |
| 137 | return true |
| 138 | } |
| 139 | raw, ok := j.Steps[probeStepAdmitting+strconv.Itoa(k)] |
| 140 | if !ok || h == nil { |
| 141 | return false |
| 142 | } |
| 143 | var a probeAdmitting |
| 144 | if err := json.Unmarshal([]byte(raw), &a); err != nil { |
| 145 | // Unreadable: never risk a second admission under k. |
| 146 | return true |
| 147 | } |
| 148 | if h.Session != a.Session { |
| 149 | return true |
| 150 | } |
| 151 | if len(h.Admissions) > a.Admissions { |
| 152 | last := h.Admissions[len(h.Admissions)-1] |
| 153 | return last.Box == l.cfg().Box && last.Mode == "cold" |
| 154 | } |
| 155 | return false |
| 156 | } |
| 157 | |
| 158 | // livePass reports the activator's live probe entry when it is this move's |
| 159 | // and its writer is the claim's current one: an attempt that passed though |
| 160 | // its probe_passed_<k> never became durable (fix round 2, ruling 2). |
| 161 | func (l *landingRun) livePass(h *hostclaim.Holder) (activator.ProbeResult, bool) { |
| 162 | live := l.cfg().ProbeLive |
| 163 | if live == nil || h == nil || h.WriterBox != l.cfg().Box { |
| 164 | return activator.ProbeResult{}, false |
| 165 | } |
| 166 | op, res, ok := live(l.rec.Host) |
| 167 | if !ok || op != l.g.ID || res.Session != h.Session || res.Stamp != h.Stamp || res.Generation != h.Generation { |
| 168 | return activator.ProbeResult{}, false |
| 169 | } |
| 170 | return res, true |
| 171 | } |
| 172 | |
| 173 | // journalPassed writes probe_passed_<k> for the passing writer res. |
| 174 | func (l *landingRun) journalPassed(k int, res activator.ProbeResult) error { |
| 175 | if err := l.fault("before-journal-passed"); err != nil { |
| 176 | return err |
| 177 | } |
| 178 | raw, err := json.Marshal(probePassed{Session: res.Session, Stamp: res.Stamp, Generation: res.Generation}) |
| 179 | if err != nil { |
| 180 | return err |
| 181 | } |
| 182 | j, err := l.journal() |
| 183 | if err != nil { |
| 184 | return err |
| 185 | } |
| 186 | j.Mark(probeStepPassed+strconv.Itoa(k), string(raw)) |
| 187 | return j.Commit(l.cfg().DataDir) |
| 188 | } |
| 189 | |
| 190 | // failAttempt journals attempt k over. |
| 191 | func (l *landingRun) failAttempt(k int) error { |
| 192 | j, err := l.journal() |
| 193 | if err != nil { |
| 194 | return err |
| 195 | } |
| 196 | if j.Done(probeStepFailed + strconv.Itoa(k)) { |
| 197 | return nil |
| 198 | } |
| 199 | j.Mark(probeStepFailed+strconv.Itoa(k), l.cfg().Now().UTC().Format(time.RFC3339Nano)) |
| 200 | return j.Commit(l.cfg().DataDir) |
| 201 | } |
| 202 | |
| 203 | // passed writes the phase steps of a passing attempt: record, then claim. |
| 204 | func (l *landingRun) passed(ctx context.Context) (bool, error) { |
| 205 | if err := l.fault("before-record-probed"); err != nil { |
| 206 | return false, err |
| 207 | } |
| 208 | return l.advance(ctx, provision.LandProbing, provision.LandProbed, nil) |
| 209 | } |
| 210 | |
| 211 | // passedWriterLives checks a journalled pass against the claim's holder and |
| 212 | // the activator's live probe entry. |
| 213 | func (l *landingRun) passedWriterLives(obs hostclaim.Observed, raw string) bool { |
| 214 | var p probePassed |
| 215 | if err := json.Unmarshal([]byte(raw), &p); err != nil || p.Session == 0 { |
| 216 | return false |
| 217 | } |
| 218 | h := obs.Claim.Holder |
| 219 | if h == nil || h.Session != p.Session || h.Stamp != p.Stamp || h.Generation != p.Generation || h.WriterBox != l.cfg().Box { |
| 220 | return false |
| 221 | } |
| 222 | live := l.cfg().ProbeLive |
| 223 | if live == nil { |
| 224 | return false |
| 225 | } |
| 226 | op, res, ok := live(l.rec.Host) |
| 227 | return ok && op == l.g.ID && res.Session == p.Session && res.Stamp == p.Stamp && res.Generation == p.Generation |
| 228 | } |
| 229 | |
| 230 | // probing makes the landing's probe attempts (see the file comment). |
| 231 | func (l *landingRun) probing(ctx context.Context) (bool, error) { |
| 232 | cfg := l.cfg() |
| 233 | obs, err := l.syncClaimPhase(ctx) |
| 234 | if err != nil { |
| 235 | return false, err |
| 236 | } |
| 237 | if err := l.requirePhase(obs, provision.LandProbing); err != nil { |
| 238 | return false, err |
| 239 | } |
| 240 | if cfg.Probe == nil { |
| 241 | l.w.notice(l.rec.Host+"#probe", "moves: %s: landing %s waits at probing: this box has no probe wake", l.rec.Host, l.g.ID) |
| 242 | return false, nil |
| 243 | } |
| 244 | m := l.rec.Move |
| 245 | if m.CutAttempts < 1 { |
| 246 | return l.markRollback(ctx, provision.RollbackProbeUnpriced) |
| 247 | } |
| 248 | n := moves.AttemptsFor(moves.ProbeLanding, m.CutAttempts) |
| 249 | j, err := l.journal() |
| 250 | if err != nil { |
| 251 | return false, err |
| 252 | } |
| 253 | raw, ok := j.Steps[landStepInstall] |
| 254 | if !ok { |
| 255 | return false, fmt.Errorf("move %s: probing without an install manifest in the journal", l.g.ID) |
| 256 | } |
| 257 | var manifest moves.InstallManifest |
| 258 | if err := json.Unmarshal([]byte(raw), &manifest); err != nil { |
| 259 | return false, fmt.Errorf("move %s: the journalled install manifest: %w", l.g.ID, err) |
| 260 | } |
| 261 | counted := 0 |
| 262 | if m.Landing != nil { |
| 263 | counted = m.Landing.Attempts |
| 264 | } |
| 265 | k := counted |
| 266 | if k > 0 { |
| 267 | if raw, ok := j.Steps[probeStepPassed+strconv.Itoa(k)]; ok { |
| 268 | if l.passedWriterLives(obs, raw) { |
| 269 | log.Printf("moves: %s: landing %s attempt %d passed; its phase writes are retried", l.rec.Host, l.g.ID, k) |
| 270 | return l.passed(ctx) |
| 271 | } |
| 272 | log.Printf("moves: %s: landing %s attempt %d passed, but its writer is gone; the attempt is consumed", l.rec.Host, l.g.ID, k) |
| 273 | if err := l.failAttempt(k); err != nil { |
| 274 | return false, err |
| 275 | } |
| 276 | k++ |
| 277 | } else if l.attemptSpent(j, k, obs.Claim.Holder) { |
| 278 | if !j.Done(probeStepFailed + strconv.Itoa(k)) { |
| 279 | if res, ok := l.livePass(obs.Claim.Holder); ok { |
| 280 | log.Printf("moves: %s: landing %s attempt %d passed (its live writer is the claim's); its phase writes follow", l.rec.Host, l.g.ID, k) |
| 281 | if err := l.journalPassed(k, res); err != nil { |
| 282 | return false, err |
| 283 | } |
| 284 | return l.passed(ctx) |
| 285 | } |
| 286 | } |
| 287 | if err := l.failAttempt(k); err != nil { |
| 288 | return false, err |
| 289 | } |
| 290 | k++ |
| 291 | } |
| 292 | } else { |
| 293 | k = 1 |
| 294 | } |
| 295 | for ; k <= n; k++ { |
| 296 | if l.pastDeadline() { |
| 297 | return l.markRollback(ctx, provision.RollbackLandingDeadline) |
| 298 | } |
| 299 | if k > counted { |
| 300 | ok, err := l.transition(ctx, "probe attempt "+strconv.Itoa(k), provision.MoveProbeAttempt(l.g, k)) |
| 301 | if !ok { |
| 302 | return false, err |
| 303 | } |
| 304 | counted = k |
| 305 | } |
| 306 | // The claim's writer before the call: whether the attempt admitted is |
| 307 | // the claim's to say after a crash (ruling 1). |
| 308 | before, err := l.claim(ctx) |
| 309 | if err != nil { |
| 310 | return false, err |
| 311 | } |
| 312 | admitting, err := json.Marshal(probeAdmitting{Session: before.Claim.Holder.Session, Admissions: len(before.Claim.Holder.Admissions)}) |
| 313 | if err != nil { |
| 314 | return false, err |
| 315 | } |
| 316 | // Under the journal's one lock (moves.UpdateJournal): the probe |
| 317 | // owners and the paging rules write this journal too. |
| 318 | if err := moves.UpdateJournal(cfg.DataDir, l.rec.Host, l.g.ID, func(j *moves.Journal) error { |
| 319 | j.Phase, j.Attempt = provision.LandProbing, k |
| 320 | j.Mark(probeStepAdmitting+strconv.Itoa(k), string(admitting)) |
| 321 | return nil |
| 322 | }); err != nil { |
| 323 | return false, err |
| 324 | } |
| 325 | if err := l.fault("before-probe-" + strconv.Itoa(k)); err != nil { |
| 326 | return false, err |
| 327 | } |
| 328 | deadline, _ := landingDeadlineOf(m) |
| 329 | if deadline == nil { |
| 330 | return false, fmt.Errorf("move %s: the landing has no deadline", l.g.ID) |
| 331 | } |
| 332 | res, err := cfg.Probe(ctx, activator.ProbeConfig{Host: l.rec.Host, Op: l.g.ID, HoldOp: l.g.ID, Deadline: *deadline, Prepared: &manifest, Inputs: m.Inputs}) |
| 333 | var changed *ProbeInputsChanged |
| 334 | switch { |
| 335 | case err == nil: |
| 336 | log.Printf("moves: %s: landing %s passed its probe on attempt %d of %d", l.rec.Host, l.g.ID, k, n) |
| 337 | if err := l.journalPassed(k, res); err != nil { |
| 338 | return false, err |
| 339 | } |
| 340 | return l.passed(ctx) |
| 341 | case errors.As(err, &changed): |
| 342 | log.Printf("moves: %s: landing %s probe attempt %d: %v", l.rec.Host, l.g.ID, k, err) |
| 343 | return l.markRollback(ctx, changed.Reason) |
| 344 | case errors.Is(err, activator.ErrProbeFailed) || errors.Is(err, activator.ErrAttemptOver): |
| 345 | log.Printf("moves: %s: landing %s probe attempt %d of %d is over: %v", l.rec.Host, l.g.ID, k, n, err) |
| 346 | if err := l.failAttempt(k); err != nil { |
| 347 | return false, err |
| 348 | } |
| 349 | default: |
| 350 | return false, fmt.Errorf("probe attempt %d: %w", k, err) |
| 351 | } |
| 352 | } |
| 353 | if l.pastDeadline() { |
| 354 | return l.markRollback(ctx, provision.RollbackLandingDeadline) |
| 355 | } |
| 356 | return l.markRollback(ctx, provision.RollbackProbeAttempts) |
| 357 | } |
| 358 | |
amadan