amadan
sim: hostlife: claim rev 6.10, the admitting CAS before the app start, a session per agent process, unfinished generations with live baselines, the library's admission shape, startup PUTs in the ledger, declarations as the only exemptions, the uncovered return (rev 7.11)
Joint round 10 (R10-1 to R10-9) and claim rev 6.10 (3f916b05). The AdmitWriter CAS comes first; the start is the agent's own loop (restore, app with a retry policy of five attempts and a doubling backoff, first upload, marker, anchor, open); a session belongs to the agent process (localDB.Epoch), so a restarted agent neither replicates nor certifies for it and stops an app it finds running. A cold admission's generation is unfinished until generation.json names it: no chore at the CAS, the marker's publication queues them, a lagging marker is read through the writers journal as a chain of unfinished admissions, any other disagreement alerts; the journal survives the release and the start of a move or a push, and an uncovered predecessor is owed. admitInput is shaped like E's Admission (KeepStamp, BaselineCoversPredecessor, warm with an empty NewGeneration, refused in creating, deleting, parking, purging, placing); serving chores use session:<n>, the root is @root. The faults exempt nothing; INV-14 checks alert delivery; INV-15 refuses declaring a write the remote holds. First uploads and anchors enter the ledger with sticky uncertainty and captured bytes. The budget prices a restore per probe attempt, the admission and a fresh upload per retry. The baseline logs COUNTERS as JSON. Found by the rerun: a rollback is clean only when the seal recorded the source's replica as covering its stopped database (seal-return), a seal built from the remote after a lost disk does not cover the predecessor, a pre-flip round never forces a sync on a session whose start is unfinished or whose agent is gone, dead probes restart, a declared-lost box's timers are declared, a retired stamp's leftover may be parked. Rules admit-before-start, baseline-live, seal-return (95); witnesses wAckBeforeAdmission, wBaselineParked, wFirstUploadRetry, wAnchorRetry, wSealReturn (57). start-check and start-check-claim-gen are caught alone again and leave expected-outcomes.txt.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
11 files changed,
+1255
−318
internal/sim/hostlife/background_test.go+11 −12internal/sim/hostlife/budget_test.go+63 −13internal/sim/hostlife/console_test.go+29 −6internal/sim/hostlife/expected-outcomes.txt+0 −2internal/sim/hostlife/ltx_test.go+23 −2internal/sim/hostlife/moves_test.go+194 −21internal/sim/hostlife/protocol_test.go+593 −238internal/sim/hostlife/rules_test.go+3 −0internal/sim/hostlife/sim_test.go+54 −6internal/sim/hostlife/witness_test.go+222 −13internal/sim/hostlife/world_test.go+63 −5
diff --git a/internal/sim/hostlife/background_test.go b/internal/sim/hostlife/background_test.go| index a12609c38..ae03d1aa1 100644 |
| --- a/internal/sim/hostlife/background_test.go |
| +++ b/internal/sim/hostlife/background_test.go |
| @@ -48,6 +48,13 @@ func (w *world) idleStop(b *box) { |
| } |
| } |
| b.Procs = keep |
| + if b.open != nil { |
| + // the session's start is still finishing its PUTs: the agent's |
| + // replicator opens when they land and ships what the app wrote, |
| + // then closes (rev 7.11 §7.14); no certificate now |
| + w.stats["idle stop before replication opened"]++ |
| + return |
| + } |
| // the clean close (claim rev 6.9 §6.4a): drain, then the certificate |
| if b.DB != nil { |
| b.DB.Closing = true |
| @@ -354,12 +361,9 @@ func (w *world) fault() { |
| // in every operation that has this box write (R8-2), not only |
| // while serving |
| if c := w.claim; w.fail["disk"] && c != nil && c.Holder != nil && c.Op.Exec == b.ID && b.DB != nil && w.rng.Intn(3) == 0 { |
| - for i := range w.acks { |
| - a := &w.acks[i] |
| - if a.Box == b.ID && a.Inc == b.DB.Inc && !w.isDurable(a.W) { |
| - a.Exempt = true |
| - } |
| - } |
| + // the fault removes physical state only (R10-6): the |
| + // acknowledged, unuploaded writes are declared lost by the |
| + // next admission's writes_lost record, never exempted here |
| b.DB = nil |
| b.Procs = nil // nothing survives the disk: a descendant with no database would acknowledge writes into nowhere |
| w.stats["disk-lost with the crash"]++ |
| @@ -442,12 +446,7 @@ func (w *world) fault() { |
| // durable write must come back (§14.2.3). |
| c := w.claim |
| if w.fail["disk"] && c != nil && c.Holder != nil && c.Op.Kind == "serving" && c.Op.Exec == b.ID && b.DB != nil && !hasMain(b) && len(b.Procs) == 0 { |
| - for i := range w.acks { |
| - a := &w.acks[i] |
| - if a.Box == b.ID && a.Inc == b.DB.Inc && !w.isDurable(a.W) { |
| - a.Exempt = true |
| - } |
| - } |
| + // physical loss only (R10-6): the next admission declares |
| b.DB = nil |
| w.stats["disk-lost"]++ |
| w.logf("box %d lost its database file", b.ID) |
diff --git a/internal/sim/hostlife/budget_test.go b/internal/sim/hostlife/budget_test.go| index 3987ae088..36407a965 100644 |
| --- a/internal/sim/hostlife/budget_test.go |
| +++ b/internal/sim/hostlife/budget_test.go |
| @@ -18,9 +18,18 @@ import "testing" |
| // the verification of the database the build produced (a disk read: its |
| // sha256 against the install manifest), which is what the `restored` |
| // phase does instead of a second cold restore. |
| +// Rev 7.11 prices what the new protocol does at every probe attempt (R10-8): |
| +// the admitted session purges the built database and restores the |
| +// prepared generation again before the app starts (RestoreAttempt, one S3 |
| +// read and one disk write of the whole database), plus the admission's |
| +// own overhead (the CAS, the purge, generation.json: Admit); a retried |
| +// probe is a fresh generation whose first upload is the whole database |
| +// again (FreshUpload, one disk read and one S3 write), so every attempt |
| +// after the first costs that too. |
| type cutEstimate struct { |
| Stop, Checkpoint, Mirror, Archive, Render, Open float64 |
| - RestoreStaged, SnapshotBuild, Anchor, Verify float64 // target-side, whole database |
| + RestoreStaged, SnapshotBuild, Anchor, Verify float64 // target-side build, whole database |
| + RestoreAttempt, Admit, FreshUpload float64 // per probe attempt (rev 7.11) |
| Probe, Settle float64 // one attempt |
| } |
| @@ -29,6 +38,7 @@ const ( |
| probeAttemptsMax = 3 // §7.7: never more than three attempts |
| probeSeconds = 30.0 |
| settleSeconds = 20.0 |
| + admitSeconds = 2.0 // the admitting CAS, the purge and generation.json (rev 7.11) |
| mirrorBarMiB = 16.0 // §7.3: the final round is at most this much |
| ) |
| @@ -36,17 +46,28 @@ func (e cutEstimate) fixed() float64 { |
| return e.Stop + e.Checkpoint + e.Mirror + e.Archive + e.RestoreStaged + e.SnapshotBuild + e.Anchor + e.Verify + e.Render + e.Open |
| } |
| -// attemptsAdmitted is T11 as rev 7.4 states it: the attempts permitted are |
| -// what the window leaves after the fixed costs, at most three; the |
| -// instance is eligible when that is at least one, and the probe loop |
| -// (§7.7) runs at most that many. |
| -func (e cutEstimate) attemptsAdmitted(window float64) int { |
| - n := int((window - e.fixed()) / (e.Probe + e.Settle)) |
| - if n > probeAttemptsMax { |
| - n = probeAttemptsMax |
| +// attempt prices the i-th probe attempt (from 1): the restore of the |
| +// prepared generation and the admission overhead, the probe and its |
| +// settle; from the second attempt on, the fresh generation's first upload |
| +// as well (rev 7.11, R10-8). |
| +func (e cutEstimate) attempt(i int) float64 { |
| + c := e.RestoreAttempt + e.Admit + e.Probe + e.Settle |
| + if i > 1 { |
| + c += e.FreshUpload |
| } |
| - if n < 0 { |
| - n = 0 |
| + return c |
| +} |
| + |
| +// attemptsAdmitted is T11 as rev 7.4 states it, repriced in rev 7.11: the |
| +// attempts permitted are those whose summed costs fit the window after |
| +// the fixed costs, at most three; the instance is eligible when that is |
| +// at least one, and the probe loop (§7.7) runs at most that many. |
| +func (e cutEstimate) attemptsAdmitted(window float64) int { |
| + left := window - e.fixed() |
| + n := 0 |
| + for n < probeAttemptsMax && left >= e.attempt(n+1) { |
| + left -= e.attempt(n + 1) |
| + n++ |
| } |
| return n |
| } |
| @@ -73,7 +94,14 @@ func estimate(dbMiB float64, r rates) cutEstimate { |
| SnapshotBuild: dbMiB/r.disk + dbMiB/r.s3, |
| Anchor: dbMiB/r.disk + dbMiB/r.s3, |
| Verify: dbMiB / r.disk, |
| - Probe: probeSeconds, Settle: settleSeconds, |
| + // rev 7.11: the probe's admission restores the prepared generation |
| + // again (purge, then the pinned snapshot from regional S3 to disk), |
| + // with the admission's overhead; a retried probe uploads the whole |
| + // database once more as its fresh generation's root |
| + RestoreAttempt: dbMiB/r.s3 + dbMiB/r.disk, |
| + Admit: admitSeconds, |
| + FreshUpload: dbMiB/r.disk + dbMiB/r.s3, |
| + Probe: probeSeconds, Settle: settleSeconds, |
| } |
| } |
| @@ -112,11 +140,32 @@ func TestReadinessBudget(t *testing.T) { |
| t.Fatalf("the example no longer illustrates R5-5") |
| } |
| if n := big.attemptsAdmitted(cutWindowDefault); n != 1 { |
| - t.Fatalf("a 2 GiB database at the example rates admits %d attempts, want 1 (fixed %.1f s)", n, big.fixed()) |
| + t.Fatalf("a 2 GiB database at the example rates admits %d attempts, want 1 (fixed %.1f s, first attempt %.1f s)", n, big.fixed(), big.attempt(1)) |
| } |
| if big.fixed() < 53 || big.fixed() > 54 { |
| t.Fatalf("the example's fixed costs are %.1f s; the spec says 53.6 s", big.fixed()) |
| } |
| + // rev 7.11 (R10-8): the first attempt restores the prepared generation |
| + // again and pays the admission: 12.3 + 2 + 30 + 20 = 64.3 s, so the |
| + // example runs to 117.9 s and admits one attempt by 2.1 s; a second |
| + // attempt would add 76.6 s (a fresh generation's full upload too) |
| + if a := big.attempt(1); a < 64 || a > 65 { |
| + t.Fatalf("the example's first attempt costs %.1f s; the spec says 64.3 s", a) |
| + } |
| + if tot := big.fixed() + big.attempt(1); tot < 117 || tot > 118.5 { |
| + t.Fatalf("the example runs to %.1f s with one attempt; the spec says 117.9 s", tot) |
| + } |
| + // the review's 2.5 GiB (R10-8): rev 7.10's arithmetic admitted it at |
| + // 113.8 s; with the probe's own restore it is 129.2 s, past the window, |
| + // and refused |
| + bigger := estimate(2560, exampleRates) |
| + old := bigger.fixed() + probeSeconds + settleSeconds |
| + if old < 113 || old > 114.5 { |
| + t.Fatalf("rev 7.10's arithmetic for 2.5 GiB is %.1f s; the review says 113.8 s", old) |
| + } |
| + if n := bigger.attemptsAdmitted(cutWindowDefault); n != 0 { |
| + t.Fatalf("2.5 GiB at the example rates admits %d attempts, want 0 (%.1f s with one attempt)", n, bigger.fixed()+bigger.attempt(1)) |
| + } |
| // and a cross-region snapshot transfer, if a builder did that instead, |
| // is refused by the same arithmetic |
| naive := big |
| @@ -132,6 +181,7 @@ func TestReadinessBudget(t *testing.T) { |
| r := rates{cross: 10, s3: 200, disk: 200} |
| three := estimate(3072, r) |
| three.RestoreStaged, three.SnapshotBuild, three.Anchor, three.Verify = 3072/200.0, 3072/200.0, 3072/200.0, 0 // rev 7.6's arithmetic |
| + three.RestoreAttempt, three.Admit, three.FreshUpload = 0, 0, 0 // rev 7.6's arithmetic knew no per-attempt restore |
| if n := three.attemptsAdmitted(cutWindowDefault); n != 1 { |
| t.Fatalf("rev 7.6's three passes admit %d attempts for the review's workload, want 1 (the finding's premise; fixed %.2f s)", n, three.fixed()) |
| } |
diff --git a/internal/sim/hostlife/console_test.go b/internal/sim/hostlife/console_test.go| index 61ab14485..0b6771276 100644 |
| --- a/internal/sim/hostlife/console_test.go |
| +++ b/internal/sim/hostlife/console_test.go |
| @@ -243,7 +243,7 @@ func (w *world) createSteps(p *consolePlan) bool { |
| } |
| w.serial++ |
| if !w.cas(c, func(n *claim) bool { |
| - n.Holder = &holder{Inc: p.Inc, App: p.App, Serial: w.serial, Location: locationFor(p.Inc, false)} |
| + n.Holder = &holder{Inc: p.Inc, App: p.App, Serial: w.serial, Location: locationFor(p.Inc, false), WriterBox: -1} // no writer yet |
| n.Op = opState{Kind: "creating", ID: p.OpID, Exec: -1, NewInc: p.Inc, NewApp: p.App, NewRegion: p.Region, ReqAt: w.now} |
| n.Regions, n.Boxes, n.Complete, n.Obl = []int{p.Region}, []int{b.ID}, true, nil |
| return true |
| @@ -404,7 +404,8 @@ func (w *world) moveTo(r int) { |
| } |
| n.Op = opState{Kind: "moving", ID: id, Phase: "leases", Exec: src.ID, Parts: []int{src.ID, tgt.ID}, |
| Src: src.ID, Tgt: tgt.ID, SrcRegion: src.Region, TgtRegion: r, ReqAt: w.now, Auto: true, App: c.Holder.App, |
| - SrcGen: c.Holder.Gen} // the source's generation, which a clean rollback's return names again (rev 7.10) |
| + SrcGen: c.Holder.Gen, // the source's generation, which a clean rollback's return names again (rev 7.10) |
| + Writers: n.Op.Writers} // the writers journal is carried into every operation (rev 7.11 §5.1 step 9) |
| return true |
| }) { |
| w.logf("move %s inc %d from box %d to box %d", id, c.Holder.Inc, src.ID, tgt.ID) |
| @@ -432,7 +433,7 @@ func (w *world) userPush() { |
| if n.Op.Kind != "serving" { |
| return false |
| } |
| - n.Op = opState{Kind: "pushing", ID: id, Phase: "stopping", Exec: b.ID, Parts: []int{b.ID}, Src: b.ID, Tgt: b.ID, SrcRegion: b.Region, TgtRegion: b.Region, ReqAt: w.now} |
| + n.Op = opState{Kind: "pushing", ID: id, Phase: "stopping", Exec: b.ID, Parts: []int{b.ID}, Src: b.ID, Tgt: b.ID, SrcRegion: b.Region, TgtRegion: b.Region, ReqAt: w.now, Writers: n.Op.Writers} |
| return true |
| }) { |
| w.logf("push %s on box %d", id, b.ID) |
| @@ -1124,6 +1125,16 @@ func (w *world) operator() { |
| return |
| } |
| } |
| + // a host marked failed by the retry policy (rev 7.11 §7.14): the operator |
| + // looks at it and clears the mark; the next attempt is a new session |
| + for _, b := range w.liveBoxes() { |
| + if b.hostFailed && w.now >= b.failedAt+horizonH/2 { |
| + b.hostFailed, b.failCounted, b.startFails, b.nextStart = false, false, 0, 0 |
| + w.stats["failed hosts cleared"]++ |
| + w.logf("operator cleared the failed host on box %d", b.ID) |
| + return |
| + } |
| + } |
| // a destroyed box that owes only local chores: declare it lost (claim |
| // §7.2 removes its obligations); compute fencing is the box being gone |
| for _, o := range c.Obl { |
| @@ -1209,11 +1220,23 @@ func (w *world) operator() { |
| } |
| return true |
| }) { |
| + // the lost target's acknowledged writes are declared by the source's |
| + // admission (writes_lost for the target's uncertified session, or a |
| + // sealed baseline), never exempted here (R10-6). Its registered |
| + // timers went with it and no return lane will carry them: the |
| + // declaration names them, and alerts (seed 4120 / 30,000) |
| + w.logf("operator declared box %d lost for %s", t.ID, op.ID) |
| + n := 0 |
| for i := range w.acks { |
| - if w.acks[i].Box == t.ID && w.acks[i].At >= op.FlipAt { |
| - w.acks[i].Exempt = true |
| + x := &w.acks[i] |
| + if x.Box == t.ID && x.Inc == c.Holder.Inc && x.Kind == "timer" && !x.Exempt && !w.delivered[x.W] { |
| + x.Exempt = true |
| + n++ |
| } |
| } |
| - w.logf("operator declared box %d lost for %s", t.ID, op.ID) |
| + if n > 0 { |
| + w.stats["timers declared lost"] += n |
| + w.alert("box %d declared lost for %s: %d acknowledged timers registered on it are lost with it", t.ID, op.ID, n) |
| + } |
| } |
| } |
diff --git a/internal/sim/hostlife/expected-outcomes.txt b/internal/sim/hostlife/expected-outcomes.txt| index 5bbbdea22..a5b6c00b1 100644 |
| --- a/internal/sim/hostlife/expected-outcomes.txt |
| +++ b/internal/sim/hostlife/expected-outcomes.txt |
| @@ -10,8 +10,6 @@ readiness-quiet missed unreachable under claim rev 4: acknowledgements wait out |
| watchdog-independent missed caught with declare-lost-compute (witness; sweep 3 / 100,000 on rev 7.2); alone, compute fencing makes it redundant |
| exec-admission missed one of four admission fences (claim rev 6.8.1 adds exec-writer); caught with hold-op-open, start-check-claim-gen and exec-writer |
| hold-op-open missed one of four admission fences; caught with exec-admission, start-check-claim-gen and exec-writer |
| -start-check-claim-gen missed one of four admission fences; caught with exec-admission, hold-op-open and exec-writer |
| -start-check missed subsumed since rev 7.10: a box restores only its own region's remote of holder.generation, so with the whole start check gone the stale source of a cross-region move finds nothing to restore and cannot start beside the target; the witness is retired, the generation clause (start-check-claim-gen) keeps its witness |
| touched-incomplete missed caught with legacy-discovery |
| door-reread missed subsumed by C3a and, since rev 7.8, by writer epochs (claim rev 6.6): a recreated incarnation writes into a fresh generation the listing never selects; its witness is retired |
| door-by-version missed subsumed by C3a and, since rev 7.8, by writer epochs (claim rev 6.6): a recreated incarnation writes into a fresh generation the listing never selects; its witness is retired |
diff --git a/internal/sim/hostlife/ltx_test.go b/internal/sim/hostlife/ltx_test.go| index 0c486368e..e2575dcd2 100644 |
| --- a/internal/sim/hostlife/ltx_test.go |
| +++ b/internal/sim/hostlife/ltx_test.go |
| @@ -272,13 +272,27 @@ func apply(state map[string]int, f *blob) { |
| // compaction, snapshot rotation and retention run as the store's |
| // monitors would. |
| func (w *world) replicate(b *box) { |
| - if !w.agentUp(b) || b.DB == nil { |
| + if !w.agentUp(b) || b.DB == nil || b.open != nil { |
| + return // replication is not open until the admitted session's start completes (rev 7.11 §7.14) |
| + } |
| + if !b.Old && b.DB.Epoch != b.Epoch { |
| + // the replicator that was admitted died with its agent; a new agent |
| + // process is a new session (claim rev 6.10) and replicates nothing |
| + // for the old one: its wake is a new admission |
| + w.stats["replication refused: the session's agent is gone"]++ |
| return |
| } |
| if !hasMain(b) { |
| w.closeStep(b) |
| return |
| } |
| + w.flush(b) |
| + w.compact(b) |
| +} |
| + |
| +// flush is one sync of the replicator: the pending writes become the next |
| +// local level-0 file, and the queue is shipped. |
| +func (w *world) flush(b *box) { |
| db := b.DB |
| if len(db.Pending) > 0 { |
| // the sync writes this TXID's file locally; the position advances |
| @@ -289,7 +303,6 @@ func (w *world) replicate(b *box) { |
| db.Pos, db.Pending = bl.Max, nil |
| } |
| w.ship(b) |
| - w.compact(b) |
| } |
| // closedKey is a writer session's clean-close certificate (claim rev 6.8 |
| @@ -365,6 +378,14 @@ func (w *world) closeStep(b *box) { |
| if db == nil || !db.Closing || hasMain(b) { |
| return |
| } |
| + if b.open != nil || !b.Old && db.Epoch != b.Epoch { |
| + // the session's start has not opened replication, or its replicator |
| + // died with its agent: nothing certifies (rev 7.11; seed 2663 / |
| + // 30,000 certified position 0 over two unshipped writes) |
| + db.Closing, db.CloseReq = false, nil |
| + w.stats["clean close refused: start unfinished or agent gone"]++ |
| + return |
| + } |
| if len(b.Procs) > 0 { |
| // nothing of the host may run when the session certifies: a |
| // descendant that outlived the unit stop (an escapee outside the |
diff --git a/internal/sim/hostlife/moves_test.go b/internal/sim/hostlife/moves_test.go| index 5cc2902b4..eefb687a4 100644 |
| --- a/internal/sim/hostlife/moves_test.go |
| +++ b/internal/sim/hostlife/moves_test.go |
| @@ -158,6 +158,15 @@ func (w *world) role(b *box, c *claim) { |
| } else if c == nil || !w.isParticipant(c, b) { |
| b.ExecUntil = 0 |
| } |
| + if !b.Old && b.DB != nil && b.DB.Epoch != b.Epoch && len(b.Procs) > 0 { |
| + // the agent restarted under a running app whose session's |
| + // replicator died with the old agent (claim rev 6.10: a new agent |
| + // process is a new session): the app is stopped, and the next wake |
| + // is a new admission that declares what it acknowledged meanwhile |
| + b.Procs = nil |
| + w.stats["app stopped: its session's agent is gone"]++ |
| + w.logf("box %d: the agent restarted under a running app of a dead session; the app is stopped", b.ID) |
| + } |
| if c == nil { |
| return |
| } |
| @@ -180,9 +189,12 @@ func (w *world) role(b *box, c *claim) { |
| return |
| } |
| if b.open != nil { |
| - // an admitted writer finishing its start (rev 7.10 §7.14) |
| + // an admitted session finishing its start (rev 7.11 §7.14); the |
| + // operation's own step runs once the app answered readiness |
| w.resumeStart(b, c) |
| - return |
| + if !hasMain(b) && !b.hostFailed { |
| + return // the operation's own step runs once the app answered readiness, or once the host is marked failed |
| + } |
| } |
| switch c.Op.Kind { |
| case "moving": |
| @@ -268,23 +280,70 @@ func (w *world) snapshotBlobOf(b *box, c *claim, gen string) (*blob, *blob) { |
| // the final checkpoint writes the pending writes as the next local |
| // level-0 file and uploads what it can (litestream SyncAndWait); |
| // the snapshot is then at that position, and carries every page |
| - // whether or not the upload has landed |
| - if len(b.DB.Pending) > 0 { |
| - bl := w.deltaOf(b.DB, b.DB.Inc) |
| - b.DB.Queue = append(b.DB.Queue, bl) |
| - b.DB.Pos, b.DB.Pending = bl.Max, nil |
| + // whether or not the upload has landed. While the admitted session's |
| + // start has not opened replication there is no replicator to sync: |
| + // the snapshot carries the pages as they stand and nothing is |
| + // uploaded into the live generation (rev 7.11; seed 17056 / 30,000: |
| + // a round's forced sync wrote the first upload's key with other |
| + // bytes while the start's own PUT was still in flight) |
| + if b.open == nil && (b.Old || b.DB.Epoch == b.Epoch) { |
| + if len(b.DB.Pending) > 0 { |
| + bl := w.deltaOf(b.DB, b.DB.Inc) |
| + b.DB.Queue = append(b.DB.Queue, bl) |
| + b.DB.Pos, b.DB.Pending = bl.Max, nil |
| + } |
| + w.ship(b) |
| } |
| - w.ship(b) |
| db = snapshotOf(b.DB, b.DB.Inc) |
| - } else if r := w.restoreCalc(b.Region, c.Holder.Inc, c.Holder.Location, gen); r != nil { |
| - db = snapshotOf(r, c.Holder.Inc) |
| + } else if sv := w.sealVersion(c.Op.SealDB); sv != nil && c.Op.Kind == "restoring" { |
| + // a target that never restored a database returns the seal it was |
| + // given, as staged, pinned by version (rev 7.11 §7.8) |
| + cp := *sv.B |
| + db = &cp |
| } else { |
| - db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}} |
| + // no local database (a lost disk): the remote as the next admission |
| + // would restore it, a lagging marker read through the journal (rev |
| + // 7.11; seeds 3987 and 14322 / 30,000 sealed an empty database over |
| + // a generation whose marker had not landed, and the push lost the |
| + // legacy host's writes) |
| + var r *localDB |
| + if gen == c.Holder.Gen { |
| + r = w.restoreAsNext(b, c) |
| + } else { |
| + r = w.restoreCalc(b.Region, c.Holder.Inc, c.Holder.Location, gen) |
| + } |
| + if r != nil { |
| + db = snapshotOf(r, c.Holder.Inc) |
| + } else if !w.hasFiles(b, c.Holder.Location, gen) && !w.hasFiles(b, c.Holder.Location, c.Holder.Gen) { |
| + db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}} // nothing was ever written |
| + } else { |
| + w.stats["seal refused: nothing to restore"]++ |
| + return nil, nil // the remote cannot be read: no seal |
| + } |
| } |
| files := &blob{Kind: "files", Inc: c.Holder.Inc, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...)} |
| + if b.Files == nil { |
| + if fv := w.sealVersion(c.Op.SealFiles); fv != nil && c.Op.Kind == "restoring" { |
| + cp := *fv.B |
| + files = &cp |
| + } |
| + } |
| return db, files |
| } |
| +// sealVersion is the staged seal object a journal ref names, or nil. |
| +func (w *world) sealVersion(r ref) *version { |
| + if r.Ver == 0 { |
| + return nil |
| + } |
| + for _, v := range w.buckets[r.Region].objs[r.Key] { |
| + if v.ID == r.Ver { |
| + return v |
| + } |
| + } |
| + return nil |
| +} |
| + |
| func (w *world) moveSource(b *box, c *claim) { |
| op := c.Op |
| switch op.Phase { |
| @@ -392,6 +451,11 @@ func (w *world) moveSource(b *box, c *claim) { |
| return |
| } |
| db, files := w.snapshotBlob(b, c) |
| + if db == nil { |
| + w.logf("%s: the source has no database and its remote cannot be restored; reverting", op.ID) |
| + w.startRevert(c) |
| + return |
| + } |
| if op.SealDB.Ver == 0 { |
| r := &request{kind: rPut, region: op.TgtRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.ID), b: db} |
| w.maybeStallDuplicate(b, op, r) |
| @@ -408,10 +472,21 @@ func (w *world) moveSource(b *box, c *claim) { |
| w.maybeStallDuplicate(b, op, r) |
| v := w.issue(b.ID, r) |
| if v > 0 { |
| + // does the source's own replica cover the stopped database? The |
| + // final checkpoint's upload may still be in flight or lost; the |
| + // seal holds every page regardless. A clean rollback restores the |
| + // source from its replica only when it does (rev 7.11, INV-15; |
| + // seed 17056 / 30,000) |
| + 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 |
| + if !covered { |
| + w.stats["seal: source replica short of the stop"]++ |
| + } |
| w.cas(c, func(n *claim) bool { |
| if !sameOp(n.Op.ID, op.ID) { |
| return false |
| } |
| + n.Op.SrcCovered = covered |
| + n.Op.SealFromDB = b.DB != nil |
| n.Op.SealFiles = ref{Region: op.TgtRegion, Key: fmt.Sprintf("moves/%s/%s/files", host, op.ID), Ver: v} |
| n.Op.SealQuiet = w.now + stagingTTL + horizonH |
| if !on("staging-quiet") { |
| @@ -705,8 +780,10 @@ func (w *world) revertSource(b *box, c *claim) { |
| w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probed"; return sameOp(n.Op.ID, op.ID) }) |
| return |
| } |
| - if b.open != nil { |
| - return // admitted, waiting for its PUT: not a failed try |
| + if b.hostFailed && !b.failCounted { |
| + b.failCounted = true // a host the retry policy marked failed: one failed try for the operation |
| + } else if b.open != nil || b.hostFailed { |
| + return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try |
| } |
| w.cas(c, func(n *claim) bool { |
| n.Op.ProbeTries++ |
| @@ -716,6 +793,14 @@ func (w *world) revertSource(b *box, c *claim) { |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| case "probed": |
| + if b.open != nil { |
| + return // the source's session has not opened replication yet (rev 7.11 §7.14) |
| + } |
| + if !hasMain(b) { |
| + w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11) |
| + w.cas(c, func(n *claim) bool { n.Op.RevPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| + return |
| + } |
| if w.openRow(b, op.ID) { |
| w.cas(c, func(n *claim) bool { n.Op.RevPhase = "opened"; return sameOp(n.Op.ID, op.ID) }) |
| } |
| @@ -956,8 +1041,12 @@ func (w *world) landTarget(b *box, c *claim) { |
| w.cas(c, func(n *claim) bool { n.Op.LandPhase, n.Op.ProbedAt = "probed", w.now; return sameOp(n.Op.ID, op.ID) }) |
| return |
| } |
| + if b.hostFailed { |
| + w.rollback(c, "target would not start") // the retry policy marked the host failed (rev 7.11 §7.14) |
| + return |
| + } |
| if b.open != nil { |
| - return // admitted, waiting for its PUT: not a failed try |
| + return // admitted, its start in progress or in backoff: not a failed try |
| } |
| tries := op.ProbeTries + 1 |
| w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) }) |
| @@ -980,8 +1069,12 @@ func (w *world) landTarget(b *box, c *claim) { |
| w.cas(c, func(n *claim) bool { n.Op.ProbedAt = w.now; return sameOp(n.Op.ID, op.ID) }) |
| return |
| } |
| + if b.hostFailed { |
| + w.rollback(c, "probe died") |
| + return |
| + } |
| if b.open != nil { |
| - return // admitted, waiting for its PUT: not a failed try |
| + return // admitted, its start in progress: not a failed try |
| } |
| tries := op.ProbeTries + 1 |
| w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) }) |
| @@ -990,6 +1083,9 @@ func (w *world) landTarget(b *box, c *claim) { |
| } |
| return |
| } |
| + if b.open != nil { |
| + return // the probe's session has not opened replication yet (rev 7.11 §7.14): the settle window waits |
| + } |
| if w.now < op.ProbedAt+15 { |
| if w.fail["lostreturn"] && w.rng.Intn(2) == 0 { |
| w.traffic(b) // the biased schedule: the app does real work in the settle window |
| @@ -1049,7 +1145,11 @@ func (w *world) installLocal(b *box, c *claim, gen string, seal ref, _ string, n |
| if st == 0 { |
| continue |
| } |
| - if c.Op.Kind == "restoring" && st != b.stampOf(c.Holder.Inc) { |
| + if c.Op.Kind == "restoring" && st != b.stampOf(c.Holder.Inc) && !b.stampRetired(c.Holder.Inc, st) { |
| + // a retired stamp's leftover is fair game: a cold admission that |
| + // issued a stamp and then failed its purge leaves the old |
| + // installation's database under a newer active stamp (rev |
| + // 7.11, the CAS before the purge; seed 65090 / 100,000) |
| 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)) |
| } |
| if c.Op.Kind == "landing" && st == b.stampOf(c.Holder.Inc) { |
| @@ -1103,7 +1203,12 @@ func (w *world) release(c *claim, execBox int) { |
| if !sameOp(n.Op.ID, op.ID) { |
| return false |
| } |
| - n.Op = opState{Kind: "serving", Exec: execBox, Parts: []int{execBox}} |
| + // the writers journal survives the release (rev 7.11): an admission |
| + // whose start is unfinished at the release stays explained, so its |
| + // predecessor stays live and the lagging marker is no alert (seed |
| + // 1052 / 30,000: an abort released over a source admitted a tick |
| + // earlier) |
| + n.Op = opState{Kind: "serving", Exec: execBox, Parts: []int{execBox}, Writers: n.Op.Writers} |
| return true |
| }) { |
| w.logf("released %s to serving on box %d", op.ID, execBox) |
| @@ -1151,6 +1256,13 @@ func (w *world) releaseTarget(b *box, c *claim) { |
| if !on("dirty-return") { |
| dirty = false |
| } |
| + if !op.SrcCovered && on("seal-return") { |
| + // the source's replica stops short of its stopped database: only |
| + // the seal holds every acknowledged write, so the return carries |
| + // it whether or not the target wrote (rev 7.11 §7.8) |
| + dirty = true |
| + w.stats["rollback: return carries the seal"]++ |
| + } |
| w.cas(c, func(n *claim) bool { |
| if !sameOp(n.Op.ID, op.ID) || n.Op.Exec != b.ID { |
| return false |
| @@ -1179,13 +1291,18 @@ func (w *world) releaseTarget(b *box, c *claim) { |
| return |
| } |
| db, files := w.snapshotBlobOf(b, c, op.ID) |
| + if db == nil { |
| + return // nothing to return yet: the remote could not be read |
| + } |
| if op.ReturnDB.Ver == 0 { |
| r := &request{kind: rPut, region: op.SrcRegion, key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), b: db} |
| w.maybeStallDuplicateFor(b, op.SrcRegion, r) |
| v := w.issue(b.ID, r) |
| if v > 0 { |
| + fromDB := b.DB != nil |
| w.cas(c, func(n *claim) bool { |
| n.Op.ReturnDB = ref{Region: op.SrcRegion, Key: fmt.Sprintf("moves/%s/%s/db", host, op.RollbackID), Ver: v} |
| + n.Op.ReturnFromDB = fromDB |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| } |
| @@ -1346,10 +1463,12 @@ func (w *world) recoverSource(b *box, c *claim) { |
| w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probed"; return sameOp(n.Op.ID, op.ID) }) |
| return |
| } |
| - if b.open != nil { |
| - return // admitted, waiting for its PUT: not a failed try |
| + if b.hostFailed && !b.failCounted { |
| + b.failCounted = true // a host the retry policy marked failed: one failed try for the operation |
| + } else if b.open != nil || b.hostFailed { |
| + return // admitted, its start in progress, or a failed host awaiting the operator: not a failed try |
| } |
| - w.logf("box %d: the rollback's probe did not start (try %d, generation %q, db=%v)", b.ID, op.ProbeTries+1, want, b.DB != nil) |
| + 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) |
| w.cas(c, func(n *claim) bool { |
| n.Op.ProbeTries++ |
| if n.Op.ProbeTries >= probeTries*3 { |
| @@ -1358,6 +1477,14 @@ func (w *world) recoverSource(b *box, c *claim) { |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| case "probed": |
| + if b.open != nil { |
| + return // the source's session has not opened replication yet (rev 7.11 §7.14) |
| + } |
| + if !hasMain(b) { |
| + w.stats["probe restarted"]++ // the probe died: restarted, admitted like any other (rev 7.11) |
| + w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| + return |
| + } |
| if op.Dirty && !op.ReturnParked { |
| // holder.generation and holder.stamp moved at the probe's cold |
| // admission (claim rev 6.9); the return lane's park is owed |
| @@ -1391,6 +1518,14 @@ func (w *world) pushStep(b *box, c *claim) { |
| // the push's snapshot: the replica under the final checkpoint, plus |
| // the push's own mark, one more acknowledged transaction |
| snap, _ := w.snapshotBlobOf(b, c, c.Holder.Gen) |
| + if snap == nil { |
| + // nothing to seal: the push is abandoned before it changed |
| + // anything (rev 7.11) |
| + w.stats["push abandoned: nothing to seal"]++ |
| + w.logf("push %s abandoned: the box has no database and its remote cannot be restored", op.ID) |
| + w.release(c, b.ID) |
| + return |
| + } |
| snap.Gen = g |
| w.nextW++ |
| mark := w.nextW |
| @@ -1404,7 +1539,7 @@ func (w *world) pushStep(b *box, c *claim) { |
| if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" { |
| return false |
| } |
| - n.Op.Phase, n.Op.StopAt, n.Op.PushSnap = "park", w.now, snap |
| + n.Op.Phase, n.Op.StopAt, n.Op.PushSnap, n.Op.SealFromDB = "park", w.now, snap, b.DB != nil |
| return true |
| }) |
| case "park": |
| @@ -1462,6 +1597,18 @@ func (w *world) pushStep(b *box, c *claim) { |
| }) |
| } |
| case "probed": |
| + if b.open != nil { |
| + return // the admitted session's start is not complete: replication is not open yet (rev 7.11 §7.14) |
| + } |
| + if !hasMain(b) { |
| + // the probe died before its session opened replication or in |
| + // its settle (seed 487 / 30,000: a push released over a dead |
| + // probe whose marker never landed): the probe restarts, a cold |
| + // admission like any other |
| + w.stats["probe restarted"]++ |
| + w.cas(c, func(n *claim) bool { n.Op.Phase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| + return |
| + } |
| if w.openRow(b, op.ID) || b.Row != nil && b.Row.Status == "ready" { |
| w.release(c, b.ID) |
| } |
| @@ -1606,6 +1753,32 @@ func (w *world) deadGeneration(c *claim, r int, k string) bool { |
| if !on("live-set-region") { |
| authoritative = true |
| } |
| + if v := w.buckets[r].current(genMarkerKey(h.Location)); authoritative && v != nil && v.B.Gen != h.Gen { |
| + // the marker disagrees with holder.generation. Normal while the |
| + // latest admission is unfinished: the marker names its predecessor, |
| + // and both generations are live (claim rev 6.10, R10-2). Any other |
| + // disagreement: touch nothing under the location, and alert (§6.4) |
| + chain, ok := w.unfinishedChain(c, v.B.Gen) |
| + if ok && on("baseline-live") { |
| + for _, g := range append(chain, v.B.Gen) { |
| + if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, g, "") { |
| + return false |
| + } |
| + if isClosedKey(k) && strings.HasPrefix(k, dataKey(h.Location, g, "closed/")) { |
| + return false |
| + } |
| + } |
| + if k == genMarkerKey(h.Location) { |
| + return false |
| + } |
| + } else if !w.markerAlerted[h.Location] { |
| + w.markerAlerted[h.Location] = true |
| + 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, v.B.Gen, h.Gen) |
| + return false |
| + } else { |
| + return false |
| + } |
| + } |
| if k == genMarkerKey(h.Location) { |
| v := w.buckets[r].current(k) |
| return v != nil && (v.B.Gen != h.Gen || !authoritative) |
diff --git a/internal/sim/hostlife/protocol_test.go b/internal/sim/hostlife/protocol_test.go| index a1dc90e1c..3073e5d5f 100644 |
| --- a/internal/sim/hostlife/protocol_test.go |
| +++ b/internal/sim/hostlife/protocol_test.go |
| @@ -359,7 +359,7 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| // an old executable: no claim, no markers, no locations; it |
| // restores whatever is newest at the legacy root |
| if b.DB == nil { |
| - b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/", State: map[string]int{}} |
| + b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/", State: map[string]int{}, Epoch: b.Epoch} |
| var newest *version |
| for _, k := range w.buckets[b.Region].keys("instances/" + host + "/ltx/") { |
| if v := w.buckets[b.Region].current(k); v != nil && (newest == nil || v.ID > newest.ID) { |
| @@ -387,19 +387,35 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| // probe and a restarted probe, a reverted source, an opened rollback |
| // recovery, a push's writer. A writer still closing holds the lease: |
| // the wake waits for it |
| + // the retry policy for failed starts (rev 7.11 §7.14, claim rev |
| + // 6.10): a host marked failed gets no new session and no further |
| + // attempt until an operator clears it; a failed start within a |
| + // session waits out its backoff |
| + if b.hostFailed { |
| + w.stats["start refused: host failed"]++ |
| + return false |
| + } |
| + if w.now < b.nextStart { |
| + w.stats["start: backoff"]++ |
| + return false |
| + } |
| if o := b.open; o != nil { |
| - // an admitted writer whose start is in progress: it waits for the |
| - // PUT it needs acknowledged and starts without another CAS; an |
| - // admission that moved on (its session or generation changed) |
| - // abandons it |
| + // an admitted session whose start is in progress continues it |
| + // without another CAS; an admission that moved on (its session |
| + // or generation changed) abandons it |
| if o.Gen != h.Gen || o.Session != h.Session { |
| b.open = nil |
| - } else if !w.resumeOpen(b, c, o) { |
| - return false |
| } else { |
| - return w.startProc(b, c, want, probe, o.Mode) |
| + return w.resumeOpen(b, c, o) |
| } |
| } |
| + // the launch environment: config.env rendered and the release live |
| + // for this incarnation; until then the agent admits nothing (a |
| + // config pass has not run yet, not an app failure) |
| + if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) { |
| + w.stats["start: environment not ready"]++ |
| + return false |
| + } |
| if b.DB != nil && b.DB.Closing { |
| w.closeStep(b) |
| if b.DB.Closing { |
| @@ -459,12 +475,31 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| return false |
| } |
| } |
| + dead := []string{h.Gen} // what the new marker retires (claim rev 6.10 QueueGenerationRetire) |
| if bg != h.Gen && on("start-check") { |
| - if len(w.buckets[b.Region].keys(dataKey(loc, h.Gen, ""))) == 0 { |
| - // the baseline is the nearest ancestor that holds a file |
| - for source != bg && len(w.buckets[b.Region].keys(dataKey(loc, source, ""))) == 0 { |
| - source, _ = w.parentGen(loc, source) |
| + if !w.hasFiles(b, loc, h.Gen) { |
| + // an empty holder.generation: only an unfinished admission's, |
| + // journalled, whose marker still names its predecessor, or |
| + // the predecessor of a chain of them (claim rev 6.10); the |
| + // baseline is the nearest generation of the chain that holds |
| + // a file, else the marker's, and every generation of the chain |
| + // is dead once the new marker lands. Any other empty |
| + // generation is unexplained missing data: refuse and alert |
| + chain, ok := w.unfinishedChain(c, bg) |
| + returned := c.Op.Kind == "restoring" && b.ID == c.Op.Src && h.Gen == c.Op.SrcGen |
| + if !returned && !ok { |
| + w.alert("box %d: generation %q is empty and is not the latest admission's unfinished one (marker %q); the start is refused", b.ID, h.Gen, bg) |
| + w.stats["start refused: unexplained empty generation"]++ |
| + return false |
| } |
| + source = bg |
| + for _, g := range chain { |
| + if g != h.Gen && w.hasFiles(b, loc, g) { |
| + source = g |
| + break |
| + } |
| + } |
| + dead = append(append([]string{h.Gen}, chain[min(1, len(chain)):]...), bg) |
| w.stats["abandoned empty generation skipped"]++ |
| } else if !w.republishMarker(b, loc, h.Gen) { |
| // no earlier version names it: the admitted writer's marker |
| @@ -480,64 +515,52 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| } |
| w.stats["marker republished"]++ |
| w.logf("box %d: generation.json published again for %q, which the admitted writer never published", b.ID, h.Gen) |
| - } |
| - } |
| - // a box never reuses its local database (claim rev 6.9, R9-1): the |
| - // admitted writer purges it, and if the purge fails it does not |
| - // start; then it restores from the certified remote baseline. The |
| - // mutation of remote-baseline keeps a local database of the |
| - // generation it opens, as rev 7.9 and today's activator do |
| - if b.DB != nil && (on("remote-baseline") || b.DB.Inc != h.Inc || b.DB.Gen != adm.NewGeneration) { |
| - if b.PurgeFails { |
| - b.PurgeFails = false // a transient error: the next attempt's purge succeeds |
| - w.stats["purge failed: writer not started"]++ |
| - w.logf("box %d: the purge of its local database failed; the writer does not start", b.ID) |
| - return false |
| - } |
| - b.DB = nil |
| - b.Parked++ |
| - w.stats["local database purged"]++ |
| - } |
| - db := b.DB |
| - if db == nil { |
| - switch { |
| - case prepared: |
| - // the operation's prepared generation: its pinned snapshot and anchor |
| - db = w.restorePinned(b.Region, h.Inc, loc, want, c.Op.BaseKey, c.Op.BaseVer, c.Op.BaseTX) |
| - default: |
| - db = w.restoreCalc(b.Region, h.Inc, loc, source) |
| - if db != nil && adm.Mode == "warm" && db.Pos < adm.RemoteMaxTXID { |
| - w.logf("box %d: the restore of %q reached position %d, the certificate says %d", b.ID, source, db.Pos, adm.RemoteMaxTXID) |
| - db = nil |
| + // the marker names holder.generation now: the dead writer's |
| + // predecessor, and the chain's, are retired as its own marker |
| + // stage would have (seed 8528 / 30,000: a warm successor |
| + // continued the generation and nothing ever retired rec19) |
| + chain, _ := w.unfinishedChain(c, bg) |
| + lagDead := []string{bg} |
| + for _, g := range chain { |
| + if g != h.Gen && !hasStr(lagDead, g) { |
| + lagDead = append(lagDead, g) |
| + } |
| + } |
| + lagDead = w.uncovered(b, c, loc, h.Gen, lagDead) |
| + if !w.queueGenerationRetire(b, c, h.Session, h.Gen, b.Region, lagDead) { |
| + return false // the claim moved: next time |
| } |
| } |
| - if db == nil { |
| - w.stats["restore refused"]++ |
| - w.logf("box %d: nothing to restore for a %s admission from %q (generation.json %q)", b.ID, adm.Mode, source, bg) |
| - return false |
| - } |
| - db.Stamp = b.stampOf(h.Inc) |
| - w.logf("box %d restored %s: TXID %d at position %d, %d writes (%s admission into %q)", b.ID, dataKey(loc, source, ""), db.TX, db.Pos, len(db.Log), adm.Mode, adm.NewGeneration) |
| } |
| - b.DB = db |
| - if on("start-check") { |
| - if b.DB.Inc != h.Inc && !(h.Legacy && b.DB.Inc == 0) { |
| - w.logf("box %d start check refused: db inc %d", b.ID, b.DB.Inc) |
| - return false |
| - } |
| - // .files-gen names the generation the files were installed or |
| - // carried into (F7): the one the marker names, holder.generation, |
| - // or the one this admission opens (the admission retags them) |
| - // .files-gen (F7) is retagged by every admission with the |
| - // generation it opens; a stale tag from an earlier stay is not a |
| - // refusal, the incarnation is |
| - if b.Files != nil && b.Files.Inc != h.Inc { |
| - w.logf("box %d start check refused: files inc %d, claim inc %d", b.ID, b.Files.Inc, h.Inc) |
| - return false |
| + // the admitting CAS first (claim rev 6.10 §6.4a): before the purge, |
| + // the restore and the app start, so nothing the app acknowledges is |
| + // ever outside a session (R10-1). Everything after it is the admitted |
| + // session's start, finished by the agent's own loop (resumeOpen) |
| + if !on("admit-before-start") && w.startHook != nil { |
| + // the mutation (rev 7.10's order): the app is already running |
| + // and may acknowledge work before the admitting CAS; a crash |
| + // here leaves the acknowledgement in a database the next warm |
| + // admission purges silently (R10-1) |
| + if w.startHook(b) { |
| + return false // the witness's crash |
| } |
| } |
| + if !w.admitCommit(b, c, adm) { |
| + return false |
| + } |
| + nh := w.claim.Holder |
| + o := &openState{Gen: nh.Gen, Want: want, Session: nh.Session, Mode: adm.Mode, Prepared: prepared, Probe: probe, Source: source, Dead: dead, RemoteMax: adm.RemoteMaxTXID, stage: "restore"} |
| + b.open = o |
| + return w.resumeOpen(b, c, o) |
| } |
| - // what the app sees |
| + return w.startProc(b, c, want, probe, "legacy") |
| +} |
| + |
| +// appStart is the app's start on its restored database and its readiness |
| +// (§7.7): what the app sees, the launch environment and the readiness |
| +// answer. It starts the processes on success. |
| +func (w *world) appStart(b *box, c *claim, want string, probe bool, mode string) bool { |
| + h := c.Holder |
| w.foreignWrites(b.DB, h.Inc, fmt.Sprintf("box %d start", b.ID)) |
| if b.DB.Inc != 0 && b.DB.Inc != h.Inc { |
| w.violate("box %d started inc %d over a database of inc %d", b.ID, h.Inc, b.DB.Inc) |
| @@ -550,7 +573,7 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| } |
| if b.Files == nil { |
| w.nextID++ |
| - b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: want, Env: w.nextID, Done: true, Stamp: b.stampOf(h.Inc)} |
| + b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: b.DB.Gen, Env: w.nextID, Done: true, Stamp: b.stampOf(h.Inc)} |
| } |
| if b.DB.Key == 0 { |
| b.DB.Key = b.Files.Key // unsealed data (never booted) adopts the key |
| @@ -559,8 +582,12 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| w.violate("INV-4: box %d started inc %d with instance.key %d over data sealed with %d", b.ID, h.Inc, b.Files.Key, b.DB.Key) |
| return false |
| } |
| + if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) { |
| + w.stats["start: environment not ready"]++ |
| + return false // not an app failure: the agent waits for its config pass |
| + } |
| // FailApp is the injected "the target cannot run it" fault: landing probes only |
| - healthy := b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc && !(probe && b.FailApp && c.Op.Kind == "landing") |
| + healthy := !(probe && b.FailApp && c.Op.Kind == "landing") |
| // §7.7: the probe checks the launch environment at spawn. The forward |
| // path requires the inputs pinned at admission. A recovery probe (the |
| // source's revert or rollback) runs on the inputs it rendered now and |
| @@ -578,63 +605,418 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| w.logf("box %d unhealthy start inc %d: cfg=%d live=%s/%d fail=%v kind=%s", b.ID, h.Inc, b.Config, b.Live, b.LiveInc, b.FailApp, c.Op.Kind) |
| return false |
| } |
| - if !b.Old { |
| - // the admitting CAS, after readiness and before any upload (claim rev |
| - // 6.9 §6.4a); the writer then opens under the session and generation |
| - // it produced. The replica reads its remote position at its first |
| - // sync (litestream's Replica.Sync initializes it lazily), RemotePos -1 |
| - if !w.admitCommit(b, c, adm) { |
| - return false |
| - } |
| - db := b.DB |
| - nh := w.claim.Holder |
| - db.Gen, db.Session, db.Stamp = nh.Gen, nh.Session, nh.Stamp |
| - if b.Files != nil { |
| - // the carried files are tagged with the installing CAS's stamp |
| - // at the CAS (§6.4), whether or not the start completes: the |
| - // record activates that stamp from its next read (seed 62 / 300) |
| - b.Files.Gen, b.Files.Stamp = nh.Gen, nh.Stamp |
| - } |
| - db.Pending, db.Queue, db.InFlight, db.Issued, db.Uncertain, db.Closing, db.CloseReq = nil, nil, nil, nil, false, false, nil |
| - db.RemotePos, db.FirstFull = -1, true // litestream's first sync after a cold initialization is a full upload (fact R5-1c) |
| - o := &openState{Gen: nh.Gen, Want: want, Session: nh.Session, Mode: adm.Mode, Prepared: adm.Mode == "cold" && adm.NewGeneration == want && want != h.Gen, DB: db} |
| - switch { |
| - case adm.Mode == "cold" && o.Prepared: |
| - db.Pos = w.maxL0(w.buckets[b.Region], dataKey(loc, db.Gen, "")) // the prepared generation's anchor |
| - o.stage = "marker" |
| - case adm.Mode == "cold": |
| - // a fresh generation is empty until the writer's first upload, |
| - // the full file [1, 1] of the restored baseline (fact R5-1c), |
| - // made after the CAS and before the marker: generation.json |
| - // never names a generation without a root (INV-11), so a writer |
| - // that dies before its first upload leaves an empty, unpublished |
| - // generation whose parent stays the restore's baseline (§7.14) |
| - db.Pos = 0 |
| - o.stage = "marker" |
| - if len(db.Log) > 0 || len(db.State) > 0 { |
| + return w.startProc(b, c, want, probe, mode) |
| +} |
| + |
| +// the retry policy for failed app starts within a session (rev 7.11 |
| +// §7.14): up to startAttemptsMax consecutive failures, the next attempt |
| +// after a doubling backoff from startBackoff0 steps; then the host is |
| +// marked failed, alerted, and gets no new sessions until an operator |
| +// clears it. The sum of the backoffs is 15 steps; a start that fails |
| +// five times in a row is the release or the configuration, which only |
| +// an operator or a push changes, not the dependency that was slow. |
| +const ( |
| + startAttemptsMax = 5 |
| + startBackoff0 = 1 |
| +) |
| + |
| +// resumeOpen advances an admitted session's start (rev 7.11 §7.14, claim |
| +// rev 6.10): purge and restore, app start and readiness, then the PUTs |
| +// before replication opens (a fresh generation's full first upload, the |
| +// marker, a warm restore's anchor), each acknowledged before the next, |
| +// retried with the same bytes after the client's timeout, and every one |
| +// of them in the session's ledger (R10-7). It returns true once replication |
| +// is open. The retire chore for a fresh generation's predecessor is queued |
| +// only once the marker names the new generation (R10-2). |
| +func (w *world) resumeOpen(b *box, c *claim, o *openState) bool { |
| + cur := w.claim |
| + if cur == nil || cur.Holder == nil || cur.Holder.Gen != o.Gen || cur.Holder.Session != o.Session || on("exec-writer") && cur.Op.Exec != b.ID { |
| + // the session moved on, or op.exec was handed to another box by a |
| + // CAS citing this box's stop evidence (a rollback after the retry |
| + // policy marked the host failed, seed 25581 / 30,000): the start is |
| + // abandoned, never resumed |
| + b.open = nil |
| + w.stats["open: abandoned"]++ |
| + return false |
| + } |
| + h := cur.Holder |
| + loc := h.Location |
| + bk := w.buckets[b.Region] |
| + for { |
| + switch o.stage { |
| + case "restore": |
| + // a box never reuses its local database (claim rev 6.9, R9-1): |
| + // the purge, error-returning; then the restore from the remote. |
| + // The mutation of remote-baseline keeps a local database of the |
| + // generation it opens, as rev 7.9 and today's activator do |
| + if b.DB != nil && (on("remote-baseline") || b.DB.Inc != h.Inc || b.DB.Gen != o.Gen) { |
| + if b.PurgeFails { |
| + b.PurgeFails = false // a transient error: the next attempt's purge succeeds |
| + w.stats["purge failed: writer not started"]++ |
| + w.logf("box %d: the purge of its local database failed; the start waits", b.ID) |
| + return false |
| + } |
| + b.DB = nil |
| + b.Parked++ |
| + w.stats["local database purged"]++ |
| + } |
| + db := b.DB |
| + if db == nil { |
| + switch { |
| + case o.Prepared: |
| + db = w.restorePinned(b.Region, h.Inc, loc, o.Gen, cur.Op.BaseKey, cur.Op.BaseVer, cur.Op.BaseTX) |
| + default: |
| + db = w.restoreCalc(b.Region, h.Inc, loc, o.Source) |
| + if db != nil && o.Mode == "warm" && db.Pos < o.RemoteMax { |
| + w.alert("box %d: the restore of %q reached position %d, the certificate says %d; the remote is incomplete", b.ID, o.Source, db.Pos, o.RemoteMax) |
| + db = nil |
| + } |
| + } |
| + if db == nil { |
| + w.stats["restore refused"]++ |
| + w.logf("box %d: nothing to restore for a %s admission from %q", b.ID, o.Mode, o.Source) |
| + return false |
| + } |
| + w.logf("box %d restored %s: TXID %d at position %d, %d writes (%s admission into %q)", b.ID, dataKey(loc, o.Source, ""), db.TX, db.Pos, len(db.Log), o.Mode, o.Gen) |
| + } |
| + db.Gen, db.Session, db.Stamp, db.Epoch = o.Gen, o.Session, h.Stamp, b.Epoch |
| + db.Pending, db.Queue, db.InFlight, db.Issued, db.Uncertain, db.Closing, db.CloseReq = nil, nil, nil, nil, false, false, nil |
| + db.RemotePos, db.FirstFull = -1, true // litestream's first sync after a cold initialization is a full upload (fact R5-1c) |
| + if b.Files != nil { |
| + b.Files.Gen, b.Files.Stamp = o.Gen, h.Stamp // the carried files, retagged (§6.4) |
| + } |
| + switch { |
| + case o.Mode == "cold" && o.Prepared: |
| + db.Pos = w.maxL0(bk, dataKey(loc, o.Gen, "")) // the prepared generation's anchor |
| + case o.Mode == "cold": |
| + db.Pos = 0 // a fresh generation: empty until the first upload |
| + } |
| + b.DB = db |
| + o.DB = db |
| + o.stage = "app" |
| + case "app": |
| + // the app starts on the restored database and answers its |
| + // readiness path; replication is not open yet, so nothing it |
| + // acknowledges is outside the session (R10-1). A failed start is |
| + // retried by the same agent under the same session, with backoff, |
| + // up to the limit |
| + if b.DB == nil { |
| + o.stage = "restore" // the database went with a lost disk: restore again |
| + continue |
| + } |
| + if !hasMain(b) { |
| + if !(b.Config == h.Inc && b.Live == h.App && b.LiveInc == h.Inc) { |
| + return false // the environment: wait, no failure counted |
| + } |
| + if !w.appStart(b, cur, o.Want, o.Probe, o.Mode) { |
| + b.startFails++ |
| + b.nextStart = w.now + startBackoff0<<(b.startFails-1) |
| + w.stats["start failed"]++ |
| + if b.startFails >= startAttemptsMax { |
| + b.hostFailed, b.failedAt = true, w.now |
| + b.open = nil // no session continues on a failed host; the operator's clearing leads to a fresh admission |
| + w.stats["host marked failed"]++ |
| + w.alert("box %d: the host failed to start %d times; marked failed, no new sessions until an operator clears it", b.ID, b.startFails) |
| + } |
| + return false |
| + } |
| + b.startFails = 0 |
| + if on("admit-before-start") && w.startHook != nil { |
| + if w.startHook(b) { // a witness's background work right after the start, inside the session |
| + return false // the witness's crash |
| + } |
| + } |
| + } |
| + db := o.DB |
| + switch { |
| + case o.Mode == "cold" && !o.Prepared && (len(db.Log) > 0 || len(db.State) > 0 || len(db.Pending) > 0): |
| o.stage = "first" |
| + case o.Mode == "cold": |
| + o.stage = "marker" |
| + case db.Pos > w.maxL0(bk, dataKey(loc, o.Gen, "")) && on("restore-anchor"): |
| + o.stage = "anchor" // the restored position is above the level-0 maximum: the anchor [N, N] (§5.1 step 7) |
| + default: |
| + o.stage = "open" |
| + } |
| + case "first", "anchor": |
| + db := o.DB |
| + var key string |
| + if o.blob == nil { |
| + // the bytes are captured once; a retry carries the same bytes |
| + // (R10-7). Rev 7.11's first build rebuilt the file at the retry |
| + // and so carried a write the original did not; when the |
| + // original landed last it replaced the retry's version and the |
| + // write, durable in the retry's, was gone from the restore |
| + // (seed 7574 / 30,000) |
| + if o.stage == "first" { |
| + o.blob = w.deltaOf(db, h.Inc) |
| + } else { |
| + o.blob = anchorOf(db, h.Inc) |
| + } |
| + o.pendAt = len(db.Pending) |
| + } |
| + if o.stage == "first" { |
| + key = ltxKey(loc, o.Gen, 0, 1, 1) |
| + } else { |
| + key = ltxKey(loc, o.Gen, 0, db.Pos, db.Pos) |
| + } |
| + mk := func() *request { return &request{kind: rPut, region: b.Region, key: key, b: o.blob} } |
| + clientOK := o.req != nil && o.req.done && o.req.doneAt-o.req.at < uploadTimeout |
| + if o.req == nil || (!clientOK && w.now-o.req.at >= uploadTimeout) { |
| + if o.req != nil { |
| + // the client gave up (a timeout, or a completion it never |
| + // saw: a lost response, seed 571 / 30,000) and retries the |
| + // same bytes: the session is uncertain for good (R10-7, R9-5) |
| + db.Uncertain = true |
| + w.stats["open: PUT retried"]++ |
| + } |
| + r := mk() |
| + if o.stage == "anchor" { |
| + w.stats["restore anchors"]++ |
| + } else { |
| + w.stats["first uploads"]++ |
| + } |
| + v := w.issue(b.ID, r) |
| + db.track(r, v) |
| + o.req = r |
| + if v > 0 { |
| + o.req = nil |
| + } |
| + } |
| + if o.req != nil && !(o.req.done && o.req.doneAt-o.req.at < uploadTimeout) { |
| + // the client has not seen success: a server completion after |
| + // the wait is a lost response, not an acknowledgement |
| + if o.stage == "anchor" && !on("anchor-acked") { |
| + // the mutation opens before the anchor is acknowledged |
| + } else { |
| + w.stats["open: waits for its "+o.stage]++ |
| + return false |
| + } |
| + } |
| + if cv := bk.current(key); cv != nil { |
| + w.markDurable(key, cv.B) |
| + } |
| + // the writes acknowledged since the capture are still pending |
| + if o.pendAt <= len(db.Pending) { |
| + db.Pending = append([]int(nil), db.Pending[o.pendAt:]...) |
| } |
| - case db.Pos > w.maxL0(w.buckets[b.Region], dataKey(loc, db.Gen, "")) && on("restore-anchor"): |
| - o.stage = "anchor" // the restored position is above the level-0 maximum: the anchor [N, N], after the CAS (§5.1 step 7) |
| - default: |
| - o.stage = "start" |
| + if o.stage == "first" { |
| + db.Pos, db.FirstFull = 1, false |
| + o.stage = "marker" |
| + } else { |
| + o.stage = "open" |
| + } |
| + o.req, o.blob, o.pendAt = nil, nil, 0 |
| + case "marker": |
| + db := o.DB |
| + bg, _ := w.bucketGen(b, loc) |
| + if bg != o.Gen { |
| + ver, k, tx := 0, "", 0 |
| + if o.Prepared { |
| + ver, k, tx = cur.Op.BaseVer, cur.Op.BaseKey, cur.Op.BaseTX |
| + } |
| + if o.req == nil || o.req.done || w.now-o.req.at >= uploadTimeout { |
| + if o.req != nil && !o.req.done { |
| + w.stats["open: marker PUT retried"]++ |
| + } |
| + v := w.publishGen(b, loc, o.Gen, h.Inc, ver, k, tx) |
| + o.req = w.lastIssued |
| + if v > 0 { |
| + o.req = nil |
| + } |
| + w.logf("box %d: marker PUT for %q issued (acknowledged %v)", b.ID, o.Gen, v > 0) |
| + } |
| + if bg, _ = w.bucketGen(b, loc); bg != o.Gen { |
| + w.stats["open: waits for the marker"]++ |
| + w.logf("box %d: admitted into %q, waiting for its marker", b.ID, db.Gen) |
| + return false |
| + } |
| + } |
| + o.req = nil |
| + // the marker names the new generation: the predecessor of a fresh |
| + // generation is dead from here and its retirement is queued |
| + // (claim rev 6.10 QueueGenerationRetire, R10-2); an operation's |
| + // prepared generation leaves its predecessor to the operation |
| + if o.Mode == "cold" && !o.Prepared { |
| + // the predecessor, the empty generations of a chain of |
| + // unfinished admissions, and any generation of this |
| + // operation's writers journal that still holds objects: an |
| + // earlier writer of the operation that died after its marker |
| + // landed but before it queued the chore left its own |
| + // predecessor behind (seed 899 / 30,000) |
| + dead := w.uncovered(b, cur, loc, o.Gen, append([]string(nil), o.Dead...)) |
| + if !w.queueGenerationRetire(b, cur, o.Session, o.Gen, b.Region, dead) { |
| + return false // the claim moved: next time |
| + } |
| + } |
| + o.stage = "open" |
| + case "open": |
| + if !hasMain(b) && b.DB != nil { |
| + // the app was stopped idle while the start was still |
| + // finishing its PUTs (seed 2663 / 30,000): the agent's |
| + // replicator ships what the app wrote, then closes cleanly |
| + b.open = nil |
| + w.flush(b) |
| + b.DB.Closing = true |
| + w.closeStep(b) |
| + w.stats["replication opened"]++ |
| + return false |
| + } |
| + b.open = nil |
| + w.stats["replication opened"]++ |
| + return true |
| } |
| - b.DB = nil |
| - b.open = o |
| - if !w.resumeOpen(b, c, o) { |
| + } |
| +} |
| + |
| +// queueGenerationRetire is claim rev 6.10's QueueGenerationRetire |
| +// (observedSession, publishedGeneration, region, retireGeneration, newID): |
| +// once generation.json names holder.generation, the admitted writer queues |
| +// the park-generation chore for the generation it replaced, owned by the |
| +// region, under the session's op id "session:<holder.session>". The |
| +// mutation of baseline-live queues it at the admitting CAS instead, as |
| +// rev 7.10 did, before the new generation has a root. |
| +func (w *world) queueGenerationRetire(b *box, c *claim, session int, published string, region int, dead []string) bool { |
| + return w.cas(c, func(n *claim) bool { |
| + if n.Holder == nil || n.Holder.Session != session || n.Holder.Gen != published { |
| return false |
| } |
| + id := c.Op.ID |
| + if n.Op.Kind == "serving" || id == "" { |
| + id = fmt.Sprintf("session:%d", session) |
| + } |
| + for i := range n.Op.Writers { |
| + if e := &n.Op.Writers[i]; e.Mode == "cold" && e.Session <= session { |
| + e.Retired = true // its predecessor is parked or in this dead set |
| + } |
| + } |
| + next: |
| + for _, retire := range dead { |
| + if retire == published { |
| + continue |
| + } |
| + ret := retire |
| + if ret == "" { |
| + ret = "@root" // the sentinel for the origin generation (claim rev 6.10) |
| + } |
| + for _, o := range n.Obl { |
| + if o.Kind == "park-generation" && o.Region == region && o.Ret == ret && o.Prefix == n.Holder.Location { |
| + continue next // already queued |
| + } |
| + } |
| + n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: id, Kind: "park-generation", Region: region, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}) |
| + } |
| + return true |
| + }) |
| +} |
| + |
| +// uncovered adds to dead the predecessor of every cold admission of the |
| +// operation's writers journal whose chores were never queued (its writer |
| +// died after its marker landed, before the chore CAS) and that still |
| +// holds objects (seeds 899 and 98677 / 100,000). Every entry is marked |
| +// covered by the CAS that queues them. |
| +func (w *world) uncovered(b *box, c *claim, loc, published string, dead []string) []string { |
| + for i := range c.Op.Writers { |
| + e := &c.Op.Writers[i] |
| + if e.Mode == "cold" && !e.Retired && e.Gen != published && e.Pred.Gen != published && !hasStr(dead, e.Pred.Gen) && w.hasFiles(b, loc, e.Pred.Gen) { |
| + dead = append(dead, e.Pred.Gen) |
| + } |
| + } |
| + return dead |
| +} |
| + |
| +// hasFiles: the generation holds at least one object (an LTX file or a |
| +// certificate) in b's region; for the origin generation "", the objects |
| +// directly under the location, not those of its g-<id>/ generations. |
| +func (w *world) hasFiles(b *box, loc, gen string) bool { |
| + for _, k := range w.buckets[b.Region].keys(dataKey(loc, gen, "")) { |
| + if genOf(loc, k) == gen { |
| + return true |
| + } |
| + } |
| + return false |
| +} |
| + |
| +func hasStr(xs []string, s string) bool { |
| + for _, x := range xs { |
| + if x == s { |
| + return true |
| + } |
| } |
| - return w.startProc(b, c, want, probe, adm.Mode) |
| + return false |
| } |
| -// resumeStart is the agent's own loop finishing an admitted writer's start |
| -// (rev 7.10 §7.14): the activator process holds the admission and waits |
| -// for the PUT it needs, without a wake. An admission the claim moved past |
| -// is abandoned. |
| +// restoreAsNext is what the next admission on e would restore (the oracle's |
| +// baseline, rev 7.11): holder.generation when the marker names it, or when |
| +// it holds objects (the admitted writer republishes its marker first, from |
| +// the operation's build journal for a prepared generation); behind an |
| +// explained lagging marker, the nearest generation of the chain that holds |
| +// objects, else the marker's. |
| +func (w *world) restoreAsNext(e *box, c *claim) *localDB { |
| + h := c.Holder |
| + loc := h.Location |
| + bg, _ := w.bucketGen(e, loc) |
| + if bg == h.Gen { |
| + return w.restoreCalc(e.Region, h.Inc, loc, h.Gen) |
| + } |
| + if w.hasFiles(e, loc, h.Gen) { |
| + ver, key, tx := 0, "", 0 |
| + if h.Gen == installGen(&c.Op) && c.Op.BaseKey != "" { |
| + ver, key, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX |
| + } |
| + return w.restoreWith(e.Region, h.Inc, loc, h.Gen, key, ver, tx) |
| + } |
| + chain, ok := w.unfinishedChain(c, bg) |
| + if !ok { |
| + return w.restoreCalc(e.Region, h.Inc, loc, h.Gen) // unexplained: the restore refuses, as the start does |
| + } |
| + for _, g := range chain { |
| + if g != h.Gen && w.hasFiles(e, loc, g) { |
| + return w.restoreWith(e.Region, h.Inc, loc, g, "", 0, 0) // a fresh generation rooted at [1, 1] |
| + } |
| + } |
| + return w.restoreCalc(e.Region, h.Inc, loc, bg) |
| +} |
| + |
| +// unfinishedChain explains a marker that lags holder.generation (claim rev |
| +// 6.10, R10-2): the writers journal must show holder.generation to be the |
| +// latest admission's, and every generation between it and the one the |
| +// marker names (bg) to be an earlier cold admission's whose predecessor is |
| +// the next; each died before its marker (seed 28216 / 30,000: two in a |
| +// row, the second's purge failed after its CAS). The chain is returned |
| +// newest first, holder.generation included, bg excluded; ok is false when |
| +// the disagreement is unexplained. |
| +func (w *world) unfinishedChain(c *claim, bg string) ([]string, bool) { |
| + h := c.Holder |
| + if bg == h.Gen { |
| + return nil, true |
| + } |
| + gen := h.Gen |
| + var chain []string |
| + for i := len(c.Op.Writers) - 1; i >= 0; i-- { |
| + e := &c.Op.Writers[i] |
| + if e.Gen != gen { |
| + return nil, false // the journal does not explain holder.generation |
| + } |
| + if e.Mode != "cold" { |
| + continue // a warm session of the same generation |
| + } |
| + chain = append(chain, gen) |
| + if e.Pred.Gen == bg { |
| + return chain, true |
| + } |
| + gen = e.Pred.Gen |
| + } |
| + return nil, false |
| +} |
| + |
| +// resumeStart is the agent's own loop finishing an admitted session's |
| +// start (rev 7.11 §7.14): the activator process holds the admission and |
| +// advances the stages without a wake, within the retry policy's backoff. |
| +// An admission the claim moved past is abandoned. |
| func (w *world) resumeStart(b *box, c *claim) { |
| o := b.open |
| - if o == nil || c == nil || c.Holder == nil || hasMain(b) { |
| + if o == nil || c == nil || c.Holder == nil { |
| return |
| } |
| if o.Gen != c.Holder.Gen || o.Session != c.Holder.Session { |
| @@ -642,9 +1024,10 @@ func (w *world) resumeStart(b *box, c *claim) { |
| w.stats["open: abandoned"]++ |
| return |
| } |
| - if w.resumeOpen(b, c, o) { |
| - w.startProc(b, c, o.Want, false, o.Mode) |
| + if w.now < b.nextStart || b.hostFailed { |
| + return // the retry policy's backoff, or a failed host |
| } |
| + w.resumeOpen(b, c, o) |
| } |
| // startProc starts the host's processes on b for the admitted writer. |
| @@ -834,9 +1217,16 @@ type admitInput struct { |
| ObservedGeneration string |
| Box int |
| Mode string // warm or cold; "unguarded" is the rev 7.7 mutation |
| - NewGeneration string |
| + NewGeneration string // cold only; "" for warm (claim rev 6.10) |
| PredecessorCertified bool |
| RemoteMaxTXID int |
| + // KeepStamp (claim rev 6.9.1): a cold admission keeps holder.stamp only |
| + // when no writer was admitted since it was issued (stamp > session) |
| + KeepStamp bool |
| + // BaselineCoversPredecessor (claim rev 6.9.1): the baseline is a seal |
| + // of the stopped predecessor, so nothing is lost and no writes_lost is |
| + // recorded; the journal entry says baseline: sealed |
| + BaselineCoversPredecessor bool |
| } |
| // admitWriter decides the admission for a start on b under the claim it |
| @@ -856,10 +1246,11 @@ func (w *world) admitWriter(b *box, c *claim, want string, probe bool) (admitInp |
| return a, false |
| } |
| switch c.Op.Kind { |
| - case "deleting", "parking", "purging", "creating": |
| + case "creating", "deleting", "parking", "purging", "placing": |
| w.stats["admission refused: op kind"]++ |
| return a, false |
| } |
| + a.KeepStamp = h.Stamp > h.Session |
| if !on("admit-everywhere") && c.Op.Kind != "serving" { |
| a.Mode, a.NewGeneration = "unguarded", want // the mutation: writer sessions for ordinary wakes only (rev 7.8) |
| if want == "" { |
| @@ -876,115 +1267,57 @@ func (w *world) admitWriter(b *box, c *claim, want string, probe bool) (admitInp |
| // it): never re-entered, since a dead writer's uploads may still |
| // land in it (seeds 43 and 87 / 300) |
| if a.PredecessorCertified || !on("recovery-generation") { |
| - a.Mode, a.NewGeneration = "warm", h.Gen |
| + a.Mode, a.NewGeneration = "warm", "" // warm: NewGeneration empty (claim rev 6.10) |
| return a, true |
| } |
| a.Mode, a.NewGeneration = "cold", w.newID("rec") |
| return a, true |
| } |
| - a.Mode, a.NewGeneration = "cold", want // the operation's prepared generation, opened for the first time |
| + // the operation's prepared generation, opened for the first time; built |
| + // from a seal of the stopped predecessor (a landing from the source's |
| + // seal, a dirty return from the target's, a push from its own), so the |
| + // baseline covers it (claim rev 6.9.1) when the seal was built from the |
| + // stopped database itself; a seal built from the remote because the |
| + // disk was gone stops at the remote's position, and the predecessor's |
| + // acknowledged writes beyond it are declared (rev 7.11; seed 34260 / |
| + // 100,000) |
| + covers := c.Op.SealFromDB |
| + if c.Op.Kind == "restoring" { |
| + covers = c.Op.ReturnFromDB |
| + } |
| + a.Mode, a.NewGeneration, a.BaselineCoversPredecessor = "cold", want, covers |
| return a, true |
| } |
| -// resumeOpen advances an admitted writer's start (rev 7.10 §7.14): each |
| -// stage issues one PUT and waits for its acknowledgement, retrying the |
| -// same bytes after the client's timeout; a stage whose effect is visible |
| -// (the file current, the marker naming the generation) is done whatever |
| -// became of the request. It returns true once the process may start, with |
| -// the database installed on the box. |
| -func (w *world) resumeOpen(b *box, c *claim, o *openState) bool { |
| - h := c.Holder |
| - loc := h.Location |
| - bk := w.buckets[b.Region] |
| - db := o.DB |
| - for { |
| - var key string |
| - var mk func() *request |
| - visible := func() bool { return false } |
| - switch o.stage { |
| - case "first": |
| - key = ltxKey(loc, db.Gen, 0, 1, 1) |
| - mk = func() *request { return &request{kind: rPut, region: b.Region, key: key, b: w.deltaOf(db, h.Inc)} } |
| - visible = func() bool { return bk.current(key) != nil } |
| - case "marker": |
| - bg, _ := w.bucketGen(b, loc) |
| - visible = func() bool { bg, _ = w.bucketGen(b, loc); return bg == db.Gen } |
| - if bg != db.Gen { |
| - ver, k, tx := 0, "", 0 |
| - if o.Prepared { |
| - ver, k, tx = c.Op.BaseVer, c.Op.BaseKey, c.Op.BaseTX |
| - } |
| - if o.req == nil || o.req.done || w.now-o.req.at >= uploadTimeout { |
| - if o.req != nil && !o.req.done { |
| - w.stats["open: marker PUT retried"]++ |
| - } |
| - v := w.publishGen(b, loc, db.Gen, h.Inc, ver, k, tx) |
| - o.req = w.lastIssued |
| - if v > 0 { |
| - o.req = nil |
| - } |
| - } |
| - if !visible() { |
| - w.stats["open: waits for the marker"]++ |
| - w.logf("box %d: admitted into %q, waiting for its marker", b.ID, db.Gen) |
| - return false |
| - } |
| - } |
| - o.stage, o.req = "start", nil |
| - continue |
| - case "anchor": |
| - key = ltxKey(loc, db.Gen, 0, db.Pos, db.Pos) |
| - mk = func() *request { return &request{kind: rPut, region: b.Region, key: key, b: anchorOf(db, h.Inc)} } |
| - visible = func() bool { return bk.current(key) != nil } |
| - case "start": |
| - if o.stage == "start" { |
| - b.DB = db |
| - b.open = nil |
| - if o.Mode == "cold" && !o.Prepared && (len(db.Log) > 0 || len(db.State) > 0) { |
| - db.Pos, db.FirstFull, db.Pending = 1, false, nil |
| - } |
| - return true |
| - } |
| - } |
| - // first and anchor: one PUT, acknowledged before the next stage |
| - if !visible() { |
| - if o.req == nil || (!o.req.done && w.now-o.req.at >= uploadTimeout) { |
| - if o.req != nil { |
| - w.stats["open: PUT retried"]++ |
| - } |
| - r := mk() |
| - if o.stage == "anchor" { |
| - w.stats["restore anchors"]++ |
| - } else { |
| - w.stats["first uploads"]++ |
| - } |
| - v := w.issue(b.ID, r) |
| - o.req = r |
| - if v > 0 { |
| - o.req = nil |
| - } |
| - } |
| - if !visible() { |
| - if o.stage == "anchor" && !on("anchor-acked") { |
| - // the mutation opens before the anchor is acknowledged |
| - } else { |
| - w.stats["open: waits for its "+o.stage]++ |
| - return false |
| - } |
| - } |
| - } |
| - if o.stage == "first" { |
| - w.markDurable(key, bk.current(key).B) |
| - o.stage = "marker" |
| - } else { |
| - o.stage = "start" |
| +// lastWriter is the newest writers journal entry, or nil. |
| +func lastWriter(op *opState) *writerEntry { |
| + if len(op.Writers) == 0 { |
| + return nil |
| + } |
| + return &op.Writers[len(op.Writers)-1] |
| +} |
| + |
| +// durableInLineage: W is durable in gen or one of its ancestors within loc. |
| +func (w *world) durableInLineage(loc, gen string, x int) bool { |
| + seen := map[string]bool{} |
| + for p := dataKey(loc, gen, ""); p != "" && !seen[p]; p = w.genParent[p] { |
| + seen[p] = true |
| + if w.durable[p][x] { |
| + return true |
| } |
| - o.req = nil |
| } |
| + return false |
| +} |
| + |
| +// alert raises an operator alert (counted; the oracle checks every |
| +// writes_lost record raised one, R10-6). |
| +func (w *world) alert(format string, args ...any) { |
| + w.alerts++ |
| + w.logf("alert: "+format, args...) |
| } |
| -// admitCommit is the AdmitWriter CAS (claim rev 6.9 §6.4a), made after |
| -// readiness and before any upload. It is bound to the observed op.id and |
| +// admitCommit is the AdmitWriter CAS (claim rev 6.10 §6.4a), made before |
| +// the purge, the restore and the app start. It is bound to the observed op.id and |
| // phase (R9-4; rule admit-bind-op), the observed session and generation, |
| // and op.exec. It writes holder.session, holder.writer_box and the journal |
| // entry; in cold mode also holder.generation, holder.stamp (the installing |
| @@ -1009,7 +1342,7 @@ func (w *world) admitCommit(b *box, c *claim, a admitInput) bool { |
| return false |
| } |
| switch n.Op.Kind { |
| - case "deleting", "parking", "purging", "creating": |
| + case "creating", "deleting", "parking", "purging", "placing": |
| return false |
| } |
| } |
| @@ -1018,35 +1351,42 @@ func (w *world) admitCommit(b *box, c *claim, a admitInput) bool { |
| } |
| n.Holder.Session = w.nextGen() |
| n.Holder.WriterBox = b.ID |
| - e := writerEntry{Session: n.Holder.Session, Box: b.ID, Gen: a.NewGeneration, Stamp: n.Holder.Stamp, Mode: a.Mode, |
| + gen := a.NewGeneration |
| + if a.Mode != "cold" { |
| + gen = n.Holder.Gen |
| + } |
| + e := writerEntry{Session: n.Holder.Session, Box: b.ID, Gen: gen, Stamp: n.Holder.Stamp, Mode: a.Mode, Baseline: "remote", |
| Pred: pred{Session: a.ObservedSession, Box: h.WriterBox, Gen: h.Gen, Stamp: h.Stamp}} |
| if a.Mode == "cold" { |
| w.newGen(n.Holder.Location, h.Gen, a.NewGeneration) |
| n.Holder.Gen = a.NewGeneration |
| - if !(n.Op.Kind == "landing" && a.NewGeneration == installGen(&n.Op) && n.Holder.Stamp == n.Op.FlipGen) { |
| + if !a.KeepStamp { |
| // a new installation: a new stamp (the installing CAS, claim |
| - // rev 6.9 §6.4a, §6.4): a fresh generation, a push's prepared |
| - // generation, a dirty rollback's return generation (claim rev |
| - // 6.5: never the predecessor's stamp). The landing's prepared |
| - // generation is the installation the flip CAS already made |
| - // current with its stamp F, which the target's record |
| - // activated before the probe; another stamp here would refuse |
| - // the probe by its own record (seed 1117 / 30,000) |
| + // rev 6.9 §6.4a, §6.4). KeepStamp (claim rev 6.9.1): the stamp |
| + // stays only when no writer was admitted since it was issued |
| + // (holder.stamp > holder.session): a landing's first probe |
| + // after the flip, whose record the target activated before the |
| + // probe; a restarted probe, a push and a dirty return take a |
| + // new one |
| n.Holder.Stamp = w.issueStamp(n, b.ID) |
| } |
| e.Stamp = n.Holder.Stamp |
| - if !a.PredecessorCertified && on("writes-lost") { |
| + if a.BaselineCoversPredecessor { |
| + e.Baseline = "sealed" // nothing to declare: the seal holds every acknowledged write (claim rev 6.9.1) |
| + } else if !a.PredecessorCertified && on("writes-lost") { |
| lost = &writesLost{Box: h.WriterBox, Gen: h.Gen, Session: h.Session, RemoteMaxTXID: a.RemoteMaxTXID} |
| e.Lost = lost |
| } |
| - if a.NewGeneration != installGen(&n.Op) || n.Op.Kind == "serving" { |
| - // a recovery into a fresh generation: the predecessor is |
| - // dead and parked as a chore after the dead writer's horizon; |
| - // an operation's prepared generation leaves its predecessor |
| - // to the operation's own finalization or rollback |
| - // in this box's region: the predecessor's files are where this |
| - // box restored them from |
| - n.Obl = append(n.Obl, w.newObl(n, "park-generation", b.Region, -1, n.Holder.Location, h.Gen, w.now+horizonH)) |
| + // no retire chore here: the predecessor is the baseline of an |
| + // unfinished admission until generation.json names the new |
| + // generation; the chore is queued then (claim rev 6.10, R10-2). |
| + // The mutation of baseline-live queues it now, as rev 7.10 did |
| + if !on("baseline-live") && !a.BaselineCoversPredecessor { |
| + ret := h.Gen |
| + if ret == "" { |
| + ret = "@root" |
| + } |
| + n.Obl = append(n.Obl, obligation{ID: w.newID("ob"), Op: fmt.Sprintf("session:%d", n.Holder.Session), Kind: "park-generation", Region: b.Region, Box: -1, Prefix: n.Holder.Location, Ret: ret, Quiet: w.now + horizonH}) |
| } |
| } |
| n.Op.Writers = append(n.Op.Writers, e) |
| @@ -1059,9 +1399,9 @@ func (w *world) admitCommit(b *box, c *claim, a admitInput) bool { |
| } |
| w.stats["admissions: "+a.Mode]++ |
| switch w.claim.Op.Kind { |
| - case "deleting", "parking", "purging", "creating": |
| + case "creating", "deleting", "parking", "purging", "placing": |
| // INV-13 (R9-4): no writer is admitted while the host is being |
| - // deleted, parked, purged or created |
| + // created, deleted, parked, purged or placed |
| w.violate("INV-13: box %d was admitted as a writer while the host is %s (%s)", b.ID, w.claim.Op.Kind, w.claim.Op.ID) |
| } |
| if a.Mode == "cold" { |
| @@ -1073,14 +1413,25 @@ func (w *world) admitCommit(b *box, c *claim, a admitInput) bool { |
| n := 0 |
| for i := range w.acks { |
| x := &w.acks[i] |
| - if x.Box == lost.Box && x.Inc == h.Inc && x.Kind == "write" && !x.Exempt && !w.isDurable(x.W) { |
| + if x.Box == lost.Box && x.Inc == h.Inc && x.Kind == "write" && !x.Exempt && !w.durableInLineage(h.Location, a.NewGeneration, x.W) { |
| + declaredLost := contains(w.claim.Lost, lost.Box) || w.claim.Op.LostAt != 0 && w.claim.Op.LostBox == lost.Box |
| + if w.isDurable(x.W) && !declaredLost { |
| + // INV-15 (rev 7.11): a write the host's remote holds is never |
| + // declared lost while its writer's box is still with us; the |
| + // lineage must adopt the generation that holds it (a dirty |
| + // rollback's return lane), not declare it. A box declared |
| + // lost may leave such writes behind: the record's |
| + // remote_max_txid names where an operator finds them |
| + w.violate("INV-15: box %d declared write %d lost, which is durable in the remote outside the new lineage (box %d's session %d)", b.ID, x.W, lost.Box, lost.Session) |
| + } |
| x.Exempt = true |
| n++ |
| } |
| } |
| w.stats["writes_lost declared"]++ |
| w.stats["writes declared lost"] += n |
| - w.logf("box %d: alert writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes", b.ID, lost.Box, lost.Gen, lost.Session, lost.RemoteMaxTXID, n) |
| + w.lostAlert++ |
| + w.alert("box %d: writes_lost {box %d, generation %q, session %d, remote_max_txid %d}: %d acknowledged writes", b.ID, lost.Box, lost.Gen, lost.Session, lost.RemoteMaxTXID, n) |
| } |
| return true |
| } |
| @@ -1288,7 +1639,11 @@ func (w *world) doObligation(b *box, c *claim, o obligation) { |
| if o.Kind == "park-generation" { |
| // a dead generation within the live location (a cold recovery's |
| // old generation, claim rev 6.6 §6.4a): only its own keys |
| - keep = func(k string) bool { return genOf(o.Prefix, k) != o.Ret } |
| + retire := o.Ret |
| + if retire == "@root" { |
| + retire = "" // the origin generation's sentinel (claim rev 6.10) |
| + } |
| + keep = func(k string) bool { return genOf(o.Prefix, k) != retire } |
| } |
| if !w.parkStep(b, o.Prefix, o.Dest(), quiet, keep) { |
| return |
diff --git a/internal/sim/hostlife/rules_test.go b/internal/sim/hostlife/rules_test.go| index 234d091d3..6db65b688 100644 |
| --- a/internal/sim/hostlife/rules_test.go |
| +++ b/internal/sim/hostlife/rules_test.go |
| @@ -96,6 +96,8 @@ var allRules = []string{ |
| "exec-writer", // claim rev 6.8.1 (D1): the AdmitWriter CAS requires op.exec to be this box, in every operation; the mutation admits any box the phase table reaches |
| "remote-baseline", // claim rev 6.9 §6.4a (R9-1): a box never reuses its local database; every admitted writer purges it and restores from the certified remote baseline; the mutation keeps a local database of the generation it opens, as today's activator does |
| "writes-lost", // claim rev 6.9 §6.4a (R9-2): a cold admission whose observed session had no certificate records writes_lost {box: holder.writer_box, generation, session, remote_max_txid} and alerts; the mutation loses the writes silently |
| + "admit-before-start", // claim rev 6.10 §6.4a (R10-1): the admitting CAS comes before the app starts, so nothing the app acknowledges is outside a session; the mutation admits after readiness, as rev 7.10 did, and a background acknowledgement before the CAS is lost silently |
| + "baseline-live", // claim rev 6.10 (R10-2): an unfinished admission's predecessor is live until generation.json names the new generation; the retire chore is queued then; the mutation queues it at the CAS and sweeps the predecessor as dead |
| "admit-bind-op", // claim rev 6.9 §6.4a (R9-4): the AdmitWriter CAS is bound to the observed op.id and phase, serving's "" compared literally; the mutation compares session and generation alone |
| "valid-ltx", // rev 7.9 §5.1 step 9 (R8-3): an empty generation gets no LTX file, never a zero-TXID snapshot or anchor; the mutation writes [1,0] and [0,0] |
| "close-evidence", // claim rev 6.8 §6.4a (R8-5): a certificate needs every request of the writer's lifetime returned with success and no retry after a timeout; the mutation looks at the level-0 queue alone |
| @@ -116,6 +118,7 @@ var allRules = []string{ |
| "revert-journal", // J-N11 |
| "rollback-journal", // R2-3 |
| "dirty-return", // INV-2 |
| + "seal-return", // rev 7.11 §7.8: a rollback whose source replica stops short of the stopped database returns the seal |
| "lost-return-park", // rev 7 §7.10 (model 27): declare-lost parks a return lane the lost target had begun (rev 7.2: a biased schedule stops the target mid-return) |
| "target-park", // R2-5/INV-10 |
| "abort-target-owner", // J-N13 |
diff --git a/internal/sim/hostlife/sim_test.go b/internal/sim/hostlife/sim_test.go| index f88f63bbc..1e8f88640 100644 |
| --- a/internal/sim/hostlife/sim_test.go |
| +++ b/internal/sim/hostlife/sim_test.go |
| @@ -1,6 +1,7 @@ |
| package hostlife |
| import ( |
| + "encoding/json" |
| "fmt" |
| "math/rand" |
| "os" |
| @@ -17,7 +18,7 @@ const ( |
| func newWorld(seed int64, verbose bool) *world { |
| w := &world{seed: seed, rng: &counting{r: rand.New(rand.NewSource(seed))}, verbose: verbose, |
| stores: map[string]bool{app0: true, app1: true}, cfgVer: map[string]int{app0: 1, app1: 1}, purgeRes: map[string]bool{}, purgeStep: map[string]int{}, |
| - control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}} |
| + control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, markerAlerted: map[string]bool{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}} |
| for r := 0; r < nRegions; r++ { |
| w.buckets[r] = newBucket() |
| } |
| @@ -84,7 +85,7 @@ func (w *world) legacyStart() { |
| w.legacyIncs[inc] = true |
| key := 7 |
| b.Row = &row{Inc: inc, App: app0, Status: "ready"} |
| - b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}} |
| + b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}, Epoch: b.Epoch} |
| for i := 0; i < 3; i++ { |
| w.nextW++ |
| w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: 0, Kind: "write"}) |
| @@ -112,7 +113,7 @@ func (w *world) legacyStart() { |
| // that earlier move also left the old box's local state behind: |
| // a database, a files directory and env files, and no row |
| ob := w.regionBox(other) |
| - ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Pos: 1} |
| + ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Pos: 1, Epoch: ob.Epoch} |
| ob.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true} |
| } |
| } |
| @@ -212,6 +213,10 @@ func (w *world) heal(final bool) { |
| continue |
| } |
| b.Partitioned, b.Frozen, b.AgentPaused, b.AgentDown, b.PurgeFails = false, false, false, false, false |
| + if b.hostFailed { |
| + b.hostFailed, b.failCounted, b.startFails, b.nextStart = false, false, 0, 0 // the operator cleared the failed host |
| + w.stats["failed hosts cleared"]++ |
| + } |
| } |
| if final { |
| for _, b := range w.boxes { |
| @@ -398,6 +403,16 @@ func (w *world) residueOf(exceptStragglers bool) []string { |
| live[dataKey(loc, gen, "")] = true // the generation's prefix: every LTX file of the generation |
| live[genMarkerKey(loc)] = true |
| live[dataKey(loc, gen, "closed/")] = true // the live generation's certificates, every session's (claim rev 6.8 §6.4a) |
| + // an unfinished admission's predecessor is its baseline until the |
| + // marker names the new generation (claim rev 6.10, R10-2) |
| + if bg, _ := w.bucketGen(e, loc); bg != gen { |
| + if chain, ok := w.unfinishedChain(c, bg); ok { |
| + for _, g := range append(chain, bg) { |
| + live[dataKey(loc, g, "")] = true |
| + live[dataKey(loc, g, "closed/")] = true |
| + } |
| + } |
| + } |
| } |
| } |
| for r := 0; r < nRegions; r++ { |
| @@ -424,14 +439,28 @@ func (w *world) residueOf(exceptStragglers bool) []string { |
| // owed: an open park obligation names this region and a prefix of k. |
| func (w *world) owed(r int, k string) bool { |
| - if w.claim == nil { |
| + c := w.claim |
| + if c == nil { |
| return false |
| } |
| - for _, o := range w.claim.Obl { |
| + for _, o := range c.Obl { |
| if o.Region == r && o.Prefix != "" && strings.HasPrefix(k, o.Prefix) { |
| return true |
| } |
| } |
| + // a cold admission whose writer died after its marker landed and before |
| + // it queued the retire chores (rev 7.11): the writers journal still |
| + // shows its predecessor uncovered, and the next admission's marker |
| + // retires it (§5.1 step 9); until then the predecessor is owed, as a |
| + // pending chore's keys are (seeds 4159, 26817, 34163 / 100,000) |
| + if h := c.Holder; h != nil && strings.HasPrefix(k, h.Location) { |
| + g := genOf(h.Location, k) |
| + for i := range c.Op.Writers { |
| + if e := &c.Op.Writers[i]; e.Mode == "cold" && !e.Retired && e.Pred.Gen == g { |
| + return true |
| + } |
| + } |
| + } |
| return false |
| } |
| @@ -521,6 +550,19 @@ func (w *world) final() { |
| local := e.DB |
| if local == nil { |
| local = w.restoreCalc(e.Region, c.Holder.Inc, c.Holder.Location, c.Holder.Gen) |
| + // the next admission would be cold if the session is uncertified, |
| + // and would declare the writer box's acknowledged, unuploaded |
| + // writes lost (R10-6: the faults exempt nothing themselves); the |
| + // oracle expects what that admission would serve |
| + if w.certificate(e, c) == nil { |
| + for i := range w.acks { |
| + x := &w.acks[i] |
| + if x.Box == c.Holder.WriterBox && x.Inc == c.Holder.Inc && x.Kind == "write" && !x.Exempt && !w.isDurable(x.W) { |
| + x.Exempt = true |
| + w.stats["final: writes the next admission would declare lost"]++ |
| + } |
| + } |
| + } |
| } |
| if local == nil { |
| w.violate("final: the serving box %d has no database and nothing to restore", e.ID) |
| @@ -528,6 +570,9 @@ func (w *world) final() { |
| } |
| w.checkLocal(c, e, local) |
| w.checkDurable(c, "final") |
| + if w.lostAlert != w.stats["writes_lost declared"] { |
| + w.violate("INV-14: %d writes_lost records, %d alerts delivered (R10-6)", w.stats["writes_lost declared"], w.lostAlert) |
| + } |
| } |
| // checkDurable is the cold-restore promise, checked at every release to |
| @@ -546,7 +591,7 @@ func (w *world) checkDurable(c *claim, when string) { |
| } |
| before := w.fingerprint() |
| defer w.pure("INV-2 oracle", before) |
| - db := w.restoreCalc(e.Region, c.Holder.Inc, c.Holder.Location, c.Holder.Gen) |
| + db := w.restoreAsNext(e, c) |
| need := map[int]bool{} |
| seen := map[string]bool{} |
| for p := dataKey(c.Holder.Location, c.Holder.Gen, ""); p != "" && !seen[p]; p = w.genParent[p] { |
| @@ -870,6 +915,9 @@ func TestHostLifecycle(t *testing.T) { |
| } |
| t.Logf("%d seeds, %d failed, %d completed operations; mutated=%q", n, failed, ops, mutated) |
| t.Logf("outcomes: %v", stats) |
| + if js, err := json.Marshal(stats); err == nil { |
| + t.Logf("COUNTERS %s", js) // structured, for the spec's generator (R10-9) |
| + } |
| for k, c := range kinds { |
| t.Logf(" %4d × %s (first seed %d)", c, k, firstSeed[k]) |
| } |
diff --git a/internal/sim/hostlife/witness_test.go b/internal/sim/hostlife/witness_test.go| index c6d558ee5..0673908ec 100644 |
| --- a/internal/sim/hostlife/witness_test.go |
| +++ b/internal/sim/hostlife/witness_test.go |
| @@ -65,8 +65,8 @@ var witnesses = []witness{ |
| "a staging round completes after the landing's staging park acknowledged; retired at the seal rather than after the writers' horizon, the round is residue at the release (INV-10)", wStagingQuiet}, |
| {"claim-cas-opid", "claim-cas-opid,revalidate-before-effect", "store of app", |
| "redundant alone: a stalled source goroutine with the rollback's last read resumes after a purge claimed the host; revalidation stops it, and so does the op-id condition on its release CAS; with both gone the release overwrites the purge and the store is torn down under a serving host", wClaimCasOpID}, |
| - {"dirty-return", "dirty-return", "INV-2 (local)", |
| - "a target that wrote during its probe is rolled back; without the return its writes are lost (INV-2)", wDirtyReturn}, |
| + {"dirty-return", "dirty-return", "INV-15", |
| + "the target writes during its probe and replicates; the rollback is dirty and the return lane carries the write to the source; treated as clean, the source's admission declares a write lost that the host's remote holds and whose writer was never lost (INV-15, rev 7.11; rev 7.10 saw it as INV-2 on the serving database)", wDirtyReturn}, |
| {"commit-before-open", "commit-before-open", "INV-5", |
| "a public request reaches the target's edge while its probe runs; a row opened at the probe routes it before the commit (INV-5, R3-5)", wCommitBeforeOpen}, |
| {"exec-admission", "exec-admission,hold-op-open,start-check-claim-gen,exec-writer", "INV-1", |
| @@ -146,14 +146,25 @@ var witnesses = []witness{ |
| "a writer acknowledges a write and crashes before uploading it; the next wake is cold and restores the remote, which lacks the write: the admission must record writes_lost {box, generation, session, remote_max_txid} and alert; silently, the serving database differs from the acknowledged writes (claim rev 6.9, R9-2)", wSilentLoss}, |
| {"admit-bind-op", "admit-bind-op,claim-cas-opid", "INV-13", |
| "a wake's admission closure is computed under serving (op.id \"\"); a delete begins, preserving session, generation and executor; the stale closure resumes: bound to the observed op.id, phase and kind it is refused; the model's general CAS condition on op.id (C2, claim-cas-opid) refuses it too, so the two must go together for a writer to be admitted into a host being deleted (claim rev 6.9, R9-4)", wStaleClosureDelete}, |
| + // rev 7.11 |
| + {"admit-before-start", "admit-before-start", "INV-2", |
| + "a background worker acknowledges a write right after the app starts, and the box crashes before replication opens; admitted before the start, the acknowledgement is the session's and the next cold admission declares it; admitted after readiness (rev 7.10's order), it is in a database the next warm admission purges silently (claim rev 6.10, R10-1)", wAckBeforeAdmission}, |
| + {"baseline-live", "baseline-live", "INV-2", |
| + "a cold admission into a fresh generation; the box crashes before the first upload, so the generation is empty and the marker names the predecessor; the predecessor is the unfinished admission's baseline and stays live until the marker names the new generation; with the retire chore queued at the CAS, the predecessor is parked and the recovery restores nothing (claim rev 6.10, R10-2)", wBaselineParked}, |
| + {"seal-return", "seal-return", "INV-15", |
| + "the source acknowledges a write whose upload is still in flight when the cut stops it: the seal carries the write, the source's replica does not; the landing is rolled back before its probe; the return must carry the seal, and treated as clean the source's cold admission declares the write lost while the landing's built generation holds it (rev 7.11 §7.8)", wSealReturn}, |
| + {"close-evidence", "close-evidence", "certificate while", |
| + "a fresh generation's first upload is held past the client's timeout and its retry lands; the session's clean stop must write no certificate (R10-7)", wFirstUploadRetry}, |
| + {"close-evidence", "close-evidence", "certificate while", |
| + "a warm admission's anchor PUT is held past the client's timeout and its retry lands; the session's clean stop must write no certificate (R10-7)", wAnchorRetry}, |
| {"close-evidence", "close-evidence", "certificate while", |
| "a compaction's level-1 PUT is held past the client's timeout; the next compaction pass retries and lands; the original completes afterwards; the session's clean stop must write no certificate, since a request of its lifetime was retried after a timeout; looking at the level-0 queue alone it certifies (R9-5)", wMaintenanceRetry}, |
| {"close-evidence", "close-evidence", "certificate while", |
| "a snapshot PUT completes on the server after the client gave up on it, a lost response; the client never observed success, so the session must stay uncertain and certify nothing; counting server completion it certifies (R9-5)", wLostResponse}, |
| {"close-evidence", "close-evidence", "certificate while", |
| "an upload times out and its retry succeeds while the original stays unresolved; the writer's clean close must write no certificate, because a request of its lifetime is unaccounted for; looking at the level-0 queue alone it certifies (claim rev 6.8 §6.4a, R8-5)", wRetryUnresolved}, |
| - {"build-completion", "build-completion", "before its snapshot", |
| - "a push's build with its snapshot PUT, then its anchor PUT, held, and the agent crashing in between: generation.json is published only once both are acknowledged; published from the intent it names artifacts that have not landed (R7-4)", wBuildCrashes}, |
| + {"build-completion", "build-completion", "INV-2", |
| + "a push's build with its snapshot, then its anchor, held, the agent crashing in between each; since rev 7.11 the marker is the admission's, so a build completed from its intent gives the probe nothing to restore from its pinned snapshot, and the next session restores an empty generation (INV-2); rev 7.10 caught it at the marker (INV-11)", wBuildCrashes}, |
| } |
| // TestWitnesses is the deterministic half of the mutation evidence. |
| @@ -302,7 +313,7 @@ func dump(t *testing.T, w *world) { |
| func newWitness() *world { |
| w := &world{seed: -1, rng: quiet{}, verbose: true, now: 100, |
| stores: map[string]bool{app0: true, app1: true}, cfgVer: map[string]int{app0: 1, app1: 1}, purgeRes: map[string]bool{}, purgeStep: map[string]int{}, |
| - control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}} |
| + control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, markerAlerted: map[string]bool{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}} |
| for r := 0; r < nRegions; r++ { |
| w.buckets[r] = newBucket() |
| b := &box{ID: r, Region: r, HasKey: true} |
| @@ -319,7 +330,7 @@ func newWitness() *world { |
| func (w *world) serve(b *box, inc int, app string) { |
| loc := locationFor(inc, false) |
| key := 1000 + inc |
| - db := &localDB{Inc: inc, Loc: loc, Key: key, State: map[string]int{}} |
| + db := &localDB{Inc: inc, Loc: loc, Key: key, State: map[string]int{}, Epoch: b.Epoch} |
| for i := 0; i < 2; i++ { |
| w.nextW++ |
| w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: w.now, Kind: "write"}) |
| @@ -366,7 +377,7 @@ func (w *world) legacy(b *box, other int) { |
| w.legacyIncs[inc] = true |
| key := 7 |
| b.Row = &row{Inc: inc, App: app0, Status: "ready"} |
| - b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}} |
| + b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}, Epoch: b.Epoch} |
| for i := 0; i < 2; i++ { |
| w.nextW++ |
| w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: w.now, Kind: "write"}) |
| @@ -544,6 +555,9 @@ func (w *world) completeHeld(key string) int { |
| for _, r := range w.pending { |
| if (r.key == key || strings.HasSuffix(key, "/") && strings.HasPrefix(r.key, key)) && r.doneAt == 1<<30 { |
| r.doneAt = w.now // the response arrives now (the close's evidence reads the completion time, R9-5) |
| + if w.now-r.at > horizonH { |
| + r.beyond = true // a completion past the horizon: the sweep's, as the biased schedule's are (§6.4) |
| + } |
| w.complete(r) |
| w.logf("witness: held %v %s r%d completes now", r.kind, r.key, r.region) |
| n++ |
| @@ -1385,12 +1399,7 @@ func (w *world) loseDisk(b *box) { |
| if b.DB == nil { |
| return |
| } |
| - for i := range w.acks { |
| - a := &w.acks[i] |
| - if a.Box == b.ID && a.Inc == b.DB.Inc && !w.isDurable(a.W) { |
| - a.Exempt = true |
| - } |
| - } |
| + // physical loss only (R10-6): the next admission declares the writes |
| b.DB = nil |
| w.logf("witness: box %d lost its database file", b.ID) |
| } |
| @@ -2074,3 +2083,203 @@ func wLostResponse(t *testing.T, w *world) { |
| } |
| w.finish(t) |
| } |
| + |
| +// ---------- rev 7.11 ---------- |
| + |
| +// ackInto acknowledges one write into b's database without a main process: |
| +// a background worker's work (R10-1). |
| +func (w *world) ackInto(b *box, inc int) int { |
| + db := b.DB |
| + w.nextW++ |
| + wr := w.nextW |
| + w.commit(db, wr) |
| + w.acks = append(w.acks, ack{W: wr, Inc: inc, Box: b.ID, At: w.now, Kind: "write"}) |
| + w.writeInc[wr] = inc |
| + w.logf("witness: a background worker on box %d acknowledged %d", b.ID, wr) |
| + return wr |
| +} |
| + |
| +// wAckBeforeAdmission (claim rev 6.10, R10-1): a background acknowledgement |
| +// right after the app starts, then a crash before replication opens. |
| +func wAckBeforeAdmission(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + w.idleStop(b) // a certified close: the next admission is warm |
| + armed := true |
| + w.startHook = func(x *box) bool { |
| + if x != b || !armed || x.DB == nil { |
| + return false |
| + } |
| + armed = false |
| + w.ackInto(x, 1) |
| + w.crash(x) |
| + return true |
| + } |
| + w.wake(b) |
| + w.startHook = nil |
| + w.recovered(t, b, "woken after the crash") |
| + w.checkLocal(w.claim, b, b.DB) |
| + w.finish(t) |
| +} |
| + |
| +// wBaselineParked (claim rev 6.10, R10-2): a crash between the admitting |
| +// CAS and the first upload, then the predecessor's retirement. |
| +func wBaselineParked(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + w.crash(b) // no certificate: the next admission is cold, into a fresh generation |
| + armed := true |
| + w.startHook = func(x *box) bool { |
| + if x != b || !armed { |
| + return false |
| + } |
| + armed = false |
| + w.crash(x) // before the first upload: the new generation stays empty |
| + return true |
| + } |
| + w.wake(b) |
| + w.startHook = nil |
| + if g := w.claim.Holder.Gen; g == "" || len(w.buckets[b.Region].keys(dataKey(w.claim.Holder.Location, g, ""))) != 0 { |
| + t.Fatalf("witness: expected an empty fresh generation, have %q: %s", g, w.state()) |
| + } |
| + // the predecessor's retirement, queued at the CAS under the mutation, |
| + // runs out its quiet time |
| + w.reach(t, 200, "chores settled", func() bool { return len(w.claim.Obl) == 0 }, w.act(b), w.recoveryPass) |
| + w.recovered(t, b, "woken after the crash") |
| + w.checkLocal(w.claim, b, b.DB) |
| + w.checkDurable(w.claim, "after the recovery") |
| + w.finish(t) |
| +} |
| + |
| +// wFirstUploadRetry (R10-7): the fresh generation's first upload retried. |
| +func wFirstUploadRetry(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + w.crash(b) |
| + var original *request |
| + w.hold = func(r *request) bool { |
| + if r.kind == rPut && r.b != nil && r.b.Level == 0 && r.b.Min == 1 && r.b.Max == 1 && r.b.Gen != "" && original == nil { |
| + original = r |
| + return true // the first upload's first attempt alone is held |
| + } |
| + return r == original |
| + } |
| + w.wake(b) |
| + for i := 0; i <= uploadTimeout+1; i++ { |
| + w.tick(w.act(b)) // the client times out; the retry lands and the start completes |
| + } |
| + if original == nil { |
| + t.Fatalf("witness: no first upload was issued: %s", w.state()) |
| + } |
| + if !hasMain(b) { |
| + t.Fatalf("witness: the writer did not start after the retry: %s", w.state()) |
| + } |
| + loc, gen := w.claim.Holder.Location, w.claim.Holder.Gen |
| + w.idleStop(b) |
| + sk := closedKey(loc, gen, b.DB.Session) |
| + for i := 0; i < 10; i++ { |
| + w.tick(func() { w.replicate(b) }) |
| + if w.buckets[b.Region].current(sk) != nil { |
| + w.violate("INV-2 (witness): box %d wrote its certificate while its first upload %s was unresolved (R10-7)", b.ID, original.key) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + w.completeHeld(original.key) |
| + w.finish(t) |
| +} |
| + |
| +// wAnchorRetry (R10-7): a warm admission's anchor retried. |
| +func wAnchorRetry(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + loc := w.claim.Holder.Location |
| + // the generation's newest snapshot stands above its level-0 maximum |
| + // (retention expired the files above it): a warm restore lands at the |
| + // snapshot's position and needs the anchor there |
| + snap := snapshotOf(b.DB, 1) |
| + snap.Max = 2 |
| + w.nextW++ |
| + wr := w.nextW |
| + w.acks = append(w.acks, ack{W: wr, Inc: 1, Box: b.ID, At: w.now, Kind: "write"}) |
| + w.writeInc[wr] = 1 |
| + w.txs[wr] = tx{W: wr, TX: 3, Key: fmt.Sprintf("k%d", wr), Val: wr} |
| + snap.Pages[fmt.Sprintf("k%d", wr)] = wr |
| + snap.TXs = append(snap.TXs, wr) |
| + k := snapKey(loc, "", 2) |
| + w.buckets[b.Region].objs[k] = append(w.buckets[b.Region].objs[k], &version{ID: w.newVer(), B: snap}) |
| + w.markDurable(k, snap) |
| + w.buckets[b.Region].objs[closedKey(loc, "", 1)][0].B.TX = 2 // the certificate names the snapshot's position |
| + var original *request |
| + w.hold = func(r *request) bool { |
| + if r.kind == rPut && r.b != nil && r.b.Level == 0 && r.b.Min == 2 && r.b.Max == 2 && original == nil { |
| + original = r |
| + return true // the anchor's first attempt alone is held |
| + } |
| + return r == original |
| + } |
| + w.wake(b) |
| + for i := 0; i <= uploadTimeout+1; i++ { |
| + w.tick(w.act(b)) |
| + } |
| + if original == nil { |
| + t.Fatalf("witness: no anchor PUT was issued: %s%s", w.state(), w.boxesState()) |
| + } |
| + if !hasMain(b) { |
| + t.Fatalf("witness: the writer did not start after the anchor retry: %s", w.state()) |
| + } |
| + w.idleStop(b) |
| + sk := closedKey(loc, "", b.DB.Session) |
| + for i := 0; i < 10; i++ { |
| + w.tick(func() { w.replicate(b) }) |
| + if w.buckets[b.Region].current(sk) != nil { |
| + w.violate("INV-2 (witness): box %d wrote its certificate while its anchor %s was unresolved (R10-7)", b.ID, original.key) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + w.completeHeld(original.key) |
| + w.finish(t) |
| +} |
| + |
| +// wSealReturn (rev 7.11 §7.8): the source's last upload is in flight at |
| +// the cut; the rollback's return must carry the seal. |
| +func wSealReturn(t *testing.T, w *world) { |
| + src, tgt := w.boxes[0], w.boxes[1] |
| + w.serve(src, 1, app0) |
| + w.wake(src) |
| + loc, gen0 := w.claim.Holder.Location, w.claim.Holder.Gen |
| + w.hold = func(r *request) bool { |
| + // every level-0 upload of the source's own generation is held: its |
| + // replica stops short of the stopped database |
| + return r.kind == rPut && r.region == src.Region && r.b != nil && r.b.Level == 0 && genOf(loc, r.key) == gen0 |
| + } |
| + w.appWrite(src) |
| + w.replicate(src) |
| + w.moveTo(tgt.Region) |
| + if !w.reach(t, 600, "landing built", func() bool { |
| + c := w.claim |
| + return c.Op.Kind == "landing" && (c.Op.LandPhase == "local" || c.Op.LandPhase == "config" || c.Op.LandPhase == "probing") |
| + }, w.act(src, tgt), w.recoveryPass) { |
| + return |
| + } |
| + if w.claim.Op.SrcCovered { |
| + t.Fatalf("witness: the seal found the source's replica complete; the held upload was not held: %s", w.state()) |
| + } |
| + w.rollback(w.claim, "configuration changed during the move") |
| + // the source's recovery admission decides with the replica still short |
| + w.reach(t, 800, "source admitted after the rollback", func() bool { |
| + c := w.claim |
| + return len(w.violations) > 0 || c.Op.Kind == "serving" || c.Op.Kind == "restoring" && (c.Op.SrcPhase == "probed" || c.Op.SrcPhase == "opened") |
| + }, w.act(src, tgt), w.recoveryPass, w.runConsole) |
| + w.hold = nil |
| + w.completeHeld(dataKey(loc, gen0, "")) |
| + w.finish(t) |
| +} |
diff --git a/internal/sim/hostlife/world_test.go b/internal/sim/hostlife/world_test.go| index 18028fcb0..59e8415d4 100644 |
| --- a/internal/sim/hostlife/world_test.go |
| +++ b/internal/sim/hostlife/world_test.go |
| @@ -267,6 +267,12 @@ type localDB struct { |
| TX int // the lineage's transaction id: the oracle's order of acknowledged writes |
| Pos int // the replication position: litestream's TXID, the count of files shipped, which every LTX range is numbered in (R5-1) |
| Pending []int // writes since Pos (what the next LTX file ships) |
| + // Epoch is the agent process that was admitted for this database: a |
| + // session belongs to the agent's replicator, and a new agent process |
| + // is a new session (claim rev 6.10), so a restarted agent neither |
| + // replicates nor certifies for it (seed 8528 / 30,000: the restarted |
| + // agent's first sync was skipped at the dead session's remote position) |
| + Epoch int |
| // the local level-0 files not yet confirmed uploaded, oldest first: |
| // litestream writes each sync's file locally and the replica uploads |
| // them in TXID order, never past one that has not completed (rev 7.6) |
| @@ -359,11 +365,21 @@ type box struct { |
| // writer is admitted only if its stamp equals the active one. |
| Stamps map[int]*stamps |
| Parked int // count of locally parked artifacts |
| - // open is an admitted writer's start in progress (rev 7.10 §7.14): the |
| - // claim CAS is done and the PUT it must have acknowledged before the |
| - // process starts (its first upload, its marker, its anchor) is |
| - // outstanding; local bookkeeping, not a claim transition |
| + // open is an admitted session's start in progress (rev 7.11 §7.14): |
| + // the claim CAS is done; the purge, the restore, the app start and |
| + // readiness, the first upload, the marker and the anchor follow, each |
| + // retried by the same agent process; the session is the agent's, and a |
| + // new agent process is a new session (claim rev 6.10) |
| open *openState |
| + // the retry policy for failed app starts within a session (rev 7.11 |
| + // §7.14): consecutive failures, the earliest next attempt, and the |
| + // host marked failed once the limit is reached (no new sessions until |
| + // an operator clears it) |
| + startFails int |
| + nextStart int |
| + hostFailed bool |
| + failCounted bool // the operation's probe counted this host failure as one failed try |
| + failedAt int // when the host was marked failed; the operator looks after a while |
| // PurgeFails is the fault: the purge of the local database an admitted |
| // writer must make before it starts returns an error (claim rev 6.9) |
| PurgeFails bool |
| @@ -418,6 +434,12 @@ func (b *box) stampOf(inc int) int { |
| return 0 |
| } |
| +// stampRetired: st is in the record's retired_stamps for inc. |
| +func (b *box) stampRetired(inc, st int) bool { |
| + s := b.Stamps[inc] |
| + return s != nil && contains(s.Retired, st) |
| +} |
| + |
| // install is activation (claim rev 6.5 §6.4): the box's record for inc |
| // takes holder.stamp as the installing CAS wrote it (a create's serving |
| // CAS, a landing's flip, an adoption, a reinstall, a dirty rollback's |
| @@ -509,6 +531,11 @@ func (b *box) admitsStamp(inc, stamp int) bool { |
| func (b *box) forget() { |
| b.Epoch++ // every goroutine of the old agent is gone |
| + // a new agent process is a new session (claim rev 6.10): an admitted |
| + // start in progress is gone with its restored database and its |
| + // outstanding requests (R10-2), and the start retry count with it |
| + b.open = nil |
| + b.startFails, b.nextStart = 0, 0 |
| b.deferred = nil |
| b.snap, b.stopped, b.stale = nil, nil, nil |
| b.mem = map[string]bool{} |
| @@ -606,6 +633,8 @@ type writerEntry struct { |
| Mode string // warm or cold (claim rev 6.9) |
| Pred pred |
| Lost *writesLost // a cold admission whose observed session had no certificate (claim rev 6.9, R9-2) |
| + Baseline string // remote, or sealed when a seal covers the stopped predecessor (claim rev 6.9.1, BaselineCoversPredecessor) |
| + Retired bool // this cold admission's predecessor has had its retire chore queued (by this writer's marker, or a successor's) |
| } |
| // pred is the predecessor an admission journals: the session, writer box, |
| @@ -626,9 +655,15 @@ type openState struct { |
| Session int |
| Mode string |
| Prepared bool |
| + Probe bool |
| + Source string // the generation the restore reads |
| + Dead []string // what the new marker retires: the predecessor and, behind a lagging marker, every empty generation of the unfinished admissions between (claim rev 6.10) |
| + RemoteMax int // the certificate's max_txid a warm restore must reach |
| DB *localDB |
| - stage string |
| + stage string // restore, app, first, marker, anchor, open |
| req *request |
| + blob *blob // the first upload's or the anchor's bytes, captured once: a retry carries the same bytes (R10-7; seed 7574 / 30,000) |
| + pendAt int // the pending writes the captured blob covers; later acknowledgements stay pending |
| } |
| // writesLost is claim rev 6.9's writes_lost record: the writer whose |
| @@ -679,6 +714,20 @@ type opState struct { |
| ReturnFiles ref |
| ReturnQuiet int |
| ReturnParked bool // the return lane's park-staging chore is owed (rev 7.10) |
| + // SrcCovered (rev 7.11): at the seal, the source's own replica of its |
| + // generation reached the stopped database's position, so a clean |
| + // rollback may restore the source from it; otherwise the seal holds |
| + // writes the replica does not, and the return carries the seal |
| + SrcCovered bool |
| + // SealFromDB (rev 7.11): the seal was built from the stopped database |
| + // itself, every page of it, so it covers every write the predecessor |
| + // acknowledged; false when the disk was gone and the seal is the |
| + // remote as it stood, which stops at the remote's position. The |
| + // prepared generation's admission declares writes_lost in that case |
| + // (seed 34260 / 100,000). ReturnFromDB is the same for a dirty |
| + // return's seal of the target |
| + SealFromDB bool |
| + ReturnFromDB bool |
| LostBox int |
| LostAt int |
| RollbackID string |
| @@ -823,6 +872,15 @@ type world struct { |
| // for is accepted and completes only when the witness says so |
| // (witness_test.go). The random sweep never sets it. |
| hold func(r *request) bool |
| + // alerts raised to the operator (writes_lost, a marker disagreement, a |
| + // failed host, an uncertified close): counted for the oracle that every |
| + // writes_lost record was alerted (R10-6) |
| + alerts int |
| + lostAlert int |
| + markerAlerted map[string]bool |
| + // startHook runs right after the app starts, before replication opens: |
| + // a witness's background worker acknowledging work (R10-1) |
| + startHook func(b *box) bool // true: the hook crashed the box |
| // lastIssued is the request issue saw last, for a caller that needs |
| // the request object behind a helper's return value (resumeOpen) |
| lastIssued *request |