| 1 | package hostlife |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "regexp" |
| 6 | "strings" |
| 7 | ) |
| 8 | |
| 9 | const host = "h" |
| 10 | |
| 11 | // ---------- small helpers ---------- |
| 12 | |
| 13 | func (w *world) agentUp(b *box) bool { |
| 14 | return b != nil && !b.Destroyed && !b.Frozen && !b.AgentDown && !b.AgentPaused |
| 15 | } |
| 16 | |
| 17 | func (w *world) localNow(b *box) int { return w.now + b.Skew } |
| 18 | |
| 19 | // readClaim is a fresh read by a box agent (C4). It fails when the box |
| 20 | // cannot reach the control bucket (C6: a failed read never acts). |
| 21 | func (w *world) readClaim(b *box) (*claim, bool) { |
| 22 | if !w.agentUp(b) || b.Partitioned { |
| 23 | return nil, false |
| 24 | } |
| 25 | c := w.claim.clone() |
| 26 | if c != nil && c.Holder != nil { |
| 27 | c.readStamp = b.stampOf(c.Holder.Inc) // the installation stamp this read carries (claim rev 6.5 §6.4) |
| 28 | } |
| 29 | b.snap = c.clone() |
| 30 | return c, true |
| 31 | } |
| 32 | |
| 33 | // admissionClaim is what admission sees: fresh, unless the rule is mutated, |
| 34 | // in which case the box's last cached snapshot. |
| 35 | func (w *world) admissionClaim(b *box) (*claim, bool) { |
| 36 | if on("fresh-admission") { |
| 37 | return w.readClaim(b) |
| 38 | } |
| 39 | // the mutation: admission decides from an older read (a cache, or a |
| 40 | // goroutine that read before it paused) |
| 41 | if b.stale != nil && w.agentUp(b) { |
| 42 | return b.stale.clone(), true |
| 43 | } |
| 44 | if b.snap != nil && w.agentUp(b) { |
| 45 | return b.snap.clone(), true |
| 46 | } |
| 47 | return w.readClaim(b) |
| 48 | } |
| 49 | |
| 50 | // cas writes the claim if nobody wrote since `old` was read. On an ETag |
| 51 | // conflict it re-reads once and re-applies, but only if the op id is |
| 52 | // unchanged (C2); the mutation drops that condition. |
| 53 | func (w *world) cas(old *claim, mut func(c *claim) bool) bool { |
| 54 | cur := w.claim |
| 55 | curVer := 0 |
| 56 | if cur != nil { |
| 57 | curVer = cur.Ver |
| 58 | } |
| 59 | oldVer := 0 |
| 60 | if old != nil { |
| 61 | oldVer = old.Ver |
| 62 | } |
| 63 | base := old |
| 64 | if curVer != oldVer { |
| 65 | if old == nil || cur == nil { |
| 66 | return false |
| 67 | } |
| 68 | if on("claim-cas-opid") && cur.Op.ID != old.Op.ID { |
| 69 | return false |
| 70 | } |
| 71 | base = cur |
| 72 | } |
| 73 | n := base.clone() |
| 74 | if n == nil { |
| 75 | n = &claim{} |
| 76 | } |
| 77 | if !mut(n) { |
| 78 | return false |
| 79 | } |
| 80 | n.Ver = curVer + 1 |
| 81 | w.claim = n |
| 82 | return true |
| 83 | } |
| 84 | |
| 85 | func (w *world) isParticipant(c *claim, b *box) bool { |
| 86 | return c != nil && c.Op.Kind != "serving" && c.Op.Kind != "" && contains(c.Op.Parts, b.ID) |
| 87 | } |
| 88 | |
| 89 | func liveProcs(b *box) []*proc { |
| 90 | if b.Destroyed { |
| 91 | return nil |
| 92 | } |
| 93 | return b.Procs |
| 94 | } |
| 95 | |
| 96 | func hasMain(b *box) bool { |
| 97 | for _, p := range b.Procs { |
| 98 | if p.Main { |
| 99 | return true |
| 100 | } |
| 101 | } |
| 102 | return false |
| 103 | } |
| 104 | |
| 105 | func setKey(xs []int) string { return intsKey(xs) } |
| 106 | |
| 107 | func (w *world) timerDigest(b *box) string { |
| 108 | if on("timer-canonical") { |
| 109 | return fmt.Sprintf("%s/%d", setKey(b.Timers), b.Attempt) |
| 110 | } |
| 111 | return fmt.Sprintf("%s/%s/%d", setKey(b.Timers), setKey(b.History), b.Attempt) |
| 112 | } |
| 113 | |
| 114 | func (w *world) digestOf(b *box) digest { |
| 115 | d := digest{Valid: true} |
| 116 | if b.DB != nil { |
| 117 | d.Writes = sum(b.DB.State) |
| 118 | } |
| 119 | if b.Files != nil { |
| 120 | d.Files = fmt.Sprintf("%d/%d/%d", b.Files.Inc, b.Files.Key, b.Files.Env) |
| 121 | } |
| 122 | d.Timers = w.timerDigest(b) |
| 123 | return d |
| 124 | } |
| 125 | |
| 126 | // ---------- stop evidence (claim §6.1) ---------- |
| 127 | |
| 128 | // evidence kills the host's process tree on b and reports whether stop |
| 129 | // evidence holds. It reads only current system state unless mutated. |
| 130 | func (w *world) evidence(b *box, opID string) bool { |
| 131 | if b.Destroyed { |
| 132 | return false |
| 133 | } |
| 134 | var keep []*proc |
| 135 | killed := len(b.Procs) |
| 136 | for _, p := range b.Procs { |
| 137 | if on("evidence-whole-tree") { |
| 138 | continue // cleanup-only reopen kills the cgroup and every identified stray |
| 139 | } |
| 140 | if p.Strict || p.Main { |
| 141 | continue // cgroup kill, and SIGTERM to the main |
| 142 | } |
| 143 | keep = append(keep, p) // an escaped descendant survives both |
| 144 | } |
| 145 | b.Procs = keep |
| 146 | if !on("evidence-from-state") { |
| 147 | // evidence only from this agent's memory: a stop it watched, or a |
| 148 | // remembered history. After a restart neither exists, and an empty |
| 149 | // box answers ErrStopUnproven forever (R2-2). |
| 150 | if b.stopped == nil { |
| 151 | b.stopped = map[string]bool{} |
| 152 | } |
| 153 | if killed > 0 { |
| 154 | b.stopped[opID] = true |
| 155 | } |
| 156 | if !b.stopped[opID] { |
| 157 | return false |
| 158 | } |
| 159 | } |
| 160 | ok := !on("evidence-whole-tree") || len(b.Procs) == 0 |
| 161 | if ok { |
| 162 | if b.StopEvidence == nil { |
| 163 | b.StopEvidence = map[string]bool{} |
| 164 | } |
| 165 | b.StopEvidence[opID] = true |
| 166 | } |
| 167 | return ok // the mutation certifies an empty cgroup, blind to escapees |
| 168 | } |
| 169 | |
| 170 | // ---------- restore and spawn ---------- |
| 171 | |
| 172 | // restoreCalc is the restore contract as a pure calculation (moves spec |
| 173 | // §5.1 steps 4, 5 and 8; rev 7.7, R6-1): it reads region r's bucket and |
| 174 | // writes nothing, so every oracle may call it. It returns the database the |
| 175 | // contract yields, with Pos the position it restored to, or nil when the |
| 176 | // contract refuses. With restore-by-generation off it is the mutation: the |
| 177 | // newest object under the location. |
| 178 | func (w *world) restoreCalc(r int, inc int, loc, gen string) *localDB { |
| 179 | return w.restoreWith(r, inc, loc, gen, "", 0, -1) |
| 180 | } |
| 181 | |
| 182 | // restorePinned is the restore of an operation's prepared generation before |
| 183 | // its marker is published (rev 7.10 §7.14): the snapshot identity comes |
| 184 | // from the operation's build journal instead of generation.json; tx 0 is |
| 185 | // an empty generation. |
| 186 | func (w *world) restorePinned(r int, inc int, loc, gen, snapKey string, snapVer, tx int) *localDB { |
| 187 | return w.restoreWith(r, inc, loc, gen, snapKey, snapVer, tx) |
| 188 | } |
| 189 | |
| 190 | func (w *world) restoreWith(r int, inc int, loc, gen, pinKey string, pinVer, pinTX int) *localDB { |
| 191 | bk := w.buckets[r] |
| 192 | db := &localDB{Inc: inc, Loc: loc, Gen: gen, State: map[string]int{}} |
| 193 | if !on("restore-by-generation") { |
| 194 | var best *version |
| 195 | for _, k := range bk.keys(loc) { |
| 196 | if !isDataKey(k) { |
| 197 | continue |
| 198 | } |
| 199 | for _, v := range bk.objs[k] { |
| 200 | if !v.Marker && (best == nil || v.ID > best.ID) { |
| 201 | best = v |
| 202 | } |
| 203 | } |
| 204 | } |
| 205 | if best != nil { |
| 206 | apply(db.State, best.B) |
| 207 | db.Pos, db.Key = best.B.Max, best.B.SealKey |
| 208 | db.Log = txsOf(best.B) |
| 209 | db.TX = w.lineageTX(best.B) |
| 210 | if best.B.Inc != 0 { |
| 211 | db.Inc = best.B.Inc |
| 212 | } |
| 213 | } |
| 214 | return db |
| 215 | } |
| 216 | prefix := dataKey(loc, gen, "") |
| 217 | cur := 0 |
| 218 | var applied []*blob |
| 219 | pinnedTX := 0 |
| 220 | if pinTX >= 0 { |
| 221 | pinnedTX = pinTX // the operation's build journal pins the root |
| 222 | } else if gen != "" { |
| 223 | m := bk.current(genMarkerKey(loc)) |
| 224 | if m == nil || m.B.Gen != gen { |
| 225 | return nil // the marker does not name this generation: refuse |
| 226 | } |
| 227 | pinnedTX = m.B.TX |
| 228 | } |
| 229 | // the root: the newest current snapshot of the generation at or past the |
| 230 | // pinned one (for the origin generation, which has no pinned snapshot, |
| 231 | // litestream's own rule: the newest snapshot); the pinned version only |
| 232 | // when no successor is current |
| 233 | var root *blob |
| 234 | if on("root-advance") || gen == "" { |
| 235 | for _, s := range w.currentLTX(bk, prefix, 9) { |
| 236 | if s.B.Max >= pinnedTX && (root == nil || s.B.Max > root.Max) { |
| 237 | root = s.B |
| 238 | } |
| 239 | } |
| 240 | } |
| 241 | if root == nil && gen != "" && pinnedTX > 0 { |
| 242 | key, ver := pinKey, pinVer |
| 243 | if pinTX < 0 { |
| 244 | m := bk.current(genMarkerKey(loc)) |
| 245 | key, ver = m.B.SnapKey, m.B.Base |
| 246 | } |
| 247 | snap := bk.byID(key, ver) |
| 248 | if snap == nil || snap.Marker || snap.B.Max != pinnedTX { |
| 249 | return nil // the snapshot version is missing or not at the pinned TXID |
| 250 | } |
| 251 | root = snap.B |
| 252 | } |
| 253 | if root != nil { |
| 254 | apply(db.State, root) |
| 255 | cur = root.Max |
| 256 | applied = append(applied, root) |
| 257 | db.Key = root.SealKey |
| 258 | if root.Inc != 0 { |
| 259 | db.Inc = root.Inc |
| 260 | } |
| 261 | } |
| 262 | files, gap := w.plan(bk, prefix, cur) |
| 263 | for _, f := range files { |
| 264 | apply(db.State, f) |
| 265 | cur = f.Max |
| 266 | applied = append(applied, f) |
| 267 | if db.Key == 0 { |
| 268 | db.Key = f.SealKey |
| 269 | } |
| 270 | if f.Inc != 0 { |
| 271 | db.Inc = f.Inc |
| 272 | } |
| 273 | } |
| 274 | if gap && on("restore-gap-closed") { |
| 275 | return nil // a gap after cur: refuse (litestream: "non-contiguous ltx files") |
| 276 | } |
| 277 | if len(applied) == 0 && gen != "" && pinnedTX > 0 { |
| 278 | return nil |
| 279 | } |
| 280 | // the oracle's bookkeeping: the writes this database holds, which its |
| 281 | // own later snapshots and anchors carry forward |
| 282 | db.Log = txsOf(applied...) |
| 283 | db.TX = w.lineageTX(applied...) |
| 284 | db.Pos = cur |
| 285 | return db |
| 286 | } |
| 287 | |
| 288 | // restoreInit is the activator's cold restore: the calculation above, then |
| 289 | // the restore-to-replication initialization (§5.1 step 7, R5-1, R6-1). |
| 290 | // Litestream's open takes a fresh box's position from the newest remote |
| 291 | // level-0 file, or 0. When the restored position is above that (a |
| 292 | // snapshot-only install), the activator writes the anchor, every page as |
| 293 | // the level-0 file [cur, cur], and opens replication only once that PUT |
| 294 | // is acknowledged (rule anchor-acked); the position is then read again |
| 295 | // from the remote listing, never assumed. A PUT the client has waited |
| 296 | // uploadTimeout steps for is retried with the same bytes. It returns nil |
| 297 | // while the anchor is outstanding: the wake tries again. The mutation of |
| 298 | // restore-anchor writes no anchor; the mutation of anchor-acked opens at |
| 299 | // once with whatever the listing shows. |
| 300 | func (w *world) restoreInit(b *box, inc int, loc, gen string) *localDB { |
| 301 | db := w.restoreCalc(b.Region, inc, loc, gen) |
| 302 | if db == nil { |
| 303 | w.logf("box %d restore refused under %s", b.ID, dataKey(loc, gen, "")) |
| 304 | w.stats["restore refused"]++ |
| 305 | return nil |
| 306 | } |
| 307 | db.Stamp = b.stampOf(inc) |
| 308 | bk := w.buckets[b.Region] |
| 309 | prefix := dataKey(loc, gen, "") |
| 310 | cur := db.Pos |
| 311 | remote := w.maxL0(bk, prefix) |
| 312 | if cur > remote && on("restore-anchor") { |
| 313 | key := ltxKey(loc, gen, 0, cur, cur) |
| 314 | acked := w.ackedPut(b, key, anchorOf(db, inc), false) |
| 315 | w.stats["restore anchors"]++ |
| 316 | if !acked && on("anchor-acked") { |
| 317 | w.logf("box %d waits for its anchor %s to be acknowledged before opening replication", b.ID, key) |
| 318 | w.stats["anchor waits"]++ |
| 319 | return nil |
| 320 | } |
| 321 | remote = w.maxL0(bk, prefix) // durable evidence: what the listing shows now |
| 322 | } |
| 323 | db.FirstFull = true // litestream's first sync after a cold initialization is a full upload (fact R5-1c) |
| 324 | db.RemotePos = remote |
| 325 | db.Pos = remote |
| 326 | w.logf("box %d restored %s: TXID %d at position %d (key %d, stamp %d)", b.ID, prefix, db.TX, db.Pos, db.Key, db.Stamp) |
| 327 | return db |
| 328 | } |
| 329 | |
| 330 | // stampOK is admission's local check (claim rev 6.5 §6.4): the worker's |
| 331 | // read of the claim carries the installation stamp active on this box at |
| 332 | // that read; it is admitted only if that is still the active stamp. A |
| 333 | // landing probe refused here by its own installation is the fault R3-3 |
| 334 | // named: the oracle records it rather than let the move roll back. |
| 335 | func (w *world) stampOK(b *box, c *claim, probe string) bool { |
| 336 | if b.admitsStamp(c.Holder.Inc, c.readStamp) { |
| 337 | return true |
| 338 | } |
| 339 | if probe != "" && c.Op.Kind == "landing" && c.Op.Tgt == b.ID && c.Ver == w.claim.Ver { |
| 340 | w.violate("INV-3: the landing probe of %s on box %d was refused by the box's own installation record (stamp %d, active %d)", c.Op.ID, b.ID, c.readStamp, b.stampOf(c.Holder.Inc)) |
| 341 | } |
| 342 | return false |
| 343 | } |
| 344 | |
| 345 | // bucketGen reads generation.json at loc in b's own bucket: the generation |
| 346 | // and incarnation it names, or origin when absent. |
| 347 | func (w *world) bucketGen(b *box, loc string) (string, int) { |
| 348 | v := w.buckets[b.Region].current(genMarkerKey(loc)) |
| 349 | if v == nil { |
| 350 | return "", 0 |
| 351 | } |
| 352 | return v.B.Gen, v.B.Inc |
| 353 | } |
| 354 | |
| 355 | // spawn starts the app on b for the holder incarnation. want is the |
| 356 | // generation the claim expects: the holder's current generation for an |
| 357 | // ordinary wake, or the generation an operation is installing for its |
| 358 | // probe. It returns true if the app answered its readiness path. |
| 359 | func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| 360 | h := c.Holder |
| 361 | loc := h.Location |
| 362 | var adm admitInput // the admission this call decided (claim rev 6.9 §6.4a) |
| 363 | if b.Old { |
| 364 | // an old executable: no claim, no markers, no locations; it |
| 365 | // restores whatever is newest at the legacy root |
| 366 | if b.DB == nil { |
| 367 | b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/", State: map[string]int{}, Epoch: b.Epoch} |
| 368 | var newest *version |
| 369 | for _, k := range w.buckets[b.Region].keys("instances/" + host + "/ltx/") { |
| 370 | if v := w.buckets[b.Region].current(k); v != nil && (newest == nil || v.ID > newest.ID) { |
| 371 | newest = v |
| 372 | } |
| 373 | } |
| 374 | if newest != nil { |
| 375 | apply(b.DB.State, newest.B) |
| 376 | b.DB.Pos, b.DB.Key = newest.B.Max, newest.B.SealKey |
| 377 | b.DB.Log = txsOf(newest.B) |
| 378 | b.DB.TX = w.lineageTX(newest.B) |
| 379 | if newest.B.Inc != 0 { |
| 380 | b.DB.Inc = newest.B.Inc |
| 381 | } |
| 382 | } |
| 383 | } |
| 384 | } else { |
| 385 | if b.actFence != "" { |
| 386 | // the activator's fence on the host (a probe stop's, ruling 75) |
| 387 | // refuses every start until its owner lifts it |
| 388 | w.stats["start refused: the activator's fence"]++ |
| 389 | return false |
| 390 | } |
| 391 | if b.foreignFence != "" { |
| 392 | w.violate("INV-32: box %d started the host while %s's activator fence, which its owner never lifted, closed it", b.ID, b.foreignFence) |
| 393 | return false |
| 394 | } |
| 395 | bg, binc := w.bucketGen(b, loc) |
| 396 | if on("start-check") && bg != "" && binc != h.Inc { |
| 397 | w.logf("box %d start check refused: generation.json %q inc %d, claim inc %d", b.ID, bg, binc, h.Inc) |
| 398 | return false |
| 399 | } |
| 400 | // every path that opens replication is admitted here (claim rev 6.9 |
| 401 | // §6.4a, R8-2): an ordinary wake, a committed landing's serving, a |
| 402 | // probe and a restarted probe, a reverted source, an opened rollback |
| 403 | // recovery, a push's writer. A writer still closing holds the lease: |
| 404 | // the wake waits for it |
| 405 | // the retry policy for failed starts (rev 7.11 §7.14, claim rev |
| 406 | // 6.10): a host marked failed gets no new session and no further |
| 407 | // attempt until an operator clears it; a failed start within a |
| 408 | // session waits out its backoff |
| 409 | if b.hostFailed { |
| 410 | w.stats["start refused: host failed"]++ |
| 411 | return false |
| 412 | } |
| 413 | if w.now < b.nextStart { |
| 414 | w.stats["start: backoff"]++ |
| 415 | return false |
| 416 | } |
| 417 | if o := b.open; o != nil { |
| 418 | // an admitted session whose start is in progress continues it |
| 419 | // without another CAS; an admission that moved on (its session |
| 420 | // or generation changed) abandons it |
| 421 | if o.Gen != h.Gen || o.Session != h.Session { |
| 422 | b.open = nil |
| 423 | } else { |
| 424 | return w.resumeOpen(b, c, o) |
| 425 | } |
| 426 | } |
| 427 | // the launch environment: config.env rendered and the release live |
| 428 | // for this incarnation; until then the agent admits nothing (a |
| 429 | // config pass has not run yet, not an app failure) |
| 430 | if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) { |
| 431 | w.stats["start: environment not ready"]++ |
| 432 | return false |
| 433 | } |
| 434 | if b.DB != nil && b.DB.Closing { |
| 435 | w.closeStep(b) |
| 436 | if b.DB.Closing { |
| 437 | w.stats["wake waited for a clean close"]++ |
| 438 | w.logf("box %d: start waits for the previous session's clean close", b.ID) |
| 439 | return false |
| 440 | } |
| 441 | } |
| 442 | var ok bool |
| 443 | if adm, ok = w.admitWriter(b, c, want, probe); !ok { |
| 444 | return false |
| 445 | } |
| 446 | // the generation to restore from is holder.generation. generation.json |
| 447 | // may lag it: an admitted writer that never published (it died between |
| 448 | // its CAS and its marker, or the marker PUT was not acknowledged, rev |
| 449 | // 7.10 §7.14). A lagging marker for an operation's prepared generation |
| 450 | // is published again by op.exec before anything else, from the |
| 451 | // operation's pinned snapshot; a generation with no file at all is an |
| 452 | // abandoned fresh one, left to the sweep, and the baseline is the |
| 453 | // generation the marker names, its parent, which holds the data |
| 454 | source := h.Gen |
| 455 | pin := -1 // the marker's generation, through generation.json |
| 456 | prepared := adm.Mode == "cold" && adm.NewGeneration == want && want != h.Gen |
| 457 | if prepared { |
| 458 | // a cold admission into the operation's prepared generation: the |
| 459 | // marker names the predecessor and the admission publishes the |
| 460 | // new one; the generation clause has nothing to compare |
| 461 | bg = h.Gen |
| 462 | } |
| 463 | if !on("start-check-claim-gen") { |
| 464 | // rev 5's start check: the box's own marker decides, so the |
| 465 | // generation it restores is the one its region's generation.json |
| 466 | // names, whatever the claim says |
| 467 | source = bg |
| 468 | bg = h.Gen |
| 469 | } |
| 470 | if bg != h.Gen && on("start-check") { |
| 471 | switch { |
| 472 | case (h.WriterBox == b.ID || c.Op.Kind == "restoring" && b.ID == c.Op.Src) && w.ancestorGen(loc, h.Gen, bg): |
| 473 | // the marker lags behind this box's own admission, or behind |
| 474 | // the return that named the source's generation again: |
| 475 | // handled below. Another box's lagging marker is the start |
| 476 | // check's refusal (the generation clause, J2-R2-8): a stale |
| 477 | // box whose region holds none of holder.generation's files is |
| 478 | // not the admitted writer that died between its CAS and its |
| 479 | // marker |
| 480 | case w.ancestorGen(loc, bg, h.Gen): |
| 481 | // the marker names a descendant: a clean rollback returned the |
| 482 | // claim to the source's generation and abandoned the target's; |
| 483 | // the marker names the source's generation again, as it did |
| 484 | // before, so the restore contract reads it (rev 7.10 §7.14) |
| 485 | if !w.republishMarker(b, loc, h.Gen) { |
| 486 | return false |
| 487 | } |
| 488 | bg = h.Gen |
| 489 | default: |
| 490 | w.logf("box %d start check refused: generation.json %q, claim %q", b.ID, bg, h.Gen) |
| 491 | return false |
| 492 | } |
| 493 | } |
| 494 | if bg != h.Gen && on("start-check") { |
| 495 | if !w.hasFiles(b, loc, h.Gen) { |
| 496 | // an empty holder.generation: only an unfinished admission's, |
| 497 | // journalled, whose marker still names its predecessor, or |
| 498 | // the predecessor of a chain of them (claim rev 6.10); the |
| 499 | // baseline is the nearest generation of the chain that holds |
| 500 | // a file, else the marker's, and every generation of the chain |
| 501 | // is dead once the new marker lands. Any other empty |
| 502 | // generation is unexplained missing data: refuse and alert |
| 503 | chain, ok := w.unfinishedChain(c, bg) |
| 504 | returned := c.Op.Kind == "restoring" && b.ID == c.Op.Src && h.Gen == c.Op.SrcGen |
| 505 | if !returned && !ok { |
| 506 | w.alert("box %d: generation %q is empty and is not the latest admission's unfinished one (marker %q); the start is refused", b.ID, h.Gen, bg) |
| 507 | w.stats["start refused: unexplained empty generation"]++ |
| 508 | return false |
| 509 | } |
| 510 | // the recovery walk (claim rev 6.22 §6.2a, R13-1): newest |
| 511 | // first over the run's links, a foreign link passed over, a |
| 512 | // covered local one refused, the nearest local one holding an |
| 513 | // object the baseline, else the marker's |
| 514 | run, _ := w.unfinishedRun(c, bg) |
| 515 | src, ok := w.runBaseline(b, c, run, bg) |
| 516 | if !ok { |
| 517 | return false |
| 518 | } |
| 519 | source = src |
| 520 | if src != bg && on("restore-selected-baseline") { |
| 521 | // a generation of the run that holds an object, selected |
| 522 | // while the marker still names the run's baseline: restored |
| 523 | // as the oracle restores it, rooted at [1, 1], never through |
| 524 | // the marker (R14-1: rev 7.14 selected it and then refused it) |
| 525 | pin = 0 |
| 526 | w.stats["restore: run generation rooted"]++ |
| 527 | } |
| 528 | _ = chain |
| 529 | w.stats["abandoned empty generation skipped"]++ |
| 530 | } else if !w.republishMarker(b, loc, h.Gen) { |
| 531 | // no earlier version names it: the admitted writer's marker |
| 532 | // never landed; from the operation's build journal for its |
| 533 | // prepared generation, else a fresh generation rooted at [1,1] |
| 534 | ver, key, tx := 0, "", 0 |
| 535 | if h.Gen == installGen(&c.Op) && c.Op.BaseKey != "" { |
| 536 | ver, key, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX |
| 537 | } |
| 538 | if w.publishGen(b, loc, h.Gen, h.Inc, ver, key, tx) <= 0 { |
| 539 | w.logf("box %d: generation.json for %q not acknowledged", b.ID, h.Gen) |
| 540 | return false |
| 541 | } |
| 542 | w.stats["marker republished"]++ |
| 543 | w.logf("box %d: generation.json published again for %q, which the admitted writer never published", b.ID, h.Gen) |
| 544 | // the marker names holder.generation now: the dead writer's |
| 545 | // predecessor, and the chain's, are retired as its own marker |
| 546 | // stage would have (seed 8528 / 30,000: a warm successor |
| 547 | // continued the generation and nothing ever retired rec19). |
| 548 | // The walk over holder.admissions finds them (claim rev 6.14) |
| 549 | if !w.publishGeneration(b, c, c.Op.ID, phaseOf(&c.Op), h.Session, h.Gen, b.Region, w.newID("ob")) { |
| 550 | return false // the claim moved: next time |
| 551 | } |
| 552 | } |
| 553 | } |
| 554 | // the admitting CAS first (claim rev 6.10 §6.4a): before the purge, |
| 555 | // the restore and the app start, so nothing the app acknowledges is |
| 556 | // ever outside a session (R10-1). Everything after it is the admitted |
| 557 | // session's start, finished by the agent's own loop (resumeOpen) |
| 558 | if !on("admit-before-start") && w.startHook != nil { |
| 559 | // the mutation (rev 7.10's order): the app is already running |
| 560 | // and may acknowledge work before the admitting CAS; a crash |
| 561 | // here leaves the acknowledgement in a database the next warm |
| 562 | // admission purges silently (R10-1) |
| 563 | if w.startHook(b) { |
| 564 | return false // the witness's crash |
| 565 | } |
| 566 | } |
| 567 | if !w.admitCommit(b, c, adm) { |
| 568 | return false |
| 569 | } |
| 570 | nh := w.claim.Holder |
| 571 | o := &openState{Gen: nh.Gen, Want: want, Session: nh.Session, Mode: adm.Mode, Prepared: prepared, Probe: probe, Source: source, SourcePin: pin, RemoteMax: adm.RemoteMaxTXID, stage: "restore"} |
| 572 | b.open = o |
| 573 | return w.resumeOpen(b, c, o) |
| 574 | } |
| 575 | return w.startProc(b, c, want, probe, "legacy") |
| 576 | } |
| 577 | |
| 578 | // appStart is the app's start on its restored database and its readiness |
| 579 | // (§7.7): what the app sees, the launch environment and the readiness |
| 580 | // answer. It starts the processes on success. |
| 581 | func (w *world) appStart(b *box, c *claim, want string, probe bool, mode string) bool { |
| 582 | h := c.Holder |
| 583 | w.foreignWrites(b.DB, h.Inc, fmt.Sprintf("box %d start", b.ID)) |
| 584 | if b.DB.Inc != 0 && b.DB.Inc != h.Inc { |
| 585 | w.violate("box %d started inc %d over a database of inc %d", b.ID, h.Inc, b.DB.Inc) |
| 586 | } |
| 587 | if b.Files != nil && b.Files.Inc != h.Inc { |
| 588 | w.violate("box %d started inc %d with files (instance.key) of inc %d", b.ID, h.Inc, b.Files.Inc) |
| 589 | } |
| 590 | if b.Config != 0 && b.Config != h.Inc { |
| 591 | w.violate("box %d started inc %d with config.env rendered for inc %d", b.ID, h.Inc, b.Config) |
| 592 | } |
| 593 | if b.Files == nil { |
| 594 | w.nextID++ |
| 595 | b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: b.DB.Gen, Env: w.nextID, Done: true, Stamp: b.stampOf(h.Inc)} |
| 596 | } |
| 597 | if b.DB.Key == 0 { |
| 598 | b.DB.Key = b.Files.Key // unsealed data (never booted) adopts the key |
| 599 | } |
| 600 | if b.DB.Key != b.Files.Key { |
| 601 | w.violate("INV-4: box %d started inc %d with instance.key %d over data sealed with %d", b.ID, h.Inc, b.Files.Key, b.DB.Key) |
| 602 | return false |
| 603 | } |
| 604 | if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) { |
| 605 | w.stats["start: environment not ready"]++ |
| 606 | return false // not an app failure: the agent waits for its config pass |
| 607 | } |
| 608 | // FailApp is the injected "the target cannot run it" fault: landing probes only |
| 609 | healthy := !(probe && b.FailApp && (c.Op.Kind == "landing" || c.Op.Kind == "relocating" && b.ID != c.Op.Src)) // and relocation probes (region HA, Task 5) |
| 610 | // §7.7: the probe checks the launch environment at spawn. The forward |
| 611 | // path requires the inputs pinned at admission. A recovery probe (the |
| 612 | // source's revert or rollback) runs on the inputs it rendered now and |
| 613 | // records them (rev 6, J2-R2-5); the mutation keeps rev 5's rule, which |
| 614 | // pins recoveries too. |
| 615 | if probe && c.Op.PinVer != 0 && b.CfgVer != c.Op.PinVer { |
| 616 | switch { |
| 617 | case c.Op.Kind == "landing": |
| 618 | healthy = false |
| 619 | case (c.Op.Kind == "reverting" || c.Op.Kind == "restoring") && !on("recovery-current-inputs"): |
| 620 | healthy = false |
| 621 | } |
| 622 | } |
| 623 | if !healthy { |
| 624 | w.logf("box %d unhealthy start inc %d: cfg=%d live=%s/%d fail=%v kind=%s", b.ID, h.Inc, b.Config, b.Live, b.LiveInc, b.FailApp, c.Op.Kind) |
| 625 | return false |
| 626 | } |
| 627 | return w.startProc(b, c, want, probe, mode) |
| 628 | } |
| 629 | |
| 630 | // the retry policy for failed app starts within a session (rev 7.11 |
| 631 | // §7.14): up to startAttemptsMax consecutive failures, the next attempt |
| 632 | // after a doubling backoff from startBackoff0 steps; then the host is |
| 633 | // marked failed, alerted, and gets no new sessions until an operator |
| 634 | // clears it. The sum of the backoffs is 15 steps; a start that fails |
| 635 | // five times in a row is the release or the configuration, which only |
| 636 | // an operator or a push changes, not the dependency that was slow. |
| 637 | const ( |
| 638 | startAttemptsMax = 5 |
| 639 | startBackoff0 = 1 |
| 640 | ) |
| 641 | |
| 642 | // resumeOpen advances an admitted session's start (rev 7.11 §7.14, claim |
| 643 | // rev 6.10): purge and restore, app start and readiness, then the PUTs |
| 644 | // before replication opens (a fresh generation's full first upload, the |
| 645 | // marker, a warm restore's anchor), each acknowledged before the next, |
| 646 | // retried with the same bytes after the client's timeout, and every one |
| 647 | // of them in the session's ledger (R10-7). It returns true once replication |
| 648 | // is open. The retire chore for a fresh generation's predecessor is queued |
| 649 | // only once the marker names the new generation (R10-2). |
| 650 | func (w *world) resumeOpen(b *box, c *claim, o *openState) bool { |
| 651 | cur := w.claim |
| 652 | if cur == nil || cur.Holder == nil || cur.Holder.Gen != o.Gen || cur.Holder.Session != o.Session || on("exec-writer") && cur.Op.Exec != b.ID { |
| 653 | // the session moved on, or op.exec was handed to another box by a |
| 654 | // CAS citing this box's stop evidence (a rollback after the retry |
| 655 | // policy marked the host failed, seed 25581 / 30,000): the start is |
| 656 | // abandoned, never resumed |
| 657 | b.open = nil |
| 658 | w.stats["open: abandoned"]++ |
| 659 | return false |
| 660 | } |
| 661 | h := cur.Holder |
| 662 | loc := h.Location |
| 663 | bk := w.buckets[b.Region] |
| 664 | for { |
| 665 | switch o.stage { |
| 666 | case "restore": |
| 667 | // a box never reuses its local database (claim rev 6.9, R9-1): |
| 668 | // the purge, error-returning; then the restore from the remote. |
| 669 | // The mutation of remote-baseline keeps a local database of the |
| 670 | // generation it opens, as rev 7.9 and today's activator do |
| 671 | if b.DB != nil && (on("remote-baseline") || b.DB.Inc != h.Inc || b.DB.Gen != o.Gen) { |
| 672 | if b.PurgeFails { |
| 673 | b.PurgeFails = false // a transient error: the next attempt's purge succeeds |
| 674 | w.stats["purge failed: writer not started"]++ |
| 675 | w.logf("box %d: the purge of its local database failed; the start waits", b.ID) |
| 676 | return false |
| 677 | } |
| 678 | b.DB = nil |
| 679 | b.Parked++ |
| 680 | w.stats["local database purged"]++ |
| 681 | } |
| 682 | db := b.DB |
| 683 | if db == nil { |
| 684 | switch { |
| 685 | case o.Prepared: |
| 686 | db = w.restorePinned(b.Region, h.Inc, loc, o.Gen, cur.Op.BaseKey, cur.Op.BaseVer, cur.Op.BaseTX) |
| 687 | default: |
| 688 | // the same selection semantics as the oracle's restoreAsNext |
| 689 | // (R14-1): the marker's generation through generation.json, or |
| 690 | // a run generation the walk selected, rooted at [1, 1] |
| 691 | db = w.restoreWith(b.Region, h.Inc, loc, o.Source, "", 0, o.SourcePin) |
| 692 | if db != nil && o.Mode == "warm" && db.Pos < o.RemoteMax { |
| 693 | w.alert("box %d: the restore of %q reached position %d, the certificate says %d; the remote is incomplete", b.ID, o.Source, db.Pos, o.RemoteMax) |
| 694 | db = nil |
| 695 | } |
| 696 | } |
| 697 | if db == nil { |
| 698 | w.stats["restore refused"]++ |
| 699 | w.logf("box %d: nothing to restore for a %s admission from %q", b.ID, o.Mode, o.Source) |
| 700 | if o.startedAt == 0 { |
| 701 | o.startedAt = w.now |
| 702 | } |
| 703 | if o.Probe && on("probe-window") && w.now-o.startedAt >= probeStartWindow { |
| 704 | // a probe attempt whose restore cannot proceed within its |
| 705 | // window fails like one whose app will not start (§7.7) |
| 706 | b.open = nil |
| 707 | w.stats["probe attempt failed: restore"]++ |
| 708 | w.probeAttemptFailed(b, cur) |
| 709 | } |
| 710 | return false |
| 711 | } |
| 712 | w.logf("box %d restored %s: TXID %d at position %d, %d writes (%s admission into %q)", b.ID, dataKey(loc, o.Source, ""), db.TX, db.Pos, len(db.Log), o.Mode, o.Gen) |
| 713 | } |
| 714 | db.Gen, db.Session, db.Stamp, db.Epoch = o.Gen, o.Session, h.Stamp, b.Epoch |
| 715 | db.Pending, db.Queue, db.InFlight, db.Issued, db.Uncertain, db.Closing, db.CloseReq = nil, nil, nil, nil, false, false, nil |
| 716 | db.RemotePos, db.FirstFull = -1, true // litestream's first sync after a cold initialization is a full upload (fact R5-1c) |
| 717 | if b.Files != nil { |
| 718 | b.Files.Gen, b.Files.Stamp = o.Gen, h.Stamp // the carried files, retagged (§6.4) |
| 719 | } |
| 720 | switch { |
| 721 | case o.Mode == "cold" && o.Prepared: |
| 722 | db.Pos = w.maxL0(bk, dataKey(loc, o.Gen, "")) // the prepared generation's anchor |
| 723 | case o.Mode == "cold": |
| 724 | db.Pos = 0 // a fresh generation: empty until the first upload |
| 725 | } |
| 726 | b.DB = db |
| 727 | o.DB = db |
| 728 | o.stage = "app" |
| 729 | case "app": |
| 730 | // the app starts on the restored database and answers its |
| 731 | // readiness path; replication is not open yet, so nothing it |
| 732 | // acknowledges is outside the session (R10-1). A failed start is |
| 733 | // retried by the same agent under the same session, with backoff, |
| 734 | // up to the limit |
| 735 | if b.DB == nil { |
| 736 | o.stage = "restore" // the database went with a lost disk: restore again |
| 737 | continue |
| 738 | } |
| 739 | if !hasMain(b) { |
| 740 | if o.startedAt == 0 { |
| 741 | o.startedAt = w.now |
| 742 | if o.Probe { |
| 743 | b.lastProbe = probeTiming{startedAt: w.now} // INV-19 reads it at the probe's acceptance |
| 744 | } |
| 745 | } |
| 746 | late := o.Probe && on("probe-window") && w.now-o.startedAt >= probeStartWindow |
| 747 | if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) || w.now < o.startedAt+b.envUnreadyFor { |
| 748 | if late { |
| 749 | // a probe attempt's readiness window has passed with the |
| 750 | // environment never ready: the attempt fails (§7.7) |
| 751 | b.open = nil |
| 752 | b.startFails, b.nextStart = 0, 0 |
| 753 | w.stats["probe attempt failed: environment"]++ |
| 754 | w.probeAttemptFailed(b, cur) |
| 755 | } |
| 756 | return false // the environment: wait, no failure counted |
| 757 | } |
| 758 | if late { |
| 759 | // the deadline is checked before the start (rev 7.13 §7.7, |
| 760 | // R12-2): an attempt whose window passed while the agent was |
| 761 | // paused, or between a failed start and its retry, does not |
| 762 | // start the app at all; a start that succeeded here would |
| 763 | // have made an expired attempt the admitted writer |
| 764 | b.open = nil |
| 765 | b.startFails, b.nextStart = 0, 0 |
| 766 | w.stats["probe attempt failed: deadline before the start"]++ |
| 767 | w.probeAttemptFailed(b, cur) |
| 768 | return false |
| 769 | } |
| 770 | if !w.appStart(b, cur, o.Want, o.Probe, o.Mode) { |
| 771 | b.startFails++ |
| 772 | b.nextStart = w.now + startBackoff0<<(b.startFails-1) |
| 773 | w.stats["start failed"]++ |
| 774 | if o.Probe && (b.startFails >= startAttemptsMax || on("probe-window") && w.now-o.startedAt >= probeStartWindow) { |
| 775 | // a probe attempt's readiness window is the budget's 30 s |
| 776 | // (§7.3, §7.7): the retry policy's restarts share it, and |
| 777 | // when the fifth fails or the window passes the attempt |
| 778 | // fails; the operation counts one of its budgeted tries, |
| 779 | // and the host is not marked failed by a probe |
| 780 | b.open = nil |
| 781 | b.startFails, b.nextStart = 0, 0 |
| 782 | w.stats["probe attempt failed: start"]++ |
| 783 | w.probeAttemptFailed(b, cur) |
| 784 | return false |
| 785 | } |
| 786 | if b.startFails >= startAttemptsMax { |
| 787 | b.hostFailed, b.failedAt = true, w.now |
| 788 | b.open = nil // no session continues on a failed host; the operator's clearing leads to a fresh admission |
| 789 | w.stats["host marked failed"]++ |
| 790 | w.alert("box %d: the host failed to start %d times; marked failed, no new sessions until an operator clears it", b.ID, b.startFails) |
| 791 | } |
| 792 | return false |
| 793 | } |
| 794 | b.startFails = 0 |
| 795 | if o.Probe { |
| 796 | // and before readiness is accepted (R12-2): the app answered |
| 797 | // after the deadline; it is stopped, the attempt fails, and |
| 798 | // nothing it did is a session's (replication is not open) |
| 799 | b.lastProbe.readyAt = w.now |
| 800 | if on("probe-window") && w.now-o.startedAt >= probeStartWindow { |
| 801 | var keep []*proc |
| 802 | for _, p := range b.Procs { |
| 803 | if !p.Main && !p.Strict { |
| 804 | keep = append(keep, p) |
| 805 | } |
| 806 | } |
| 807 | b.Procs = keep |
| 808 | b.open = nil |
| 809 | b.startFails, b.nextStart = 0, 0 |
| 810 | w.stats["probe attempt failed: readiness after the deadline"]++ |
| 811 | w.probeAttemptFailed(b, cur) |
| 812 | return false |
| 813 | } |
| 814 | } |
| 815 | if on("admit-before-start") && w.startHook != nil { |
| 816 | if w.startHook(b) { // a witness's background work right after the start, inside the session |
| 817 | return false // the witness's crash |
| 818 | } |
| 819 | } |
| 820 | } |
| 821 | db := o.DB |
| 822 | switch { |
| 823 | case o.Mode == "cold" && !o.Prepared && (len(db.Log) > 0 || len(db.State) > 0 || len(db.Pending) > 0): |
| 824 | o.stage = "first" |
| 825 | case o.Mode == "cold": |
| 826 | o.stage = "marker" |
| 827 | case db.Pos > w.maxL0(bk, dataKey(loc, o.Gen, "")) && on("restore-anchor"): |
| 828 | o.stage = "anchor" // the restored position is above the level-0 maximum: the anchor [N, N] (§5.1 step 7) |
| 829 | default: |
| 830 | o.stage = "open" |
| 831 | } |
| 832 | case "first", "anchor": |
| 833 | db := o.DB |
| 834 | var key string |
| 835 | if o.blob == nil { |
| 836 | // the bytes are captured once; a retry carries the same bytes |
| 837 | // (R10-7). Rev 7.11's first build rebuilt the file at the retry |
| 838 | // and so carried a write the original did not; when the |
| 839 | // original landed last it replaced the retry's version and the |
| 840 | // write, durable in the retry's, was gone from the restore |
| 841 | // (seed 7574 / 30,000) |
| 842 | if o.stage == "first" { |
| 843 | o.blob = w.deltaOf(db, h.Inc) |
| 844 | } else { |
| 845 | o.blob = anchorOf(db, h.Inc) |
| 846 | } |
| 847 | o.pendAt = len(db.Pending) |
| 848 | } |
| 849 | if o.stage == "first" { |
| 850 | key = ltxKey(loc, o.Gen, 0, 1, 1) |
| 851 | } else { |
| 852 | key = ltxKey(loc, o.Gen, 0, db.Pos, db.Pos) |
| 853 | } |
| 854 | mk := func() *request { return &request{kind: rPut, region: b.Region, key: key, b: o.blob} } |
| 855 | clientOK := o.req != nil && o.req.done && o.req.doneAt-o.req.at < uploadTimeout |
| 856 | if o.req == nil || (!clientOK && w.now-o.req.at >= uploadTimeout) { |
| 857 | if o.req != nil { |
| 858 | // the client gave up (a timeout, or a completion it never |
| 859 | // saw: a lost response, seed 571 / 30,000) and retries the |
| 860 | // same bytes: the session is uncertain for good (R10-7, R9-5) |
| 861 | db.Uncertain = true |
| 862 | w.stats["open: PUT retried"]++ |
| 863 | } |
| 864 | r := mk() |
| 865 | if o.stage == "anchor" { |
| 866 | w.stats["restore anchors"]++ |
| 867 | } else { |
| 868 | w.stats["first uploads"]++ |
| 869 | } |
| 870 | v := w.issue(b.ID, r) |
| 871 | db.track(r, v) |
| 872 | o.req = r |
| 873 | if v > 0 { |
| 874 | o.req = nil |
| 875 | } |
| 876 | } |
| 877 | if o.req != nil && !(o.req.done && o.req.doneAt-o.req.at < uploadTimeout) { |
| 878 | // the client has not seen success: a server completion after |
| 879 | // the wait is a lost response, not an acknowledgement |
| 880 | if o.stage == "anchor" && !on("anchor-acked") { |
| 881 | // the mutation opens before the anchor is acknowledged |
| 882 | } else { |
| 883 | w.stats["open: waits for its "+o.stage]++ |
| 884 | return false |
| 885 | } |
| 886 | } |
| 887 | if cv := bk.current(key); cv != nil { |
| 888 | w.markDurable(key, cv.B) |
| 889 | } |
| 890 | // the writes acknowledged since the capture are still pending |
| 891 | if o.pendAt <= len(db.Pending) { |
| 892 | db.Pending = append([]int(nil), db.Pending[o.pendAt:]...) |
| 893 | } |
| 894 | if o.stage == "first" { |
| 895 | db.Pos, db.FirstFull = 1, false |
| 896 | o.stage = "marker" |
| 897 | } else { |
| 898 | o.stage = "open" |
| 899 | } |
| 900 | o.req, o.blob, o.pendAt = nil, nil, 0 |
| 901 | case "marker": |
| 902 | db := o.DB |
| 903 | bg, _ := w.bucketGen(b, loc) |
| 904 | if bg != o.Gen { |
| 905 | ver, k, tx := 0, "", 0 |
| 906 | if o.Prepared { |
| 907 | ver, k, tx = cur.Op.BaseVer, cur.Op.BaseKey, cur.Op.BaseTX |
| 908 | } |
| 909 | if o.req == nil || o.req.done || w.now-o.req.at >= uploadTimeout { |
| 910 | if o.req != nil && !o.req.done { |
| 911 | w.stats["open: marker PUT retried"]++ |
| 912 | } |
| 913 | v := w.publishGen(b, loc, o.Gen, h.Inc, ver, k, tx) |
| 914 | o.req = w.lastIssued |
| 915 | if v > 0 { |
| 916 | o.req = nil |
| 917 | } |
| 918 | w.logf("box %d: marker PUT for %q issued (acknowledged %v)", b.ID, o.Gen, v > 0) |
| 919 | } |
| 920 | if bg, _ = w.bucketGen(b, loc); bg != o.Gen { |
| 921 | w.stats["open: waits for the marker"]++ |
| 922 | w.logf("box %d: admitted into %q, waiting for its marker", b.ID, db.Gen) |
| 923 | return false |
| 924 | } |
| 925 | } |
| 926 | o.req = nil |
| 927 | // the marker names the new generation: every cold admission, |
| 928 | // a prepared generation's included, publishes it (claim rev |
| 929 | // 6.14 §6.4a item 5; rev 7.12 left prepared generations to the |
| 930 | // operation, R12-3). The walk over holder.admissions covers what |
| 931 | // this region may retire: the predecessor, the empty |
| 932 | // generations of a chain of unfinished admissions, an earlier |
| 933 | // dead writer's predecessor (seed 899 / 30,000); a predecessor |
| 934 | // whose objects are another region's stays uncovered for that |
| 935 | // region's whole-location park |
| 936 | if o.Mode == "cold" { |
| 937 | if !w.publishGeneration(b, cur, cur.Op.ID, phaseOf(&cur.Op), o.Session, o.Gen, b.Region, w.newID("ob")) { |
| 938 | return false // the claim moved: next time |
| 939 | } |
| 940 | } |
| 941 | o.stage = "open" |
| 942 | if w.openYield != nil && w.openYield(b) { |
| 943 | return false // the marker is published; replication is not open yet |
| 944 | } |
| 945 | case "open": |
| 946 | if !hasMain(b) && b.DB != nil { |
| 947 | // the app was stopped idle while the start was still |
| 948 | // finishing its PUTs (seed 2663 / 30,000): the agent's |
| 949 | // replicator ships what the app wrote, then closes cleanly |
| 950 | b.open = nil |
| 951 | w.flush(b) |
| 952 | b.DB.Closing = true |
| 953 | w.closeStep(b) |
| 954 | w.stats["replication opened"]++ |
| 955 | return false |
| 956 | } |
| 957 | b.open = nil |
| 958 | w.stats["replication opened"]++ |
| 959 | return true |
| 960 | } |
| 961 | } |
| 962 | } |
| 963 | |
| 964 | // queueGenerationRetire is claim rev 6.10's QueueGenerationRetire |
| 965 | // (observedSession, publishedGeneration, region, retireGeneration, newID): |
| 966 | // once generation.json names holder.generation, the admitted writer queues |
| 967 | // the park-generation chore for the generation it replaced, owned by the |
| 968 | // region, under the session's op id "session:<holder.session>". The |
| 969 | // mutation of baseline-live queues it at the admitting CAS instead, as |
| 970 | // rev 7.10 did, before the new generation has a root. |
| 971 | // publishGeneration is claim rev 6.14's PublishGeneration(observedOpID, |
| 972 | // observedPhase, observedSession, publishedGeneration, region, newID) |
| 973 | // (§6.4a item 5): the one CAS that retires generations, made after |
| 974 | // generation.json names the published one, bound to the observed |
| 975 | // operation and phase like the admission, legal while serving and inside |
| 976 | // any operation a writer runs in (R11-5). It walks every uncovered cold |
| 977 | // record of holder.admissions: a predecessor that is the published |
| 978 | // generation is covered with no chore (it is live again: a clean return); |
| 979 | // one with no objects anywhere (predecessor_region "") is covered with no |
| 980 | // chore; one whose objects are this region's gets a deduplicated |
| 981 | // park-generation chore, op = op.id inside an operation and "session:<n>" |
| 982 | // while serving, and is covered; one whose objects are another region's |
| 983 | // stays uncovered, with no chore, for that region's whole-location park |
| 984 | // (CoverRegionPredecessors) or its own publication: a chore queued into |
| 985 | // the source's region from the target's publication would park the |
| 986 | // generation the rollback restores from (seeds 33646, 40148, 1296 / rev |
| 987 | // 7.12; R12-3). The mutation of foreign-region-uncovered queues it there. |
| 988 | // Only predecessors recorded in holder.admissions are ever retired, and |
| 989 | // "" and "@root" are one origin (R11-2). Retention is conservative (claim |
| 990 | // rev 6.14 item 8): covered records are trimmed from the front only, to |
| 991 | // eight, and an uncovered record is never dropped, nor anything behind it. |
| 992 | func (w *world) publishGeneration(b *box, c *claim, observedOpID, observedPhase string, session int, published string, region int, newID string) bool { |
| 993 | return w.cas(c, func(n *claim) bool { |
| 994 | if n.Holder == nil || n.Holder.Session != session || n.Holder.Gen != published { |
| 995 | return false |
| 996 | } |
| 997 | named := false |
| 998 | for _, e := range n.Holder.Admissions { |
| 999 | if e.Gen == published { |
| 1000 | named = true |
| 1001 | } |
| 1002 | } |
| 1003 | if !named { |
| 1004 | return false // an entry must name the published generation (claim rev 6.22 item 5; any entry, as E reads it) |
| 1005 | } |
| 1006 | if n.Op.ID != observedOpID || phaseOf(&n.Op) != observedPhase { |
| 1007 | return false // bound to the operation and phase it observed |
| 1008 | } |
| 1009 | id := n.Op.ID |
| 1010 | if n.Op.Kind == "serving" || id == "" { |
| 1011 | id = fmt.Sprintf("session:%d", session) |
| 1012 | } |
| 1013 | // the walk is over the uncovered entries up to and including the |
| 1014 | // published one, warm records included (claim rev 6.21 item 5, as |
| 1015 | // E processes them; rev 7.13 skipped warm records): the publishing |
| 1016 | // session is the latest admitted, so no entry lies beyond it |
| 1017 | for i := range n.Holder.Admissions { |
| 1018 | e := &n.Holder.Admissions[i] |
| 1019 | if e.Session > session { |
| 1020 | break // beyond the published entry: the bound, never reached under the CAS's session check |
| 1021 | } |
| 1022 | if e.Covered { |
| 1023 | continue |
| 1024 | } |
| 1025 | switch { |
| 1026 | case e.Predecessor.Gen == published: |
| 1027 | // the predecessor is the live generation again (a clean |
| 1028 | // return's source generation, a republished lagging marker): |
| 1029 | // left uncovered, no chore. Covered here, as claim rev 6.14 |
| 1030 | // item 5's first bullet reads, nothing would retire it once a |
| 1031 | // later admission supersedes it, since that admission's |
| 1032 | // predecessor is the last record, not the generation it |
| 1033 | // restored from (seed 41914 / 100,000: the source's |
| 1034 | // republished marker covered the target's record while the |
| 1035 | // source's generation was live again, and the generation was |
| 1036 | // residue at the end; a reading for the claim) |
| 1037 | w.stats["publish: predecessor live, left uncovered"]++ |
| 1038 | continue |
| 1039 | case e.PredRegion < 0: |
| 1040 | e.Covered = true // it holds nothing anywhere and no writer can add to it |
| 1041 | w.stats["publish: predecessor empty everywhere"]++ |
| 1042 | continue |
| 1043 | case e.PredRegion != region && on("foreign-region-uncovered"): |
| 1044 | w.stats["publish: predecessor left to its region"]++ |
| 1045 | continue // uncovered: that region's whole-location park covers it |
| 1046 | } |
| 1047 | chore := region |
| 1048 | if e.PredRegion != region { |
| 1049 | chore = e.PredRegion // the mutation: queued into the other region |
| 1050 | w.stats["publish: chore queued into another region (mutation)"]++ |
| 1051 | } |
| 1052 | // an empty dead generation still gets its chore: a dead writer's |
| 1053 | // late upload or certificate may land in it within the chore's |
| 1054 | // horizon (seeds 15273 and 20810 / 30,000) |
| 1055 | ret := e.Predecessor.Gen |
| 1056 | if ret == "" { |
| 1057 | ret = "@root" // the sentinel for the origin generation (claim rev 6.10) |
| 1058 | } |
| 1059 | e.Covered = true |
| 1060 | queued := false |
| 1061 | for _, o := range n.Obl { |
| 1062 | if o.Kind == "park-generation" && o.Region == chore && o.Ret == ret && o.Prefix == n.Holder.Location { |
| 1063 | queued = true // already queued |
| 1064 | } |
| 1065 | } |
| 1066 | if queued { |
| 1067 | continue |
| 1068 | } |
| 1069 | oid := newID |
| 1070 | if hasObl(n.Obl, oid) { |
| 1071 | oid = fmt.Sprintf("%s-%s", newID, ret) // one caller id, one chore per dead generation |
| 1072 | } |
| 1073 | // the chore's scope (claim rev 6.15 §2, R14-5): the publication is |
| 1074 | // the evidence that the predecessor is dead, taken now |
| 1075 | n.Obl = append(n.Obl, w.scoped(obligation{ID: oid, Op: id, Kind: "park-generation", Region: chore, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}, w.now, n.Holder.Inc)) |
| 1076 | } |
| 1077 | // retention is E's (claim rev 6.14 item 8, R12-6): covered entries |
| 1078 | // are dropped from the front only while more than eight entries |
| 1079 | // remain; an uncovered head stops the trimming |
| 1080 | for len(n.Holder.Admissions) > 8 && n.Holder.Admissions[0].Covered { |
| 1081 | n.Holder.Admissions = n.Holder.Admissions[1:] |
| 1082 | } |
| 1083 | return true |
| 1084 | }) |
| 1085 | } |
| 1086 | |
| 1087 | // coverRegionPredecessors is claim rev 6.14's CoverRegionPredecessors |
| 1088 | // (§6.4a item 6), composed into the CAS that queues a region's |
| 1089 | // whole-location park (the commit's park of the source region, a |
| 1090 | // rollback's park of the target region): every uncovered cold record |
| 1091 | // whose predecessor's objects are that region's is covered, with no |
| 1092 | // chore, since the park retires the whole location there (R12-3, note 5). |
| 1093 | // The mutation of cover-at-park leaves them uncovered, and a later |
| 1094 | // publication in the region (or none) is their only cover. |
| 1095 | func (w *world) coverRegionPredecessors(n *claim, region int, why string) { |
| 1096 | if n.Holder == nil || !on("cover-at-park") { |
| 1097 | return |
| 1098 | } |
| 1099 | for i := range n.Holder.Admissions { |
| 1100 | e := &n.Holder.Admissions[i] |
| 1101 | if !e.Covered && e.PredRegion == region && e.Predecessor.Gen != n.Holder.Gen { |
| 1102 | e.Covered = true |
| 1103 | w.stats["admissions covered at the "+why+" park"]++ |
| 1104 | } |
| 1105 | } |
| 1106 | } |
| 1107 | |
| 1108 | // sweptPredecessors is the generation set the at-rest sweep parks in a |
| 1109 | // region, read from the uncovered records whose predecessors are there: |
| 1110 | // at the queuing CAS, the predecessors with a sweepable object; at the |
| 1111 | // acknowledgement CAS (retired true), those whose objects it retired on |
| 1112 | // the way (had objects, has none). An empty predecessor, with nothing to |
| 1113 | // sweep, is in neither set and waits for its publication's chore. |
| 1114 | func (w *world) sweptPredecessors(n *claim, region int, retired bool) []string { |
| 1115 | if n.Holder == nil { |
| 1116 | return nil |
| 1117 | } |
| 1118 | bk := w.buckets[region] |
| 1119 | loc := n.Holder.Location |
| 1120 | var gens []string |
| 1121 | for _, e := range n.Holder.Admissions { |
| 1122 | if e.Covered || e.PredRegion != region || e.Predecessor.Gen == n.Holder.Gen || hasStr(gens, e.Predecessor.Gen) { |
| 1123 | continue |
| 1124 | } |
| 1125 | if retired { |
| 1126 | if w.hadFilesIn(region, loc, e.Predecessor.Gen) && !w.hasFilesIn(region, loc, e.Predecessor.Gen) { |
| 1127 | gens = append(gens, e.Predecessor.Gen) |
| 1128 | } |
| 1129 | continue |
| 1130 | } |
| 1131 | for _, k := range bk.keys(dataKey(loc, e.Predecessor.Gen, "")) { |
| 1132 | if genOf(loc, k) == e.Predecessor.Gen && w.sweepable(n, region, k) { |
| 1133 | gens = append(gens, e.Predecessor.Gen) |
| 1134 | break |
| 1135 | } |
| 1136 | } |
| 1137 | } |
| 1138 | return gens |
| 1139 | } |
| 1140 | |
| 1141 | // coverParkedPredecessors is claim rev 6.21's CoverParkedPredecessors(op, |
| 1142 | // phase, region, generations) (§6.4a item 6): the at-rest sweep covers |
| 1143 | // what it parks, selectively, in the CAS that queues its park-sweep and in |
| 1144 | // the one that acknowledges it; only the records whose predecessors are |
| 1145 | // among the generations it parks in the region are covered, never every |
| 1146 | // predecessor in the region. |
| 1147 | func (w *world) coverParkedPredecessors(n *claim, region int, gens []string) { |
| 1148 | if n.Holder == nil || !on("cover-at-park") { |
| 1149 | return |
| 1150 | } |
| 1151 | for i := range n.Holder.Admissions { |
| 1152 | e := &n.Holder.Admissions[i] |
| 1153 | if !e.Covered && e.PredRegion == region && e.Predecessor.Gen != n.Holder.Gen && hasStr(gens, e.Predecessor.Gen) { |
| 1154 | e.Covered = true |
| 1155 | w.stats["admissions covered at the sweep park"]++ |
| 1156 | } |
| 1157 | } |
| 1158 | } |
| 1159 | |
| 1160 | // hadFilesIn: the generation held an object in region r at some time: a |
| 1161 | // key under its prefix, whatever versions remain (a parked key keeps its |
| 1162 | // entry with none). With hasFilesIn it tells a parked generation from one |
| 1163 | // that never had anything (INV-17). |
| 1164 | func (w *world) hadFilesIn(r int, loc, gen string) bool { |
| 1165 | for _, k := range w.buckets[r].keys(dataKey(loc, gen, "")) { |
| 1166 | if genOf(loc, k) == gen && !isFilesKey(k) { // a files manifest is no database object (region HA §5.6) |
| 1167 | return true |
| 1168 | } |
| 1169 | } |
| 1170 | return false |
| 1171 | } |
| 1172 | |
| 1173 | // hasFilesIn: the generation holds a current object (an LTX file or a |
| 1174 | // certificate, not a delete marker) in region r. |
| 1175 | func (w *world) hasFilesIn(r int, loc, gen string) bool { |
| 1176 | bk := w.buckets[r] |
| 1177 | for _, k := range bk.keys(dataKey(loc, gen, "")) { |
| 1178 | if genOf(loc, k) == gen && bk.current(k) != nil && !isFilesKey(k) { // a files manifest is no database object (region HA §5.6) |
| 1179 | return true |
| 1180 | } |
| 1181 | } |
| 1182 | return false |
| 1183 | } |
| 1184 | |
| 1185 | func hasObl(obl []obligation, id string) bool { |
| 1186 | for _, o := range obl { |
| 1187 | if o.ID == id { |
| 1188 | return true |
| 1189 | } |
| 1190 | } |
| 1191 | return false |
| 1192 | } |
| 1193 | |
| 1194 | // hasFiles: the generation holds at least one object (an LTX file or a |
| 1195 | // certificate) in b's region; for the origin generation "", the objects |
| 1196 | // directly under the location, not those of its g-<id>/ generations. |
| 1197 | func (w *world) hasFiles(b *box, loc, gen string) bool { |
| 1198 | return w.hasFilesIn(b.Region, loc, gen) |
| 1199 | } |
| 1200 | |
| 1201 | // runBaseline walks an unfinished run in b's region (claim rev 6.22 §6.2a, |
| 1202 | // R13-1): each record's predecessor is a link below it, newest first. A |
| 1203 | // link whose generation keeps its objects in another region is passed |
| 1204 | // over, covered or not (a clean return's fresh source admission after the |
| 1205 | // target's park was queued walks R, then the target's T in the other |
| 1206 | // region, then the source's S); a link in this region must be uncovered, |
| 1207 | // since a covered local generation is retired or being retired here, so |
| 1208 | // reaching one as a needed baseline refuses, with an alert. The baseline |
| 1209 | // is the newest link in this region that holds an object, else the |
| 1210 | // marker's generation (bg, the last link). The mutation of |
| 1211 | // covered-baseline-refused ignores the covered flag, as rev 7.13 did, and |
| 1212 | // selects by objects alone. |
| 1213 | func (w *world) runBaseline(b *box, c *claim, run []admissionRecord, bg string) (string, bool) { |
| 1214 | loc := c.Holder.Location |
| 1215 | for _, r := range run { |
| 1216 | link := r.Predecessor.Gen |
| 1217 | if r.PredRegion != b.Region { |
| 1218 | continue // another region's, or holding nothing anywhere: passed over |
| 1219 | } |
| 1220 | if r.Covered && on("covered-baseline-refused") { |
| 1221 | w.alert("box %d: generation %q, a baseline the restore of %q needs, is covered in this region (retired or being retired); the start is refused", b.ID, link, c.Holder.Gen) |
| 1222 | w.stats["start refused: covered local baseline"]++ |
| 1223 | return "", false |
| 1224 | } |
| 1225 | if w.hasFiles(b, loc, link) { |
| 1226 | return link, true |
| 1227 | } |
| 1228 | } |
| 1229 | return bg, true |
| 1230 | } |
| 1231 | |
| 1232 | func hasStr(xs []string, s string) bool { |
| 1233 | for _, x := range xs { |
| 1234 | if x == s { |
| 1235 | return true |
| 1236 | } |
| 1237 | } |
| 1238 | return false |
| 1239 | } |
| 1240 | |
| 1241 | // restoreAsNext is what the next admission on e would restore (the oracle's |
| 1242 | // baseline, rev 7.11): holder.generation when the marker names it, or when |
| 1243 | // it holds objects (the admitted writer republishes its marker first, from |
| 1244 | // the operation's build journal for a prepared generation); behind an |
| 1245 | // explained lagging marker, the nearest generation of the chain that holds |
| 1246 | // objects, else the marker's. |
| 1247 | func (w *world) restoreAsNext(e *box, c *claim) *localDB { |
| 1248 | h := c.Holder |
| 1249 | loc := h.Location |
| 1250 | bg, _ := w.bucketGen(e, loc) |
| 1251 | if bg == h.Gen { |
| 1252 | return w.restoreCalc(e.Region, h.Inc, loc, h.Gen) |
| 1253 | } |
| 1254 | if w.hasFiles(e, loc, h.Gen) { |
| 1255 | ver, key, tx := 0, "", 0 |
| 1256 | if h.Gen == installGen(&c.Op) && c.Op.BaseKey != "" { |
| 1257 | ver, key, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX |
| 1258 | } |
| 1259 | return w.restoreWith(e.Region, h.Inc, loc, h.Gen, key, ver, tx) |
| 1260 | } |
| 1261 | chain, ok := w.unfinishedChain(c, bg) |
| 1262 | if !ok { |
| 1263 | return w.restoreCalc(e.Region, h.Inc, loc, h.Gen) // unexplained: the restore refuses, as the start does |
| 1264 | } |
| 1265 | _ = chain |
| 1266 | run, _ := w.unfinishedRun(c, bg) |
| 1267 | src, fine := w.runBaseline(e, c, run, bg) |
| 1268 | if !fine { |
| 1269 | return nil // the start refuses: a covered local baseline (R13-1) |
| 1270 | } |
| 1271 | if src != bg { |
| 1272 | return w.restoreWith(e.Region, h.Inc, loc, src, "", 0, 0) // a fresh generation rooted at [1, 1] |
| 1273 | } |
| 1274 | return w.restoreCalc(e.Region, h.Inc, loc, bg) |
| 1275 | } |
| 1276 | |
| 1277 | // unfinishedRun is unfinishedChain's records: the cold admissions of the |
| 1278 | // run, newest first, each one's predecessor the link below it; the last |
| 1279 | // record's predecessor is the marker's generation. |
| 1280 | func (w *world) unfinishedRun(c *claim, bg string) ([]admissionRecord, bool) { |
| 1281 | h := c.Holder |
| 1282 | if bg == h.Gen { |
| 1283 | return nil, true |
| 1284 | } |
| 1285 | gen := h.Gen |
| 1286 | var run []admissionRecord |
| 1287 | for i := len(c.Holder.Admissions) - 1; i >= 0; i-- { |
| 1288 | e := c.Holder.Admissions[i] |
| 1289 | if e.Gen != gen { |
| 1290 | return nil, false |
| 1291 | } |
| 1292 | if e.Mode != "cold" { |
| 1293 | continue |
| 1294 | } |
| 1295 | run = append(run, e) |
| 1296 | if e.Predecessor.Gen == bg { |
| 1297 | return run, true |
| 1298 | } |
| 1299 | gen = e.Predecessor.Gen |
| 1300 | } |
| 1301 | return nil, false |
| 1302 | } |
| 1303 | |
| 1304 | // unfinishedChain explains a marker that lags holder.generation (claim rev |
| 1305 | // 6.10, R10-2): the writers journal must show holder.generation to be the |
| 1306 | // latest admission's, and every generation between it and the one the |
| 1307 | // marker names (bg) to be an earlier cold admission's whose predecessor is |
| 1308 | // the next; each died before its marker (seed 28216 / 30,000: two in a |
| 1309 | // row, the second's purge failed after its CAS). The chain is returned |
| 1310 | // newest first, holder.generation included, bg excluded; ok is false when |
| 1311 | // the disagreement is unexplained. |
| 1312 | func (w *world) unfinishedChain(c *claim, bg string) ([]string, bool) { |
| 1313 | h := c.Holder |
| 1314 | if bg == h.Gen { |
| 1315 | return nil, true |
| 1316 | } |
| 1317 | gen := h.Gen |
| 1318 | var chain []string |
| 1319 | for i := len(c.Holder.Admissions) - 1; i >= 0; i-- { |
| 1320 | e := &c.Holder.Admissions[i] |
| 1321 | if e.Gen != gen { |
| 1322 | return nil, false // the journal does not explain holder.generation |
| 1323 | } |
| 1324 | if e.Mode != "cold" { |
| 1325 | continue // a warm session of the same generation |
| 1326 | } |
| 1327 | chain = append(chain, gen) |
| 1328 | if e.Predecessor.Gen == bg { |
| 1329 | return chain, true |
| 1330 | } |
| 1331 | gen = e.Predecessor.Gen |
| 1332 | } |
| 1333 | return nil, false |
| 1334 | } |
| 1335 | |
| 1336 | // probeAttemptFailed counts a failed probe attempt against the operation's |
| 1337 | // budget (§7.3, §7.7): a landing rolls back at probeTries (three), a |
| 1338 | // revert or a rollback's source is stuck at three times that (nine, rev |
| 1339 | // 7.13 R12-5); the next attempt, if any, is a new cold admission into a |
| 1340 | // fresh generation. A probe never marks the host failed (claim rev 6.14): |
| 1341 | // the retry limit of §7.14 belongs to a serving wake. |
| 1342 | func (w *world) probeAttemptFailed(b *box, c *claim) { |
| 1343 | id := c.Op.ID |
| 1344 | w.cas(c, func(n *claim) bool { |
| 1345 | if !sameOp(n.Op.ID, id) { |
| 1346 | return false |
| 1347 | } |
| 1348 | n.Op.ProbeTries++ |
| 1349 | switch n.Op.Kind { |
| 1350 | case "reverting": |
| 1351 | if n.Op.ProbeTries >= probeTries*3 { |
| 1352 | n.Op.RevPhase = "stuck" |
| 1353 | } |
| 1354 | case "restoring": |
| 1355 | if n.Op.ProbeTries >= probeTries*3 { |
| 1356 | n.Op.SrcPhase = "stuck" |
| 1357 | } |
| 1358 | } |
| 1359 | return true |
| 1360 | }) |
| 1361 | if cur := w.claim; cur != nil && sameOp(cur.Op.ID, id) && cur.Op.Kind == "landing" && cur.Op.ProbeTries >= probeTries { |
| 1362 | w.rollback(cur, "target would not start") |
| 1363 | } |
| 1364 | w.logf("box %d: probe attempt %d failed within its readiness window (%s/%s)", b.ID, w.claim.Op.ProbeTries, c.Op.Kind, id) |
| 1365 | } |
| 1366 | |
| 1367 | // resumeStart is the agent's own loop finishing an admitted session's |
| 1368 | // start (rev 7.11 §7.14): the activator process holds the admission and |
| 1369 | // advances the stages without a wake, within the retry policy's backoff. |
| 1370 | // An admission the claim moved past is abandoned. |
| 1371 | func (w *world) resumeStart(b *box, c *claim) { |
| 1372 | o := b.open |
| 1373 | if o == nil || c == nil || c.Holder == nil { |
| 1374 | return |
| 1375 | } |
| 1376 | if o.Gen != c.Holder.Gen || o.Session != c.Holder.Session { |
| 1377 | b.open = nil |
| 1378 | w.stats["open: abandoned"]++ |
| 1379 | return |
| 1380 | } |
| 1381 | if w.now < b.nextStart || b.hostFailed { |
| 1382 | return // the retry policy's backoff, or a failed host |
| 1383 | } |
| 1384 | w.resumeOpen(b, c, o) |
| 1385 | } |
| 1386 | |
| 1387 | // startProc starts the host's processes on b for the admitted writer. |
| 1388 | func (w *world) startProc(b *box, c *claim, want string, probe bool, mode string) bool { |
| 1389 | h := c.Holder |
| 1390 | strict := !b.Old && (on("strict-spawn") || w.rng.Intn(2) == 0) |
| 1391 | gen := want |
| 1392 | if b.DB != nil { |
| 1393 | gen = b.DB.Gen |
| 1394 | } |
| 1395 | b.Procs = append(b.Procs, &proc{Main: true, Strict: strict, Inc: h.Inc, Gen: gen, Session: h.Session, Box: b.ID}) |
| 1396 | if w.rng.Intn(2) == 0 { |
| 1397 | b.Procs = append(b.Procs, &proc{Main: false, Strict: strict, Inc: h.Inc, Gen: gen, Session: h.Session, Box: b.ID}) |
| 1398 | } |
| 1399 | w.logf("box %d spawned inc %d gen %q (%s, probe=%v strict=%v) claim=%s/%s exec=%d", b.ID, h.Inc, gen, mode, probe, strict, c.Op.Kind, c.Op.ID, c.Op.Exec) |
| 1400 | return true |
| 1401 | } |
| 1402 | |
| 1403 | // ---------- admission (C4, §19.1 of the moves spec) ---------- |
| 1404 | |
| 1405 | func (w *world) admits(b *box, c *claim, probe string) bool { |
| 1406 | if c == nil || c.Holder == nil || b.Row == nil || b.Row.Inc != c.Holder.Inc { |
| 1407 | return false |
| 1408 | } |
| 1409 | if on("lost-quarantine") && contains(c.Lost, b.ID) { |
| 1410 | return false |
| 1411 | } |
| 1412 | if on("exec-admission") && c.Op.Exec != b.ID { |
| 1413 | return false |
| 1414 | } |
| 1415 | if !w.stampOK(b, c, probe) { |
| 1416 | return false // a retired installation's worker (claim rev 6.5 §6.4) |
| 1417 | } |
| 1418 | if w.isParticipant(c, b) && b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil { |
| 1419 | return false // self-fenced until the lease is renewed |
| 1420 | } |
| 1421 | if !w.isParticipant(c, b) { |
| 1422 | b.ExecUntil = 0 // a fresh read with no operation ends the fence (§7.1) |
| 1423 | } |
| 1424 | op := c.Op |
| 1425 | // the admission table's role column: in this model each phase's role |
| 1426 | // is a box id, and it is op.exec in every phase, so exec-admission |
| 1427 | // governs both |
| 1428 | role := func(id int) bool { return !on("exec-admission") || b.ID == id } |
| 1429 | switch op.Kind { |
| 1430 | case "serving": |
| 1431 | return probe == "" && b.Row.Status == "ready" |
| 1432 | case "moving": |
| 1433 | return probe == "" && role(op.Src) && op.CutAt == 0 && b.Row.Status == "ready" |
| 1434 | case "reverting": |
| 1435 | if probe != "" { |
| 1436 | return probe == op.ID && role(op.Src) && op.RevPhase == "probing" |
| 1437 | } |
| 1438 | return role(op.Src) && op.RevPhase == "opened" && b.Row.Status == "ready" |
| 1439 | case "landing": |
| 1440 | if probe != "" { |
| 1441 | return probe == op.ID && role(op.Tgt) && op.LandPhase == "probing" |
| 1442 | } |
| 1443 | return role(op.Tgt) && op.Committed && b.Row.Status == "ready" |
| 1444 | case "restoring": |
| 1445 | if probe != "" { |
| 1446 | return probe == op.RollbackID && role(op.Src) && op.SrcPhase == "probing" |
| 1447 | } |
| 1448 | return role(op.Src) && op.SrcPhase == "opened" && b.Row.Status == "ready" |
| 1449 | case "pushing": |
| 1450 | return probe == op.ID && op.Phase == "probing" |
| 1451 | case "relocating": |
| 1452 | if op.Stuck != "" && on("reloc-stuck-refuses-admission") { |
| 1453 | return false // relocationAdmission: a stuck operation admits nobody, the source's wake included (ruling 9, R1-16; rev 4, Astra R1-2) |
| 1454 | } |
| 1455 | if probe != "" { |
| 1456 | return probe == op.ID && role(op.Tgt) && op.Phase == "probing" |
| 1457 | } |
| 1458 | if role(op.Src) && (op.Phase == "starting" || op.Phase == "releasing" && op.Release == nil) { |
| 1459 | return b.Row.Status == "ready" && b.relFence != op.ID // the source serves until its admission fence closes (§5.3, §5.8) |
| 1460 | } |
| 1461 | return role(op.Tgt) && op.Phase == "opened" && b.Row.Status == "ready" |
| 1462 | } |
| 1463 | return false |
| 1464 | } |
| 1465 | |
| 1466 | // certificate is the clean-close certificate of the observed session |
| 1467 | // (claim rev 6.9 §6.4a): closed/<holder.session>.json exists for |
| 1468 | // holder.generation and its max_txid is the generation's remote maximum |
| 1469 | // TXID over every level. Any op.exec box may continue on it (claim rev |
| 1470 | // 6.8.2: the certificate proves every upload of the session completed, |
| 1471 | // whichever box made them). The mutation of writer-session takes the |
| 1472 | // newest certificate of any session (rev 7.8's single marker); the |
| 1473 | // mutation of closed-marker ignores max_txid. |
| 1474 | func (w *world) certificate(b *box, c *claim) *version { |
| 1475 | h := c.Holder |
| 1476 | bk := w.buckets[b.Region] |
| 1477 | prefix := dataKey(h.Location, h.Gen, "") |
| 1478 | var m *version |
| 1479 | if on("writer-session") { |
| 1480 | m = bk.current(closedKey(h.Location, h.Gen, h.Session)) |
| 1481 | } else { |
| 1482 | for _, k := range bk.keys(prefix) { |
| 1483 | if isClosedKey(k) { |
| 1484 | if v := bk.current(k); v != nil && (m == nil || v.ID > m.ID) { |
| 1485 | m = v |
| 1486 | } |
| 1487 | } |
| 1488 | } |
| 1489 | } |
| 1490 | if m == nil { |
| 1491 | w.logf("box %d: no certificate for session %d of %s", b.ID, h.Session, prefix) |
| 1492 | return nil |
| 1493 | } |
| 1494 | if m.B.Seq != b.ID { |
| 1495 | w.stats["certificate of another box"]++ // continued all the same (claim rev 6.8.2) |
| 1496 | } |
| 1497 | if on("closed-marker") && m.B.TX != w.remoteMax(bk, prefix) { |
| 1498 | w.logf("box %d: certificate %d for %s is stale (max_txid %d, remote %d)", b.ID, m.B.Session, prefix, m.B.TX, w.remoteMax(bk, prefix)) |
| 1499 | w.stats["stale certificate"]++ |
| 1500 | return nil |
| 1501 | } |
| 1502 | return m |
| 1503 | } |
| 1504 | |
| 1505 | // installGen is the generation an operation has prepared for its writer: a |
| 1506 | // landing's or a push's g-<op id>/, a dirty rollback's return generation. |
| 1507 | // Its probe's cold admission makes it holder.generation (claim rev 6.9: |
| 1508 | // holder.generation is the only authoritative generation, in every kind |
| 1509 | // and phase). |
| 1510 | func installGen(op *opState) string { |
| 1511 | switch op.Kind { |
| 1512 | case "restoring": |
| 1513 | return op.RollbackID |
| 1514 | case "relocating": |
| 1515 | return op.Prepared // P, fresh per installation (region HA §5.8) |
| 1516 | } |
| 1517 | return op.ID |
| 1518 | } |
| 1519 | |
| 1520 | // republishMarker writes generation.json for gen again, with the content |
| 1521 | // its last marker had: a clean rollback's return to the source generation. |
| 1522 | func (w *world) republishMarker(b *box, loc, gen string) bool { |
| 1523 | bk := w.buckets[b.Region] |
| 1524 | var last *version |
| 1525 | for _, v := range bk.objs[genMarkerKey(loc)] { |
| 1526 | if !v.Marker && v.B != nil && v.B.Gen == gen && (last == nil || v.ID > last.ID) { |
| 1527 | last = v |
| 1528 | } |
| 1529 | } |
| 1530 | if last == nil && gen != "" { |
| 1531 | w.logf("box %d: no earlier marker names %q", b.ID, gen) |
| 1532 | return false |
| 1533 | } |
| 1534 | inc, base, key, tx := w.claim.Holder.Inc, 0, "", 0 |
| 1535 | if last != nil { |
| 1536 | inc, base, key, tx = last.B.Inc, last.B.Base, last.B.SnapKey, last.B.TX |
| 1537 | } |
| 1538 | if w.publishGen(b, loc, gen, inc, base, key, tx) <= 0 { |
| 1539 | w.logf("box %d: generation.json for %q not acknowledged", b.ID, gen) |
| 1540 | return false |
| 1541 | } |
| 1542 | w.stats["marker republished after a clean rollback"]++ |
| 1543 | w.logf("box %d: generation.json names %q again after the rollback", b.ID, gen) |
| 1544 | return true |
| 1545 | } |
| 1546 | |
| 1547 | // parentGen is the generation gen was admitted from within loc, as newGen |
| 1548 | // recorded it; ok is false at the origin. |
| 1549 | func (w *world) parentGen(loc, gen string) (string, bool) { |
| 1550 | p, ok := w.genParent[dataKey(loc, gen, "")] |
| 1551 | if !ok { |
| 1552 | return "", false |
| 1553 | } |
| 1554 | if p == loc { |
| 1555 | return "", true |
| 1556 | } |
| 1557 | return strings.TrimSuffix(strings.TrimPrefix(p, loc+"g-"), "/"), true |
| 1558 | } |
| 1559 | |
| 1560 | // ancestorGen: anc is gen or one of its ancestors within loc. |
| 1561 | func (w *world) ancestorGen(loc, gen, anc string) bool { |
| 1562 | for { |
| 1563 | if gen == anc { |
| 1564 | return true |
| 1565 | } |
| 1566 | p, ok := w.parentGen(loc, gen) |
| 1567 | if !ok { |
| 1568 | return false |
| 1569 | } |
| 1570 | gen = p |
| 1571 | } |
| 1572 | } |
| 1573 | |
| 1574 | // phaseOf is the operation phase an admission observes and binds to (R9-4). |
| 1575 | func phaseOf(op *opState) string { |
| 1576 | return op.Phase + "/" + op.LandPhase + "/" + op.SrcPhase + "/" + op.RevPhase |
| 1577 | } |
| 1578 | |
| 1579 | // admitInput is claim rev 6.9's AdmitWriter input (Admission in the claim library): what the box observed, |
| 1580 | // the mode it decided and the generation it opens. |
| 1581 | type admitInput struct { |
| 1582 | ObservedOpID string |
| 1583 | ObservedPhase string |
| 1584 | ObservedSession int |
| 1585 | ObservedGeneration string |
| 1586 | Box int |
| 1587 | Mode string // warm or cold; "unguarded" is the rev 7.7 mutation |
| 1588 | NewGeneration string // cold only; "" for warm (claim rev 6.10) |
| 1589 | PredecessorCertified bool |
| 1590 | RemoteMaxTXID int |
| 1591 | // KeepStamp (claim rev 6.9.1): a cold admission keeps holder.stamp only |
| 1592 | // when no writer was admitted since it was issued (stamp > session) |
| 1593 | KeepStamp bool |
| 1594 | // Coverage (claim rev 6.12) replaces rev 6.9.1's BaselineCoversPredecessor |
| 1595 | // of the stopped predecessor, so nothing is lost and no writes_lost is |
| 1596 | // recorded; the journal entry says baseline: sealed |
| 1597 | // "" (writes_lost unless certified), "sealed" (a from_db seal of the |
| 1598 | // stopped predecessor), "clean-return" (a clean rollback with |
| 1599 | // src_covered: the unchanged target and the source's complete replica |
| 1600 | // establish preservation, R11-4) |
| 1601 | Coverage string |
| 1602 | // Region is where the new writer keeps its generation's objects (its |
| 1603 | // box's region); PredecessorRegion where the predecessor's are, -1 for |
| 1604 | // a predecessor holding nothing anywhere (claim rev 6.14 §6.4a item 1) |
| 1605 | Region int |
| 1606 | PredecessorRegion int |
| 1607 | } |
| 1608 | |
| 1609 | // admitWriter decides the admission for a start on b under the claim it |
| 1610 | // read (claim rev 6.9 §6.4a, R8-2, R9-3): it makes no CAS and issues |
| 1611 | // nothing. It requires op.exec to be this box (rule exec-writer) and |
| 1612 | // refuses every kind in which no writer may open (deleting, parking, |
| 1613 | // purging, creating; R9-4). The mode is warm when the certificate of the |
| 1614 | // observed session is intact and the generation to open is |
| 1615 | // holder.generation; else cold, into the operation's prepared generation |
| 1616 | // (want, when it is not holder.generation) or a fresh one. |
| 1617 | func (w *world) admitWriter(b *box, c *claim, want string, probe bool) (admitInput, bool) { |
| 1618 | h := c.Holder |
| 1619 | a := admitInput{ObservedOpID: c.Op.ID, ObservedPhase: phaseOf(&c.Op), ObservedSession: h.Session, ObservedGeneration: h.Gen, Box: b.ID, Region: b.Region, PredecessorRegion: w.predecessorRegion(h)} |
| 1620 | if on("exec-writer") && c.Op.Exec != b.ID { |
| 1621 | w.stats["admission refused: not op.exec"]++ |
| 1622 | w.logf("box %d: admission refused, op.exec is %d (%s/%s)", b.ID, c.Op.Exec, c.Op.Kind, c.Op.ID) |
| 1623 | return a, false |
| 1624 | } |
| 1625 | switch c.Op.Kind { |
| 1626 | case "creating", "deleting", "parking", "purging", "placing": |
| 1627 | w.stats["admission refused: op kind"]++ |
| 1628 | return a, false |
| 1629 | } |
| 1630 | if c.Op.Kind == "relocating" && c.Op.Stuck != "" && on("reloc-stuck-refuses-admission") { |
| 1631 | w.stats["admission refused: relocation stuck"]++ |
| 1632 | return a, false |
| 1633 | } |
| 1634 | if c.Op.Kind == "relocating" && !(probe && c.Op.Phase == "probing") && on("reloc-admits-probe-only") { |
| 1635 | // relocatingAdmission (1b-next generation.go): a relocating claim |
| 1636 | // admits only op.exec's cold probe in probing; the source in |
| 1637 | // starting or releasing keeps the session it has and admits no new |
| 1638 | // one (L18 joint round 2 finding 3) |
| 1639 | w.stats["admission refused: relocating admits only the probe"]++ |
| 1640 | return a, false |
| 1641 | } |
| 1642 | a.KeepStamp = h.Stamp > h.Session |
| 1643 | if !on("admit-everywhere") && c.Op.Kind != "serving" { |
| 1644 | a.Mode, a.NewGeneration = "unguarded", want // the mutation: writer sessions for ordinary wakes only (rev 7.8) |
| 1645 | if want == "" { |
| 1646 | a.NewGeneration = h.Gen |
| 1647 | } |
| 1648 | return a, true |
| 1649 | } |
| 1650 | bk := w.buckets[b.Region] |
| 1651 | a.RemoteMaxTXID = w.remoteMax(bk, dataKey(h.Location, h.Gen, "")) |
| 1652 | a.PredecessorCertified = w.certificate(b, c) != nil |
| 1653 | if w.completedPrior(b, c) { |
| 1654 | // the first serving wake after a completion (N6) is claimServingOwners' |
| 1655 | // Select: always cold into a fresh generation, PredecessorCertified |
| 1656 | // refused, the predecessor's writes_lost recorded; the probe's stop |
| 1657 | // receipt serves the handover alone (L18 joint round 2 finding 1) |
| 1658 | a.PredecessorCertified = false |
| 1659 | } |
| 1660 | |
| 1661 | if want == "" || want == h.Gen || w.ancestorGen(h.Location, h.Gen, want) { |
| 1662 | // holder.generation, or an operation's prepared generation that a |
| 1663 | // writer already opened (it is holder.generation or an ancestor of |
| 1664 | // it): never re-entered, since a dead writer's uploads may still |
| 1665 | // land in it (seeds 43 and 87 / 300) |
| 1666 | if a.PredecessorCertified || !on("recovery-generation") { |
| 1667 | a.Mode, a.NewGeneration = "warm", "" // warm: NewGeneration empty (claim rev 6.10) |
| 1668 | return a, true |
| 1669 | } |
| 1670 | a.Mode, a.NewGeneration = "cold", w.newID("rec") |
| 1671 | if l := lastAdmission(h); c.Op.Kind == "restoring" && c.Op.DirtyKnown && !c.Op.Dirty && c.Op.LostAt == 0 && c.Op.SrcCovered && b.ID == c.Op.Src && h.Gen == c.Op.SrcGen && |
| 1672 | l != nil && (l.Box == c.Op.Tgt || l.Session == c.Op.CutSession) && on("clean-return-identity") { |
| 1673 | // (a return made clean by a declared-lost target is not covered: |
| 1674 | // the target may have written, and its session is declared; and |
| 1675 | // only the target's probe session or the source's own cut |
| 1676 | // session is covered by the seal's src_covered: a later session |
| 1677 | // of the recovered source that dies uncertified is declared as |
| 1678 | // any other, seed 95491 / 100,000) |
| 1679 | // a clean rollback's return: the target wrote nothing and the |
| 1680 | // source's replica covered the stop, so the predecessor's writes |
| 1681 | // are preserved and nothing is declared (claim rev 6.12, R11-4) |
| 1682 | a.Coverage = "clean-return" |
| 1683 | } |
| 1684 | return a, true |
| 1685 | } |
| 1686 | // the operation's prepared generation, opened for the first time; built |
| 1687 | // from a seal of the stopped predecessor (a landing from the source's |
| 1688 | // seal, a dirty return from the target's, a push from its own), so the |
| 1689 | // baseline covers it (claim rev 6.9.1) when the seal was built from the |
| 1690 | // stopped database itself; a seal built from the remote because the |
| 1691 | // disk was gone stops at the remote's position, and the predecessor's |
| 1692 | // acknowledged writes beyond it are declared (rev 7.11; seed 34260 / |
| 1693 | // 100,000) |
| 1694 | covers := c.Op.SealFromDB |
| 1695 | if c.Op.Kind == "restoring" { |
| 1696 | covers = c.Op.ReturnFromDB |
| 1697 | } |
| 1698 | a.Mode, a.NewGeneration = "cold", want |
| 1699 | if c.Op.Kind == "relocating" { |
| 1700 | // released (claim §6.4a item 2, L9): only on the admission that |
| 1701 | // opens P, and only when the predecessor is exactly the box and |
| 1702 | // session whose flushed release this operation journaled and whose |
| 1703 | // final_txid the pin reached; a failover's first admission and every |
| 1704 | // later attempt declare (coverage ""). The mutation of |
| 1705 | // reloc-released-coverage grants it to every first opening of P |
| 1706 | r := c.Op.Release |
| 1707 | predBox, predSession, predGen := h.WriterBox, h.Session, h.Gen // the predecessor's identity as admitCommit journals it |
| 1708 | if l := lastAdmission(h); l != nil { |
| 1709 | predBox, predSession, predGen = l.Box, l.Session, l.Gen |
| 1710 | } |
| 1711 | switch { |
| 1712 | case !on("reloc-released-coverage"): |
| 1713 | a.Coverage = "released" |
| 1714 | case r != nil && r.Flushed && c.Op.Pin != nil && c.Op.Pin.TX >= r.FinalTX && predBox == r.Box && predSession == r.Session && predGen == r.Gen: |
| 1715 | a.Coverage = "released" |
| 1716 | } |
| 1717 | return a, true |
| 1718 | } |
| 1719 | if covers { |
| 1720 | a.Coverage = "sealed" |
| 1721 | } |
| 1722 | return a, true |
| 1723 | } |
| 1724 | |
| 1725 | // predecessorRegion is the next admission's predecessor_region (claim rev |
| 1726 | // 6.14 item 1): the region of the last admitted writer's box, where its |
| 1727 | // generation's objects are; before any record, the adopted or created |
| 1728 | // host's writer box, else wherever holder.generation holds an object. It |
| 1729 | // is -1, nothing to retire, when it holds none anywhere. A generation |
| 1730 | // whose objects were already parked, or that never held one, keeps its |
| 1731 | // region: a dead writer's late upload may still land in it, and the chore |
| 1732 | // a later publication queues there is what parks it (seeds 15273 and |
| 1733 | // 20810 / 30,000). |
| 1734 | func (w *world) predecessorRegion(h *holder) int { |
| 1735 | box, gen := h.WriterBox, h.Gen |
| 1736 | if l := lastAdmission(h); l != nil { |
| 1737 | box, gen = l.Box, l.Gen |
| 1738 | } |
| 1739 | for _, b := range w.boxes { |
| 1740 | if b.ID == box && box >= 0 { |
| 1741 | return b.Region |
| 1742 | } |
| 1743 | } |
| 1744 | for x := range w.buckets { |
| 1745 | if w.hasFilesIn(x, h.Location, gen) { |
| 1746 | return x |
| 1747 | } |
| 1748 | } |
| 1749 | return -1 |
| 1750 | } |
| 1751 | |
| 1752 | // lastAdmission is the newest entry of holder.admissions, or nil: the next |
| 1753 | // admission's predecessor (claim rev 6.12). |
| 1754 | func lastAdmission(h *holder) *admissionRecord { |
| 1755 | if h == nil || len(h.Admissions) == 0 { |
| 1756 | return nil |
| 1757 | } |
| 1758 | return &h.Admissions[len(h.Admissions)-1] |
| 1759 | } |
| 1760 | |
| 1761 | // durableInLineage: W is durable in gen or one of its ancestors within loc. |
| 1762 | func (w *world) durableInLineage(loc, gen string, x int) bool { |
| 1763 | seen := map[string]bool{} |
| 1764 | for p := dataKey(loc, gen, ""); p != "" && !seen[p]; p = w.genParent[p] { |
| 1765 | seen[p] = true |
| 1766 | if w.durable[p][x] { |
| 1767 | return true |
| 1768 | } |
| 1769 | } |
| 1770 | return false |
| 1771 | } |
| 1772 | |
| 1773 | // alert raises an operator alert (counted; the oracle checks every |
| 1774 | // writes_lost record raised one, R10-6). |
| 1775 | func (w *world) alert(format string, args ...any) { |
| 1776 | w.alerts++ |
| 1777 | w.logf("alert: "+format, args...) |
| 1778 | } |
| 1779 | |
| 1780 | // admitCommit is the AdmitWriter CAS (claim rev 6.10 §6.4a), made before |
| 1781 | // the purge, the restore and the app start. It is bound to the observed op.id and |
| 1782 | // phase (R9-4; rule admit-bind-op), the observed session and generation, |
| 1783 | // and op.exec. It writes holder.session, holder.writer_box and the journal |
| 1784 | // entry; in cold mode also holder.generation, holder.stamp (the installing |
| 1785 | // CAS), a writes_lost record when the observed session had no certificate |
| 1786 | // (R9-2), and a park-generation chore for a fresh generation's predecessor. |
| 1787 | func (w *world) admitCommit(b *box, c *claim, a admitInput) bool { |
| 1788 | if a.Mode == "unguarded" { |
| 1789 | return true |
| 1790 | } |
| 1791 | h := c.Holder |
| 1792 | var lost *writesLost |
| 1793 | var lostTgt []*writesLost |
| 1794 | var last *admissionRecord // the last admitted writer before this CAS, the predecessor's identity (claim rev 6.12, R11-4) |
| 1795 | var entry admissionRecord |
| 1796 | ok := w.cas(c, func(n *claim) bool { |
| 1797 | if n.Holder == nil || n.Holder.Session != a.ObservedSession || n.Holder.Gen != a.ObservedGeneration { |
| 1798 | return false |
| 1799 | } |
| 1800 | if on("admit-bind-op") { |
| 1801 | // bound to the observed op.id and phase (serving's "" compared |
| 1802 | // literally) and to the kinds in which a writer may open: a |
| 1803 | // closure computed under serving fails once any operation |
| 1804 | // started, a delete included (R9-4) |
| 1805 | if n.Op.ID != a.ObservedOpID || phaseOf(&n.Op) != a.ObservedPhase { |
| 1806 | return false |
| 1807 | } |
| 1808 | switch n.Op.Kind { |
| 1809 | case "creating", "deleting", "parking", "purging", "placing": |
| 1810 | return false |
| 1811 | } |
| 1812 | } |
| 1813 | if on("exec-writer") && n.Op.Exec != b.ID { |
| 1814 | return false |
| 1815 | } |
| 1816 | // the predecessor is the last writer in holder.admissions, with its |
| 1817 | // own session, box and generation; never the generation an |
| 1818 | // operation chose for the restore (claim rev 6.12, R11-4: a clean |
| 1819 | // return resets holder.generation to the source's, but the |
| 1820 | // predecessor is still the target's writer). The mutation takes |
| 1821 | // rev 7.11's holder fields and journals an identity that never |
| 1822 | // admitted a writer |
| 1823 | last = nil |
| 1824 | if l := lastAdmission(n.Holder); l != nil { |
| 1825 | cp := *l |
| 1826 | last = &cp |
| 1827 | } |
| 1828 | p := pred{Session: a.ObservedSession, Box: h.WriterBox, Gen: h.Gen, Stamp: h.Stamp} |
| 1829 | if last != nil && on("clean-return-identity") { |
| 1830 | p = pred{Session: last.Session, Box: last.Box, Gen: last.Gen, Stamp: last.Stamp} |
| 1831 | } |
| 1832 | n.Holder.Session = w.nextGen() |
| 1833 | n.Holder.WriterBox = b.ID |
| 1834 | gen := a.NewGeneration |
| 1835 | if a.Mode != "cold" { |
| 1836 | gen = n.Holder.Gen |
| 1837 | } |
| 1838 | e := admissionRecord{Session: n.Holder.Session, Box: b.ID, Gen: gen, Stamp: n.Holder.Stamp, Mode: a.Mode, Coverage: a.Coverage, Predecessor: p, Region: a.Region, PredRegion: a.PredecessorRegion} |
| 1839 | if a.Mode != "cold" { |
| 1840 | // born covered (claim rev 6.22 item 5): a warm entry's predecessor |
| 1841 | // is its own generation, which nothing retires; the next cold |
| 1842 | // admission's record names it as predecessor and queues its chore |
| 1843 | e.Covered = true |
| 1844 | w.stats["admissions born covered: warm"]++ |
| 1845 | } |
| 1846 | if on("cover-at-park") { |
| 1847 | // born covered (claim rev 6.22 item 5, rev 7.13's reading): the |
| 1848 | // claim already holds a whole-location park for the |
| 1849 | // predecessor's region, so the predecessor is being retired there |
| 1850 | // and this record needs no chore. A rollback's return is admitted |
| 1851 | // after the release queued the target region's park, and |
| 1852 | // CoverRegionPredecessors in that CAS could not see this record |
| 1853 | for _, o := range n.Obl { |
| 1854 | // narrowed (claim rev 6.24 item 5): a park scoped exactly to |
| 1855 | // holder.location, or an unscoped park-replica; a scoped park of |
| 1856 | // an old incarnation or of a staging prefix does not count |
| 1857 | exact := o.Kind == "park-superseded" && o.Prefix == n.Holder.Location && (o.Scope == nil || o.Scope.Location == scopeLocation(n.Holder.Location)) |
| 1858 | unscopedReplica := o.Kind == "park-replica" && o.Scope == nil |
| 1859 | if (exact || unscopedReplica) && o.Region == e.PredRegion { |
| 1860 | e.Covered = true |
| 1861 | w.stats["admissions born covered: whole-location park pending"]++ |
| 1862 | break |
| 1863 | } |
| 1864 | } |
| 1865 | } |
| 1866 | if a.Mode == "cold" { |
| 1867 | w.newGen(n.Holder.Location, h.Gen, a.NewGeneration) |
| 1868 | n.Holder.Gen = a.NewGeneration |
| 1869 | if !a.KeepStamp { |
| 1870 | // a new installation: a new stamp (the installing CAS, claim |
| 1871 | // rev 6.9 §6.4a, §6.4). KeepStamp (claim rev 6.9.1): the stamp |
| 1872 | // stays only when no writer was admitted since it was issued |
| 1873 | // (holder.stamp > holder.session): a landing's first probe |
| 1874 | // after the flip, whose record the target activated before the |
| 1875 | // probe; a restarted probe, a push and a dirty return take a |
| 1876 | // new one |
| 1877 | n.Holder.Stamp = w.issueStamp(n, b.ID) |
| 1878 | } |
| 1879 | e.Stamp = n.Holder.Stamp |
| 1880 | if a.Coverage == "" && !a.PredecessorCertified && p.Box >= 0 && on("writes-lost") { |
| 1881 | // only an uncovered, uncertified predecessor is declared |
| 1882 | // (claim rev 6.12): a sealed or clean-return coverage holds |
| 1883 | // every write it acknowledged. A created incarnation's first |
| 1884 | // admission has no predecessor writer (holder.writer_box is |
| 1885 | // empty before the first writer, claim rev 6.24): nothing is |
| 1886 | // declared, since a declaration always names a box (INV-24) |
| 1887 | lost = &writesLost{Box: p.Box, Gen: p.Gen, Session: p.Session, RemoteMaxTXID: a.RemoteMaxTXID} |
| 1888 | e.Lost = lost |
| 1889 | } else if a.Coverage == "" && !a.PredecessorCertified && p.Box < 0 { |
| 1890 | w.stats["admission: no predecessor writer, nothing declared"]++ |
| 1891 | } |
| 1892 | if n.Op.Kind == "restoring" && n.Op.LostAt != 0 && b.ID == n.Op.Src && on("writes-lost") { |
| 1893 | // additional_writes_lost (claim rev 6.22 item 2, R13-3, R14-2): |
| 1894 | // beside the ordinary writes_lost for the predecessor, every |
| 1895 | // session the lost target was admitted into since the cut (a |
| 1896 | // probe session that certified and a warm continuation that |
| 1897 | // did not are two; no return lane carries either's writes, |
| 1898 | // seed 33373 / 100,000), and the source's cut session when |
| 1899 | // src_covered is false (the seal in the lost region was the |
| 1900 | // one holder of its writes beyond the replica, §7.10; seeds |
| 1901 | // 16395 and 27191 / 30,000). The list and the ordinary record |
| 1902 | // are unique by (box, session): a target destroyed before its |
| 1903 | // first admission leaves the source's cut session as the |
| 1904 | // predecessor itself, declared once (R14-2; rev 7.14 declared |
| 1905 | // it twice through a separate field). The mutation of |
| 1906 | // unique-declarations is rev 7.14's separate path |
| 1907 | lostTgt = nil |
| 1908 | add := func(x *writesLost) { |
| 1909 | if on("unique-declarations") { |
| 1910 | if lost != nil && lost.Box == x.Box && lost.Session == x.Session { |
| 1911 | return |
| 1912 | } |
| 1913 | for _, y := range lostTgt { |
| 1914 | if y.Box == x.Box && y.Session == x.Session { |
| 1915 | return |
| 1916 | } |
| 1917 | } |
| 1918 | } |
| 1919 | lostTgt = append(lostTgt, x) |
| 1920 | } |
| 1921 | for _, q := range n.Holder.Admissions { |
| 1922 | if q.Box == n.Op.Tgt && q.Session > n.Op.CutSession { |
| 1923 | add(&writesLost{Box: q.Box, Gen: q.Gen, Session: q.Session, RemoteMaxTXID: a.RemoteMaxTXID}) |
| 1924 | } |
| 1925 | } |
| 1926 | if !n.Op.SrcCovered && !n.Op.Dirty && h.Gen == n.Op.SrcGen { |
| 1927 | // the persisted cut-session identity, never a search by generation |
| 1928 | add(&writesLost{Box: n.Op.Src, Gen: n.Op.SrcGen, Session: n.Op.CutSession, RemoteMaxTXID: a.RemoteMaxTXID}) |
| 1929 | w.stats["writes_lost: source cut session behind a lost target"]++ |
| 1930 | } |
| 1931 | e.AdditionalWritesLost = lostTgt |
| 1932 | } |
| 1933 | // no retire chore here: the predecessor is the baseline of an |
| 1934 | // unfinished admission until generation.json names the new |
| 1935 | // generation; the chore is queued then (claim rev 6.10, R10-2). |
| 1936 | // The mutation of baseline-live queues it now, as rev 7.10 did |
| 1937 | if !on("baseline-live") && a.Coverage != "sealed" && a.Coverage != "released" { |
| 1938 | ret := h.Gen |
| 1939 | if ret == "" { |
| 1940 | ret = "@root" |
| 1941 | } |
| 1942 | n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: fmt.Sprintf("session:%d", n.Holder.Session), Kind: "park-generation", Region: b.Region, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}, w.now, n.Holder.Inc)) |
| 1943 | } |
| 1944 | } |
| 1945 | if n.Op.Kind == "relocating" && n.Op.Prepared != "" && gen == n.Op.Prepared { |
| 1946 | n.Op.PreparedOpened = gen // prepared_opened, journaled with the admission that opens P (1b-next rev 12 S1) |
| 1947 | } |
| 1948 | n.Holder.Admissions = append(n.Holder.Admissions, e) |
| 1949 | entry = e |
| 1950 | return true |
| 1951 | }) |
| 1952 | if !ok { |
| 1953 | w.logf("box %d: %s admission CAS refused (session %d, op %s/%s exec %d)", b.ID, a.Mode, h.Session, c.Op.Kind, c.Op.ID, c.Op.Exec) |
| 1954 | w.stats["admission CAS refused"]++ |
| 1955 | return false |
| 1956 | } |
| 1957 | w.stats["admissions: "+a.Mode]++ |
| 1958 | if last != nil && (entry.Predecessor.Session != last.Session || entry.Predecessor.Box != last.Box || entry.Predecessor.Gen != last.Gen) { |
| 1959 | // INV-16 (rev 7.12, R11-4): the journalled predecessor is the last |
| 1960 | // admitted writer; a tuple that never admitted a writer (the |
| 1961 | // target's session and box with the source's generation, after a |
| 1962 | // clean return) is a false record |
| 1963 | w.violate("INV-16: box %d journalled predecessor {session %d, box %d, generation %q}, but the last admitted writer was {session %d, box %d, generation %q}", b.ID, entry.Predecessor.Session, entry.Predecessor.Box, entry.Predecessor.Gen, last.Session, last.Box, last.Gen) |
| 1964 | } |
| 1965 | if a.Coverage != "" { |
| 1966 | w.stats["admissions covered: "+a.Coverage]++ |
| 1967 | } |
| 1968 | switch w.claim.Op.Kind { |
| 1969 | case "creating", "deleting", "parking", "purging", "placing": |
| 1970 | // INV-13 (R9-4): no writer is admitted while the host is being |
| 1971 | // created, deleted, parked, purged or placed |
| 1972 | w.violate("INV-13: box %d was admitted as a writer while the host is %s (%s)", b.ID, w.claim.Op.Kind, w.claim.Op.ID) |
| 1973 | } |
| 1974 | if a.Mode == "cold" { |
| 1975 | w.logf("box %d: cold admission %s from generation %q (session %d %scertified, op %s/%s)", b.ID, a.NewGeneration, h.Gen, h.Session, map[bool]string{true: "", false: "un"}[a.PredecessorCertified], c.Op.Kind, c.Op.ID) |
| 1976 | } |
| 1977 | if lost != nil { |
| 1978 | // the writes the predecessor acknowledged and never uploaded are |
| 1979 | // lost, declared and alerted (claim rev 6.9), never silently |
| 1980 | n := w.exemptDeclared(b, lost, h, a.NewGeneration, true) |
| 1981 | w.stats["writes_lost declared"]++ |
| 1982 | w.stats["writes declared lost"] += n |
| 1983 | if d := c.Op.Completed; n > 0 && c.Op.Kind == "serving" && d != nil && d.Box == b.ID && lost.Session == d.Session && lost.Box == d.Box { |
| 1984 | // INV-34 (ruling 87): the probe stop flushed every write its |
| 1985 | // session acknowledged, so the first serving wake's writes_lost, |
| 1986 | // bounded by the remote's last shipped transaction, names none |
| 1987 | w.violate("INV-34: box %d's first serving wake after relocation %s declared %d writes the stopped probe session %d acknowledged lost (ruling 87: the probe stop flushes first)", b.ID, d.Op, n, d.Session) |
| 1988 | } |
| 1989 | w.lostAlert++ |
| 1990 | w.alert("box %d: writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes", b.ID, lost.Box, lost.Gen, lost.Session, lost.RemoteMaxTXID, n) |
| 1991 | } |
| 1992 | for _, x := range lostTgt { |
| 1993 | // one alert per declaration (claim rev 6.22 item 2) |
| 1994 | n := w.exemptDeclared(b, x, h, a.NewGeneration, true) |
| 1995 | w.stats["writes_lost declared"]++ |
| 1996 | w.stats["writes declared lost"] += n |
| 1997 | if x.Box == w.claim.Op.Tgt { |
| 1998 | w.stats["writes_lost: another session of a lost target"]++ |
| 1999 | } |
| 2000 | w.lostAlert++ |
| 2001 | w.alert("box %d: writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes (additional_writes_lost behind the lost target)", b.ID, x.Box, x.Gen, x.Session, x.RemoteMaxTXID, n) |
| 2002 | } |
| 2003 | if lost != nil || len(lostTgt) > 0 { |
| 2004 | // INV-21 (rev 7.15, R14-2): the declarations of one admission are |
| 2005 | // unique by (box, session), the ordinary record and the list together. |
| 2006 | // INV-24 (claim rev 6.24, R16-3): every declaration names a box; the |
| 2007 | // legacy identity, session 0, comes only with generation "" (@root), |
| 2008 | // the adopted host's pre-claim writer, and is valid in both fields |
| 2009 | seen := map[[2]int]bool{} |
| 2010 | all := append([]*writesLost{}, lostTgt...) |
| 2011 | if lost != nil { |
| 2012 | all = append(all, lost) |
| 2013 | } |
| 2014 | for _, x := range all { |
| 2015 | k := [2]int{x.Box, x.Session} |
| 2016 | if seen[k] { |
| 2017 | w.violate("INV-21: box %d declared {box %d, session %d} twice in one admission (writes_lost and additional_writes_lost)", b.ID, x.Box, x.Session) |
| 2018 | } |
| 2019 | seen[k] = true |
| 2020 | if x.Box < 0 || x.Session == 0 && x.Gen != "" { |
| 2021 | w.violate("INV-24: box %d declared {box %d, generation %q, session %d}: a declaration names a box, and session 0 is the legacy identity in the origin generation only", b.ID, x.Box, x.Gen, x.Session) |
| 2022 | } |
| 2023 | if x.Session == 0 { |
| 2024 | w.stats["writes_lost: the legacy identity (session 0)"]++ |
| 2025 | } |
| 2026 | } |
| 2027 | } |
| 2028 | return true |
| 2029 | } |
| 2030 | |
| 2031 | // exemptDeclared applies a writes_lost record to the acknowledgements: the |
| 2032 | // writes the declared session acknowledged and never uploaded into the |
| 2033 | // new lineage are exempt from the loss oracles, since the record and its |
| 2034 | // alert declared them (claim rev 6.9, R10-6). It returns how many. With |
| 2035 | // inv15 it checks INV-15 for the predecessor's record. |
| 2036 | func (w *world) exemptDeclared(b *box, l *writesLost, h *holder, newGen string, inv15 bool) int { |
| 2037 | n := 0 |
| 2038 | for i := range w.acks { |
| 2039 | x := &w.acks[i] |
| 2040 | if !w.declaredBy(x, l, h.Inc) || w.durableInLineage(h.Location, newGen, x.W) { |
| 2041 | continue |
| 2042 | } |
| 2043 | if inv15 { |
| 2044 | // a rollback whose target was declared lost has no return lane: |
| 2045 | // what the seal holds in the target's region is the operator's |
| 2046 | // (§7.10), and the source declares what its own replica lacks |
| 2047 | // (seed 8035 / 30,000) |
| 2048 | declaredLost := contains(w.claim.Lost, l.Box) || w.claim.Op.LostAt != 0 |
| 2049 | if w.isDurable(x.W) && !declaredLost { |
| 2050 | // INV-15 (rev 7.11): a write the host's remote holds is never |
| 2051 | // declared lost while its writer's box is still with us; the |
| 2052 | // lineage must adopt the generation that holds it (a dirty |
| 2053 | // rollback's return lane), not declare it. A box declared |
| 2054 | // lost may leave such writes behind: the record's |
| 2055 | // remote_max_txid names where an operator finds them |
| 2056 | w.violate("INV-15: box %d declared write %d lost, which is durable in the remote outside the new lineage (box %d's session %d)", b.ID, x.W, l.Box, l.Session) |
| 2057 | } |
| 2058 | } |
| 2059 | x.Exempt = true |
| 2060 | n++ |
| 2061 | } |
| 2062 | return n |
| 2063 | } |
| 2064 | |
| 2065 | // declaredBy: acknowledgement x is one the writes_lost record covers: a |
| 2066 | // write of the host's incarnation on the record's box, acknowledged by |
| 2067 | // the writer session the record names, into the generation it names (rev |
| 2068 | // 7.13 §14.2.3, R12-1). A declaration for one session exempts nothing |
| 2069 | // another session acknowledged: those are undeclared losses still. The |
| 2070 | // mutation of session-scoped-loss is rev 7.12's box-wide match. |
| 2071 | func (w *world) declaredBy(x *ack, l *writesLost, inc int) bool { |
| 2072 | if x.Box != l.Box || x.Inc != inc || x.Kind != "write" || x.Exempt { |
| 2073 | return false |
| 2074 | } |
| 2075 | if !on("session-scoped-loss") { |
| 2076 | return true |
| 2077 | } |
| 2078 | return x.Session == l.Session && x.Gen == l.Gen |
| 2079 | } |
| 2080 | |
| 2081 | // probeAccepted is INV-19 (rev 7.13 §7.7, R12-2), checked where a probe |
| 2082 | // is accepted: the attempt whose app this is answered readiness within |
| 2083 | // its window of its first start; an attempt that answered at or after |
| 2084 | // the deadline is never the probed writer. |
| 2085 | func (w *world) probeAccepted(b *box, c *claim) { |
| 2086 | p := b.lastProbe |
| 2087 | if p.startedAt > 0 && p.readyAt >= p.startedAt && p.readyAt-p.startedAt >= probeStartWindow { |
| 2088 | w.violate("INV-19: box %d's probe was accepted %d steps after its first start (%s/%s), beyond the %d-step window", b.ID, p.readyAt-p.startedAt, c.Op.Kind, c.Op.ID, probeStartWindow) |
| 2089 | } |
| 2090 | } |
| 2091 | |
| 2092 | // ---------- parking (claim §6.2) ---------- |
| 2093 | |
| 2094 | // parkStep advances one park of `prefix` in b's own bucket into dest. It |
| 2095 | // returns true when the location is verified empty after the quiet time. |
| 2096 | // keep, when set, names keys the park must leave alone. |
| 2097 | func (w *world) parkStep(b *box, prefix, dest string, quiet int, scope *parkScope, keep func(string) bool) bool { |
| 2098 | return w.parkDepth(b, prefix, dest, quiet, keep, scope, 0) |
| 2099 | } |
| 2100 | |
| 2101 | // member is INV-23's second half (claim rev 6.24 §2; R16-2, R17-1): every |
| 2102 | // reference a scoped park archives or deletes, in a manifest it builds |
| 2103 | // and in a persisted manifest it replays, must be within the scope. It |
| 2104 | // is checked for every reference, whatever rules are on: the listing's |
| 2105 | // confinement (rule scope-membership) is the implementation, this is the |
| 2106 | // oracle. An out-of-scope reference is a violation and is not acted on. |
| 2107 | func (w *world) member(scope *parkScope, id, what string, r ref) bool { |
| 2108 | if scope == nil || inScope(r.Key, scope.Location) { |
| 2109 | return true |
| 2110 | } |
| 2111 | w.violate("INV-23: chore %s (scope %s, incarnation %d) would %s %s@%d, outside its scope", id, scope.Location, scope.Inc, what, r.Key, r.Ver) |
| 2112 | return false |
| 2113 | } |
| 2114 | |
| 2115 | func (w *world) parkDepth(b *box, prefix, dest string, quiet int, keep func(string) bool, scope *parkScope, depth int) bool { |
| 2116 | if depth > 8 { |
| 2117 | return false // a park that cannot finish: reported as non-convergence |
| 2118 | } |
| 2119 | bk := w.buckets[b.Region] |
| 2120 | id := dest |
| 2121 | skip := func(k string) bool { |
| 2122 | return (strings.HasSuffix(k, "lock.json") && on("park-skip-lock")) || (keep != nil && keep(k)) |
| 2123 | } |
| 2124 | // the parker takes the location's replica lease first (claim §6.2 step 1): |
| 2125 | // its own lock.json, renewed on every visit |
| 2126 | w.issue(b.ID, &request{kind: rPut, region: b.Region, key: prefix + "lock.json", b: &blob{Kind: "lock", Text: fmt.Sprintf("box%d@%d", b.ID, w.now)}}) |
| 2127 | manKey := dest + "manifest" |
| 2128 | if !on("retirement-ids") { |
| 2129 | // one destination per operation, shared by every retirement event |
| 2130 | if i := strings.Index(dest, "/"+host+"/"); i >= 0 { |
| 2131 | manKey = dest[:i+len(host)+2] + "manifest" |
| 2132 | dest = dest[:i+len(host)+2] |
| 2133 | } |
| 2134 | } |
| 2135 | man := bk.current(manKey) |
| 2136 | if man == nil { |
| 2137 | // the snapshot (claim rev 4 §6.2 step 2): the archive set is every |
| 2138 | // non-marker version, current and noncurrent; the deletion set is |
| 2139 | // every version, delete markers included. The mutation archives |
| 2140 | // and deletes only current versions, so older ones resurface. |
| 2141 | var archive, dels []ref |
| 2142 | for _, k := range bk.keys(prefix) { |
| 2143 | if skip(k) { |
| 2144 | continue |
| 2145 | } |
| 2146 | vs := bk.objs[k] |
| 2147 | for i, v := range vs { |
| 2148 | current := i == len(vs)-1 |
| 2149 | if !on("park-all-versions") && !current { |
| 2150 | continue |
| 2151 | } |
| 2152 | r := ref{Region: b.Region, Key: k, Ver: v.ID} |
| 2153 | if !w.member(scope, id, "delete", r) { |
| 2154 | continue // out of scope: never deleted, never archived |
| 2155 | } |
| 2156 | dels = append(dels, r) |
| 2157 | if !v.Marker { |
| 2158 | archive = append(archive, r) |
| 2159 | } |
| 2160 | } |
| 2161 | } |
| 2162 | // copy each archive-set version by id; a destination keyed by the |
| 2163 | // source version is idempotent, so progress accumulates |
| 2164 | all := true |
| 2165 | for _, r := range archive { |
| 2166 | dk := dest + strings.TrimPrefix(r.Key, "instances/"+host+"/") + fmt.Sprintf("@%d", r.Ver) |
| 2167 | if bk.current(dk) != nil { |
| 2168 | continue |
| 2169 | } |
| 2170 | all = false |
| 2171 | w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: dk, srcRegion: b.Region, srcKey: r.Key, srcVer: r.Ver}) |
| 2172 | } |
| 2173 | if !all { |
| 2174 | return false |
| 2175 | } |
| 2176 | // the manifest last, once every copy exists, with both sets |
| 2177 | w.issue(b.ID, &request{kind: rPut, region: b.Region, key: manKey, b: &blob{Kind: "manifest", Refs: archive, Dels: dels}}) |
| 2178 | return false |
| 2179 | } |
| 2180 | // replay the manifest's deletion set, by version id, in one batch per |
| 2181 | // visit; a persisted manifest's references are checked against the |
| 2182 | // scope as the fresh set was (R17-1: a manifest naming another |
| 2183 | // incarnation's keys under a valid scope is refused, not replayed) |
| 2184 | issued := false |
| 2185 | for _, r := range man.B.Refs { |
| 2186 | w.member(scope, id, "archive", r) |
| 2187 | } |
| 2188 | for _, r := range man.B.Dels { |
| 2189 | if !w.member(scope, id, "delete", r) || bk.byID(r.Key, r.Ver) == nil { |
| 2190 | continue |
| 2191 | } |
| 2192 | if on("park-by-version") { |
| 2193 | w.issue(b.ID, &request{kind: rDelVer, region: b.Region, key: r.Key, ver: r.Ver}) |
| 2194 | } else { |
| 2195 | w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: r.Key}) |
| 2196 | } |
| 2197 | issued = true |
| 2198 | } |
| 2199 | if issued { |
| 2200 | return false |
| 2201 | } |
| 2202 | if on("quiet-before-ack") { |
| 2203 | if w.now < quiet { |
| 2204 | return false |
| 2205 | } |
| 2206 | // a straggler written after the snapshot: park it as a new event |
| 2207 | for _, k := range bk.keys(prefix) { |
| 2208 | if skip(k) { |
| 2209 | continue |
| 2210 | } |
| 2211 | for _, v := range bk.objs[k] { |
| 2212 | if !v.Marker { |
| 2213 | return w.parkDepth(b, prefix, dest+"s/", quiet, keep, scope, depth+1) |
| 2214 | } |
| 2215 | } |
| 2216 | } |
| 2217 | } |
| 2218 | return true |
| 2219 | } |
| 2220 | |
| 2221 | // localPark retires b's local state for incarnation inc under the host |
| 2222 | // guard: a retired marker first, then each artifact renamed aside, one per |
| 2223 | // call, so a crash can fall between any two (claim §6.3). It reports |
| 2224 | // whether nothing is left. |
| 2225 | // localPark is claim rev 6.5's retire-installation (§6.3): the active |
| 2226 | // stamp moves into retired_stamps and becomes none, then every artifact is |
| 2227 | // parked. It serves a delete, a move away and the quarantine; a landing |
| 2228 | // parks its predecessor's leftovers with retire-artifacts instead. |
| 2229 | func (w *world) localPark(b *box, inc int) bool { |
| 2230 | if on("hostguard") && b.stampOf(inc) != 0 { |
| 2231 | b.retireStamp(inc) // retire-installation: active moves into retired_stamps (claim rev 6.5 §6.4) |
| 2232 | return false |
| 2233 | } |
| 2234 | switch { |
| 2235 | case b.DB != nil: |
| 2236 | b.DB = nil |
| 2237 | case b.Files != nil: |
| 2238 | b.Files = nil |
| 2239 | case b.Config != 0: |
| 2240 | b.Config = 0 |
| 2241 | case b.Live != "": |
| 2242 | b.Live, b.LiveInc = "", 0 |
| 2243 | case len(b.Timers) > 0 || len(b.History) > 0 || b.Attempt != 0: |
| 2244 | b.Timers, b.History, b.Attempt = nil, nil, 0 |
| 2245 | default: |
| 2246 | b.receipt = nil // a relocation's install journal goes with the installation |
| 2247 | return true |
| 2248 | } |
| 2249 | b.Parked++ |
| 2250 | return false |
| 2251 | } |
| 2252 | |
| 2253 | // ---------- obligations ---------- |
| 2254 | |
| 2255 | func (w *world) ownerOf(o obligation) *box { |
| 2256 | if o.Kind == "park-local" { |
| 2257 | b := w.boxByID(o.Box) |
| 2258 | if b == nil || b.Destroyed { |
| 2259 | return nil |
| 2260 | } |
| 2261 | return b |
| 2262 | } |
| 2263 | return w.regionBox(o.Region) |
| 2264 | } |
| 2265 | |
| 2266 | // doObligation runs one step of an obligation owned by b. |
| 2267 | func (w *world) doObligation(b *box, c *claim, o obligation) { |
| 2268 | switch o.Kind { |
| 2269 | case "park-local": |
| 2270 | if !w.evidence(b, o.Op) { |
| 2271 | w.logf("box %d: park-local %s waits for stop evidence", b.ID, o.ID) |
| 2272 | return |
| 2273 | } |
| 2274 | inc := 0 |
| 2275 | if c.Holder != nil { |
| 2276 | inc = c.Holder.Inc |
| 2277 | } |
| 2278 | if b.DB != nil && b.DB.Inc != 0 { |
| 2279 | inc = b.DB.Inc |
| 2280 | } |
| 2281 | if !w.localPark(b, inc) { |
| 2282 | if w.calm { |
| 2283 | w.logf("box %d: park-local %s parked an artifact of inc %d (stamp %d, db=%v files=%v cfg=%d live=%q)", b.ID, o.ID, inc, b.stampOf(inc), b.DB != nil, b.Files != nil, b.Config, b.Live) |
| 2284 | } |
| 2285 | return // the next artifact on a later step |
| 2286 | } |
| 2287 | w.logf("box %d: park-local %s done", b.ID, o.ID) |
| 2288 | if b.Row != nil && b.Row.HoldOp == o.Op || b.Row != nil && (c.Holder == nil || b.ID != c.Op.Exec) { |
| 2289 | b.Row = nil |
| 2290 | } |
| 2291 | default: |
| 2292 | quiet := o.Quiet |
| 2293 | if o.Scope != nil && !scopeLocationRe.MatchString(o.Scope.Location) { |
| 2294 | // INV-23 (claim rev 6.23 §2, R15-1): a scope's location is exactly |
| 2295 | // one of the grammar's four forms |
| 2296 | w.violate("INV-23: chore %s (%s) carries a scope location %q outside the claim's grammar", o.ID, o.Kind, o.Scope.Location) |
| 2297 | } |
| 2298 | // membership (claim rev 6.24 §2, R16-1, R16-2): a park executes |
| 2299 | // strictly within its scope's namespace. Every key the chore would |
| 2300 | // archive must be in scope (INV-23's second half, checked on the |
| 2301 | // manifest's sources), and under the rule the chore's listing is |
| 2302 | // confined to the scope: the legacy instances/<host> scope excludes |
| 2303 | // child incarnations, an instances scope never touches moves/, a |
| 2304 | // staging lane's scope only its own lane. The mutation of |
| 2305 | // scope-membership is rev 7.16's listing of the whole park prefix |
| 2306 | var member func(string) bool |
| 2307 | if o.Scope != nil && o.Kind != "park-local" { |
| 2308 | loc := o.Scope.Location |
| 2309 | member = func(k string) bool { return inScope(k, loc) } |
| 2310 | } |
| 2311 | if o.Kind == "park-replica" || o.Kind == "park-superseded" || o.Kind == "park-staging" { |
| 2312 | // the owner first proves the region's box runs nothing for the |
| 2313 | // host, when it is not the current executor (a whole-location |
| 2314 | // park; a staging lane's bound is its seal's, persisted) |
| 2315 | if (o.Kind == "park-replica" || o.Kind == "park-superseded") && (c.Op.Exec != b.ID || c.Holder == nil) { |
| 2316 | if !w.evidence(b, o.Op) { |
| 2317 | return |
| 2318 | } |
| 2319 | if w.perRegion > 1 && on("park-evidence-every-box") { |
| 2320 | // every box of the region publishes its own stop evidence |
| 2321 | // (hostops collectEvidence over the inventory); the owner's |
| 2322 | // alone proves nothing about its sibling. The mutation |
| 2323 | // parks on the owner's evidence alone, and a sibling |
| 2324 | // writer's replicator ships into the parked region |
| 2325 | // (seed 819; wParkEvidenceEveryBox) |
| 2326 | for _, x := range w.regionBoxes(b.Region) { |
| 2327 | if x != b && !contains(c.Lost, x.ID) && !x.StopEvidence[o.Op] { |
| 2328 | return |
| 2329 | } |
| 2330 | } |
| 2331 | } |
| 2332 | } |
| 2333 | // a replicator's PUT signed before that evidence can land up to |
| 2334 | // the horizon later: verify only after it (this model's finding). |
| 2335 | // The evidence time is the chore's persisted park scope (claim |
| 2336 | // rev 6.15 §2, R13-4), bound in a CAS before the first step runs, |
| 2337 | // so a retry after an agent restart keeps the original time; the |
| 2338 | // mutation keeps rev 7.13's box-local map, which the restart |
| 2339 | // clears, and the horizon starts again from the restart's clock |
| 2340 | if o.Scope == nil && on("persisted-park-scope") { |
| 2341 | inc := 0 |
| 2342 | if c.Holder != nil { |
| 2343 | inc = c.Holder.Inc |
| 2344 | } |
| 2345 | // the chore was queued unscoped (a delete's or adoption's |
| 2346 | // park-replica): the evidence is the stop evidence first in |
| 2347 | // hand, now; a staging lane never comes here under the rule |
| 2348 | scope := &parkScope{Location: scopeLocation(o.Prefix), Inc: inc, EvidenceAt: w.now, RequestLimit: requestLimitH} |
| 2349 | w.cas(c, func(n *claim) bool { |
| 2350 | for i := range n.Obl { |
| 2351 | if n.Obl[i].ID == o.ID && n.Obl[i].Scope == nil { |
| 2352 | n.Obl[i].Scope = scope |
| 2353 | return true |
| 2354 | } |
| 2355 | } |
| 2356 | return false |
| 2357 | }) |
| 2358 | w.stats["park scope bound"]++ |
| 2359 | return // the next step runs under the persisted scope |
| 2360 | } |
| 2361 | if b.evidAt == nil { |
| 2362 | b.evidAt = map[string]int{} |
| 2363 | } |
| 2364 | if _, ok := b.evidAt[o.ID]; !ok { |
| 2365 | b.evidAt[o.ID] = w.now |
| 2366 | } |
| 2367 | // the horizon after the evidence is H, the signature-age |
| 2368 | // allowance plus the request bound (R14-3): the scope persists the |
| 2369 | // request bound, never the whole horizon. One deadline for every |
| 2370 | // scope, a staging lane's included (claim rev 6.23 §2, R15-1): a |
| 2371 | // listing later than evidence_at + H, the evidence being the |
| 2372 | // persisted original bound, never the cleanup's start; rev 7.15 |
| 2373 | // bound a staging park at cleanup and then ignored the bound for |
| 2374 | // the seal's quiet time |
| 2375 | deadline := b.evidAt[o.ID] + horizonH |
| 2376 | if o.Scope != nil { |
| 2377 | deadline = o.Scope.deadline() |
| 2378 | } |
| 2379 | if (on("quiet-before-ack") || o.Kind == "park-staging") && deadline > quiet { |
| 2380 | quiet = deadline |
| 2381 | } |
| 2382 | } |
| 2383 | if o.Kind == "park-generation" && o.Scope != nil && on("quiet-before-ack") { |
| 2384 | // a dead generation's chore under its scope (R14-5): the |
| 2385 | // publication's evidence time plus H, never less than the |
| 2386 | // queued quiet time |
| 2387 | if q := o.Scope.deadline(); q > quiet { |
| 2388 | quiet = q |
| 2389 | } |
| 2390 | } |
| 2391 | if o.Kind != "park-local" && o.Scope == nil && on("persisted-park-scope") { |
| 2392 | // INV-22 (rev 7.15, R14-5): every remote chore runs its steps under |
| 2393 | // a persisted scope, bound at its queuing or before its first step |
| 2394 | w.violate("INV-22: remote chore %s (%s %s) is about to run with no persisted park scope", o.ID, o.Kind, o.Prefix) |
| 2395 | } |
| 2396 | var keep func(string) bool |
| 2397 | if member != nil && on("scope-membership") { |
| 2398 | keep = func(k string) bool { return !member(k) } // confined to the scope |
| 2399 | } |
| 2400 | if o.Kind == "park-sweep" { |
| 2401 | inner := keep |
| 2402 | keep = func(k string) bool { return !w.sweepable(w.claim, b.Region, k) || inner != nil && inner(k) } |
| 2403 | quiet = 0 |
| 2404 | if o.Scope != nil && on("quiet-before-ack") { |
| 2405 | // the sweep's scope (claim rev 6.23 §6.2a): a park-sweep is a |
| 2406 | // retirement event, parked at once and acknowledged only after a |
| 2407 | // verification listing later than evidence_at + H, H the |
| 2408 | // signature-age allowance plus the request bound; the straggler |
| 2409 | // pass is that listing; every sweep acknowledgement waits it |
| 2410 | quiet = o.Scope.deadline() |
| 2411 | } |
| 2412 | } |
| 2413 | if o.Kind == "park-generation" { |
| 2414 | // a dead generation within the live location (a cold recovery's |
| 2415 | // old generation, claim rev 6.6 §6.4a): only its own keys |
| 2416 | retire := o.Ret |
| 2417 | if retire == "@root" { |
| 2418 | retire = "" // the origin generation's sentinel (claim rev 6.10) |
| 2419 | } |
| 2420 | inner := keep |
| 2421 | keep = func(k string) bool { |
| 2422 | if genOf(o.Prefix, k) != retire || inner != nil && inner(k) { |
| 2423 | return true |
| 2424 | } |
| 2425 | if w.filesLive(w.claim, b.Region, k) { |
| 2426 | w.markSweepOwed(b.Region, k) |
| 2427 | return true |
| 2428 | } |
| 2429 | return false |
| 2430 | } |
| 2431 | } |
| 2432 | if !w.parkStep(b, o.Prefix, o.Dest(), quiet, o.Scope, keep) { |
| 2433 | return |
| 2434 | } |
| 2435 | } |
| 2436 | deadline := horizonH |
| 2437 | if b.evidAt != nil { |
| 2438 | deadline = b.evidAt[o.ID] + horizonH |
| 2439 | } |
| 2440 | if o.Scope != nil { |
| 2441 | deadline = o.Scope.deadline() |
| 2442 | } |
| 2443 | w.cas(c, func(n *claim) bool { |
| 2444 | for i, x := range n.Obl { |
| 2445 | if x.ID == o.ID { |
| 2446 | if (o.Kind == "park-replica" || o.Kind == "park-superseded") && on("quiet-before-ack") && o.Region >= 0 && deadline > n.Quiet[o.Region] { |
| 2447 | n.Quiet[o.Region] = deadline // quiet_after survives free |
| 2448 | } |
| 2449 | if o.Kind == "park-superseded" && n.Holder != nil && o.Prefix == n.Holder.Location { |
| 2450 | // INV-17 (rev 7.13, R12-3): the whole location in this region |
| 2451 | // is retired now; every record whose predecessor kept its |
| 2452 | // objects here was covered by the CAS that queued the park |
| 2453 | // (item 6) or born covered under it. One still uncovered |
| 2454 | // would wait for a publication in this region that may never |
| 2455 | // come, with a chore for a prefix already parked |
| 2456 | for _, e := range n.Holder.Admissions { |
| 2457 | if !e.Covered && e.PredRegion == o.Region && e.Predecessor.Gen != n.Holder.Gen { |
| 2458 | w.violate("INV-17: the whole-location park %s of region %d retired admission %d's predecessor %q, and the record is uncovered", o.ID, o.Region, e.Session, e.Predecessor.Gen) |
| 2459 | break |
| 2460 | } |
| 2461 | } |
| 2462 | } |
| 2463 | if o.Kind == "park-sweep" && n.Holder != nil { |
| 2464 | // CoverParkedPredecessors at the acknowledgement (claim rev 6.21 |
| 2465 | // item 6): what the sweep retired on its way |
| 2466 | w.coverParkedPredecessors(n, o.Region, w.sweptPredecessors(n, o.Region, true)) |
| 2467 | } |
| 2468 | n.Obl = append(n.Obl[:i:i], n.Obl[i+1:]...) |
| 2469 | if o.Ret == "legacy" && n.oblFor("adopt") == 0 { |
| 2470 | n.Complete = true // every region has acknowledged |
| 2471 | } |
| 2472 | if n.Holder == nil && n.Op.Kind == "parking" && n.Op.ParkSweep != "" && n.oblFor(n.Op.ID) == 0 { |
| 2473 | // FreeAfterPark (claim rev 6.15 §6.2a): the holderless |
| 2474 | // sweep's last acknowledgement frees the claim again in the |
| 2475 | // same CAS, keeping quiet_after |
| 2476 | n.Op = opState{} |
| 2477 | w.stats["free after park"]++ |
| 2478 | } |
| 2479 | return true |
| 2480 | } |
| 2481 | } |
| 2482 | return false |
| 2483 | }) |
| 2484 | w.logf("box %d finished obligation %s %s %s", b.ID, o.ID, o.Kind, o.Prefix) |
| 2485 | } |
| 2486 | |
| 2487 | func (o obligation) Dest() string { |
| 2488 | if o.Kind == "park-sweep" { |
| 2489 | if o.Scope != nil { |
| 2490 | // the archive names the scope's incarnation (claim rev 6.24 §2; |
| 2491 | // R16-2: rev 7.16's destination ignored it) |
| 2492 | return fmt.Sprintf("parked/sweep/%s/%d/%s/", host, o.Scope.Inc, o.ID) |
| 2493 | } |
| 2494 | return fmt.Sprintf("parked/sweep/%s/%s/", host, o.ID) |
| 2495 | } |
| 2496 | if o.Kind == "park-replica" { |
| 2497 | return fmt.Sprintf("parked/%s/%s/%s/", o.Ret, host, o.ID) |
| 2498 | } |
| 2499 | return fmt.Sprintf("superseded/%s/%s/%s/", o.Op, host, o.ID) |
| 2500 | } |
| 2501 | |
| 2502 | func (w *world) newObl(c *claim, kind string, region, boxID int, prefix, ret string, quiet int) obligation { |
| 2503 | return obligation{ID: w.newID("ob"), Op: c.Op.ID, Kind: kind, Region: region, Box: boxID, Prefix: prefix, Ret: ret, Quiet: quiet} |
| 2504 | } |
| 2505 | |
| 2506 | // scoped attaches a chore's persisted park scope at its queuing CAS (claim |
| 2507 | // rev 6.23 §2): the location is the park prefix without its trailing "/", |
| 2508 | // evidenceAt the original evidence or fencing bound the caller cites, inc |
| 2509 | // the archive incarnation. Under the mutation of persisted-park-scope the |
| 2510 | // chore is queued unscoped and the runner binds the cleanup's own clock, |
| 2511 | // as rev 7.13 did. |
| 2512 | func (w *world) scoped(o obligation, evidenceAt, inc int) obligation { |
| 2513 | if !on("persisted-park-scope") { |
| 2514 | return o |
| 2515 | } |
| 2516 | o.Scope = &parkScope{Location: scopeLocation(o.Prefix), Inc: inc, EvidenceAt: evidenceAt, RequestLimit: requestLimitH} |
| 2517 | return o |
| 2518 | } |
| 2519 | |
| 2520 | // scopeLocation is the scope grammar's form of a park prefix. |
| 2521 | func scopeLocation(prefix string) string { return strings.TrimSuffix(prefix, "/") } |
| 2522 | |
| 2523 | // opScope is the scope an operation's own park runs under (a landing's |
| 2524 | // install park of the target region's earlier stay, a dirty return's and |
| 2525 | // a push's park of what their install supersedes): holder.location, the |
| 2526 | // holder's incarnation, the operation's own evidence time (R16-1, R17-1). |
| 2527 | func (w *world) opScope(c *claim, prefix string) *parkScope { |
| 2528 | inc := 0 |
| 2529 | if c.Holder != nil { |
| 2530 | inc = c.Holder.Inc |
| 2531 | } |
| 2532 | return &parkScope{Location: scopeLocation(prefix), Inc: inc, EvidenceAt: w.now, RequestLimit: requestLimitH} |
| 2533 | } |
| 2534 | |
| 2535 | // confine is the keep function that confines a park's listing to its |
| 2536 | // scope: a legacy instances/<host> scope excludes child incarnations. |
| 2537 | func confine(prefix string) func(string) bool { |
| 2538 | loc := scopeLocation(prefix) |
| 2539 | return func(k string) bool { return !inScope(k, loc) } |
| 2540 | } |
| 2541 | |
| 2542 | // scopeOf is the scope location a key belongs to (claim rev 6.24 §2): a |
| 2543 | // staging key's is moves/<host> (or its own lane, moves/<host>/<op id>, |
| 2544 | // which inScope tests by prefix); an instance key's is its incarnation |
| 2545 | // location, or the legacy instances/<host> for a level directory or |
| 2546 | // generation directly under the host prefix. A child incarnation never |
| 2547 | // belongs to the legacy scope, and nothing under moves/ belongs to an |
| 2548 | // instances scope. |
| 2549 | func scopeOf(k string) string { |
| 2550 | if strings.HasPrefix(k, "moves/"+host+"/") { |
| 2551 | return "moves/" + host |
| 2552 | } |
| 2553 | return scopeLocation(locationOf(k)) |
| 2554 | } |
| 2555 | |
| 2556 | // inScope: key k is within the scope location loc. |
| 2557 | func inScope(k, loc string) bool { |
| 2558 | if strings.HasPrefix(loc, "moves/"+host+"/") { |
| 2559 | return strings.HasPrefix(k, loc+"/") // one staging lane |
| 2560 | } |
| 2561 | return scopeOf(k) == loc |
| 2562 | } |
| 2563 | |
| 2564 | // incOf is the archive incarnation of a scope location (claim rev 6.24 |
| 2565 | // §2): an incarnation location's own; for the legacy location or the |
| 2566 | // staging sweep, the holder's incarnation, else the newest retired. |
| 2567 | func incOf(c *claim, loc string) int { |
| 2568 | if strings.HasPrefix(loc, "instances/"+host+"/") { |
| 2569 | n := 0 |
| 2570 | fmt.Sscanf(strings.TrimPrefix(loc, "instances/"+host+"/"), "%d", &n) |
| 2571 | if n > 0 { |
| 2572 | return n |
| 2573 | } |
| 2574 | } |
| 2575 | if c.Holder != nil { |
| 2576 | return c.Holder.Inc |
| 2577 | } |
| 2578 | if len(c.Retired) > 0 { |
| 2579 | return c.Retired[len(c.Retired)-1] |
| 2580 | } |
| 2581 | return 0 |
| 2582 | } |
| 2583 | |
| 2584 | // scopeLocationRe is claim rev 6.23's grammar for a scope's location |
| 2585 | // (the model's incarnation ids are decimal, the claim's 64 hex): exactly |
| 2586 | // instances/<host>, instances/<host>/<incarnation>, moves/<host>/<op id> |
| 2587 | // or moves/<host>, nothing else, no trailing "/". |
| 2588 | var scopeLocationRe = regexp.MustCompile(`^(instances/` + host + `(/[0-9]+)?|moves/` + host + `(/[A-Za-z0-9-]+)?)$`) |
| 2589 | |
| 2590 | // stagingBound is a staging lane's evidence_at: the seal bound, the time |
| 2591 | // after which no staging writer, its lease holder included, can still |
| 2592 | // write; the lane's quiet time is that bound plus H, so the bound is the |
| 2593 | // quiet time less H (the staging-quiet mutation's quiet time of "now" |
| 2594 | // gives a bound already past). |
| 2595 | func stagingBound(quiet int) int { return quiet - horizonH } |
| 2596 | |
| 2597 | func (c *claim) oblFor(op string) int { |
| 2598 | n := 0 |
| 2599 | for _, o := range c.Obl { |
| 2600 | if o.Op == op { |
| 2601 | n++ |
| 2602 | } |
| 2603 | } |
| 2604 | return n |
| 2605 | } |
| 2606 | |
| 2607 | // sameOp is the op.id condition every transition carries (claim C2). The |
| 2608 | // mutation drops it everywhere at once, CAS retry and closures alike. |
| 2609 | func sameOp(a, b string) bool { return !on("claim-cas-opid") || a == b } |
| 2610 | |
amadan