amadan

carlos / platform Public

sim: hostlife: LTX restore, installation stamps, region-qualified live set, purge retirement (rev 7.5)

Joint round 4 and claim rev 6.2 (30d572c4):

- R4-1: the database is a map of keys to values; transactions insert,
  update or delete; replica objects are LTX files with litestream's
  semantics (ltx_test.go): a transaction range, the final page image of
  every key touched, the checksum after Max, levels 0 and 1, a level-9
  snapshot; compaction merges level 0 into level 1 and deletes the
  inputs. Restore is litestream v0.5.14's planner over the generation's
  current files from the snapshot generation.json names (key, version,
  TXID): overlap allowed, highest max first, a gap refused, the checksum
  verified. The sealed transfer is a verified final snapshot. The oracle
  replays acknowledged transactions into the state a restore must equal.
  Rules restore-overlap and restore-gap-closed, with witnesses (a
  compacted [10,15] over a snapshot at 11 restores through 15; a genuine
  gap refuses).
- R4-2: installation stamps per claim rev 6.2 §6.4: each box keeps
  {active_stamp | none, retired_stamps} per incarnation; a writer carries
  the stamp it selected under and is admitted only if it equals the
  active one; retirement moves active to retired; a landing sets its own
  stamp (the flip CAS's gen) under the guard on a fresh read. The op
  counter ordering and the empty-op-means-now rule are gone. Rule
  fence-stamp, witness: a stale config pass across UK->US->UK. The oracle
  records a landing probe refused by its own installation record.
- R4-3: the live set is {authoritative region, location, generation};
  matching keys elsewhere are residue. Rule live-set-region, witness: an
  old US replication PUT for a legacy host serving in UK.
- R4-4: deleteSteps records the record version in op.intent, deletes
  conditionally, and moves purging -> parking only with the journal at
  torn-down; a purged host never returns to serving.
- Random generator: marker-hold and targeted freeze schedules, a
  retention-boundary fixture, the probe-fence oracle.
- The absent-claim branch of orphan cleanup is explicit (legacy, never
  an orphan; claim rev 6.2 §5.1).
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
claude 5de710b2fc9ad76cda8cba96d2a459d683987b58 parent 0e41b68
10 files changed, +1022 −287
  • internal/sim/hostlife/background_test.go +35 −15
  • internal/sim/hostlife/console_test.go +47 −8
  • internal/sim/hostlife/expected-outcomes.txt +1 −0
  • internal/sim/hostlife/ltx_test.go +337 −0
  • internal/sim/hostlife/moves_test.go +94 −60
  • internal/sim/hostlife/protocol_test.go +77 −87
  • internal/sim/hostlife/rules_test.go +5 −1
  • internal/sim/hostlife/sim_test.go +89 −38
  • internal/sim/hostlife/witness_test.go +200 −23
  • internal/sim/hostlife/world_test.go +137 −55
diff --git a/internal/sim/hostlife/background_test.go b/internal/sim/hostlife/background_test.go
index 120d2ced5..9945c580d 100644
--- a/internal/sim/hostlife/background_test.go
+++ b/internal/sim/hostlife/background_test.go
@@ -15,7 +15,9 @@ func (w *world) traffic(b *box) {
w.nextW++
wr := w.nextW
if b.DB != nil && b.DB.Inc == p.Inc {
- b.DB.Writes = append(b.DB.Writes, wr)
+ w.commit(b.DB, wr)
+ } else {
+ w.txs[wr] = tx{W: wr} // into a purged or foreign database: nowhere
}
// acknowledged to the client or the outside world either way: a
// write into a purged or foreign database is a lost write
@@ -29,17 +31,6 @@ func (w *world) traffic(b *box) {
}
}
-// replicate: the serving process's ordinary litestream replication. The
-// claim's C3 exception: these are plain PUTs to reused keys.
-func (w *world) replicate(b *box) {
- if !w.agentUp(b) || !hasMain(b) || b.DB == nil {
- return
- }
- bl := &blob{Kind: "ltx", Inc: b.DB.Inc, Gen: b.DB.Gen, Writes: append([]int(nil), b.DB.Writes...), SealKey: b.DB.Key}
- // into the location and generation the database was restored from
- w.issue(b.ID, &request{kind: rPut, region: b.Region, key: dataKey(b.DB.Loc, b.DB.Gen, "ltx"), b: bl})
-}
-
// idleStop: the activator's ordinary idle stop (main only; a descendant
// keeps running, as today).
func (w *world) idleStop(b *box) {
@@ -206,7 +197,9 @@ 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"}
- delete(b.Fence, h.Inc) // the serving executor, on a fresh read: the fence is replaced
+ 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)
+ }
}
} else if b.Row != nil && !(on("participant-rows") && w.ownsLocalObligation(c, b)) {
w.retireRow(b)
@@ -233,7 +226,7 @@ func (w *world) passPlan(b *box, kind string) {
if kind == "config" && !b.HasKey {
return
}
- b.plans = append(b.plans, &plan{Kind: kind, Inc: b.Row.Inc, App: b.Row.App, Ver: w.cfgVer[b.Row.App], At: w.now + w.pauseLen(), Seq: w.nextID})
+ b.plans = append(b.plans, &plan{Kind: kind, Inc: b.Row.Inc, App: b.Row.App, Ver: w.cfgVer[b.Row.App], At: w.now + w.pauseLen(), Stamp: b.stampOf(b.Row.Inc)})
}
// passExec runs one independent process's write. It needs the box running,
@@ -257,13 +250,22 @@ func (w *world) passExec(b *box) {
return
}
c := w.claim
- if c == nil || c.Holder == nil || c.Holder.Inc != p.Inc || b.Row == nil || b.Row.Inc != p.Inc || w.fenced(b, p.Inc, p.Seq) {
+ if c == nil || c.Holder == nil || c.Holder.Inc != p.Inc || b.Row == nil || b.Row.Inc != p.Inc {
return
}
if p.Kind == "config" && p.Ver != w.cfgVer[p.App] {
return // the selected bundle is no longer the held one
}
}
+ // the local record (claim rev 6.2 §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
+ }
+ if p.Stamp != b.stampOf(p.Inc) {
+ // the oracle: a writer of a retired installation wrote host state
+ w.violate("INV-6: a %s pass of a retired installation (stamp %d, active %d) wrote host state for inc %d on box %d", p.Kind, p.Stamp, b.stampOf(p.Inc), p.Inc, b.ID)
+ }
switch p.Kind {
case "config":
b.Config, b.CfgVer = p.Inc, p.Ver
@@ -354,6 +356,24 @@ func (w *world) fault() {
b.Partitioned = !b.Partitioned
}
case 3:
+ if w.fail["freezetgt"] {
+ // the biased schedule (rev 7.5): freeze the landing target in
+ // its probe or settle window, and keep it frozen until the
+ // source runs again, if a declare-lost let it
+ c := w.claim
+ if c != nil && c.Op.Kind == "landing" && b.ID == c.Op.Tgt && !b.Frozen && (c.Op.LandPhase == "probing" || c.Op.LandPhase == "probed") {
+ b.Frozen = true
+ w.logf("box %d frozen in its settle window (biased schedule)", b.ID)
+ break
+ }
+ if b.Frozen && c != nil && (c.Op.Kind == "landing" || c.Op.Kind == "restoring") && b.ID == c.Op.Tgt {
+ if s := w.boxByID(c.Op.Src); s != nil && hasMain(s) {
+ b.Frozen = false
+ w.logf("box %d resumed beside the recovered source (biased schedule)", b.ID)
+ }
+ break
+ }
+ }
if w.fail["freeze"] {
b.Frozen = !b.Frozen
}
diff --git a/internal/sim/hostlife/console_test.go b/internal/sim/hostlife/console_test.go
index b2c701142..31a1f339c 100644
--- a/internal/sim/hostlife/console_test.go
+++ b/internal/sim/hostlife/console_test.go
@@ -341,13 +341,18 @@ func (w *world) userDelete() {
if e := w.execBox(c); e != nil && e.DB != nil && hasMain(e) && w.rng.Intn(2) == 0 {
// the replicator's last PUT before the delete's stop evidence:
// accepted, then completing after the host is recreated
- bl := &blob{Kind: "ltx", Inc: e.DB.Inc, Gen: e.DB.Gen, Writes: append([]int(nil), e.DB.Writes...), SealKey: e.DB.Key}
- w.pending = append(w.pending, &request{kind: rPut, region: e.Region, key: dataKey(e.DB.Loc, e.DB.Gen, "ltx"), b: bl,
+ bl := snapshotOf(e.DB, e.DB.Inc)
+ bl.Kind, bl.Level = "ltx", 0 // a level-0 file covering everything so far
+ w.pending = append(w.pending, &request{kind: rPut, region: e.Region, key: ltxKey(e.DB.Loc, e.DB.Gen, 0, 1, bl.Max), b: bl,
doneAt: w.now + 40*horizonH, beyond: true, by: fmt.Sprintf("box:%d", e.ID)})
}
}
+ recVer := 0
+ if w.rec != nil {
+ recVer = w.rec.Ver
+ }
if w.cas(c, func(n *claim) bool {
- n.Op = opState{Kind: "deleting", ID: id, Exec: -1, ReqAt: w.now}
+ n.Op = opState{Kind: "deleting", ID: id, Exec: -1, ReqAt: w.now, RecVer: recVer} // intent: the record's ETag (claim §5.2 step 1)
return true
}) {
w.console = append(w.console, &consolePlan{Kind: "delete", OpID: id})
@@ -562,13 +567,36 @@ func (w *world) orphanSteps(p *consolePlan) bool {
}
}
-// deleteSteps: record DeleteIfMatch, then parking with obligations, then
-// free once every obligation is acknowledged (claim §5.2).
+// 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,
+// 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
+// and the purge journal at torn-down or later. A purged host never returns
+// to serving.
func (w *world) deleteSteps(c *claim) bool {
switch c.Op.Kind {
case "deleting", "purging":
+ if c.Op.Kind == "purging" && on("purge-journal") && w.purgeStep[c.Holder.App] < 4 {
+ return false // the journal is not at torn-down yet
+ }
if w.rec != nil && w.rec.Inc == c.Holder.Inc {
- w.rec = nil
+ if c.Op.RecVer == 0 {
+ // the intent: the exact record version to delete
+ ver := w.rec.Ver
+ w.cas(c, func(n *claim) bool {
+ if !sameOp(n.Op.ID, c.Op.ID) || n.Op.Kind != c.Op.Kind {
+ return false
+ }
+ n.Op.RecVer = ver
+ return true
+ })
+ return false
+ }
+ if w.rec.Ver == c.Op.RecVer {
+ w.rec = nil // DeleteIfMatch
+ }
return false
}
if w.rec != nil {
@@ -576,7 +604,7 @@ func (w *world) deleteSteps(c *claim) bool {
}
inc := c.Holder.Inc
w.cas(c, func(n *claim) bool {
- if !sameOp(n.Op.ID, c.Op.ID) {
+ if !sameOp(n.Op.ID, c.Op.ID) || n.Op.Kind != c.Op.Kind {
return false
}
n.Op.Kind = "parking"
@@ -865,8 +893,19 @@ func (w *world) doorAuthorize() {
if b == nil {
return
}
+ // 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/")) {
+ if v := bk.current(k); v != nil && (key == "" || v.ID > bk.current(key).ID) {
+ key = k
+ }
+ }
+ if key == "" {
+ return
+ }
d := &doorPlan{Region: b.Region, Loc: c.Holder.Location, Gen: c.Holder.Gen, Inc: c.Holder.Inc, OpID: c.Op.ID,
- Key: dataKey(c.Holder.Location, c.Holder.Gen, "ltx"), Until: w.now + 1 + w.pauseLen()}
+ Key: key, Until: w.now + 1 + w.pauseLen()}
if w.fail["recreate"] && w.rng.Intn(2) == 0 {
d.Adv = true
if w.rng.Intn(2) == 0 {
diff --git a/internal/sim/hostlife/expected-outcomes.txt b/internal/sim/hostlife/expected-outcomes.txt
index 6cc69ce95..c81903152 100644
--- a/internal/sim/hostlife/expected-outcomes.txt
+++ b/internal/sim/hostlife/expected-outcomes.txt
@@ -17,3 +17,4 @@ 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)
registration-check missed a cheap extra: subsumed by purge-claim-fence and fence-etag together
+fence-stamp missed caught with hostguard (witness): alone, the host guard's fresh re-validation refuses the stale writer's bundle; the stamp is what refuses it when that check is gone
diff --git a/internal/sim/hostlife/ltx_test.go b/internal/sim/hostlife/ltx_test.go
new file mode 100644
index 000000000..8d4a1ac7d
--- /dev/null
+++ b/internal/sim/hostlife/ltx_test.go
@@ -0,0 +1,337 @@
+package hostlife
+
+import (
+ "fmt"
+ "hash/fnv"
+ "sort"
+ "strings"
+)
+
+// The database and its replica, with litestream's LTX semantics (rev 7.5,
+// R4-1).
+//
+// 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.
+
+type tx struct {
+ W int // the global write id, which is what the application acknowledged
+ TX int // the lineage's transaction id; 0 if the write never reached a database
+ Key string // the key written
+ 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.
+func sum(state map[string]int) string {
+ keys := make([]string, 0, len(state))
+ for k, v := range state {
+ if v != 0 {
+ keys = append(keys, k)
+ }
+ }
+ sort.Strings(keys)
+ h := fnv.New64a()
+ for _, k := range keys {
+ fmt.Fprintf(h, "%s=%d;", k, state[k])
+ }
+ return fmt.Sprintf("%016x", h.Sum64())
+}
+
+func cloneState(m map[string]int) map[string]int {
+ out := make(map[string]int, len(m))
+ for k, v := range m {
+ if v != 0 {
+ out[k] = v
+ }
+ }
+ return out
+}
+
+func statesEqual(a, b map[string]int) (string, bool) {
+ for k, v := range a {
+ if v != 0 && b[k] != v {
+ return fmt.Sprintf("key %s expected %d, got %d", k, v, b[k]), false
+ }
+ }
+ for k, v := range b {
+ if v != 0 && a[k] != v {
+ return fmt.Sprintf("key %s expected %d, got %d", k, a[k], v), false
+ }
+ }
+ return "", true
+}
+
+// commit applies acknowledged write w to db as a transaction: mostly an
+// insert of a new key named by the write; every fifth write updates the
+// newest live key, every fifth-but-one deletes the oldest, so restore
+// fidelity depends on final page images, not on a list of ids. It records
+// the transaction for the oracle.
+func (w *world) commit(db *localDB, wr int) {
+ if db.State == nil {
+ db.State = map[string]int{}
+ }
+ t := tx{W: wr, TX: db.TX + 1, Key: fmt.Sprintf("k%d", wr), Val: wr}
+ switch wr % 5 {
+ case 3:
+ if k := newestKey(db.State); k != "" {
+ t.Key, t.Val = k, wr
+ }
+ case 4:
+ if k := oldestKey(db.State); k != "" {
+ t.Key, t.Val = k, 0
+ }
+ }
+ if t.Val == 0 {
+ delete(db.State, t.Key)
+ } else {
+ db.State[t.Key] = t.Val
+ }
+ db.TX++
+ db.Pending = append(db.Pending, wr)
+ db.Log = append(db.Log, wr)
+ w.txs[wr] = t
+}
+
+func keyID(k string) int {
+ n := 0
+ fmt.Sscanf(k, "k%d", &n)
+ return n
+}
+
+func newestKey(state map[string]int) string {
+ best, id := "", 0
+ for k, v := range state {
+ if v != 0 && keyID(k) > id {
+ best, id = k, keyID(k)
+ }
+ }
+ return best
+}
+
+func oldestKey(state map[string]int) string {
+ best, id := "", 0
+ for k, v := range state {
+ if v != 0 && (best == "" || keyID(k) < id) {
+ best, id = k, keyID(k)
+ }
+ }
+ return best
+}
+
+// ltxKey and snapKey are litestream's layout under a generation:
+// ltx/<level>/<min>-<max>; 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)
+}
+
+func snapKey(loc, gen string, max int) string { return ltxKey(loc, gen, 9, 1, max) }
+
+// 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 {
+ 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
+}
+
+// genPrefixOf is the generation prefix an LTX key belongs to.
+func genPrefixOf(k string) string {
+ i := strings.Index(k, "/ltx/")
+ if i < 0 {
+ return ""
+ }
+ return k[:i+1]
+}
+
+// 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.
+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...)}
+}
+
+// deltaOf is the level-0 file of the transactions db has not replicated
+// yet: the final image of every key they touched.
+func (w *world) deltaOf(db *localDB, inc int) *blob {
+ pages := map[string]int{}
+ 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...)}
+}
+
+// apply overlays an LTX file's page images onto state.
+func apply(state map[string]int, f *blob) {
+ for k, v := range f.Pages {
+ if v == 0 {
+ delete(state, k)
+ } else {
+ state[k] = v
+ }
+ }
+}
+
+// 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.
+func (w *world) replicate(b *box) {
+ if !w.agentUp(b) || !hasMain(b) || b.DB == nil {
+ return
+ }
+ db := b.DB
+ inc := db.Inc
+ 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.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
+ for _, k := range bk.keys(prefix) {
+ if v := bk.current(k); v != nil {
+ inputs = append(inputs, in{k, v})
+ }
+ }
+ 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{}}
+ max := 0
+ var keys []string
+ for _, x := range inputs {
+ v := x.v
+ keys = append(keys, x.k)
+ if max != 0 && !(v.B.Min <= max+1 && v.B.Max > max) {
+ return // non-contiguous inputs: litestream refuses to compact them
+ }
+ apply(out.Pages, v.B)
+ for k, val := range v.B.Pages {
+ if val == 0 {
+ out.Pages[k] = 0 // a delete stays a delete in the compacted image
+ }
+ }
+ out.TXs = append(out.TXs, v.B.TXs...)
+ out.Post = v.B.Post
+ if v.B.Max > max {
+ max = 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
+ }
+ w.stats["ltx compactions"]++
+ }
+}
+
+// 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:
+// min <= cur+1 && max > cur), preferring the highest max, then the lowest
+// min, then the higher level; stop when nothing extends it. The mutation
+// of restore-overlap is rev 7.4's rule, min == cur+1, which discards a
+// compacted continuation. With restore-gap-closed, a current file that
+// starts beyond cur+1 once nothing extends the range is a gap and the
+// restore refuses (litestream: "non-contiguous ltx files"); the mutation
+// returns what it has.
+func (w *world) plan(bk *bucket, prefix string, cur int) (files []*blob, gap bool) {
+ 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 v := bk.current(k); v != nil {
+ cands = append(cands, v.B)
+ }
+ }
+ for {
+ var best *blob
+ for _, f := range cands {
+ ext := f.Min <= cur+1 && f.Max > cur
+ if !on("restore-overlap") {
+ ext = f.Min == cur+1
+ }
+ if !ext {
+ continue
+ }
+ if best == nil || f.Max > best.Max || f.Max == best.Max && (f.Min < best.Min || f.Min == best.Min && f.Level > best.Level) {
+ best = f
+ }
+ }
+ if best == nil {
+ break
+ }
+ files = append(files, best)
+ cur = best.Max
+ }
+ for _, f := range cands {
+ if f.Min > cur+1 {
+ gap = true
+ }
+ }
+ return files, gap
+}
+
+// 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
+// to upto, what a cold restore that reached upto must hold; with upto <
+// 0 it replays every acknowledged write, a write that never reached a
+// database counting as the insert the application believed it made, which
+// is what the serving box's own database must hold.
+func (w *world) expectState(inc int, upto int) map[string]int {
+ state := map[string]int{}
+ for _, a := range w.acks {
+ if a.Inc != inc || a.Exempt || a.Kind != "write" {
+ continue
+ }
+ t := w.txs[a.W]
+ if upto >= 0 && (t.TX == 0 || t.TX > upto) {
+ continue
+ }
+ 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 state
+}
diff --git a/internal/sim/hostlife/moves_test.go b/internal/sim/hostlife/moves_test.go
index 0df9ad223..023ee3aab 100644
--- a/internal/sim/hostlife/moves_test.go
+++ b/internal/sim/hostlife/moves_test.go
@@ -86,7 +86,7 @@ func (w *world) role(b *box, c *claim) {
if b.DB != nil {
inc = b.DB.Inc
}
- if w.localPark(b, inc, w.nextID) { // a quarantine retires every installation so far
+ if w.localPark(b, inc) { // the quarantine retires the installation
b.Row = nil
}
}
@@ -171,25 +171,17 @@ func (w *world) snapshotBlob(b *box, c *claim) (*blob, *blob) {
}
func (w *world) snapshotBlobOf(b *box, c *claim, gen string) (*blob, *blob) {
- db := &blob{Kind: "snap", Inc: c.Holder.Inc}
- if r := w.restore(b, c.Holder.Inc, c.Holder.Location, gen); r != nil {
- db.Writes, db.SealKey = r.Writes, r.Key
- }
+ // the sealed transfer is a verified final snapshot (rev 7.5 §5.1, R4-1):
+ // the local database after its final checkpoint, every page, TXIDs 1
+ // to its current one; with no local database (a lost disk, or never
+ // woken) the replica restored as it stands
+ var db *blob
if b.DB != nil {
- // the final checkpoint: the local database continues the replica's
- // history (R3-1); a local history that does not is never staged
- switch {
- case extends(b.DB.Writes, db.Writes):
- db.Writes = append([]int(nil), b.DB.Writes...)
- case extends(db.Writes, b.DB.Writes):
- default:
- w.logf("box %d snapshot: the local database does not continue the replica", b.ID)
- w.stats["snapshot refused: conflicting history"]++
- }
- if b.DB.Key != 0 {
- db.SealKey = b.DB.Key
- }
- db.Inc = b.DB.Inc
+ 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)}
}
files := &blob{Kind: "files", Inc: c.Holder.Inc, Files: b.Files.clone(), Timers: append([]int(nil), b.Timers...)}
return db, files
@@ -664,6 +656,7 @@ 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)
return true
})
case r.MoveState == "committed" && op.Kind == "landing" && !op.Committed:
@@ -718,14 +711,18 @@ func (w *world) landTarget(b *box, c *claim) {
}
switch op.LandPhase {
case "checking":
- // a new installation of this incarnation here replaces the local
- // retired marker, under the guard, on a fresh read showing this
- // operation current and this box its target (claim rev 6 §6.4,
- // R3-3): a move back to a region the instance left
- if _, fenced := b.Fence[c.Holder.Inc]; fenced && on("fence-by-installation") {
- if fc, ok := w.readClaim(b); ok && fc.Op.ID == op.ID && fc.Op.Kind == "landing" && fc.Op.Tgt == b.ID {
- b.Fence[c.Holder.Inc] = w.opSeq(op.ID) - 1
- w.logf("box %d: the fence for inc %d now admits %s", b.ID, c.Holder.Inc, op.ID)
+ // 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
+ // §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)
}
}
w.cas(c, func(n *claim) bool {
@@ -751,7 +748,7 @@ func (w *world) landTarget(b *box, c *claim) {
src := op.SealDB
if cur := w.buckets[b.Region].current(src.Key); cur != nil && cur.ID != src.Ver {
w.stats["diag: staging newer than seal at install"]++
- if sv := w.buckets[b.Region].byID(src.Key, src.Ver); sv != nil && len(cur.B.Writes) < len(sv.B.Writes) {
+ if sv := w.buckets[b.Region].byID(src.Key, src.Ver); sv != nil && cur.B.Max < sv.B.Max {
w.stats["diag: ...and older content"]++
}
}
@@ -763,15 +760,25 @@ func (w *world) landTarget(b *box, c *claim) {
}
src.Ver = cur.ID
}
- // into this move's own generation: a copy that completes late can
- // only ever add an object to this generation
- v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: dataKey(loc, op.ID, "base"), srcRegion: b.Region, srcKey: src.Key, srcVer: src.Ver})
+ // into this move's own generation, as its snapshot at the sealed
+ // TXID: a copy that completes late can only ever add an object to
+ // this generation
+ sv := w.buckets[b.Region].byID(src.Key, src.Ver)
+ if sv == nil {
+ w.rollback(c, "staging missing")
+ return
+ }
+ 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 version (R3-1)
- w.cas(c, func(n *claim) bool { n.Op.LandPhase, n.Op.BaseVer = "published", v; return sameOp(n.Op.ID, op.ID) })
+ // the install manifest records the snapshot's key, version and TXID (R3-1, R4-1)
+ 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
+ return sameOp(n.Op.ID, op.ID)
+ })
}
case "published":
- v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer)
+ v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX)
if v > 0 {
w.cas(c, func(n *claim) bool { n.Op.LandPhase = "local"; return sameOp(n.Op.ID, op.ID) })
}
@@ -1102,12 +1109,20 @@ func (w *world) recoverSource(b *box, c *claim) {
w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "install-copy"; return sameOp(n.Op.ID, op.ID) })
}
case "install-copy":
- v := w.issue(b.ID, &request{kind: rCopy, region: b.Region, key: dataKey(c.Holder.Location, op.RollbackID, "base"), srcRegion: b.Region, srcKey: op.ReturnDB.Key, srcVer: op.ReturnDB.Ver})
+ sv := w.buckets[b.Region].byID(op.ReturnDB.Key, op.ReturnDB.Ver)
+ if sv == nil {
+ return // the lane is gone: wait (stuck, in time)
+ }
+ 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 {
- w.cas(c, func(n *claim) bool { n.Op.SrcPhase, n.Op.BaseVer = "published", v; return sameOp(n.Op.ID, op.ID) })
+ 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
+ return sameOp(n.Op.ID, op.ID)
+ })
}
case "published":
- v := w.publishGen(b, c.Holder.Location, op.RollbackID, c.Holder.Inc, op.BaseVer)
+ v := w.publishGen(b, c.Holder.Location, op.RollbackID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX)
if v > 0 {
w.cas(c, func(n *claim) bool { n.Op.SrcPhase = "local"; return sameOp(n.Op.ID, op.ID) })
}
@@ -1164,23 +1179,24 @@ func (w *world) pushStep(b *box, c *claim) {
if !w.evidence(b, op.ID) {
return
}
- var writes []int
- if r := w.restore(b, c.Holder.Inc, loc, c.Holder.Gen); r != nil {
- writes = r.Writes // the replica, under the final checkpoint
- }
- if b.DB != nil {
- writes = unionInts(writes, b.DB.Writes)
- }
+ // the push's snapshot: the replica under the final checkpoint, plus
+ // the push's own mark, one more acknowledged transaction
+ snap, _ := w.snapshotBlobOf(b, c, c.Holder.Gen)
+ snap.Gen = op.ID
w.nextW++
mark := w.nextW
w.acks = append(w.acks, ack{W: mark, Inc: c.Holder.Inc, Box: b.ID, At: w.now, Kind: "write"})
w.writeInc[mark] = c.Holder.Inc
- writes = append(writes, mark)
+ 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 {
if !sameOp(n.Op.ID, op.ID) || n.Op.Phase != "stopping" {
return false
}
- n.Op.Phase, n.Op.StopAt, n.Op.PushWrites = "park", w.now, writes
+ n.Op.Phase, n.Op.StopAt, n.Op.PushSnap = "park", w.now, snap
return true
})
case "park":
@@ -1191,16 +1207,20 @@ func (w *world) pushStep(b *box, c *claim) {
if on("quiet-before-ack") && w.now < c.Quiet[b.Region] {
return // installs wait out quiet_after (claim §6.2a)
}
- key := 0
+ snap := op.PushSnap
if b.Files != nil {
- key = b.Files.Key
+ snap.SealKey = b.Files.Key
}
- v := w.issue(b.ID, &request{kind: rPut, region: b.Region, key: dataKey(loc, op.ID, "base"), b: &blob{Kind: "base", Inc: c.Holder.Inc, Gen: op.ID, Writes: op.PushWrites, SealKey: key}})
+ 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 {
- w.cas(c, func(n *claim) bool { n.Op.Phase, n.Op.BaseVer = "published", v; return sameOp(n.Op.ID, op.ID) })
+ w.cas(c, func(n *claim) bool {
+ n.Op.Phase, n.Op.BaseVer, n.Op.BaseKey, n.Op.BaseTX = "published", v, key, snap.Max
+ return sameOp(n.Op.ID, op.ID)
+ })
}
case "published":
- v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer)
+ v := w.publishGen(b, loc, op.ID, c.Holder.Inc, op.BaseVer, op.BaseKey, op.BaseTX)
if v > 0 {
w.cas(c, func(n *claim) bool { n.Op.Phase = "restore"; return sameOp(n.Op.ID, op.ID) })
}
@@ -1355,17 +1375,26 @@ func locationOf(k string) string {
return p
}
-// deadGeneration: within holder.location, a generation other than
-// holder.generation, or a marker that no longer names it.
+// 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
+// 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
+// in every region.
func (w *world) deadGeneration(c *claim, r int, k string) bool {
h := c.Holder
- if k == dataKey(h.Location, h.Gen, "base") || k == dataKey(h.Location, h.Gen, "ltx") {
- return false
+ e := w.execBox(c)
+ authoritative := e != nil && e.Region == r
+ if !on("live-set-region") {
+ authoritative = true
}
if k == genMarkerKey(h.Location) {
v := w.buckets[r].current(k)
- e := w.execBox(c)
- return v != nil && (v.B.Gen != h.Gen || e == nil || e.Region != r)
+ return v != nil && (v.B.Gen != h.Gen || !authoritative)
+ }
+ if strings.HasPrefix(k, dataKey(h.Location, h.Gen, "ltx/")) {
+ return !authoritative
}
return isDataKey(k)
}
@@ -1386,9 +1415,14 @@ func (w *world) legacyRetired(c *claim) bool {
// (moves rev 7.1 §5.1, model 29). generation.json is a reused key, so an
// earlier publication that completes late would otherwise put an older
// generation back, and every wake would refuse.
-func (w *world) publishGen(b *box, loc, gen string, inc int, base int) int {
- // the marker names the generation's snapshot by version id (R3-1)
- r := &request{kind: rPut, region: b.Region, key: genMarkerKey(loc), b: &blob{Kind: "gen", Gen: gen, Inc: inc, Base: base}}
+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)
+ 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
+ // completes at its worst moment, once a newer marker is current
+ r.hold = true
+ }
if on("generation-conditional") {
r.ifSet = true
if cur := w.buckets[b.Region].current(r.key); cur != nil {
diff --git a/internal/sim/hostlife/protocol_test.go b/internal/sim/hostlife/protocol_test.go
index 5b41f7687..6ee60967c 100644
--- a/internal/sim/hostlife/protocol_test.go
+++ b/internal/sim/hostlife/protocol_test.go
@@ -22,6 +22,9 @@ func (w *world) readClaim(b *box) (*claim, bool) {
return nil, false
}
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)
+ }
b.snap = c.clone()
return c, true
}
@@ -110,7 +113,7 @@ func (w *world) timerDigest(b *box) string {
func (w *world) digestOf(b *box) digest {
d := digest{Valid: true}
if b.DB != nil {
- d.Writes = setKey(b.DB.Writes)
+ d.Writes = sum(b.DB.State)
}
if b.Files != nil {
d.Files = fmt.Sprintf("%d/%d/%d", b.Files.Inc, b.Files.Key, b.Files.Env)
@@ -162,15 +165,19 @@ 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 6 §5.1). It replays every replica object of
-// that generation, as litestream replays every LTX file of a generation:
-// base snapshots and later LTX alike, current and noncurrent. It never
-// looks at another location or generation. The mutation restores the
-// newest single object anywhere under the location instead, which is what
-// a prefix restore without generation identity does.
+// 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.
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}
+ db := &localDB{Inc: inc, Loc: loc, Gen: gen, State: map[string]int{}}
if !on("restore-by-generation") {
var best *version
for _, k := range bk.keys(loc) {
@@ -184,93 +191,74 @@ func (w *world) restore(b *box, inc int, loc, gen string) *localDB {
}
}
if best != nil {
- db.Writes = append([]int(nil), best.B.Writes...)
- db.Key = best.B.SealKey
+ apply(db.State, best.B)
+ db.TX, db.Key = best.B.Max, best.B.SealKey
if best.B.Inc != 0 {
db.Inc = best.B.Inc
}
}
return db
}
- // the chain (rev 7.4 §5.1, R3-1): the generation's snapshot, named by
- // version id in generation.json, then the LTX objects of the
- // generation whose history extends what the chain holds, longest last.
- // A version whose history is a prefix of the chain is an older
- // checkpoint or a duplicate and is skipped; one that is neither is a
- // conflict, and the restore refuses rather than guess. Rev 7.3's model
- // took the union of every version's writes, which no restore can do.
- var history []int
- found := false
+ cur := 0
+ var last *blob
if gen != "" {
m := bk.current(genMarkerKey(loc))
if m == nil || m.B.Gen != gen {
return nil // the marker does not name this generation: refuse
}
- base := bk.byID(dataKey(loc, gen, "base"), m.B.Base)
- if base == nil || base.Marker {
- w.stats["restore refused: snapshot missing"]++
+ 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
}
- history = append([]int(nil), base.B.Writes...)
- db.Key = base.B.SealKey
- if base.B.Inc != 0 {
- db.Inc = base.B.Inc
+ 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
}
- found = true
}
- for _, v := range bk.objs[dataKey(loc, gen, "ltx")] {
- if v.Marker {
- continue
+ files, gap := w.plan(bk, dataKey(loc, gen, "ltx/"), cur)
+ for _, f := range files {
+ apply(db.State, f)
+ cur, last = f.Max, f
+ if db.Key == 0 {
+ db.Key = f.SealKey
}
- found = true
- switch {
- case extends(v.B.Writes, history):
- history = append([]int(nil), v.B.Writes...)
- if db.Key == 0 {
- db.Key = v.B.SealKey
- }
- if v.B.Inc != 0 {
- db.Inc = v.B.Inc
- }
- case extends(history, v.B.Writes):
- // an older checkpoint, or a duplicate: skipped
- default:
- w.logf("box %d restore refused: %s version %d does not continue the chain", b.ID, dataKey(loc, gen, "ltx"), v.ID)
- w.stats["restore refused: conflicting history"]++
- return nil
+ if f.Inc != 0 {
+ db.Inc = f.Inc
}
}
- if !found && gen != "" {
- return nil // an installed generation with nothing in it: refuse
+ 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.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"]++
+ return nil
}
- db.Writes = history
+ if last == nil && gen != "" {
+ return nil
+ }
+ db.TX = cur
return db
}
-// fenced: the local retired marker refuses a writer for incarnation inc
-// carrying operation sequence seq (claim rev 6 §6.4, R3-3). The mutation
-// is rev 5.2's marker, which fences the whole incarnation for good.
-func (w *world) fenced(b *box, inc, seq int) bool {
- f, ok := b.Fence[inc]
- if !ok {
- return false
- }
- if !on("fence-by-installation") {
+// stampOK is admission's local check (claim rev 6.2 §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
+// named: the oracle records it rather than let the move roll back.
+func (w *world) stampOK(b *box, c *claim, probe string) bool {
+ if b.admitsStamp(c.Holder.Inc, c.readStamp) {
return true
}
- return seq <= f
-}
-
-func unionInts(a, b []int) []int {
- seen := map[int]bool{}
- var out []int
- for _, x := range append(append([]int(nil), a...), b...) {
- if !seen[x] {
- seen[x] = true
- out = append(out, x)
- }
+ if probe != "" && c.Op.Kind == "landing" && c.Op.Tgt == b.ID && c.Ver == w.claim.Ver {
+ w.violate("INV-3: the landing probe of %s on box %d was refused by the box's own installation record (stamp %d, active %d)", c.Op.ID, b.ID, c.readStamp, b.stampOf(c.Holder.Inc))
}
- return out
+ return false
}
// bucketGen reads generation.json at loc in b's own bucket: the generation
@@ -294,11 +282,18 @@ func (w *world) spawn(b *box, c *claim, want string, probe bool) bool {
// an old executable: no claim, no markers, no locations; it
// restores whatever is newest at the legacy root
if b.DB == nil {
- b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/"}
- if v := w.buckets[b.Region].current("instances/" + host + "/ltx"); v != nil {
- b.DB.Writes, b.DB.Key = append([]int(nil), v.B.Writes...), v.B.SealKey
- if v.B.Inc != 0 {
- b.DB.Inc = v.B.Inc
+ b.DB = &localDB{Inc: h.Inc, Loc: "instances/" + host + "/", State: map[string]int{}}
+ var newest *version
+ for _, k := range w.buckets[b.Region].keys("instances/" + host + "/ltx/") {
+ if v := w.buckets[b.Region].current(k); v != nil && (newest == nil || v.ID > newest.ID) {
+ newest = v
+ }
+ }
+ if newest != nil {
+ apply(b.DB.State, newest.B)
+ b.DB.TX, b.DB.Key = newest.B.Max, newest.B.SealKey
+ if newest.B.Inc != 0 {
+ b.DB.Inc = newest.B.Inc
}
}
}
@@ -401,8 +396,8 @@ func (w *world) admits(b *box, c *claim, probe string) bool {
if on("exec-admission") && c.Op.Exec != b.ID {
return false
}
- if w.fenced(b, c.Holder.Inc, w.opSeq(c.Op.ID)) {
- return false // retired here by this operation or a later one (claim rev 6 §6.4)
+ if !w.stampOK(b, c, probe) {
+ return false // a retired installation's worker (claim rev 6.2 §6.4)
}
if w.isParticipant(c, b) && b.ExecUntil > 0 && w.localNow(b) > b.ExecUntil {
return false // self-fenced until the lease is renewed
@@ -549,15 +544,10 @@ 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.
-func (w *world) localPark(b *box, inc int, seq int) bool {
- if on("hostguard") {
- if f, ok := b.Fence[inc]; !ok || f < seq {
- if b.Fence == nil {
- b.Fence = map[int]int{}
- }
- b.Fence[inc] = seq // {incarnation, op of the installation being retired}
- return false
- }
+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)
+ return false
}
switch {
case b.DB != nil:
@@ -604,7 +594,7 @@ func (w *world) doObligation(b *box, c *claim, o obligation) {
if b.DB != nil && b.DB.Inc != 0 {
inc = b.DB.Inc
}
- if !w.localPark(b, inc, w.opSeq(o.Op)) {
+ if !w.localPark(b, inc) {
return // the next artifact on a later step
}
if b.Row != nil && b.Row.HoldOp == o.Op || b.Row != nil && (c.Holder == nil || b.ID != c.Op.Exec) {
diff --git a/internal/sim/hostlife/rules_test.go b/internal/sim/hostlife/rules_test.go
index 4071d38f5..5d2eeec70 100644
--- a/internal/sim/hostlife/rules_test.go
+++ b/internal/sim/hostlife/rules_test.go
@@ -54,7 +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 §6.4 (R3-3): the retired marker is {incarnation, op}; a new installation replaces it; the mutation fences the whole incarnation for good
+ "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
"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
@@ -79,10 +80,13 @@ var allRules = []string{
"participant-rows", // claim §2/R2-4
"stale-hold-release", // this model's finding: release a hold whose op ended
"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
"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
// 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 557a99bc9..e6e8687ac 100644
--- a/internal/sim/hostlife/sim_test.go
+++ b/internal/sim/hostlife/sim_test.go
@@ -17,12 +17,12 @@ const (
func newWorld(seed int64, verbose bool) *world {
w := &world{seed: seed, rng: rand.New(rand.NewSource(seed)), verbose: verbose,
stores: map[string]bool{app0: true, app1: true}, cfgVer: map[string]int{app0: 1, app1: 1}, purgeRes: map[string]bool{}, purgeStep: map[string]int{},
- control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}}
+ control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}}
for r := 0; r < nRegions; r++ {
w.buckets[r] = newBucket()
}
w.admit = map[string]*admission{app0: {Reg: map[string]regEntry{}}, app1: {Reg: map[string]regEntry{}}}
- for _, f := range []string{"partition", "freeze", "pause", "destroy", "disk", "recreate", "pausecreate", "lostreturn"} {
+ for _, f := range []string{"partition", "freeze", "pause", "destroy", "disk", "recreate", "pausecreate", "lostreturn", "markerhold", "freezetgt", "retention"} {
w.fail[f] = w.rng.Intn(2) == 0
}
for r := 0; r < nRegions; r++ {
@@ -41,10 +41,34 @@ func newWorld(seed int64, verbose bool) *world {
}
if w.rng.Intn(3) == 0 {
w.legacyStart()
+ } else if w.fail["retention"] {
+ w.retentionStart()
}
return w
}
+// retentionStart: the host has been recreated more times than
+// retired_incarnations remembers (claim rev 6.2 §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++ {
+ w.nextInc++
+ w.claim.Retired = retire(w.claim.Retired, w.nextInc)
+ }
+ 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.logf("retention fixture: incarnation %d's replica in region %d, past the cap", old, r)
+}
+
// legacyStart: a host that exists before the release. Its data is untagged,
// a best-effort-placed descendant may be running, and an earlier
// (pre-release) move left a replica in another region.
@@ -56,21 +80,28 @@ func (w *world) legacyStart() {
w.rec = &record{Ver: w.newVer(), Inc: inc, App: app0, Region: r}
w.legacyIncs[inc] = true
key := 7
- var writes []int
+ b.Row = &row{Inc: inc, App: app0, Status: "ready"}
+ b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}}
for i := 0; i < 3; i++ {
w.nextW++
- writes = append(writes, w.nextW)
w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: 0, Kind: "write"})
w.writeInc[w.nextW] = inc
+ w.commit(b.DB, w.nextW)
}
- b.Row = &row{Inc: inc, App: app0, Status: "ready"}
- b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, Writes: writes}
b.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true}
b.Config, b.CfgVer, b.Live, b.LiveInc = inc, 1, app0, inc
- w.buckets[r].objs["instances/"+host+"/ltx"] = []*version{{ID: w.newVer(), B: &blob{Kind: "ltx", Inc: 0, Writes: writes, SealKey: key}}}
- w.markDurable("instances/"+host+"/ltx", &blob{Writes: writes})
+ // 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)
+ 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}}
other := (r + 1 + w.rng.Intn(nRegions-1)) % nRegions
- w.buckets[other].objs["instances/"+host+"/ltx"] = []*version{{ID: w.newVer(), B: &blob{Kind: "ltx", Inc: 0, Writes: writes[:1], SealKey: key}}}
+ w.buckets[other].objs[ltxKey(b.DB.Loc, "", 0, 1, 1)] = []*version{{ID: w.newVer(), B: part}}
if w.rng.Intn(2) == 0 {
b.Procs = []*proc{{Main: true, Strict: false, Inc: inc, Box: b.ID}, {Main: false, Strict: false, Inc: inc, Box: b.ID}}
}
@@ -78,7 +109,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, Writes: writes[:1]}
+ ob.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: cloneState(part.Pages), TX: 1, Rep: 1}
ob.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true}
}
}
@@ -304,8 +335,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, "base")] = true
- live[dataKey(loc, gen, "ltx")] = true
+ live[dataKey(loc, gen, "ltx/")] = true // a prefix: every current LTX file of the generation
live[genMarkerKey(loc)] = true
}
}
@@ -313,7 +343,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]) {
+ if strings.HasSuffix(k, "lock.json") || (r == serving && (live[k] || live[genPrefixOf(k)+"ltx/"] && genPrefixOf(k) != "")) {
continue
}
if exceptStragglers && w.owed(r, k) {
@@ -356,6 +386,8 @@ func hasNonMarker(vs []*version) bool {
// canStart is the start check and admission, computed without starting
// anything: the final oracle never writes.
func (w *world) canStart(e *box, c *claim) string {
+ c = c.clone()
+ c.readStamp = e.stampOf(c.Holder.Inc) // as a fresh read by e would carry it
if !w.admits(e, c, "") {
return "not admitted"
}
@@ -416,14 +448,16 @@ func (w *world) final() {
w.violate("final: the serving box %d has no database and nothing to restore", e.ID)
return
}
- w.checkAcks(c, e, local.Writes, "local")
+ w.checkLocal(c, e, local)
w.checkDurable(c, "final")
}
// checkDurable is the cold-restore promise, checked at every release to
// serving and at the end: a restore of the claim's location and generation
-// in the serving region holds every acknowledged write that a completed
-// replica object ever held. It reads the bucket and writes nothing.
+// in the serving region holds the effect of every acknowledged write that
+// a completed replica object ever covered, and equals the database the
+// acknowledged writes up to its TXID produce. It reads the bucket and
+// writes nothing.
func (w *world) checkDurable(c *claim, when string) {
if c == nil || c.Holder == nil {
return
@@ -441,53 +475,70 @@ func (w *world) checkDurable(c *claim, when string) {
need[x] = true
}
}
- have := map[int]bool{}
- if db != nil {
- for _, x := range db.Writes {
- have[x] = true
- }
- w.foreignWrites(db, c.Holder.Inc, "cold restore, "+when)
- }
for _, a := range w.acks {
if a.Inc != c.Holder.Inc || a.Exempt || a.Kind != "write" || !need[a.W] {
continue
}
- if !have[a.W] {
+ 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)
return
}
}
+ if db == nil {
+ return
+ }
+ w.foreignWrites(db, c.Holder.Inc, "cold restore, "+when)
+ if why, ok := statesEqual(w.expectState(c.Holder.Inc, db.TX), db.State); !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, "ltx/")))
+ }
+}
+
+// ltxShape describes the current LTX files under prefix, for diagnosis.
+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")
+ continue
+ }
+ parts = append(parts, fmt.Sprintf("%s:%v", strings.TrimPrefix(k, prefix), v.B.Pages))
+ }
+ return "files " + strings.Join(parts, " ")
}
// foreignWrites: a database restored or started for incarnation inc holds
-// no write another incarnation acknowledged (C1, INV-4, claim C3a).
+// no key another incarnation wrote (C1, INV-4, claim C3a).
func (w *world) foreignWrites(db *localDB, inc int, how string) {
- for _, x := range db.Writes {
- if i, ok := w.writeInc[x]; ok && i != inc {
- w.violate("C3a (%s): incarnation %d's database holds write %d of incarnation %d", how, inc, x, i)
+ for k, v := range db.State {
+ if v == 0 {
+ continue
+ }
+ if i, ok := w.writeInc[keyID(k)]; ok && i != inc {
+ w.violate("C3a (%s): incarnation %d's database holds key %s of incarnation %d", how, inc, k, i)
return
}
}
}
-func (w *world) checkAcks(c *claim, e *box, writes []int, how string) {
- have := map[int]bool{}
- for _, x := range writes {
- have[x] = true
+// checkLocal: the serving box's database holds the effect of every
+// acknowledged, non-exempt write of the incarnation (INV-2), and every
+// acknowledged timer is delivered or still registered.
+func (w *world) checkLocal(c *claim, e *box, db *localDB) {
+ if why, ok := statesEqual(w.expectState(c.Holder.Inc, -1), db.State); !ok {
+ w.violate("INV-2 (local): the serving database differs from the acknowledged writes: %s", why)
+ return
}
timers := map[int]bool{}
for _, x := range e.Timers {
timers[x] = true
}
for _, a := range w.acks {
- if a.Inc != c.Holder.Inc || a.Exempt {
+ if a.Inc != c.Holder.Inc || a.Exempt || a.Kind != "timer" {
continue
}
- if a.Kind == "write" && !have[a.W] {
- w.violate("INV-2 (%s): acknowledged write %d (box %d, t=%d) lost", how, a.W, a.Box, a.At)
- return
- }
- if a.Kind == "timer" && how == "local" && !w.delivered[a.W] && !timers[a.W] {
+ if !w.delivered[a.W] && !timers[a.W] {
w.violate("INV-2: acknowledged timer %d (box %d, t=%d) lost", a.W, a.Box, a.At)
return
}
diff --git a/internal/sim/hostlife/witness_test.go b/internal/sim/hostlife/witness_test.go
index ee3bb080b..000688553 100644
--- a/internal/sim/hostlife/witness_test.go
+++ b/internal/sim/hostlife/witness_test.go
@@ -87,7 +87,7 @@ var witnesses = []witness{
{"registration-check", "registration-check,purge-claim-fence", "store of app",
"the same, with the fence gone instead", wRegistrationCheck},
// rev 7.4
- {"fence-by-installation", "fence-by-installation", "INV-3 (witness): instance back on box 0",
+ {"fence-by-installation", "fence-by-installation", "refused by the box's own installation record",
"a round trip UK→US→UK (R3-3): the second landing replaces the retired marker UK wrote after the first move; a marker for the whole incarnation refuses the probe, the move back rolls back, and the instance never serves on UK again", wRoundTrip},
{"orphan-order", "orphan-order", "orphan cleanup deleted the record",
"the cleanup reads the claim first and pauses; a create removes the orphan, claims and publishes; the cleanup reads that record, judges it by the stale claim and deletes it (R3-2)", wOrphanOrder},
@@ -95,6 +95,15 @@ var witnesses = []witness{
"a replication PUT of incarnation 1 completes after 33 recreates dropped it from retired_incarnations; a sweep reading that list never finds it (R3-4)", wSweepNamespace},
{"recovery-current-inputs", "recovery-current-inputs", "stuck: source could not recover",
"a bundle is published after admission and the target writes; the dirty recovery renders current inputs; gated on move.inputs it never passes its probe (R2-5, §7.8)", wRecoveryInputs},
+ // rev 7.5
+ {"fence-stamp", "fence-stamp,hostguard", "INV-6",
+ "a config pass selected under the first installation pauses across a retire-and-reinstall (UK→US→UK) and a bundle change; the host guard's fresh check refuses its stale bundle, the stamp refuses its retired installation; with both gone it writes host state for a retired installation (R4-2)", wStaleWriter},
+ {"live-set-region", "live-set-region", "residue never retired",
+ "a legacy host serving in UK with an earlier replica in US; discovery parks US; an old US replication PUT lands after; the live set is region-qualified, so the US sweep parks it; exempted by key alone it stays for good (R4-3)", wLiveSetRegion},
+ {"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},
}
// TestWitnesses is the deterministic half of the mutation evidence.
@@ -243,7 +252,7 @@ func dump(t *testing.T, w *world) {
func newWitness() *world {
w := &world{seed: -1, rng: quiet{}, verbose: true, now: 100,
stores: map[string]bool{app0: true, app1: true}, cfgVer: map[string]int{app0: 1, app1: 1}, purgeRes: map[string]bool{}, purgeStep: map[string]int{},
- control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}}
+ control: newBucket(), delivered: map[int]bool{}, durable: map[string]map[int]bool{}, writeInc: map[int]int{}, txs: map[int]tx{}, legacyIncs: map[int]bool{}, purgeOp: map[string]string{}, genParent: map[string]string{}, execLease: map[int]execLease{}, fail: map[string]bool{}, stats: map[string]int{}}
for r := 0; r < nRegions; r++ {
w.buckets[r] = newBucket()
b := &box{ID: r, Region: r, HasKey: true}
@@ -260,12 +269,12 @@ func newWitness() *world {
func (w *world) serve(b *box, inc int, app string) {
loc := locationFor(inc, false)
key := 1000 + inc
- var writes []int
+ db := &localDB{Inc: inc, Loc: loc, Key: key, State: map[string]int{}}
for i := 0; i < 2; i++ {
w.nextW++
- writes = append(writes, w.nextW)
w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: w.now, Kind: "write"})
w.writeInc[w.nextW] = inc
+ w.commit(db, w.nextW)
}
if inc > w.nextInc {
w.nextInc = inc
@@ -282,11 +291,13 @@ func (w *world) serve(b *box, inc int, app string) {
w.claim = n
w.rec = &record{Ver: w.newVer(), Inc: inc, App: app, Region: b.Region}
b.Row = &row{Inc: inc, App: app, Status: "ready"}
- b.DB = &localDB{Inc: inc, Loc: loc, Key: key, Writes: append([]int(nil), writes...)}
+ 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
- k := dataKey(loc, "", "ltx")
- bl := &blob{Kind: "ltx", Inc: inc, Writes: append([]int(nil), writes...), SealKey: key}
+ b.install(inc, n.Ver) // the create's installation stamp
+ bl := w.deltaOf(db, inc)
+ db.Rep, db.Pending = db.TX, 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)
w.logf("witness: inc %d of app %s serving on box %d", inc, app, b.ID)
@@ -300,21 +311,24 @@ func (w *world) legacy(b *box, other int) {
w.rec = &record{Ver: w.newVer(), Inc: inc, App: app0, Region: b.Region}
w.legacyIncs[inc] = true
key := 7
- var writes []int
+ b.Row = &row{Inc: inc, App: app0, Status: "ready"}
+ b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, State: map[string]int{}}
for i := 0; i < 2; i++ {
w.nextW++
- writes = append(writes, w.nextW)
w.acks = append(w.acks, ack{W: w.nextW, Inc: inc, Box: b.ID, At: w.now, Kind: "write"})
w.writeInc[w.nextW] = inc
+ w.commit(b.DB, w.nextW)
}
- b.Row = &row{Inc: inc, App: app0, Status: "ready"}
- b.DB = &localDB{Inc: inc, Loc: locationFor(inc, true), Key: key, Writes: writes}
b.Files = &filesDir{Inc: inc, Key: key, Env: 1, Done: true}
b.Config, b.CfgVer, b.Live, b.LiveInc = inc, 1, app0, inc
- k := "instances/" + host + "/ltx"
- w.buckets[b.Region].objs[k] = []*version{{ID: w.newVer(), B: &blob{Kind: "ltx", Inc: 0, Writes: writes, SealKey: key}}}
- w.markDurable(k, &blob{Writes: writes})
- w.buckets[other].objs[k] = []*version{{ID: w.newVer(), B: &blob{Kind: "ltx", Inc: 0, Writes: writes[:1], SealKey: key}}}
+ 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)
+ 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}}
+ w.buckets[other].objs[ltxKey(b.DB.Loc, "", 0, 1, 1)] = []*version{{ID: w.newVer(), B: part}}
}
// ---------- the schedule ----------
@@ -447,7 +461,9 @@ func (w *world) appWrite(b *box) int {
w.nextW++
wr := w.nextW
if b.DB != nil && b.DB.Inc == p.Inc {
- b.DB.Writes = append(b.DB.Writes, wr)
+ w.commit(b.DB, wr)
+ } else {
+ w.txs[wr] = tx{W: wr}
}
w.acks = append(w.acks, ack{W: wr, Inc: p.Inc, Box: b.ID, At: w.now, Kind: "write"})
w.writeInc[wr] = p.Inc
@@ -455,17 +471,21 @@ func (w *world) appWrite(b *box) int {
return wr
}
-// holdKey makes the witness hold every request for key until completeHeld.
+// holdKey makes the witness hold every request for key, or under a prefix
+// ending in "/", until completeHeld.
func (w *world) holdKey(key string) {
- w.hold = func(r *request) bool { return r.key == key }
+ w.hold = func(r *request) bool {
+ return r.key == key || strings.HasSuffix(key, "/") && strings.HasPrefix(r.key, key)
+ }
}
-// completeHeld completes the requests the witness held for key, now.
+// completeHeld completes the requests the witness held for key (or
+// prefix), now.
func (w *world) completeHeld(key string) int {
n := 0
var keep []*request
for _, r := range w.pending {
- if r.key == key && r.doneAt == 1<<30 {
+ if (r.key == key || strings.HasSuffix(key, "/") && strings.HasPrefix(r.key, key)) && r.doneAt == 1<<30 {
w.complete(r)
w.logf("witness: held %v %s r%d completes now", r.kind, r.key, r.region)
n++
@@ -792,6 +812,7 @@ func (w *world) recreate(t *testing.T, r, inc int) {
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) })
w.appWrite(b)
+ w.appWrite(b) // two transactions: its first LTX file has the same key as the fixture's
w.replicate(b)
}
@@ -857,8 +878,8 @@ func wRoundTrip(t *testing.T, w *world) {
c := w.claim
return c.Op.Kind == "serving" && c.Op.Exec == us.ID && w.rec != nil && w.rec.MoveID == "" && len(c.Obl) == 0
}, w.act(uk, us), w.recoveryPass)
- if _, fenced := uk.Fence[1]; !fenced {
- t.Fatalf("witness: box 0 wrote no retired marker: %s%s", w.state(), w.boxesState())
+ if st := uk.Stamps[1]; st == nil || st.Active != 0 || len(st.Retired) == 0 {
+ t.Fatalf("witness: box 0 did not retire its installation: %+v %s%s", st, w.state(), w.boxesState())
}
w.moveTo(uk.Region)
if w.claim.Op.Kind != "moving" {
@@ -917,7 +938,8 @@ func wSweepNamespace(t *testing.T, w *world) {
if !hasMain(b) {
t.Fatalf("witness: the host did not wake: %s", w.state())
}
- key := dataKey(w.claim.Holder.Location, "", "ltx")
+ w.appWrite(b)
+ key := dataKey(w.claim.Holder.Location, "", "ltx/")
w.holdKey(key)
w.replicate(b)
w.hold = nil
@@ -961,3 +983,158 @@ func wRecoveryInputs(t *testing.T, w *world) {
w.rollback(w.claim, "settle check failed")
w.finish(t)
}
+
+// ---------- rev 7.5 ----------
+
+// roundTrip moves the host UK→US→UK and requires it to serve on UK again,
+// running.
+func (w *world) roundTrip(t *testing.T, uk, us *box) {
+ w.moveTo(us.Region)
+ w.reach(t, 1500, "serving on box 1 after the first move", func() bool {
+ c := w.claim
+ return c.Op.Kind == "serving" && c.Op.Exec == us.ID && w.rec != nil && w.rec.MoveID == "" && len(c.Obl) == 0
+ }, w.act(uk, us), w.recoveryPass)
+ w.moveTo(uk.Region)
+ if w.claim.Op.Kind != "moving" {
+ t.Fatalf("witness: the move back did not start: %s", w.state())
+ }
+ w.reach(t, 1500, "instance back on box 0, running", func() bool {
+ c := w.claim
+ return c.Op.Kind == "serving" && c.Op.Exec == uk.ID && hasMain(uk)
+ }, w.act(uk, us), w.recoveryPass, w.operator)
+}
+
+// wStaleWriter (claim rev 6.2 §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) {
+ uk, us := w.boxes[0], w.boxes[1]
+ w.serve(uk, 1, app0)
+ w.passPlan(uk, "config")
+ if len(uk.plans) != 1 {
+ t.Fatalf("witness: no plan")
+ }
+ p := uk.plans[0]
+ p.At = 1 << 30 // paused
+ w.roundTrip(t, uk, us)
+ if len(w.violations) > 0 {
+ return
+ }
+ w.cfgVer[app0]++
+ p.At = w.now // the pass resumes and takes the guard
+ w.tick(func() { w.passExec(uk) })
+ w.finish(t)
+}
+
+// wLiveSetRegion (claim rev 6.2 §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.
+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"})
+ 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 {
+ t.Fatalf("witness: no held PUT")
+ }
+ w.finish(t)
+}
+
+// pushed installs a push generation on b: the host serves from a snapshot
+// at its current TXID, then runs again, so the witness can shape the
+// generation's LTX files by hand.
+func (w *world) pushed(t *testing.T, b *box) (loc, gen string) {
+ w.userPush()
+ 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) {
+ t.Fatalf("witness: the host did not wake after the push: %s%s", w.state(), w.boxesState())
+ }
+ 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 {
+ 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 {
+ 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)
+ }
+ }
+ return out
+}
+
+// wRestoreOverlap (R4-1): after a push the generation's snapshot ends at
+// TXID 11 (two writes, eight more, the push's mark). The host writes four
+// more; compaction has left one current file [10,15] and no level-0
+// files. The disk is lost and the host wakes cold.
+func wRestoreOverlap(t *testing.T, w *world) {
+ b := w.boxes[0]
+ w.serve(b, 1, app0)
+ w.wake(b)
+ for i := 0; i < 8; i++ {
+ w.appWrite(b)
+ }
+ w.replicate(b)
+ loc, gen := w.pushed(t, b)
+ 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)
+ }
+ 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
+ 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.reach(t, 20, "woken cold", func() bool { return hasMain(b) }, func() { w.wake(b) })
+ w.finish(t)
+}
+
+// wRestoreGap (R4-1): the same generation with a genuine gap: the file
+// [12,12] never landed and [13,13] is current. The restore must refuse.
+func wRestoreGap(t *testing.T, w *world) {
+ b := w.boxes[0]
+ w.serve(b, 1, app0)
+ w.wake(b)
+ for i := 0; i < 8; i++ {
+ w.appWrite(b)
+ }
+ w.replicate(b)
+ loc, gen := w.pushed(t, b)
+ 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
+ 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)
+ }
+}
diff --git a/internal/sim/hostlife/world_test.go b/internal/sim/hostlife/world_test.go
index beaae28e6..7e1299163 100644
--- a/internal/sim/hostlife/world_test.go
+++ b/internal/sim/hostlife/world_test.go
@@ -64,31 +64,29 @@ func retire(xs []int, inc int) []int {
return xs
}
-// extends: history a is b with zero or more later transactions appended
-// (R3-1: a replica object continues the chain only if its history is an
-// extension of what the chain holds).
-func extends(a, b []int) bool {
- if len(a) < len(b) {
- return false
- }
- for i := range b {
- if a[i] != b[i] {
- return false
- }
- }
- return true
-}
-
// ---------- storage ----------
type blob struct {
Kind string // "ltx", "base", "files", "snap", "lock", "manifest", "lease", "report", "gen"
Inc int // incarnation tag; 0 = untagged legacy
Gen string // lineage that wrote it
- Base int // a "gen" marker: the version id of the generation's snapshot (R3-1)
Seq int
SealKey int // the instance.key the data is sealed with
- Writes []int
+ // 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
+ 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
+ Base int
+ SnapKey string
+ TX int
Files *filesDir
Timers []int
Refs []ref // a park manifest's archive set
@@ -198,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.HasSuffix(k, "/base") || strings.HasSuffix(k, "/ltx"))
+ return strings.HasPrefix(k, "instances/") && strings.Contains(k, "/ltx/")
}
// ---------- requests (late completion) ----------
@@ -225,6 +223,7 @@ type request struct {
beyond bool // completes after H
ifSet bool // a conditional PUT: If-Match ifVer, or If-None-Match when ifVer is 0
ifVer int
+ hold bool // the biased schedule holds it until its worst moment (rev 7.5: a generation marker)
by string
// result, for an immediate completion
resultVer int
@@ -249,12 +248,16 @@ func (f *filesDir) clone() *filesDir {
}
type localDB struct {
- Inc int
- Loc string // the holder location it was restored from (claim C3a)
- Gen string // <dbPath>.gen
- Key int // the key it was sealed with
- Writes []int
- Seq int
+ Inc int
+ Loc string // the holder location it was restored from (claim C3a)
+ 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)
+ Log []int // every write applied here, for the oracle
+ Seq int
}
func (d *localDB) clone() *localDB {
@@ -262,7 +265,9 @@ func (d *localDB) clone() *localDB {
return nil
}
c := *d
- c.Writes = append([]int(nil), d.Writes...)
+ c.State = cloneState(d.State)
+ c.Pending = append([]int(nil), d.Pending...)
+ c.Log = append([]int(nil), d.Log...)
return &c
}
@@ -310,12 +315,11 @@ type box struct {
History []int // delivered history (disposable)
Attempt int // runner bookkeeping
Procs []*proc
- // Fence is the local retired marker, per incarnation (claim rev 6 §6.4,
- // R3-3): {incarnation, op}. It refuses a writer for that incarnation
- // whose operation is at or before the fence's; a new installation of
- // the incarnation by a later operation replaces it under the guard.
- // The value is the sequence of the fence's op (opSeq).
- Fence map[int]int
+ // Stamps is the local record per host incarnation (claim rev 6.2 §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.
+ Stamps map[int]*stamps
Parked int // count of locally parked artifacts
ExecUntil int // local exec-lease file (watchdog input); 0 none
@@ -342,27 +346,75 @@ type plan struct {
App string
Ver int // the config bundle version it selected
At int // it takes the guard no earlier than this
- // Seq is the operation identity the pass validated under (R2-3): the
- // sequence of the newest operation when it selected the row. The
- // host guard refuses it once the local fence is at or past it.
- Seq int
-}
-
-// opSeq orders operations: every op id the model mints carries the
-// global counter, so a later operation has a larger sequence. The serving
-// state has no op id and sorts as "now". Claim rev 6's marker names the
-// op id of the installation; the ordering it needs is this one.
-func (w *world) opSeq(id string) int {
- if id == "" || id == "sweep" || id == "adopt" {
- return w.nextID
- }
- n := 0
- for _, r := range id {
- if r >= '0' && r <= '9' {
- n = n*10 + int(r-'0')
- }
+ // 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
+ // 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
+// stamps of the installations retired here.
+type stamps struct {
+ Active int
+ Retired []int
+}
+
+func (b *box) stampOf(inc int) int {
+ if s := b.Stamps[inc]; s != nil {
+ return s.Active
+ }
+ 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.
+func (b *box) install(inc, gen int) {
+ if b.Stamps == nil {
+ b.Stamps = map[int]*stamps{}
+ }
+ if b.Stamps[inc] == nil {
+ b.Stamps[inc] = &stamps{}
}
- return n
+ b.Stamps[inc].Active = gen
+}
+
+// retireStamp moves the active stamp into the retired set and sets none.
+func (b *box) retireStamp(inc int) {
+ if b.Stamps == nil {
+ b.Stamps = map[int]*stamps{}
+ }
+ if b.Stamps[inc] == nil {
+ b.Stamps[inc] = &stamps{}
+ }
+ if a := b.Stamps[inc].Active; a != 0 {
+ b.Stamps[inc].Retired = append(b.Stamps[inc].Retired, a)
+ }
+ 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.
+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
+ }
+ if !on("fence-by-installation") && len(s.Retired) > 0 {
+ return false
+ }
+ if s.Active == 0 {
+ return false
+ }
+ if !on("fence-stamp") {
+ return true
+ }
+ return stamp == s.Active
}
func (b *box) forget() {
@@ -442,7 +494,11 @@ type opState struct {
LostBox int
LostAt int
RollbackID string
- PushWrites []int
+ 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 claim gen of the flip CAS: the landing's installation stamp (claim rev 6.2 §6.4)
Committed bool
Opened bool
ParkLocalNow bool
@@ -491,6 +547,10 @@ type claim struct {
// Cancelled is cancelled_creates (claim rev 4 §5.1, J2-R2-7): creates
// recovery cancelled, so a late record PUT can be removed. It survives free.
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)
+ readStamp int
}
type cancelledCreate struct {
@@ -591,6 +651,7 @@ type world struct {
pending []*request
acks []ack
writeInc map[int]int // which incarnation acknowledged each write
+ txs map[int]tx // every write as a transaction (ltx_test.go)
durable map[string]map[int]bool // per generation prefix: writes a completed replica object holds
genParent map[string]string // generation prefix -> the one it superseded
nextW int
@@ -643,6 +704,15 @@ func (w *world) newVer() int {
return w.ver
}
+// 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).
+func (w *world) nextGen() int {
+ if w.claim == nil {
+ return 1
+ }
+ return w.claim.Ver + 1
+}
+
func (w *world) boxByID(id int) *box {
for _, b := range w.boxes {
if b.ID == id {
@@ -694,7 +764,6 @@ func (c *claim) clone() *claim {
n.Holder = &h
}
n.Op.Parts = append([]int(nil), c.Op.Parts...)
- n.Op.PushWrites = append([]int(nil), c.Op.PushWrites...)
n.Op.StopFiles = c.Op.StopFiles.clone()
n.Regions = append([]int(nil), c.Regions...)
n.Boxes = append([]int(nil), c.Boxes...)
@@ -748,6 +817,14 @@ func (w *world) issue(actorBox int, r *request) int {
w.logf("box %d issued %v %s r%d (held by the witness)", actorBox, r.kind, r.key, r.region)
return 0
}
+ if r.hold {
+ // the biased schedule: accepted, completing at its worst moment or
+ // long after everything else
+ r.doneAt, r.by, r.beyond = w.now+40*horizonH, fmt.Sprintf("box:%d", actorBox), true
+ w.pending = append(w.pending, r)
+ w.logf("box %d issued %v %s r%d (held for its worst moment)", actorBox, r.kind, r.key, r.region)
+ return 0
+ }
if w.rng.Intn(10) < 7 {
return w.complete(r)
}
@@ -830,11 +907,11 @@ func (w *world) markDurable(key string, b *blob) {
if b == nil || !isDataKey(key) {
return
}
- p := key[:strings.LastIndex(key, "/")+1]
+ p := genPrefixOf(key)
if w.durable[p] == nil {
w.durable[p] = map[int]bool{}
}
- for _, x := range b.Writes {
+ for _, x := range b.TXs {
w.durable[p][x] = true
}
}
@@ -863,6 +940,11 @@ func (w *world) newGen(loc, old, gen string) {
// is the interleaving C3a exists for (a recreate while the previous
// incarnation's requests are still in flight).
func (w *world) worstMoment(r *request) bool {
+ if r.hold && r.b != nil && r.b.Kind == "gen" {
+ // a held marker publication lands once a newer marker is current
+ cur := w.buckets[r.region].current(r.key)
+ return cur != nil && cur.B.Gen != r.b.Gen
+ }
if !w.fail["recreate"] || r.kind != rPut || r.b == nil || r.b.Inc == 0 || !isDataKey(r.key) {
return false
}