amadan
sim: litestreamfacts: the pinned library's replication, checksum and retention facts (R5-1 to R5-3)
Test-only, driving litestream v0.5.14 with the file replica client as internal/lease does. Each test asserts a fact the moves spec's restore contract rests on: a snapshot-only install restarts replication at TXID 1 and a cold restore loses the writes; a level-0 anchor [N,N] beside the snapshot makes the pinned library continue at N+1; the -txid sidecar is follow-mode only; retention keeps the newest level-0 file, so an ordinary replica needs no anchor; every pinned writer sets HeaderFlagNoChecksum and PostApplyChecksum is 0 (per-file CRC-64 only); snapshot retention expires the continuation from an old snapshot after two rotations, so a restore rooted there fails closed while one from the newest snapshot succeeds. Facts recorded in ~/.cache/nearest-region/litestream-facts.md.
Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
1 file changed,
+583
−0
internal/sim/litestreamfacts/facts_test.go+583 −0
diff --git a/internal/sim/litestreamfacts/facts_test.go b/internal/sim/litestreamfacts/facts_test.go| new file mode 100644 |
| index 000000000..d3dc027e3 |
| --- /dev/null |
| +++ b/internal/sim/litestreamfacts/facts_test.go |
| @@ -0,0 +1,583 @@ |
| +// 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. |
| +// |
| +// The observed facts are recorded in ~/.cache/nearest-region/litestream-facts.md. |
| +package litestreamfacts |
| + |
| +import ( |
| + "bytes" |
| + "context" |
| + "database/sql" |
| + "errors" |
| + "fmt" |
| + "io" |
| + "os" |
| + "path/filepath" |
| + "strings" |
| + "testing" |
| + "time" |
| + |
| + "github.com/benbjohnson/litestream" |
| + lsfile "github.com/benbjohnson/litestream/file" |
| + "github.com/superfly/ltx" |
| + _ "modernc.org/sqlite" |
| +) |
| + |
| +// openDB opens path under litestream against client, with its own metadata |
| +// directory, no background monitors (every sync is explicit), and returns |
| +// the DB. The caller closes it. |
| +func openDB(t *testing.T, path, meta string, client litestream.ReplicaClient) *litestream.DB { |
| + t.Helper() |
| + db := litestream.NewDB(path) |
| + db.SetMetaPath(meta) |
| + db.MonitorInterval = time.Hour |
| + db.Replica = litestream.NewReplicaWithClient(db, client) |
| + db.Replica.SyncInterval = time.Hour |
| + if err := db.Open(); err != nil { |
| + t.Fatalf("open %s: %v", path, err) |
| + } |
| + return db |
| +} |
| + |
| +func closeDB(t *testing.T, db *litestream.DB) { |
| + t.Helper() |
| + if err := db.Close(context.Background()); err != nil { |
| + t.Fatalf("close: %v", err) |
| + } |
| +} |
| + |
| +// write inserts n rows, each in its own transaction, so each is a |
| +// transaction litestream can see. |
| +func write(t *testing.T, path string, n int) { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + if _, err := d.Exec(`CREATE TABLE IF NOT EXISTS t (id INTEGER PRIMARY KEY, body TEXT)`); err != nil { |
| + t.Fatal(err) |
| + } |
| + for i := 0; i < n; i++ { |
| + if _, err := d.Exec(`INSERT INTO t (body) VALUES (?)`, strings.Repeat("x", 100)); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| +} |
| + |
| +func count(t *testing.T, path string) int { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + var n int |
| + if err := d.QueryRow(`SELECT count(*) FROM t`).Scan(&n); err != nil { |
| + t.Fatal(err) |
| + } |
| + return n |
| +} |
| + |
| +func sync(t *testing.T, db *litestream.DB) { |
| + t.Helper() |
| + if err := db.SyncAndWait(context.Background()); err != nil { |
| + t.Fatalf("sync: %v", err) |
| + } |
| +} |
| + |
| +// files lists one level of a replica, in litestream's order. |
| +func files(t *testing.T, client litestream.ReplicaClient, level int) []*ltx.FileInfo { |
| + t.Helper() |
| + itr, err := client.LTXFiles(context.Background(), level, 0, false) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + infos, err := ltx.SliceFileIterator(itr) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + return infos |
| +} |
| + |
| +func ranges(infos []*ltx.FileInfo) string { |
| + var out []string |
| + for _, f := range infos { |
| + out = append(out, fmt.Sprintf("[%d,%d]", f.MinTXID, f.MaxTXID)) |
| + } |
| + if len(out) == 0 { |
| + return "none" |
| + } |
| + return strings.Join(out, " ") |
| +} |
| + |
| +// copyFile copies one LTX file between replica clients, as the install's |
| +// CopyObject would. |
| +func copyFile(t *testing.T, from, to litestream.ReplicaClient, info *ltx.FileInfo) { |
| + t.Helper() |
| + ctx := context.Background() |
| + rc, err := from.OpenLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rc.Close() |
| + if _, err := to.WriteLTXFile(ctx, info.Level, info.MinTXID, info.MaxTXID, rc); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// restore restores client's newest state to out, litestream's own planner. |
| +func restore(t *testing.T, client litestream.ReplicaClient, out string) error { |
| + t.Helper() |
| + r := litestream.NewReplicaWithClient(nil, client) |
| + opt := litestream.NewRestoreOptions() |
| + opt.OutputPath = out |
| + return r.Restore(context.Background(), opt) |
| +} |
| + |
| +// decode reads an LTX file's header and trailer and verifies it. |
| +func decode(t *testing.T, client litestream.ReplicaClient, info *ltx.FileInfo) (ltx.Header, ltx.Trailer) { |
| + t.Helper() |
| + rc, err := client.OpenLTXFile(context.Background(), info.Level, info.MinTXID, info.MaxTXID, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rc.Close() |
| + b, err := io.ReadAll(rc) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + dec := ltx.NewDecoder(bytes.NewReader(b)) |
| + if err := dec.Verify(); err != nil { |
| + t.Fatalf("verify %s: %v", ltx.FormatFilename(info.MinTXID, info.MaxTXID), err) |
| + } |
| + 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) { |
| + t.Helper() |
| + ctx := context.Background() |
| + rc, err := from.OpenLTXFile(ctx, snap.Level, snap.MinTXID, snap.MaxTXID, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rc.Close() |
| + dec := ltx.NewDecoder(rc) |
| + if err := dec.DecodeHeader(); err != nil { |
| + t.Fatal(err) |
| + } |
| + hdr := dec.Header() |
| + var buf bytes.Buffer |
| + enc, err := ltx.NewEncoder(&buf) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + // WALOffset at the WAL header: litestream's verify reads the position |
| + // from the newest local level-0 file and, finding the WAL does not |
| + // match it, takes a fresh snapshot at the next TXID; an offset below |
| + // the header size is refused outright ("prev WAL offset is less than |
| + // the header size", observed) |
| + if err := enc.EncodeHeader(ltx.Header{ |
| + Version: ltx.Version, |
| + Flags: ltx.HeaderFlagNoChecksum, |
| + PageSize: hdr.PageSize, |
| + Commit: hdr.Commit, |
| + MinTXID: snap.MaxTXID, |
| + MaxTXID: snap.MaxTXID, |
| + Timestamp: time.Now().UnixMilli(), |
| + WALOffset: litestream.WALHeaderSize, |
| + }); err != nil { |
| + t.Fatal(err) |
| + } |
| + page := make([]byte, hdr.PageSize) |
| + for { |
| + var ph ltx.PageHeader |
| + if err := dec.DecodePage(&ph, page); err == io.EOF { |
| + break |
| + } else if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := enc.EncodePage(ph, page); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + if err := dec.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := enc.Close(); err != nil { |
| + t.Fatal(err) |
| + } |
| + if _, err := to.WriteLTXFile(ctx, 0, snap.MaxTXID, snap.MaxTXID, &buf); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// 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) { |
| + t.Helper() |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "src.db") |
| + write(t, path, 1) |
| + client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| + db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| + sync(t, db) |
| + write(t, path, n-1) |
| + sync(t, db) |
| + snap, err := db.Snapshot(context.Background()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + closeDB(t, db) |
| + t.Logf("source: L0 %s; snapshot [%d,%d]", ranges(files(t, client, 0)), snap.MinTXID, snap.MaxTXID) |
| + return client, snap |
| +} |
| + |
| +// wipe removes the local database, its WAL and SHM, and litestream's |
| +// metadata: a lost disk. |
| +func wipe(t *testing.T, path, meta string) { |
| + t.Helper() |
| + for _, p := range []string{path, path + "-wal", path + "-shm", path + "-journal"} { |
| + os.Remove(p) |
| + } |
| + if err := os.RemoveAll(meta); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne: install only the level-9 |
| +// snapshot [1,N], restore it, start replication with fresh metadata and no |
| +// remote level 0. The next upload carries TXID 1, not N+1, and a cold |
| +// restore afterwards loses the new writes. |
| +func TestFact_R5_1_SnapshotOnlyInstallRestartsAtOne(t *testing.T) { |
| + src, snap := source(t, 5) |
| + dir := t.TempDir() |
| + gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| + copyFile(t, src, gen, snap) |
| + out := filepath.Join(dir, "out.db") |
| + if err := restore(t, gen, out); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := count(t, out); n != 5 { |
| + t.Fatalf("restored %d rows, want 5", n) |
| + } |
| + meta := filepath.Join(dir, "meta") |
| + db := openDB(t, out, meta, gen) |
| + write(t, out, 2) |
| + sync(t, db) |
| + closeDB(t, db) |
| + l0 := files(t, gen, 0) |
| + t.Logf("FACT R5-1a: after a snapshot-only install (snapshot [1,%d]) the first upload is level 0 %s", snap.MaxTXID, ranges(l0)) |
| + if len(l0) == 0 || l0[0].MinTXID != 1 { |
| + t.Fatalf("expected replication to restart at TXID 1, got %s", ranges(l0)) |
| + } |
| + 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-1b: a cold restore after that holds %d rows of 7: the planner roots at the snapshot and skips level-0 files below its TXID", n) |
| + if n != 5 { |
| + t.Fatalf("expected the two new rows to be lost (5 rows), got %d", n) |
| + } |
| +} |
| + |
| +// TestFact_R5_1_AnchorL0ContinuesFromN: the remedy. The install also writes |
| +// a level-0 file [N,N] with every page. Litestream's open fetches the |
| +// newest remote level-0 file as its local position, so the next upload |
| +// carries N+1 and a cold restore has the new writes. |
| +func TestFact_R5_1_AnchorL0ContinuesFromN(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) |
| + out := filepath.Join(dir, "out.db") |
| + if err := restore(t, gen, out); err != nil { |
| + t.Fatal(err) |
| + } |
| + if n := count(t, out); n != 5 { |
| + t.Fatalf("restored %d rows, want 5", n) |
| + } |
| + meta := filepath.Join(dir, "meta") |
| + db := openDB(t, out, meta, gen) |
| + write(t, out, 2) |
| + 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)) |
| + } |
| + 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) |
| + } |
| +} |
| + |
| +// TestFact_R5_1_TXIDSidecarIsFollowOnly: a -txid sidecar beside the |
| +// restored database does not set the replication position of an |
| +// ordinary open. |
| +func TestFact_R5_1_TXIDSidecarIsFollowOnly(t *testing.T) { |
| + src, snap := source(t, 5) |
| + dir := t.TempDir() |
| + gen := lsfile.NewReplicaClient(filepath.Join(dir, "gen")) |
| + copyFile(t, src, gen, snap) |
| + out := filepath.Join(dir, "out.db") |
| + if err := restore(t, gen, out); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := litestream.WriteTXIDFile(out, snap.MaxTXID); err != nil { |
| + t.Fatal(err) |
| + } |
| + db := openDB(t, out, filepath.Join(dir, "meta"), gen) |
| + write(t, out, 1) |
| + sync(t, db) |
| + closeDB(t, db) |
| + l0 := files(t, gen, 0) |
| + t.Logf("FACT R5-1e: with a -txid sidecar of %d, the first upload is level 0 %s", snap.MaxTXID, ranges(l0)) |
| + if len(l0) == 0 || l0[0].MinTXID != 1 { |
| + t.Fatalf("the sidecar unexpectedly set the position: %s", ranges(l0)) |
| + } |
| +} |
| + |
| +// TestFact_R5_1_L0ExpiryAndAFreshBox: what a fresh box's position is 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. |
| +func TestFact_R5_1_L0ExpiryAndAFreshBox(t *testing.T) { |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "src.db") |
| + write(t, path, 1) |
| + client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| + db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| + for i := 0; i < 3; i++ { |
| + write(t, path, 2) |
| + sync(t, db) |
| + } |
| + if _, err := db.Compact(context.Background(), 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + snap, err := db.Snapshot(context.Background()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + // TXID-based retention at the snapshot level |
| + if err := db.EnforceRetentionByTXID(context.Background(), 0, snap.MaxTXID+1); err != nil { |
| + t.Fatal(err) |
| + } |
| + byTXID := ranges(files(t, client, 0)) |
| + // time-based level-0 expiry, as the store's L0 retention monitor runs it |
| + db.L0Retention = time.Millisecond |
| + time.Sleep(20 * time.Millisecond) |
| + if err := db.EnforceL0RetentionByTime(context.Background()); err != nil { |
| + 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))) |
| + out := filepath.Join(dir, "out.db") |
| + if err := restore(t, client, out); err != nil { |
| + t.Fatal(err) |
| + } |
| + 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)) |
| + 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) |
| + } |
| +} |
| + |
| +// TestFact_R5_2_Checksums: what the pinned writers put in LTX headers and |
| +// trailers: the level-0 writer, the snapshot writer and the compactor. |
| +func TestFact_R5_2_Checksums(t *testing.T) { |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "src.db") |
| + write(t, path, 1) |
| + client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| + db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| + for i := 0; i < 3; i++ { |
| + write(t, path, 2) |
| + sync(t, db) |
| + } |
| + snap, err := db.Snapshot(context.Background()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + l1, err := db.Compact(context.Background(), 1) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + closeDB(t, db) |
| + check := func(name string, info *ltx.FileInfo) { |
| + hdr, tr := decode(t, client, info) |
| + t.Logf("FACT R5-2 %s [%d,%d]: HeaderFlagNoChecksum=%v PreApplyChecksum=%#x PostApplyChecksum=%#x FileChecksum!=0=%v (Verify passed: CRC-64 of the file)", |
| + name, info.MinTXID, info.MaxTXID, hdr.NoChecksum(), uint64(hdr.PreApplyChecksum), uint64(tr.PostApplyChecksum), tr.FileChecksum != 0) |
| + if !hdr.NoChecksum() || tr.PostApplyChecksum != 0 || tr.FileChecksum == 0 { |
| + t.Fatalf("%s: unexpected checksum fields", name) |
| + } |
| + } |
| + l0 := files(t, client, 0) |
| + check("level 0", l0[len(l0)-1]) |
| + check("snapshot", snap) |
| + check("level 1 (compactor)", l1) |
| +} |
| + |
| +// TestFact_R5_3_SnapshotRetention: newer snapshots appear, and snapshot |
| +// retention expires older snapshots and the lower-level ranges below the |
| +// oldest remaining one. Afterwards no current file continues from the |
| +// original snapshot's TXID, so a restore rooted there finds a gap, while |
| +// litestream's own restore from the newest snapshot succeeds. |
| +func TestFact_R5_3_SnapshotRetention(t *testing.T) { |
| + dir := t.TempDir() |
| + path := filepath.Join(dir, "src.db") |
| + write(t, path, 1) |
| + client := lsfile.NewReplicaClient(filepath.Join(dir, "replica")) |
| + db := openDB(t, path, filepath.Join(dir, "meta"), client) |
| + sync(t, db) |
| + snap1, err := db.Snapshot(context.Background()) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + // keep the original snapshot's bytes: on S3 it would be a pinned, |
| + // noncurrent version; the file client deletes outright |
| + rc, err := client.OpenLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, 0, 0) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + snap1Bytes, _ := io.ReadAll(rc) |
| + rc.Close() |
| + // two more rotations: writes, a level-1 compaction, a snapshot, as the |
| + // store's monitors would over days |
| + var snap2 *ltx.FileInfo |
| + for round := 0; round < 2; round++ { |
| + for i := 0; i < 3; i++ { |
| + write(t, path, 2) |
| + sync(t, db) |
| + } |
| + if _, err := db.Compact(context.Background(), 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + if snap2, err = db.Snapshot(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + before := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| + // snapshot retention keeps the newest snapshot; minSnapshotTXID is the |
| + // MaxTXID of the snapshot before the oldest kept one, and the store |
| + // then expires levels 1 and up below it (level 0 expires by time, |
| + // L0Retention, 5 minutes by default: here it is enforced by TXID too) |
| + minSnap, err := db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second)) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := db.EnforceRetentionByTXID(context.Background(), 1, minSnap); err != nil { |
| + t.Fatal(err) |
| + } |
| + if err := db.EnforceRetentionByTXID(context.Background(), 0, minSnap); err != nil { |
| + t.Fatal(err) |
| + } |
| + after := fmt.Sprintf("L9 %s; L1 %s; L0 %s", ranges(files(t, client, 9)), ranges(files(t, client, 1)), ranges(files(t, client, 0))) |
| + t.Logf("FACT R5-3a: before retention: %s", before) |
| + t.Logf("FACT R5-3b: after snapshot retention (minSnapshotTXID %d): %s", minSnap, after) |
| + if len(files(t, client, 9)) != 1 || files(t, client, 9)[0].MaxTXID != snap2.MaxTXID { |
| + t.Fatalf("expected only the newest snapshot to remain") |
| + } |
| + continuation := func() bool { |
| + // any current file with MinTXID <= snap1.MaxTXID+1 and MaxTXID > snap1.MaxTXID |
| + for _, lvl := range []int{0, 1, 2, 3} { |
| + for _, f := range files(t, client, lvl) { |
| + if f.MinTXID <= snap1.MaxTXID+1 && f.MaxTXID > snap1.MaxTXID { |
| + return true |
| + } |
| + } |
| + } |
| + 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()) |
| + // 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++ { |
| + write(t, path, 2) |
| + sync(t, db) |
| + } |
| + if _, err := db.Compact(context.Background(), 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + if snap2, err = db.Snapshot(context.Background()); err != nil { |
| + t.Fatal(err) |
| + } |
| + if minSnap, err = db.EnforceSnapshotRetention(context.Background(), time.Now().Add(time.Second)); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, lvl := range []int{1, 0} { |
| + if err := db.EnforceRetentionByTXID(context.Background(), lvl, minSnap); err != nil { |
| + t.Fatal(err) |
| + } |
| + } |
| + closeDB(t, db) |
| + 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) |
| + // 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")) |
| + if _, err := rooted.WriteLTXFile(context.Background(), 9, snap1.MinTXID, snap1.MaxTXID, bytes.NewReader(snap1Bytes)); err != nil { |
| + t.Fatal(err) |
| + } |
| + for _, lvl := range []int{0, 1, 2, 3} { |
| + for _, f := range files(t, client, lvl) { |
| + copyFile(t, client, rooted, f) |
| + } |
| + } |
| + 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) |
| + } |
| + } 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) |
| + } |
| +} |