amadan
sim: litestreamfacts: exact assertions, anchor replay bytes, the anchored lineage through the Store (R6-2, R6-5)
Every fact is asserted exactly and fails on any alternative behaviour (R6-2): the newest level-0 file retained by identity through TXID- and time-based retention, the next TXID exactly N+1 after an anchored or an ordinary restore, every row restored (a run that kept 7 of 9 fails), the original continuation gone after the second rotation, the rooted restore refused with litestream's own "non-contiguous ltx files" message for the gap after TXID 1, and the newest restore's exact row count. TestFact_R6_5_AnchorReplayBytes: the anchor's bytes are identical for the same inputs and differ with the header timestamp, so a replayed PUT must persist the timestamp or reuse the object (R6-5). TestGate_AnchoredLineageThroughStore: one anchored lineage through internal/lease's Store shape (metadata directory, replica client, the compaction ladder, snapshot and L0 retention monitors) with every interval shortened: three rounds of eight writes with level 1-3 compaction, snapshot rotation and time-based level-0 expiry running, then disk loss and a cold restore; exact positions (every sync advances by one, a reopened box continues at the newest remote level-0 TXID plus one) and exact row counts. The package doc says what the file client cannot show and that the repository has no S3 fake litestream can use.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 file changed,
+202
−48
internal/sim/litestreamfacts/facts_test.go+202 −48
diff --git a/internal/sim/litestreamfacts/facts_test.go b/internal/sim/litestreamfacts/facts_test.go| index d3dc027e3..86cac14bf 100644 |
| --- a/internal/sim/litestreamfacts/facts_test.go |
| +++ b/internal/sim/litestreamfacts/facts_test.go |
| @@ -1,10 +1,24 @@ |
| // Package litestreamfacts establishes, by observation of the pinned |
| // libraries (litestream v0.5.14, superfly/ltx v0.5.1), the facts the |
| -// move-any-app spec's restore contract rests on (moves spec rev 7.6 §5.1; |
| -// joint round 5, R5-1 to R5-3). It is test-only and drives litestream |
| -// exactly as internal/lease does, with the file replica client the lease |
| -// tests use. Each test logs what it observed and asserts the fact the spec |
| -// relies on, so a library upgrade that changes the fact fails here first. |
| +// move-any-app spec's restore contract rests on (moves spec rev 7.7 §5.1; |
| +// joint rounds 5 and 6, R5-1 to R5-3, R6-2, R6-5). It is test-only and |
| +// drives litestream as internal/lease does, with the file replica client |
| +// the lease tests use. Each test asserts the exact fact the spec relies on |
| +// and fails on any alternative behaviour, so a library upgrade that |
| +// changes a fact fails here first (R6-2): a passing run is the contract, |
| +// not a record of one observation. TestGate_AnchoredLineageThroughStore |
| +// takes one anchored lineage through the production Store configuration |
| +// (internal/lease's Held.Replicate, intervals shortened) with repeated |
| +// writes, compaction, snapshot rotation, time-based level-0 expiry, disk |
| +// loss and repeated cold restores, asserting exact positions and contents. |
| +// |
| +// What the file client cannot show (R6 review): S3 listings order and age |
| +// objects by LastModified, the file client by the encoded LTX timestamp |
| +// kept as the file's mtime; S3 keeps versions and delete markers, the file |
| +// client replaces and deletes outright; S3 requests can complete late. The |
| +// repository has no S3 fake that litestream's client can talk to (the |
| +// lease tests' fakeMirror is a MirrorStore seam), so those remain the |
| +// model's (internal/sim/hostlife) and §15 step 8's. |
| // |
| // The observed facts are recorded in ~/.cache/nearest-region/litestream-facts.md. |
| package litestreamfacts |
| @@ -13,7 +27,6 @@ import ( |
| "bytes" |
| "context" |
| "database/sql" |
| - "errors" |
| "fmt" |
| "io" |
| "os" |
| @@ -159,10 +172,12 @@ func decode(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) ( |
| return dec.Header(), dec.Trailer() |
| } |
| -// anchorL0 re-encodes a snapshot [1,N] as a level-0 file [N,N] with every |
| -// page: the file litestream's position initialization fetches, so that |
| -// replication continues from N (R5-1, the remedy under test). |
| -func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, to litestream.ReplicaClient) { |
| +// encodeAnchor re-encodes a snapshot [1,N] as a level-0 file [N,N] with |
| +// every page: the file litestream's position initialization fetches, so |
| +// that replication continues from N (R5-1, the remedy under test). The |
| +// header timestamp is an input (R6-5): with the same inputs the bytes are |
| +// identical, which is what lets a replayed PUT repeat the same object. |
| +func encodeAnchor(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, ts int64) []byte { |
| t.Helper() |
| ctx := context.Background() |
| rc, err := from.OpenLTXFile(ctx, snap.Level, snap.MinTXID, snap.MaxTXID, 0, 0) |
| @@ -192,7 +207,7 @@ func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, t |
| Commit: hdr.Commit, |
| MinTXID: snap.MaxTXID, |
| MaxTXID: snap.MaxTXID, |
| - Timestamp: time.Now().UnixMilli(), |
| + Timestamp: ts, |
| WALOffset: litestream.WALHeaderSize, |
| }); err != nil { |
| t.Fatal(err) |
| @@ -215,11 +230,34 @@ func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, t |
| if err := enc.Close(); err != nil { |
| t.Fatal(err) |
| } |
| - if _, err := to.WriteLTXFile(ctx, 0, snap.MaxTXID, snap.MaxTXID, &buf); err != nil { |
| + return buf.Bytes() |
| +} |
| + |
| +// anchorL0 writes the anchor for snap into to. |
| +func anchorL0(t *testing.T, from litestream.ReplicaClient, snap *ltx.FileInfo, to litestream.ReplicaClient) { |
| + t.Helper() |
| + b := encodeAnchor(t, from, snap, time.Now().UnixMilli()) |
| + if _, err := to.WriteLTXFile(context.Background(), 0, snap.MaxTXID, snap.MaxTXID, bytes.NewReader(b)); err != nil { |
| t.Fatal(err) |
| } |
| } |
| +// mustRanges fails unless the level's current files are exactly want. |
| +func mustRanges(t *testing.T, client litestream.ReplicaClient, level int, want string) { |
| + t.Helper() |
| + if got := ranges(files(t, client, level)); got != want { |
| + t.Fatalf("level %d: have %s, want exactly %s", level, got, want) |
| + } |
| +} |
| + |
| +// mustCount fails unless the database at path holds exactly want rows. |
| +func mustCount(t *testing.T, path string, want int) { |
| + t.Helper() |
| + if n := count(t, path); n != want { |
| + t.Fatalf("%s: %d rows, want exactly %d", filepath.Base(path), n, want) |
| + } |
| +} |
| + |
| // source builds a replicated source database with n rows and a level-9 |
| // snapshot of it, and returns the client and the snapshot's info. |
| func source(t *testing.T, n int) (litestream.ReplicaClient, *ltx.FileInfo) { |
| @@ -314,21 +352,17 @@ func TestFact_R5_1_AnchorL0ContinuesFromN(t *testing.T) { |
| sync(t, db) |
| closeDB(t, db) |
| l0 := files(t, gen, 0) |
| - t.Logf("FACT R5-1c: with a level-0 anchor [%d,%d] beside the snapshot, level 0 after two writes is %s", snap.MaxTXID, snap.MaxTXID, ranges(l0)) |
| - last := l0[len(l0)-1] |
| - if last.MaxTXID <= snap.MaxTXID { |
| - t.Fatalf("expected replication to continue past %d, got %s", snap.MaxTXID, ranges(l0)) |
| - } |
| + t.Logf("FACT R5-1c: with a level-0 anchor [%d,%d] beside the snapshot, level 0 after two writes and one sync is %s", snap.MaxTXID, snap.MaxTXID, ranges(l0)) |
| + // exactly: the anchor, then the next TXID N+1 and nothing else (R6-2) |
| + n := snap.MaxTXID |
| + mustRanges(t, gen, 0, fmt.Sprintf("[%d,%d] [%d,%d]", n, n, n+1, n+1)) |
| wipe(t, out, meta) |
| out2 := filepath.Join(dir, "out2.db") |
| if err := restore(t, gen, out2); err != nil { |
| t.Fatal(err) |
| } |
| - n := count(t, out2) |
| - t.Logf("FACT R5-1d: a cold restore after that holds %d rows of 7", n) |
| - if n != 7 { |
| - t.Fatalf("expected the new rows restored (7), got %d", n) |
| - } |
| + t.Logf("FACT R5-1d: a cold restore after that holds %d rows of 7", count(t, out2)) |
| + mustCount(t, out2, 7) |
| } |
| // TestFact_R5_1_TXIDSidecarIsFollowOnly: a -txid sidecar beside the |
| @@ -357,12 +391,13 @@ func TestFact_R5_1_TXIDSidecarIsFollowOnly(t *testing.T) { |
| } |
| } |
| -// TestFact_R5_1_L0ExpiryAndAFreshBox: what a fresh box's position is after |
| +// TestFact_R5_1_L0ExpiryAndAFreshBox: a fresh box's position after |
| // litestream's own level-0 expiry. TXID-based retention keeps the newest |
| -// level-0 file; time-based retention (L0Retention, the store's monitor) |
| -// is the path that could remove it. The test runs the time-based path and |
| -// records what the next upload after a restore carries, and whether a |
| -// cold restore afterwards keeps the new writes. |
| +// level-0 file, and so does time-based retention (L0Retention, the store's |
| +// monitor); a fresh box restored from the replica therefore continues at |
| +// N+1 and a cold restore afterwards holds every row. Each of those is |
| +// asserted exactly (R6-2): a library that removed the newest level-0 file, |
| +// restarted numbering or lost the new writes fails here. |
| func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) { |
| dir := t.TempDir() |
| path := filepath.Join(dir, "src.db") |
| @@ -384,7 +419,12 @@ func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) { |
| if err := db.EnforceRetentionByTXID(context.Background(), 0, snap.MaxTXID+1); err != nil { |
| t.Fatal(err) |
| } |
| + n := snap.MaxTXID // the newest level-0 file's TXID, the position |
| + newest := fmt.Sprintf("[%d,%d]", n, n) |
| byTXID := ranges(files(t, client, 0)) |
| + if byTXID != newest { |
| + t.Fatalf("TXID-based retention at %d left level 0 %s, want exactly the newest file %s retained", n+1, byTXID, newest) |
| + } |
| // time-based level-0 expiry, as the store's L0 retention monitor runs it |
| db.L0Retention = time.Millisecond |
| time.Sleep(20 * time.Millisecond) |
| @@ -392,29 +432,27 @@ func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) { |
| t.Fatal(err) |
| } |
| closeDB(t, db) |
| - t.Logf("FACT R5-1f: after TXID-based level-0 retention at %d the level-0 listing is %s; after time-based L0 expiry it is %s (L9 %s; L1 %s)", snap.MaxTXID+1, byTXID, ranges(files(t, client, 0)), ranges(files(t, client, 9)), ranges(files(t, client, 1))) |
| + t.Logf("FACT R5-1f: after TXID-based level-0 retention at %d the level-0 listing is %s; after time-based L0 expiry it is %s (L9 %s; L1 %s)", n+1, byTXID, ranges(files(t, client, 0)), ranges(files(t, client, 9)), ranges(files(t, client, 1))) |
| + mustRanges(t, client, 0, newest) // the newest level-0 file survives time-based expiry too, by identity |
| out := filepath.Join(dir, "out.db") |
| if err := restore(t, client, out); err != nil { |
| t.Fatal(err) |
| } |
| + mustCount(t, out, 7) |
| meta := filepath.Join(dir, "meta2") |
| db2 := openDB(t, out, meta, client) |
| write(t, out, 2) |
| sync(t, db2) |
| closeDB(t, db2) |
| - l0 := files(t, client, 0) |
| - t.Logf("FACT R5-1g: on a fresh box restored at TXID %d, the first upload is level 0 %s", snap.MaxTXID, ranges(l0)) |
| + t.Logf("FACT R5-1g: on a fresh box restored at TXID %d, the first upload is level 0 %s", n, ranges(files(t, client, 0))) |
| + mustRanges(t, client, 0, fmt.Sprintf("%s [%d,%d]", newest, n+1, n+1)) // exactly N+1: a restart at 1 fails here |
| wipe(t, out, meta) |
| out2 := filepath.Join(dir, "out2.db") |
| if err := restore(t, client, out2); err != nil { |
| t.Fatal(err) |
| } |
| - n := count(t, out2) |
| - t.Logf("FACT R5-1h: a cold restore after that holds %d rows of 9", n) |
| - restarted := len(l0) > 0 && l0[0].MinTXID == 1 |
| - if restarted && n != 7 || !restarted && n != 9 { |
| - t.Fatalf("inconsistent: restarted=%v rows=%d", restarted, n) |
| - } |
| + t.Logf("FACT R5-1h: a cold restore after that holds %d rows of 9", count(t, out2)) |
| + mustCount(t, out2, 9) // every row: a run that kept 7 of 9 fails here |
| } |
| // TestFact_R5_2_Checksums: what the pinned writers put in LTX headers and |
| @@ -524,7 +562,10 @@ func TestFact_R5_3_SnapshotRetention(t *testing.T) { |
| } |
| return false |
| } |
| - t.Logf("FACT R5-3c: after one retention pass a continuation from the original snapshot [1,%d] remains: %v (the level-1 file that starts at 1 is above minSnapshotTXID)", snap1.MaxTXID, continuation()) |
| + t.Logf("FACT R5-3c: after one retention pass a continuation from the original snapshot [1,%d] remains: %v (the level-1 file that starts at 1 ends at minSnapshotTXID and deletion is strictly below)", snap1.MaxTXID, continuation()) |
| + if !continuation() { |
| + t.Fatalf("expected the level-1 file starting at 1 to survive the first pass (it ends at minSnapshotTXID %d)", minSnap) |
| + } |
| // one more rotation: the level-1 file starting at 1 now ends below the |
| // new minSnapshotTXID and is expired |
| for i := 0; i < 3; i++ { |
| @@ -550,6 +591,10 @@ func TestFact_R5_3_SnapshotRetention(t *testing.T) { |
| after2 := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| cont := continuation() |
| t.Logf("FACT R5-3f: after a second rotation and retention (minSnapshotTXID %d): %s; a continuation from the original snapshot remains: %v", minSnap, after2, cont) |
| + if cont { |
| + t.Fatalf("expected every continuation from the original snapshot [1,%d] to be expired after the second rotation", snap1.MaxTXID) |
| + } |
| + mustRanges(t, client, 9, fmt.Sprintf("[1,%d]", snap2.MaxTXID)) // only the newest snapshot |
| // restore rooted at the original snapshot: the original's bytes plus the |
| // current lower levels, without the newer snapshot |
| rooted := lsfile.NewReplicaClient(filepath.Join(dir, "rooted")) |
| @@ -564,20 +609,129 @@ func TestFact_R5_3_SnapshotRetention(t *testing.T) { |
| errRooted := restore(t, rooted, filepath.Join(dir, "rooted.db")) |
| t.Logf("FACT R5-3d: restore rooted at the original snapshot after retention: %v", errRooted) |
| errNewest := restore(t, client, filepath.Join(dir, "newest.db")) |
| - t.Logf("FACT R5-3e: restore from the newest snapshot: %v (rows %d)", errNewest, func() int { |
| - if errNewest != nil { |
| - return -1 |
| - } |
| - return count(t, filepath.Join(dir, "newest.db")) |
| - }()) |
| if errNewest != nil { |
| t.Fatal(errNewest) |
| } |
| - if cont { |
| - if errRooted != nil { |
| - t.Fatalf("a continuation remained but the rooted restore failed: %v", errRooted) |
| + t.Logf("FACT R5-3e: restore from the newest snapshot: ok (rows %d)", count(t, filepath.Join(dir, "newest.db"))) |
| + mustCount(t, filepath.Join(dir, "newest.db"), 19) // 1 row, then 3 rounds of 3 syncs of 2 rows |
| + // the rooted restore fails closed, with litestream's own error (R6-2) |
| + want := fmt.Sprintf("non-contiguous ltx files: have up to %016x but next file starts at %016x", uint64(snap1.MaxTXID), 5) |
| + if errRooted == nil || !strings.Contains(errRooted.Error(), want) { |
| + t.Fatalf("expected the rooted restore to fail closed with %q, got %v", want, errRooted) |
| + } |
| +} |
| + |
| +// TestFact_R6_5_AnchorReplayBytes: the anchor's bytes depend on the header |
| +// timestamp. Encoded twice with the same inputs the bytes are identical; |
| +// with another timestamp they differ, so a builder that replays the anchor |
| +// PUT (a restarted install) must persist the timestamp it used, or reuse |
| +// the object it already wrote (moves spec §5.1 step 2, §7.6 building). |
| +func TestFact_R6_5_AnchorReplayBytes(t *testing.T) { |
| + src, snap := source(t, 5) |
| + ts := time.Now().UnixMilli() |
| + a := encodeAnchor(t, src, snap, ts) |
| + b := encodeAnchor(t, src, snap, ts) |
| + c := encodeAnchor(t, src, snap, ts+1) |
| + t.Logf("FACT R6-5: anchor [%d,%d], %d bytes; same timestamp: identical=%v; timestamp+1 ms: identical=%v", snap.MaxTXID, snap.MaxTXID, len(a), bytes.Equal(a, b), bytes.Equal(a, c)) |
| + if !bytes.Equal(a, b) { |
| + t.Fatalf("the same inputs produced different anchor bytes") |
| + } |
| + if bytes.Equal(a, c) { |
| + t.Fatalf("a different timestamp produced identical bytes: the premise of persisting it is gone, review the contract") |
| + } |
| + // the replayed object is accepted by the decoder and equals the first |
| + dec := ltx.NewDecoder(bytes.NewReader(b)) |
| + if err := dec.Verify(); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// TestGate_AnchoredLineageThroughStore: one anchored lineage through the |
| +// production Store configuration (internal/lease Held.Replicate: a DB with |
| +// its own metadata directory, a replica client, litestream.NewStore with |
| +// the compaction ladder, the L0 retention monitor), with every interval |
| +// shortened from hours to fractions of a second so the monitors run inside |
| +// the test: repeated writes, level-1 to level-3 compaction, snapshot |
| +// rotation and retention, time-based level-0 expiry, then disk loss and a |
| +// cold restore, three times over. Exact positions (every sync advances the |
| +// position by one, and a reopened box continues at exactly the newest |
| +// remote level-0 TXID plus one) and exact row counts are asserted. |
| +func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| + src, snap := source(t, 5) |
| + dir := t.TempDir() |
| + gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| + copyFile(t, src, gen, snap) |
| + anchorL0(t, src, snap, gen) |
| + path := filepath.Join(dir, "db") |
| + if err := restore(t, gen, path); err != nil { |
| + t.Fatal(err) |
| + } |
| + rows := 5 |
| + pos := int(snap.MaxTXID) |
| + mustCount(t, path, rows) |
| + // internal/lease's ladder is L1 1h, L2 6h, L3 24h, the snapshot interval |
| + // 24h, L0 retention 5 min checked hourly, the DB monitor 10 s: here each |
| + // is shortened by the same shape |
| + levels := litestream.CompactionLevels{ |
| + {Level: 0, Interval: 0}, |
| + {Level: 1, Interval: 150 * time.Millisecond}, |
| + {Level: 2, Interval: 400 * time.Millisecond}, |
| + {Level: 3, Interval: 900 * time.Millisecond}, |
| + } |
| + ctx := context.Background() |
| + for round := 0; round < 3; round++ { |
| + meta := filepath.Join(dir, fmt.Sprintf("meta%d", round)) |
| + db := litestream.NewDB(path) |
| + db.SetMetaPath(meta) |
| + db.MonitorInterval = 50 * time.Millisecond |
| + db.Replica = litestream.NewReplicaWithClient(db, gen) |
| + db.Replica.SyncInterval = 50 * time.Millisecond |
| + store := litestream.NewStore([]*litestream.DB{db}, levels) |
| + store.SnapshotInterval = 300 * time.Millisecond |
| + store.SnapshotRetention = 400 * time.Millisecond |
| + store.SetL0Retention(200 * time.Millisecond) // NewStore copied the default into the DB; SetL0Retention propagates |
| + store.L0RetentionCheckInterval = 50 * time.Millisecond |
| + if err := store.Open(ctx); err != nil { |
| + t.Fatalf("round %d: open store: %v", round, err) |
| + } |
| + l0Before := len(files(t, gen, 0)) |
| + for i := 0; i < 8; i++ { |
| + write(t, path, 1) |
| + rows++ |
| + sync(t, db) |
| + pos++ |
| + p, err := db.Pos() |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if int(p.TXID) != pos { |
| + t.Fatalf("round %d write %d: position %d, want exactly %d (a reopened box continues at the newest remote level-0 TXID plus one)", round, i, p.TXID, pos) |
| + } |
| + time.Sleep(120 * time.Millisecond) // the monitors run between writes |
| + } |
| + time.Sleep(700 * time.Millisecond) // compaction, rotation, retention, L0 expiry catch up |
| + if err := store.Close(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9) |
| + t.Logf("GATE round %d: position %d, rows %d; L0 %s (%d files, %d before); L1 %s; L2 %s; L3 %s; L9 %s", round, pos, rows, ranges(l0), len(l0), l0Before, ranges(l1), ranges(files(t, gen, 2)), ranges(files(t, gen, 3)), ranges(l9)) |
| + if len(l0) == 0 || int(l0[len(l0)-1].MaxTXID) != pos { |
| + t.Fatalf("round %d: newest remote level-0 file is %s, want MaxTXID exactly %d", round, ranges(l0), pos) |
| + } |
| + if len(l0) >= l0Before+8 { |
| + t.Fatalf("round %d: time-based level-0 expiry removed nothing: %d files", round, len(l0)) |
| + } |
| + if len(l1) == 0 { |
| + t.Fatalf("round %d: no level-1 compaction ran", round) |
| + } |
| + if len(l9) == 0 || int(l9[len(l9)-1].MaxTXID) <= int(snap.MaxTXID) { |
| + t.Fatalf("round %d: no snapshot rotated past the installed one: L9 %s", round, ranges(l9)) |
| + } |
| + // disk loss, cold restore: exact contents |
| + wipe(t, path, meta) |
| + if err := restore(t, gen, path); err != nil { |
| + t.Fatalf("round %d: cold restore: %v", round, err) |
| } |
| - } else if errRooted == nil || !(strings.Contains(errRooted.Error(), "non-contiguous") || errors.Is(errRooted, litestream.ErrTxNotAvailable)) { |
| - t.Fatalf("expected the rooted restore to fail closed on the gap, got %v", errRooted) |
| + mustCount(t, path, rows) |
| } |
| } |