amadan
sim: hostlife: writer epochs (claim rev 6.6), identity oracle, newest level-0, build completion (rev 7.8)
R7-1 (claim rev 6.6 §6.4a, b10547e1): a clean stop drains its uploads,
then writes closed.json {stamp, max_txid} once (If-None-Match); a wake
continues holder.generation only if the marker exists, its max_txid is
the generation's remote maximum and its stamp is the box's active stamp,
and the writer consumes the marker as it opens; anything else is lost
replication state and a cold recovery: an installing CAS writes a fresh
holder.generation and holder.stamp, the box builds the new generation
(snapshot, anchor and the installer's closed.json, each acknowledged)
from the warm database or the old generation restored, publishes it and
replicates only into it; the old generation is parked as a chore
(park-generation) after the dead writer's horizon. The upload loop skips
a local file at or below the remote position as litestream does (the
initialization/remote-position race), and the first upload after a cold
initialization carries every page. Rules recovery-generation,
closed-marker, close-waits; witnesses for both completion orders, the
race, the R7-2 counterexample, a crash before the marker, a stale marker
and a close that must wait.
R7-2: the cold-restore oracle requires every durable write by identity
in the restored files' log and compares the state with the replay of
exactly the writes those files carry; the per-key mask is gone. A full
file (snapshot, anchor, first upload after cold init, their compaction)
replaces the state on apply and carries every write it holds.
R7-3: every retention path keeps the newest level-0 file (rule
newest-l0, witness with the local position ahead of the uploads).
R7-4: the builds of a landing, a push and a dirty rollback journal the
snapshot, then the anchor, each acknowledged, before publishing
generation.json; INV-11 refuses a marker naming artifacts that have not
landed (rule build-completion, witness with crashes between the steps).
The door witnesses are retired: a recreated incarnation's first wake is a
cold recovery into a fresh generation, which a listing scoped to its
authorized generation never selects.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
9 files changed,
+928
−110
internal/sim/hostlife/background_test.go+4 −1internal/sim/hostlife/expected-outcomes.txt+2 −2internal/sim/hostlife/ltx_test.go+102 −10internal/sim/hostlife/moves_test.go+191 −7internal/sim/hostlife/protocol_test.go+88 −20internal/sim/hostlife/rules_test.go+5 −0internal/sim/hostlife/sim_test.go+60 −31internal/sim/hostlife/witness_test.go+413 −24internal/sim/hostlife/world_test.go+63 −15
diff --git a/internal/sim/hostlife/background_test.go b/internal/sim/hostlife/background_test.go| index f345765d4..c5f5555ca 100644 |
| --- a/internal/sim/hostlife/background_test.go |
| +++ b/internal/sim/hostlife/background_test.go |
| @@ -45,6 +45,9 @@ func (w *world) idleStop(b *box) { |
| } |
| } |
| b.Procs = keep |
| + // the clean close (claim rev 6.6 §6.4a): drain, then closed.json |
| + b.DB.Closing = true |
| + w.closeStep(b) |
| } |
| // wake: a request reaches this box's edge for the host. |
| @@ -193,7 +196,7 @@ func (w *world) reconcileClaimed(b *box, c *claim) { |
| return |
| } |
| switch c.Op.Kind { |
| - case "serving", "creating", "pushing": |
| + case "serving", "creating", "pushing", "recovering": |
| 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) |
diff --git a/internal/sim/hostlife/expected-outcomes.txt b/internal/sim/hostlife/expected-outcomes.txt| index 6cc69ce95..1b79c19bd 100644 |
| --- a/internal/sim/hostlife/expected-outcomes.txt |
| +++ b/internal/sim/hostlife/expected-outcomes.txt |
| @@ -14,6 +14,6 @@ start-check-claim-gen missed one of three admission fences; caught with exec-adm |
| start-check missed caught with exec-admission and hold-op-open |
| strict-spawn missed caught with evidence-whole-tree |
| touched-incomplete missed caught with legacy-discovery |
| -door-reread missed subsumed by C3a: caught with location-per-incarnation |
| -door-by-version missed subsumed by C3a: caught with location-per-incarnation (witness), and with door-reread as well (sweep) |
| +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 |
diff --git a/internal/sim/hostlife/ltx_test.go b/internal/sim/hostlife/ltx_test.go| index ae6fb2d50..08c5d76a4 100644 |
| --- a/internal/sim/hostlife/ltx_test.go |
| +++ b/internal/sim/hostlife/ltx_test.go |
| @@ -199,7 +199,7 @@ func isLevelDir(child string) bool { |
| // writes it holds. The caller ships pending writes first, as litestream |
| // syncs before it snapshots. |
| 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, |
| + return &blob{Kind: "snap", Level: 9, Min: 1, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| } |
| @@ -208,15 +208,22 @@ func snapshotOf(db *localDB, inc int) *blob { |
| // first sync, a full-page file [1, 1], as litestream writes it. |
| func (w *world) deltaOf(db *localDB, inc int) *blob { |
| pages := map[string]int{} |
| - if db.Pos == 0 { |
| + full := db.Pos == 0 || db.FirstFull |
| + if full { |
| pages = cloneState(db.State) |
| } |
| for _, wr := range db.Pending { |
| t := w.txs[wr] |
| pages[t.Key] = db.State[t.Key] // the image now: 0 when deleted since |
| } |
| - return &blob{Kind: "ltx", Level: 0, Min: db.Pos + 1, Max: db.Pos + 1, Inc: inc, Gen: db.Gen, SealKey: db.Key, |
| - Pages: pages, TXs: append([]int(nil), db.Pending...)} |
| + // the oracle's list of the writes the file carries: a full file carries |
| + // the effect of every write applied here, not only the pending ones |
| + txs := db.Pending |
| + if full { |
| + txs = db.Log |
| + } |
| + return &blob{Kind: "ltx", Level: 0, Min: db.Pos + 1, Max: db.Pos + 1, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: full, |
| + Pages: pages, TXs: append([]int(nil), txs...)} |
| } |
| // anchorOf is the level-0 anchor a cold restore writes before replication |
| @@ -224,12 +231,20 @@ func (w *world) deltaOf(db *localDB, inc int) *blob { |
| // every page of the restored database as the file [cur, cur], which |
| // litestream's open takes as its position. |
| func anchorOf(db *localDB, inc int) *blob { |
| - return &blob{Kind: "ltx", Level: 0, Min: db.Pos, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, |
| + return &blob{Kind: "ltx", Level: 0, Min: db.Pos, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, Full: true, |
| Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| } |
| -// apply overlays an LTX file's page images onto state. |
| +// apply overlays an LTX file's page images onto state; a full file replaces |
| +// it, so a key the full file does not carry is gone. |
| func apply(state map[string]int, f *blob) { |
| + if f.Full { |
| + for k := range state { |
| + if _, ok := f.Pages[k]; !ok { |
| + delete(state, k) |
| + } |
| + } |
| + } |
| for k, v := range f.Pages { |
| if v == 0 { |
| delete(state, k) |
| @@ -245,7 +260,11 @@ func apply(state map[string]int, f *blob) { |
| // compaction, snapshot rotation and retention run as the store's |
| // monitors would. |
| func (w *world) replicate(b *box) { |
| - if !w.agentUp(b) || !hasMain(b) || b.DB == nil { |
| + if !w.agentUp(b) || b.DB == nil { |
| + return |
| + } |
| + if !hasMain(b) { |
| + w.closeStep(b) |
| return |
| } |
| db := b.DB |
| @@ -253,6 +272,7 @@ func (w *world) replicate(b *box) { |
| // the sync writes this TXID's file locally; the position advances |
| // at once, and later writes go to the next TXID |
| bl := w.deltaOf(db, db.Inc) |
| + db.FirstFull = false |
| db.Queue = append(db.Queue, bl) |
| db.Pos, db.Pending = bl.Max, nil |
| } |
| @@ -260,6 +280,43 @@ 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") } |
| + |
| +// 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. |
| +func (w *world) closeStep(b *box) { |
| + db := b.DB |
| + if db == nil || !db.Closing || hasMain(b) { |
| + return |
| + } |
| + // the marker is the live generation's: once an operation has taken the |
| + // writer over (a push, a move, a delete), or the claim has moved on, |
| + // the generation is dead or being replaced and the close is dropped |
| + if c := w.claim; c == nil || c.Holder == nil || c.Op.Kind != "serving" || c.Op.Exec != b.ID || c.Holder.Gen != db.Gen || c.Holder.Inc != db.Inc { |
| + db.Closing, db.CloseReq = false, nil |
| + return |
| + } |
| + w.ship(b) |
| + drained := len(db.Queue) == 0 && (db.InFlight == nil || db.InFlight.done) |
| + if !drained && on("close-waits") { |
| + 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}} |
| + w.issue(b.ID, db.CloseReq) |
| + w.stats["clean closes"]++ |
| + } |
| + if db.CloseReq.done { |
| + db.Closing, db.CloseReq = false, nil |
| + } |
| +} |
| + |
| // uploadTimeout is the client's patience with one level-0 upload, in |
| // steps, before litestream retries it. |
| const uploadTimeout = 8 |
| @@ -288,6 +345,18 @@ func (w *world) ship(b *box) { |
| } |
| } |
| 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 { |
| + w.markDurable(ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max), bl) |
| + w.stats["uploads skipped at the remote position"]++ |
| + db.InFlight = nil |
| + db.Queue = db.Queue[1:] |
| + continue |
| + } |
| 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.InFlight = nil |
| @@ -333,7 +402,7 @@ func (w *world) compact(b *box) { |
| if len(inputs) < 3 { |
| return |
| } |
| - out := &blob{Kind: "ltx", Level: 1, Min: inputs[0].B.Min, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}} |
| + out := &blob{Kind: "ltx", Level: 1, Min: inputs[0].B.Min, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}, Full: inputs[0].B.Full} |
| max := 0 |
| var keys []string |
| for _, v := range inputs { |
| @@ -358,8 +427,14 @@ func (w *world) compact(b *box) { |
| } |
| // the inputs are deleted bar the newest: litestream's level-0 retention |
| // never removes the newest level-0 file, which is where a fresh box |
| - // reads its position (facts R5-1f to h; R6 review) |
| - for _, k := range keys[:len(keys)-1] { |
| + // reads its position (facts R5-1f to h; R6 review). The mutation of |
| + // newest-l0 is rev 7.6's: every input, and below every L0 below the |
| + // previous snapshot |
| + keep := keys[len(keys)-1] |
| + for _, k := range keys { |
| + if k == keep && on("newest-l0") { |
| + continue |
| + } |
| w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: k}) |
| } |
| w.stats["ltx compactions"]++ |
| @@ -384,8 +459,12 @@ func (w *world) compact(b *box) { |
| w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: snapKey(db.Loc, db.Gen, s.B.Max)}) |
| } |
| } |
| + newest := w.maxL0(bk, prefix) |
| for _, level := range []int{1, 0} { |
| for _, f := range w.currentLTX(bk, prefix, level) { |
| + if level == 0 && f.B.Max == newest && on("newest-l0") { |
| + 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)}) |
| } |
| @@ -393,6 +472,19 @@ func (w *world) compact(b *box) { |
| } |
| } |
| +// remoteMax is the generation's remote maximum TXID over every level. |
| +func (w *world) remoteMax(bk *bucket, prefix string) int { |
| + max := 0 |
| + for _, level := range []int{0, 1, 9} { |
| + for _, f := range w.currentLTX(bk, prefix, level) { |
| + if f.B.Max > max { |
| + max = f.B.Max |
| + } |
| + } |
| + } |
| + return max |
| +} |
| + |
| // plan is litestream's restore planner (replica.go CalcRestorePlan, |
| // v0.5.14) over the current files under prefix, from TXID cur: take the |
| // file that extends the range, overlap allowed (ltx.IsContiguous: |
diff --git a/internal/sim/hostlife/moves_test.go b/internal/sim/hostlife/moves_test.go| index 23e157b1a..6782bf8d7 100644 |
| --- a/internal/sim/hostlife/moves_test.go |
| +++ b/internal/sim/hostlife/moves_test.go |
| @@ -189,6 +189,10 @@ 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) |
| @@ -852,13 +856,23 @@ func (w *world) landTarget(b *box, c *claim) { |
| } |
| key := snapKey(loc, op.ID, 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 { |
| - // the install manifest records the snapshot's key, version and TXID (R3-1, R4-1) |
| + if v > 0 || !on("build-completion") { |
| + // the build journal records the snapshot's key, version and TXID |
| + // (R3-1, R4-1); the anchor follows, then completion (R7-4). The |
| + // mutation publishes from the intent, before the artifacts landed |
| + next := "install-anchor" |
| + if !on("build-completion") { |
| + next = "published" |
| + } |
| w.cas(c, func(n *claim) bool { |
| - n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", v, key, sv.B.Max |
| + n.Op.LandPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| } |
| + case "install-anchor": |
| + if w.anchorAcked(b, loc, op.ID, 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) |
| if v > 0 { |
| @@ -1063,6 +1077,8 @@ 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"]++ |
| } |
| @@ -1224,12 +1240,20 @@ func (w *world) recoverSource(b *box, c *claim) { |
| } |
| key := snapKey(c.Holder.Location, op.RollbackID, 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 { |
| + if v > 0 || !on("build-completion") { |
| + next := "install-anchor" |
| + if !on("build-completion") { |
| + next = "published" |
| + } |
| w.cas(c, func(n *claim) bool { |
| - n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", v, key, sv.B.Max |
| + n.Op.SrcPhase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, sv.B.Max |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| } |
| + case "install-anchor": |
| + if w.anchorAcked(b, c.Holder.Location, op.RollbackID, 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) |
| if v > 0 { |
| @@ -1322,12 +1346,20 @@ func (w *world) pushStep(b *box, c *claim) { |
| } |
| key := snapKey(loc, op.ID, snap.Max) |
| v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: snap}) |
| - if v > 0 { |
| + if v > 0 || !on("build-completion") { |
| + next := "install-anchor" |
| + if !on("build-completion") { |
| + next = "published" |
| + } |
| w.cas(c, func(n *claim) bool { |
| - n.Op.Phase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", v, key, snap.Max |
| + n.Op.Phase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = next, v, key, snap.Max |
| return sameOp(n.Op.ID, op.ID) |
| }) |
| } |
| + case "install-anchor": |
| + if w.anchorAcked(b, loc, op.ID, 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) |
| if v > 0 { |
| @@ -1510,9 +1542,33 @@ 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 |
| + } |
| return isDataKey(k) |
| } |
| +// genOf is the generation within loc that key k belongs to: "" for the |
| +// origin layout (level directories and closed.json directly under loc), |
| +// the op id for g-<op>/..., and "-" for a key of no generation (markers, |
| +// the lock). |
| +func genOf(loc, k string) string { |
| + rest := strings.TrimPrefix(k, loc) |
| + if strings.HasPrefix(rest, "g-") { |
| + if i := strings.Index(rest, "/"); i > 0 { |
| + return rest[2:i] |
| + } |
| + return "-" |
| + } |
| + if isDataKey(k) && genPrefixOf(k) == loc || rest == "closed.json" { |
| + return "" |
| + } |
| + return "-" |
| +} |
| + |
| // legacyRetired: the host once held a legacy incarnation, now retired, so |
| // the bare namespace outside holder.location is a retired location. |
| func (w *world) legacyRetired(c *claim) bool { |
| @@ -1531,6 +1587,16 @@ func (w *world) legacyRetired(c *claim) bool { |
| // generation back, and every wake would refuse. |
| func (w *world) publishGen(b *box, loc, gen string, inc int, base int, key string, tx int) int { |
| // the marker names the generation's snapshot by key, version id and TXID (R3-1, R4-1) |
| + // INV-11 (R7-4): generation.json is published only from a completed |
| + // build: the snapshot version it names exists and the anchor at its |
| + // 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) |
| + } |
| 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 { |
| // the biased schedule (rev 7.5): this publication is accepted and |
| @@ -1545,3 +1611,121 @@ func (w *world) publishGen(b *box, loc, gen string, inc int, base int, key strin |
| } |
| return w.issue(b.ID, r) |
| } |
| + |
| +// anchorAcked is §5.1 step 7 inside a build (R6-1, R7-4): the anchor |
| +// [tx, tx], every page of the snapshot the build installed, is PUT into the |
| +// generation and the build goes on only once the PUT is acknowledged. The |
| +// mutation of restore-anchor writes none; the mutation of anchor-acked |
| +// does not wait. |
| +func (w *world) anchorAcked(b *box, loc, gen, snapKey string, snapVer, tx int) bool { |
| + if !on("restore-anchor") || tx == 0 { |
| + return true // no transactions yet: there is no position to anchor |
| + } |
| + sv := w.buckets[b.Region].byID(snapKey, snapVer) |
| + if sv == nil { |
| + return false |
| + } |
| + db := &localDB{Inc: sv.B.Inc, Loc: loc, Gen: gen, Key: sv.B.SealKey, Pos: tx, State: cloneState(sv.B.Pages), Log: append([]int(nil), sv.B.TXs...)} |
| + acked := w.ackedPut(b, ltxKey(loc, gen, 0, tx, tx), anchorOf(db, sv.B.Inc), false) |
| + 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": |
| + var base *localDB |
| + if b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == op.BaseGen { |
| + base = b.DB |
| + } else if b.DB != nil && b.DB.Inc == h.Inc && b.DB.Gen == op.ID { |
| + base = b.DB // a replayed build after a crash: the restored file is there |
| + } 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 !w.ackedPut(b, key, snap, false) { |
| + return |
| + } |
| + sv := w.buckets[b.Region].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 8d8c9bbc8..61c0cd2b4 100644 |
| --- a/internal/sim/hostlife/protocol_test.go |
| +++ b/internal/sim/hostlife/protocol_test.go |
| @@ -288,26 +288,8 @@ func (w *world) restoreInit(b *box, inc int, loc, gen string) *localDB { |
| remote := w.maxL0(bk, prefix) |
| if cur > remote && on("restore-anchor") { |
| key := ltxKey(loc, gen, 0, cur, cur) |
| - acked, inflight := false, false |
| - for _, r := range b.anchors[key] { |
| - if r.done { |
| - acked = true |
| - } else if w.now-r.at < uploadTimeout { |
| - inflight = true |
| - } |
| - } |
| - if !acked && !inflight { |
| - r := &request{kind: rPut, region: b.Region, key: key, b: anchorOf(db, inc)} |
| - if v := w.issue(b.ID, r); v > 0 { |
| - acked = true |
| - } else { |
| - if b.anchors == nil { |
| - b.anchors = map[string][]*request{} |
| - } |
| - b.anchors[key] = append(b.anchors[key], r) |
| - } |
| - w.stats["restore anchors"]++ |
| - } |
| + acked := w.ackedPut(b, key, anchorOf(db, inc), false) |
| + w.stats["restore anchors"]++ |
| if !acked && on("anchor-acked") { |
| w.logf("box %d waits for its anchor %s to be acknowledged before opening replication", b.ID, key) |
| w.stats["anchor waits"]++ |
| @@ -315,6 +297,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.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 |
| @@ -352,6 +335,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 |
| if b.Old { |
| // an old executable: no claim, no markers, no locations; it |
| // restores whatever is newest at the legacy root |
| @@ -392,6 +376,23 @@ 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) |
| + return false |
| + } |
| + consume = m |
| + } |
| if b.DB == nil { |
| b.DB = w.restoreInit(b, h.Inc, loc, want) |
| if b.DB == nil { |
| @@ -450,6 +451,11 @@ 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}) |
| + } |
| 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}) |
| if w.rng.Intn(2) == 0 { |
| @@ -507,10 +513,67 @@ 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 |
| + } |
| + h := c.Holder |
| + bk := w.buckets[b.Region] |
| + prefix := dataKey(h.Location, h.Gen, "") |
| + m := bk.current(closedKey(h.Location, h.Gen)) |
| + if m == nil { |
| + w.logf("box %d: no closed.json for %s: lost replication state", b.ID, prefix) |
| + return "lost", nil |
| + } |
| + if !on("closed-marker") { |
| + return "intact", 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 "intact", m |
| +} |
| + |
| +// 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) { |
| + id := w.newID("rec") |
| + if w.cas(c, func(n *claim) bool { |
| + if n.Op.Kind != "serving" || n.Op.Exec != b.ID { |
| + 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) |
| + 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"]++ |
| + } |
| +} |
| + |
| // ---------- parking (claim §6.2) ---------- |
| // parkStep advances one park of `prefix` in b's own bucket into dest. It |
| @@ -706,6 +769,11 @@ func (w *world) doObligation(b *box, c *claim, o obligation) { |
| keep = func(k string) bool { return !w.sweepable(w.claim, b.Region, k) } |
| quiet = 0 |
| } |
| + if o.Kind == "park-generation" { |
| + // a dead generation within the live location (a cold recovery's |
| + // old generation, claim rev 6.6 §6.4a): only its own keys |
| + keep = func(k string) bool { return genOf(o.Prefix, k) != o.Ret } |
| + } |
| if !w.parkStep(b, o.Prefix, o.Dest(), quiet, keep) { |
| return |
| } |
diff --git a/internal/sim/hostlife/rules_test.go b/internal/sim/hostlife/rules_test.go| index 9ba303251..bc15fa65c 100644 |
| --- a/internal/sim/hostlife/rules_test.go |
| +++ b/internal/sim/hostlife/rules_test.go |
| @@ -86,6 +86,11 @@ var allRules = []string{ |
| "root-advance", // rev 7.6 §5.1 (R5-3): the restore roots at the newest current successor snapshot at or past the pinned one; the mutation roots at the pinned snapshot only |
| "anchor-acked", // rev 7.7 §5.1 step 7 (R6-1): replication opens only once the anchor PUT is acknowledged, and the position is read from the listing; the mutation opens at once with whatever the listing shows |
| "return-stamp", // claim rev 6.5 (R6-4): the CAS that returns execution to the source writes holder.stamp = the source installation's stamp from the journal; the mutation leaves the target's |
| + "recovery-generation", // claim rev 6.6 §6.4a (R7-1): a wake with lost replication state is a cold recovery, an installing CAS into a fresh generation; the mutation restarts in the same generation (rev 7.7) |
| + "closed-marker", // claim rev 6.6 §6.4a (R7-1): a generation is continued only if closed.json exists with max_txid at the remote maximum and the box's active stamp, and the marker is consumed; the mutation takes any marker as intact |
| + "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 |
| "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 912d1c6a0..48158cc77 100644 |
| --- a/internal/sim/hostlife/sim_test.go |
| +++ b/internal/sim/hostlife/sim_test.go |
| @@ -340,6 +340,7 @@ 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) |
| } |
| } |
| for r := 0; r < nRegions; r++ { |
| @@ -497,11 +498,20 @@ func (w *world) checkDurable(c *claim, when string) { |
| need[x] = true |
| } |
| } |
| + have := map[int]bool{} |
| + if db != nil { |
| + for _, x := range db.Log { |
| + have[x] = true |
| + } |
| + } |
| for _, a := range w.acks { |
| if a.Inc != c.Holder.Inc || a.Exempt || a.Kind != "write" || !need[a.W] { |
| continue |
| } |
| - if db == nil || w.txs[a.W].TX > db.TX { |
| + // coverage by write identity (R7-2): a required write is present |
| + // only if a file the restore applied carried it; a maximum TXID |
| + // says nothing after a fork |
| + if db == nil || !have[a.W] { |
| w.violate("INV-2 (cold restore, %s): durable write %d (box %d, t=%d) lost (%s)", when, a.W, a.Box, a.At, w.ltxShape(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, ""))) |
| return |
| } |
| @@ -510,46 +520,65 @@ func (w *world) checkDurable(c *claim, when string) { |
| return |
| } |
| w.foreignWrites(db, c.Holder.Inc, "cold restore, "+when) |
| - // a write declared lost with a disk (exempt) is not required anywhere, |
| - // but its upload may still land and the planner may still apply it, so |
| - // the keys such writes touched are indeterminate for this comparison; |
| - // every other key must match (rev 7.7: seed 68282, a delete of the |
| - // oldest key in an exempt write that landed after the cold restore) |
| - masked := w.exemptKeys(c.Holder.Inc) |
| - if why, ok := statesEqual(unmask(w.expectState(c.Holder.Inc, db.TX), masked), unmask(db.State, masked)); !ok { |
| - w.violate("INV-2 (cold restore, %s): the restored database differs from the acknowledged writes: %s (restored TXID %d; %s)", when, why, db.TX, w.ltxShape(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, ""))) |
| + // the restored state must be the replay, in acknowledgement order, of |
| + // exactly the writes its files carried (R7-2): a write declared lost |
| + // with a disk that landed anyway is replayed because it is there, a |
| + // required write that is missing was caught above, and no key is ever |
| + // masked; a later required write re-establishes its key's value |
| + if why, ok := statesEqual(w.expectFrom(c.Holder.Inc, have), db.State); !ok { |
| + w.violate("INV-2 (cold restore, %s): the restored database differs from the replay of the writes it carries: %s (restored TXID %d; %s; %s)", when, why, db.TX, w.keyHistory(c.Holder.Inc, why, have), w.ltxShape(e.Region, dataKey(c.Holder.Location, c.Holder.Gen, ""))) |
| } |
| } |
| -// exemptKeys are the keys touched by writes of inc that a disk loss |
| -// declared lost (exempt acknowledgements). |
| -func (w *world) exemptKeys(inc int) map[string]bool { |
| - out := map[string]bool{} |
| +// keyHistory lists, for diagnosis, the acknowledged writes of inc that |
| +// touched the key named in a statesEqual message: W, lineage TXID, value, |
| +// whether the restored files carried it, whether it is exempt. |
| +func (w *world) keyHistory(inc int, why string, have map[int]bool) string { |
| + key := "" |
| + if i := strings.Index(why, "key "); i >= 0 { |
| + rest := why[i+4:] |
| + if j := strings.Index(rest, " "); j > 0 { |
| + key = rest[:j] |
| + } |
| + } |
| + var out []string |
| for _, a := range w.acks { |
| - if a.Inc == inc && a.Exempt && a.Kind == "write" { |
| - t := w.txs[a.W] |
| - k := t.Key |
| - if k == "" { |
| - k = fmt.Sprintf("k%d", a.W) |
| - } |
| - out[k] = true |
| + if a.Inc != inc || a.Kind != "write" { |
| + continue |
| } |
| + t := w.txs[a.W] |
| + k := t.Key |
| + if k == "" { |
| + k = fmt.Sprintf("k%d", a.W) |
| + } |
| + if k != key { |
| + continue |
| + } |
| + out = append(out, fmt.Sprintf("W%d:TX%d=%d have=%v exempt=%v", a.W, t.TX, t.Val, have[a.W], a.Exempt)) |
| } |
| - return out |
| + return "history of " + key + ": " + strings.Join(out, " ") |
| } |
| -// unmask is state without the masked keys, for a comparison. |
| -func unmask(state map[string]int, masked map[string]bool) map[string]int { |
| - if len(masked) == 0 { |
| - return state |
| - } |
| - out := map[string]int{} |
| - for k, v := range state { |
| - if !masked[k] { |
| - out[k] = v |
| +// expectFrom replays, in acknowledgement order, the acknowledged writes of |
| +// inc that the restored files carried (have), into the state a correct |
| +// restore of those files must equal. |
| +func (w *world) expectFrom(inc int, have map[int]bool) map[string]int { |
| + state := map[string]int{} |
| + for _, a := range w.acks { |
| + if a.Inc != inc || a.Kind != "write" || !have[a.W] { |
| + continue |
| + } |
| + t := w.txs[a.W] |
| + if t.Key == "" { |
| + t.Key, t.Val = fmt.Sprintf("k%d", a.W), a.W |
| + } |
| + if t.Val == 0 { |
| + delete(state, t.Key) |
| + } else { |
| + state[t.Key] = t.Val |
| } |
| } |
| - return out |
| + return state |
| } |
| // counting is the sweep's random source, counting its draws so the oracle |
diff --git a/internal/sim/hostlife/witness_test.go b/internal/sim/hostlife/witness_test.go| index 7ac881efa..5b52049c7 100644 |
| --- a/internal/sim/hostlife/witness_test.go |
| +++ b/internal/sim/hostlife/witness_test.go |
| @@ -78,10 +78,6 @@ var witnesses = []witness{ |
| "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", |
| "redundant alone: discovery at adoption parks the other regions; without discovery, an incomplete touched set must park every region at delete, or a pre-release replica is residue after free (INV-10)", wTouchedIncomplete}, |
| - {"door-reread", "door-reread,location-per-incarnation", "door:", |
| - "subsumed by C3a: a listing authorized for the old incarnation selects, after a recreate into the shared location, the new incarnation's object; O2 refuses it", wDoorReread}, |
| - {"door-by-version", "door-by-version,location-per-incarnation", "door:", |
| - "subsumed by C3a: a handle signed before the delete opens, after a recreate into the shared location, the key's current object unless bound to the version it selected", wDoorByVersion}, |
| {"registration-check", "registration-check,fence-etag", "store of app", |
| "a cheap extra: a creator paused across a whole purge resumes; the fence and the ETag condition refuse it without the re-read; with the ETag condition gone the re-read is what stops it", wRegistrationCheck}, |
| {"registration-check", "registration-check,purge-claim-fence", "store of app", |
| @@ -114,6 +110,23 @@ var witnesses = []witness{ |
| "a push installs a snapshot alone and its anchor PUT is slow; litestream reads its position from the remote level-0 listing, so the host must not open replication until the anchor is acknowledged; opened at once it runs from TXID 1 below the snapshot and a cold restore loses the writes (R6-1)", wAnchorHeld}, |
| {"return-stamp", "return-stamp", "issued for box", |
| "a dirty rollback whose source lost its local stamp record; the return CAS writes the source installation's stamp back and the source re-initializes from it; left as the target's stamp, the source activates an installation that was never its own (claim rev 6.5, R6-4)", wReturnLostRecord}, |
| + // rev 7.8 |
| + {"recovery-generation", "recovery-generation", "INV-2", |
| + "writer A's level-0 upload [N+1] is in flight when A dies with its disk; B recovers cold into a fresh generation and replicates its own [N+1]; A's upload lands after B's; in the same generation it would replace B's file and the next cold restore loses B's acknowledged write (R7-1, completion order A after B)", wLateUploadAfter}, |
| + {"recovery-generation", "recovery-generation", "INV-2", |
| + "as above, but A's upload lands after B initialized its position and before B's own upload; in the same generation litestream's upload loop takes the remote [N+1] as already uploaded and skips B's, which is then lost (R7-1, the initialization/remote-position race)", wLateUploadRace}, |
| + {"recovery-generation", "recovery-generation", "INV-2", |
| + "A's in-flight write updated K and was declared lost with A's disk; B durably updates K; A's stale upload lands and restores K's old value; coverage by write identity sees B's write missing where a TXID bound and a per-key mask did not (R7-2)", wExemptKeyRestored}, |
| + {"recovery-generation", "recovery-generation", "no cold recovery", |
| + "a clean stop whose uploads all returned, then a crash before closed.json was written: no marker, so the next wake is a cold recovery into a fresh generation (claim rev 6.6 §6.4a)", wCrashBeforeMarker}, |
| + {"closed-marker", "closed-marker", "INV-2", |
| + "a stale closed.json (max_txid behind the remote maximum) beside a dead writer's in-flight upload; the stale marker forces a cold recovery; taken as intact, the box continues in place and the late upload replaces its file (claim rev 6.6 §6.4a, R7-1)", wStaleMarker}, |
| + {"close-waits", "close-waits", "before every upload returned", |
| + "a clean stop with an upload still in flight: closed.json is written only once the upload returned; written at once, the marker vouches for an upload that may still fail or land late (claim rev 6.6 §6.4a)", wCloseWaits}, |
| + {"newest-l0", "newest-l0", "newest level-0", |
| + "the local position runs ahead of the uploads; two compactions and snapshot rotations later, retention below the previous snapshot would remove the newest uploaded level-0 file, which litestream never does (R7-3)", wNewestL0}, |
| + {"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}, |
| } |
| // TestWitnesses is the deterministic half of the mutation evidence. |
| @@ -311,6 +324,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}}} |
| w.logf("witness: inc %d of app %s serving on box %d", inc, app, b.ID) |
| } |
| @@ -337,6 +353,8 @@ 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}}} |
| 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}} |
| @@ -821,14 +839,23 @@ func (w *world) recreate(t *testing.T, r, inc int) { |
| return c.Holder != nil && c.Holder.Inc == inc && c.Op.Kind == "serving" && b.Row != nil |
| }, w.runConsole, w.actAll(), w.recoveryPass) |
| b.Config, b.CfgVer, b.Live, b.LiveInc = inc, w.cfgVer[app0], app0, inc // config sync and adoption have run |
| - w.reach(t, 200, "new incarnation running", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + // the first wake of the new incarnation's empty generation is a cold |
| + // recovery (claim rev 6.6 §6.4a): the box's role runs it, then the wake |
| + // admits the writer into the recovered generation |
| + if !w.reach(t, 400, "new incarnation running", func() bool { return hasMain(b) }, w.actAll(), func() { w.wake(b) }) { |
| + t.Fatalf("witness: the new incarnation did not run: %s", w.state()) |
| + } |
| w.appWrite(b) |
| w.appWrite(b) // two transactions: its first LTX file has the same key as the fixture's |
| w.replicate(b) |
| } |
| // wDoorReread (claim §6.5 O2): a listing authorized before the delete |
| -// selects after the recreate. |
| +// selects after the recreate. Retired from the table in rev 7.8: under |
| +// claim rev 6.6's writer epochs a recreated incarnation's first wake is a |
| +// cold recovery into a fresh generation, which a listing scoped to the |
| +// generation it was authorized for never selects, so the scenario holds |
| +// with C3a dropped too; C3a stays as defence in depth (§14.2.5). |
| func wDoorReread(t *testing.T, w *world) { |
| b0 := w.boxes[0] |
| w.serve(b0, 1, app0) |
| @@ -1066,8 +1093,9 @@ func (w *world) pushed(t *testing.T, b *box) (loc, gen string) { |
| gen = w.claim.Op.ID |
| w.reach(t, 400, "push released", func() bool { return w.claim.Op.Kind == "serving" }, w.act(b)) |
| w.idleStop(b) |
| - w.wake(b) |
| - if !hasMain(b) { |
| + // the wake after the push: under a mutation it may be a cold recovery |
| + // first (claim rev 6.6 §6.4a), which the box's role runs |
| + if !w.reach(t, 60, "woken after the push", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) { |
| t.Fatalf("witness: the host did not wake after the push: %s%s", w.state(), w.boxesState()) |
| } |
| return w.claim.Holder.Location, gen |
| @@ -1116,7 +1144,7 @@ func wRestoreOverlap(t *testing.T, w *world) { |
| w.idleStop(b) |
| b.DB = nil // the disk is lost; the next wake restores cold |
| w.logf("witness: box %d lost its database; the generation holds the snapshot at %d and [%d,%d]", b.ID, n, n-1, n+2) |
| - w.reach(t, 20, "woken cold", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + w.reach(t, 60, "woken cold", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) |
| w.finish(t) |
| } |
| @@ -1159,18 +1187,25 @@ func wSnapshotOnlyInstall(t *testing.T, w *world) { |
| w.appWrite(b) |
| } |
| w.replicate(b) |
| - w.pushed(t, b) |
| - if len(w.violations) > 0 { |
| + // the push's probe process goes on serving after the release (no |
| + // clean close in between: a clean close would write a marker whose |
| + // max_txid betrays the missing anchor and force a recovery first) |
| + w.userPush() |
| + w.reach(t, 400, "push released", func() bool { return w.claim.Op.Kind == "serving" }, w.act(b)) |
| + if len(w.violations) > 0 || !hasMain(b) { |
| return |
| } |
| for i := 0; i < 2; i++ { |
| w.appWrite(b) |
| w.replicate(b) |
| } |
| - w.idleStop(b) |
| - b.DB = nil // the disk is lost |
| + // the writer dies with its disk (a clean close would write a marker |
| + // whose max_txid, 2, is behind the snapshot's 3 and force a cold |
| + // recovery from the warm database, which hides the missing anchor) |
| + w.crash(b) |
| + w.loseDisk(b) |
| w.logf("witness: box %d lost its database after replicating past the push's snapshot", b.ID) |
| - w.reach(t, 20, "woken cold", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + w.reach(t, 60, "woken cold", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) |
| w.finish(t) |
| } |
| @@ -1200,7 +1235,7 @@ func wRootAdvance(t *testing.T, w *world) { |
| w.idleStop(b) |
| b.DB = nil // the disk is lost |
| w.logf("witness: box %d lost its database after four rotations", b.ID) |
| - w.reach(t, 20, "woken cold", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + w.reach(t, 60, "woken cold", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) |
| w.finish(t) |
| } |
| @@ -1220,19 +1255,20 @@ func wAnchorHeld(t *testing.T, w *world) { |
| w.userPush() |
| gen := w.claim.Op.ID |
| loc := w.claim.Holder.Location |
| - w.reach(t, 400, "push published", func() bool { |
| - return w.claim.Op.Kind == "pushing" && w.claim.Op.Phase == "restore" |
| + // the build (R7-4): the snapshot is acknowledged, the anchor is next; |
| + // the witness holds the anchor PUT there |
| + w.reach(t, 400, "push building its anchor", func() bool { |
| + return w.claim.Op.Kind == "pushing" && w.claim.Op.Phase == "install-anchor" |
| }, w.act(b)) |
| - m := w.buckets[b.Region].current(genMarkerKey(loc)) |
| - if m == nil || m.B.Gen != gen { |
| - t.Fatalf("witness: the push did not publish its generation: %s", w.state()) |
| - } |
| - n := m.B.TX |
| + n := w.claim.Op.BaseTX |
| anchor := ltxKey(loc, gen, 0, n, n) |
| w.holdKey(anchor) |
| prefix := dataKey(loc, gen, "") |
| for i := 0; i < 40; i++ { |
| w.tick(w.act(b)) |
| + if len(w.violations) > 0 { |
| + return // INV-11: published before the anchor landed |
| + } |
| for _, f := range w.currentLTX(w.buckets[b.Region], prefix, 0) { |
| if f.B.Min != n { |
| w.violate("INV-2 (witness): box %d replicated [%d,%d] before its anchor [%d,%d] was acknowledged", b.ID, f.B.Min, f.B.Max, n, n) |
| @@ -1250,14 +1286,14 @@ func wAnchorHeld(t *testing.T, w *world) { |
| } |
| w.reach(t, 400, "push released", func() bool { return w.claim.Op.Kind == "serving" }, w.act(b)) |
| w.idleStop(b) |
| - w.reach(t, 20, "woken after the push", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + w.reach(t, 60, "woken after the push", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) |
| for i := 0; i < 2; i++ { |
| w.appWrite(b) |
| w.replicate(b) |
| } |
| w.idleStop(b) |
| b.DB = nil // the disk is lost |
| - w.reach(t, 20, "woken cold", func() bool { return hasMain(b) }, func() { w.wake(b) }) |
| + w.reach(t, 60, "woken cold", func() bool { return hasMain(b) }, w.act(b), func() { w.wake(b) }) |
| w.finish(t) |
| } |
| @@ -1294,3 +1330,356 @@ func wReturnLostRecord(t *testing.T, w *world) { |
| w.violate("INV-6 (witness): the pre-move stamp %d did not retire when %d activated on box %d", s, act, src.ID) |
| } |
| } |
| + |
| +// ---------- rev 7.8 ---------- |
| + |
| +// crash is the agent dying with its processes (an OOM kill, a crash): what |
| +// it had in flight is unaccounted for, and the restarted agent knows |
| +// nothing of it (forget). Nothing cleans up, so no closed.json. |
| +func (w *world) crash(b *box) { |
| + b.Procs = nil |
| + b.forget() |
| + w.logf("witness: box %d's agent crashed", b.ID) |
| +} |
| + |
| +// loseDisk is the disk-loss fault: the database file is gone, and every |
| +// acknowledged write not yet durable anywhere is declared lost (exempt). |
| +func (w *world) loseDisk(b *box) { |
| + if b.DB == nil { |
| + return |
| + } |
| + for i := range w.acks { |
| + a := &w.acks[i] |
| + if a.Box == b.ID && a.Inc == b.DB.Inc && !w.isDurable(a.W) { |
| + a.Exempt = true |
| + } |
| + } |
| + b.DB = nil |
| + w.logf("witness: box %d lost its database file", b.ID) |
| +} |
| + |
| +// appWriteKey is one acknowledged application write of key = val (0 is a |
| +// delete) by b's main process. |
| +func (w *world) appWriteKey(b *box, key string, val int) int { |
| + if !hasMain(b) { |
| + panic("appWriteKey: no main process") |
| + } |
| + p := b.Procs[0] |
| + w.nextW++ |
| + wr := w.nextW |
| + db := b.DB |
| + t := tx{W: wr, Key: key, Val: val} |
| + if val == 0 { |
| + delete(db.State, key) |
| + } else { |
| + db.State[key] = val |
| + } |
| + db.TX++ |
| + t.TX = db.TX |
| + db.Pending = append(db.Pending, wr) |
| + db.Log = append(db.Log, wr) |
| + w.txs[wr] = t |
| + w.acks = append(w.acks, ack{W: wr, Inc: p.Inc, Box: b.ID, At: w.now, Kind: "write"}) |
| + w.writeInc[wr] = p.Inc |
| + w.logf("witness: box %d wrote %d: %s = %d", b.ID, wr, key, val) |
| + return wr |
| +} |
| + |
| +// recovered drives the box until it runs again: a cold recovery first, |
| +// when its replication state was lost, then the wake. |
| +func (w *world) recovered(t *testing.T, b *box, what string) { |
| + if !w.reach(t, 120, what, func() bool { return hasMain(b) }, w.act(b), w.recoveryPass, func() { w.wake(b) }) { |
| + t.Fatalf("witness: %s: the host did not run again: %s%s", what, w.state(), w.boxesState()) |
| + } |
| +} |
| + |
| +// deadWriter is the common opening of the R7-1 witnesses: the serving box |
| +// writes once more, its level-0 upload for that write is held in flight, |
| +// and the box dies with its disk. It returns the held key. |
| +func (w *world) deadWriter(t *testing.T, b *box) string { |
| + 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) |
| + w.holdKey(key) |
| + w.replicate(b) // A's upload: accepted, outstanding |
| + w.hold = nil |
| + w.crash(b) |
| + w.loseDisk(b) |
| + return key |
| +} |
| + |
| +// wLateUploadAfter (R7-1): A's outstanding upload lands after B uploaded |
| +// its own file for the same TXID. |
| +func wLateUploadAfter(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + key := w.deadWriter(t, b) |
| + w.recovered(t, b, "B running after A's death") |
| + w.appWrite(b) |
| + w.replicate(b) // B's upload |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: A's upload was not held") |
| + } |
| + w.checkDurable(w.claim, "after A's late upload landed") // before any snapshot of the warm database repairs the replica |
| + w.finish(t) |
| +} |
| + |
| +// wLateUploadRace (R7-1): A's outstanding upload lands after B |
| +// initialized its position and before B's own upload for that TXID. |
| +func wLateUploadRace(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + key := w.deadWriter(t, b) |
| + w.recovered(t, b, "B running after A's death") |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: A's upload was not held") |
| + } |
| + w.appWrite(b) |
| + w.replicate(b) // B's upload, or litestream's skip at the remote position |
| + w.checkDurable(w.claim, "after B's upload") |
| + w.finish(t) |
| +} |
| + |
| +// wExemptKeyRestored (R7-2): A's in-flight write updated K; B durably |
| +// updates K; A's stale upload lands and restores A's value. |
| +func wExemptKeyRestored(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + loc := w.claim.Holder.Location |
| + k := fmt.Sprintf("k%d", b.DB.Log[0]) // a key the fixture wrote |
| + w.appWriteKey(b, k, 99) |
| + key := ltxKey(loc, "", 0, b.DB.Pos+1, b.DB.Pos+1) |
| + w.holdKey(key) |
| + w.replicate(b) |
| + w.hold = nil |
| + w.crash(b) |
| + w.loseDisk(b) // A's update of K is declared lost |
| + w.recovered(t, b, "B running after A's death") |
| + w.appWriteKey(b, k, 77) // B's durable update of K |
| + w.replicate(b) |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: A's upload was not held") |
| + } |
| + w.checkDurable(w.claim, "after A's stale upload landed") |
| + w.finish(t) |
| +} |
| + |
| +// wCrashBeforeMarker (claim rev 6.6 §6.4a): a clean stop drains its |
| +// uploads, then the agent crashes before closed.json is written; the next |
| +// wake must be a cold recovery. |
| +func wCrashBeforeMarker(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + gen := w.claim.Holder.Gen |
| + w.appWrite(b) |
| + key := ltxKey(w.claim.Holder.Location, "", 0, b.DB.Pos+1, b.DB.Pos+1) |
| + w.holdKey(key) |
| + w.idleStop(b) // closing: the upload is outstanding, no marker yet |
| + 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") |
| + } |
| + w.recovered(t, b, "running after the crash") |
| + if w.claim.Holder.Gen == gen { |
| + w.violate("INV-2 (witness): no cold recovery after a crash left generation %q without closed.json; box %d continued it", gen, b.ID) |
| + return |
| + } |
| + w.finish(t) |
| +} |
| + |
| +// wStaleMarker (claim rev 6.6 §6.4a): a closed.json whose max_txid is |
| +// behind the remote maximum, beside a dead writer's outstanding upload. |
| +func wStaleMarker(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + 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}}) |
| + w.recovered(t, b, "B running after A's death") |
| + w.appWrite(b) |
| + w.replicate(b) |
| + if w.completeHeld(key) == 0 { |
| + t.Fatalf("witness: A's upload was not held") |
| + } |
| + w.checkDurable(w.claim, "after A's late upload landed") |
| + w.finish(t) |
| +} |
| + |
| +// deadWriterContinues: the running writer writes once more with its |
| +// upload held, then dies with its disk; the held key is returned. |
| +func (w *world) deadWriterContinues(b *box, loc string) string { |
| + w.appWrite(b) |
| + key := ltxKey(loc, b.DB.Gen, 0, b.DB.Pos+1, b.DB.Pos+1) |
| + w.holdKey(key) |
| + w.replicate(b) |
| + w.hold = nil |
| + w.crash(b) |
| + w.loseDisk(b) |
| + return key |
| +} |
| + |
| +// wCloseWaits (claim rev 6.6 §6.4a): a clean stop with an upload in |
| +// flight writes no closed.json until the upload returned. |
| +func wCloseWaits(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) |
| + w.holdKey(key) |
| + w.idleStop(b) |
| + for i := 0; i < 20; i++ { |
| + 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) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + if w.completeHeld(key) == 0 { |
| + 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") |
| + } |
| + // the writer continues the generation: no upload of its previous run |
| + // can be outstanding, every one returned before the marker |
| + w.recovered(t, b, "continued after the clean close") |
| + if w.claim.Holder.Gen != "" { |
| + t.Fatalf("witness: a cold recovery after a clean close") |
| + } |
| + w.appWrite(b) |
| + w.replicate(b) |
| + w.finish(t) |
| +} |
| + |
| +// wNewestL0 (R7-3): the local position runs ahead of the uploads across |
| +// two compactions and snapshot rotations. |
| +func wNewestL0(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + loc := w.claim.Holder.Location |
| + prefix := dataKey(loc, "", "") |
| + held := map[string]bool{} |
| + w.hold = func(r *request) bool { return held[r.key] } |
| + k := func(n int) string { return ltxKey(loc, "", 0, n, n) } |
| + // position 2 is already uploaded (the fixture at 1, one write at 2) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + // uploads from 3 on are held; the writer reaches position 10 |
| + held[k(3)] = true |
| + for i := 3; i <= 10; i++ { |
| + w.appWrite(b) |
| + w.replicate(b) |
| + } |
| + // 3 lands; 4 and 5 ship; 6 is held: compaction sees {2,3,4,5}, keeps 5, |
| + // rotates a snapshot at the local position 10 |
| + held[k(6)] = true |
| + delete(held, k(3)) |
| + w.completeHeld(k(3)) |
| + w.replicate(b) |
| + if got := w.maxL0(w.buckets[b.Region], prefix); got == 0 { |
| + w.violate("INV-2 (witness): compaction removed the newest level-0 file of %s; litestream keeps it, and a fresh box reads its position there (%s)", prefix, w.ltxShape(b.Region, prefix)) |
| + return |
| + } else if got != 5 { |
| + t.Fatalf("witness: newest uploaded level-0 file is %d, want 5 (%s)", got, w.ltxShape(b.Region, prefix)) |
| + } |
| + // the writer goes on to 11; 6 lands and 7 to 11 ship but 12 is held: |
| + // compaction sees {5,6,...}, rotates a snapshot at 11 and expires |
| + // below the previous snapshot, 10 |
| + w.appWrite(b) |
| + w.replicate(b) |
| + held[k(12)] = true |
| + delete(held, k(6)) |
| + w.completeHeld(k(6)) |
| + w.replicate(b) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + w.hold = nil |
| + newest := w.maxL0(w.buckets[b.Region], prefix) |
| + snaps := w.currentLTX(w.buckets[b.Region], prefix, 9) |
| + if len(snaps) == 0 { |
| + t.Fatalf("witness: no snapshot rotated: %s", w.ltxShape(b.Region, prefix)) |
| + } |
| + if newest == 0 { |
| + w.violate("INV-2 (witness): retention removed the newest level-0 file of %s; litestream keeps it, and a fresh box reads its position there (%s)", prefix, w.ltxShape(b.Region, prefix)) |
| + return |
| + } |
| + w.completeHeld(k(12)) |
| + w.finish(t) |
| +} |
| + |
| +// wBuildCrashes (R7-4): a push's build with its snapshot PUT, then its |
| +// anchor PUT, held, the agent crashing in between each. |
| +func wBuildCrashes(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + for i := 0; i < 2; i++ { |
| + w.appWrite(b) |
| + } |
| + w.replicate(b) |
| + w.userPush() |
| + gen := w.claim.Op.ID |
| + loc := w.claim.Holder.Location |
| + w.reach(t, 400, "push installing", func() bool { |
| + return w.claim.Op.Kind == "pushing" && w.claim.Op.Phase == "install" |
| + }, w.act(b)) |
| + n := w.claim.Op.PushSnap.Max |
| + snap := snapKey(loc, gen, n) |
| + anchor := ltxKey(loc, gen, 0, n, n) |
| + published := func() bool { |
| + m := w.buckets[b.Region].current(genMarkerKey(loc)) |
| + return m != nil && m.B.Gen == gen |
| + } |
| + w.holdKey(snap) |
| + for i := 0; i < 12; i++ { |
| + w.tick(w.act(b)) |
| + if len(w.violations) > 0 { |
| + return // INV-11 |
| + } |
| + if i == 5 { |
| + w.crash(b) // between the intent and the snapshot's acknowledgement |
| + } |
| + if published() { |
| + w.violate("INV-11 (witness): generation.json names %s before its snapshot %s landed", gen, snap) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + w.completeHeld(snap) |
| + w.reach(t, 100, "push building its anchor", func() bool { return w.claim.Op.Phase == "install-anchor" }, w.act(b)) |
| + w.holdKey(anchor) |
| + for i := 0; i < 12; i++ { |
| + w.tick(w.act(b)) |
| + if len(w.violations) > 0 { |
| + return |
| + } |
| + if i == 5 { |
| + w.crash(b) // between the snapshot's and the anchor's acknowledgement |
| + } |
| + if published() { |
| + w.violate("INV-11 (witness): generation.json names %s before its anchor %s landed", gen, anchor) |
| + return |
| + } |
| + } |
| + w.hold = nil |
| + w.completeHeld(anchor) |
| + w.reach(t, 400, "push released", func() bool { return w.claim.Op.Kind == "serving" }, w.act(b)) |
| + w.crash(b) // after completion: the marker stands, the generation is complete |
| + w.recovered(t, b, "running after the push") |
| + w.appWrite(b) |
| + w.replicate(b) |
| + w.finish(t) |
| +} |
diff --git a/internal/sim/hostlife/world_test.go b/internal/sim/hostlife/world_test.go| index d28294236..2f1d3fafc 100644 |
| --- a/internal/sim/hostlife/world_test.go |
| +++ b/internal/sim/hostlife/world_test.go |
| @@ -82,16 +82,23 @@ type blob struct { |
| Max int |
| Pages map[string]int |
| TXs []int |
| + // Full: the file carries every page (a snapshot, an anchor, litestream's |
| + // first sync after a cold initialization, a compaction of one of those), |
| + // so applying it replaces the state; a key absent from it is gone |
| + Full bool |
| // a "gen" marker (R3-1, R4-1): the generation's snapshot by key, version |
| // id and TXID |
| Base int |
| SnapKey string |
| TX 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" 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 |
| } |
| type ref struct { |
| @@ -265,9 +272,17 @@ type localDB struct { |
| Queue []*blob |
| InFlight *request // the head's upload, as the client sees it |
| InFlightAt 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 |
| + // FirstFull: the next level-0 file carries every page (litestream's |
| + // 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 |
| + // 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 |
| } |
| func (d *localDB) clone() *localDB { |
| @@ -332,10 +347,10 @@ 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 |
| - // anchors are the level-0 anchor PUTs this box's activator has issued |
| - // and not yet seen acknowledged, by key (rev 7.7, R6-1): replication |
| - // opens only once one of them is done, and a request the client has |
| - // waited uploadTimeout steps for is retried with the same bytes. |
| + // 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 |
| + // for is retried with the same bytes. |
| anchors map[string][]*request |
| ExecUntil int // local exec-lease file (watchdog input); 0 none |
| @@ -481,10 +496,42 @@ func (b *box) forget() { |
| b.pendingTimers = nil |
| if b.DB != nil { |
| b.DB.InFlight = nil // the upload's client is gone; the restarted replica uploads the same file again |
| + // a writer that was draining before its clean close is gone with |
| + // its uploads unaccounted for: no closed.json (claim rev 6.6 §6.4a) |
| + b.DB.Closing, b.DB.CloseReq = false, nil |
| } |
| b.anchors = nil // the activator issues the anchor again, the same bytes |
| } |
| +// ackedPut is a PUT the activator must see acknowledged before it goes on |
| +// (R6-1): it issues the request once, retries the same bytes after |
| +// uploadTimeout steps while none has returned, and reports whether one has. |
| +func (w *world) ackedPut(b *box, key string, bl *blob, ifNone bool) bool { |
| + acked, inflight := false, false |
| + for _, r := range b.anchors[key] { |
| + if r.done { |
| + acked = true |
| + } else if w.now-r.at < uploadTimeout { |
| + inflight = true |
| + } |
| + } |
| + if acked || inflight { |
| + return acked |
| + } |
| + r := &request{kind: rPut, region: b.Region, key: key, b: bl, ifSet: ifNone} |
| + if v := w.issue(b.ID, r); v > 0 { |
| + return true |
| + } |
| + if r.done { |
| + return true // refused (412): the object is already there, written once |
| + } |
| + if b.anchors == nil { |
| + b.anchors = map[string][]*request{} |
| + } |
| + b.anchors[key] = append(b.anchors[key], r) |
| + return false |
| +} |
| + |
| // ---------- control plane ---------- |
| type holder struct { |
| @@ -562,9 +609,10 @@ type opState struct { |
| 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) |
| + 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) |
| Committed bool |
| Opened bool |
| ParkLocalNow bool |