amadan
sim: hostlife: replication position, level-0 anchor, snapshot rotation, claim rev 6.3 stamps (rev 7.6)
Joint round 5 on the model, from the pinned-library facts in internal/sim/litestreamfacts: R5-1 the replication position (Pos) is separate from the lineage TXID. A cold restore reads its position from the newest current level-0 file and, when it restored past that, writes the level-0 anchor [cur,cur] before replication starts (rule restore-anchor; witness: snapshot-only install, write, replicate, local loss, cold restore). R5-2 no whole-state checksum: the pinned writers set NoChecksum, so the model verifies by per-file ranges and the restored state only. R5-3 snapshot rotation and retention as the library does them: a compaction at the snapshot level writes a new snapshot, keeps only the newest, and expires level-1 and level-0 files below the previous snapshot's Max. The restore roots at the newest current successor snapshot at or past the pinned one (rule root-advance; witness). R5-4 claim rev 6.3 (8faa89a4): holder.stamp is written by every installing CAS (create step 4, adoption, flip, reinstall, dirty rollback, push commit); a box's local record is initialized only from a fresh claim read under the guard that names it op.exec or landing participant; no local record refuses the write; two retirements, retire-artifacts(S) and retire-installation. R5-5 the readiness budget prices snapshot construction and the cross-region transfer, with the sealed snapshot built on the target from staged material. R5-6 the retention fixture populates 2..33 and asserts the injected incarnation absent; LTX keys follow the S3 client's <prefix>/<4-hex-level>/<min>-<max>.ltx. The ship step advances the position as soon as the upload is issued (a retry carries the same bytes): seed 143 showed a late-landing upload leaving a numbering gap.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
10 files changed,
+601
−228
internal/sim/hostlife/background_test.go+3 −5internal/sim/hostlife/budget_test.go+46 −7internal/sim/hostlife/console_test.go+7 −2internal/sim/hostlife/ltx_test.go+206 −67internal/sim/hostlife/moves_test.go+60 −18internal/sim/hostlife/protocol_test.go+84 −37internal/sim/hostlife/rules_test.go+5 −3internal/sim/hostlife/sim_test.go+26 −17internal/sim/hostlife/witness_test.go+99 −39internal/sim/hostlife/world_test.go+65 −33
diff --git a/internal/sim/hostlife/background_test.go b/internal/sim/hostlife/background_test.go| index 9945c580d..7988c317a 100644 |
| --- a/internal/sim/hostlife/background_test.go |
| +++ b/internal/sim/hostlife/background_test.go |
| @@ -150,7 +150,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)}, |
| + 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.3 §2) |
| Op: opState{Kind: "serving", Exec: eb.ID, Parts: []int{eb.ID}}, Regions: []int{w.rec.Region}, Boxes: []int{eb.ID}, Complete: false} |
| if on("legacy-discovery") { |
| // every other region parks whatever it holds: it cannot be the |
| @@ -197,9 +197,7 @@ func (w *world) reconcileClaimed(b *box, c *claim) { |
| } |
| if b.Row == nil && c.Op.Kind == "serving" { |
| b.Row = &row{Inc: h.Inc, App: h.App, Status: "ready"} |
| - if b.stampOf(h.Inc) == 0 { |
| - b.install(h.Inc, c.Ver) // a create's installation: its stamp is the gen that made it serving here (claim rev 6.2 §6.4) |
| - } |
| + w.initRecord(b, c) // the local record from holder.stamp, on this fresh read (claim rev 6.3 §6.4) |
| } |
| } else if b.Row != nil && !(on("participant-rows") && w.ownsLocalObligation(c, b)) { |
| w.retireRow(b) |
| @@ -257,7 +255,7 @@ func (w *world) passExec(b *box) { |
| return // the selected bundle is no longer the held one |
| } |
| } |
| - // the local record (claim rev 6.2 §6.4): the writer carries the stamp |
| + // the local record (claim rev 6.3 §6.4): the writer carries the stamp |
| // it selected under, admitted only if it is still the active one |
| if !b.admitsStamp(p.Inc, p.Stamp) { |
| return |
diff --git a/internal/sim/hostlife/budget_test.go b/internal/sim/hostlife/budget_test.go| index 7208e9c7c..54f5142a0 100644 |
| --- a/internal/sim/hostlife/budget_test.go |
| +++ b/internal/sim/hostlife/budget_test.go |
| @@ -2,15 +2,21 @@ package hostlife |
| import "testing" |
| -// The readiness arithmetic of moves spec §7.3 T11 (rev 7.4, R3-7), in |
| +// The readiness arithmetic of moves spec §7.3 T11 (rev 7.6, R3-7, R5-5), in |
| // seconds. Rev 7.3's T11 required every one of three probe attempts, each |
| // with its settle, to fit the window beside the fixed costs: 150 s of |
| // probing alone, against a default window of 120 s, so no default move |
| -// was ever eligible. Rev 7.4 makes the permitted attempt count depend on |
| -// the admitted budget. |
| +// was ever eligible. Rev 7.4 made the permitted attempt count depend on |
| +// the admitted budget. Rev 7.5 then made the sealed transfer a full |
| +// snapshot but still priced the final mirror at the 16 MiB bar (R5-5). |
| +// Rev 7.6 prices what the cut really does: the final delta mirror within |
| +// the bar, then on the target a restore of the staged replica, the |
| +// construction of the generation's snapshot and its level-0 anchor, each |
| +// at the measured local rate for the whole database. |
| type cutEstimate struct { |
| - Stop, Checkpoint, Mirror, Archive, Copy, Restore, Render, Open float64 |
| - Probe, Settle float64 // one attempt |
| + Stop, Checkpoint, Mirror, Archive, Render, Open float64 |
| + RestoreStaged, SnapshotBuild, Anchor float64 // target-side, local, whole database |
| + Probe, Settle float64 // one attempt |
| } |
| const ( |
| @@ -18,10 +24,11 @@ const ( |
| probeAttemptsMax = 3 // §7.7: never more than three attempts |
| probeSeconds = 30.0 |
| settleSeconds = 20.0 |
| + mirrorBarMiB = 16.0 // §7.3: the final round is at most this much |
| ) |
| func (e cutEstimate) fixed() float64 { |
| - return e.Stop + e.Checkpoint + e.Mirror + e.Archive + e.Copy + e.Restore + e.Render + e.Open |
| + return e.Stop + e.Checkpoint + e.Mirror + e.Archive + e.RestoreStaged + e.SnapshotBuild + e.Anchor + e.Render + e.Open |
| } |
| // attemptsAdmitted is T11 as rev 7.4 states it: the attempts permitted are |
| @@ -44,9 +51,22 @@ func (e cutEstimate) fitsRev73(window float64) bool { |
| return e.fixed()+probeAttemptsMax*(e.Probe+e.Settle) <= window |
| } |
| +// estimate prices a cut for a database of dbMiB over a cross-region path |
| +// of crossMiBps and a local disk of localMiBps, as rev 7.6 §7.3 does. |
| +func estimate(dbMiB, crossMiBps, localMiBps float64) cutEstimate { |
| + return cutEstimate{ |
| + Stop: 2, Checkpoint: 1, Archive: 2, Render: 1, Open: 5, |
| + Mirror: mirrorBarMiB / crossMiBps, |
| + RestoreStaged: dbMiB / localMiBps, |
| + SnapshotBuild: dbMiB / localMiBps, |
| + Anchor: dbMiB / localMiBps, |
| + Probe: probeSeconds, Settle: settleSeconds, |
| + } |
| +} |
| + |
| func TestReadinessBudget(t *testing.T) { |
| // a tiny instance: every fixed cost at the floor of its measured rate |
| - tiny := cutEstimate{Stop: 2, Checkpoint: 1, Mirror: 2, Archive: 2, Copy: 1, Restore: 1, Render: 1, Open: 5, Probe: probeSeconds, Settle: settleSeconds} |
| + tiny := estimate(1, 10, 200) |
| if tiny.fitsRev73(cutWindowDefault) { |
| t.Fatalf("rev 7.3's formula admits a tiny instance in the default window: R3-7 is stale") |
| } |
| @@ -66,4 +86,23 @@ func TestReadinessBudget(t *testing.T) { |
| if n := tiny.attemptsAdmitted(600); n != probeAttemptsMax { |
| t.Fatalf("a 600 s window admits %d attempts, want %d", n, probeAttemptsMax) |
| } |
| + // R5-5: a 2 GiB database over a 10 MiB/s cross-region path. Rev 7.5's |
| + // full-snapshot transfer alone takes 205 s and could never fit; rev |
| + // 7.6 ships the final delta within the bar and builds the snapshot on |
| + // the target at the local rate |
| + big := estimate(2048, 10, 200) |
| + fullTransfer := 2048.0 / 10 |
| + if fullTransfer < cutWindowDefault { |
| + t.Fatalf("the example no longer illustrates R5-5") |
| + } |
| + if n := big.attemptsAdmitted(cutWindowDefault); n != 1 { |
| + t.Fatalf("a 2 GiB database over 10 MiB/s with the snapshot built on the target admits %d attempts, want 1 (fixed %.1f s)", n, big.fixed()) |
| + } |
| + // and a cross-region snapshot transfer, if a builder did that instead, |
| + // is refused by the same arithmetic |
| + naive := big |
| + naive.Mirror, naive.RestoreStaged, naive.SnapshotBuild = fullTransfer, 0, 0 |
| + if n := naive.attemptsAdmitted(cutWindowDefault); n != 0 { |
| + t.Fatalf("a full cross-region snapshot transfer admits %d attempts, want 0", n) |
| + } |
| } |
diff --git a/internal/sim/hostlife/console_test.go b/internal/sim/hostlife/console_test.go| index 31a1f339c..3b3244b7a 100644 |
| --- a/internal/sim/hostlife/console_test.go |
| +++ b/internal/sim/hostlife/console_test.go |
| @@ -315,6 +315,7 @@ func (w *world) createSteps(p *consolePlan) bool { |
| return false |
| } |
| n.Op = opState{Kind: "serving", Exec: b.ID, Parts: []int{b.ID}} |
| + n.Holder.Stamp = w.nextGen() // create step 4: holder.stamp is this CAS's gen (claim rev 6.3 §2) |
| return true |
| }) { |
| delete(ad.Reg, p.OpID) |
| @@ -569,7 +570,7 @@ func (w *world) orphanSteps(p *consolePlan) bool { |
| // deleteSteps: record DeleteIfMatch on the version the intent names, then |
| // parking with obligations, then free once every obligation is |
| -// acknowledged (claim §5.2). Under a purge (claim rev 6.2 §5.3 step 4, |
| +// acknowledged (claim §5.2). Under a purge (claim rev 6.3 §5.3 step 4, |
| // R4-4) the same steps run under the purge's own claim: the record version |
| // is recorded in op.intent first, the delete is conditional on it, and |
| // purging → parking is legal only with the record authoritatively gone |
| @@ -896,7 +897,10 @@ func (w *world) doorAuthorize() { |
| // the listing will select the generation's newest LTX object |
| bk := w.buckets[b.Region] |
| key := "" |
| - for _, k := range bk.keys(dataKey(c.Holder.Location, c.Holder.Gen, "ltx/")) { |
| + for _, k := range bk.keys(dataKey(c.Holder.Location, c.Holder.Gen, "")) { |
| + if !isDataKey(k) || genPrefixOf(k) != dataKey(c.Holder.Location, c.Holder.Gen, "") { |
| + continue |
| + } |
| if v := bk.current(k); v != nil && (key == "" || v.ID > bk.current(key).ID) { |
| key = k |
| } |
| @@ -1021,6 +1025,7 @@ func (w *world) recoveryPass() { |
| return false |
| } |
| n.Op = opState{Kind: "serving", Exec: b.ID, Parts: []int{b.ID}} |
| + n.Holder.Stamp = w.nextGen() // the recovery pass finishing create step 4 writes holder.stamp too |
| return true |
| }) { |
| delete(w.admit[c.Holder.App].Reg, c.Op.ID) |
diff --git a/internal/sim/hostlife/ltx_test.go b/internal/sim/hostlife/ltx_test.go| index 8d4a1ac7d..54d97a9c8 100644 |
| --- a/internal/sim/hostlife/ltx_test.go |
| +++ b/internal/sim/hostlife/ltx_test.go |
| @@ -3,24 +3,43 @@ package hostlife |
| import ( |
| "fmt" |
| "hash/fnv" |
| + "regexp" |
| "sort" |
| "strings" |
| ) |
| -// The database and its replica, with litestream's LTX semantics (rev 7.5, |
| -// R4-1). |
| +// The database and its replica, with the pinned litestream's LTX |
| +// behaviour (rev 7.6, R4-1, R5-1 to R5-3; the facts in |
| +// internal/sim/litestreamfacts). |
| // |
| -// A database is a map of keys to values. A transaction inserts, updates or |
| -// deletes one key and advances the database's TXID, the lineage's own |
| -// counter (litestream's TXID), which a restore continues. A replica object |
| -// is an LTX file: a transaction range [Min, Max], the final page image of |
| -// every key touched within it, and the checksum of the whole database |
| -// after Max. It never carries the transaction list: compaction merges |
| -// files into one covering [Min of first, Max of last] with the final |
| -// images, and deletes the inputs, so individual transactions disappear |
| -// from the current listing. A snapshot is the file [1, TX] with every |
| -// page. The oracle alone knows which acknowledged transaction each object |
| -// covers (blob.TXs), for durability; restore never reads it. |
| +// A database is a map of keys to values with two counters: the lineage |
| +// transaction id (TX), the oracle's order of acknowledged writes, and the |
| +// replication position (Pos), litestream's TXID, which counts shipped LTX |
| +// files and is what every LTX range is numbered in. A replica object is an |
| +// LTX file: a level (0 and 1, 9 for a snapshot), a range in replication |
| +// numbering, the final page image of every key it touches, and, for the |
| +// oracle only, the writes it covers. Pinned litestream writes no database |
| +// checksum into any file (R5-2): a file carries a CRC-64 of itself, which |
| +// the model's files never fail, and nothing else; restore verifies by |
| +// decoding, not by a state checksum. |
| +// |
| +// Replication (R5-1): each shipped file carries one TXID, Pos+1; a |
| +// database with position 0 ships a full-page level-0 file [1, 1] first, |
| +// as litestream's first sync does. Litestream takes a fresh box's position |
| +// from the newest remote level-0 file, and from nothing else: a sidecar is |
| +// follow-mode only, and a snapshot does not count. So a cold restore |
| +// writes a level-0 anchor [cur, cur] with every page before replication |
| +// starts, and the position is cur; without it the position is the newest |
| +// remote level-0 file's, or 0, and replication restarts below the |
| +// snapshot, where the planner never looks. |
| +// |
| +// Rotation and retention (R5-3): the executor's litestream compacts three |
| +// level-0 files into one level-1 file (deleting the inputs), writes a new |
| +// snapshot after each compaction, keeps only the newest snapshot, and |
| +// expires level-1 and level-0 files whose range ends below the previous |
| +// snapshot's TXID. A restore roots at the newest current snapshot of the |
| +// generation at or past the pinned one, and falls back to the pinned |
| +// version only when no successor is current. |
| type tx struct { |
| W int // the global write id, which is what the application acknowledged |
| @@ -29,8 +48,8 @@ type tx struct { |
| Val int // the value; 0 is a delete |
| } |
| -// sum is the database checksum: litestream's rolling checksum stands in |
| -// for a digest of the sorted state. |
| +// sum is a digest of the state, for the model's own digests (stop |
| +// evidence, the settle check); no replica file carries it (R5-2). |
| func sum(state map[string]int) string { |
| keys := make([]string, 0, len(state)) |
| for k, v := range state { |
| @@ -127,51 +146,86 @@ func oldestKey(state map[string]int) string { |
| return best |
| } |
| -// ltxKey and snapKey are litestream's layout under a generation: |
| -// ltx/<level>/<min>-<max>; a snapshot is level 9 from TXID 1. |
| +// ltxKey is the pinned S3 client's layout under a generation prefix: |
| +// <prefix>/<level as four hex digits>/<min>-<max>.ltx, sixteen hex digits |
| +// each (litestream s3/replica_client.go, ltx.FormatFilename). A snapshot |
| +// is level 9 from TXID 1. |
| func ltxKey(loc, gen string, level, min, max int) string { |
| - return fmt.Sprintf("%sltx/%d/%06d-%06d", dataKey(loc, gen, ""), level, min, max) |
| + return fmt.Sprintf("%s%04x/%016x-%016x.ltx", dataKey(loc, gen, ""), level, min, max) |
| } |
| func snapKey(loc, gen string, max int) string { return ltxKey(loc, gen, 9, 1, max) } |
| +var ltxKeyRe = regexp.MustCompile(`^(.*/)([0-9a-f]{4})/([0-9a-f]{16})-([0-9a-f]{16})\.ltx$`) |
| + |
| // ltxLevel parses the level and range from an LTX key; ok is false for |
| // any other key. |
| func ltxLevel(k string) (level, min, max int, ok bool) { |
| - i := strings.Index(k, "/ltx/") |
| - if i < 0 { |
| + m := ltxKeyRe.FindStringSubmatch(k) |
| + if m == nil { |
| return 0, 0, 0, false |
| } |
| - n, err := fmt.Sscanf(k[i+5:], "%d/%d-%d", &level, &min, &max) |
| - return level, min, max, err == nil && n == 3 |
| + fmt.Sscanf(m[2], "%x", &level) |
| + fmt.Sscanf(m[3], "%x", &min) |
| + fmt.Sscanf(m[4], "%x", &max) |
| + return level, min, max, true |
| } |
| -// genPrefixOf is the generation prefix an LTX key belongs to. |
| +// genPrefixOf is the generation prefix an LTX key belongs to: the key |
| +// without its level directory and file name. |
| func genPrefixOf(k string) string { |
| - i := strings.Index(k, "/ltx/") |
| - if i < 0 { |
| + m := ltxKeyRe.FindStringSubmatch(k) |
| + if m == nil { |
| return "" |
| } |
| - return k[:i+1] |
| + return m[1] |
| +} |
| + |
| +// isLevelDir: a four-hex-digit child is litestream's level directory. |
| +func isLevelDir(child string) bool { |
| + if len(child) != 4 { |
| + return false |
| + } |
| + for _, r := range child { |
| + if !(r >= '0' && r <= '9' || r >= 'a' && r <= 'f') { |
| + return false |
| + } |
| + } |
| + return true |
| } |
| -// snapshotOf is the snapshot file of a database: every page, TXIDs 1 to |
| -// its current one, and the oracle's list of the writes it applied. |
| +// snapshotOf is the snapshot file of a database at its replication |
| +// 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. |
| func snapshotOf(db *localDB, inc int) *blob { |
| - return &blob{Kind: "snap", Level: 9, Min: 1, Max: db.TX, Inc: inc, Gen: db.Gen, SealKey: db.Key, |
| - Pages: cloneState(db.State), Post: sum(db.State), TXs: append([]int(nil), db.Log...)} |
| + return &blob{Kind: "snap", Level: 9, Min: 1, Max: db.Pos, Inc: inc, Gen: db.Gen, SealKey: db.Key, |
| + Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| } |
| -// deltaOf is the level-0 file of the transactions db has not replicated |
| -// yet: the final image of every key they touched. |
| +// deltaOf is the next level-0 file: one TXID, Pos+1, with the final |
| +// image of every key the pending writes touched. At position 0 it is the |
| +// 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 { |
| + 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.Rep + 1, Max: db.TX, Inc: inc, Gen: db.Gen, SealKey: db.Key, |
| - Pages: pages, Post: sum(db.State), TXs: append([]int(nil), db.Pending...)} |
| + 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...)} |
| +} |
| + |
| +// anchorOf is the level-0 anchor a cold restore writes before replication |
| +// starts (R5-1, the fact in litestreamfacts TestFact_R5_1_AnchorL0ContinuesFromN): |
| +// 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, |
| + Pages: cloneState(db.State), TXs: append([]int(nil), db.Log...)} |
| } |
| // apply overlays an LTX file's page images onto state. |
| @@ -186,10 +240,10 @@ func apply(state map[string]int, f *blob) { |
| } |
| // replicate is the serving process's litestream replication (claim C3's |
| -// exception: plain PUTs). It ships the pending transactions as one level-0 |
| -// file, retries the same range widened if the PUT timed out, and compacts |
| -// level 0 into level 1 once three files are current, deleting the inputs |
| -// after the compacted file is written, as litestream does. |
| +// exception: plain PUTs). It ships the pending writes as one level-0 file |
| +// at Pos+1 (an upload that times out is retried with the same bytes, so |
| +// the position advances at once), then runs compaction, snapshot |
| +// rotation and retention as the store's monitors would. |
| func (w *world) replicate(b *box) { |
| if !w.agentUp(b) || !hasMain(b) || b.DB == nil { |
| return |
| @@ -199,41 +253,53 @@ func (w *world) replicate(b *box) { |
| if len(db.Pending) > 0 { |
| bl := w.deltaOf(db, inc) |
| key := ltxKey(db.Loc, db.Gen, 0, bl.Min, bl.Max) |
| - if v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: bl}); v > 0 { |
| - db.Rep, db.Pending = db.TX, nil |
| - } |
| + w.issue(b.ID, &request{kind: rPut, region: b.Region, key: key, b: bl}) |
| + // the file is this TXID's whether the upload has completed or is |
| + // still in flight: litestream retries the same bytes, and later |
| + // writes go to the next TXID. A retry that lands after the |
| + // original is an identical version of the same key |
| + db.Pos, db.Pending = bl.Max, nil |
| } |
| w.compact(b) |
| } |
| -// compact merges the current level-0 files of b's generation into one |
| -// level-1 file when there are at least three, then deletes the inputs. |
| -// Contiguity is checked as ltx.Compactor does; a compacted file covers |
| -// [Min of first, Max of last] with the final image of every page touched. |
| -func (w *world) compact(b *box) { |
| - db := b.DB |
| - bk := w.buckets[b.Region] |
| - prefix := dataKey(db.Loc, db.Gen, "") + "ltx/0/" |
| - type in struct { |
| - k string |
| - v *version |
| - } |
| - var inputs []in |
| +// currentLTX lists the current LTX files of a generation at one level, by |
| +// range. |
| +func (w *world) currentLTX(bk *bucket, prefix string, level int) []*version { |
| + var out []*version |
| for _, k := range bk.keys(prefix) { |
| + l, _, _, ok := ltxLevel(k) |
| + if !ok || l != level || genPrefixOf(k) != prefix { |
| + continue |
| + } |
| if v := bk.current(k); v != nil { |
| - inputs = append(inputs, in{k, v}) |
| + out = append(out, v) |
| } |
| } |
| + sort.Slice(out, func(i, j int) bool { return out[i].B.Min < out[j].B.Min }) |
| + return out |
| +} |
| + |
| +// compact merges the current level-0 files of b's generation into one |
| +// level-1 file when there are at least three, deletes the inputs, then |
| +// writes a new snapshot and enforces retention (litestream's store after |
| +// a compaction at the snapshot level): only the newest snapshot stays, |
| +// and level-1 and level-0 files whose range ends below the previous |
| +// snapshot's TXID are deleted. Every delete is a key delete: the versions |
| +// stay noncurrent, which is what keeps the pinned snapshot version |
| +// reachable (R5-3). Contiguity is checked as ltx.Compactor does. |
| +func (w *world) compact(b *box) { |
| + db := b.DB |
| + bk := w.buckets[b.Region] |
| + prefix := dataKey(db.Loc, db.Gen, "") |
| + inputs := w.currentLTX(bk, prefix, 0) |
| if len(inputs) < 3 { |
| return |
| } |
| - sort.Slice(inputs, func(i, j int) bool { return inputs[i].v.B.Min < inputs[j].v.B.Min }) |
| - out := &blob{Kind: "ltx", Level: 1, Min: inputs[0].v.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{}} |
| max := 0 |
| var keys []string |
| - for _, x := range inputs { |
| - v := x.v |
| - keys = append(keys, x.k) |
| + for _, v := range inputs { |
| if max != 0 && !(v.B.Min <= max+1 && v.B.Max > max) { |
| return // non-contiguous inputs: litestream refuses to compact them |
| } |
| @@ -244,17 +310,46 @@ func (w *world) compact(b *box) { |
| } |
| } |
| out.TXs = append(out.TXs, v.B.TXs...) |
| - out.Post = v.B.Post |
| if v.B.Max > max { |
| max = v.B.Max |
| } |
| + 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 { |
| - for _, k := range keys { |
| - w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: k}) // a key delete: the versions stay noncurrent |
| + 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 { |
| + return |
| + } |
| + for _, k := range keys { |
| + w.issue(b.ID, &request{kind: rDelKey, region: b.Region, key: k}) |
| + } |
| + w.stats["ltx compactions"]++ |
| + // snapshot rotation: a new snapshot at the position, then retention |
| + prev := 0 |
| + for _, s := range w.currentLTX(bk, prefix, 9) { |
| + if s.B.Max > prev { |
| + prev = s.B.Max |
| + } |
| + } |
| + 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 { |
| + return |
| + } |
| + w.stats["snapshot rotations"]++ |
| + // keep the newest snapshot; minSnapshotTXID is the previous one's |
| + // MaxTXID (litestream db.EnforceSnapshotRetention); expire levels 1 |
| + // 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)}) |
| + } |
| + } |
| + for _, level := range []int{1, 0} { |
| + for _, f := range w.currentLTX(bk, prefix, level) { |
| + 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)}) |
| + } |
| } |
| - w.stats["ltx compactions"]++ |
| } |
| } |
| @@ -272,8 +367,8 @@ func (w *world) plan(bk *bucket, prefix string, cur int) (files []*blob, gap boo |
| var cands []*blob |
| for _, k := range bk.keys(prefix) { |
| level, _, _, ok := ltxLevel(k) |
| - if !ok || level == 9 { |
| - continue // snapshots are named by the marker, never planned |
| + if !ok || level == 9 || genPrefixOf(k) != prefix { |
| + continue // snapshots are the root, never planned |
| } |
| if v := bk.current(k); v != nil { |
| cands = append(cands, v.B) |
| @@ -307,6 +402,18 @@ func (w *world) plan(bk *bucket, prefix string, cur int) (files []*blob, gap boo |
| return files, gap |
| } |
| +// maxL0 is the newest current level-0 file's TXID of a generation: what |
| +// litestream's open takes as a fresh box's position (R5-1). |
| +func (w *world) maxL0(bk *bucket, prefix string) int { |
| + max := 0 |
| + for _, f := range w.currentLTX(bk, prefix, 0) { |
| + if f.B.Max > max { |
| + max = f.B.Max |
| + } |
| + } |
| + return max |
| +} |
| + |
| // expectState replays the incarnation's acknowledged, non-exempt writes, |
| // in acknowledgement order, into the database a correct restore must |
| // equal. With upto >= 0 it replays the transactions with lineage TXID 1 |
| @@ -335,3 +442,35 @@ func (w *world) expectState(inc int, upto int) map[string]int { |
| } |
| return state |
| } |
| + |
| +// txsOf is the union, in order, of the writes the files cover: the oracle's |
| +// log of a restored database. |
| +func txsOf(files ...*blob) []int { |
| + seen := map[int]bool{} |
| + var out []int |
| + for _, f := range files { |
| + for _, wr := range f.TXs { |
| + if !seen[wr] { |
| + seen[wr] = true |
| + out = append(out, wr) |
| + } |
| + } |
| + } |
| + return out |
| +} |
| + |
| +// lineageTX is the oracle's lineage TXID of a restored state: the highest |
| +// transaction id among the writes the applied files cover. |
| +func (w *world) lineageTX(files ...*blob) int { |
| + max := 0 |
| + for _, f := range files { |
| + for _, wr := range f.TXs { |
| + if t := w.txs[wr].TX; t > max { |
| + max = t |
| + } |
| + } |
| + } |
| + return max |
| +} |
| + |
| +var _ = strings.HasPrefix |
diff --git a/internal/sim/hostlife/moves_test.go b/internal/sim/hostlife/moves_test.go| index 023ee3aab..b2799cd0f 100644 |
| --- a/internal/sim/hostlife/moves_test.go |
| +++ b/internal/sim/hostlife/moves_test.go |
| @@ -48,6 +48,31 @@ func (w *world) workerMode(b *box, obl bool) { |
| w.role(b, c) |
| } |
| +// initRecord is claim rev 6.3 §6.4's initialization (R5-4): a box creates |
| +// or refreshes its local stamp record only from a fresh claim read, and |
| +// only when that claim names it as op.exec or as the landing participant; |
| +// it sets active_stamp = holder.stamp. A create's first wake, adoption on |
| +// the serving box, and a lost record on restart all go through here. A |
| +// stamp this box has retired is never made active again. |
| +func (w *world) initRecord(b *box, c *claim) { |
| + if c == nil || c.Holder == nil || c.Holder.Stamp == 0 || c.Ver != w.claim.Ver { |
| + return |
| + } |
| + inc := c.Holder.Inc |
| + if !(c.Op.Exec == b.ID || c.Op.Kind == "landing" && c.Op.Tgt == b.ID) { |
| + return |
| + } |
| + if b.stampOf(inc) == c.Holder.Stamp { |
| + return |
| + } |
| + if s := b.Stamps[inc]; s != nil && contains(s.Retired, c.Holder.Stamp) { |
| + return |
| + } |
| + b.install(inc, c.Holder.Stamp) |
| + c.readStamp = c.Holder.Stamp |
| + w.logf("box %d: local record for inc %d initialized from holder.stamp %d", b.ID, inc, c.Holder.Stamp) |
| +} |
| + |
| // resumeStale is a stalled worker goroutine of b resuming with a claim it |
| // read long ago (rev 4's actor). With claim C3's revalidation it re-reads |
| // under its lease, sees the world moved, and stops; the mutation acts on |
| @@ -66,6 +91,7 @@ func (w *world) resumeStale(b *box, old *claim) { |
| // role is the worker's exec-lease renewal, quarantine and operation role, |
| // acting on the claim c it read. |
| func (w *world) role(b *box, c *claim) { |
| + w.initRecord(b, c) |
| // exec lease (claim §7.1) |
| if w.isParticipant(c, b) && !contains(c.Lost, b.ID) { |
| if b.ExecUntil == 0 || b.ExecUntil-w.localNow(b) < execRenew { |
| @@ -177,11 +203,19 @@ func (w *world) snapshotBlobOf(b *box, c *claim, gen string) (*blob, *blob) { |
| // woken) the replica restored as it stands |
| var db *blob |
| if b.DB != nil { |
| + // the final checkpoint ships the pending writes as the next level-0 |
| + // file and waits for the upload (litestream SyncAndWait); the |
| + // snapshot is then at that position |
| + if len(b.DB.Pending) > 0 { |
| + bl := w.deltaOf(b.DB, b.DB.Inc) |
| + w.issue(b.ID, &request{kind: rPut, region: b.Region, key: ltxKey(b.DB.Loc, b.DB.Gen, 0, bl.Min, bl.Max), b: bl}) |
| + b.DB.Pos, b.DB.Pending = bl.Max, nil |
| + } |
| db = snapshotOf(b.DB, b.DB.Inc) |
| } else if r := w.restore(b, c.Holder.Inc, c.Holder.Location, gen); r != nil { |
| db = snapshotOf(r, c.Holder.Inc) |
| } else { |
| - db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}, Post: sum(nil)} |
| + db = &blob{Kind: "snap", Level: 9, Min: 1, Inc: c.Holder.Inc, Pages: map[string]int{}} |
| } |
| files := &blob{Kind: "files", Inc: c.Holder.Inc, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...)} |
| return db, files |
| @@ -656,7 +690,8 @@ func (w *world) followRecord(c *claim) bool { |
| return false |
| } |
| n.Op.Kind, n.Op.Exec, n.Op.LandPhase, n.Op.FlipAt = "landing", op.Tgt, "checking", w.now |
| - n.Op.FlipGen = w.nextGen() // the gen this CAS will have: the landing's installation stamp (claim rev 6.2 §6.4) |
| + n.Holder.Stamp = w.nextGen() // holder.stamp: the gen this CAS produces, the landing's installation stamp (claim rev 6.3 §2) |
| + n.Op.FlipGen = n.Holder.Stamp |
| return true |
| }) |
| case r.MoveState == "committed" && op.Kind == "landing" && !op.Committed: |
| @@ -714,15 +749,15 @@ func (w *world) landTarget(b *box, c *claim) { |
| // the new installation sets this box's active stamp to its own |
| // captured stamp, the claim gen of the CAS that made the landing |
| // current, under the guard, on a fresh read showing its operation |
| - // current and this box the landing participant (claim rev 6.2 |
| + // current and this box the landing participant (claim rev 6.3 |
| // §6.4, R3-3, R4-2): a move back to a region the instance left is |
| // admitted at its own stamp while the retired installation's |
| // writers stay refused |
| if fc, ok := w.readClaim(b); ok && fc.Op.ID == op.ID && fc.Op.Kind == "landing" && fc.Op.Tgt == b.ID { |
| retiredHere := b.Stamps[c.Holder.Inc] != nil && len(b.Stamps[c.Holder.Inc].Retired) > 0 |
| if on("fence-by-installation") || !retiredHere { |
| - b.install(c.Holder.Inc, op.FlipGen) |
| - w.logf("box %d: installation stamp %d for inc %d (%s)", b.ID, op.FlipGen, c.Holder.Inc, op.ID) |
| + b.install(c.Holder.Inc, fc.Holder.Stamp) |
| + w.logf("box %d: installation stamp %d for inc %d (%s), from holder.stamp", b.ID, fc.Holder.Stamp, c.Holder.Inc, op.ID) |
| } |
| } |
| w.cas(c, func(n *claim) bool { |
| @@ -802,6 +837,7 @@ func (w *world) landTarget(b *box, c *claim) { |
| case "restore": |
| if b.DB == nil || b.DB.Gen != op.ID { |
| if b.DB != nil { |
| + b.retireArtifacts(c.Holder.Inc, b.DB.Stamp) |
| b.Parked++ |
| } |
| b.DB = w.restore(b, c.Holder.Inc, loc, op.ID) |
| @@ -887,8 +923,15 @@ func (w *world) installLocal(b *box, c *claim, gen string, seal ref, _ string, n |
| return |
| } |
| if b.Files == nil || b.Files.Gen != gen { |
| - // fresh extraction: park whatever is there first |
| - if b.DB != nil || b.Files != nil { |
| + // fresh extraction: park whatever is there first, as |
| + // retire-artifacts of the predecessor's stamp (claim rev 6.3 |
| + // §6.4): the new installation's active stamp is untouched |
| + if b.DB != nil { |
| + b.retireArtifacts(c.Holder.Inc, b.DB.Stamp) |
| + b.Parked++ |
| + } |
| + if b.Files != nil { |
| + b.retireArtifacts(c.Holder.Inc, b.Files.Stamp) |
| b.Parked++ |
| } |
| b.DB = nil |
| @@ -909,7 +952,7 @@ func (w *world) installLocal(b *box, c *claim, gen string, seal ref, _ string, n |
| return |
| } |
| f := src.B.Files.clone() |
| - f.Gen, f.Done = gen, false |
| + f.Gen, f.Done, f.Stamp = gen, false, b.stampOf(c.Holder.Inc) |
| if !on("markers-last") { |
| f.Done = true |
| } |
| @@ -1155,6 +1198,7 @@ func (w *world) recoverSource(b *box, c *claim) { |
| w.cas(c, func(n *claim) bool { |
| w.newGen(n.Holder.Location, n.Holder.Gen, op.RollbackID) |
| n.Holder.Gen = op.RollbackID |
| + n.Holder.Stamp = w.nextGen() // a dirty rollback installs a new generation: a new stamp, never the predecessor's (claim rev 6.3 §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) |
| }) |
| @@ -1189,7 +1233,6 @@ func (w *world) pushStep(b *box, c *claim) { |
| w.writeInc[mark] = c.Holder.Inc |
| snap.Max++ |
| snap.Pages[fmt.Sprintf("k%d", mark)] = mark |
| - snap.Post = sum(snap.Pages) |
| snap.TXs = append(snap.TXs, mark) |
| w.txs[mark] = tx{W: mark, TX: snap.Max, Key: fmt.Sprintf("k%d", mark), Val: mark} |
| w.cas(c, func(n *claim) bool { |
| @@ -1226,6 +1269,7 @@ func (w *world) pushStep(b *box, c *claim) { |
| } |
| case "restore": |
| if b.DB != nil && b.DB.Gen != op.ID { |
| + b.retireArtifacts(c.Holder.Inc, b.DB.Stamp) |
| b.Parked++ |
| b.DB = nil |
| } |
| @@ -1256,6 +1300,7 @@ func (w *world) pushStep(b *box, c *claim) { |
| n.Op.Phase = "probed" |
| w.newGen(n.Holder.Location, n.Holder.Gen, op.ID) |
| n.Holder.Gen = op.ID |
| + n.Holder.Stamp = w.nextGen() // a push's commit installs a new generation: a new stamp (claim rev 6.3 §2) |
| q := op.StopAt + horizonH |
| if q > n.Quiet[b.Region] { |
| n.Quiet[b.Region] = q |
| @@ -1362,13 +1407,10 @@ func locationOf(k string) string { |
| rest := strings.TrimPrefix(k, p) |
| if i := strings.Index(rest, "/"); i > 0 { |
| child := rest[:i] |
| - digits := len(child) > 0 |
| - for _, r := range child { |
| - if r < '0' || r > '9' { |
| - digits = false |
| - } |
| - } |
| - if digits { |
| + // a level directory (0000 to 0009, litestream's layout) directly |
| + // under the host is the legacy location; anything else is an |
| + // incarnation location (claim rev 6 §6.2a) |
| + if !isLevelDir(child) { |
| return p + child + "/" |
| } |
| } |
| @@ -1377,7 +1419,7 @@ func locationOf(k string) string { |
| // deadGeneration: within holder.location, everything but the live set. |
| // The live set is {authoritative region, holder.location, |
| -// holder.generation} (claim rev 6.2 §6.2a, R4-3): the generation's |
| +// holder.generation} (claim rev 6.3 §6.2a, R4-3): the generation's |
| // objects are live only in the authoritative region's bucket, the region |
| // of the holder's current installation; a matching key in any other |
| // region is residue. The mutation of live-set-region exempts matching keys |
| @@ -1393,7 +1435,7 @@ func (w *world) deadGeneration(c *claim, r int, k string) bool { |
| v := w.buckets[r].current(k) |
| return v != nil && (v.B.Gen != h.Gen || !authoritative) |
| } |
| - if strings.HasPrefix(k, dataKey(h.Location, h.Gen, "ltx/")) { |
| + if isDataKey(k) && genPrefixOf(k) == dataKey(h.Location, h.Gen, "") { |
| return !authoritative |
| } |
| return isDataKey(k) |
diff --git a/internal/sim/hostlife/protocol_test.go b/internal/sim/hostlife/protocol_test.go| index 5a293f31c..daefde62f 100644 |
| --- a/internal/sim/hostlife/protocol_test.go |
| +++ b/internal/sim/hostlife/protocol_test.go |
| @@ -23,7 +23,7 @@ func (w *world) readClaim(b *box) (*claim, bool) { |
| } |
| c := w.claim.clone() |
| if c != nil && c.Holder != nil { |
| - c.readStamp = b.stampOf(c.Holder.Inc) // the installation stamp this read carries (claim rev 6.2 §6.4) |
| + c.readStamp = b.stampOf(c.Holder.Inc) // the installation stamp this read carries (claim rev 6.3 §6.4) |
| } |
| b.snap = c.clone() |
| return c, true |
| @@ -165,16 +165,24 @@ func (w *world) evidence(b *box, opID string) bool { |
| // ---------- restore and spawn ---------- |
| // restore rebuilds a database from location loc, generation gen, in b's own |
| -// bucket (claim C3a; moves rev 7.5 §5.1, R4-1), exactly as litestream's |
| -// restore would: it opens the snapshot that generation.json names (by key, |
| -// version id and TXID), then applies the LTX files the planner selects |
| -// (ltx_test.go plan), verifies the database checksum against the last |
| -// file's, and fails closed on a gap, a missing or mismatched snapshot, or |
| -// a checksum mismatch. It never reads another location or generation, and |
| -// it never knows which transactions a file holds. The mutation of |
| -// restore-by-generation restores the newest single object anywhere under |
| -// the location, which is what a prefix restore without generation identity |
| -// does. |
| +// bucket (claim C3a; moves rev 7.6 §5.1, R4-1, R5-1, R5-3), as the pinned |
| +// litestream does: it roots at the newest current snapshot of the |
| +// generation at or past the pinned one (the executor's own rotation |
| +// writes successors; the pinned version is the fallback when none is |
| +// current), applies the files the planner selects (ltx_test.go plan), |
| +// and fails closed on a gap or a missing root. There is no database |
| +// checksum to compare (R5-2): a file is verified by decoding, and the |
| +// restored database by SQLite's integrity check, which the model's files |
| +// never fail. Then, when no level-0 file at or past the restored position |
| +// is current (a snapshot-only install), the activator writes the level-0 |
| +// anchor [cur, cur] before replication starts, so litestream's position |
| +// is cur (R5-1); the mutation of restore-anchor leaves the position to |
| +// litestream's open, which takes the newest remote level-0 file, or |
| +// nothing. The mutation of root-advance roots at the pinned snapshot only |
| +// (rev 7.5), and refuses once retention has expired its continuation. The |
| +// mutation of restore-by-generation restores the newest single object |
| +// anywhere under the location, which is what a prefix restore without |
| +// generation identity does. |
| func (w *world) restore(b *box, inc int, loc, gen string) *localDB { |
| bk := w.buckets[b.Region] |
| db := &localDB{Inc: inc, Loc: loc, Gen: gen, State: map[string]int{}} |
| @@ -192,36 +200,62 @@ func (w *world) restore(b *box, inc int, loc, gen string) *localDB { |
| } |
| if best != nil { |
| apply(db.State, best.B) |
| - db.TX, db.Rep, db.Key = best.B.Max, best.B.Max, best.B.SealKey |
| + db.Pos, db.Key = best.B.Max, best.B.SealKey |
| + db.Log = txsOf(best.B) |
| + db.TX = w.lineageTX(best.B) |
| if best.B.Inc != 0 { |
| db.Inc = best.B.Inc |
| } |
| } |
| + db.Stamp = b.stampOf(inc) |
| return db |
| } |
| + prefix := dataKey(loc, gen, "") |
| cur := 0 |
| - var last *blob |
| + var applied []*blob |
| + pinnedTX := 0 |
| if gen != "" { |
| m := bk.current(genMarkerKey(loc)) |
| if m == nil || m.B.Gen != gen { |
| return nil // the marker does not name this generation: refuse |
| } |
| + pinnedTX = m.B.TX |
| + } |
| + // the root: the newest current snapshot of the generation at or past the |
| + // pinned one (for the origin generation, which has no pinned snapshot, |
| + // litestream's own rule: the newest snapshot); the pinned version only |
| + // when no successor is current |
| + var root *blob |
| + if on("root-advance") || gen == "" { |
| + for _, s := range w.currentLTX(bk, prefix, 9) { |
| + if s.B.Max >= pinnedTX && (root == nil || s.B.Max > root.Max) { |
| + root = s.B |
| + } |
| + } |
| + } |
| + if root == nil && gen != "" { |
| + 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 { |
| w.stats["restore refused: snapshot missing or not at the marker's TXID"]++ |
| return nil |
| } |
| - apply(db.State, snap.B) |
| - cur, last = snap.B.Max, snap.B |
| - db.Key = snap.B.SealKey |
| - if snap.B.Inc != 0 { |
| - db.Inc = snap.B.Inc |
| + root = snap.B |
| + } |
| + if root != nil { |
| + apply(db.State, root) |
| + cur = root.Max |
| + applied = append(applied, root) |
| + db.Key = root.SealKey |
| + if root.Inc != 0 { |
| + db.Inc = root.Inc |
| } |
| } |
| - files, gap := w.plan(bk, dataKey(loc, gen, "ltx/"), cur) |
| + files, gap := w.plan(bk, prefix, cur) |
| for _, f := range files { |
| apply(db.State, f) |
| - cur, last = f.Max, f |
| + cur = f.Max |
| + applied = append(applied, f) |
| if db.Key == 0 { |
| db.Key = f.SealKey |
| } |
| @@ -230,27 +264,33 @@ func (w *world) restore(b *box, inc int, loc, gen string) *localDB { |
| } |
| } |
| if gap && on("restore-gap-closed") { |
| - w.logf("box %d restore refused: a gap after TXID %d under %s", b.ID, cur, dataKey(loc, gen, "ltx/")) |
| + w.logf("box %d restore refused: a gap after TXID %d under %s", b.ID, cur, prefix) |
| w.stats["restore refused: gap"]++ |
| return nil |
| } |
| - if last != nil && sum(db.State) != last.Post { |
| - w.logf("box %d restore refused: checksum mismatch after TXID %d", b.ID, cur) |
| - w.stats["restore refused: checksum"]++ |
| + if len(applied) == 0 && gen != "" { |
| return nil |
| } |
| - if last == nil && gen != "" { |
| - return nil |
| + // the oracle's bookkeeping: the writes this database holds, which its |
| + // own later snapshots and anchors carry forward |
| + db.Log = txsOf(applied...) |
| + db.TX = w.lineageTX(applied...) |
| + db.Stamp = b.stampOf(inc) |
| + // restore-to-replication initialization (R5-1, litestreamfacts |
| + // TestFact_R5_1_*): litestream's open takes a fresh box's position from |
| + // the newest remote level-0 file, or 0. When that is below the restored |
| + // position (a snapshot-only install), the activator writes the anchor, |
| + // every page as the level-0 file [cur, cur], before replication starts |
| + db.Pos = w.maxL0(bk, prefix) |
| + if cur > 0 && db.Pos < cur && on("restore-anchor") { |
| + db.Pos = cur |
| + w.issue(b.ID, &request{kind: rPut, region: b.Region, key: ltxKey(loc, gen, 0, cur, cur), b: anchorOf(db, inc)}) |
| + w.stats["restore anchors"]++ |
| } |
| - // the restored database's replication position is where the restore |
| - // reached: its next LTX file starts at cur+1 (litestream's -txid |
| - // sidecar). Rev 7.5's first run shipped [1, TX] with only the new pages |
| - // after a cold restore; the checksum check refused it, correctly |
| - db.TX, db.Rep = cur, cur |
| return db |
| } |
| -// stampOK is admission's local check (claim rev 6.2 §6.4): the worker's |
| +// stampOK is admission's local check (claim rev 6.3 §6.4): the worker's |
| // read of the claim carries the installation stamp active on this box at |
| // that read; it is admitted only if that is still the active stamp. A |
| // landing probe refused here by its own installation is the fault R3-3 |
| @@ -295,7 +335,9 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| } |
| if newest != nil { |
| apply(b.DB.State, newest.B) |
| - b.DB.TX, b.DB.Rep, b.DB.Key = newest.B.Max, newest.B.Max, newest.B.SealKey |
| + b.DB.Pos, b.DB.Key = newest.B.Max, newest.B.SealKey |
| + b.DB.Log = txsOf(newest.B) |
| + b.DB.TX = w.lineageTX(newest.B) |
| if newest.B.Inc != 0 { |
| b.DB.Inc = newest.B.Inc |
| } |
| @@ -315,8 +357,9 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| return false |
| } |
| // a warm database from an earlier stay is parked, and the box |
| - // restores cold (§5.1) |
| + // restores cold (§5.1): retire-artifacts of its stamp |
| if b.DB != nil && (b.DB.Gen != bg || b.DB.Loc != loc) { |
| + b.retireArtifacts(h.Inc, b.DB.Stamp) |
| b.Parked++ |
| b.DB = nil |
| } |
| @@ -351,7 +394,7 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool { |
| } |
| if b.Files == nil { |
| w.nextID++ |
| - b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: want, Env: w.nextID, Done: true} |
| + b.Files = &filesDir{Inc: h.Inc, Key: w.nextID, Gen: want, Env: w.nextID, Done: true, Stamp: b.stampOf(h.Inc)} |
| } |
| if b.DB.Key == 0 { |
| b.DB.Key = b.Files.Key // unsealed data (never booted) adopts the key |
| @@ -401,7 +444,7 @@ func (w *world) admits(b *box, c *claim, probe string) bool { |
| return false |
| } |
| if !w.stampOK(b, c, probe) { |
| - return false // a retired installation's worker (claim rev 6.2 §6.4) |
| + return false // a retired installation's worker (claim rev 6.3 §6.4) |
| } |
| if w.isParticipant(c, b) && b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil { |
| return false // self-fenced until the lease is renewed |
| @@ -548,9 +591,13 @@ func (w *world) parkDepth(b *box, prefix, dest string, quiet int, keep func(stri |
| // guard: a retired marker first, then each artifact renamed aside, one per |
| // call, so a crash can fall between any two (claim §6.3). It reports |
| // whether nothing is left. |
| +// localPark is claim rev 6.3's retire-installation (§6.3): the active |
| +// stamp moves into retired_stamps and becomes none, then every artifact is |
| +// parked. It serves a delete, a move away and the quarantine; a landing |
| +// parks its predecessor's leftovers with retire-artifacts instead. |
| func (w *world) localPark(b *box, inc int) bool { |
| if on("hostguard") && b.stampOf(inc) != 0 { |
| - b.retireStamp(inc) // the active stamp moves into retired_stamps (claim rev 6.2 §6.4) |
| + b.retireStamp(inc) // retire-installation: active moves into retired_stamps (claim rev 6.3 §6.4) |
| return false |
| } |
| switch { |
diff --git a/internal/sim/hostlife/rules_test.go b/internal/sim/hostlife/rules_test.go| index 5d2eeec70..8fc006790 100644 |
| --- a/internal/sim/hostlife/rules_test.go |
| +++ b/internal/sim/hostlife/rules_test.go |
| @@ -54,8 +54,8 @@ var allRules = []string{ |
| "park-skip-lock", // §6.2 step 2/J-N6 |
| "retirement-ids", // §6.2/J-D5/J-N7 |
| "hostguard", // §6.4/J-N8 |
| - "fence-by-installation", // claim rev 6.2 §6.4 (R3-3): a new installation of a retired incarnation sets its own active stamp; the mutation is rev 5.2's marker, retired for good |
| - "fence-stamp", // claim rev 6.2 §6.4 (R4-2): a writer is admitted only if its stamp equals the active stamp; the mutation admits any writer for the incarnation while one is active |
| + "fence-by-installation", // claim rev 6.3 §6.4 (R3-3): a new installation of a retired incarnation sets its own active stamp; the mutation is rev 5.2's marker, retired for good |
| + "fence-stamp", // claim rev 6.3 §6.4 (R4-2): a writer is admitted only if its stamp equals the active stamp; the mutation admits any writer for the incarnation while one is active |
| "purge-reserve-first", // §5.3/J-N9 |
| "purge-adopts", // this model's finding: purge claims unclaimed legacy records first |
| "purge-journal", // this model's finding: the reservation is the purge's journal |
| @@ -82,11 +82,13 @@ var allRules = []string{ |
| "restore-by-generation", // C3a + moves rev 6 §5.1: restore replays the claim's generation only |
| "restore-overlap", // rev 7.5 §5.1 (R4-1): the planner takes a file whose range overlaps the one restored so far (litestream's rule); the mutation requires min == current+1 and discards a compacted continuation |
| "restore-gap-closed", // rev 7.5 §5.1 (R4-1): a current file that starts beyond current+1 once nothing extends the range is a gap, and the restore refuses; the mutation returns what it has |
| + "restore-anchor", // rev 7.6 §5.1 (R5-1): a cold restore with no level-0 file at its position writes the anchor [cur,cur] before replication starts; the mutation leaves the position to litestream's open |
| + "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 |
| "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>/ |
| "sweep-namespace", // claim rev 6 §6.2a (R3-4): the sweep discovers locations from the namespace; the mutation reads the capped retired_incarnations |
| - "live-set-region", // claim rev 6.2 §6.2a (R4-3): the live set is {authoritative region, location, generation}; the mutation exempts matching keys in every region |
| + "live-set-region", // claim rev 6.3 §6.2a (R4-3): the live set is {authoritative region, location, generation}; the mutation exempts matching keys in every region |
| // moves spec |
| "seal-by-version", // J-N2: landing copies sealed versions |
| "staging-quiet", // J-N2: staging retired after its writers' horizon |
diff --git a/internal/sim/hostlife/sim_test.go b/internal/sim/hostlife/sim_test.go| index e6e8687ac..a544e3019 100644 |
| --- a/internal/sim/hostlife/sim_test.go |
| +++ b/internal/sim/hostlife/sim_test.go |
| @@ -48,24 +48,27 @@ func newWorld(seed int64, verbose bool) *world { |
| } |
| // retentionStart: the host has been recreated more times than |
| -// retired_incarnations remembers (claim rev 6.2 §2 caps it at 32), and a |
| +// retired_incarnations remembers (claim rev 6.3 §2 caps it at 32), and a |
| // replication object of an incarnation the list forgot is still in a |
| // bucket (rev 7.5, R4 §3: the retention-boundary fixture). Only a sweep |
| // that discovers locations from the namespace parks it. |
| func (w *world) retentionStart() { |
| w.claim = &claim{Ver: w.newVer()} |
| - for i := 0; i < retiredCap; i++ { |
| + old := 1 // an incarnation older than every one the list remembers |
| + w.nextInc = old |
| + for i := 0; i < retiredCap+1; i++ { |
| w.nextInc++ |
| - w.claim.Retired = retire(w.claim.Retired, w.nextInc) |
| + w.claim.Retired = retire(w.claim.Retired, w.nextInc) // 2 to 33: the cap drops nothing but 1 was never listed |
| + } |
| + if contains(w.claim.Retired, old) || len(w.claim.Retired) != retiredCap { |
| + panic("retention fixture: the injected incarnation must be absent from a full list (R5-6)") |
| } |
| - old := 1 // an incarnation older than every one the list remembers |
| - w.nextInc++ |
| r := w.rng.Intn(nRegions) |
| k := ltxKey(locationFor(old, false), "", 0, 1, 1) |
| // a request that completed long after its incarnation was retired: a |
| // straggler, which only the sweep retires (INV-10 at an operation's end |
| // exempts it, as it exempts every completion after H) |
| - w.buckets[r].objs[k] = []*version{{ID: w.newVer(), Straggler: true, B: &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: old, Pages: map[string]int{"k0": 1}, Post: sum(map[string]int{"k0": 1})}}} |
| + w.buckets[r].objs[k] = []*version{{ID: w.newVer(), Straggler: true, B: &blob{Kind: "ltx", Level: 0, Min: 1, Max: 1, Inc: old, Pages: map[string]int{"k0": 1}}}} |
| w.logf("retention fixture: incarnation %d's replica in region %d, past the cap", old, r) |
| } |
| @@ -93,13 +96,13 @@ func (w *world) legacyStart() { |
| // the pre-release replica: one level-0 file of everything, untagged |
| full := w.deltaOf(b.DB, 0) |
| full.Inc = 0 |
| - b.DB.Rep, b.DB.Pending = b.DB.TX, nil |
| - k := ltxKey(b.DB.Loc, "", 0, 1, b.DB.TX) |
| + b.DB.Pos, b.DB.Pending = 1, nil |
| + k := ltxKey(b.DB.Loc, "", 0, 1, 1) |
| w.buckets[r].objs[k] = []*version{{ID: w.newVer(), B: full}} |
| w.markDurable(k, full) |
| // an earlier move's replica in another region: the first transaction only |
| 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}, Post: sum(map[string]int{first.Key: first.Val}), TXs: []int{first.W}} |
| + 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}} |
| other := (r + 1 + w.rng.Intn(nRegions-1)) % nRegions |
| w.buckets[other].objs[ltxKey(b.DB.Loc, "", 0, 1, 1)] = []*version{{ID: w.newVer(), B: part}} |
| if w.rng.Intn(2) == 0 { |
| @@ -109,7 +112,7 @@ func (w *world) legacyStart() { |
| // that earlier move also left the old box's local state behind: |
| // a database, a files directory and env files, and no row |
| ob := w.regionBox(other) |
| - ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Rep: 1} |
| + ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Pos: 1} |
| ob.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true} |
| } |
| } |
| @@ -335,7 +338,7 @@ func (w *world) residueOf(exceptStragglers bool) []string { |
| if e := w.execBox(c); e != nil { |
| serving = e.Region |
| loc, gen := c.Holder.Location, c.Holder.Gen |
| - live[dataKey(loc, gen, "ltx/")] = true // a prefix: every current LTX file of the generation |
| + live[dataKey(loc, gen, "")] = true // the generation's prefix: every LTX file of the generation |
| live[genMarkerKey(loc)] = true |
| } |
| } |
| @@ -343,7 +346,7 @@ func (w *world) residueOf(exceptStragglers bool) []string { |
| 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] || live[genPrefixOf(k)+"ltx/"] && genPrefixOf(k) != "")) { |
| + if strings.HasSuffix(k, "lock.json") || (r == serving && (live[k] || isDataKey(k) && live[genPrefixOf(k)])) { |
| continue |
| } |
| if exceptStragglers && w.owed(r, k) { |
| @@ -480,7 +483,7 @@ func (w *world) checkDurable(c *claim, when string) { |
| continue |
| } |
| if db == nil || w.txs[a.W].TX > db.TX { |
| - w.violate("INV-2 (cold restore, %s): durable write %d (box %d, t=%d) lost", when, a.W, a.Box, a.At) |
| + 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 |
| } |
| } |
| @@ -498,12 +501,18 @@ func (w *world) ltxShape(r int, prefix string) string { |
| var parts []string |
| bk := w.buckets[r] |
| for _, k := range bk.keys(prefix) { |
| - v := bk.current(k) |
| - if v == nil { |
| - parts = append(parts, strings.TrimPrefix(k, prefix)+":deleted") |
| + if !isDataKey(k) { |
| continue |
| } |
| - parts = append(parts, fmt.Sprintf("%s:%v", strings.TrimPrefix(k, prefix), v.B.Pages)) |
| + var vs []string |
| + for _, v := range bk.objs[k] { |
| + if v.Marker { |
| + vs = append(vs, "M") |
| + } else { |
| + vs = append(vs, fmt.Sprintf("v%d", v.ID)) |
| + } |
| + } |
| + parts = append(parts, strings.TrimPrefix(k, prefix)+":"+strings.Join(vs, ",")) |
| } |
| return "files " + strings.Join(parts, " ") |
| } |
diff --git a/internal/sim/hostlife/witness_test.go b/internal/sim/hostlife/witness_test.go| index 000688553..f90054a65 100644 |
| --- a/internal/sim/hostlife/witness_test.go |
| +++ b/internal/sim/hostlife/witness_test.go |
| @@ -103,7 +103,12 @@ var witnesses = []witness{ |
| {"restore-overlap", "restore-overlap", "INV-2", |
| "a generation's snapshot ends at TXID 11 and the only continuation is a compacted file [10,15]; litestream's planner takes it and restores through 15; rev 7.4's rule (min == current+1) discards it and the wake loses four acknowledged writes (R4-1)", wRestoreOverlap}, |
| {"restore-gap-closed", "restore-gap-closed", "INV-2 (cold restore)", |
| - "a generation's snapshot ends at TXID 11, the file [12,12] is missing and [13,13] is current: a genuine gap; the restore refuses; returning what it has would wake the host without durable transactions (R4-1)", wRestoreGap}, |
| + "a generation's snapshot ends at TXID n, the file [n+1,n+1] is missing and [n+2,n+2] is current: a genuine gap; the restore refuses; returning what it has would wake the host without durable transactions (R4-1)", wRestoreGap}, |
| + // rev 7.6 |
| + {"restore-anchor", "restore-anchor", "INV-2", |
| + "a snapshot-only install (a push), then writes and replication, then the disk is lost: without the level-0 anchor litestream's position restarts at 1, the new files sit below the snapshot, and the cold restore loses the writes (R5-1; litestreamfacts TestFact_R5_1_*)", wSnapshotOnlyInstall}, |
| + {"root-advance", "root-advance", "INV-3 (witness): woken cold", |
| + "after a push, four compaction-and-snapshot rotations with retention; the disk is lost; a restore rooted at the pinned snapshot finds its continuation expired and refuses, while one rooted at the newest current successor snapshot succeeds (R5-3; litestreamfacts TestFact_R5_3_SnapshotRetention)", wRootAdvance}, |
| } |
| // TestWitnesses is the deterministic half of the mutation evidence. |
| @@ -285,7 +290,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} |
| + n.Holder = &holder{Inc: inc, App: app, Serial: w.serial, Location: loc, Stamp: n.Ver} // the serving CAS wrote holder.stamp |
| 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 |
| @@ -294,9 +299,10 @@ func (w *world) serve(b *box, inc int, app string) { |
| b.DB = db |
| b.Files = &filesDir{Inc: inc, Key: key, Env: key, Done: true} |
| b.Config, b.CfgVer, b.Live, b.LiveInc = inc, w.cfgVer[app], app, inc |
| - b.install(inc, n.Ver) // the create's installation stamp |
| + b.install(inc, n.Ver) // the create's installation stamp, initialized from holder.stamp |
| + db.Stamp = n.Ver |
| bl := w.deltaOf(db, inc) |
| - db.Rep, db.Pending = db.TX, nil |
| + db.Pos, db.Pending = 1, nil |
| 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) |
| @@ -322,12 +328,12 @@ func (w *world) legacy(b *box, other int) { |
| b.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true} |
| b.Config, b.CfgVer, b.Live, b.LiveInc = inc, 1, app0, inc |
| full := w.deltaOf(b.DB, 0) |
| - b.DB.Rep, b.DB.Pending = b.DB.TX, nil |
| - k := ltxKey(b.DB.Loc, "", 0, 1, b.DB.TX) |
| + b.DB.Pos, b.DB.Pending = 1, nil |
| + k := ltxKey(b.DB.Loc, "", 0, 1, 1) |
| w.buckets[b.Region].objs[k] = []*version{{ID: w.newVer(), B: full}} |
| w.markDurable(k, full) |
| 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}, Post: sum(map[string]int{first.Key: first.Val}), TXs: []int{first.W}} |
| + 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}} |
| } |
| @@ -939,7 +945,7 @@ func wSweepNamespace(t *testing.T, w *world) { |
| t.Fatalf("witness: the host did not wake: %s", w.state()) |
| } |
| w.appWrite(b) |
| - key := dataKey(w.claim.Holder.Location, "", "ltx/") |
| + key := dataKey(w.claim.Holder.Location, "", "") + "0000/" |
| w.holdKey(key) |
| w.replicate(b) |
| w.hold = nil |
| @@ -1004,7 +1010,7 @@ func (w *world) roundTrip(t *testing.T, uk, us *box) { |
| }, w.act(uk, us), w.recoveryPass, w.operator) |
| } |
| -// wStaleWriter (claim rev 6.2 §6.4, R4-2): a config-sync pass selects the |
| +// wStaleWriter (claim rev 6.3 §6.4, R4-2): a config-sync pass selects the |
| // row under the first installation and pauses; the host moves away and |
| // back; a new bundle is published; the pass takes the guard. |
| func wStaleWriter(t *testing.T, w *world) { |
| @@ -1026,7 +1032,7 @@ func wStaleWriter(t *testing.T, w *world) { |
| w.finish(t) |
| } |
| -// wLiveSetRegion (claim rev 6.2 §6.2a, R4-3): a legacy host serves in UK |
| +// wLiveSetRegion (claim rev 6.3 §6.2a, R4-3): a legacy host serves in UK |
| // (box 0) with a pre-release replica in US (box 1). Adoption's discovery |
| // parks US. A replication PUT the old US executor issued before the |
| // retirement completes after it. |
| @@ -1034,14 +1040,14 @@ func wLiveSetRegion(t *testing.T, w *world) { |
| uk, us := w.boxes[0], w.boxes[1] |
| w.legacy(uk, us.Region) |
| // the old US executor's last replication PUT, accepted, completing late |
| - late := &blob{Kind: "ltx", Level: 0, Min: 1, Max: 2, Inc: 0, SealKey: 7, Pages: cloneState(uk.DB.State), Post: sum(uk.DB.State)} |
| - w.pending = append(w.pending, &request{kind: rPut, region: us.Region, key: ltxKey("instances/"+host+"/", "", 0, 1, 2), b: late, doneAt: 1 << 30, by: "box:1"}) |
| + late := &blob{Kind: "ltx", Level: 0, Min: 2, Max: 2, Inc: 0, SealKey: 7, Pages: cloneState(uk.DB.State)} |
| + w.pending = append(w.pending, &request{kind: rPut, region: us.Region, key: ltxKey("instances/"+host+"/", "", 0, 2, 2), b: late, doneAt: 1 << 30, by: "box:1"}) |
| w.tick(func() { w.reconcile(uk) }) // adoption |
| if w.claim == nil { |
| t.Fatalf("witness: not adopted") |
| } |
| w.reach(t, 400, "discovery settled", func() bool { return w.claim.oblFor("adopt") == 0 }, w.actAll(), w.recoveryPass) |
| - if w.completeHeld(ltxKey("instances/"+host+"/", "", 0, 1, 2)) != 1 { |
| + if w.completeHeld(ltxKey("instances/"+host+"/", "", 0, 2, 2)) != 1 { |
| t.Fatalf("witness: no held PUT") |
| } |
| w.finish(t) |
| @@ -1062,20 +1068,19 @@ func (w *world) pushed(t *testing.T, b *box) (loc, gen string) { |
| return w.claim.Holder.Location, gen |
| } |
| -// compacted builds the level-1 file [min, max] of b's lineage from the |
| -// oracle's transactions: the final image of every key touched in the |
| -// range, the checksum at max. The witness places it where compaction |
| -// would, and removes the level-0 files it replaces. |
| -func (w *world) compacted(b *box, min, max int) *blob { |
| +// compactedPending builds the level-1 file [min, max], in replication |
| +// numbering, that compaction would have left for b's pending writes: the |
| +// final image of every key they touched. The witness places it where |
| +// compaction would and advances the position past it. |
| +func (w *world) compactedPending(b *box, min, max int) *blob { |
| db := b.DB |
| - out := &blob{Kind: "ltx", Level: 1, Min: min, Max: max, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}, Post: sum(db.State)} |
| - for _, wr := range db.Log { |
| + out := &blob{Kind: "ltx", Level: 1, Min: min, Max: max, Inc: db.Inc, Gen: db.Gen, SealKey: db.Key, Pages: map[string]int{}} |
| + for _, wr := range db.Pending { |
| t := w.txs[wr] |
| - if t.TX >= min && t.TX <= max { |
| - out.Pages[t.Key] = db.State[t.Key] |
| - out.TXs = append(out.TXs, wr) |
| - } |
| + out.Pages[t.Key] = db.State[t.Key] |
| + out.TXs = append(out.TXs, wr) |
| } |
| + db.Pos, db.Pending = max, nil |
| return out |
| } |
| @@ -1095,19 +1100,17 @@ func wRestoreOverlap(t *testing.T, w *world) { |
| if len(w.violations) > 0 { |
| return |
| } |
| - if b.DB.TX != 11 { |
| - t.Fatalf("witness: expected TXID 11 after the push, got %d", b.DB.TX) |
| - } |
| + n := b.DB.Pos // the snapshot's TXID after the push, in replication numbering |
| for i := 0; i < 4; i++ { |
| w.appWrite(b) |
| } |
| - f := w.compacted(b, 10, 15) |
| - w.buckets[b.Region].objs[ltxKey(loc, gen, 1, 10, 15)] = []*version{{ID: w.newVer(), B: f}} |
| - w.markDurable(ltxKey(loc, gen, 1, 10, 15), f) |
| - b.DB.Rep, b.DB.Pending = 15, nil |
| + // compaction has left one file [n-1, n+2] overlapping the snapshot |
| + f := w.compactedPending(b, n-1, n+2) |
| + w.buckets[b.Region].objs[ltxKey(loc, gen, 1, n-1, n+2)] = []*version{{ID: w.newVer(), B: f}} |
| + w.markDurable(ltxKey(loc, gen, 1, n-1, n+2), f) |
| 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 11 and [10,15]", b.ID) |
| + 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.finish(t) |
| } |
| @@ -1126,15 +1129,72 @@ func wRestoreGap(t *testing.T, w *world) { |
| if len(w.violations) > 0 { |
| return |
| } |
| - w.appWrite(b) // TXID 12: its file never lands |
| - b.DB.Rep, b.DB.Pending = 12, nil |
| - w.appWrite(b) // TXID 13 |
| - f := w.compacted(b, 13, 13) |
| - w.buckets[b.Region].objs[ltxKey(loc, gen, 0, 13, 13)] = []*version{{ID: w.newVer(), B: f}} |
| - b.DB.Rep, b.DB.Pending = 13, nil |
| + n := b.DB.Pos |
| + w.appWrite(b) // its file, [n+1, n+1], never lands |
| + b.DB.Pos, b.DB.Pending = n+1, nil |
| + w.appWrite(b) |
| + f := w.compactedPending(b, n+2, n+2) // [n+2, n+2] is current |
| + w.buckets[b.Region].objs[ltxKey(loc, gen, 0, n+2, n+2)] = []*version{{ID: w.newVer(), B: f}} |
| w.idleStop(b) |
| db := w.restore(b, 1, loc, gen) |
| if db != nil { |
| - w.violate("INV-2 (cold restore): a restore across the gap after TXID 11 returned a database at TXID %d", db.TX) |
| + w.violate("INV-2 (cold restore): a restore across the gap after TXID %d returned a database at position %d", n, db.Pos) |
| + } |
| +} |
| + |
| +// ---------- rev 7.6 ---------- |
| + |
| +// wSnapshotOnlyInstall (R5-1): the push installs a snapshot; the host |
| +// wakes, writes, replicates, loses its disk and wakes cold. |
| +func wSnapshotOnlyInstall(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + for i := 0; i < 3; i++ { |
| + w.appWrite(b) |
| + } |
| + w.replicate(b) |
| + w.pushed(t, b) |
| + if len(w.violations) > 0 { |
| + return |
| + } |
| + for i := 0; i < 2; i++ { |
| + w.appWrite(b) |
| + w.replicate(b) |
| + } |
| + w.idleStop(b) |
| + b.DB = nil // the disk is lost |
| + 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.finish(t) |
| +} |
| + |
| +// wRootAdvance (R5-3): after the push, enough writes and replication for |
| +// compaction to rotate snapshots four times, retention expiring the files |
| +// that continued from the pinned snapshot; then the disk is lost. |
| +func wRootAdvance(t *testing.T, w *world) { |
| + b := w.boxes[0] |
| + w.serve(b, 1, app0) |
| + w.wake(b) |
| + w.appWrite(b) |
| + w.replicate(b) |
| + loc, gen := w.pushed(t, b) |
| + if len(w.violations) > 0 { |
| + return |
| + } |
| + for round := 0; round < 4; round++ { |
| + for i := 0; i < 3; i++ { |
| + w.appWrite(b) |
| + w.replicate(b) |
| + } |
| } |
| + snaps := w.currentLTX(w.buckets[b.Region], dataKey(loc, gen, ""), 9) |
| + if len(snaps) == 0 || snaps[len(snaps)-1].B.Max <= w.buckets[b.Region].current(genMarkerKey(loc)).B.TX { |
| + t.Fatalf("witness: no successor snapshot rotated in: %d snapshots", len(snaps)) |
| + } |
| + 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.finish(t) |
| } |
diff --git a/internal/sim/hostlife/world_test.go b/internal/sim/hostlife/world_test.go| index 372891001..023196568 100644 |
| --- a/internal/sim/hostlife/world_test.go |
| +++ b/internal/sim/hostlife/world_test.go |
| @@ -1,7 +1,7 @@ |
| // Package hostlife is a deterministic, randomized model of the host |
| // lifecycle claim (docs/superpowers/specs/2026-10-02-host-lifecycle-claim-design.md, |
| -// rev 6.2, 30d572c4) and Phase 1 of move-any-app |
| -// (docs/superpowers/specs/2026-10-02-move-any-app-design.md, rev 7.5). |
| +// rev 6.3, 8faa89a4) and Phase 1 of move-any-app |
| +// (docs/superpowers/specs/2026-10-02-move-any-app-design.md, rev 7.6). |
| // |
| // Since joint round 2 the model encodes what the specs say and nothing |
| // stronger: restore reads the holder's location and the generation the |
| @@ -72,15 +72,15 @@ type blob struct { |
| Gen string // lineage that wrote it |
| Seq int |
| SealKey int // the instance.key the data is sealed with |
| - // an LTX file (ltx_test.go, R4-1): its level (0 and 1, 9 for a snapshot), |
| - // transaction range, the final image of every key it touches (0 is a |
| - // delete), the database checksum after Max, and, for the oracle only, |
| - // the writes it covers |
| + // an LTX file (ltx_test.go, R4-1, R5-2): its level (0 and 1, 9 for a |
| + // snapshot), its range in replication numbering, the final image of |
| + // every key it touches (0 is a delete), and, for the oracle only, the |
| + // writes it covers. No database checksum: the pinned writers put none |
| + // in a file (litestreamfacts, R5-2) |
| Level int |
| Min int |
| Max int |
| Pages map[string]int |
| - Post string |
| TXs []int |
| // a "gen" marker (R3-1, R4-1): the generation's snapshot by key, version |
| // id and TXID |
| @@ -196,7 +196,7 @@ func dataKey(loc, gen, kind string) string { |
| func genMarkerKey(loc string) string { return loc + "generation.json" } |
| func isDataKey(k string) bool { |
| - return strings.HasPrefix(k, "instances/") && strings.Contains(k, "/ltx/") |
| + return strings.HasPrefix(k, "instances/") && ltxKeyRe.MatchString(k) |
| } |
| // ---------- requests (late completion) ---------- |
| @@ -232,11 +232,12 @@ type request struct { |
| // ---------- local box state ---------- |
| type filesDir struct { |
| - Inc int |
| - Key int // the app's instance.key identity |
| - Gen string // .files-gen marker |
| - Env int // admin token identity in <host>.env |
| - Done bool // markers written last |
| + Inc int |
| + Key int // the app's instance.key identity |
| + Gen string // .files-gen marker |
| + Env int // admin token identity in <host>.env |
| + Done bool // markers written last |
| + Stamp int // the installation stamp that produced it (claim rev 6.3 §6.3) |
| } |
| func (f *filesDir) clone() *filesDir { |
| @@ -253,10 +254,11 @@ type localDB struct { |
| Gen string // <dbPath>.gen |
| Key int // the key it was sealed with |
| State map[string]int |
| - TX int // the lineage's transaction id (litestream TXID) |
| - Rep int // replicated through this TXID |
| - Pending []int // writes since Rep (what the next LTX file ships) |
| + TX int // the lineage's transaction id: the oracle's order of acknowledged writes |
| + Pos int // the replication position: litestream's TXID, the count of files shipped, which every LTX range is numbered in (R5-1) |
| + Pending []int // writes since Pos (what the next LTX file ships) |
| Log []int // every write applied here, for the oracle |
| + Stamp int // the installation stamp that produced this artifact (claim rev 6.3 §6.3) |
| Seq int |
| } |
| @@ -315,7 +317,7 @@ type box struct { |
| History []int // delivered history (disposable) |
| Attempt int // runner bookkeeping |
| Procs []*proc |
| - // Stamps is the local record per host incarnation (claim rev 6.2 §6.4, |
| + // Stamps is the local record per host incarnation (claim rev 6.3 §6.4, |
| // R4-2): {active_stamp | none, retired_stamps}. A stamp is the claim |
| // gen captured at the CAS that made an installation current here. A |
| // writer is admitted only if its stamp equals the active one. |
| @@ -347,13 +349,13 @@ type plan struct { |
| Ver int // the config bundle version it selected |
| At int // it takes the guard no earlier than this |
| // Stamp is the installation stamp the pass read when it selected the |
| - // row (claim rev 6.2 §6.4, R2-3): the host guard admits the write only |
| + // row (claim rev 6.3 §6.4, R2-3): the host guard admits the write only |
| // if it is still the active stamp. |
| Stamp int |
| } |
| // stamps is a box's local record for one incarnation of the host (claim |
| -// rev 6.2 §6.4): the active installation's stamp (0 is none) and the |
| +// rev 6.3 §6.4): the active installation's stamp (0 is none) and the |
| // stamps of the installations retired here. |
| type stamps struct { |
| Active int |
| @@ -367,9 +369,11 @@ func (b *box) stampOf(inc int) int { |
| return 0 |
| } |
| -// install sets the active stamp for inc to the claim gen captured at the |
| -// CAS that made this installation current here (a create's serving CAS, a |
| -// landing's flip). Under the guard; the caller has read the claim fresh. |
| +// install sets the active stamp for inc to holder.stamp as the installing |
| +// CAS wrote it (claim rev 6.3 §6.4: a create's serving CAS, a landing's |
| +// flip, an adoption, a reinstall, a dirty rollback's install, a push |
| +// commit), or as initRecord read it fresh under the guard. Under the |
| +// guard. |
| func (b *box) install(inc, gen int) { |
| if b.Stamps == nil { |
| b.Stamps = map[int]*stamps{} |
| @@ -380,7 +384,28 @@ func (b *box) install(inc, gen int) { |
| b.Stamps[inc].Active = gen |
| } |
| -// retireStamp moves the active stamp into the retired set and sets none. |
| +// retireArtifacts is claim rev 6.3's retire-artifacts(S): the artifacts |
| +// tagged S are parked (the caller does that) and S joins retired_stamps; |
| +// active_stamp is untouched. A landing uses it on its predecessor's |
| +// leftovers. |
| +func (b *box) retireArtifacts(inc, stamp int) { |
| + if stamp == 0 { |
| + return |
| + } |
| + if b.Stamps == nil { |
| + b.Stamps = map[int]*stamps{} |
| + } |
| + if b.Stamps[inc] == nil { |
| + b.Stamps[inc] = &stamps{} |
| + } |
| + if !contains(b.Stamps[inc].Retired, stamp) && stamp != b.Stamps[inc].Active { |
| + b.Stamps[inc].Retired = append(b.Stamps[inc].Retired, stamp) |
| + } |
| +} |
| + |
| +// retireStamp is claim rev 6.3's retire-installation: the active stamp |
| +// moves into retired_stamps and becomes none; the caller parks every |
| +// artifact. Used at a delete, a move away, and the quarantine. |
| func (b *box) retireStamp(inc int) { |
| if b.Stamps == nil { |
| b.Stamps = map[int]*stamps{} |
| @@ -394,16 +419,17 @@ func (b *box) retireStamp(inc int) { |
| b.Stamps[inc].Active = 0 |
| } |
| -// admitsStamp is the local record's admission (claim rev 6.2 §6.4): the |
| -// writer's stamp must equal the active stamp; none active refuses all. |
| -// The mutation of fence-stamp admits any writer for the incarnation while |
| -// an installation is active (rev 7.4's behaviour); the mutation of |
| -// fence-by-installation is rev 5.2's marker: once retired, never active |
| -// again. |
| +// admitsStamp is the local record's admission (claim rev 6.3 §6.4): the |
| +// writer's stamp must equal the active stamp; none active refuses all, |
| +// and so does no record at all (R5-4: a writer is never admitted before |
| +// initialization). The mutation of fence-stamp admits any writer for the |
| +// incarnation while an installation is active (rev 7.4's behaviour); the |
| +// mutation of fence-by-installation is rev 5.2's marker: once retired, |
| +// never active again. |
| func (b *box) admitsStamp(inc, stamp int) bool { |
| s := b.Stamps[inc] |
| if s == nil { |
| - return true // never installed or retired here: nothing to compare with |
| + return false |
| } |
| if !on("fence-by-installation") && len(s.Retired) > 0 { |
| return false |
| @@ -441,6 +467,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 |
| + // Stamp is holder.stamp (claim rev 6.3 §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 |
| + // dirty rollback's install, a push's commit. The claim is where a box |
| + // recovers its local record from. |
| + Stamp int |
| } |
| type obligation struct { |
| @@ -498,7 +530,7 @@ type opState struct { |
| 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 claim gen of the flip CAS: the landing's installation stamp (claim rev 6.2 §6.4) |
| + FlipGen int // the holder.stamp the flip CAS wrote: the landing's installation stamp (claim rev 6.3 §6.4) |
| Committed bool |
| Opened bool |
| ParkLocalNow bool |
| @@ -549,7 +581,7 @@ type claim struct { |
| Cancelled []cancelledCreate |
| // readStamp is model bookkeeping on a box's read of the claim: the |
| // installation stamp active on that box at the read, which a worker |
| - // carries with the claim it validated under (claim rev 6.2 §6.4) |
| + // carries with the claim it validated under (claim rev 6.3 §6.4) |
| readStamp int |
| } |
| @@ -705,7 +737,7 @@ func (w *world) newVer() int { |
| } |
| // nextGen is the gen the claim CAS in progress will carry: what a writer |
| -// captures as its installation stamp (claim rev 6.2 §6.4). |
| +// captures as its installation stamp (claim rev 6.3 §6.4). |
| func (w *world) nextGen() int { |
| if w.claim == nil { |
| return 1 |