| 1 | package hostlife |
| 2 | |
| 3 | import ( |
| 4 | "fmt" |
| 5 | "strings" |
| 6 | ) |
| 7 | |
| 8 | // worker is one box agent's lifecycle worker: exec-lease renewal, the |
| 9 | // obligations it owns, then its role in the current operation. One call |
| 10 | // performs at most one protocol step. |
| 11 | func (w *world) worker(b *box) { w.workerMode(b, false) } |
| 12 | |
| 13 | // obligationRunner is a separate goroutine from the role worker: a box |
| 14 | // runs its chores and its operation role concurrently (CC11 is what keeps |
| 15 | // them from meeting in one namespace). |
| 16 | func (w *world) obligationRunner(b *box) { w.workerMode(b, true) } |
| 17 | |
| 18 | func (w *world) workerMode(b *box, obl bool) { |
| 19 | if b.Old { |
| 20 | return // a pre-floor executable knows nothing of claims |
| 21 | } |
| 22 | c, ok := w.readClaim(b) |
| 23 | if !ok { |
| 24 | if obl && w.claim != nil && len(w.claim.Obl) > 0 && w.calm { |
| 25 | w.logf("box %d: obligation run skipped, the agent cannot read the claim (down=%v paused=%v frozen=%v partitioned=%v destroyed=%v)", b.ID, b.AgentDown, b.AgentPaused, b.Frozen, b.Partitioned, b.Destroyed) |
| 26 | } |
| 27 | return |
| 28 | } |
| 29 | if obl { |
| 30 | if c == nil || (contains(c.Lost, b.ID) && on("lost-quarantine")) { |
| 31 | if c != nil && w.calm { |
| 32 | w.logf("box %d: obligation run refused, the box is declared lost", b.ID) |
| 33 | } |
| 34 | return |
| 35 | } |
| 36 | for _, o := range c.Obl { |
| 37 | if w.ownsObligation(o, b) { |
| 38 | w.doObligation(b, c, o) |
| 39 | return |
| 40 | } |
| 41 | } |
| 42 | w.sweep(b, c) |
| 43 | return |
| 44 | } |
| 45 | if b.stale != nil && w.rng.Intn(6) == 0 { |
| 46 | // a stalled worker of this box resumes with the claim it read long ago |
| 47 | old := b.stale |
| 48 | b.stale = nil |
| 49 | w.resumeStale(b, old) |
| 50 | return |
| 51 | } else if w.rng.Intn(10) == 0 { |
| 52 | b.stale = c.clone() |
| 53 | } |
| 54 | changed, pending := w.fenceStep(b, c) |
| 55 | if changed { |
| 56 | return // the regional worker's termination fence (region HA §5.9), before its role |
| 57 | } |
| 58 | // a relocation participant's fence this worker could not consume yet |
| 59 | // (a dead target with no spare before its deadline): Worker.Step |
| 60 | // returns w.replace(...) before the phase switch, so the regional |
| 61 | // transitions wait; the participant work that does not go through |
| 62 | // Worker.Step still runs (rev 8, Astra round 5) |
| 63 | b.fencePending = pending |
| 64 | w.role(b, c) |
| 65 | b.fencePending = false |
| 66 | } |
| 67 | |
| 68 | // initRecord is claim rev 6.5 §6.4's initialization (R5-4): a box creates |
| 69 | // or refreshes its local stamp record only from a fresh claim read, and |
| 70 | // only when that claim names it as op.exec or as the landing participant; |
| 71 | // it sets active_stamp = holder.stamp. A create's first wake, adoption on |
| 72 | // the serving box, and a lost record on restart all go through here. A |
| 73 | // stamp this box has retired is never made active again. |
| 74 | func (w *world) initRecord(b *box, c *claim) { |
| 75 | if c == nil || c.Holder == nil || c.Holder.Stamp == 0 || c.Ver != w.claim.Ver { |
| 76 | return |
| 77 | } |
| 78 | inc := c.Holder.Inc |
| 79 | if !(c.Op.Exec == b.ID || c.Op.Kind == "landing" && c.Op.Tgt == b.ID) { |
| 80 | return |
| 81 | } |
| 82 | if b.stampOf(inc) == c.Holder.Stamp { |
| 83 | return |
| 84 | } |
| 85 | if s := b.Stamps[inc]; s != nil && contains(s.Retired, c.Holder.Stamp) { |
| 86 | return |
| 87 | } |
| 88 | w.activate(b, inc, c.Holder.Stamp, c.Holder.Gen) |
| 89 | c.readStamp = c.Holder.Stamp |
| 90 | w.logf("box %d: local record for inc %d initialized from holder.stamp %d", b.ID, inc, c.Holder.Stamp) |
| 91 | } |
| 92 | |
| 93 | // activate is activation on box b (claim rev 6.5 §6.4): the old active |
| 94 | // stamp retires, the new one is set, and the artifacts of the generation |
| 95 | // the claim names, installed under the predecessor's stamp, are tagged |
| 96 | // with the new one (R6-4: a dirty rollback's return database is installed |
| 97 | // under S and tagged R when R activates). The oracle refuses a stamp the |
| 98 | // claim issued for another box's installation (INV-6). |
| 99 | func (w *world) activate(b *box, inc, stamp int, gen string) { |
| 100 | if bx, ok := w.stampBox[stamp]; ok && bx != b.ID { |
| 101 | w.violate("INV-6: box %d activates stamp %d, issued for box %d's installation (claim rev 6.5: holder.stamp names the installation op.exec should run)", b.ID, stamp, bx) |
| 102 | } |
| 103 | b.install(inc, stamp) |
| 104 | if b.DB != nil && b.DB.Inc == inc && b.DB.Gen == gen { |
| 105 | b.DB.Stamp = stamp |
| 106 | } |
| 107 | if b.Files != nil && b.Files.Inc == inc && b.Files.Gen == gen { |
| 108 | b.Files.Stamp = stamp |
| 109 | } |
| 110 | } |
| 111 | |
| 112 | // issueStamp is the stamp an installing CAS writes into holder.stamp: the |
| 113 | // gen the CAS produces, recorded as box's (claim rev 6.5 §2). |
| 114 | func (w *world) issueStamp(n *claim, box int) int { |
| 115 | s := w.nextGen() |
| 116 | if w.stampBox == nil { |
| 117 | w.stampBox = map[int]int{} |
| 118 | } |
| 119 | w.stampBox[s] = box |
| 120 | return s |
| 121 | } |
| 122 | |
| 123 | func dbStamp(d *localDB) int { |
| 124 | if d == nil { |
| 125 | return 0 |
| 126 | } |
| 127 | return d.Stamp |
| 128 | } |
| 129 | |
| 130 | func filesStamp(f *filesDir) int { |
| 131 | if f == nil { |
| 132 | return 0 |
| 133 | } |
| 134 | return f.Stamp |
| 135 | } |
| 136 | |
| 137 | // resumeStale is a stalled worker goroutine of b resuming with a claim it |
| 138 | // read long ago (rev 4's actor). With claim C3's revalidation it re-reads |
| 139 | // under its lease, sees the world moved, and stops; the mutation acts on |
| 140 | // the old read. A witness calls this directly (witness_test.go). |
| 141 | func (w *world) resumeStale(b *box, old *claim) { |
| 142 | c, ok := w.readClaim(b) |
| 143 | if !ok { |
| 144 | return |
| 145 | } |
| 146 | if on("revalidate-before-effect") && !sameStep(old, c) { |
| 147 | return |
| 148 | } |
| 149 | // the revalidation re-reads the box's own record as well: the stamp the |
| 150 | // resumed step carries is the one active now (claim rev 6.5 §6.4; seed |
| 151 | // 67671 / 100,000: a stalled probe step carried the stamp active at its |
| 152 | // read, before a cold admission's installing CAS, and the box's record |
| 153 | // refused its own probe) |
| 154 | old = old.clone() |
| 155 | old.readStamp = c.readStamp |
| 156 | w.role(b, old) |
| 157 | } |
| 158 | |
| 159 | // role is the worker's exec-lease renewal, quarantine and operation role, |
| 160 | // acting on the claim c it read. |
| 161 | func (w *world) role(b *box, c *claim) { |
| 162 | w.initRecord(b, c) |
| 163 | w.finishParticipation(b, c) // a relocation controller whose participation ended (rev 4) |
| 164 | // exec lease (claim §7.1) |
| 165 | if w.isParticipant(c, b) && !contains(c.Lost, b.ID) { |
| 166 | if b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil { |
| 167 | w.scopedCleanup(b, c) // the relocation controller's lease was lost (1b-next rev 14 finding 1) |
| 168 | } |
| 169 | if b.ExecUntil == 0 || b.ExecUntil-w.localNow(b) < execRenew { |
| 170 | b.ExecUntil = w.localNow(b) + execTTL |
| 171 | w.execLease[b.ID] = execLease{Op: c.Op.ID, Expires: w.now + execTTL} |
| 172 | } |
| 173 | if c.Op.Kind == "relocating" { |
| 174 | w.recordAck(b, c) // exec_ack.<box>, durable (rev 4) |
| 175 | if cur := w.claim; cur != nil && cur.Ver != c.Ver { |
| 176 | nc := cur.clone() |
| 177 | nc.readStamp = c.readStamp |
| 178 | c = nc |
| 179 | } |
| 180 | } |
| 181 | } else if c == nil || !w.isParticipant(c, b) { |
| 182 | b.ExecUntil = 0 |
| 183 | } |
| 184 | if !b.Old && b.DB != nil && b.DB.Epoch != b.Epoch && len(b.Procs) > 0 { |
| 185 | // the agent restarted under a running app whose session's |
| 186 | // replicator died with the old agent (claim rev 6.10: a new agent |
| 187 | // process is a new session): the app is stopped, and the next wake |
| 188 | // is a new admission that declares what it acknowledged meanwhile |
| 189 | b.Procs = nil |
| 190 | w.stats["app stopped: its session's agent is gone"]++ |
| 191 | w.logf("box %d: the agent restarted under a running app of a dead session; the app is stopped", b.ID) |
| 192 | } |
| 193 | if c == nil { |
| 194 | return |
| 195 | } |
| 196 | if contains(c.Lost, b.ID) && on("lost-quarantine") { |
| 197 | // a quarantined box retires whatever it kept before anything else |
| 198 | if b.Row != nil || b.DB != nil || b.Files != nil || len(b.Procs) > 0 { |
| 199 | if w.evidence(b, "quarantine") { |
| 200 | inc := 0 |
| 201 | if b.DB != nil { |
| 202 | inc = b.DB.Inc |
| 203 | } |
| 204 | if w.localPark(b, inc) { // the quarantine retires the installation |
| 205 | b.Row = nil |
| 206 | b.owedDischarge = "" // a lost box's participation binds nothing (ruling 77) |
| 207 | } |
| 208 | } |
| 209 | } |
| 210 | return |
| 211 | } |
| 212 | if c.Holder == nil { |
| 213 | return |
| 214 | } |
| 215 | if b.open != nil { |
| 216 | // an admitted session finishing its start (rev 7.11 §7.14); the |
| 217 | // operation's own step runs once the app answered readiness |
| 218 | w.resumeStart(b, c) |
| 219 | if !hasMain(b) && !b.hostFailed { |
| 220 | return // the operation's own step runs once the app answered readiness, or once the host is marked failed |
| 221 | } |
| 222 | } |
| 223 | switch c.Op.Kind { |
| 224 | case "moving": |
| 225 | if b.ID == c.Op.Src { |
| 226 | w.moveSource(b, c) |
| 227 | } else if b.ID == c.Op.Tgt { |
| 228 | w.moveReport(b, c) |
| 229 | } |
| 230 | case "reverting": |
| 231 | if b.ID == c.Op.Src { |
| 232 | w.revertSource(b, c) |
| 233 | } |
| 234 | case "landing": |
| 235 | if b.ID == c.Op.Tgt { |
| 236 | w.landTarget(b, c) |
| 237 | } else if b.ID == c.Op.Src { |
| 238 | w.holdSourceRow(b, c) |
| 239 | } |
| 240 | case "restoring": |
| 241 | if b.ID == c.Op.Tgt { |
| 242 | w.releaseTarget(b, c) |
| 243 | } else if b.ID == c.Op.Src { |
| 244 | w.recoverSource(b, c) |
| 245 | } |
| 246 | case "pushing": |
| 247 | if b.ID == c.Op.Src { |
| 248 | w.pushStep(b, c) |
| 249 | } |
| 250 | case "relocating": |
| 251 | w.relocate(b, c) |
| 252 | case "serving": |
| 253 | w.completionPass(b, c) // a relocation completed on this box and not yet marked (ruling 80) |
| 254 | } |
| 255 | } |
| 256 | |
| 257 | // convertToRelocating is the termination fence's conversion of a pre-flip |
| 258 | // move (moves rev 7.20 §7.10; claim rev 6.28 §4): one CAS records the |
| 259 | // fence (fence.<box>), sets exec none, keeps holder.stamp and |
| 260 | // holder.generation, starts relocating at pinning in the source's region, |
| 261 | // touches no existing obligation, and queues the staging lane's park for |
| 262 | // the target region when the move had not queued it yet (as an abort |
| 263 | // would). The fenced source is on lost and leaves the participants. |
| 264 | func (w *world) convertToRelocating(c *claim, src *box) { |
| 265 | op := c.Op |
| 266 | if w.cas(c, func(n *claim) bool { |
| 267 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.Committed { |
| 268 | return false |
| 269 | } |
| 270 | n.Op.Kind, n.Op.Phase, n.Op.Exec, n.Op.FenceBox = "relocating", "pinning", -1, src.ID |
| 271 | n.Op.RelocReason, n.Op.FenceAt, n.Op.Tgt, n.Op.Src = "failover", w.now, -1, src.ID |
| 272 | n.Op.Pin, n.Op.Prepared, n.Op.Receipt, n.Op.Release = nil, "", nil, nil |
| 273 | // the files the relocation installs: region HA §5.6's files |
| 274 | // publisher kept the source's files directory in S3 as it served |
| 275 | // (L17); this model has no publisher, so the fence takes the |
| 276 | // source's directory as published (a stand-in, moves §14.2.6) |
| 277 | if src.Files != nil { |
| 278 | n.Op.StopFiles = src.Files.clone() |
| 279 | } |
| 280 | var parts []int |
| 281 | for _, p := range n.Op.Parts { |
| 282 | if p != src.ID { |
| 283 | parts = append(parts, p) |
| 284 | } |
| 285 | } |
| 286 | n.Op.Parts = parts |
| 287 | n.Lost = addUniq(n.Lost, src.ID) |
| 288 | if c.oblFor(op.ID) == 0 { |
| 289 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc)) |
| 290 | } |
| 291 | return true |
| 292 | }) { |
| 293 | w.stats["relocation: source fenced before the flip"]++ |
| 294 | w.logf("move %s: source box %d terminated before the flip; converted to relocating at pinning in region %d", op.ID, src.ID, op.SrcRegion) |
| 295 | } |
| 296 | } |
| 297 | |
| 298 | // relocate is the relocation's worker on a box of the operation's region |
| 299 | // (region HA §5.8's phase table; claim rev 6.30 §4): every live, enrolled |
| 300 | // box of the region that is not on lost may act, and each phase's CAS is |
| 301 | // conditional on the operation, the phase and the ETag, so two workers |
| 302 | // racing make the same CAS and one fails. The phases live in |
| 303 | // regionha_test.go. A stuck operation waits for the operator. |
| 304 | func (w *world) relocate(b *box, c *claim) { |
| 305 | op := c.Op |
| 306 | if b.Region != op.SrcRegion || b.Destroyed || contains(c.Lost, b.ID) || !b.Enrolled || op.Stuck != "" || w.haOff { |
| 307 | return // with region HA off no relocation pass runs, only the completion pass (ruling 82) |
| 308 | } |
| 309 | if b.fencePending && on("reloc-fence-pending-blocks") { |
| 310 | // Worker.Step returned w.replace(...) before its phase switch (rev |
| 311 | // 8, Astra round 5): no regional transition while a participant's |
| 312 | // fence is pending. A live executor's own release sequence is the |
| 313 | // participant's (1b-next Task 7), not Worker.Step's, so it runs. |
| 314 | // The mutation goes on into the phase switch, as rev 7 did |
| 315 | w.stats["relocation: phase held, a participant's fence is pending"]++ |
| 316 | if (op.Phase == "releasing" || op.Phase == "probe-failed") && b.ID == op.Exec && (op.Release == nil || op.Release.Box != op.Exec) { |
| 317 | w.relocJournalRelease(b, c, true) |
| 318 | } |
| 319 | return |
| 320 | } |
| 321 | switch op.Phase { |
| 322 | case "starting": |
| 323 | w.relocStarting(b, c) |
| 324 | case "releasing": |
| 325 | w.relocReleasing(b, c) |
| 326 | case "pinning": |
| 327 | w.relocPinning(b, c) |
| 328 | case "installing": |
| 329 | w.relocInstalling(b, c) |
| 330 | case "installed": |
| 331 | w.relocInstalled(b, c) |
| 332 | case "probing": |
| 333 | w.relocProbing(b, c) |
| 334 | case "probe-failed": |
| 335 | w.relocProbeFailed(b, c) |
| 336 | case "probed": |
| 337 | w.relocSettle(b, c) |
| 338 | } |
| 339 | } |
| 340 | |
| 341 | type execLease struct { |
| 342 | Op string |
| 343 | Expires int |
| 344 | } |
| 345 | |
| 346 | func (w *world) leasesSeen(c *claim) bool { |
| 347 | for _, id := range c.Op.Parts { |
| 348 | l, ok := w.execLease[id] |
| 349 | if !ok || l.Op != c.Op.ID || l.Expires < w.now { |
| 350 | return false |
| 351 | } |
| 352 | } |
| 353 | return true |
| 354 | } |
| 355 | |
| 356 | // stagingLease: the shared staging-work lease (claim C3), per destination |
| 357 | // region: the target's for a move's staging, the source's for a dirty |
| 358 | // rollback's return lane. |
| 359 | func (w *world) takeStaging(c *claim, b *box) bool { |
| 360 | return w.takeStagingFor(c.Op.TgtRegion, c.Op.ID, b) |
| 361 | } |
| 362 | |
| 363 | func (w *world) takeStagingFor(region int, opID string, b *box) bool { |
| 364 | l := w.staging[region] |
| 365 | if l.Op != "" && l.Op != opID && l.Until >= w.now { |
| 366 | return false |
| 367 | } |
| 368 | w.staging[region] = stagingLease{Op: opID, Box: b.ID, Until: w.now + stagingTTL} |
| 369 | return true |
| 370 | } |
| 371 | |
| 372 | type stagingLease struct { |
| 373 | Op string |
| 374 | Box int |
| 375 | Until int |
| 376 | } |
| 377 | |
| 378 | // snapshotBlob is a staging round: the source's final checkpoint over its |
| 379 | // replica, which is what the mirror copies (§7.3). With no local database |
| 380 | // (a lost disk, or never woken) it is the replica alone. |
| 381 | func (w *world) snapshotBlob(b *box, c *claim) (*blob, *blob) { |
| 382 | return w.snapshotBlobOf(b, c, c.Holder.Gen) |
| 383 | } |
| 384 | |
| 385 | func (w *world) snapshotBlobOf(b *box, c *claim, gen string) (*blob, *blob) { |
| 386 | // the sealed transfer is a verified final snapshot (rev 7.5 §5.1, R4-1): |
| 387 | // the local database after its final checkpoint, every page, TXIDs 1 |
| 388 | // to its current one; with no local database (a lost disk, or never |
| 389 | // woken) the replica restored as it stands |
| 390 | var db *blob |
| 391 | if b.DB != nil { |
| 392 | // the final checkpoint writes the pending writes as the next local |
| 393 | // level-0 file and uploads what it can (litestream SyncAndWait); |
| 394 | // the snapshot is then at that position, and carries every page |
| 395 | // whether or not the upload has landed. While the admitted session's |
| 396 | // start has not opened replication there is no replicator to sync: |
| 397 | // the snapshot carries the pages as they stand and nothing is |
| 398 | // uploaded into the live generation (rev 7.11; seed 17056 / 30,000: |
| 399 | // a round's forced sync wrote the first upload's key with other |
| 400 | // bytes while the start's own PUT was still in flight) |
| 401 | if b.open == nil && (b.Old || b.DB.Epoch == b.Epoch) { |
| 402 | if len(b.DB.Pending) > 0 { |
| 403 | bl := w.deltaOf(b.DB, b.DB.Inc) |
| 404 | b.DB.Queue = append(b.DB.Queue, bl) |
| 405 | b.DB.Pos, b.DB.Pending = bl.Max, nil |
| 406 | } |
| 407 | w.ship(b) |
| 408 | } |
| 409 | db = snapshotOf(b.DB, b.DB.Inc) |
| 410 | } else if sv := w.sealVersion(c.Op.SealDB); sv != nil && c.Op.Kind == "restoring" { |
| 411 | // a target that never restored a database returns the seal it was |
| 412 | // given, as staged, pinned by version (rev 7.11 §7.8) |
| 413 | cp := *sv.B |
| 414 | db = &cp |
| 415 | } else { |
| 416 | // no local database (a lost disk): the remote as the next admission |
| 417 | // would restore it, a lagging marker read through the journal (rev |
| 418 | // 7.11; seeds 3987 and 14322 / 30,000 sealed an empty database over |
| 419 | // a generation whose marker had not landed, and the push lost the |
| 420 | // legacy host's writes) |
| 421 | var r *localDB |
| 422 | if gen == c.Holder.Gen { |
| 423 | r = w.restoreAsNext(b, c) |
| 424 | } else { |
| 425 | r = w.restoreCalc(b.Region, c.Holder.Inc, c.Holder.Location, gen) |
| 426 | } |
| 427 | if r != nil { |
| 428 | db = snapshotOf(r, c.Holder.Inc) |
| 429 | } else if !w.hasFiles(b, c.Holder.Location, gen) && !w.hasFiles(b, c.Holder.Location, c.Holder.Gen) { |
| 430 | db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}} // nothing was ever written |
| 431 | } else { |
| 432 | w.stats["seal refused: nothing to restore"]++ |
| 433 | return nil, nil // the remote cannot be read: no seal |
| 434 | } |
| 435 | } |
| 436 | files := &blob{Kind: "files", Inc: c.Holder.Inc, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...)} |
| 437 | if b.Files == nil { |
| 438 | if fv := w.sealVersion(c.Op.SealFiles); fv != nil && c.Op.Kind == "restoring" { |
| 439 | cp := *fv.B |
| 440 | files = &cp |
| 441 | } |
| 442 | } |
| 443 | return db, files |
| 444 | } |
| 445 | |
| 446 | // sealVersion is the staged seal object a journal ref names, or nil. |
| 447 | func (w *world) sealVersion(r ref) *version { |
| 448 | if r.Ver == 0 { |
| 449 | return nil |
| 450 | } |
| 451 | for _, v := range w.buckets[r.Region].objs[r.Key] { |
| 452 | if v.ID == r.Ver { |
| 453 | return v |
| 454 | } |
| 455 | } |
| 456 | return nil |
| 457 | } |
| 458 | |
| 459 | func (w *world) moveSource(b *box, c *claim) { |
| 460 | op := c.Op |
| 461 | switch op.Phase { |
| 462 | case "leases": |
| 463 | if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == "" { |
| 464 | // the record half of the request, after the claim's intent (C2): |
| 465 | // a crash between is finished here or by the recovery pass |
| 466 | w.rec.MoveID, w.rec.MoveState = op.ID, "" |
| 467 | return |
| 468 | } |
| 469 | if w.leasesSeen(c) { |
| 470 | w.cas(c, func(n *claim) bool { n.Op.Phase = "preparing"; return sameOp(n.Op.ID, op.ID) }) |
| 471 | } |
| 472 | case "preparing", "ready": |
| 473 | if !contains(c.Regions, op.TgtRegion) { |
| 474 | w.cas(c, func(n *claim) bool { n.Regions = addUniq(n.Regions, op.TgtRegion); return true }) |
| 475 | return |
| 476 | } |
| 477 | srcReady := !on("source-readiness") || (b.Config == c.Holder.Inc && b.Live == c.Holder.App && b.LiveInc == c.Holder.Inc && b.HasKey) |
| 478 | if op.Phase == "ready" && op.ReportOK && w.now-op.ReportAt < 60 && op.Auto && srcReady { |
| 479 | // auto-cut: admission is atomic with abort eligibility |
| 480 | if w.rec == nil || w.rec.MoveID != op.ID { |
| 481 | return |
| 482 | } |
| 483 | deadline := w.now + cutBudget |
| 484 | if !on("cut-deadline") { |
| 485 | deadline = 1 << 30 |
| 486 | } |
| 487 | w.cas(c, func(n *claim) bool { |
| 488 | if !sameOp(n.Op.ID, op.ID) || n.Op.CutAt != 0 || n.Op.Kind != "moving" { |
| 489 | return false |
| 490 | } |
| 491 | n.Op.CutAt = w.now |
| 492 | n.Op.CutDeadline = deadline |
| 493 | n.Op.PinVer = w.cfgVer[c.Holder.App] // move.inputs, pinned at admission |
| 494 | n.Op.Phase = "stopping" |
| 495 | return true |
| 496 | }) |
| 497 | return |
| 498 | } |
| 499 | if w.now-op.ReqAt > readyWait { |
| 500 | // readiness never came: abort on its own (no human) |
| 501 | w.logf("abort %s: phase=%s report=%v@%d srcReady=%v auto=%v", op.ID, op.Phase, op.ReportOK, op.ReportAt, srcReady, op.Auto) |
| 502 | w.abortMove(c, b) |
| 503 | return |
| 504 | } |
| 505 | if !w.takeStaging(c, b) { |
| 506 | return |
| 507 | } |
| 508 | db, files := w.snapshotBlob(b, c) |
| 509 | r1 := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), b: db} |
| 510 | r2 := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), b: files} |
| 511 | if w.rng.Intn(3) == 0 { |
| 512 | // this round revalidated under the staging-work lease just now, |
| 513 | // and its goroutine pauses before signing (J-N2): it may sign |
| 514 | // after the final round and the seal, but only before its lease's |
| 515 | // SafeUntil (claim C3) |
| 516 | until := w.staging[op.TgtRegion].Until |
| 517 | for _, r := range []*request{r1, r2} { |
| 518 | w.stalled = append(w.stalled, &stalledSign{box: b.ID, epoch: b.Epoch, req: r, signAt: w.now + w.pauseSign(), until: until}) |
| 519 | } |
| 520 | return |
| 521 | } |
| 522 | v1 := w.issue(b.ID, r1) |
| 523 | v2 := w.issue(b.ID, r2) |
| 524 | if v1 > 0 && v2 > 0 && op.Phase == "preparing" { |
| 525 | w.cas(c, func(n *claim) bool { n.Op.Phase = "ready"; return sameOp(n.Op.ID, op.ID) }) |
| 526 | } |
| 527 | case "aborting": |
| 528 | if c.oblFor(op.ID) == 0 { |
| 529 | w.closeMove(c, op.Src) |
| 530 | } |
| 531 | case "stopping": |
| 532 | if b.Row != nil { |
| 533 | b.Row.Status, b.Row.HoldOp = "moving", op.ID |
| 534 | } |
| 535 | if w.now > op.CutDeadline { |
| 536 | w.startRevert(c) |
| 537 | return |
| 538 | } |
| 539 | if !w.evidence(b, op.ID) { |
| 540 | return |
| 541 | } |
| 542 | d := w.digestOf(b) |
| 543 | files := b.Files.clone() |
| 544 | repl, next := "", 0 |
| 545 | if b.Row != nil { |
| 546 | repl, next = b.Row.Repl, b.Row.Next |
| 547 | } |
| 548 | w.cas(c, func(n *claim) bool { |
| 549 | if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" { |
| 550 | return false |
| 551 | } |
| 552 | n.Op.Phase, n.Op.StopAt, n.Op.StopDigest, n.Op.StopFiles = "stopped", w.now, d, files |
| 553 | n.Op.CutSession = n.Holder.Session // the session src_covered speaks for (rev 7.12) |
| 554 | n.Op.Repl, n.Op.Next = repl, next |
| 555 | n.Op.Exec = -1 // K3: the source gives up execution, citing its evidence |
| 556 | return true |
| 557 | }) |
| 558 | case "stopped": |
| 559 | if w.now > op.CutDeadline || b.FailArchive { |
| 560 | w.startRevert(c) |
| 561 | return |
| 562 | } |
| 563 | if !w.takeStaging(c, b) { |
| 564 | return |
| 565 | } |
| 566 | db, files := w.snapshotBlob(b, c) |
| 567 | if db == nil { |
| 568 | w.logf("%s: the source has no database and its remote cannot be restored; reverting", op.ID) |
| 569 | w.startRevert(c) |
| 570 | return |
| 571 | } |
| 572 | if op.SealDB.Ver == 0 { |
| 573 | r := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), b: db} |
| 574 | w.maybeStallDuplicate(b, op, r) |
| 575 | v := w.issue(b.ID, r) |
| 576 | if v > 0 { |
| 577 | w.cas(c, func(n *claim) bool { |
| 578 | n.Op.SealDB = ref{Region: op.TgtRegion, Key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), Ver: v} |
| 579 | return sameOp(n.Op.ID, op.ID) |
| 580 | }) |
| 581 | } |
| 582 | return |
| 583 | } |
| 584 | r := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), b: files} |
| 585 | w.maybeStallDuplicate(b, op, r) |
| 586 | v := w.issue(b.ID, r) |
| 587 | if v > 0 { |
| 588 | // does the source's own replica cover the stopped database? The |
| 589 | // final checkpoint's upload may still be in flight or lost; the |
| 590 | // seal holds every page regardless. A clean rollback restores the |
| 591 | // source from its replica only when it does (rev 7.11, INV-15; |
| 592 | // seed 17056 / 30,000) |
| 593 | covered := b.DB == nil || b.open == nil && len(b.DB.Pending) == 0 && w.remoteMax(w.buckets[b.Region], dataKey(c.Holder.Location, c.Holder.Gen, "")) >= b.DB.Pos |
| 594 | if !covered { |
| 595 | w.stats["seal: source replica short of the stop"]++ |
| 596 | } |
| 597 | w.cas(c, func(n *claim) bool { |
| 598 | if !sameOp(n.Op.ID, op.ID) { |
| 599 | return false |
| 600 | } |
| 601 | n.Op.SrcCovered = covered |
| 602 | n.Op.SealFromDB = b.DB != nil |
| 603 | n.Op.SealFiles = ref{Region: op.TgtRegion, Key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), Ver: v} |
| 604 | n.Op.SealQuiet = w.now + stagingTTL + horizonH |
| 605 | if !on("staging-quiet") { |
| 606 | n.Op.SealQuiet = w.now |
| 607 | } |
| 608 | n.Op.Phase = "sealed" |
| 609 | return true |
| 610 | }) |
| 611 | } |
| 612 | case "sealed": |
| 613 | // the flip, record first (§7.5): one CAS on the record, then the |
| 614 | // claim follows on a later step. A crash between leaves nobody |
| 615 | // with execution, which is safe; whoever next sees the record |
| 616 | // completes the claim (rev 6 §19 K4). |
| 617 | if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == op.ID && w.rec.MoveState == "landing" { |
| 618 | w.followRecord(c) |
| 619 | return |
| 620 | } |
| 621 | if w.rec != nil && w.rec.Inc == c.Holder.Inc && w.rec.MoveID == op.ID && w.rec.MoveState == "" && op.StopAt != 0 { |
| 622 | w.rec.Region, w.rec.MoveState = op.TgtRegion, "landing" |
| 623 | w.rec.Ver = w.newVer() // every record write has a fresh ETag |
| 624 | w.logf("flip %s: record names region %d", op.ID, op.TgtRegion) |
| 625 | } |
| 626 | } |
| 627 | } |
| 628 | |
| 629 | func (w *world) moveReport(b *box, c *claim) { |
| 630 | op := c.Op |
| 631 | if op.Phase != "preparing" && op.Phase != "ready" { |
| 632 | return |
| 633 | } |
| 634 | if w.now-op.ReportAt < 20 { |
| 635 | return |
| 636 | } |
| 637 | ok := b.HasKey || !on("readiness-key") |
| 638 | if on("readiness-quiet") && w.now < c.Quiet[op.TgtRegion] { |
| 639 | ok = false // T12: the landing would wait out quiet_after inside the cut window |
| 640 | } |
| 641 | w.cas(c, func(n *claim) bool { |
| 642 | if !sameOp(n.Op.ID, op.ID) { |
| 643 | return false |
| 644 | } |
| 645 | n.Op.ReportAt, n.Op.ReportOK = w.now, ok |
| 646 | return true |
| 647 | }) |
| 648 | } |
| 649 | |
| 650 | func (w *world) abortMove(c *claim, by *box) { |
| 651 | op := c.Op |
| 652 | w.cas(c, func(n *claim) bool { |
| 653 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.CutAt != 0 { |
| 654 | return false |
| 655 | } |
| 656 | n.Op.Phase = "aborting" |
| 657 | if on("abort-target-owner") { |
| 658 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc)) |
| 659 | } |
| 660 | return true |
| 661 | }) |
| 662 | if !on("abort-target-owner") && by != nil { |
| 663 | // the source tries to retire remote staging itself |
| 664 | w.issue(by.ID, &request{kind: rDelVer, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), ver: 1}) |
| 665 | } |
| 666 | } |
| 667 | |
| 668 | func (w *world) startRevert(c *claim) { |
| 669 | op := c.Op |
| 670 | w.cas(c, func(n *claim) bool { |
| 671 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" { |
| 672 | return false |
| 673 | } |
| 674 | n.Op.Kind, n.Op.Exec = "reverting", op.Src |
| 675 | if op.StopAt == 0 { |
| 676 | n.Op.RevPhase = "evidence" |
| 677 | } else { |
| 678 | n.Op.RevPhase = "verify" |
| 679 | } |
| 680 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", w.now+stagingTTL+horizonH), w.now+stagingTTL, c.Holder.Inc)) |
| 681 | return true |
| 682 | }) |
| 683 | } |
| 684 | |
| 685 | func (w *world) holdRow(b *box, status, holdOp string, c *claim) { |
| 686 | if b.Row == nil { |
| 687 | b.Row = &row{Inc: c.Holder.Inc, App: c.Holder.App} |
| 688 | } |
| 689 | b.Row.Status, b.Row.HoldOp = status, holdOp |
| 690 | if w.rng.Intn(25) == 0 { |
| 691 | // a paused goroutine's duplicate of this write, executed later |
| 692 | b.deferred = append(b.deferred, &localWrite{kind: "hold", status: status, holdOp: holdOp, epoch: b.Epoch, at: w.now + 1 + w.pauseLen()}) |
| 693 | } |
| 694 | } |
| 695 | |
| 696 | // openRow is the conditional open (moves §5.2). |
| 697 | func (w *world) openRow(b *box, holdOp string) bool { |
| 698 | if w.rng.Intn(6) == 0 { |
| 699 | d := &localWrite{kind: "open", holdOp: holdOp, epoch: b.Epoch, at: w.now + 1 + w.pauseLen()} |
| 700 | if w.rng.Intn(2) == 0 { |
| 701 | // the adversary keeps this one paused until its worst moment, |
| 702 | // or a long stall's end |
| 703 | d.worst, d.at = true, w.now+4000 |
| 704 | } |
| 705 | b.deferred = append(b.deferred, d) |
| 706 | } |
| 707 | return w.openNow(b, holdOp) |
| 708 | } |
| 709 | |
| 710 | func (w *world) openNow(b *box, holdOp string) bool { |
| 711 | if b.Row == nil { |
| 712 | return false |
| 713 | } |
| 714 | if on("hold-op-open") && b.Row.HoldOp != holdOp { |
| 715 | return false |
| 716 | } |
| 717 | b.Row.Status, b.Row.HoldOp = "ready", "" |
| 718 | return true |
| 719 | } |
| 720 | |
| 721 | // localWrite is a host-specific registry write whose goroutine passed its |
| 722 | // revalidation and then paused: it lands later, and only the conditions |
| 723 | // on the write itself (hold_op) stand between it and a newer operation. |
| 724 | type localWrite struct { |
| 725 | kind string // "open" or "hold" |
| 726 | status string |
| 727 | holdOp string |
| 728 | epoch int |
| 729 | at int // the latest it lands |
| 730 | worst bool // the scheduler wakes it at the worst moment before at |
| 731 | } |
| 732 | |
| 733 | // pauseLen is how long a paused goroutine sleeps before its registry |
| 734 | // write lands: mostly short, sometimes as long as a whole operation. A |
| 735 | // crash or reboot (b.Epoch) kills it first. |
| 736 | func (w *world) pauseLen() int { |
| 737 | if w.rng.Intn(4) == 0 { |
| 738 | return w.rng.Intn(2000) |
| 739 | } |
| 740 | return w.rng.Intn(150) |
| 741 | } |
| 742 | |
| 743 | // pauseSign is how long a staging goroutine pauses between revalidating |
| 744 | // under the staging-work lease and signing: it may outlast the lease, and |
| 745 | // then SafeUntil (claim C3) is all that stops the signature. |
| 746 | func (w *world) pauseSign() int { return w.rng.Intn(2*stagingTTL + horizonH + 1) } |
| 747 | |
| 748 | // maybeStallDuplicate models a lost-response retry in the cut's own |
| 749 | // staging round: the first attempt's goroutine paused before signing and |
| 750 | // the agent retried. The paused attempt may sign after the seal. |
| 751 | func (w *world) maybeStallDuplicate(b *box, op opState, r *request) { |
| 752 | w.maybeStallDuplicateFor(b, op.TgtRegion, r) |
| 753 | } |
| 754 | |
| 755 | func (w *world) maybeStallDuplicateFor(b *box, region int, r *request) { |
| 756 | if w.rng.Intn(3) != 0 { |
| 757 | return |
| 758 | } |
| 759 | dup := *r |
| 760 | w.stalled = append(w.stalled, &stalledSign{box: b.ID, epoch: b.Epoch, req: &dup, signAt: w.now + w.pauseSign(), until: w.staging[region].Until}) |
| 761 | } |
| 762 | |
| 763 | type stalledSign struct { |
| 764 | box int |
| 765 | epoch int |
| 766 | req *request |
| 767 | signAt int |
| 768 | until int |
| 769 | checked bool // passed its SafeUntil check and signed, not yet sent |
| 770 | checkedAt int |
| 771 | retry bool // a retry of a send that failed transiently |
| 772 | } |
| 773 | |
| 774 | // runStalled lets paused goroutines resume: deferred registry writes and |
| 775 | // staging rounds that pause between revalidation and signing. |
| 776 | func (w *world) runStalled() { |
| 777 | var keep []*stalledSign |
| 778 | for _, s := range w.stalled { |
| 779 | b := w.boxByID(s.box) |
| 780 | if b == nil || b.Destroyed || b.Epoch != s.epoch { |
| 781 | continue // the goroutine died with its agent |
| 782 | } |
| 783 | if s.signAt > w.now || b.Frozen || b.AgentPaused { |
| 784 | keep = append(keep, s) |
| 785 | continue |
| 786 | } |
| 787 | if s.checked { |
| 788 | // the request is sent now, after a pause that a whole-VM |
| 789 | // freeze can stretch without bound. Signed at the check, its |
| 790 | // timestamp is the check's, and S3 refuses it once older than |
| 791 | // the signature-age limit (model 18); signed at send time, it |
| 792 | // is accepted however late |
| 793 | resigned := s.retry && !on("sdk-no-resign") // the SDK's retry refreshed the signature |
| 794 | if on("sign-at-check") && !resigned && w.now > s.checkedAt+sigAge { |
| 795 | w.stats["stale signature refused"]++ |
| 796 | continue |
| 797 | } |
| 798 | if !s.retry && w.rng.Intn(3) == 0 { |
| 799 | // a transient error: the SDK retries the same request later, |
| 800 | // after a pause a frozen box can stretch (CC19) |
| 801 | s.retry, s.signAt = true, w.now+1+w.pauseLen() |
| 802 | keep = append(keep, s) |
| 803 | continue |
| 804 | } |
| 805 | w.issue(b.ID, s.req) |
| 806 | continue |
| 807 | } |
| 808 | margin := maxSkew + maxPause |
| 809 | if on("sign-at-check") { |
| 810 | margin = maxSkew + sigAge // SafeUntil leaves room for the signature's whole valid life |
| 811 | } |
| 812 | if on("sign-before-safeuntil") && w.localNow(b) > s.until-margin { |
| 813 | continue // checked SafeUntil immediately before signing: too late |
| 814 | } |
| 815 | // the check passed and the request is signed; the goroutine may |
| 816 | // still pause before sending it |
| 817 | s.checked, s.checkedAt, s.signAt = true, w.now, w.now+w.rng.Intn(maxPause+1) |
| 818 | keep = append(keep, s) |
| 819 | } |
| 820 | w.stalled = keep |
| 821 | for _, b := range w.boxes { |
| 822 | if b.Destroyed || b.Frozen || b.AgentPaused || b.AgentDown { |
| 823 | continue |
| 824 | } |
| 825 | var rest []*localWrite |
| 826 | for _, d := range b.deferred { |
| 827 | if d.epoch != b.Epoch { |
| 828 | continue |
| 829 | } |
| 830 | // an adversarial wake-up: a stale open resumes while a newer |
| 831 | // operation holds the row, at its worst moment: after that |
| 832 | // operation committed and stopped re-asserting the hold |
| 833 | worst := d.worst && d.kind == "open" && b.Row != nil && b.Row.HoldOp != "" && b.Row.HoldOp != d.holdOp && |
| 834 | w.claim != nil && w.claim.Op.Committed && w.rng.Intn(3) == 0 |
| 835 | if d.at > w.now && !worst { |
| 836 | rest = append(rest, d) |
| 837 | continue |
| 838 | } |
| 839 | switch d.kind { |
| 840 | case "open": |
| 841 | w.openNow(b, d.holdOp) |
| 842 | case "hold": |
| 843 | if b.Row != nil && (!on("hold-op-open") || b.Row.HoldOp == "" || b.Row.HoldOp == d.holdOp) { |
| 844 | b.Row.Status, b.Row.HoldOp = d.status, d.holdOp |
| 845 | } |
| 846 | } |
| 847 | } |
| 848 | b.deferred = rest |
| 849 | } |
| 850 | } |
| 851 | |
| 852 | func (w *world) revertSource(b *box, c *claim) { |
| 853 | op := c.Op |
| 854 | if !on("revert-journal") && !b.mem[op.ID+"rev"] && (op.RevPhase == "probing" || op.RevPhase == "probed") { |
| 855 | // the journal is not trusted after a restart: verify again |
| 856 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = "verify"; return sameOp(n.Op.ID, op.ID) }) |
| 857 | return |
| 858 | } |
| 859 | if op.RevPhase != "probed" && op.RevPhase != "opened" { |
| 860 | w.holdRow(b, "restoring", op.ID, c) |
| 861 | } |
| 862 | switch op.RevPhase { |
| 863 | case "evidence": |
| 864 | if w.evidence(b, op.ID) { |
| 865 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 866 | } |
| 867 | case "verify": |
| 868 | if !w.evidence(b, op.ID) { |
| 869 | return |
| 870 | } |
| 871 | ph := "probing" |
| 872 | if b.DB == nil { |
| 873 | // the source lost its disk since it stopped (R8-2): nothing to |
| 874 | // verify; its writer is admitted like any lost replication |
| 875 | // state, a warm continuation of the intact generation or a cold |
| 876 | // admission (claim rev 6.8 §6.4a) |
| 877 | w.logf("revert: box %d's database is gone; the probe restores it", b.ID) |
| 878 | w.stats["revert: source database lost"]++ |
| 879 | } else if w.digestOf(b) != op.StopDigest { |
| 880 | ph = "stuck" |
| 881 | w.logf("revert: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest) |
| 882 | } |
| 883 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = ph; return sameOp(n.Op.ID, op.ID) }) |
| 884 | case "probing": |
| 885 | b.mem[op.ID+"rev"] = true |
| 886 | if !w.admits(b, c, op.ID) { |
| 887 | return |
| 888 | } |
| 889 | if b.HasKey { // RenderHeld + MaterializeHeld before every probe |
| 890 | b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc |
| 891 | } |
| 892 | if hasMain(b) || w.spawn(b, c, c.Holder.Gen, true) { |
| 893 | w.probeAccepted(b, c) |
| 894 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probed"; return sameOp(n.Op.ID, op.ID) }) |
| 895 | return |
| 896 | } |
| 897 | if b.hostFailed && !b.failCounted { |
| 898 | b.failCounted = true // a host the retry policy marked failed: one failed try for the operation |
| 899 | } else if b.open != nil || b.hostFailed { |
| 900 | return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try |
| 901 | } |
| 902 | w.cas(c, func(n *claim) bool { |
| 903 | n.Op.ProbeTries++ |
| 904 | if n.Op.ProbeTries >= probeTries*3 { |
| 905 | n.Op.RevPhase = "stuck" |
| 906 | } |
| 907 | return sameOp(n.Op.ID, op.ID) |
| 908 | }) |
| 909 | case "probed": |
| 910 | if b.open != nil { |
| 911 | return // the source's session has not opened replication yet (rev 7.11 §7.14) |
| 912 | } |
| 913 | if !hasMain(b) { |
| 914 | w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11) |
| 915 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 916 | return |
| 917 | } |
| 918 | if w.openRow(b, op.ID) { |
| 919 | w.cas(c, func(n *claim) bool { n.Op.RevPhase = "opened"; return sameOp(n.Op.ID, op.ID) }) |
| 920 | } |
| 921 | case "opened": |
| 922 | if c.oblFor(op.ID) == 0 { |
| 923 | w.closeMove(c, op.Src) |
| 924 | } |
| 925 | } |
| 926 | } |
| 927 | |
| 928 | func (w *world) holdSourceRow(b *box, c *claim) { |
| 929 | if b.Row != nil && !c.Op.Committed { |
| 930 | b.Row.Status, b.Row.HoldOp = "moving", c.Op.ID |
| 931 | } |
| 932 | } |
| 933 | |
| 934 | // rollback is record first (§7.8): one CAS on the record, landing → |
| 935 | // rolling_back, guarded by move id. The commit is a CAS on the same record |
| 936 | // field, so exactly one of them wins. The claim follows on a later step. |
| 937 | func (w *world) rollback(c *claim, why string) { |
| 938 | op := c.Op |
| 939 | r := w.rec |
| 940 | if r == nil || r.Inc != c.Holder.Inc || r.MoveID != op.ID || r.MoveState != "landing" { |
| 941 | return |
| 942 | } |
| 943 | r.MoveState, r.Region, r.RollbackID, r.Reason = "rolling_back", op.SrcRegion, w.newID("rb"), why |
| 944 | r.Ver = w.newVer() |
| 945 | w.logf("rollback %s: %s", op.ID, why) |
| 946 | w.stats["why: "+why+" @"+op.LandPhase]++ |
| 947 | } |
| 948 | |
| 949 | // followRecord completes the claim half of a record transition the claim |
| 950 | // has not caught up with: the flip, the commit, a rollback. Any actor that |
| 951 | // reads both may do it (rev 6 §19 K4): the CAS is fully determined by the |
| 952 | // record and is conditional on the op id and the claim's prior state. |
| 953 | func (w *world) followRecord(c *claim) bool { |
| 954 | r := w.rec |
| 955 | if c == nil || c.Holder == nil || r == nil || r.Inc != c.Holder.Inc || r.MoveID == "" || r.MoveID != c.Op.ID { |
| 956 | return false |
| 957 | } |
| 958 | op := c.Op |
| 959 | switch { |
| 960 | case r.MoveState == "landing" && op.Kind == "moving" && op.Phase == "sealed": |
| 961 | return w.cas(c, func(n *claim) bool { |
| 962 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "moving" || n.Op.Phase != "sealed" { |
| 963 | return false |
| 964 | } |
| 965 | n.Op.Kind, n.Op.Exec, n.Op.LandPhase, n.Op.FlipAt = "landing", op.Tgt, "checking", w.now |
| 966 | n.Op.SrcStamp = n.Holder.Stamp // the source installation's stamp, kept in the journal for the return (claim rev 6.5) |
| 967 | n.Op.SrcGen = n.Holder.Gen // the source's generation at the cut, which a clean rollback's return names again (rev 7.10) |
| 968 | n.Holder.Stamp = w.issueStamp(n, op.Tgt) // holder.stamp: the gen this CAS produces, the landing's installation stamp (claim rev 6.5 §2) |
| 969 | n.Op.FlipGen = n.Holder.Stamp |
| 970 | return true |
| 971 | }) |
| 972 | case r.MoveState == "committed" && op.Kind == "landing" && !op.Committed: |
| 973 | return w.cas(c, func(n *claim) bool { |
| 974 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "landing" || n.Op.Committed { |
| 975 | return false |
| 976 | } |
| 977 | n.Op.Committed = true // holder.generation moved at the probe's cold admission (claim rev 6.9) |
| 978 | q := op.StopAt + horizonH |
| 979 | if !on("quiet-before-ack") { |
| 980 | q = 0 |
| 981 | } |
| 982 | n.Obl = append(n.Obl, |
| 983 | w.scoped(w.newObl(n, "park-superseded", op.SrcRegion, -1, c.Holder.Location, "", q), op.StopAt, c.Holder.Inc), |
| 984 | w.newObl(n, "park-local", -1, op.Src, "", "", 0), |
| 985 | w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", op.SealQuiet), stagingBound(op.SealQuiet), c.Holder.Inc)) |
| 986 | if q > n.Quiet[op.SrcRegion] { |
| 987 | n.Quiet[op.SrcRegion] = q |
| 988 | } |
| 989 | // the source region's whole location is parked: its uncovered |
| 990 | // predecessors are covered here, in the same CAS, with no chore |
| 991 | // (claim rev 6.14 §6.4a item 6, CoverRegionPredecessors) |
| 992 | w.coverRegionPredecessors(n, op.SrcRegion, "commit") |
| 993 | return true |
| 994 | }) |
| 995 | case r.MoveState == "rolling_back" && op.Kind == "landing": |
| 996 | return w.cas(c, func(n *claim) bool { |
| 997 | if !sameOp(n.Op.ID, op.ID) || n.Op.Kind != "landing" || n.Op.Committed { |
| 998 | return false |
| 999 | } |
| 1000 | n.Op.Kind, n.Op.RollbackID, n.Op.SrcPhase, n.Op.Phase, n.Op.Reason = "restoring", r.RollbackID, "waiting", "releasing", r.Reason |
| 1001 | return true |
| 1002 | }) |
| 1003 | } |
| 1004 | return false |
| 1005 | } |
| 1006 | |
| 1007 | func (w *world) landTarget(b *box, c *claim) { |
| 1008 | op := c.Op |
| 1009 | if c.Op.Exec != b.ID { |
| 1010 | return |
| 1011 | } |
| 1012 | loc := c.Holder.Location |
| 1013 | g := installGen(&op) // the generation this landing prepares; the probe's cold admission makes it holder.generation (claim rev 6.9) |
| 1014 | if !op.Committed { |
| 1015 | if w.rec != nil && w.rec.MoveID == op.ID && w.rec.MoveState != "landing" { |
| 1016 | w.followRecord(c) // a commit or rollback the claim has not caught up with |
| 1017 | return |
| 1018 | } |
| 1019 | w.holdRow(b, "landing", op.ID, c) |
| 1020 | if w.now > op.CutDeadline { |
| 1021 | w.rollback(c, "cut deadline") // N15: every phase before the commit |
| 1022 | return |
| 1023 | } |
| 1024 | } |
| 1025 | switch op.LandPhase { |
| 1026 | case "checking": |
| 1027 | // the new installation sets this box's active stamp to its own |
| 1028 | // captured stamp, the claim gen of the CAS that made the landing |
| 1029 | // current, under the guard, on a fresh read showing its operation |
| 1030 | // current and this box the landing participant (claim rev 6.5 |
| 1031 | // §6.4, R3-3, R4-2): a move back to a region the instance left is |
| 1032 | // admitted at its own stamp while the retired installation's |
| 1033 | // writers stay refused |
| 1034 | if fc, ok := w.readClaim(b); ok && fc.Op.ID == op.ID && fc.Op.Kind == "landing" && fc.Op.Tgt == b.ID { |
| 1035 | retiredHere := b.Stamps[c.Holder.Inc] != nil && len(b.Stamps[c.Holder.Inc].Retired) > 0 |
| 1036 | if on("fence-by-installation") || !retiredHere { |
| 1037 | b.install(c.Holder.Inc, fc.Holder.Stamp) |
| 1038 | w.logf("box %d: installation stamp %d for inc %d (%s), from holder.stamp", b.ID, fc.Holder.Stamp, c.Holder.Inc, op.ID) |
| 1039 | } |
| 1040 | } |
| 1041 | w.cas(c, func(n *claim) bool { |
| 1042 | n.Boxes = addUniq(n.Boxes, b.ID) |
| 1043 | n.Op.LandPhase = "install-park" |
| 1044 | return sameOp(n.Op.ID, op.ID) |
| 1045 | }) |
| 1046 | case "install-park": |
| 1047 | if !w.evidence(b, op.ID) { |
| 1048 | return |
| 1049 | } |
| 1050 | if w.now < c.Quiet[b.Region] && !b.mem[op.ID+"quietwait"] { |
| 1051 | // INV-9 (T12): an admitted cut never waits out quiet_after; the |
| 1052 | // readiness report said the region was quiet |
| 1053 | b.mem[op.ID+"quietwait"] = true |
| 1054 | w.violate("INV-9: the cut of %s waits on quiet_after[%d] until %d", op.ID, b.Region, c.Quiet[b.Region]) |
| 1055 | } |
| 1056 | // an earlier stay of this incarnation, in its location here |
| 1057 | if w.parkStep(b, loc, fmt.Sprintf("superseded/%s/%s/install/", op.ID, host), c.Quiet[b.Region], w.opScope(c, loc), w.keepFilesLive(b.Region, confine(loc))) { |
| 1058 | w.cas(c, func(n *claim) bool { n.Op.LandPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) }) |
| 1059 | } |
| 1060 | case "install-copy": |
| 1061 | src := op.SealDB |
| 1062 | if cur := w.buckets[b.Region].current(src.Key); cur != nil && cur.ID != src.Ver { |
| 1063 | w.stats["diag: staging newer than seal at install"]++ |
| 1064 | if sv := w.buckets[b.Region].byID(src.Key, src.Ver); sv != nil && cur.B.Max < sv.B.Max { |
| 1065 | w.stats["diag: ...and older content"]++ |
| 1066 | } |
| 1067 | } |
| 1068 | if !on("seal-by-version") { |
| 1069 | cur := w.buckets[b.Region].current(src.Key) |
| 1070 | if cur == nil { |
| 1071 | w.rollback(c, "staging missing") |
| 1072 | return |
| 1073 | } |
| 1074 | src.Ver = cur.ID |
| 1075 | } |
| 1076 | // into this move's own generation, as its snapshot at the sealed |
| 1077 | // TXID: a copy that completes late can only ever add an object to |
| 1078 | // this generation |
| 1079 | sv := w.buckets[b.Region].byID(src.Key, src.Ver) |
| 1080 | if sv == nil { |
| 1081 | w.rollback(c, "staging missing") |
| 1082 | return |
| 1083 | } |
| 1084 | if sv.B.Max == 0 && on("valid-ltx") { |
| 1085 | // an empty source (R8-3): no LTX file can carry TXID 0, so the |
| 1086 | // generation is installed empty, its marker naming no snapshot |
| 1087 | w.cas(c, func(n *claim) bool { |
| 1088 | n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0 |
| 1089 | return sameOp(n.Op.ID, op.ID) |
| 1090 | }) |
| 1091 | return |
| 1092 | } |
| 1093 | key := snapKey(loc, g, sv.B.Max) |
| 1094 | v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: src.Key, srcVer: src.Ver}) |
| 1095 | if v > 0 || !on("build-completion") { |
| 1096 | // the build journal records the snapshot's key, version and TXID |
| 1097 | // (R3-1, R4-1); the anchor follows, then completion (R7-4). The |
| 1098 | // mutation publishes from the intent, before the artifacts landed |
| 1099 | next := "install-anchor" |
| 1100 | if !on("build-completion") { |
| 1101 | next = "published" |
| 1102 | } |
| 1103 | w.cas(c, func(n *claim) bool { |
| 1104 | n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max |
| 1105 | return sameOp(n.Op.ID, op.ID) |
| 1106 | }) |
| 1107 | } |
| 1108 | case "install-anchor": |
| 1109 | if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) { |
| 1110 | w.cas(c, func(n *claim) bool { n.Op.LandPhase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| 1111 | } |
| 1112 | case "published": |
| 1113 | // the build is complete; generation.json is published by the probe's |
| 1114 | // cold admission, after readiness and before its first upload (claim |
| 1115 | // rev 6.9 §6.4a) |
| 1116 | w.cas(c, func(n *claim) bool { n.Op.LandPhase = "local"; return sameOp(n.Op.ID, op.ID) }) |
| 1117 | case "local": |
| 1118 | w.installLocal(b, c, g, op.SealFiles, op.LandPhase, func(n *claim) { n.Op.LandPhase = "config" }) |
| 1119 | case "config": |
| 1120 | if !b.HasKey && on("config-before-probe") { |
| 1121 | w.rollback(c, "configuration did not render") |
| 1122 | return |
| 1123 | } |
| 1124 | if b.HasKey { |
| 1125 | b.Config, b.CfgVer = c.Holder.Inc, w.cfgVer[c.Holder.App] |
| 1126 | if b.CfgVer != op.PinVer { |
| 1127 | // `configured`: the bundle changed after admission (§7.6) |
| 1128 | w.rollback(c, "configuration changed during the move") |
| 1129 | return |
| 1130 | } |
| 1131 | } |
| 1132 | b.Live, b.LiveInc = c.Holder.App, c.Holder.Inc |
| 1133 | w.cas(c, func(n *claim) bool { n.Op.LandPhase = "restore"; return sameOp(n.Op.ID, op.ID) }) |
| 1134 | case "restore": |
| 1135 | // the installed digest, for the settle check's dirty verdict, from a |
| 1136 | // restore of the prepared generation computed and not kept: the |
| 1137 | // probe's admission restores the database it starts on (claim rev |
| 1138 | // 6.9: a box never reuses a local database) |
| 1139 | tmp := *b |
| 1140 | tmp.DB = w.restorePinned(b.Region, c.Holder.Inc, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) |
| 1141 | if tmp.DB == nil { |
| 1142 | w.rollback(c, "restore failed") |
| 1143 | return |
| 1144 | } |
| 1145 | d := w.digestOf(&tmp) |
| 1146 | w.cas(c, func(n *claim) bool { n.Op.Installed, n.Op.LandPhase = d, "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 1147 | case "probing": |
| 1148 | if w.now > op.CutDeadline { |
| 1149 | w.rollback(c, "cut deadline") |
| 1150 | return |
| 1151 | } |
| 1152 | if !w.admits(b, c, op.ID) { |
| 1153 | return |
| 1154 | } |
| 1155 | if hasMain(b) || w.spawn(b, c, g, true) { |
| 1156 | w.probeAccepted(b, c) |
| 1157 | if !on("commit-before-open") { |
| 1158 | w.openRow(b, op.ID) |
| 1159 | } |
| 1160 | w.cas(c, func(n *claim) bool { n.Op.LandPhase, n.Op.ProbedAt = "probed", w.now; return sameOp(n.Op.ID, op.ID) }) |
| 1161 | return |
| 1162 | } |
| 1163 | if b.hostFailed { |
| 1164 | w.rollback(c, "target would not start") // the retry policy marked the host failed (rev 7.11 §7.14) |
| 1165 | return |
| 1166 | } |
| 1167 | if b.open != nil { |
| 1168 | return // admitted, its start in progress or in backoff: not a failed try |
| 1169 | } |
| 1170 | tries := op.ProbeTries + 1 |
| 1171 | w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) }) |
| 1172 | if tries >= probeTries { |
| 1173 | w.rollback(c, "target would not start") |
| 1174 | } |
| 1175 | case "probed": |
| 1176 | if !op.Committed { |
| 1177 | if w.now > op.CutDeadline { |
| 1178 | w.rollback(c, "cut deadline") |
| 1179 | return |
| 1180 | } |
| 1181 | if !hasMain(b) { |
| 1182 | // the probe died in its settle window (a crash, an agent |
| 1183 | // restart): the landing restarts only the probe (§7.14), a |
| 1184 | // writer admitted like any other (claim rev 6.9 §6.4a, R8-2): |
| 1185 | // cold, into a fresh generation restored from the dead probe's |
| 1186 | if w.spawn(b, c, c.Holder.Gen, true) { |
| 1187 | w.stats["probe restarted"]++ |
| 1188 | w.cas(c, func(n *claim) bool { n.Op.ProbedAt = w.now; return sameOp(n.Op.ID, op.ID) }) |
| 1189 | return |
| 1190 | } |
| 1191 | if b.hostFailed { |
| 1192 | w.rollback(c, "probe died") |
| 1193 | return |
| 1194 | } |
| 1195 | if b.open != nil { |
| 1196 | return // admitted, its start in progress: not a failed try |
| 1197 | } |
| 1198 | tries := op.ProbeTries + 1 |
| 1199 | w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) }) |
| 1200 | if tries >= probeTries { |
| 1201 | w.rollback(c, "probe died") |
| 1202 | } |
| 1203 | return |
| 1204 | } |
| 1205 | if b.open != nil { |
| 1206 | return // the probe's session has not opened replication yet (rev 7.11 §7.14): the settle window waits |
| 1207 | } |
| 1208 | if w.now < op.ProbedAt+15 { |
| 1209 | if w.fail["lostreturn"] && w.rng.Intn(2) == 0 { |
| 1210 | w.traffic(b) // the biased schedule: the app does real work in the settle window |
| 1211 | } |
| 1212 | return // the settle window between the two readiness checks |
| 1213 | } |
| 1214 | // the settle window (§7.7): an app that answered once, did real |
| 1215 | // work, and then fails its second readiness check. The biased |
| 1216 | // schedule (model 27) makes every target such an app, so the |
| 1217 | // rollback is dirty and a return lane follows |
| 1218 | if (b.FailSettle || w.fail["lostreturn"]) && hasMain(b) && w.digestOf(b) != op.Installed && w.rng.Intn(2) == 0 { |
| 1219 | w.logf("box %d failed the settle check for %s", b.ID, op.ID) |
| 1220 | w.rollback(c, "settle check failed") |
| 1221 | return |
| 1222 | } |
| 1223 | // the commit, record first (§7.9): a CAS on the same record |
| 1224 | // field the rollback uses, conditional on the deadline |
| 1225 | r := w.rec |
| 1226 | if r != nil && r.MoveID == op.ID && r.MoveState == "landing" && w.now <= op.CutDeadline { |
| 1227 | r.MoveState = "committed" |
| 1228 | r.Ver = w.newVer() |
| 1229 | w.logf("commit %s: record committed", op.ID) |
| 1230 | if w.now > op.CutAt+cutBudget { |
| 1231 | w.violate("INV-9: %s committed at %d, cut at %d (budget %d)", op.ID, w.now, op.CutAt, cutBudget) |
| 1232 | } |
| 1233 | } |
| 1234 | return |
| 1235 | } |
| 1236 | if !op.Opened { |
| 1237 | if w.openRow(b, op.ID) || b.Row != nil && b.Row.Status == "ready" { |
| 1238 | w.cas(c, func(n *claim) bool { n.Op.Opened = true; return sameOp(n.Op.ID, op.ID) }) |
| 1239 | } |
| 1240 | return |
| 1241 | } |
| 1242 | if c.oblFor(op.ID) == 0 { |
| 1243 | w.closeMove(c, op.Tgt) |
| 1244 | } |
| 1245 | } |
| 1246 | } |
| 1247 | |
| 1248 | // installLocal is the local install, journaled so the files marker is |
| 1249 | // written last (R2-6). seal names the files archive version. |
| 1250 | func (w *world) installLocal(b *box, c *claim, gen string, seal ref, _ string, next func(n *claim)) { |
| 1251 | if b.Files != nil && b.Files.Gen == gen && b.Files.Done && b.Files.Inc == c.Holder.Inc { |
| 1252 | if w.perRegion > 1 && !w.preparedManifestHere(b, c, gen) { |
| 1253 | return // moves §7.6 (L17, R18-1) in the two-box shape: the generation's prepared manifest first |
| 1254 | } |
| 1255 | // the marker vouches for everything before it |
| 1256 | w.cas(c, func(n *claim) bool { next(n); return true }) |
| 1257 | return |
| 1258 | } |
| 1259 | if b.Files == nil || b.Files.Gen != gen { |
| 1260 | // fresh extraction: park whatever is there first, as |
| 1261 | // retire-artifacts of the predecessor's stamp (claim rev 6.5 |
| 1262 | // §6.4): the new installation's active stamp is untouched. A |
| 1263 | // landing parks an earlier stay, never its own installation; a |
| 1264 | // dirty rollback's install on the source parks the pre-move |
| 1265 | // database while its stamp is still the active one (R6-4) |
| 1266 | for _, st := range []int{dbStamp(b.DB), filesStamp(b.Files)} { |
| 1267 | if st == 0 { |
| 1268 | continue |
| 1269 | } |
| 1270 | if c.Op.Kind == "restoring" && st != b.stampOf(c.Holder.Inc) && !b.stampRetired(c.Holder.Inc, st) { |
| 1271 | // a retired stamp's leftover is fair game: a cold admission that |
| 1272 | // issued a stamp and then failed its purge leaves the old |
| 1273 | // installation's database under a newer active stamp (rev |
| 1274 | // 7.11, the CAS before the purge; seed 65090 / 100,000) |
| 1275 | w.violate("INV-6: the dirty install on box %d parks an artifact tagged %d while the active stamp is %d (claim rev 6.5 §6.4)", b.ID, st, b.stampOf(c.Holder.Inc)) |
| 1276 | } |
| 1277 | if c.Op.Kind == "landing" && st == b.stampOf(c.Holder.Inc) { |
| 1278 | w.violate("INV-6: landing on box %d would park an artifact tagged with its own active stamp %d", b.ID, st) |
| 1279 | } |
| 1280 | w.retireArtifacts(b, c.Holder.Inc, st) |
| 1281 | } |
| 1282 | b.DB = nil |
| 1283 | b.Files = nil |
| 1284 | src := w.buckets[b.Region].byID(seal.Key, seal.Ver) |
| 1285 | if !on("seal-by-version") { |
| 1286 | src = w.buckets[b.Region].current(seal.Key) |
| 1287 | } |
| 1288 | if src == nil || src.B.Files == nil { |
| 1289 | return |
| 1290 | } |
| 1291 | if !on("files-carried") { |
| 1292 | // the kpmg outage: nothing carries the files directory, and the |
| 1293 | // landing carries on regardless |
| 1294 | b.pendingTimers = append([]int(nil), src.B.Timers...) |
| 1295 | b.Timers = append([]int(nil), b.pendingTimers...) |
| 1296 | w.cas(c, func(n *claim) bool { next(n); return true }) |
| 1297 | return |
| 1298 | } |
| 1299 | f := src.B.Files.clone() |
| 1300 | f.Gen, f.Done, f.Stamp = gen, false, b.stampOf(c.Holder.Inc) |
| 1301 | if !on("markers-last") { |
| 1302 | f.Done = true |
| 1303 | } |
| 1304 | b.Files = f |
| 1305 | b.pendingTimers = append([]int(nil), src.B.Timers...) |
| 1306 | return |
| 1307 | } |
| 1308 | // files extracted, marker not yet written: env, timers, columns, marker |
| 1309 | b.Timers = append([]int(nil), b.pendingTimers...) |
| 1310 | if src := w.buckets[b.Region].byID(seal.Key, seal.Ver); src != nil { |
| 1311 | b.Timers = append([]int(nil), src.B.Timers...) |
| 1312 | } |
| 1313 | b.Attempt = 0 |
| 1314 | b.History = nil |
| 1315 | if b.Row != nil { |
| 1316 | b.Row.Repl, b.Row.Next = c.Op.Repl, c.Op.Next |
| 1317 | } |
| 1318 | b.Files.Done = true |
| 1319 | if w.perRegion > 1 && !w.preparedManifestHere(b, c, gen) { |
| 1320 | return // the first branch finishes once the prepared manifest is acknowledged |
| 1321 | } |
| 1322 | w.cas(c, func(n *claim) bool { next(n); return true }) |
| 1323 | } |
| 1324 | |
| 1325 | // moveClose reports whether op is a move's close: a committed landing, an |
| 1326 | // opened rollback or revert, or an abort. Pushes and relocations release on |
| 1327 | // their own terms. |
| 1328 | func moveClose(op opState) bool { |
| 1329 | switch op.Kind { |
| 1330 | case "landing", "restoring", "reverting": |
| 1331 | return true |
| 1332 | case "moving": |
| 1333 | return op.Phase == "aborting" |
| 1334 | } |
| 1335 | return false |
| 1336 | } |
| 1337 | |
| 1338 | // closeExec is the box a move's close releases to: the target after a |
| 1339 | // landing, the source otherwise (§19 K8). |
| 1340 | func closeExec(op opState) int { |
| 1341 | if op.Kind == "landing" { |
| 1342 | return op.Tgt |
| 1343 | } |
| 1344 | return op.Src |
| 1345 | } |
| 1346 | |
| 1347 | // closeMove is a move's close by the box that runs it, record first (moves |
| 1348 | // rev 7.23 §7.5 and §19 K8, Paul's ruling M4). With nothing owed under the |
| 1349 | // op, the record's CAS clears move into last_move and names the executor's |
| 1350 | // region; only then does the claim release. The box may stop between the |
| 1351 | // halves (every third step here, a deterministic schedule that draws |
| 1352 | // nothing from the seed's stream). It never comes back to a record without a |
| 1353 | // move, so a closed record whose claim still holds the op is the recovery |
| 1354 | // pass's to release, from last_move (fixRecord). |
| 1355 | func (w *world) closeMove(c *claim, execBox int) { |
| 1356 | op := c.Op |
| 1357 | if c.oblFor(op.ID) != 0 { |
| 1358 | return |
| 1359 | } |
| 1360 | if !on("record-first-close") { |
| 1361 | // rev 7.22: the claim first; the recovery pass wrote the record |
| 1362 | // half after the release |
| 1363 | w.release(c, execBox) |
| 1364 | return |
| 1365 | } |
| 1366 | if !w.closeRecord(c, execBox) { |
| 1367 | return // closed already (last_move): not the box's any more |
| 1368 | } |
| 1369 | if w.now%3 == 0 { |
| 1370 | w.stats["close: stopped between the record and the release"]++ |
| 1371 | return |
| 1372 | } |
| 1373 | w.release(c, execBox) |
| 1374 | } |
| 1375 | |
| 1376 | // closeRecord is the close's record half: the CAS that clears move into |
| 1377 | // last_move and names the executor's region, conditional on the record still |
| 1378 | // carrying this move for the claim's incarnation. |
| 1379 | func (w *world) closeRecord(c *claim, execBox int) bool { |
| 1380 | r, op := w.rec, c.Op |
| 1381 | if r == nil || c.Holder == nil || r.Inc != c.Holder.Inc || r.MoveID != op.ID { |
| 1382 | return false |
| 1383 | } |
| 1384 | if b := w.boxByID(execBox); b != nil { |
| 1385 | r.Region = b.Region |
| 1386 | } |
| 1387 | r.MoveID, r.MoveState, r.RollbackID, r.LastMove = "", "", "", op.ID |
| 1388 | r.Ver = w.newVer() |
| 1389 | w.logf("close %s: record closed (last_move)", op.ID) |
| 1390 | return true |
| 1391 | } |
| 1392 | |
| 1393 | func (w *world) release(c *claim, execBox int) { |
| 1394 | op := c.Op |
| 1395 | if w.cas(c, func(n *claim) bool { |
| 1396 | if !sameOp(n.Op.ID, op.ID) { |
| 1397 | return false |
| 1398 | } |
| 1399 | if op.Kind == "relocating" && (n.Op.Kind != "relocating" || n.Op.Phase != op.Phase || n.Op.Exec != execBox) { |
| 1400 | return false // CompleteRelocation's requireRelocation (L18 finding 2) |
| 1401 | } |
| 1402 | // holder.admissions survives the release by living in the holder |
| 1403 | // (claim rev 6.12): an admission whose start is unfinished at the |
| 1404 | // release stays explained (seed 1052 / 30,000) |
| 1405 | if op.Kind == "relocating" { |
| 1406 | w.relocationLeftovers(n, op) // the abandoned installations' park-local chores (claim §4) |
| 1407 | } |
| 1408 | n.Op = opState{Kind: "serving", Exec: execBox, Parts: []int{execBox}} |
| 1409 | if op.Kind == "relocating" { |
| 1410 | // completed_relocation (ruling 56): the box and the probe's |
| 1411 | // session, unmarked until the completion pass discharges it |
| 1412 | n.Op.Completed = &completedEntry{Op: op.ID, Box: execBox, Session: n.Holder.Session} |
| 1413 | } |
| 1414 | return true |
| 1415 | }) { |
| 1416 | w.logf("released %s to serving on box %d", op.ID, execBox) |
| 1417 | if b := w.boxByID(execBox); b != nil && op.Kind == "relocating" { |
| 1418 | b.owedDischarge = op.ID // ground truth for INV-31 |
| 1419 | } |
| 1420 | if moveClose(op) && (w.rec == nil || w.rec.LastMove != op.ID || w.rec.MoveID == op.ID) { |
| 1421 | // K8 as rev 7.23 amends it: never a serving claim over a live |
| 1422 | // move. Judged only for a release CAS that succeeded: one the |
| 1423 | // op-id guard refused released nothing (fix round 2) |
| 1424 | w.violate("K8: %s released the claim before its record closed (record first, rev 7.23)", op.ID) |
| 1425 | } |
| 1426 | w.opsDone++ |
| 1427 | // a move's record closed before this (closeMove, rev 7.23); a |
| 1428 | // push's or a relocation's record half follows on a later step |
| 1429 | w.checkDurable(w.claim, "at the release of "+op.ID) |
| 1430 | if r := w.residueAtEnd(); len(r) > 0 { |
| 1431 | w.violate("INV-10: %s ended with residue %v", op.ID, r) |
| 1432 | } |
| 1433 | switch { |
| 1434 | case op.Kind == "landing": |
| 1435 | w.stats["move-done"]++ |
| 1436 | case op.Kind == "restoring" && op.LostBox != 0: |
| 1437 | w.stats["declared-lost"]++ |
| 1438 | case op.Kind == "restoring" && op.Dirty: |
| 1439 | w.stats["rollback-dirty"]++ |
| 1440 | case op.Kind == "restoring": |
| 1441 | w.stats["rollback-clean"]++ |
| 1442 | case op.Kind == "reverting": |
| 1443 | w.stats["revert"]++ |
| 1444 | case op.Kind == "moving": |
| 1445 | w.stats["abort"]++ |
| 1446 | case op.Kind == "pushing": |
| 1447 | w.stats["push"]++ |
| 1448 | case op.Kind == "relocating": |
| 1449 | w.stats["relocation"]++ |
| 1450 | } |
| 1451 | } |
| 1452 | } |
| 1453 | |
| 1454 | func (w *world) releaseTarget(b *box, c *claim) { |
| 1455 | op := c.Op |
| 1456 | if op.Exec == b.ID { |
| 1457 | w.holdRow(b, "releasing", op.RollbackID, c) |
| 1458 | if !w.evidence(b, op.RollbackID) { |
| 1459 | return |
| 1460 | } |
| 1461 | dirty := false |
| 1462 | if (op.LandPhase == "probing" || op.LandPhase == "probed") && b.DB != nil { |
| 1463 | // a target without a database never opened: nothing to return. |
| 1464 | // Since rev 7.10 the probe's admission restores the database, so |
| 1465 | // a target whose probe never started has none (seed 62 / 300: |
| 1466 | // a spurious dirty return sealed an empty database and the |
| 1467 | // source restored it) |
| 1468 | dirty = w.digestOf(b) != op.Installed |
| 1469 | } |
| 1470 | if !on("dirty-return") { |
| 1471 | dirty = false |
| 1472 | } |
| 1473 | if !op.SrcCovered && on("seal-return") { |
| 1474 | // the source's replica stops short of its stopped database: only |
| 1475 | // the seal holds every acknowledged write, so the return carries |
| 1476 | // it whether or not the target wrote (rev 7.11 §7.8) |
| 1477 | dirty = true |
| 1478 | w.stats["rollback: return carries the seal"]++ |
| 1479 | } |
| 1480 | w.cas(c, func(n *claim) bool { |
| 1481 | if !sameOp(n.Op.ID, op.ID) || n.Op.Exec != b.ID { |
| 1482 | return false |
| 1483 | } |
| 1484 | n.Op.Exec, n.Op.DirtyKnown, n.Op.Dirty, n.Op.Phase, n.Op.TgtStopAt = -1, true, dirty, "released", w.now |
| 1485 | q := w.now + horizonH |
| 1486 | if on("target-park") { |
| 1487 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-superseded", op.TgtRegion, -1, c.Holder.Location, "", q), w.now, c.Holder.Inc)) |
| 1488 | w.coverRegionPredecessors(n, op.TgtRegion, "rollback") // the target region's whole location is parked (claim rev 6.14 item 6) |
| 1489 | } |
| 1490 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.TgtRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.ID), "", op.SealQuiet), stagingBound(op.SealQuiet), c.Holder.Inc)) |
| 1491 | if !dirty { |
| 1492 | n.Obl = append(n.Obl, w.newObl(n, "park-local", -1, b.ID, "", "", 0)) |
| 1493 | } |
| 1494 | if q > n.Quiet[op.TgtRegion] { |
| 1495 | n.Quiet[op.TgtRegion] = q |
| 1496 | } |
| 1497 | return true |
| 1498 | }) |
| 1499 | return |
| 1500 | } |
| 1501 | if op.Dirty && op.Phase == "released" && op.Exec == -1 { |
| 1502 | // return the target's writes over the approved lane (M17), under |
| 1503 | // the staging-work lease of the source's region, with the same |
| 1504 | // paused signers and SafeUntil rule as the forward path |
| 1505 | if !w.takeStagingFor(op.SrcRegion, op.RollbackID, b) { |
| 1506 | return |
| 1507 | } |
| 1508 | db, files := w.snapshotBlobOf(b, c, op.ID) |
| 1509 | if db == nil { |
| 1510 | return // nothing to return yet: the remote could not be read |
| 1511 | } |
| 1512 | if op.ReturnDB.Ver == 0 { |
| 1513 | r := &request{kind: rPut, region: op.SrcRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), b: db} |
| 1514 | w.maybeStallDuplicateFor(b, op.SrcRegion, r) |
| 1515 | v := w.issue(b.ID, r) |
| 1516 | if v > 0 { |
| 1517 | fromDB := b.DB != nil |
| 1518 | w.cas(c, func(n *claim) bool { |
| 1519 | n.Op.ReturnDB = ref{Region: op.SrcRegion, Key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), Ver: v} |
| 1520 | n.Op.ReturnFromDB = fromDB |
| 1521 | return sameOp(n.Op.ID, op.ID) |
| 1522 | }) |
| 1523 | } |
| 1524 | w.loseReturning(b) // the first PUT of the lane is out, accepted or in flight |
| 1525 | return |
| 1526 | } |
| 1527 | r := &request{kind: rPut, region: op.SrcRegion, key: fmt.Sprintf("moves/%s/%s/files", host, op.RollbackID), b: files} |
| 1528 | w.maybeStallDuplicateFor(b, op.SrcRegion, r) |
| 1529 | v := w.issue(b.ID, r) |
| 1530 | if v > 0 { |
| 1531 | w.cas(c, func(n *claim) bool { |
| 1532 | if !sameOp(n.Op.ID, op.ID) { |
| 1533 | return false |
| 1534 | } |
| 1535 | n.Op.ReturnFiles = ref{Region: op.SrcRegion, Key: fmt.Sprintf("moves/%s/%s/files", host, op.RollbackID), Ver: v} |
| 1536 | n.Op.ReturnQuiet = w.now + stagingTTL + horizonH // the forward path's staging horizon (§7.3) |
| 1537 | if !on("staging-quiet") { |
| 1538 | n.Op.ReturnQuiet = w.now |
| 1539 | } |
| 1540 | n.Op.Phase = "returned" |
| 1541 | n.Obl = append(n.Obl, w.newObl(n, "park-local", -1, b.ID, "", "", 0)) |
| 1542 | return true |
| 1543 | }) |
| 1544 | return |
| 1545 | } |
| 1546 | w.loseReturning(b) // the second PUT is in flight and the return is not yet journaled |
| 1547 | } |
| 1548 | } |
| 1549 | |
| 1550 | // loseReturning is the biased schedule for model 27 (rev 7.2). In half the |
| 1551 | // seeds the cloud stops the target the moment it has begun its dirty |
| 1552 | // return and before the return is complete: a return PUT is accepted, or |
| 1553 | // still in flight, in the source's region, and the box that would finish |
| 1554 | // the lane is gone. Declare-lost then meets a partial return lane, which |
| 1555 | // the uniform schedule reached once in 100,000 seeds on rev 7 and never on |
| 1556 | // rev 7.1. A stopped box is what the operator's compute fencing verifies |
| 1557 | // (§7.10), so it is declared lost on the operator's next call. |
| 1558 | func (w *world) loseReturning(b *box) { |
| 1559 | if !w.fail["lostreturn"] || w.rng.Intn(2) != 0 { |
| 1560 | return |
| 1561 | } |
| 1562 | b.Destroyed, b.Procs = true, nil |
| 1563 | w.stats["biased: target stopped mid-return"]++ |
| 1564 | w.logf("box %d stopped mid-return (biased schedule)", b.ID) |
| 1565 | } |
| 1566 | |
| 1567 | func (w *world) recoverSource(b *box, c *claim) { |
| 1568 | op := c.Op |
| 1569 | if !on("rollback-journal") && !b.mem[op.RollbackID] && op.SrcPhase != "waiting" && op.SrcPhase != "opened" && op.SrcPhase != "stuck" { |
| 1570 | // without a durable journal a restarted source starts over |
| 1571 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "claimed"; return sameOp(n.Op.ID, op.ID) }) |
| 1572 | b.mem[op.RollbackID] = true |
| 1573 | return |
| 1574 | } |
| 1575 | b.mem[op.RollbackID] = true |
| 1576 | switch op.SrcPhase { |
| 1577 | case "waiting": |
| 1578 | if b.Row != nil && !op.Committed { |
| 1579 | b.Row.Status, b.Row.HoldOp = "restoring", op.RollbackID |
| 1580 | } |
| 1581 | if op.Exec == -1 && op.DirtyKnown && (!op.Dirty || op.Phase == "returned") { |
| 1582 | w.cas(c, func(n *claim) bool { |
| 1583 | if !sameOp(n.Op.ID, op.ID) || n.Op.Exec != -1 { |
| 1584 | return false |
| 1585 | } |
| 1586 | n.Op.Exec, n.Op.SrcPhase = op.Src, "claimed" |
| 1587 | // return to source (claim rev 6.5): the CAS that makes the |
| 1588 | // source's existing installation current again writes its |
| 1589 | // stamp back, from the journal; never the target's F |
| 1590 | if on("return-stamp") { |
| 1591 | n.Holder.Stamp = op.SrcStamp |
| 1592 | } |
| 1593 | // and, for a clean rollback, the source's generation (rev |
| 1594 | // 7.10 §7.14): the target's probe moved holder.generation to |
| 1595 | // its own; the target wrote nothing, and the source's probe |
| 1596 | // restores from the predecessor generation's remote in its |
| 1597 | // own region |
| 1598 | if !op.Dirty && n.Holder.Gen != op.SrcGen { |
| 1599 | n.Holder.Gen = op.SrcGen // "" is the origin generation |
| 1600 | } |
| 1601 | return true |
| 1602 | }) |
| 1603 | } |
| 1604 | case "claimed": |
| 1605 | w.holdRow(b, "restoring", op.RollbackID, c) |
| 1606 | next := "verify" |
| 1607 | if op.Dirty { |
| 1608 | next = "install-park" |
| 1609 | } |
| 1610 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = next; return sameOp(n.Op.ID, op.ID) }) |
| 1611 | case "verify": |
| 1612 | if !w.evidence(b, op.RollbackID) { |
| 1613 | return |
| 1614 | } |
| 1615 | ph := "config" |
| 1616 | if b.DB == nil { |
| 1617 | // the source lost its disk since it stopped (R8-2): see the |
| 1618 | // revert's verify above |
| 1619 | w.logf("recover: box %d's database is gone; the probe restores it", b.ID) |
| 1620 | w.stats["rollback: source database lost"]++ |
| 1621 | } else if w.digestOf(b) != op.StopDigest { |
| 1622 | ph = "stuck" |
| 1623 | w.logf("recover: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest) |
| 1624 | } |
| 1625 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = ph; return sameOp(n.Op.ID, op.ID) }) |
| 1626 | case "install-park": |
| 1627 | if !w.evidence(b, op.RollbackID) { |
| 1628 | return |
| 1629 | } |
| 1630 | if on("dirty-return-needs-admission") && !targetAdmitted(c) { |
| 1631 | // movesinstall.go's parking (fix rounds 3 and 4): the target never |
| 1632 | // admitted this move, so holder.generation is still the pre-move |
| 1633 | // one, the source's live state; there is no residue, and nothing |
| 1634 | // is parked. The mutation parks as if the target had admitted |
| 1635 | w.stats["rollback: dirty return parks nothing, the target never admitted"]++ |
| 1636 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) }) |
| 1637 | return |
| 1638 | } |
| 1639 | if w.parkStep(b, c.Holder.Location, fmt.Sprintf("superseded/%s/%s/return/", op.ID, host), op.StopAt+horizonH, w.opScope(c, c.Holder.Location), w.keepFilesLive(b.Region, confine(c.Holder.Location))) { |
| 1640 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) }) |
| 1641 | } |
| 1642 | case "install-copy": |
| 1643 | sv := w.buckets[b.Region].byID(op.ReturnDB.Key, op.ReturnDB.Ver) |
| 1644 | if sv == nil { |
| 1645 | return // the lane is gone: wait (stuck, in time) |
| 1646 | } |
| 1647 | g := installGen(&op) |
| 1648 | if sv.B.Max == 0 && on("valid-ltx") { |
| 1649 | w.cas(c, func(n *claim) bool { |
| 1650 | n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0 |
| 1651 | return sameOp(n.Op.ID, op.ID) |
| 1652 | }) |
| 1653 | return |
| 1654 | } |
| 1655 | key := snapKey(c.Holder.Location, g, sv.B.Max) |
| 1656 | v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: op.ReturnDB.Key, srcVer: op.ReturnDB.Ver}) |
| 1657 | if v > 0 || !on("build-completion") { |
| 1658 | next := "install-anchor" |
| 1659 | if !on("build-completion") { |
| 1660 | next = "published" |
| 1661 | } |
| 1662 | w.cas(c, func(n *claim) bool { |
| 1663 | n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max |
| 1664 | return sameOp(n.Op.ID, op.ID) |
| 1665 | }) |
| 1666 | } |
| 1667 | case "install-anchor": |
| 1668 | if w.anchorAcked(b, c.Holder.Location, installGen(&op), op.BaseKey, op.BaseVer, op.BaseTX) { |
| 1669 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| 1670 | } |
| 1671 | case "published": |
| 1672 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "local"; return sameOp(n.Op.ID, op.ID) }) // the marker is the probe's admission's (claim rev 6.9) |
| 1673 | case "local": |
| 1674 | w.installLocal(b, c, installGen(&op), op.ReturnFiles, "", func(n *claim) { n.Op.SrcPhase = "config" }) |
| 1675 | case "config": |
| 1676 | b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc |
| 1677 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 1678 | case "probing": |
| 1679 | if !w.admits(b, c, op.RollbackID) { |
| 1680 | return |
| 1681 | } |
| 1682 | want := c.Holder.Gen |
| 1683 | if op.Dirty { |
| 1684 | want = installGen(&op) // the return generation prepared from the target's sealed copy |
| 1685 | } |
| 1686 | if hasMain(b) || w.spawn(b, c, want, true) { |
| 1687 | w.probeAccepted(b, c) |
| 1688 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probed"; return sameOp(n.Op.ID, op.ID) }) |
| 1689 | return |
| 1690 | } |
| 1691 | if b.hostFailed && !b.failCounted { |
| 1692 | b.failCounted = true // a host the retry policy marked failed: one failed try for the operation |
| 1693 | } else if b.open != nil || b.hostFailed { |
| 1694 | return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try |
| 1695 | } |
| 1696 | w.logf("box %d: the rollback's probe did not start (try %d, generation %q, db=%v, host failed %v)", b.ID, op.ProbeTries+1, want, b.DB != nil, b.hostFailed) |
| 1697 | w.cas(c, func(n *claim) bool { |
| 1698 | n.Op.ProbeTries++ |
| 1699 | if n.Op.ProbeTries >= probeTries*3 { |
| 1700 | n.Op.SrcPhase = "stuck" |
| 1701 | } |
| 1702 | return sameOp(n.Op.ID, op.ID) |
| 1703 | }) |
| 1704 | case "probed": |
| 1705 | if b.open != nil { |
| 1706 | return // the source's session has not opened replication yet (rev 7.11 §7.14) |
| 1707 | } |
| 1708 | if !hasMain(b) { |
| 1709 | w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11) |
| 1710 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 1711 | return |
| 1712 | } |
| 1713 | if op.Dirty && !op.ReturnParked { |
| 1714 | // holder.generation and holder.stamp moved at the probe's cold |
| 1715 | // admission (claim rev 6.9); the return lane's park is owed |
| 1716 | w.cas(c, func(n *claim) bool { |
| 1717 | n.Op.ReturnParked = true |
| 1718 | n.Obl = append(n.Obl, w.scoped(w.newObl(n, "park-staging", op.SrcRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.RollbackID), "", op.ReturnQuiet), stagingBound(op.ReturnQuiet), c.Holder.Inc)) |
| 1719 | return sameOp(n.Op.ID, op.ID) |
| 1720 | }) |
| 1721 | return |
| 1722 | } |
| 1723 | if w.openRow(b, op.RollbackID) { |
| 1724 | w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "opened"; return sameOp(n.Op.ID, op.ID) }) |
| 1725 | } |
| 1726 | case "opened": |
| 1727 | if c.oblFor(op.ID) == 0 { |
| 1728 | w.closeMove(c, op.Src) |
| 1729 | } |
| 1730 | } |
| 1731 | } |
| 1732 | |
| 1733 | func (w *world) pushStep(b *box, c *claim) { |
| 1734 | op := c.Op |
| 1735 | loc := c.Holder.Location |
| 1736 | g := installGen(&op) // the generation this push prepares; its probe's cold admission makes it holder.generation (claim rev 6.9) |
| 1737 | switch op.Phase { |
| 1738 | case "stopping": |
| 1739 | w.holdRow(b, "moving", op.ID, c) |
| 1740 | if !w.evidence(b, op.ID) { |
| 1741 | return |
| 1742 | } |
| 1743 | // the push's snapshot: the replica under the final checkpoint, plus |
| 1744 | // the push's own mark, one more acknowledged transaction |
| 1745 | snap, _ := w.snapshotBlobOf(b, c, c.Holder.Gen) |
| 1746 | if snap == nil { |
| 1747 | // nothing to seal: the push is abandoned before it changed |
| 1748 | // anything (rev 7.11) |
| 1749 | w.stats["push abandoned: nothing to seal"]++ |
| 1750 | w.logf("push %s abandoned: the box has no database and its remote cannot be restored", op.ID) |
| 1751 | w.release(c, b.ID) |
| 1752 | return |
| 1753 | } |
| 1754 | snap.Gen = g |
| 1755 | w.nextW++ |
| 1756 | mark := w.nextW |
| 1757 | w.acks = append(w.acks, ack{W: mark, Inc: c.Holder.Inc, Box: b.ID, Session: c.Holder.Session, Gen: c.Holder.Gen, At: w.now, Kind: "write"}) |
| 1758 | w.writeInc[mark] = c.Holder.Inc |
| 1759 | snap.Max++ |
| 1760 | snap.Pages[fmt.Sprintf("k%d", mark)] = mark |
| 1761 | snap.TXs = append(snap.TXs, mark) |
| 1762 | w.txs[mark] = tx{W: mark, TX: snap.Max, Key: fmt.Sprintf("k%d", mark), Val: mark} |
| 1763 | w.cas(c, func(n *claim) bool { |
| 1764 | if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" { |
| 1765 | return false |
| 1766 | } |
| 1767 | n.Op.Phase, n.Op.StopAt, n.Op.PushSnap, n.Op.SealFromDB = "park", w.now, snap, b.DB != nil |
| 1768 | return true |
| 1769 | }) |
| 1770 | case "park": |
| 1771 | if w.parkStep(b, loc, fmt.Sprintf("superseded/%s/%s/push/", op.ID, host), op.StopAt+horizonH, w.opScope(c, loc), w.keepFilesLive(b.Region, confine(loc))) { |
| 1772 | w.cas(c, func(n *claim) bool { n.Op.Phase = "install"; return sameOp(n.Op.ID, op.ID) }) |
| 1773 | } |
| 1774 | case "install": |
| 1775 | if on("quiet-before-ack") && w.now < c.Quiet[b.Region] { |
| 1776 | return // installs wait out quiet_after (claim §6.2a) |
| 1777 | } |
| 1778 | snap := op.PushSnap |
| 1779 | if b.Files != nil { |
| 1780 | snap.SealKey = b.Files.Key |
| 1781 | } |
| 1782 | key := snapKey(loc, g, snap.Max) |
| 1783 | v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: snap}) |
| 1784 | if v > 0 || !on("build-completion") { |
| 1785 | next := "install-anchor" |
| 1786 | if !on("build-completion") { |
| 1787 | next = "published" |
| 1788 | } |
| 1789 | w.cas(c, func(n *claim) bool { |
| 1790 | n.Op.Phase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, snap.Max |
| 1791 | return sameOp(n.Op.ID, op.ID) |
| 1792 | }) |
| 1793 | } |
| 1794 | case "install-anchor": |
| 1795 | if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) { |
| 1796 | w.cas(c, func(n *claim) bool { n.Op.Phase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| 1797 | } |
| 1798 | case "published": |
| 1799 | w.cas(c, func(n *claim) bool { n.Op.Phase = "restore"; return sameOp(n.Op.ID, op.ID) }) // the marker is the probe's admission's (claim rev 6.9) |
| 1800 | case "restore": |
| 1801 | // the probe's admission restores the database (claim rev 6.9); the |
| 1802 | // files are unchanged and carried (F7) |
| 1803 | // every probe renders config and materializes the release for the |
| 1804 | // held row first (the passes skip held rows): RenderHeld, MaterializeHeld |
| 1805 | if !b.HasKey { |
| 1806 | return |
| 1807 | } |
| 1808 | b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc |
| 1809 | w.cas(c, func(n *claim) bool { n.Op.Phase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 1810 | case "probing": |
| 1811 | if !w.admits(b, c, op.ID) { |
| 1812 | return |
| 1813 | } |
| 1814 | if hasMain(b) || w.spawn(b, c, g, true) { |
| 1815 | w.probeAccepted(b, c) |
| 1816 | w.cas(c, func(n *claim) bool { |
| 1817 | n.Op.Phase = "probed" // holder.generation and holder.stamp moved at the probe's cold admission (claim rev 6.9) |
| 1818 | q := op.StopAt + horizonH |
| 1819 | if q > n.Quiet[b.Region] { |
| 1820 | n.Quiet[b.Region] = q |
| 1821 | } |
| 1822 | return sameOp(n.Op.ID, op.ID) |
| 1823 | }) |
| 1824 | } |
| 1825 | case "probed": |
| 1826 | if b.open != nil { |
| 1827 | return // the admitted session's start is not complete: replication is not open yet (rev 7.11 §7.14) |
| 1828 | } |
| 1829 | if !hasMain(b) { |
| 1830 | // the probe died before its session opened replication or in |
| 1831 | // its settle (seed 487 / 30,000: a push released over a dead |
| 1832 | // probe whose marker never landed): the probe restarts, a cold |
| 1833 | // admission like any other |
| 1834 | w.stats["probe restarted"]++ |
| 1835 | w.cas(c, func(n *claim) bool { n.Op.Phase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| 1836 | return |
| 1837 | } |
| 1838 | if w.openRow(b, op.ID) || b.Row != nil && b.Row.Status == "ready" { |
| 1839 | w.release(c, b.ID) |
| 1840 | } |
| 1841 | } |
| 1842 | } |
| 1843 | |
| 1844 | // sameStep: the operation and every phase the worker branches on are what |
| 1845 | // the stale read said (claim C3's revalidation under the lease). |
| 1846 | func sameStep(a, b *claim) bool { |
| 1847 | if a == nil || b == nil { |
| 1848 | return a == b |
| 1849 | } |
| 1850 | x, y := a.Op, b.Op |
| 1851 | if !(x.ID == y.ID && x.Kind == y.Kind && x.Phase == y.Phase && x.LandPhase == y.LandPhase && |
| 1852 | x.SrcPhase == y.SrcPhase && x.RevPhase == y.RevPhase && x.Exec == y.Exec && x.Committed == y.Committed && |
| 1853 | x.Opened == y.Opened && len(a.Obl) == len(b.Obl) && (a.Holder == nil) == (b.Holder == nil)) { |
| 1854 | return false |
| 1855 | } |
| 1856 | // the writer state is part of the step (claim rev 6.8 §6.4a, R8-2): a |
| 1857 | // stalled worker that read a cold admission in progress must not replay |
| 1858 | // its build once the admission completed and the writer runs, since the |
| 1859 | // build takes stop evidence and replaces the database (seed 10382 / |
| 1860 | // 30,000) |
| 1861 | if a.Holder != nil && (a.Holder.Session != b.Holder.Session || a.Holder.Gen != b.Holder.Gen || a.Holder.Stamp != b.Holder.Stamp) { |
| 1862 | return false |
| 1863 | } |
| 1864 | if a.Holder != nil && a.Holder.WriterBox != b.Holder.WriterBox { |
| 1865 | return false |
| 1866 | } |
| 1867 | return a.Holder == nil || len(a.Holder.Admissions) == len(b.Holder.Admissions) |
| 1868 | } |
| 1869 | |
| 1870 | // sweep is claim rev 5's at-rest sweep (§6.2a, CC15): for a host with no |
| 1871 | // operation running, each region's box lists the locations of the host's |
| 1872 | // retired incarnations and every generation within holder.location other |
| 1873 | // than holder.generation, and parks whatever it finds as a further |
| 1874 | // retirement event, under a park-sweep obligation it owns. The claim's |
| 1875 | // wording leaves out `moves/<host>/`; `sweep-staging` (CC20) adds it. |
| 1876 | func (w *world) sweep(b *box, c *claim) { |
| 1877 | holderless := c != nil && c.Holder == nil && c.Op.Kind == "parking" && c.Op.ParkSweep != "" |
| 1878 | if !on("straggler-sweep") || c == nil || !(c.Op.Kind == "" || c.Op.Kind == "serving" || holderless) { |
| 1879 | return // a running holderless sweep admits another region's chores (R14-4); any other operation suppresses sweeps |
| 1880 | } |
| 1881 | for _, o := range c.Obl { |
| 1882 | if o.Kind == "park-sweep" && o.Region == b.Region { |
| 1883 | return // its own sweep is still running |
| 1884 | } |
| 1885 | } |
| 1886 | bk := w.buckets[b.Region] |
| 1887 | // one chore per scope location holding residue (claim rev 6.24 §2, |
| 1888 | // R16-2): the sweepable keys are grouped by the location they belong |
| 1889 | // to, an incarnation's own, the legacy instances/<host> (a level |
| 1890 | // directory or generation directly under the host prefix, never a |
| 1891 | // child incarnation) or moves/<host>; each chore's scope names that |
| 1892 | // location and its archive incarnation. The mutation of |
| 1893 | // scope-membership queues rev 7.16's two chores, instances/<host> and |
| 1894 | // moves/<host>, over the whole prefixes |
| 1895 | var prefixes []string |
| 1896 | for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} { |
| 1897 | for _, k := range bk.keys(p) { |
| 1898 | if !hasNonMarker(bk.objs[k]) || !w.sweepable(c, b.Region, k) { |
| 1899 | continue |
| 1900 | } |
| 1901 | loc := p |
| 1902 | if on("scope-membership") { |
| 1903 | loc = scopeOf(k) + "/" |
| 1904 | } |
| 1905 | if !hasStr(prefixes, loc) { |
| 1906 | prefixes = append(prefixes, loc) |
| 1907 | } |
| 1908 | } |
| 1909 | } |
| 1910 | if len(prefixes) == 0 { |
| 1911 | return |
| 1912 | } |
| 1913 | w.cas(c, func(n *claim) bool { |
| 1914 | joining := n.Holder == nil && n.Op.Kind == "parking" && n.Op.ParkSweep != "" |
| 1915 | if !(n.Op.Kind == "" || n.Op.Kind == "serving" || joining) { |
| 1916 | return false // any other operation running suppresses new sweeps (claim rev 6.15 §6.2a) |
| 1917 | } |
| 1918 | // the chore's owner: on a serving host the session, as for |
| 1919 | // PublishGeneration; on a free host a holderless parking op, {kind: |
| 1920 | // parking, intent: {park_sweep: "v1"}}, no exec and no holder, the |
| 1921 | // only holderless claim that may carry obligations, so it blocks |
| 1922 | // free's successor create until its last acknowledgement frees the |
| 1923 | // claim again (FreeAfterPark; claim rev 6.15 §6.2a, R13-4). The |
| 1924 | // mutation of holderless-park queues rev 7.13's bare "sweep" chores |
| 1925 | // on the free claim. Another region's sweep while the op runs joins |
| 1926 | // it with its own chores (a reading, §19.4) |
| 1927 | op := "sweep" |
| 1928 | switch { |
| 1929 | case n.Holder != nil: |
| 1930 | op = fmt.Sprintf("session:%d", n.Holder.Session) |
| 1931 | case joining: |
| 1932 | op = n.Op.ID |
| 1933 | w.stats["sweep joined a running holderless op"]++ |
| 1934 | case on("holderless-park"): |
| 1935 | n.Op = opState{Kind: "parking", ID: w.newID("park"), ParkSweep: "v1", Exec: -1, ReqAt: w.now} |
| 1936 | op = n.Op.ID |
| 1937 | w.stats["holderless sweeps"]++ |
| 1938 | } |
| 1939 | for _, p := range prefixes { |
| 1940 | // the scope is persisted at the queuing CAS (claim rev 6.23 §2): |
| 1941 | // the sweeping box's complete listing is the evidence, taken now; |
| 1942 | // the location and its archive incarnation (the location's own, |
| 1943 | // the holder's for the legacy location or the staging sweep, the |
| 1944 | // newest retired on a free host; claim rev 6.24 §2) |
| 1945 | n.Obl = append(n.Obl, w.scoped(obligation{ID: w.newID("ob"), Op: op, Kind: "park-sweep", Region: b.Region, Box: -1, Prefix: p, Ret: "sweep"}, w.now, incOf(n, scopeLocation(p)))) |
| 1946 | } |
| 1947 | // CoverParkedPredecessors at the queuing CAS (claim rev 6.21 item 6): |
| 1948 | // the predecessors with a sweepable object, no others |
| 1949 | w.coverParkedPredecessors(n, b.Region, w.sweptPredecessors(n, b.Region, false)) |
| 1950 | return true |
| 1951 | }) |
| 1952 | } |
| 1953 | |
| 1954 | // sweepable: is k, in region r, one of the things the at-rest sweep parks? |
| 1955 | func (w *world) sweepable(c *claim, r int, k string) bool { |
| 1956 | if strings.HasSuffix(k, "lock.json") { |
| 1957 | return false |
| 1958 | } |
| 1959 | if strings.HasPrefix(k, "moves/") { |
| 1960 | return on("sweep-staging") // CC20 |
| 1961 | } |
| 1962 | h := c.Holder |
| 1963 | if !on("sweep-namespace") { |
| 1964 | // claim rev 5's discovery: retired_incarnations (capped) and the |
| 1965 | // legacy flag. A location older than the cap is never listed |
| 1966 | if h != nil && strings.HasPrefix(k, h.Location) { |
| 1967 | return w.deadGeneration(c, r, k) |
| 1968 | } |
| 1969 | for _, inc := range c.Retired { |
| 1970 | if strings.HasPrefix(k, locationFor(inc, false)) { |
| 1971 | return true |
| 1972 | } |
| 1973 | } |
| 1974 | return w.legacyRetired(c) |
| 1975 | } |
| 1976 | // claim rev 6 §6.2a (R3-4): locations come from the namespace itself. |
| 1977 | // Each child prefix of instances/<host>/ is an incarnation location, or |
| 1978 | // the legacy location (the host prefix, litestream's level layout): |
| 1979 | // residue unless it is holder.location, and within holder.location |
| 1980 | // residue unless it is holder.generation |
| 1981 | if h != nil && locationOf(k) == h.Location { |
| 1982 | return w.deadGeneration(c, r, k) |
| 1983 | } |
| 1984 | return true |
| 1985 | } |
| 1986 | |
| 1987 | // locationOf is the location a key under instances/<host>/ belongs to: |
| 1988 | // instances/<host>/<incarnation>/ for a child prefix, else the legacy |
| 1989 | // location instances/<host>/ (a level directory directly under the host). |
| 1990 | func locationOf(k string) string { |
| 1991 | p := "instances/" + host + "/" |
| 1992 | rest := strings.TrimPrefix(k, p) |
| 1993 | if i := strings.Index(rest, "/"); i > 0 { |
| 1994 | child := rest[:i] |
| 1995 | // an incarnation id as a child prefix (the claim's 64 hex |
| 1996 | // characters; the model's decimal ids) is an incarnation location. |
| 1997 | // A level directory (0000 to 0009, litestream's layout) or a |
| 1998 | // generation prefix (g-<op>/) directly under the host belongs to |
| 1999 | // the legacy location, the host prefix itself (claim rev 6.5 §6.2a) |
| 2000 | digits := len(child) > 0 && !isLevelDir(child) |
| 2001 | for _, r := range child { |
| 2002 | if r < '0' || r > '9' { |
| 2003 | digits = false |
| 2004 | } |
| 2005 | } |
| 2006 | if digits { |
| 2007 | return p + child + "/" |
| 2008 | } |
| 2009 | } |
| 2010 | return p |
| 2011 | } |
| 2012 | |
| 2013 | // deadGeneration: within holder.location, everything but the live set. |
| 2014 | // The live set is {authoritative region, holder.location, |
| 2015 | // holder.generation} (claim rev 6.5 §6.2a, R4-3): the generation's |
| 2016 | // objects are live only in the authoritative region's bucket, the region |
| 2017 | // of the holder's current installation; a matching key in any other |
| 2018 | // region is residue. The mutation of live-set-region exempts matching keys |
| 2019 | // in every region. |
| 2020 | func (w *world) deadGeneration(c *claim, r int, k string) bool { |
| 2021 | h := c.Holder |
| 2022 | if w.filesLive(c, r, k) { |
| 2023 | return false |
| 2024 | } |
| 2025 | e := w.execBox(c) |
| 2026 | authoritative := e != nil && e.Region == r |
| 2027 | if !on("live-set-region") { |
| 2028 | authoritative = true |
| 2029 | } |
| 2030 | // an absent marker is the origin generation "" (R11-1): an adopted |
| 2031 | // host's replica has no generation.json until its first cold admission |
| 2032 | // publishes one, and that admission's unfinished chain protects the |
| 2033 | // origin exactly as the residue oracle reads it |
| 2034 | mg := "" |
| 2035 | v := w.buckets[r].current(genMarkerKey(h.Location)) |
| 2036 | if v != nil { |
| 2037 | mg = v.B.Gen |
| 2038 | } |
| 2039 | if v == nil && !on("absent-marker-origin") { |
| 2040 | // the mutation (rev 7.11's sweep): an absent marker is no |
| 2041 | // disagreement, so nothing protects the unfinished chain's origin |
| 2042 | mg = h.Gen |
| 2043 | } |
| 2044 | if authoritative && mg != h.Gen { |
| 2045 | // the marker disagrees with holder.generation. Normal while the |
| 2046 | // latest admission is unfinished: the marker names its predecessor, |
| 2047 | // and both generations are live (claim rev 6.10, R10-2). Any other |
| 2048 | // disagreement: touch nothing under the location, and alert (§6.4) |
| 2049 | chain, ok := w.unfinishedChain(c, mg) |
| 2050 | if ok && on("baseline-live") { |
| 2051 | for _, g := range append(chain, mg) { |
| 2052 | if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, g, "") { |
| 2053 | return false |
| 2054 | } |
| 2055 | if isClosedKey(k) && strings.HasPrefix(k, dataKey(h.Location, g, "closed/")) { |
| 2056 | return false |
| 2057 | } |
| 2058 | } |
| 2059 | if k == genMarkerKey(h.Location) { |
| 2060 | return false |
| 2061 | } |
| 2062 | } else if !w.markerAlerted[h.Location] { |
| 2063 | w.markerAlerted[h.Location] = true |
| 2064 | w.alert("box in region %d: generation.json names %q, the claim %q, and no unfinished admission explains it; the location is left untouched", r, mg, h.Gen) |
| 2065 | return false |
| 2066 | } else { |
| 2067 | return false |
| 2068 | } |
| 2069 | } |
| 2070 | if k == genMarkerKey(h.Location) { |
| 2071 | v := w.buckets[r].current(k) |
| 2072 | return v != nil && (v.B.Gen != h.Gen || !authoritative) |
| 2073 | } |
| 2074 | if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, h.Gen, "") { |
| 2075 | return !authoritative |
| 2076 | } |
| 2077 | if isClosedKey(k) { |
| 2078 | if strings.HasPrefix(k, dataKey(h.Location, h.Gen, "closed/")) { |
| 2079 | return !authoritative // the live generation's certificates, every session's |
| 2080 | } |
| 2081 | return true // a dead generation's certificates |
| 2082 | } |
| 2083 | if isFilesKey(k) { |
| 2084 | // a files manifest or object outside the files live set (region HA §5.6, L13): |
| 2085 | // its generation's, as the data keys are |
| 2086 | if genOf(h.Location, k) == h.Gen { |
| 2087 | return !authoritative |
| 2088 | } |
| 2089 | return true |
| 2090 | } |
| 2091 | return isDataKey(k) |
| 2092 | } |
| 2093 | |
| 2094 | // genOf is the generation within loc that key k belongs to: "" for the |
| 2095 | // origin layout (level directories and closed.json directly under loc), |
| 2096 | // the op id for g-<op>/..., and "-" for a key of no generation (markers, |
| 2097 | // the lock). |
| 2098 | func genOf(loc, k string) string { |
| 2099 | rest := strings.TrimPrefix(k, loc) |
| 2100 | if strings.HasPrefix(rest, "files/") { |
| 2101 | return "" // a files manifest of the origin generation (region HA §5.6) |
| 2102 | } |
| 2103 | if strings.HasPrefix(rest, "g-") { |
| 2104 | if i := strings.Index(rest, "/"); i > 0 { |
| 2105 | return rest[2:i] |
| 2106 | } |
| 2107 | return "-" |
| 2108 | } |
| 2109 | if isDataKey(k) && genPrefixOf(k) == loc || strings.HasPrefix(rest, "closed/") { |
| 2110 | return "" |
| 2111 | } |
| 2112 | return "-" |
| 2113 | } |
| 2114 | |
| 2115 | // legacyRetired: the host once held a legacy incarnation, now retired, so |
| 2116 | // the bare namespace outside holder.location is a retired location. |
| 2117 | func (w *world) legacyRetired(c *claim) bool { |
| 2118 | for _, inc := range c.Retired { |
| 2119 | if w.legacyIncs[inc] { |
| 2120 | return true |
| 2121 | } |
| 2122 | } |
| 2123 | return false |
| 2124 | } |
| 2125 | |
| 2126 | // publishGen writes generation.json, conditional on the marker it |
| 2127 | // supersedes: If-Match its version, or If-None-Match when there is none |
| 2128 | // (moves rev 7.1 §5.1, model 29). generation.json is a reused key, so an |
| 2129 | // earlier publication that completes late would otherwise put an older |
| 2130 | // generation back, and every wake would refuse. |
| 2131 | func (w *world) publishGen(b *box, loc, gen string, inc int, base int, key string, tx int) int { |
| 2132 | // the marker names the generation's snapshot by key, version id and TXID (R3-1, R4-1) |
| 2133 | // INV-11 (R7-4): generation.json is published only from a completed |
| 2134 | // build: the snapshot version it names exists and the anchor at its |
| 2135 | // TXID is current. A marker that names artifacts still in flight would |
| 2136 | // make the completion predicate true before the artifacts exist |
| 2137 | bk := w.buckets[b.Region] |
| 2138 | if tx > 0 { |
| 2139 | if sv := bk.byID(key, base); sv == nil || sv.Marker { |
| 2140 | w.violate("INV-11: generation.json for %s published before its snapshot %s@%d landed (R7-4)", gen, key, base) |
| 2141 | } else if on("restore-anchor") && bk.current(ltxKey(loc, gen, 0, tx, tx)) == nil { |
| 2142 | w.violate("INV-11: generation.json for %s published before its anchor [%d,%d] landed (R7-4)", gen, tx, tx) |
| 2143 | } |
| 2144 | } |
| 2145 | r := &request{kind: rPut, region: b.Region, key: genMarkerKey(loc), b: &blob{Kind: "gen", Gen: gen, Inc: inc, Base: base, SnapKey: key, TX: tx}} |
| 2146 | if w.fail["markerhold"] && w.rng.Intn(2) == 0 { |
| 2147 | // the biased schedule (rev 7.5): this publication is accepted and |
| 2148 | // completes at its worst moment, once a newer marker is current |
| 2149 | r.hold = true |
| 2150 | } |
| 2151 | if on("generation-conditional") { |
| 2152 | r.ifSet = true |
| 2153 | if cur := w.buckets[b.Region].current(r.key); cur != nil { |
| 2154 | r.ifVer = cur.ID |
| 2155 | } |
| 2156 | } |
| 2157 | return w.issue(b.ID, r) |
| 2158 | } |
| 2159 | |
| 2160 | // anchorAcked is §5.1 step 7 inside a build (R6-1, R7-4): the anchor |
| 2161 | // [tx, tx], every page of the snapshot the build installed, is PUT into the |
| 2162 | // generation and the build goes on only once the PUT is acknowledged. The |
| 2163 | // mutation of restore-anchor writes none; the mutation of anchor-acked |
| 2164 | // does not wait. |
| 2165 | func (w *world) anchorAcked(b *box, loc, gen, snapKey string, snapVer, tx int) bool { |
| 2166 | if !on("restore-anchor") || tx == 0 && on("valid-ltx") { |
| 2167 | return true // no transactions yet: there is no position to anchor (the mutation of valid-ltx writes [0,0], which the encoder refuses) |
| 2168 | } |
| 2169 | sv := w.buckets[b.Region].byID(snapKey, snapVer) |
| 2170 | if sv == nil { |
| 2171 | return false |
| 2172 | } |
| 2173 | if w.buckets[b.Region].current(ltxKey(loc, gen, 0, tx, tx)) != nil { |
| 2174 | return true // a replay finds its anchor current and reuses it |
| 2175 | } |
| 2176 | db := &localDB{Inc: sv.B.Inc, Loc: loc, Gen: gen, Key: sv.B.SealKey, Pos: tx, State: cloneState(sv.B.Pages), Log: append([]int(nil), sv.B.TXs...)} |
| 2177 | acked := w.ackedPut(b, ltxKey(loc, gen, 0, tx, tx), anchorOf(db, sv.B.Inc), false) |
| 2178 | w.stats["restore anchors"]++ |
| 2179 | return acked || !on("anchor-acked") |
| 2180 | } |
| 2181 | |
amadan