amadan
sim: hostlife: writer sessions and admission on every path (claim rev 6.8, 6.8.1), the empty generation, provenance, close evidence (rev 7.9)
Joint round 8. AdmitWriter as one CAS as the process starts (open, warm, fresh, cold modes), holder.session and the writers journal with its predecessor, per-session certificates never deleted, a cold admission inside the operation it interrupts (op.Rec, recStep), only op.exec admitted, the probe restarted in its settle window, the crash-with-disk fault in every operation, snapshots at TXID 1 for a never-replicated database and nothing for an empty one (INV-12), lineage files and the three provenance conditions, every request of a session tracked with a sticky uncertainty, the replica's lazy remote position. Rules writer-session, admit-everywhere, exec-writer, warm-provenance, valid-ltx, close-evidence; seven witnesses; the admission scenario's named failure is INV-6. Faults found by the sweep and fixed: a stalled worker replaying a finished build (the revalidation step includes the writer state), carried files untagged by the installing CAS, a source that lost its disk during a rollback, a declared-lost write that came back (the local oracle), a never-replicated seal taken for empty, the biased schedule's unbounded hold in the calm phase. Diagnostics for why an admission is cold, which baseline a build takes and what blocks convergence. litestreamfacts: TestFact_R8_3_EmptyBootstrap, neither database nor replica: the encoder refuses zero TXIDs, an empty replica restores nothing, the first sync uploads nothing, the first transaction is [1,1].
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
11 files changed,
+1065
−338
internal/sim/hostlife/background_test.go+8 −4internal/sim/hostlife/console_test.go+3 −3internal/sim/hostlife/expected-outcomes.txt+5 −4internal/sim/hostlife/ltx_test.go+102 −24internal/sim/hostlife/moves_test.go+117 −167internal/sim/hostlife/protocol_test.go+344 −61internal/sim/hostlife/rules_test.go+6 −0internal/sim/hostlife/sim_test.go+76 −8internal/sim/hostlife/witness_test.go+211 −22internal/sim/hostlife/world_test.go+134 −45internal/sim/litestreamfacts/facts_test.go+59 −0
diff --git a/internal/sim/hostlife/background_test.go b/internal/sim/hostlife/background_test.go| index dae4d547a..047da526d 100644 |
| --- a/internal/sim/hostlife/background_test.go |
| +++ b/internal/sim/hostlife/background_test.go |
| @@ -155,7 +155,7 @@ func (w *world) adopt() bool { |
| if eb == nil { |
| return false |
| } |
| - n := &claim{Ver: 1, Holder: &holder{Inc: w.rec.Inc, App: w.rec.App, Legacy: true, Location: locationFor(w.rec.Inc, true), Stamp: 1}, // adoption writes holder.stamp (claim rev 6.5 §2) |
| + n := &claim{Ver: 1, Holder: &holder{Inc: w.rec.Inc, App: w.rec.App, Legacy: true, Location: locationFor(w.rec.Inc, true), Stamp: 1, Session: 1}, // adoption writes holder.stamp (claim rev 6.5 §2); the session starts with it |
| Op: opState{Kind: "serving", Exec: eb.ID, Parts: []int{eb.ID}}, Regions: []int{w.rec.Region}, Boxes: []int{eb.ID}, Complete: false} |
| if w.stampBox == nil { |
| w.stampBox = map[int]int{} |
| @@ -198,7 +198,7 @@ func (w *world) reconcileClaimed(b *box, c *claim) { |
| return |
| } |
| switch c.Op.Kind { |
| - case "serving", "creating", "pushing", "recovering": |
| + case "serving", "creating", "pushing": |
| if c.Op.Exec == b.ID { |
| if b.Row != nil && b.Row.Status != "ready" && c.Op.Kind == "serving" && on("stale-hold-release") { |
| // a hold whose operation has ended (the claim is serving here) |
| @@ -348,7 +348,9 @@ func (w *world) fault() { |
| // a crash that takes the disk with it (an instance replaced, a |
| // volume gone): the writer's uploads may still be in flight, |
| // which is the writer-epoch hazard (claim rev 6.6 §6.4a, R7-1) |
| - if c := w.claim; w.fail["disk"] && c != nil && c.Holder != nil && c.Op.Kind == "serving" && c.Op.Exec == b.ID && b.DB != nil && w.rng.Intn(3) == 0 { |
| + // 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) { |
| @@ -356,7 +358,8 @@ func (w *world) fault() { |
| } |
| } |
| b.DB = nil |
| - b.Procs = nil // nothing survives the disk: a descendant with no database would acknowledge writes into nowhere |
| + b.lineage = nil // the lineage file goes with the disk |
| + b.Procs = nil // nothing survives the disk: a descendant with no database would acknowledge writes into nowhere |
| w.stats["disk-lost with the crash"]++ |
| w.logf("box %d lost its disk with the crash", b.ID) |
| } |
| @@ -435,6 +438,7 @@ func (w *world) fault() { |
| } |
| } |
| b.DB = nil |
| + b.lineage = nil |
| w.stats["disk-lost"]++ |
| w.logf("box %d lost its database file", b.ID) |
| } |
diff --git a/internal/sim/hostlife/console_test.go b/internal/sim/hostlife/console_test.go| index 3675012b9..be458ceb8 100644 |
| --- a/internal/sim/hostlife/console_test.go |
| +++ b/internal/sim/hostlife/console_test.go |
| @@ -330,7 +330,7 @@ func (w *world) userDelete() { |
| if c == nil || c.Holder == nil { |
| return |
| } |
| - if on("claim-op-exclusive") && c.Op.Kind != "serving" { |
| + if on("claim-op-exclusive") && c.Op.Kind != "serving" || c.Op.Rec != nil { |
| return |
| } |
| id := w.newID("del") |
| @@ -363,7 +363,7 @@ func (w *world) userDelete() { |
| func (w *world) userMove() { |
| c := w.claim |
| - if c == nil || c.Holder == nil || c.Op.Kind != "serving" { |
| + if c == nil || c.Holder == nil || c.Op.Kind != "serving" || c.Op.Rec != nil { |
| return |
| } |
| if src := w.execBox(c); src == nil || src.Destroyed { |
| @@ -412,7 +412,7 @@ func (w *world) moveTo(r int) { |
| func (w *world) userPush() { |
| c := w.claim |
| - if c == nil || c.Holder == nil || c.Op.Kind != "serving" { |
| + if c == nil || c.Holder == nil || c.Op.Kind != "serving" || c.Op.Rec != nil { |
| return |
| } |
| b := w.execBox(c) |
diff --git a/internal/sim/hostlife/expected-outcomes.txt b/internal/sim/hostlife/expected-outcomes.txt| index 1b79c19bd..93003463c 100644 |
| --- a/internal/sim/hostlife/expected-outcomes.txt |
| +++ b/internal/sim/hostlife/expected-outcomes.txt |
| @@ -8,12 +8,13 @@ config-before-probe missed liveness only: a target that cannot render config fai |
| readiness-key missed liveness only: T2 avoids a pointless cut |
| readiness-quiet missed unreachable under claim rev 4: acknowledgements wait out their own horizon and CC11 holds |
| 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 three admission fences; caught with hold-op-open and start-check-claim-gen |
| -hold-op-open missed one of three admission fences; caught with exec-admission and start-check-claim-gen |
| -start-check-claim-gen missed one of three admission fences; caught with exec-admission and hold-op-open |
| -start-check missed caught with exec-admission and hold-op-open |
| +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 caught with exec-admission, hold-op-open and exec-writer |
| strict-spawn missed caught with evidence-whole-tree |
| 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 |
| registration-check missed a cheap extra: subsumed by purge-claim-fence and fence-etag together |
| +exec-writer missed one of four admission fences (claim rev 6.8.1): the AdmitWriter CAS refuses a box that is not op.exec; caught with exec-admission, hold-op-open and start-check-claim-gen (witness) |
diff --git a/internal/sim/hostlife/ltx_test.go b/internal/sim/hostlife/ltx_test.go| index 08c5d76a4..68065a9a4 100644 |
| --- a/internal/sim/hostlife/ltx_test.go |
| +++ b/internal/sim/hostlife/ltx_test.go |
| @@ -198,8 +198,20 @@ func isLevelDir(child string) bool { |
| // position: every page, TXIDs 1 to Pos, and the oracle's list of the |
| // writes it holds. The caller ships pending writes first, as litestream |
| // syncs before it snapshots. |
| +// snapshotOf is the snapshot of a database as a seal or an installer |
| +// encodes it: [1, N], N the replication position. A database that holds |
| +// transactions but has never replicated (position 0: a warm database |
| +// whose writes are all pending, a source sealed at its first stay) is |
| +// encoded at TXID 1, the TXID litestream's first sync gives its full-page |
| +// file; only a database with no transaction at all is empty, and the |
| +// encoder refuses a file for it (R8-3; seed 19023 / 30,000 took such a |
| +// seal for empty and lost the source's first write) |
| func snapshotOf(db *localDB, inc int) *blob { |
| - return &blob{Kind: "snap", Level: 9, Min: 1, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| + n := db.Pos |
| + if n == 0 && (len(db.Log) > 0 || len(db.State) > 0) { |
| + n = 1 |
| + } |
| + return &blob{Kind: "snap", Level: 9, Min: 1, Max: n, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| } |
| @@ -280,15 +292,55 @@ func (w *world) replicate(b *box) { |
| w.compact(b) |
| } |
| -// closedKey is the generation's clean-close marker (claim rev 6.6 §6.4a). |
| -func closedKey(loc, gen string) string { return dataKey(loc, gen, "closed.json") } |
| +// closedKey is a writer session's clean-close certificate (claim rev 6.8 |
| +// §6.4a): <location>/<generation>/closed/<session>.json, written once and |
| +// never deleted; the admitting CAS moves holder.session, so an old, |
| +// delayed or replayed certificate names a dead session. |
| +func closedKey(loc, gen string, session int) string { |
| + return dataKey(loc, gen, fmt.Sprintf("closed/%d.json", session)) |
| +} |
| + |
| +// isClosedKey: a clean-close certificate of any session. |
| +func isClosedKey(k string) bool { |
| + i := strings.LastIndex(k, "/closed/") |
| + return i >= 0 && strings.HasSuffix(k, ".json") |
| +} |
| + |
| +// track records a request the writer issued that did not return at once, |
| +// for the clean close's evidence (R8-5). |
| +func (db *localDB) track(r *request, v int) { |
| + if v <= 0 && !r.done { |
| + db.Issued = append(db.Issued, r) |
| + } |
| +} |
| -// closeStep is the clean close of a stopped writer (claim rev 6.6 §6.4a): |
| -// first every upload it issued must have returned success, then it |
| -// writes closed.json {stamp, max_txid} once (If-None-Match), then it |
| -// releases its lease. A writer whose uploads have not all returned writes |
| -// no marker; a crash while draining leaves none (forget). The mutation of |
| -// close-waits writes the marker as soon as the writer stops. |
| +// settled: every request the writer issued over its lifetime has returned |
| +// success and none was ever retried after a timeout (R8-5). The mutation |
| +// of close-evidence looks at the level-0 queue alone, as rev 7.8 did. |
| +func (db *localDB) settled() bool { |
| + queued := len(db.Queue) == 0 && (db.InFlight == nil || db.InFlight.done) |
| + if !on("close-evidence") { |
| + return queued |
| + } |
| + if !queued || db.Uncertain { |
| + return false |
| + } |
| + for _, r := range db.Issued { |
| + if !r.done { |
| + return false |
| + } |
| + } |
| + return true |
| +} |
| + |
| +// closeStep is the clean close of a stopped writer (claim rev 6.8 §6.4a): |
| +// first every request it issued over its lifetime must have returned |
| +// success (R8-5), then it writes closed/<session>.json {session, box, |
| +// max_txid} once (If-None-Match), then it releases its lease. A writer |
| +// with a request unresolved, or that ever retried after a timeout, writes |
| +// no certificate; a crash while draining leaves none (forget). The |
| +// mutation of close-waits writes the certificate as soon as the writer |
| +// stops. |
| func (w *world) closeStep(b *box) { |
| db := b.DB |
| if db == nil || !db.Closing || hasMain(b) { |
| @@ -302,13 +354,18 @@ func (w *world) closeStep(b *box) { |
| return |
| } |
| w.ship(b) |
| - drained := len(db.Queue) == 0 && (db.InFlight == nil || db.InFlight.done) |
| - if !drained && on("close-waits") { |
| + if !db.settled() && on("close-waits") { |
| + if db.Uncertain && on("close-evidence") { |
| + // a request of this writer timed out and was retried: its outcome |
| + // is unknown for good, so this session never certifies |
| + db.Closing = false |
| + w.stats["clean close refused: uncertain"]++ |
| + } |
| return |
| } |
| if db.CloseReq == nil { |
| - db.CloseReq = &request{kind: rPut, region: b.Region, key: closedKey(db.Loc, db.Gen), ifSet: true, |
| - b: &blob{Kind: "closed", Gen: db.Gen, Inc: db.Inc, Stamp: db.Stamp, TX: db.Pos}} |
| + db.CloseReq = &request{kind: rPut, region: b.Region, key: closedKey(db.Loc, db.Gen, db.Session), ifSet: true, |
| + b: &blob{Kind: "closed", Gen: db.Gen, Inc: db.Inc, Stamp: db.Stamp, Seq: b.ID, Session: db.Session, TX: db.Pos}} |
| w.issue(b.ID, db.CloseReq) |
| w.stats["clean closes"]++ |
| } |
| @@ -337,20 +394,29 @@ func (w *world) ship(b *box) { |
| if w.now-db.InFlightAt < uploadTimeout { |
| return |
| } |
| + // the client gives up and retries the same bytes; the |
| + // original's outcome is unknown for good (R8-5) |
| + db.Uncertain = true |
| w.stats["ltx upload retries"]++ |
| } else { |
| + db.RemotePos = db.Queue[0].Max |
| db.InFlight = nil |
| db.Queue = db.Queue[1:] |
| continue |
| } |
| } |
| bl := db.Queue[0] |
| - // litestream's upload loop (replica.go Sync): the remote position is |
| - // the newest remote level-0 file; a local file at or below it is |
| - // taken as already uploaded and skipped, and the writer believes it |
| - // replicated (R7-1, the initialization/remote-position race: another |
| - // writer's file at this TXID makes this one vanish) |
| - if w.maxL0(w.buckets[b.Region], dataKey(db.Loc, db.Gen, "")) >= bl.Max { |
| + // litestream's upload loop (replica.go Sync): the replica reads the |
| + // remote position once, at its first sync, caches it and advances it |
| + // with each upload it completes (R8-5); a local file at or below that |
| + // position is taken as already uploaded and skipped, and the writer |
| + // believes it replicated (R7-1, the initialization/remote-position |
| + // race: another writer's file at this TXID, present at that first |
| + // read, makes this one vanish) |
| + if db.RemotePos < 0 { |
| + db.RemotePos = w.maxL0(w.buckets[b.Region], dataKey(db.Loc, db.Gen, "")) |
| + } |
| + if db.RemotePos >= bl.Max { |
| w.markDurable(ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), bl) |
| w.stats["uploads skipped at the remote position"]++ |
| db.InFlight = nil |
| @@ -359,10 +425,12 @@ func (w *world) ship(b *box) { |
| } |
| r := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), b: bl} |
| if v := w.issue(b.ID, r); v > 0 { |
| + db.RemotePos = bl.Max |
| db.InFlight = nil |
| db.Queue = db.Queue[1:] |
| continue |
| } |
| + db.track(r, 0) |
| db.InFlight, db.InFlightAt = r, w.now |
| return |
| } |
| @@ -422,7 +490,9 @@ func (w *world) compact(b *box) { |
| keys = append(keys, ltxKey(db.Loc, db.Gen, 0, v.B.Min, v.B.Max)) |
| } |
| out.Max = max |
| - if v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 1, out.Min, out.Max), b: out}); v <= 0 { |
| + l1 := &request{kind: rPut, region: b.Region, key: ltxKey(db.Loc, db.Gen, 1, out.Min, out.Max), b: out} |
| + if v := w.issue(b.ID, l1); v <= 0 { |
| + db.track(l1, v) // a late or lost compaction upload is part of the writer's close evidence (R8-5) |
| return |
| } |
| // the inputs are deleted bar the newest: litestream's level-0 retention |
| @@ -435,7 +505,8 @@ func (w *world) compact(b *box) { |
| if k == keep && on("newest-l0") { |
| continue |
| } |
| - w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: k}) |
| + dr := &request{kind: rDelKey, region: b.Region, key: k} |
| + db.track(dr, w.issue(b.ID, dr)) |
| } |
| w.stats["ltx compactions"]++ |
| // snapshot rotation: a new snapshot at the position, then retention |
| @@ -447,7 +518,12 @@ func (w *world) compact(b *box) { |
| } |
| snap := snapshotOf(db, db.Inc) |
| snap.Kind = "ltx" |
| - if snap.Max <= prev || w.issue(b.ID, &request{kind: rPut, region: b.Region, key: snapKey(db.Loc, db.Gen, snap.Max), b: snap}) <= 0 { |
| + if snap.Max <= prev { |
| + return |
| + } |
| + sr := &request{kind: rPut, region: b.Region, key: snapKey(db.Loc, db.Gen, snap.Max), b: snap} |
| + if v := w.issue(b.ID, sr); v <= 0 { |
| + db.track(sr, v) |
| return |
| } |
| w.stats["snapshot rotations"]++ |
| @@ -456,7 +532,8 @@ func (w *world) compact(b *box) { |
| // and 0 below it |
| for _, s := range w.currentLTX(bk, prefix, 9) { |
| if s.B.Max < snap.Max { |
| - w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: snapKey(db.Loc, db.Gen, s.B.Max)}) |
| + dr := &request{kind: rDelKey, region: b.Region, key: snapKey(db.Loc, db.Gen, s.B.Max)} |
| + db.track(dr, w.issue(b.ID, dr)) |
| } |
| } |
| newest := w.maxL0(bk, prefix) |
| @@ -466,7 +543,8 @@ func (w *world) compact(b *box) { |
| continue // the newest level-0 file survives every retention path (R7-3) |
| } |
| if f.B.Max < prev { |
| - w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: ltxKey(db.Loc, db.Gen, level, f.B.Min, f.B.Max)}) |
| + dr := &request{kind: rDelKey, region: b.Region, key: ltxKey(db.Loc, db.Gen, level, f.B.Min, f.B.Max)} |
| + db.track(dr, w.issue(b.ID, dr)) |
| } |
| } |
| } |
diff --git a/internal/sim/hostlife/moves_test.go b/internal/sim/hostlife/moves_test.go| index ab2339697..34afc0581 100644 |
| --- a/internal/sim/hostlife/moves_test.go |
| +++ b/internal/sim/hostlife/moves_test.go |
| @@ -166,6 +166,12 @@ func (w *world) role(b *box, c *claim) { |
| if c.Holder == nil { |
| return |
| } |
| + if c.Op.Rec != nil && c.Op.Rec.Box == b.ID { |
| + // a cold admission's build runs under the operation, by the box it |
| + // admitted, before the operation's own step (claim rev 6.8 §6.4a, R8-2) |
| + w.recStep(b, c) |
| + return |
| + } |
| switch c.Op.Kind { |
| case "moving": |
| if b.ID == c.Op.Src { |
| @@ -189,10 +195,6 @@ func (w *world) role(b *box, c *claim) { |
| } else if b.ID == c.Op.Src { |
| w.recoverSource(b, c) |
| } |
| - case "recovering": |
| - if c.Op.Exec == b.ID { |
| - w.recoverStep(b, c) |
| - } |
| case "pushing": |
| if b.ID == c.Op.Src { |
| w.pushStep(b, c) |
| @@ -667,7 +669,14 @@ func (w *world) revertSource(b *box, c *claim) { |
| return |
| } |
| ph := "probing" |
| - if w.digestOf(b) != op.StopDigest { |
| + if b.DB == nil { |
| + // the source lost its disk since it stopped (R8-2): nothing to |
| + // verify; its writer is admitted like any lost replication |
| + // state, a warm continuation of the intact generation or a cold |
| + // admission (claim rev 6.8 §6.4a) |
| + w.logf("revert: box %d's database is gone; the probe restores it", b.ID) |
| + w.stats["revert: source database lost"]++ |
| + } else if w.digestOf(b) != op.StopDigest { |
| ph = "stuck" |
| w.logf("revert: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest) |
| } |
| @@ -751,8 +760,8 @@ func (w *world) followRecord(c *claim) bool { |
| return false |
| } |
| n.Op.Committed = true |
| - w.newGen(n.Holder.Location, n.Holder.Gen, op.ID) |
| - n.Holder.Gen = op.ID // the claim names the generation restore reads (C3a) |
| + w.newGen(n.Holder.Location, n.Holder.Gen, w.opGen(&op)) |
| + n.Holder.Gen = w.opGen(&op) // the claim names the generation restore reads (C3a) |
| q := op.StopAt + horizonH |
| if !on("quiet-before-ack") { |
| q = 0 |
| @@ -784,6 +793,7 @@ func (w *world) landTarget(b *box, c *claim) { |
| return |
| } |
| loc := c.Holder.Location |
| + g := w.opGen(&op) // the generation this landing installs (R8-2: a cold admission within it may have moved it) |
| if !op.Committed { |
| if w.rec != nil && w.rec.MoveID == op.ID && w.rec.MoveState != "landing" { |
| w.followRecord(c) // a commit or rollback the claim has not caught up with |
| @@ -854,7 +864,16 @@ func (w *world) landTarget(b *box, c *claim) { |
| w.rollback(c, "staging missing") |
| return |
| } |
| - key := snapKey(loc, op.ID, sv.B.Max) |
| + if sv.B.Max == 0 && on("valid-ltx") { |
| + // an empty source (R8-3): no LTX file can carry TXID 0, so the |
| + // generation is installed empty, its marker naming no snapshot |
| + w.cas(c, func(n *claim) bool { |
| + n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0 |
| + return sameOp(n.Op.ID, op.ID) |
| + }) |
| + return |
| + } |
| + key := snapKey(loc, g, sv.B.Max) |
| v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: src.Key, srcVer: src.Ver}) |
| if v > 0 || !on("build-completion") { |
| // the build journal records the snapshot's key, version and TXID |
| @@ -870,16 +889,16 @@ func (w *world) landTarget(b *box, c *claim) { |
| }) |
| } |
| case "install-anchor": |
| - if w.anchorAcked(b, loc, op.ID, op.BaseKey, op.BaseVer, op.BaseTX) { |
| + if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) { |
| w.cas(c, func(n *claim) bool { n.Op.LandPhase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "published": |
| - v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| + v := w.publishGen(b, loc, g, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| if v > 0 { |
| - w.cas(c, func(n *claim) bool { n.Op.LandPhase = "local"; return sameOp(n.Op.ID, op.ID) }) |
| + w.cas(c, func(n *claim) bool { n.Op.LandPhase, n.Op.Fresh = "local", true; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "local": |
| - w.installLocal(b, c, op.ID, op.SealFiles, op.LandPhase, func(n *claim) { n.Op.LandPhase = "config" }) |
| + w.installLocal(b, c, g, op.SealFiles, op.LandPhase, func(n *claim) { n.Op.LandPhase = "config" }) |
| case "config": |
| if !b.HasKey && on("config-before-probe") { |
| w.rollback(c, "configuration did not render") |
| @@ -896,7 +915,7 @@ func (w *world) landTarget(b *box, c *claim) { |
| b.Live, b.LiveInc = c.Holder.App, c.Holder.Inc |
| w.cas(c, func(n *claim) bool { n.Op.LandPhase = "restore"; return sameOp(n.Op.ID, op.ID) }) |
| case "restore": |
| - if b.DB == nil || b.DB.Gen != op.ID { |
| + if b.DB == nil || b.DB.Gen != g { |
| if b.DB != nil { |
| // an earlier stay's database: retire-artifacts of its stamp, |
| // never of the installation checking activated |
| @@ -905,7 +924,7 @@ func (w *world) landTarget(b *box, c *claim) { |
| } |
| w.retireDatabase(b, c.Holder.Inc, b.DB.Stamp) |
| } |
| - if b.DB = w.restoreInit(b, c.Holder.Inc, loc, op.ID); b.DB == nil && w.restoreCalc(b.Region, c.Holder.Inc, loc, op.ID) != nil { |
| + if b.DB = w.restoreInit(b, c.Holder.Inc, loc, g); b.DB == nil && w.restoreCalc(b.Region, c.Holder.Inc, loc, g) != nil { |
| return // the anchor is outstanding: try again |
| } |
| } |
| @@ -923,7 +942,7 @@ func (w *world) landTarget(b *box, c *claim) { |
| if !w.admits(b, c, op.ID) { |
| return |
| } |
| - if hasMain(b) || w.spawn(b, c, op.ID, true) { |
| + if hasMain(b) || w.spawn(b, c, g, true) { |
| if !on("commit-before-open") { |
| w.openRow(b, op.ID) |
| } |
| @@ -941,6 +960,26 @@ func (w *world) landTarget(b *box, c *claim) { |
| w.rollback(c, "cut deadline") |
| return |
| } |
| + if !hasMain(b) { |
| + // the probe died in its settle window (a crash, an agent |
| + // restart): the landing restarts only the probe (§7.14), a |
| + // writer admitted like any other (claim rev 6.8 §6.4a, R8-2); |
| + // a cold admission's build runs first (recStep) |
| + if c.Op.Rec != nil { |
| + return |
| + } |
| + if w.spawn(b, c, g, true) { |
| + w.stats["probe restarted"]++ |
| + w.cas(c, func(n *claim) bool { n.Op.ProbedAt = w.now; return sameOp(n.Op.ID, op.ID) }) |
| + return |
| + } |
| + tries := op.ProbeTries + 1 |
| + w.cas(c, func(n *claim) bool { n.Op.ProbeTries = tries; return sameOp(n.Op.ID, op.ID) }) |
| + if tries >= probeTries { |
| + w.rollback(c, "probe died") |
| + } |
| + return |
| + } |
| 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 |
| @@ -1077,8 +1116,6 @@ func (w *world) release(c *claim, execBox int) { |
| w.stats["revert"]++ |
| case op.Kind == "moving": |
| w.stats["abort"]++ |
| - case op.Kind == "recovering": |
| - w.stats["cold-recovery-done"]++ |
| case op.Kind == "pushing": |
| w.stats["push"]++ |
| } |
| @@ -1221,7 +1258,12 @@ func (w *world) recoverSource(b *box, c *claim) { |
| return |
| } |
| ph := "config" |
| - if w.digestOf(b) != op.StopDigest { |
| + if b.DB == nil { |
| + // the source lost its disk since it stopped (R8-2): see the |
| + // revert's verify above |
| + w.logf("recover: box %d's database is gone; the probe restores it", b.ID) |
| + w.stats["rollback: source database lost"]++ |
| + } else if w.digestOf(b) != op.StopDigest { |
| ph = "stuck" |
| w.logf("recover: digest mismatch %+v vs %+v", w.digestOf(b), op.StopDigest) |
| } |
| @@ -1238,7 +1280,15 @@ func (w *world) recoverSource(b *box, c *claim) { |
| if sv == nil { |
| return // the lane is gone: wait (stuck, in time) |
| } |
| - key := snapKey(c.Holder.Location, op.RollbackID, sv.B.Max) |
| + g := w.opGen(&op) |
| + if sv.B.Max == 0 && on("valid-ltx") { |
| + w.cas(c, func(n *claim) bool { |
| + n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", 0, "", 0 |
| + return sameOp(n.Op.ID, op.ID) |
| + }) |
| + return |
| + } |
| + key := snapKey(c.Holder.Location, g, sv.B.Max) |
| v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: key, srcRegion: b.Region, srcKey: op.ReturnDB.Key, srcVer: op.ReturnDB.Ver}) |
| if v > 0 || !on("build-completion") { |
| next := "install-anchor" |
| @@ -1251,16 +1301,16 @@ func (w *world) recoverSource(b *box, c *claim) { |
| }) |
| } |
| case "install-anchor": |
| - if w.anchorAcked(b, c.Holder.Location, op.RollbackID, op.BaseKey, op.BaseVer, op.BaseTX) { |
| + if w.anchorAcked(b, c.Holder.Location, w.opGen(&op), op.BaseKey, op.BaseVer, op.BaseTX) { |
| w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "published": |
| - v := w.publishGen(b, c.Holder.Location, op.RollbackID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| + v := w.publishGen(b, c.Holder.Location, w.opGen(&op), c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| if v > 0 { |
| - w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "local"; return sameOp(n.Op.ID, op.ID) }) |
| + w.cas(c, func(n *claim) bool { n.Op.SrcPhase, n.Op.Fresh = "local", true; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "local": |
| - w.installLocal(b, c, op.RollbackID, op.ReturnFiles, "", func(n *claim) { n.Op.SrcPhase = "config" }) |
| + w.installLocal(b, c, w.opGen(&op), op.ReturnFiles, "", func(n *claim) { n.Op.SrcPhase = "config" }) |
| case "config": |
| b.Config, b.CfgVer, b.Live, b.LiveInc = c.Holder.Inc, w.cfgVer[c.Holder.App], c.Holder.App, c.Holder.Inc |
| w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probing"; return sameOp(n.Op.ID, op.ID) }) |
| @@ -1270,7 +1320,7 @@ func (w *world) recoverSource(b *box, c *claim) { |
| } |
| want := c.Holder.Gen |
| if op.Dirty { |
| - want = op.RollbackID |
| + want = w.opGen(&op) |
| } |
| if hasMain(b) || w.spawn(b, c, want, true) { |
| w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "probed"; return sameOp(n.Op.ID, op.ID) }) |
| @@ -1284,10 +1334,10 @@ func (w *world) recoverSource(b *box, c *claim) { |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| case "probed": |
| - if op.Dirty && c.Holder.Gen != op.RollbackID { |
| + if op.Dirty && c.Holder.Gen != w.opGen(&op) { |
| w.cas(c, func(n *claim) bool { |
| - w.newGen(n.Holder.Location, n.Holder.Gen, op.RollbackID) |
| - n.Holder.Gen = op.RollbackID |
| + w.newGen(n.Holder.Location, n.Holder.Gen, w.opGen(&n.Op)) |
| + n.Holder.Gen = w.opGen(&n.Op) |
| n.Holder.Stamp = w.issueStamp(n, op.Src) // a dirty rollback installs a new generation: a new stamp, never the predecessor's (claim rev 6.5 §6.4) |
| n.Obl = append(n.Obl, w.newObl(n, "park-staging", op.SrcRegion, -1, fmt.Sprintf("moves/%s/%s/", host, op.RollbackID), "", op.ReturnQuiet)) |
| return sameOp(n.Op.ID, op.ID) |
| @@ -1307,6 +1357,7 @@ func (w *world) recoverSource(b *box, c *claim) { |
| func (w *world) pushStep(b *box, c *claim) { |
| op := c.Op |
| loc := c.Holder.Location |
| + g := w.opGen(&op) // the generation this push installs (R8-2) |
| switch op.Phase { |
| case "stopping": |
| w.holdRow(b, "moving", op.ID, c) |
| @@ -1316,7 +1367,7 @@ 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) |
| - snap.Gen = op.ID |
| + snap.Gen = g |
| w.nextW++ |
| mark := w.nextW |
| w.acks = append(w.acks, ack{W: mark, Inc: c.Holder.Inc, Box: b.ID, At: w.now, Kind: "write"}) |
| @@ -1344,7 +1395,7 @@ func (w *world) pushStep(b *box, c *claim) { |
| if b.Files != nil { |
| snap.SealKey = b.Files.Key |
| } |
| - key := snapKey(loc, op.ID, snap.Max) |
| + key := snapKey(loc, g, snap.Max) |
| v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: snap}) |
| if v > 0 || !on("build-completion") { |
| next := "install-anchor" |
| @@ -1357,20 +1408,20 @@ func (w *world) pushStep(b *box, c *claim) { |
| }) |
| } |
| case "install-anchor": |
| - if w.anchorAcked(b, loc, op.ID, op.BaseKey, op.BaseVer, op.BaseTX) { |
| + if w.anchorAcked(b, loc, g, op.BaseKey, op.BaseVer, op.BaseTX) { |
| w.cas(c, func(n *claim) bool { n.Op.Phase = "published"; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "published": |
| - v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| + v := w.publishGen(b, loc, g, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX) |
| if v > 0 { |
| - w.cas(c, func(n *claim) bool { n.Op.Phase = "restore"; return sameOp(n.Op.ID, op.ID) }) |
| + w.cas(c, func(n *claim) bool { n.Op.Phase, n.Op.Fresh = "restore", true; return sameOp(n.Op.ID, op.ID) }) |
| } |
| case "restore": |
| - if b.DB != nil && b.DB.Gen != op.ID { |
| + if b.DB != nil && b.DB.Gen != g { |
| w.retireDatabase(b, c.Holder.Inc, b.DB.Stamp) // the files are carried (F7) |
| } |
| if b.DB == nil { |
| - b.DB = w.restoreInit(b, c.Holder.Inc, loc, op.ID) |
| + b.DB = w.restoreInit(b, c.Holder.Inc, loc, g) |
| } |
| if b.DB == nil { |
| return |
| @@ -1378,7 +1429,7 @@ func (w *world) pushStep(b *box, c *claim) { |
| // .files-gen last: the files are unchanged, the marker names the |
| // new generation (F7 on §5.1) |
| if b.Files != nil { |
| - b.Files.Gen = op.ID |
| + b.Files.Gen = g |
| } |
| // every probe renders config and materializes the release for the |
| // held row first (the passes skip held rows): RenderHeld, MaterializeHeld |
| @@ -1391,11 +1442,11 @@ func (w *world) pushStep(b *box, c *claim) { |
| if !w.admits(b, c, op.ID) { |
| return |
| } |
| - if hasMain(b) || w.spawn(b, c, op.ID, true) { |
| + if hasMain(b) || w.spawn(b, c, g, true) { |
| w.cas(c, func(n *claim) bool { |
| n.Op.Phase = "probed" |
| - w.newGen(n.Holder.Location, n.Holder.Gen, op.ID) |
| - n.Holder.Gen = op.ID |
| + w.newGen(n.Holder.Location, n.Holder.Gen, w.opGen(&n.Op)) |
| + n.Holder.Gen = w.opGen(&n.Op) |
| n.Holder.Stamp = w.issueStamp(n, n.Op.Exec) // a push's commit installs a new generation: a new stamp (claim rev 6.5 §2) |
| q := op.StopAt + horizonH |
| if q > n.Quiet[b.Region] { |
| @@ -1418,9 +1469,23 @@ func sameStep(a, b *claim) bool { |
| return a == b |
| } |
| x, y := a.Op, b.Op |
| - return x.ID == y.ID && x.Kind == y.Kind && x.Phase == y.Phase && x.LandPhase == y.LandPhase && |
| + if !(x.ID == y.ID && x.Kind == y.Kind && x.Phase == y.Phase && x.LandPhase == y.LandPhase && |
| x.SrcPhase == y.SrcPhase && x.RevPhase == y.RevPhase && x.Exec == y.Exec && x.Committed == y.Committed && |
| - x.Opened == y.Opened && len(a.Obl) == len(b.Obl) && (a.Holder == nil) == (b.Holder == nil) |
| + x.Opened == y.Opened && len(a.Obl) == len(b.Obl) && (a.Holder == nil) == (b.Holder == nil)) { |
| + return false |
| + } |
| + // the writer state is part of the step (claim rev 6.8 §6.4a, R8-2): a |
| + // stalled worker that read a cold admission in progress must not replay |
| + // its build once the admission completed and the writer runs, since the |
| + // build takes stop evidence and replaces the database (seed 10382 / |
| + // 30,000) |
| + if a.Holder != nil && (a.Holder.Session != b.Holder.Session || a.Holder.Gen != b.Holder.Gen || a.Holder.Stamp != b.Holder.Stamp) { |
| + return false |
| + } |
| + if (x.Rec == nil) != (y.Rec == nil) || x.Rec != nil && (x.Rec.Gen != y.Rec.Gen || x.Rec.Phase != y.Rec.Phase) { |
| + return false |
| + } |
| + return len(x.Writers) == len(y.Writers) |
| } |
| // sweep is claim rev 5's at-rest sweep (§6.2a, CC15): for a host with no |
| @@ -1542,11 +1607,11 @@ func (w *world) deadGeneration(c *claim, r int, k string) bool { |
| if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, h.Gen, "") { |
| return !authoritative |
| } |
| - if k == closedKey(h.Location, h.Gen) { |
| - return !authoritative // the live generation's clean-close marker |
| - } |
| - if strings.HasSuffix(k, "/closed.json") { |
| - return true // a dead generation's marker |
| + if isClosedKey(k) { |
| + if strings.HasPrefix(k, dataKey(h.Location, h.Gen, "closed/")) { |
| + return !authoritative // the live generation's certificates, every session's |
| + } |
| + return true // a dead generation's certificates |
| } |
| return isDataKey(k) |
| } |
| @@ -1563,7 +1628,7 @@ func genOf(loc, k string) string { |
| } |
| return "-" |
| } |
| - if isDataKey(k) && genPrefixOf(k) == loc || rest == "closed.json" { |
| + if isDataKey(k) && genPrefixOf(k) == loc || strings.HasPrefix(rest, "closed/") { |
| return "" |
| } |
| return "-" |
| @@ -1592,10 +1657,12 @@ func (w *world) publishGen(b *box, loc, gen string, inc int, base int, key strin |
| // TXID is current. A marker that names artifacts still in flight would |
| // make the completion predicate true before the artifacts exist |
| bk := w.buckets[b.Region] |
| - if sv := bk.byID(key, base); sv == nil || sv.Marker { |
| - w.violate("INV-11: generation.json for %s published before its snapshot %s@%d landed (R7-4)", gen, key, base) |
| - } else if on("restore-anchor") && tx > 0 && bk.current(ltxKey(loc, gen, 0, tx, tx)) == nil { |
| - w.violate("INV-11: generation.json for %s published before its anchor [%d,%d] landed (R7-4)", gen, tx, tx) |
| + if tx > 0 { |
| + if sv := bk.byID(key, base); sv == nil || sv.Marker { |
| + w.violate("INV-11: generation.json for %s published before its snapshot %s@%d landed (R7-4)", gen, key, base) |
| + } else if on("restore-anchor") && bk.current(ltxKey(loc, gen, 0, tx, tx)) == nil { |
| + w.violate("INV-11: generation.json for %s published before its anchor [%d,%d] landed (R7-4)", gen, tx, tx) |
| + } |
| } |
| r := &request{kind: rPut, region: b.Region, key: genMarkerKey(loc), b: &blob{Kind: "gen", Gen: gen, Inc: inc, Base: base, SnapKey: key, TX: tx}} |
| if w.fail["markerhold"] && w.rng.Intn(2) == 0 { |
| @@ -1633,120 +1700,3 @@ func (w *world) anchorAcked(b *box, loc, gen, snapKey string, snapVer, tx int) b |
| w.stats["restore anchors"]++ |
| return acked || !on("anchor-acked") |
| } |
| - |
| -// recoverStep is the cold recovery's data path (claim rev 6.6 §6.4a; |
| -// moves rev 7.8 §5.1 step 9): the installing CAS has named the new |
| -// generation and stamp; the box builds the new generation from the old |
| -// one and publishes it, then the claim returns to serving and the |
| -// ordinary wake admits the writer. |
| -// |
| -// - build: the baseline is the warm database when it is the old |
| -// generation's (it holds every acknowledged write, uploaded or not), |
| -// else the old generation restored by the pure calculation; the |
| -// snapshot [1, N] and the anchor [N, N] are PUT into the new |
| -// generation's prefix and acknowledged, then closed.json {stamp, |
| -// max_txid N} as the installer's clean close, so the first wake finds |
| -// the generation intact; |
| -// - publish: generation.json names the new generation; the operation |
| -// releases to serving. |
| -// |
| -// The database replicates only into the new generation (its Gen), from |
| -// position N, and its first upload carries every page (FirstFull). |
| -func (w *world) recoverStep(b *box, c *claim) { |
| - op := c.Op |
| - h := c.Holder |
| - loc := h.Location |
| - switch op.Phase { |
| - case "build": |
| - // nothing of the host may run while the baseline is taken: a |
| - // descendant that outlived the main process would write to the |
| - // database after the snapshot and before the generation moved |
| - // (seed 86799: a write acknowledged during the build was lost) |
| - if !w.evidence(b, op.ID) { |
| - return |
| - } |
| - var base *localDB |
| - reused := false |
| - bk := w.buckets[b.Region] |
| - if snaps := w.currentLTX(bk, dataKey(loc, op.ID, ""), 9); len(snaps) > 0 { |
| - // a replayed build (a crash after the snapshot landed, with or |
| - // without the disk): the snapshot already current in the new |
| - // generation is reused, never rebuilt from another baseline, |
| - // which could carry other bytes under the same key (R7-4, R6-5; |
| - // seed 86799: the first build's snapshot held a write the |
| - // rebuilt one from the old generation did not) |
| - sv := snaps[len(snaps)-1].B |
| - base = &localDB{Inc: h.Inc, Loc: loc, Gen: op.ID, Key: sv.SealKey, Pos: sv.Max, State: cloneState(sv.Pages), Log: txsOf(sv), TX: w.lineageTX(sv)} |
| - reused = true |
| - w.stats["cold recovery: snapshot reused on replay"]++ |
| - } else if b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == op.BaseGen { |
| - base = b.DB |
| - } else { |
| - base = w.restoreCalc(b.Region, h.Inc, loc, op.BaseGen) |
| - if base == nil { |
| - w.stats["cold recovery: baseline unrestorable"]++ |
| - return |
| - } |
| - } |
| - snap := snapshotOf(base, h.Inc) |
| - snap.Gen = op.ID |
| - if b.Files != nil { |
| - snap.SealKey = b.Files.Key |
| - } |
| - n := base.Pos |
| - key := snapKey(loc, op.ID, n) |
| - if !reused && !w.ackedPut(b, key, snap, false) { |
| - return |
| - } |
| - sv := bk.current(key) |
| - if sv == nil { |
| - return |
| - } |
| - if !w.anchorAcked(b, loc, op.ID, key, sv.ID, n) { |
| - return |
| - } |
| - closed := &blob{Kind: "closed", Gen: op.ID, Inc: h.Inc, Stamp: h.Stamp, TX: n} |
| - if !w.ackedPut(b, closedKey(loc, op.ID), closed, true) { |
| - return |
| - } |
| - // the local database is the new generation's from here |
| - db := base |
| - if db != b.DB { |
| - b.DB = db |
| - } |
| - db.Gen, db.Pending, db.Queue, db.InFlight, db.FirstFull, db.Closing, db.CloseReq = op.ID, nil, nil, nil, true, false, nil |
| - // the position is what the new generation's listing shows (its |
| - // anchor, once acknowledged), never assumed (R6-1) |
| - db.Pos = w.maxL0(w.buckets[b.Region], dataKey(loc, op.ID, "")) |
| - db.Key = snap.SealKey |
| - db.Stamp = h.Stamp // the new installation's stamp, which activation also tags |
| - if b.Files != nil { |
| - b.Files.Gen = op.ID // the files are unchanged; .files-gen names the new generation (F7) |
| - } |
| - w.cas(c, func(nc *claim) bool { |
| - nc.Op.Phase, nc.Op.BaseVer, nc.Op.BaseKey, nc.Op.BaseTX = "publish", sv.ID, key, n |
| - return sameOp(nc.Op.ID, op.ID) |
| - }) |
| - case "publish": |
| - cur := w.buckets[b.Region].current(genMarkerKey(loc)) |
| - if cur == nil || cur.B.Gen != op.ID { |
| - if v := w.publishGen(b, loc, op.ID, h.Inc, op.BaseVer, op.BaseKey, op.BaseTX); v <= 0 { |
| - return |
| - } |
| - } |
| - // the old generation is dead: nothing reads it once the marker has |
| - // moved, and the dead writer's late uploads land there (claim rev |
| - // 6.6 §6.4a). It is parked as a chore after the dead writer's |
| - // horizon, and anything later is the sweep's |
| - w.cas(c, func(nc *claim) bool { |
| - if !sameOp(nc.Op.ID, op.ID) { |
| - return false |
| - } |
| - nc.Op.Phase = "release" |
| - nc.Obl = append(nc.Obl, w.newObl(nc, "park-generation", b.Region, -1, loc, op.BaseGen, op.ReqAt+horizonH)) |
| - return true |
| - }) |
| - case "release": |
| - w.release(c, b.ID) |
| - } |
| -} |
diff --git a/internal/sim/hostlife/protocol_test.go b/internal/sim/hostlife/protocol_test.go| index 61c0cd2b4..7e6b8b1ed 100644 |
| --- a/internal/sim/hostlife/protocol_test.go |
| +++ b/internal/sim/hostlife/protocol_test.go |
| @@ -219,7 +219,7 @@ func (w *world) restoreCalc(r int, inc int, loc, gen string) *localDB { |
| } |
| } |
| } |
| - if root == nil && gen != "" { |
| + if root == nil && gen != "" && pinnedTX > 0 { |
| m := bk.current(genMarkerKey(loc)) |
| snap := bk.byID(m.B.SnapKey, m.B.Base) |
| if snap == nil || snap.Marker || snap.B.Max != m.B.TX { |
| @@ -251,7 +251,7 @@ func (w *world) restoreCalc(r int, inc int, loc, gen string) *localDB { |
| if gap && on("restore-gap-closed") { |
| return nil // a gap after cur: refuse (litestream: "non-contiguous ltx files") |
| } |
| - if len(applied) == 0 && gen != "" { |
| + if len(applied) == 0 && gen != "" && pinnedTX > 0 { |
| return nil |
| } |
| // the oracle's bookkeeping: the writes this database holds, which its |
| @@ -298,6 +298,7 @@ func (w *world) restoreInit(b *box, inc int, loc, gen string) *localDB { |
| remote = w.maxL0(bk, prefix) // durable evidence: what the listing shows now |
| } |
| db.FirstFull = true // litestream's first sync after a cold initialization is a full upload (fact R5-1c) |
| + db.RemotePos = remote |
| db.Pos = remote |
| w.logf("box %d restored %s: TXID %d at position %d (key %d, stamp %d)", b.ID, prefix, db.TX, db.Pos, db.Key, db.Stamp) |
| return db |
| @@ -335,7 +336,7 @@ func (w *world) bucketGen(b *box, loc string) (string, int) { |
| func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| h := c.Holder |
| loc := h.Location |
| - var consume *version // the closed.json the admitted writer consumes as it opens |
| + admitted := "" // the admission mode this call decided (claim rev 6.8 §6.4a) |
| if b.Old { |
| // an old executable: no claim, no markers, no locations; it |
| // restores whatever is newest at the legacy root |
| @@ -376,22 +377,20 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| w.retireDatabase(b, h.Inc, b.DB.Stamp) |
| } |
| } |
| - if !probe && c.Op.Kind == "serving" && !b.Old { |
| - // admitting a writer to holder.generation (claim rev 6.6 §6.4a). |
| - // A writer still closing holds the lease: the wake waits for it |
| - if b.DB != nil && b.DB.Closing { |
| - w.closeStep(b) |
| - if b.DB.Closing { |
| - w.stats["wake waited for a clean close"]++ |
| - return false |
| - } |
| - } |
| - state, m := w.replicationState(b, c) |
| - if state == "lost" { |
| - w.startRecovery(b, c) |
| + // every path that opens replication is admitted here (claim rev 6.8 |
| + // §6.4a, R8-2): an ordinary wake, a committed landing's serving, a |
| + // 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 |
| + if b.DB != nil && b.DB.Closing { |
| + w.closeStep(b) |
| + if b.DB.Closing { |
| + w.stats["wake waited for a clean close"]++ |
| return false |
| } |
| - consume = m |
| + } |
| + if admitted = w.admitWriter(b, c, want, probe); admitted == "" { |
| + return false |
| } |
| if b.DB == nil { |
| b.DB = w.restoreInit(b, h.Inc, loc, want) |
| @@ -451,10 +450,20 @@ 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 consume != nil { |
| - // the writer opens: the marker it continued from is consumed, so a |
| - // crash of this writer leaves no marker (claim rev 6.6 §6.4a) |
| - w.issue(b.ID, &request{kind: rDelVer, region: b.Region, key: closedKey(loc, h.Gen), ver: consume.ID}) |
| + if admitted != "" && !b.Old { |
| + // the admitting CAS, as the process starts; the writer opens under |
| + // the session it produced and writes its lineage file (claim rev 6.8 |
| + // §6.4a). The replica reads its remote position at its first sync |
| + // (litestream's Replica.Sync initializes it lazily), RemotePos -1 |
| + if !w.admitCommit(b, c, want, admitted) { |
| + return false |
| + } |
| + b.DB.Session = w.claim.Holder.Session |
| + b.DB.RemotePos = -1 |
| + if b.lineage == nil { |
| + b.lineage = map[int]lineageRec{} |
| + } |
| + b.lineage[h.Inc] = lineageRec{Gen: b.DB.Gen, Session: b.DB.Session} |
| } |
| strict := !b.Old && (on("strict-spawn") || w.rng.Intn(2) == 0) |
| b.Procs = append(b.Procs, &proc{Main: true, Strict: strict, Inc: h.Inc, Gen: want, Box: b.ID}) |
| @@ -513,65 +522,339 @@ func (w *world) admits(b *box, c *claim, probe string) bool { |
| return role(op.Src) && op.SrcPhase == "opened" && b.Row.Status == "ready" |
| case "pushing": |
| return probe == op.ID && op.Phase == "probing" |
| - case "recovering": |
| - return false // no writer until the new generation is published and the claim is serving again |
| } |
| return false |
| } |
| -// replicationState decides, at an ordinary wake, whether the box may |
| -// continue holder.generation or must recover cold (claim rev 6.6 §6.4a): |
| -// it continues only if closed.json exists for the generation, its max_txid |
| -// equals the generation's remote maximum TXID, and its stamp is the box's |
| -// own active stamp. The marker is consumed (deleted by its version) by |
| -// the writer as it opens, so a later crash of that writer leaves none; |
| -// marker returns the version to consume. Anything else is lost |
| -// replication state. The mutation of closed-marker takes any marker as |
| -// intact and consumes none; the mutation of recovery-generation never |
| -// recovers cold (rev 7.7's same-generation restart). |
| -func (w *world) replicationState(b *box, c *claim) (state string, marker *version) { |
| - if !on("recovery-generation") { |
| - return "intact", nil |
| - } |
| +// warmCertificate is the clean-close certificate that permits a warm |
| +// admission (claim rev 6.8 §6.4a): closed/<holder.session>.json exists for |
| +// the generation, its max_txid is the generation's remote maximum TXID over |
| +// every level, and its box is this box. The mutation of writer-session |
| +// takes the newest certificate of any session (rev 7.8's single marker); |
| +// the mutation of closed-marker ignores max_txid. |
| +func (w *world) warmCertificate(b *box, c *claim, gen string) *version { |
| h := c.Holder |
| bk := w.buckets[b.Region] |
| - prefix := dataKey(h.Location, h.Gen, "") |
| - m := bk.current(closedKey(h.Location, h.Gen)) |
| + prefix := dataKey(h.Location, gen, "") |
| + var m *version |
| + if on("writer-session") { |
| + m = bk.current(closedKey(h.Location, gen, h.Session)) |
| + } else { |
| + for _, k := range bk.keys(prefix) { |
| + if isClosedKey(k) { |
| + if v := bk.current(k); v != nil && (m == nil || v.ID > m.ID) { |
| + m = v |
| + } |
| + } |
| + } |
| + } |
| if m == nil { |
| - w.logf("box %d: no closed.json for %s: lost replication state", b.ID, prefix) |
| - return "lost", nil |
| + w.logf("box %d: no certificate for session %d of %s", b.ID, h.Session, prefix) |
| + return nil |
| + } |
| + if m.B.Seq != b.ID { |
| + w.stats["certificate of another box"]++ |
| + w.logf("box %d: certificate %d for %s is box %d's", b.ID, m.B.Session, prefix, m.B.Seq) |
| + return nil |
| } |
| - if !on("closed-marker") { |
| - return "intact", nil |
| + if on("closed-marker") && m.B.TX != w.remoteMax(bk, prefix) { |
| + w.logf("box %d: certificate %d for %s is stale (max_txid %d, remote %d)", b.ID, m.B.Session, prefix, m.B.TX, w.remoteMax(bk, prefix)) |
| + w.stats["stale certificate"]++ |
| + return nil |
| } |
| - if m.B.TX != w.remoteMax(bk, prefix) || m.B.Stamp != b.stampOf(h.Inc) { |
| - w.logf("box %d: closed.json for %s is stale (max_txid %d, remote %d, stamp %d, active %d)", b.ID, prefix, m.B.TX, w.remoteMax(bk, prefix), m.B.Stamp, b.stampOf(h.Inc)) |
| - w.stats["stale closed.json"]++ |
| - return "lost", nil |
| + return m |
| +} |
| + |
| +// lastWriter is the newest writers journal entry, or nil. |
| +func lastWriter(op *opState) *writerEntry { |
| + if len(op.Writers) == 0 { |
| + return nil |
| } |
| - return "intact", m |
| + return &op.Writers[len(op.Writers)-1] |
| } |
| -// startRecovery is the cold recovery's installing CAS (claim rev 6.6 |
| -// §6.4a): conditional on the observed claim, it writes a new |
| -// holder.generation (a fresh op id) and a new holder.stamp, requires |
| -// op.exec to be this box, and opens the recovering operation whose steps |
| -// (recoverStep) restore from the old generation into the new one. |
| -func (w *world) startRecovery(b *box, c *claim) { |
| +// admitWriter is AdmitWriter (claim rev 6.8 §6.4a, R8-1, R8-2): the one |
| +// claim CAS every path that opens replication makes before any upload. It |
| +// requires the observed holder.session and holder.generation, and op.exec |
| +// to be this box (claim rev 6.8.1, rule exec-writer); it writes |
| +// holder.session = its resulting gen and a writers journal entry. Three |
| +// modes, decided here and committed by admitCommit as the process starts |
| +// (so a start that fails after the decision leaves nothing admitted): |
| +// |
| +// - "open": the box's own cold admission has finished its build (the |
| +// newest entry is this box's, unopened, for gen): the writer opens it, |
| +// the CAS marking the entry opened; |
| +// - "warm": a certificate for the observed session permits a warm |
| +// admission; the CAS moves the session forward, so the certificate |
| +// names a dead session from then on; |
| +// - "fresh": the operation's freshly built generation has had no writer |
| +// (op.Fresh, an install's first probe): a cold admission whose build |
| +// was the install; |
| +// - otherwise lost replication state: a cold admission's CAS is made now |
| +// (op.Rec), its build runs under the operation (recStep), and this call |
| +// returns "" until the writer may open. |
| +func (w *world) admitWriter(b *box, c *claim, gen string, probe bool) string { |
| + h := c.Holder |
| + if on("exec-writer") && c.Op.Exec != b.ID { |
| + w.stats["admission refused: not op.exec"]++ |
| + return "" |
| + } |
| + if !on("admit-everywhere") && c.Op.Kind != "serving" { |
| + return "unguarded" // the mutation: writer epochs for ordinary wakes only (rev 7.8) |
| + } |
| + if c.Op.Rec != nil { |
| + return "" // a cold admission's build is in progress |
| + } |
| + if e := lastWriter(&c.Op); e != nil && e.Box == b.ID && e.Session == h.Session && e.Gen == gen && !e.Opened { |
| + return "open" |
| + } |
| + if !on("recovery-generation") { |
| + return "unguarded" // rev 7.7: a same-generation restart, no session |
| + } |
| + if m := w.warmCertificate(b, c, gen); m != nil { |
| + return "warm" |
| + } |
| + if c.Op.Fresh && w.opGen(&c.Op) == gen { |
| + return "fresh" // the operation's install was this writer's build |
| + } |
| + // cold: a fresh generation, built under this operation |
| id := w.newID("rec") |
| if w.cas(c, func(n *claim) bool { |
| - if n.Op.Kind != "serving" || n.Op.Exec != b.ID { |
| + if n.Holder.Session != h.Session || n.Op.Rec != nil || n.Op.Exec != b.ID && on("exec-writer") { |
| return false |
| } |
| - n.Op = opState{Kind: "recovering", ID: id, Phase: "build", Exec: b.ID, Parts: []int{b.ID}, Src: b.ID, Tgt: b.ID, SrcRegion: b.Region, TgtRegion: b.Region, ReqAt: w.now, BaseGen: n.Holder.Gen} |
| - w.newGen(n.Holder.Location, n.Holder.Gen, id) |
| - n.Holder.Gen = id |
| - n.Holder.Stamp = w.issueStamp(n, b.ID) |
| + n.Holder.Session = w.nextGen() |
| + p := pred{Session: h.Session, Box: w.predBox(c), Gen: gen, Stamp: h.Stamp} |
| + n.Op.Writers = append(n.Op.Writers, writerEntry{Session: n.Holder.Session, Box: b.ID, Gen: id, Mode: "cold", Pred: p}) |
| + n.Op.Rec = &recState{Gen: id, Base: gen, Session: n.Holder.Session, Box: b.ID, Phase: "build", Pred: p} |
| + w.newGen(n.Holder.Location, gen, id) |
| + if gen == n.Holder.Gen { |
| + n.Holder.Gen = id // the claim's generation moves with the writer it admits |
| + } else { |
| + n.Op.InstGen = id // an operation's install generation, not yet the claim's |
| + } |
| + n.Holder.Stamp = w.issueStamp(n, b.ID) // an installing CAS (claim rev 6.8 §6.4a) |
| + n.Op.Writers[len(n.Op.Writers)-1].Stamp = n.Holder.Stamp |
| + n.Op.Fresh = false |
| return true |
| }) { |
| - w.logf("box %d: cold recovery %s from generation %q (lost replication state)", b.ID, id, c.Holder.Gen) |
| - w.stats["cold recoveries"]++ |
| + w.logf("box %d: cold admission %s from generation %q (lost replication state, op %s/%s)", b.ID, id, gen, c.Op.Kind, c.Op.ID) |
| + w.stats["admissions: cold"]++ |
| + } |
| + return "" |
| +} |
| + |
| +// admitCommit is the admitting CAS for the open, warm and fresh modes, made |
| +// as the process starts; false means the claim moved and the start is |
| +// abandoned. |
| +func (w *world) admitCommit(b *box, c *claim, gen, mode string) bool { |
| + h := c.Holder |
| + if mode == "unguarded" { |
| + return true |
| + } |
| + ok := w.cas(c, func(n *claim) bool { |
| + if n.Holder.Session != h.Session || n.Op.Rec != nil || n.Op.Exec != b.ID && on("exec-writer") { |
| + return false |
| + } |
| + if mode == "open" { |
| + le := lastWriter(&n.Op) |
| + if le == nil || le.Session != h.Session || le.Opened { |
| + return false |
| + } |
| + le.Opened = true |
| + return true |
| + } |
| + n.Holder.Session = w.nextGen() |
| + n.Op.Fresh = false |
| + n.Op.Writers = append(n.Op.Writers, writerEntry{Session: n.Holder.Session, Box: b.ID, Gen: gen, Stamp: n.Holder.Stamp, Mode: mode, Opened: true, |
| + Pred: pred{Session: h.Session, Box: w.predBox(c), Gen: gen, Stamp: h.Stamp}}) |
| + return true |
| + }) |
| + if ok { |
| + w.stats["admissions: "+mode]++ |
| } |
| + return ok |
| +} |
| + |
| +// predBox is the box of the newest writer session the claim records, or |
| +// op.exec when it records none. |
| +func (w *world) predBox(c *claim) int { |
| + if e := lastWriter(&c.Op); e != nil { |
| + return e.Box |
| + } |
| + return c.Op.Exec |
| +} |
| + |
| +// opGen is the generation an operation installs: its install generation |
| +// when a cold admission within it moved it (R8-2), else its id, or the |
| +// rollback id for a rollback. |
| +func (w *world) opGen(op *opState) string { |
| + if op.InstGen != "" { |
| + return op.InstGen |
| + } |
| + if op.Kind == "restoring" { |
| + return op.RollbackID |
| + } |
| + return op.ID |
| +} |
| + |
| +// recStep is a cold admission's build (claim rev 6.8 §6.4a; moves rev 7.9 |
| +// §7.14), run under whichever operation admitted the writer, by the box it |
| +// admitted, before that operation's own step: |
| +// |
| +// - build: stop evidence first. The baseline is the warm database only |
| +// with provenance (R8-4): its lineage file names the predecessor's |
| +// {generation, session}, the predecessor's box is this box, and it |
| +// passes its integrity check (the model's database is always |
| +// consistent, WAL included); else the predecessor generation restored |
| +// by the pure calculation, and the warm database's unreplicated writes |
| +// are declared lost. A snapshot [1, N] and an anchor [N, N] go into the |
| +// new generation and are acknowledged when N > 0; an empty generation |
| +// gets no LTX file (R8-3). A replay reuses a snapshot already current. |
| +// - publish: generation.json names the new generation; the predecessor |
| +// generation is parked as a chore after the dead writer's horizon; the |
| +// build is done, and the writer opens at its next admission. |
| +func (w *world) recStep(b *box, c *claim) { |
| + rec := c.Op.Rec |
| + h := c.Holder |
| + loc := h.Location |
| + bk := w.buckets[b.Region] |
| + switch rec.Phase { |
| + case "build": |
| + if !w.evidence(b, rec.Gen) { |
| + return |
| + } |
| + var base *localDB |
| + reused := false |
| + if snaps := w.currentLTX(bk, dataKey(loc, rec.Gen, ""), 9); len(snaps) > 0 { |
| + sv := snaps[len(snaps)-1].B |
| + base = &localDB{Inc: h.Inc, Loc: loc, Gen: rec.Gen, Key: sv.SealKey, Pos: sv.Max, State: cloneState(sv.Pages), Log: txsOf(sv), TX: w.lineageTX(sv)} |
| + reused = true |
| + w.stats["cold admission: snapshot reused on replay"]++ |
| + } else if b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == rec.Base && w.warmProvenance(b, h.Inc, rec) { |
| + base = b.DB |
| + w.stats["cold admission: warm baseline"]++ |
| + } else { |
| + switch { |
| + case b.DB == nil: |
| + w.logf("box %d: cold admission %s has no local database; baseline from the remote of %q", b.ID, rec.Gen, rec.Base) |
| + case b.DB.Inc != h.Inc || b.DB.Gen != rec.Base: |
| + w.logf("box %d: cold admission %s finds a database of inc %d generation %q, not %q; baseline from the remote", b.ID, rec.Gen, b.DB.Inc, b.DB.Gen, rec.Base) |
| + } |
| + if b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == rec.Base { |
| + // a warm database without provenance: the remote wins, and |
| + // its writes that never replicated are declared lost |
| + n := 0 |
| + for i := range w.acks { |
| + a := &w.acks[i] |
| + if a.Box == b.ID && a.Inc == h.Inc && !a.Exempt && !w.isDurable(a.W) { |
| + a.Exempt = true |
| + n++ |
| + } |
| + } |
| + w.stats["cold admission: warm baseline refused, writes declared lost"] += n |
| + w.logf("box %d: warm database of %q has no provenance; %d unreplicated writes declared lost", b.ID, rec.Base, n) |
| + } |
| + base = w.restoreCalc(b.Region, h.Inc, loc, rec.Base) |
| + if base == nil { |
| + w.stats["cold admission: baseline unrestorable"]++ |
| + w.logf("box %d: cold admission %s cannot restore %q yet: %s", b.ID, rec.Gen, rec.Base, w.ltxShape(b.Region, dataKey(loc, rec.Base, ""))) |
| + return |
| + } |
| + w.stats["cold admission: remote baseline"]++ |
| + } |
| + snap := snapshotOf(base, h.Inc) |
| + n := snap.Max // the baseline's position; 1 for a database with transactions that never replicated; 0 only when empty (R8-3) |
| + if !on("valid-ltx") { |
| + n = base.Pos // the mutation: the position as is, a zero-TXID file for a never-replicated database |
| + snap.Max = n |
| + } |
| + w.logf("box %d: cold admission %s baseline: position %d, TXID %d, %d writes (reused snapshot %v, warm %v)", b.ID, rec.Gen, n, base.TX, len(base.Log), reused, base == b.DB && !reused) |
| + key := "" |
| + ver := 0 |
| + if n > 0 || !on("valid-ltx") { |
| + snap.Gen = rec.Gen |
| + if b.Files != nil { |
| + snap.SealKey = b.Files.Key |
| + } |
| + key = snapKey(loc, rec.Gen, n) |
| + if !reused && !w.ackedPut(b, key, snap, false) { |
| + return |
| + } |
| + sv := bk.current(key) |
| + if sv == nil { |
| + return |
| + } |
| + ver = sv.ID |
| + if !w.anchorAcked(b, loc, rec.Gen, key, ver, n) { |
| + return |
| + } |
| + base.Key = snap.SealKey |
| + } |
| + db := base |
| + if db != b.DB { |
| + b.DB = db |
| + } |
| + db.Gen, db.Pending, db.Queue, db.InFlight, db.FirstFull, db.Closing, db.CloseReq = rec.Gen, nil, nil, nil, true, false, nil |
| + db.Issued, db.Uncertain = nil, false |
| + db.Pos = w.maxL0(bk, dataKey(loc, rec.Gen, "")) |
| + db.RemotePos = db.Pos |
| + db.Stamp = h.Stamp |
| + if b.Files != nil { |
| + // the files are unchanged; .files-gen names the new generation (F7) |
| + // and the carried files are tagged with the installing CAS's |
| + // stamp, like the database (§6.4; seed 453 / 30,000) |
| + b.Files.Gen, b.Files.Stamp = rec.Gen, h.Stamp |
| + } |
| + w.cas(c, func(nc *claim) bool { |
| + if nc.Op.Rec == nil || nc.Op.Rec.Gen != rec.Gen { |
| + return false |
| + } |
| + nc.Op.Rec.Phase, nc.Op.Rec.BaseVer, nc.Op.Rec.BaseKey, nc.Op.Rec.BaseTX = "publish", ver, key, n |
| + return true |
| + }) |
| + case "publish": |
| + cur := bk.current(genMarkerKey(loc)) |
| + if cur == nil || cur.B.Gen != rec.Gen { |
| + if v := w.publishGen(b, loc, rec.Gen, h.Inc, rec.BaseVer, rec.BaseKey, rec.BaseTX); v <= 0 { |
| + return |
| + } |
| + } |
| + w.cas(c, func(nc *claim) bool { |
| + if nc.Op.Rec == nil || nc.Op.Rec.Gen != rec.Gen { |
| + return false |
| + } |
| + nc.Op.Rec = nil |
| + nc.Obl = append(nc.Obl, w.newObl(nc, "park-generation", b.Region, -1, loc, rec.Base, w.now+horizonH)) |
| + return true |
| + }) |
| + w.stats["cold admissions done"]++ |
| + } |
| +} |
| + |
| +// warmProvenance is claim rev 6.8 §6.4a's three conditions for taking the |
| +// warm database as a cold admission's baseline (R8-4): the local lineage |
| +// file, written by the writer that opened the database, names the |
| +// journalled predecessor's {generation, session}; the predecessor's box is |
| +// this box; and the database passes its integrity check with its WAL in |
| +// place, which here is that it stands at or beyond its own generation's |
| +// remote position (its uploads came from it; a shorter file is a truncated |
| +// one). The mutation of warm-provenance takes any database of the |
| +// predecessor generation. |
| +func (w *world) warmProvenance(b *box, inc int, rec *recState) bool { |
| + if !on("warm-provenance") { |
| + return true |
| + } |
| + l, ok := b.lineage[inc] |
| + if !ok || l.Gen != rec.Pred.Gen || l.Session != rec.Pred.Session || rec.Pred.Box != b.ID { |
| + return false |
| + } |
| + if remote := w.maxL0(w.buckets[b.Region], dataKey(b.DB.Loc, rec.Base, "")); b.DB.Pos < remote { |
| + w.logf("box %d: warm database of %q fails its integrity check: position %d behind its remote %d", b.ID, rec.Base, b.DB.Pos, remote) |
| + w.stats["cold admission: warm baseline failed integrity"]++ |
| + return false |
| + } |
| + return true |
| } |
| // ---------- parking (claim §6.2) ---------- |
diff --git a/internal/sim/hostlife/rules_test.go b/internal/sim/hostlife/rules_test.go| index bc15fa65c..623c09a4b 100644 |
| --- a/internal/sim/hostlife/rules_test.go |
| +++ b/internal/sim/hostlife/rules_test.go |
| @@ -91,6 +91,12 @@ var allRules = []string{ |
| "close-waits", // claim rev 6.6 §6.4a (R7-1): a clean close writes closed.json only after every upload returned; the mutation writes it at once |
| "newest-l0", // rev 7.8 §5.1 step 8 (R7-3): every retention path keeps the newest level-0 file; the mutation is rev 7.6's deletion of inputs and of every file below the previous snapshot |
| "build-completion", // rev 7.8 §7.6 (R7-4): generation.json is published only from a completed build, snapshot and anchor acknowledged; the mutation publishes from the intent |
| + "writer-session", // claim rev 6.8 §6.4a (R8-1): a warm admission needs the certificate of the observed holder.session and moves the session forward, so a certificate is one-use; the mutation takes the newest certificate of any session and moves nothing |
| + "admit-everywhere", // claim rev 6.8 §6.4a (R8-2): every path that opens replication is an AdmitWriter, a committed landing's serving, a probe restart, a reverted source, an opened rollback included; the mutation gates ordinary wakes only |
| + "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 |
| + "warm-provenance", // claim rev 6.8 §6.4a (R8-4): the warm database is a cold admission's baseline only with its lineage file naming the journalled predecessor's generation and session and the predecessor's box this box; the mutation takes any database of the predecessor generation |
| + "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 |
| "quiet-before-ack", // this model's finding (late writes after a park) |
| "straggler-sweep", // claim rev 5 §6.2a (CC15): the at-rest sweep |
| "sweep-staging", // CC20: the sweep also covers moves/<host>/ |
diff --git a/internal/sim/hostlife/sim_test.go b/internal/sim/hostlife/sim_test.go| index 6e921e98d..ab4254c60 100644 |
| --- a/internal/sim/hostlife/sim_test.go |
| +++ b/internal/sim/hostlife/sim_test.go |
| @@ -237,6 +237,60 @@ func (w *world) quiescent() bool { |
| return len(w.residue()) == 0 |
| } |
| +// whyNotQuiescent names, for a non-convergence report, what quiescent() |
| +// found unfinished. |
| +func (w *world) whyNotQuiescent() string { |
| + var out []string |
| + for _, r := range w.pending { |
| + out = append(out, fmt.Sprintf("pending %v %s r%d by %s issued %d due %d beyond=%v", r.kind, r.key, r.region, r.by, r.at, r.doneAt, r.beyond)) |
| + } |
| + if len(w.console) > 0 { |
| + out = append(out, fmt.Sprintf("%d console plans", len(w.console))) |
| + } |
| + if w.recPend != nil { |
| + out = append(out, "a record PUT pending") |
| + } |
| + if w.h2.Kind != "" { |
| + out = append(out, "h2 "+w.h2.Kind) |
| + } |
| + for app, ad := range w.admit { |
| + if ad.Closed { |
| + out = append(out, "admission closed for "+app) |
| + } |
| + for id, e := range ad.Reg { |
| + if w.now <= e.Until || !on("registration-recovery") { |
| + out = append(out, fmt.Sprintf("registration %s of %s live until %d", id, app, e.Until)) |
| + } |
| + } |
| + } |
| + if c := w.claim; c != nil && c.Holder != nil { |
| + if c.Op.Rec != nil { |
| + out = append(out, fmt.Sprintf("cold admission %s in phase %s", c.Op.Rec.Gen, c.Op.Rec.Phase)) |
| + } |
| + if c.Op.Kind != "serving" { |
| + out = append(out, "op "+c.Op.Kind) |
| + } |
| + for _, o := range c.Obl { |
| + out = append(out, fmt.Sprintf("obligation %s %s %s", o.ID, o.Kind, o.Prefix)) |
| + } |
| + if e := w.execBox(c); e != nil && (e.Config != c.Holder.Inc || e.Live != c.Holder.App || e.LiveInc != c.Holder.Inc) { |
| + out = append(out, "config or adoption owed") |
| + } |
| + for _, b := range w.liveBoxes() { |
| + if e := w.execBox(c); b != e && (b.Row != nil || len(b.Procs) > 0) { |
| + out = append(out, fmt.Sprintf("box %d has a row or processes", b.ID)) |
| + } |
| + } |
| + } |
| + if r := w.residue(); len(r) > 0 { |
| + out = append(out, fmt.Sprintf("residue %v", r)) |
| + } |
| + if len(out) == 0 { |
| + return "quiescent" |
| + } |
| + return strings.Join(out, "; ") |
| +} |
| + |
| // atRest: the claim, record and boxes have converged. |
| func (w *world) atRest() bool { |
| if w.h2.Kind == "creating" || w.h2.Kind == "purging" || w.h2.Kind == "fenced" { |
| @@ -259,7 +313,7 @@ func (w *world) atRest() bool { |
| if c.Holder == nil { |
| return c.Op.Kind == "" && w.rec == nil && len(c.Obl) == 0 |
| } |
| - if c.Op.Kind != "serving" || len(c.Obl) > 0 || w.rec == nil || w.rec.Inc != c.Holder.Inc || w.rec.MoveID != "" { |
| + if c.Op.Kind != "serving" || c.Op.Rec != nil || len(c.Obl) > 0 || w.rec == nil || w.rec.Inc != c.Holder.Inc || w.rec.MoveID != "" { |
| return false |
| } |
| e := w.execBox(c) |
| @@ -340,14 +394,14 @@ func (w *world) residueOf(exceptStragglers bool) []string { |
| loc, gen := c.Holder.Location, c.Holder.Gen |
| live[dataKey(loc, gen, "")] = true // the generation's prefix: every LTX file of the generation |
| live[genMarkerKey(loc)] = true |
| - live[closedKey(loc, gen)] = true // the live generation's clean-close marker (claim rev 6.6 §6.4a) |
| + live[dataKey(loc, gen, "closed/")] = true // the live generation's certificates, every session's (claim rev 6.8 §6.4a) |
| } |
| } |
| for r := 0; r < nRegions; r++ { |
| bk := w.buckets[r] |
| for _, p := range []string{"instances/" + host + "/", "moves/" + host + "/"} { |
| for _, k := range bk.keys(p) { |
| - if strings.HasSuffix(k, "lock.json") || (r == serving && (live[k] || isDataKey(k) && live[genPrefixOf(k)])) { |
| + if strings.HasSuffix(k, "lock.json") || (r == serving && (live[k] || isDataKey(k) && live[genPrefixOf(k)] || isClosedKey(k) && live[k[:strings.LastIndex(k, "closed/")+7]])) { |
| continue |
| } |
| if exceptStragglers && w.owed(r, k) { |
| @@ -427,7 +481,7 @@ func (w *world) final() { |
| w.violate("INV-10: residue never retired: %v", w.residue()) |
| return |
| } |
| - w.violate("INV-3: no convergence after %d calm steps: %s%s", calmSteps, w.state(), w.boxesState()) |
| + w.violate("INV-3: no convergence after %d calm steps: %s; %s%s", calmSteps, w.whyNotQuiescent(), w.state(), w.boxesState()) |
| return |
| } |
| c := w.claim |
| @@ -668,11 +722,22 @@ func (w *world) foreignWrites(db *localDB, inc int, how string) { |
| // acknowledged, non-exempt write of the incarnation (INV-2), and every |
| // acknowledged timer is delivered or still registered. |
| func (w *world) checkLocal(c *claim, e *box, db *localDB) { |
| - if why, ok := statesEqual(w.expectState(c.Holder.Inc, -1), db.State); !ok { |
| - have := map[int]bool{} |
| - for _, x := range db.Log { |
| - have[x] = true |
| + // the serving database is the replay of every acknowledged write that |
| + // was never declared lost, and of every write declared lost that came |
| + // back: an upload in flight at a lost disk that landed anyway, which a |
| + // later cold admission's remote baseline carried (R7-2's identity rule, |
| + // as checkDurable applies it; seed 5863 / 30,000) |
| + have := map[int]bool{} |
| + for _, x := range db.Log { |
| + have[x] = true |
| + } |
| + want := map[int]bool{} |
| + for _, a := range w.acks { |
| + if a.Inc == c.Holder.Inc && a.Kind == "write" && (!a.Exempt || have[a.W]) { |
| + want[a.W] = true |
| } |
| + } |
| + if why, ok := statesEqual(w.expectFrom(c.Holder.Inc, want), db.State); !ok { |
| w.violate("INV-2 (local): the serving database differs from the acknowledged writes: %s (%s)", why, w.keyHistory(c.Holder.Inc, why, have)) |
| return |
| } |
| @@ -720,6 +785,9 @@ func runSeed(seed int64, verbose bool) *world { |
| w.step() |
| if i == 400 || i%400 == 399 { |
| w.heal(true) |
| + if w.verbose { |
| + w.logf("calm step %d: %s", i, w.whyNotQuiescent()) |
| + } |
| } |
| } |
| w.completeDue(true) |
diff --git a/internal/sim/hostlife/witness_test.go b/internal/sim/hostlife/witness_test.go| index d38c2d0bc..14b4bb733 100644 |
| --- a/internal/sim/hostlife/witness_test.go |
| +++ b/internal/sim/hostlife/witness_test.go |
| @@ -69,11 +69,13 @@ var witnesses = []witness{ |
| "a target that wrote during its probe is rolled back; without the return its writes are lost (INV-2)", 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", "INV-1", |
| - "one of three admission fences (J2-R2-8): a paused open from an earlier operation makes the old source ready after the commit; hold_op refuses the open, the exec check refuses the wake, the generation check refuses the start; with all three gone the source serves beside the target (INV-1)", wAdmissionTriple}, |
| - {"hold-op-open", "exec-admission,hold-op-open,start-check-claim-gen", "INV-1", "one of three admission fences (see exec-admission)", wAdmissionTriple}, |
| - {"start-check-claim-gen", "exec-admission,hold-op-open,start-check-claim-gen", "INV-1", "one of three admission fences (see exec-admission)", wAdmissionTriple}, |
| - {"start-check", "exec-admission,hold-op-open,start-check", "INV-1", "the whole start check in place of its generation clause (see exec-admission)", wAdmissionTriple}, |
| + {"exec-admission", "exec-admission,hold-op-open,start-check-claim-gen,exec-writer", "INV-6", |
| + "one of four admission fences (J2-R2-8, claim rev 6.8.1): a paused open from an earlier operation makes the old source ready after the commit; hold_op refuses the open, the exec check refuses the wake, the generation check refuses the start, AdmitWriter refuses a box that is not op.exec; with all four gone the source's wake is a cold admission by a non-exec box, whose installing CAS moves holder.stamp under the target (INV-6), and the source would then serve beside the target (INV-1)", wAdmissionTriple}, |
| + {"hold-op-open", "exec-admission,hold-op-open,start-check-claim-gen,exec-writer", "INV-6", "one of four admission fences (see exec-admission)", wAdmissionTriple}, |
| + {"start-check-claim-gen", "exec-admission,hold-op-open,start-check-claim-gen,exec-writer", "INV-6", "one of four admission fences (see exec-admission)", wAdmissionTriple}, |
| + {"exec-writer", "exec-admission,hold-op-open,start-check-claim-gen,exec-writer", "INV-6", |
| + "a paused open from an earlier operation lands after the commit while the source's row is still held; the source's AdmitWriter CAS requires op.exec to be the source, which it no longer is, and refuses (claim rev 6.8.1); with the three box-side fences gone as well, the non-exec source is admitted: its cold admission moves holder.stamp under the target (INV-6) on the way to two boxes running the host (INV-1)", wAdmissionTriple}, |
| + {"start-check", "exec-admission,hold-op-open,start-check,exec-writer", "INV-6", "the whole start check in place of its generation clause (see exec-admission)", wAdmissionTriple}, |
| {"strict-spawn", "strict-spawn,evidence-whole-tree", "INV-1", |
| "redundant alone: an escaped descendant is killed by the whole-tree cleanup; with evidence blind to escapees, a non-strict descendant survives the cut and writes beside the target (INV-1)", wStrictSpawn}, |
| {"touched-incomplete", "touched-incomplete,legacy-discovery", "freed with residue", |
| @@ -129,6 +131,19 @@ var witnesses = []witness{ |
| "an adopted legacy host runs a descendant outside the cgroup; the move's stop evidence must cover the whole tree, or the escaped descendant survives the cut and two boxes run the host (§6.1, J-D3)", wEvidenceWholeTree}, |
| {"timer-freeze", "timer-freeze", "stuck", |
| "the source has a registered timer; the move stops and holds it; the runner's refused deliveries while held must not advance its bookkeeping, or a clean rollback's digest check finds the timers changed and the source is stuck (J-N14)", wTimerFreeze}, |
| + // rev 7.9 |
| + {"writer-session", "writer-session", "INV-2", |
| + "a dead session's certificate, with a max_txid that still matches the remote maximum because the next writer's upload is in flight: the session moved forward when that writer was admitted, so the certificate authorizes nothing; taken by stamp and TXID alone it readmits a writer into the generation and the late upload replaces that writer's file (claim rev 6.8 §6.4a, R8-1)", wDeadSessionCertificate}, |
| + {"admit-everywhere", "admit-everywhere", "INV-2", |
| + "a committed move's target serves while the claim is still landing; it dies with its disk and an upload in flight; the next wake must be an AdmitWriter within the landing, a cold one into a fresh generation; gated for ordinary wakes only, the target continues the generation and the late upload replaces its file (claim rev 6.8 §6.4a, R8-2)", wCommittedLandingCrash}, |
| + {"admit-everywhere", "admit-everywhere", "INV-2", |
| + "a landing's probe dies with its disk and an upload in flight; the restarted probe must be a cold AdmitWriter within the landing, into a fresh install generation; gated for ordinary wakes only, the restarted probe reuses the generation and the late upload replaces its file (R8-2)", wProbeRestart}, |
| + {"warm-provenance", "warm-provenance", "INV-2", |
| + "a cold admission finds a warm database of the predecessor generation whose lineage file names another session: a cached copy, behind the remote; without provenance it must restore from the remote; taken as the baseline, the new generation's snapshot lacks a durable write (claim rev 6.8 §6.4a, R8-4)", wWarmLineageMismatch}, |
| + {"valid-ltx", "valid-ltx", "INV-12", |
| + "a created incarnation's first wake: an empty generation, position 0; the cold admission writes no LTX file and the app's first transaction is the first upload; a snapshot [1,0] or an anchor [0,0] is a file the pinned encoder refuses (R8-3)", wEmptyBootstrap}, |
| + {"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}, |
| } |
| @@ -312,7 +327,7 @@ func (w *world) serve(b *box, inc int, app string) { |
| n = w.claim.clone() |
| n.Ver++ |
| } |
| - n.Holder = &holder{Inc: inc, App: app, Serial: w.serial, Location: loc, Stamp: n.Ver} // the serving CAS wrote holder.stamp |
| + n.Holder = &holder{Inc: inc, App: app, Serial: w.serial, Location: loc, Stamp: n.Ver, Session: n.Ver} // the serving CAS wrote holder.stamp; one writer session so far |
| n.Op = opState{Kind: "serving", Exec: b.ID, Parts: []int{b.ID}} |
| n.Regions, n.Boxes, n.Complete, n.Obl = []int{b.Region}, []int{b.ID}, true, nil |
| w.claim = n |
| @@ -328,9 +343,9 @@ func (w *world) serve(b *box, inc int, app string) { |
| k := ltxKey(loc, "", 0, bl.Min, bl.Max) |
| w.buckets[b.Region].objs[k] = append(w.buckets[b.Region].objs[k], &version{ID: w.newVer(), B: bl}) |
| w.markDurable(k, bl) |
| - // the generation is intact: its last writer closed cleanly at position |
| - // 1 under this stamp (claim rev 6.6 §6.4a) |
| - w.buckets[b.Region].objs[closedKey(loc, "")] = []*version{{ID: w.newVer(), B: &blob{Kind: "closed", Inc: inc, Stamp: n.Ver, TX: 1}}} |
| + // the generation is intact: its last writer session closed cleanly at |
| + // position 1 on this box (claim rev 6.8 §6.4a) |
| + w.buckets[b.Region].objs[closedKey(loc, "", n.Ver)] = []*version{{ID: w.newVer(), B: &blob{Kind: "closed", Inc: inc, Stamp: n.Ver, Seq: b.ID, Session: n.Ver, TX: 1}}} |
| w.logf("witness: inc %d of app %s serving on box %d", inc, app, b.ID) |
| } |
| @@ -357,8 +372,9 @@ func (w *world) legacy(b *box, other int) { |
| k := ltxKey(b.DB.Loc, "", 0, 1, 1) |
| w.buckets[b.Region].objs[k] = []*version{{ID: w.newVer(), B: full}} |
| w.markDurable(k, full) |
| - // the pre-release writer closed cleanly; adoption's stamp is 1 |
| - w.buckets[b.Region].objs[closedKey(b.DB.Loc, "")] = []*version{{ID: w.newVer(), B: &blob{Kind: "closed", Inc: inc, Stamp: 1, TX: 1}}} |
| + // the pre-release writer closed cleanly on this box; adoption's stamp |
| + // and session are 1 |
| + w.buckets[b.Region].objs[closedKey(b.DB.Loc, "", 1)] = []*version{{ID: w.newVer(), B: &blob{Kind: "closed", Inc: inc, Stamp: 1, Seq: b.ID, Session: 1, TX: 1}}} |
| first := w.txs[b.DB.Log[0]] |
| part := &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: 0, SealKey: key, Pages: map[string]int{first.Key: first.Val}, TXs: []int{first.W}} |
| w.buckets[other].objs[ltxKey(b.DB.Loc, "", 0, 1, 1)] = []*version{{ID: w.newVer(), B: part}} |
| @@ -1484,8 +1500,8 @@ func wCrashBeforeMarker(t *testing.T, w *world) { |
| w.hold = nil |
| w.completeHeld(key) // the upload returns |
| w.crash(b) // before closeStep ran again: no closed.json |
| - if on("recovery-generation") && w.buckets[b.Region].current(closedKey(w.claim.Holder.Location, gen)) != nil { |
| - t.Fatalf("witness: a marker exists after the crash") |
| + if on("recovery-generation") && w.buckets[b.Region].current(closedKey(w.claim.Holder.Location, gen, w.claim.Holder.Session)) != nil { |
| + t.Fatalf("witness: a certificate exists after the crash") |
| } |
| w.recovered(t, b, "running after the crash") |
| if w.claim.Holder.Gen == gen { |
| @@ -1503,10 +1519,12 @@ func wStaleMarker(t *testing.T, w *world) { |
| w.wake(b) // consumes the fixture's marker as it opens |
| loc := w.claim.Holder.Location |
| key := w.deadWriterContinues(b, loc) // [2,2] in flight; the remote maximum is 1 |
| - // a stale marker: a late marker PUT of an earlier close that said 0, |
| - // behind the remote maximum, beside the dead writer's outstanding upload |
| - w.buckets[b.Region].objs[closedKey(loc, "")] = append(w.buckets[b.Region].objs[closedKey(loc, "")], |
| - &version{ID: w.newVer(), B: &blob{Kind: "closed", Inc: 1, Stamp: b.stampOf(1), TX: 0}}) |
| + // a stale certificate for the current session: a late PUT of a close |
| + // that said 0, behind the remote maximum, beside the dead writer's |
| + // outstanding upload |
| + sk := closedKey(loc, "", w.claim.Holder.Session) |
| + w.buckets[b.Region].objs[sk] = append(w.buckets[b.Region].objs[sk], |
| + &version{ID: w.newVer(), B: &blob{Kind: "closed", Inc: 1, Stamp: b.stampOf(1), Seq: b.ID, Session: w.claim.Holder.Session, TX: 0}}) |
| w.recovered(t, b, "B running after A's death") |
| w.appWrite(b) |
| w.replicate(b) |
| @@ -1541,10 +1559,11 @@ func wCloseWaits(t *testing.T, w *world) { |
| key := ltxKey(loc, "", 0, b.DB.Pos+1, b.DB.Pos+1) |
| w.holdKey(key) |
| w.idleStop(b) |
| - for i := 0; i < 20; i++ { |
| + sk := closedKey(loc, "", b.DB.Session) |
| + for i := 0; i < uploadTimeout-2; i++ { // short of the client's timeout: a retry would make this session uncertain for good (R8-5) |
| w.tick(func() { w.replicate(b) }) |
| - if w.buckets[b.Region].current(closedKey(loc, "")) != nil { |
| - w.violate("INV-2 (witness): box %d wrote closed.json before every upload returned (%s is in flight)", b.ID, key) |
| + if w.buckets[b.Region].current(sk) != nil { |
| + w.violate("INV-2 (witness): box %d wrote its certificate before every upload returned (%s is in flight)", b.ID, key) |
| return |
| } |
| } |
| @@ -1553,8 +1572,8 @@ func wCloseWaits(t *testing.T, w *world) { |
| t.Fatalf("witness: the upload was not held") |
| } |
| w.tick(func() { w.replicate(b) }) |
| - if w.buckets[b.Region].current(closedKey(loc, "")) == nil { |
| - t.Fatalf("witness: no closed.json after the uploads returned") |
| + if w.buckets[b.Region].current(sk) == nil { |
| + t.Fatalf("witness: no certificate after the uploads returned") |
| } |
| // the writer continues the generation: no upload of its previous run |
| // can be outstanding, every one returned before the marker |
| @@ -1722,3 +1741,173 @@ func wTimerFreeze(t *testing.T, w *world) { |
| w.rollback(w.claim, "settle check failed") // clean: the target wrote nothing |
| w.finish(t) |
| } |
| + |
| +// ---------- rev 7.9 ---------- |
| + |
| +// wDeadSessionCertificate (claim rev 6.8 §6.4a, R8-1): a certificate of a |
| +// dead session whose max_txid still matches the remote maximum. |
| +func wDeadSessionCertificate(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) // warm: session 2 |
| + loc := w.claim.Holder.Location |
| + w.idleStop(b) // session 2 certifies {max 1}, the generation's position |
| + w.recovered(t, b, "warm continuation") // session 3; the certificates of 1 and 2 are dead |
| + key := w.deadWriterContinues(b, loc) // [2,2] in flight; A dies with its disk |
| + // the dead certificates are still there, never deleted, and session 2's |
| + // max_txid 1 equals the remote maximum, since [2,2] has not landed; |
| + // holder.session is 3, for which no certificate exists |
| + w.recovered(t, b, "B running after A's death") |
| + w.appWrite(b) |
| + w.replicate(b) // B's [2,2] |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: A's upload was not held") |
| + } |
| + w.checkDurable(w.claim, "after A's late upload landed") // two level-0 files: nothing compacted or snapshotted over A's |
| + w.finish(t) |
| +} |
| + |
| +// wCommittedLandingCrash (R8-2): the committed move's target dies with its |
| +// disk and an upload in flight while the claim is still landing. |
| +func wCommittedLandingCrash(t *testing.T, w *world) { |
| + src, tgt := w.boxes[0], w.boxes[1] |
| + w.serve(src, 1, app0) |
| + w.moveTo(tgt.Region) |
| + w.reach(t, 600, "committed and serving on the target", func() bool { |
| + return w.claim.Op.Kind == "landing" && w.claim.Op.Committed && hasMain(tgt) |
| + }, w.act(src, tgt), w.recoveryPass) |
| + if w.claim.Op.Kind != "landing" { |
| + t.Fatalf("witness: the move finalized before the target could crash: %s", w.state()) |
| + } |
| + loc := w.claim.Holder.Location |
| + key := w.deadWriterContinues(tgt, loc) // the target's upload in flight; it dies with its disk |
| + w.recovered(t, tgt, "target running again while the claim is still landing") |
| + w.appWrite(tgt) |
| + w.replicate(tgt) |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: the target's upload was not held") |
| + } |
| + w.checkDurable(w.claim, "after the dead target's late upload landed") |
| + w.finish(t) |
| +} |
| + |
| +// wProbeRestart (R8-2): the landing's probe dies with its disk and an |
| +// upload in flight; the landing restarts its probe. |
| +func wProbeRestart(t *testing.T, w *world) { |
| + src, tgt := w.boxes[0], w.boxes[1] |
| + w.serve(src, 1, app0) |
| + w.landProbed(t, src, tgt) |
| + loc := w.claim.Holder.Location |
| + w.appWrite(tgt) // the probe writes |
| + key := ltxKey(loc, tgt.DB.Gen, 0, tgt.DB.Pos+1, tgt.DB.Pos+1) |
| + w.holdKey(key) |
| + w.replicate(tgt) |
| + w.hold = nil |
| + w.crash(tgt) |
| + w.loseDisk(tgt) |
| + w.reach(t, 400, "probe restarted", func() bool { return hasMain(tgt) }, w.act(src, tgt), w.recoveryPass) |
| + if !hasMain(tgt) { |
| + t.Fatalf("witness: the probe did not restart: %s", w.state()) |
| + } |
| + wr := w.appWrite(tgt) |
| + w.replicate(tgt) |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: the dead probe's upload was not held") |
| + } |
| + // the landing's install generation, as its planner would restore it now |
| + gen := w.opGen(&w.claim.Op) |
| + db := w.restoreCalc(tgt.Region, 1, loc, gen) |
| + if db == nil || !contains(db.Log, wr) { |
| + w.violate("INV-2 (witness): the landing's generation %q lost the restarted probe's write %d after the dead probe's upload landed", gen, wr) |
| + return |
| + } |
| + w.finish(t) |
| +} |
| + |
| +// wWarmLineageMismatch (claim rev 6.8 §6.4a, R8-4): a warm database of the |
| +// predecessor generation whose lineage names another session. |
| +func wWarmLineageMismatch(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + stale := b.DB.clone() // a cached copy at this position |
| + w.appWrite(b) |
| + w.replicate(b) // one more durable write the copy lacks |
| + w.crash(b) |
| + b.DB = stale |
| + b.lineage[1] = lineageRec{Gen: stale.Gen, Session: 999} // a lineage no session of this claim wrote |
| + w.logf("witness: box %d's database is a stale copy with a foreign lineage", b.ID) |
| + w.recovered(t, b, "running after the crash") |
| + w.finish(t) |
| +} |
| + |
| +// wEmptyBootstrap (R8-3): a created incarnation's first wake. |
| +func wEmptyBootstrap(t *testing.T, w *world) { |
| + w.console = append(w.console, &consolePlan{Kind: "create", Host: "h", App: app0, OpID: w.newID("cr"), Inc: 1, Region: 0}) |
| + b := w.regionBox(0) |
| + w.reach(t, 200, "created", func() bool { |
| + c := w.claim |
| + return c != nil && c.Holder != nil && c.Holder.Inc == 1 && c.Op.Kind == "serving" && b.Row != nil |
| + }, w.runConsole, w.actAll(), w.recoveryPass) |
| + b.Config, b.CfgVer, b.Live, b.LiveInc = 1, w.cfgVer[app0], app0, 1 |
| + // the first wake: a cold admission into an empty generation, which the |
| + // mutation builds with files the encoder refuses (INV-12 at the PUT) |
| + w.reach(t, 120, "first wake", func() bool { return hasMain(b) }, w.act(b), w.recoveryPass, func() { w.wake(b) }) |
| + if len(w.violations) > 0 { |
| + return |
| + } |
| + if !hasMain(b) { |
| + t.Fatalf("witness: the created incarnation did not run: %s", w.state()) |
| + } |
| + loc := w.claim.Holder.Location |
| + gen := w.claim.Holder.Gen |
| + if n := len(w.currentLTX(w.buckets[b.Region], dataKey(loc, gen, ""), 0)) + len(w.currentLTX(w.buckets[b.Region], dataKey(loc, gen, ""), 9)); n != 0 { |
| + t.Fatalf("witness: an empty generation has %d LTX files before the first transaction", n) |
| + } |
| + w.appWrite(b) |
| + w.replicate(b) // the first transaction is the first upload, [1,1] |
| + w.idleStop(b) |
| + w.loseDisk(b) |
| + w.recovered(t, b, "woken cold") |
| + w.finish(t) |
| +} |
| + |
| +// wRetryUnresolved (R8-5): an upload times out, its retry succeeds, the |
| +// original stays unresolved. |
| +func wRetryUnresolved(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + loc := w.claim.Holder.Location |
| + w.appWrite(b) |
| + key := ltxKey(loc, "", 0, b.DB.Pos+1, b.DB.Pos+1) |
| + var original *request |
| + w.hold = func(r *request) bool { |
| + if r.key == key && original == nil { |
| + original = r |
| + return true // the first attempt alone is held |
| + } |
| + return r == original |
| + } |
| + w.replicate(b) |
| + for i := 0; i <= uploadTimeout; i++ { |
| + w.tick(func() { w.replicate(b) }) // the client times out and retries the same bytes; the retry lands |
| + } |
| + if original == nil || w.buckets[b.Region].current(key) == nil { |
| + t.Fatalf("witness: the retry did not land: %s", w.ltxShape(b.Region, dataKey(loc, "", ""))) |
| + } |
| + 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 original upload %s was unresolved (R8-5)", b.ID, key) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + w.completeHeld(key) |
| + w.finish(t) |
| +} |
diff --git a/internal/sim/hostlife/world_test.go b/internal/sim/hostlife/world_test.go| index 2f1d3fafc..5853375d5 100644 |
| --- a/internal/sim/hostlife/world_test.go |
| +++ b/internal/sim/hostlife/world_test.go |
| @@ -91,14 +91,15 @@ type blob struct { |
| Base int |
| SnapKey string |
| TX int |
| - // a "closed" marker (claim rev 6.6 §6.4a): the writer's stamp; TX is |
| - // its max_txid |
| - Stamp int |
| - Files *filesDir |
| - Timers []int |
| - Refs []ref // a park manifest's archive set |
| - Dels []ref // a park manifest's deletion set, delete markers included |
| - Text string |
| + // a "closed" certificate (claim rev 6.8 §6.4a): the writer's stamp and |
| + // session; Seq is its box and TX its max_txid |
| + Stamp int |
| + Session int |
| + Files *filesDir |
| + Timers []int |
| + Refs []ref // a park manifest's archive set |
| + Dels []ref // a park manifest's deletion set, delete markers included |
| + Text string |
| } |
| type ref struct { |
| @@ -276,13 +277,24 @@ type localDB struct { |
| // first sync after a cold initialization, fact R5-1c) |
| FirstFull bool |
| // Closing: the writer stopped cleanly and is draining its uploads |
| - // before writing closed.json (claim rev 6.6 §6.4a); CloseReq is the |
| + // before writing its certificate (claim rev 6.8 §6.4a); CloseReq is the |
| // marker PUT once issued |
| Closing bool |
| CloseReq *request |
| - Log []int // every write applied here, for the oracle |
| - Stamp int // the installation stamp that produced this artifact (claim rev 6.5 §6.3) |
| - Seq int |
| + // Session is the writer session this database was admitted under; Issued |
| + // is every request the writer issued that did not return at once (level |
| + // 0, compaction, snapshot, retention), and Uncertain is sticky once any |
| + // of them timed out and was retried (R8-5): no certificate then. |
| + Session int |
| + Issued []*request |
| + Uncertain bool |
| + // RemotePos is the remote position the pinned replica caches until an |
| + // initialization or an error: the newest remote level-0 TXID it has |
| + // seen or uploaded (R8-5) |
| + RemotePos int |
| + Log []int // every write applied here, for the oracle |
| + Stamp int // the installation stamp that produced this artifact (claim rev 6.5 §6.3) |
| + Seq int |
| } |
| func (d *localDB) clone() *localDB { |
| @@ -347,6 +359,11 @@ 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 |
| + // lineage is the local lineage file per incarnation: the generation and |
| + // session the writer that opened this database was admitted to (claim |
| + // rev 6.8 §6.4a, R8-4). Lost with the disk; never written by a stamp |
| + // re-initialization. |
| + lineage map[int]lineageRec |
| // anchors are the PUTs this box's activator must see acknowledged before |
| // it goes on (an anchor, a recovery's snapshot, a closed.json), by key |
| // (rev 7.7, R6-1): a request the client has waited uploadTimeout steps |
| @@ -547,6 +564,12 @@ type holder struct { |
| // current (moves rev 6 §5.1): "" is origin; a move, a push and a dirty |
| // rollback each install a generation named by their op id. |
| Gen string |
| + // Session is holder.session (claim rev 6.8 §6.4a, R8-1): the gen of the |
| + // AdmitWriter CAS that admitted the current writer session. Every path |
| + // that opens replication goes through that CAS, which moves the |
| + // session forward, so a clean-close certificate (written per session) |
| + // is one-use by construction. |
| + Session int |
| // Stamp is holder.stamp (claim rev 6.5 §2, R5-4): the gen produced by |
| // the CAS that made the current installation current, written by that |
| // same CAS: a create's serving CAS, adoption, a flip, a reinstall, a |
| @@ -566,6 +589,48 @@ type obligation struct { |
| Quiet int // not before |
| } |
| +// writerEntry is one AdmitWriter (claim rev 6.8 §6.4a): the session it |
| +// produced, the box and generation it admitted, the stamp, the mode (warm |
| +// or cold), whether the writer has opened, and the predecessor it observed. |
| +type writerEntry struct { |
| + Session, Box int |
| + Gen string |
| + Stamp int |
| + Mode string |
| + Opened bool |
| + Pred pred |
| +} |
| + |
| +// pred is the predecessor a cold admission journals before the recovery |
| +// stamp activates (claim rev 6.8 §6.4a, R8-4). |
| +type pred struct { |
| + Session, Box int |
| + Gen string |
| + Stamp int |
| +} |
| + |
| +// recState is a cold admission's build in progress (moves rev 7.9 §7.14): |
| +// the new generation, the predecessor generation it restores from, the |
| +// session and box admitted, the phase, the predecessor, and the snapshot |
| +// identity once written. |
| +type recState struct { |
| + Gen, Base string |
| + Session, Box int |
| + Phase string |
| + Pred pred |
| + BaseVer int |
| + BaseKey string |
| + BaseTX int |
| +} |
| + |
| +// lineageRec is the local lineage file a writer writes as it opens (claim |
| +// rev 6.8 §6.4a): the generation and session it was admitted to. A stamp |
| +// re-initialization never writes it (R8-4). |
| +type lineageRec struct { |
| + Gen string |
| + Session int |
| +} |
| + |
| type opState struct { |
| Kind string // serving, creating, deleting, parking, purging, moving, landing, reverting, restoring, pushing |
| ID string |
| @@ -582,37 +647,46 @@ type opState struct { |
| NewSerial int |
| // journal |
| - ReportAt int |
| - ReportOK bool |
| - CutAt int |
| - CutDeadline int |
| - StopAt int |
| - StopDigest digest |
| - StopFiles *filesDir |
| - SealDB ref |
| - SealFiles ref |
| - SealQuiet int |
| - LandPhase string |
| - Installed digest |
| - ProbeTries int |
| - ProbedAt int |
| - RevPhase string |
| - SrcPhase string |
| - Dirty bool |
| - DirtyKnown bool |
| - ReturnDB ref |
| - ReturnFiles ref |
| - ReturnQuiet int |
| - LostBox int |
| - LostAt int |
| - RollbackID string |
| - PushSnap *blob // a push's installed snapshot (the final checkpoint plus the push's mark) |
| - BaseKey string |
| - BaseTX int |
| - RecVer int // a delete's or purge's intent: the record version it will DeleteIfMatch (claim §5.2 step 1, §5.3 step 4) |
| - FlipGen int // the holder.stamp the flip CAS wrote: the landing's installation stamp (claim rev 6.5 §6.4) |
| - SrcStamp int // the source installation's stamp before the flip, for the return to source (claim rev 6.5) |
| - BaseGen string // a cold recovery's baseline: the generation it restores from (claim rev 6.6 §6.4a) |
| + ReportAt int |
| + ReportOK bool |
| + CutAt int |
| + CutDeadline int |
| + StopAt int |
| + StopDigest digest |
| + StopFiles *filesDir |
| + SealDB ref |
| + SealFiles ref |
| + SealQuiet int |
| + LandPhase string |
| + Installed digest |
| + ProbeTries int |
| + ProbedAt int |
| + RevPhase string |
| + SrcPhase string |
| + Dirty bool |
| + DirtyKnown bool |
| + ReturnDB ref |
| + ReturnFiles ref |
| + ReturnQuiet int |
| + LostBox int |
| + LostAt int |
| + RollbackID string |
| + PushSnap *blob // a push's installed snapshot (the final checkpoint plus the push's mark) |
| + BaseKey string |
| + BaseTX int |
| + RecVer int // a delete's or purge's intent: the record version it will DeleteIfMatch (claim §5.2 step 1, §5.3 step 4) |
| + FlipGen int // the holder.stamp the flip CAS wrote: the landing's installation stamp (claim rev 6.5 §6.4) |
| + SrcStamp int // the source installation's stamp before the flip, for the return to source (claim rev 6.5) |
| + // Writers is op.journal.writers (claim rev 6.8 §6.4a): one entry per |
| + // AdmitWriter CAS. Rec is a cold admission whose build (the new |
| + // generation's snapshot and anchor, generation.json) is still in |
| + // progress; it never replaces the operation (R8-2). InstGen is the |
| + // generation an operation is installing when a cold admission within it |
| + // moved it off the op id; "" means the op id (or the rollback id). |
| + Writers []writerEntry |
| + Rec *recState |
| + InstGen string |
| + Fresh bool // the operation's freshly built generation has had no writer yet: its first probe is admitted cold without another build |
| Committed bool |
| Opened bool |
| ParkLocalNow bool |
| @@ -884,6 +958,11 @@ func (c *claim) clone() *claim { |
| } |
| n.Op.Parts = append([]int(nil), c.Op.Parts...) |
| n.Op.StopFiles = c.Op.StopFiles.clone() |
| + n.Op.Writers = append([]writerEntry(nil), c.Op.Writers...) |
| + if c.Op.Rec != nil { |
| + r := *c.Op.Rec |
| + n.Op.Rec = &r |
| + } |
| n.Regions = append([]int(nil), c.Regions...) |
| n.Boxes = append([]int(nil), c.Boxes...) |
| n.Obl = append([]obligation(nil), c.Obl...) |
| @@ -927,6 +1006,12 @@ func (w *world) iamCheck(actorBox int, r *request) bool { |
| // (the request may still complete later). |
| func (w *world) issue(actorBox int, r *request) int { |
| r.at = w.now |
| + if r.kind == rPut && r.b != nil && isDataKey(r.key) && (r.b.Min == 0 || r.b.Max == 0) { |
| + // INV-12 (R8-3): the pinned LTX encoder refuses a zero TXID, so no |
| + // such file can exist; a position-0 snapshot [1,0] or anchor [0,0] |
| + // is a builder bug, never an object |
| + w.violate("INV-12: box %d wrote an LTX file with a zero TXID, %s, which the pinned encoder refuses (R8-3)", actorBox, r.key) |
| + } |
| if !w.iamCheck(actorBox, r) { |
| w.violate("IAM: box %d may not %v %s in region %d", actorBox, r.kind, r.key, r.region) |
| return 0 |
| @@ -963,9 +1048,13 @@ func (w *world) issue(actorBox int, r *request) int { |
| // bounds acceptance, not completion (claim C3a, J2-R2-1) |
| r.doneAt = w.now + horizonH + 1 + w.rng.Intn(10*horizonH) |
| r.beyond = true |
| - if w.fail["recreate"] && w.rng.Intn(2) == 0 { |
| + if w.fail["recreate"] && !w.calm && w.rng.Intn(2) == 0 { |
| // the biased schedule: no bound at all, so it can outlive a |
| - // delete and a recreate; it lands at its worst moment |
| + // delete and a recreate; it lands at its worst moment. Chaos |
| + // only: in the calm phase a chain of such holds, one per park |
| + // of a dead generation, can outlast the whole convergence |
| + // window (seed 20077 / 30,000), which says nothing about the |
| + // protocol |
| r.doneAt = w.now + 40*horizonH |
| } |
| } |
diff --git a/internal/sim/litestreamfacts/facts_test.go b/internal/sim/litestreamfacts/facts_test.go| index 6ec2b3f40..2f76c6e5c 100644 |
| --- a/internal/sim/litestreamfacts/facts_test.go |
| +++ b/internal/sim/litestreamfacts/facts_test.go |
| @@ -910,3 +910,62 @@ func TestFact_R7_1_UnconditionalPut(t *testing.T) { |
| } |
| t.Logf("FACT R7-1b: the key is c.Path + \"/\" + %%04x level + ltx.FormatFilename(min, max): two writers of one generation share it") |
| } |
| + |
| +// TestFact_R8_3_EmptyBootstrap: a created incarnation's first wake has |
| +// neither a database file nor a replica (R8-3). The pinned encoder refuses |
| +// the files a position-0 build would carry, a snapshot [1,0] and an anchor |
| +// [0,0], so an empty generation has no LTX file at all: litestream opened |
| +// on an absent path against an empty replica syncs nothing, a cold restore |
| +// of the empty generation has nothing to restore, and the first |
| +// transaction is the first upload, which a cold restore then holds. |
| +func TestFact_R8_3_EmptyBootstrap(t *testing.T) { |
| + for _, h := range []ltx.Header{ |
| + {Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 1, MaxTXID: 0, Timestamp: 1, WALOffset: litestream.WALHeaderSize}, |
| + {Version: ltx.Version, Flags: ltx.HeaderFlagNoChecksum, PageSize: 4096, Commit: 1, MinTXID: 0, MaxTXID: 0, Timestamp: 1}, |
| + } { |
| + var buf bytes.Buffer |
| + enc, err := ltx.NewEncoder(&buf) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + err = enc.EncodeHeader(h) |
| + t.Logf("FACT R8-3a: encoding an LTX header [%d,%d]: %v", h.MinTXID, h.MaxTXID, err) |
| + if err == nil { |
| + t.Fatalf("the encoder accepted a zero TXID [%d,%d]; the spec's empty-bootstrap rule rests on it refusing", h.MinTXID, h.MaxTXID) |
| + } |
| + } |
| + dir := t.TempDir() |
| + gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| + path := filepath.Join(dir, "new.db") |
| + meta := filepath.Join(dir, "meta") |
| + if _, err := os.Stat(path); !os.IsNotExist(err) { |
| + t.Fatalf("the database path exists before the first wake: %v", err) |
| + } |
| + if err := restore(t, gen, filepath.Join(dir, "nothing.db")); err == nil { |
| + t.Fatalf("a restore from an empty replica succeeded; the empty generation must have nothing to restore") |
| + } else { |
| + t.Logf("FACT R8-3b: a cold restore of an empty generation: %v", err) |
| + } |
| + db := openDB(t, path, meta, gen) |
| + err := db.Sync(context.Background()) |
| + t.Logf("FACT R8-3c: litestream opened on an absent database against an empty replica; first sync: %v; level 0 %s, level 9 %s", err, ranges(files(t, gen, 0)), ranges(files(t, gen, 9))) |
| + if err != nil { |
| + t.Fatalf("the first sync of an absent database failed: %v", err) |
| + } |
| + mustRanges(t, gen, 0, "none") |
| + mustRanges(t, gen, 9, "none") |
| + write(t, path, 1) // CREATE TABLE, then one row: two SQLite transactions, one sync |
| + sync(t, db) |
| + l0 := ranges(files(t, gen, 0)) |
| + t.Logf("FACT R8-3d: after the first transactions and one sync the replica's level 0 is %s, level 9 %s (a TXID is one sync's WAL frames, not one SQLite transaction)", l0, ranges(files(t, gen, 9))) |
| + if l0 != "[1,1]" { |
| + t.Fatalf("expected the first upload to be level 0 [1,1], got %s", l0) |
| + } |
| + closeDB(t, db) |
| + out := filepath.Join(dir, "out.db") |
| + if err := restore(t, gen, out); err != nil { |
| + t.Fatalf("restore after the first transactions: %v", err) |
| + } |
| + mustCount(t, out, 1) |
| + t.Logf("FACT R8-3e: a cold restore after the first transactions holds the row; the first transaction was the first upload") |
| +} |