amadan
sim: litestreamfacts: the Store gate asserts its maintenance events and row contents; the unconditional PUT fact (R7-5, R7-1)
The anchored lineage through the Store now awaits each maintenance event as a condition (a snapshot rotation, level 1 to 3 compactions, snapshot retention leaving one snapshot, time-based level-0 expiry) instead of sleeping past it, writes rows with distinct bodies, an update and a delete per round, and compares the restored table row by row; a tail forces a cold restore that must consume a level-1 file beyond the newest snapshot. TestFact_R7_1_UnconditionalPut reads the pinned S3 client's source: its PutObjectInput carries no If-Match or If-None-Match and its key is prefix, level and TXID range only, the two facts the writer-epoch hazard rests on (claim rev 6.6 §6.4a). The package doc distinguishes S3's filename ordering from its LastModified ageing.
Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
1 file changed,
+203
−28
internal/sim/litestreamfacts/facts_test.go+203 −28
diff --git a/internal/sim/litestreamfacts/facts_test.go b/internal/sim/litestreamfacts/facts_test.go| index 86cac14bf..6ec2b3f40 100644 |
| --- a/internal/sim/litestreamfacts/facts_test.go |
| +++ b/internal/sim/litestreamfacts/facts_test.go |
| @@ -12,13 +12,18 @@ |
| // 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. |
| +// What the file client cannot show (R6 and R7 reviews): S3 lists LTX |
| +// files in filename order and ages them for retention 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, and an upload |
| +// outstanding across a writer replacement is the R7-1 hazard the model |
| +// (internal/sim/hostlife) encodes. 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 and §15 step 8's. |
| +// TestFact_R7_1_UnconditionalPut reads the pinned S3 client's source for |
| +// the two facts the hazard rests on: an unconditional PutObject, and a key |
| +// of prefix, level and TXID range only. |
| // |
| // The observed facts are recorded in ~/.cache/nearest-region/litestream-facts.md. |
| package litestreamfacts |
| @@ -28,6 +33,7 @@ import ( |
| "context" |
| "database/sql" |
| "fmt" |
| + "go/build" |
| "io" |
| "os" |
| "path/filepath" |
| @@ -83,6 +89,72 @@ func write(t *testing.T, path string, n int) { |
| } |
| } |
| +// rows reads the table's contents, id -> body, so a restore can be |
| +// compared row by row, updates and deletes included (R7-5). |
| +func rows(t *testing.T, path string) map[int]string { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + rs, err := d.Query(`SELECT id, body FROM t ORDER BY id`) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer rs.Close() |
| + out := map[int]string{} |
| + for rs.Next() { |
| + var id int |
| + var body string |
| + if err := rs.Scan(&id, &body); err != nil { |
| + t.Fatal(err) |
| + } |
| + out[id] = body |
| + } |
| + return out |
| +} |
| + |
| +// mustRows fails unless the database at path holds exactly want. |
| +func mustRows(t *testing.T, path string, want map[int]string) { |
| + t.Helper() |
| + got := rows(t, path) |
| + if len(got) != len(want) { |
| + t.Fatalf("%s: %d rows, want %d", filepath.Base(path), len(got), len(want)) |
| + } |
| + for id, body := range want { |
| + if got[id] != body { |
| + t.Fatalf("%s: row %d is %q, want %q", filepath.Base(path), id, got[id], body) |
| + } |
| + } |
| +} |
| + |
| +// exec runs one statement in its own transaction. |
| +func exec(t *testing.T, path, stmt string, args ...any) { |
| + t.Helper() |
| + d, err := sql.Open("sqlite", path) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + defer d.Close() |
| + if _, err := d.Exec(stmt, args...); err != nil { |
| + t.Fatal(err) |
| + } |
| +} |
| + |
| +// waitFor polls cond until it holds or the deadline passes (R7-5: a |
| +// completion condition, not a sleep). |
| +func waitFor(t *testing.T, what string, d time.Duration, cond func() bool) { |
| + t.Helper() |
| + deadline := time.Now().Add(d) |
| + for !cond() { |
| + if time.Now().After(deadline) { |
| + t.Fatalf("timed out waiting for %s", what) |
| + } |
| + time.Sleep(20 * time.Millisecond) |
| + } |
| +} |
| + |
| func count(t *testing.T, path string) int { |
| t.Helper() |
| d, err := sql.Open("sqlite", "file:"+path+"?mode=ro") |
| @@ -651,11 +723,14 @@ func TestFact_R6_5_AnchorReplayBytes(t *testing.T) { |
| // 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. |
| +// the test: repeated writes with distinct bodies, updates and deletes, |
| +// 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 contents are asserted; each maintenance event |
| +// is awaited as a condition, not assumed after a sleep; the last round |
| +// forces a restore that consumes compacted continuation files (R7-5). |
| func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| src, snap := source(t, 5) |
| dir := t.TempDir() |
| @@ -666,9 +741,12 @@ func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| if err := restore(t, gen, path); err != nil { |
| t.Fatal(err) |
| } |
| - rows := 5 |
| + want := rows(t, path) // the source's five rows, distinct bodies |
| + if len(want) != 5 { |
| + t.Fatalf("restored %d rows, want 5", len(want)) |
| + } |
| pos := int(snap.MaxTXID) |
| - mustCount(t, path, rows) |
| + next := 6 // the next row id to insert |
| // 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 |
| @@ -679,6 +757,13 @@ func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| {Level: 3, Interval: 900 * time.Millisecond}, |
| } |
| ctx := context.Background() |
| + maxOf := func(level int) int { |
| + fs := files(t, gen, level) |
| + if len(fs) == 0 { |
| + return 0 |
| + } |
| + return int(fs[len(fs)-1].MaxTXID) |
| + } |
| for round := 0; round < 3; round++ { |
| meta := filepath.Join(dir, fmt.Sprintf("meta%d", round)) |
| db := litestream.NewDB(path) |
| @@ -694,10 +779,25 @@ func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| if err := store.Open(ctx); err != nil { |
| t.Fatalf("round %d: open store: %v", round, err) |
| } |
| + snapsBefore := maxOf(9) |
| l0Before := len(files(t, gen, 0)) |
| + // eight transactions: inserts of distinct bodies, an update and a |
| + // delete among them, each advancing the position by exactly one |
| for i := 0; i < 8; i++ { |
| - write(t, path, 1) |
| - rows++ |
| + switch { |
| + case i == 3: |
| + id := next - 2 |
| + exec(t, path, `UPDATE t SET body = ? WHERE id = ?`, fmt.Sprintf("r%d-updated-in-round-%d", id, round), id) |
| + want[id] = fmt.Sprintf("r%d-updated-in-round-%d", id, round) |
| + case i == 6: |
| + id := next - 4 |
| + exec(t, path, `DELETE FROM t WHERE id = ?`, id) |
| + delete(want, id) |
| + default: |
| + exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-round-%d", next, round)) |
| + want[next] = fmt.Sprintf("r%d-round-%d", next, round) |
| + next++ |
| + } |
| sync(t, db) |
| pos++ |
| p, err := db.Pos() |
| @@ -709,29 +809,104 @@ func TestGate_AnchoredLineageThroughStore(t *testing.T) { |
| } |
| time.Sleep(120 * time.Millisecond) // the monitors run between writes |
| } |
| - time.Sleep(700 * time.Millisecond) // compaction, rotation, retention, L0 expiry catch up |
| + // the maintenance events, each as a condition: a snapshot rotated |
| + // past the previous newest, level 1, 2 and 3 files exist, the older |
| + // snapshot was removed by retention, and level-0 expiry removed |
| + // compacted files while keeping the newest |
| + waitFor(t, "a snapshot rotation", 5*time.Second, func() bool { return maxOf(9) > snapsBefore }) |
| + for _, lvl := range []int{1, 2, 3} { |
| + waitFor(t, fmt.Sprintf("a level-%d compaction", lvl), 5*time.Second, func() bool { return len(files(t, gen, lvl)) > 0 }) |
| + } |
| + waitFor(t, "snapshot retention", 5*time.Second, func() bool { return len(files(t, gen, 9)) == 1 }) |
| + waitFor(t, "level-0 expiry", 5*time.Second, func() bool { return len(files(t, gen, 0)) < l0Before+8 }) |
| 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)) |
| + 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, len(want), 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)) |
| + if len(l9) != 1 || int(l9[0].MaxTXID) <= snapsBefore { |
| + t.Fatalf("round %d: snapshot retention did not leave exactly the newest snapshot: L9 %s", round, ranges(l9)) |
| } |
| - // disk loss, cold restore: exact contents |
| + // disk loss, cold restore: exact contents, row by row |
| wipe(t, path, meta) |
| if err := restore(t, gen, path); err != nil { |
| t.Fatalf("round %d: cold restore: %v", round, err) |
| } |
| - mustCount(t, path, rows) |
| + mustRows(t, path, want) |
| + } |
| + // a restore that must consume compacted continuation files: with the |
| + // monitors stopped, more writes are synced, compacted into level 1, and |
| + // every level-0 file but the newest is expired by time; the newest |
| + // snapshot is behind them, so the planner has to take the level-1 file |
| + meta := filepath.Join(dir, "meta-tail") |
| + db := openDB(t, path, meta, gen) |
| + for i := 0; i < 3; i++ { |
| + exec(t, path, `INSERT INTO t (id, body) VALUES (?, ?)`, next, fmt.Sprintf("r%d-tail", next)) |
| + want[next] = fmt.Sprintf("r%d-tail", next) |
| + next++ |
| + sync(t, db) |
| + pos++ |
| + } |
| + exec(t, path, `DELETE FROM t WHERE id = ?`, next-1) |
| + delete(want, next-1) |
| + sync(t, db) |
| + pos++ |
| + if _, err := db.Compact(ctx, 1); err != nil { |
| + t.Fatal(err) |
| + } |
| + db.L0Retention = time.Millisecond |
| + time.Sleep(20 * time.Millisecond) |
| + if err := db.EnforceL0RetentionByTime(ctx); err != nil { |
| + t.Fatal(err) |
| + } |
| + closeDB(t, db) |
| + l0, l1, l9 := files(t, gen, 0), files(t, gen, 1), files(t, gen, 9) |
| + t.Logf("GATE tail: position %d; L0 %s; L1 %s; L9 %s", pos, ranges(l0), ranges(l1), ranges(l9)) |
| + if len(l0) != 1 || int(l0[0].MaxTXID) != pos { |
| + t.Fatalf("tail: level 0 should hold the newest file alone, have %s", ranges(l0)) |
| + } |
| + if maxOf(9) >= pos || maxOf(1) != pos { |
| + t.Fatalf("tail: the newest snapshot (%d) must be behind the level-1 file ending at %d", maxOf(9), maxOf(1)) |
| + } |
| + wipe(t, path, meta) |
| + if err := restore(t, gen, path); err != nil { |
| + t.Fatalf("tail: cold restore: %v", err) |
| + } |
| + mustRows(t, path, want) // the rows past the snapshot came through level 1 |
| +} |
| + |
| +// TestFact_R7_1_UnconditionalPut: the pinned S3 client's PutObject carries |
| +// no If-Match or If-None-Match, and its key is prefix, level and TXID range |
| +// only, so two writers of one generation write the same key with |
| +// unconditional PUTs: the R7-1 hazard (claim rev 6.6 §6.4a). Read from |
| +// the pinned source so an upgrade that changes either fails here. |
| +func TestFact_R7_1_UnconditionalPut(t *testing.T) { |
| + pkg, err := build.Default.Import("github.com/benbjohnson/litestream/s3", "", build.FindOnly) |
| + if err != nil { |
| + t.Skipf("pinned s3 client source not found: %v", err) |
| + } |
| + src, err := os.ReadFile(filepath.Join(pkg.Dir, "replica_client.go")) |
| + if err != nil { |
| + t.Fatal(err) |
| + } |
| + text := string(src) |
| + i := strings.Index(text, "s3.PutObjectInput{") |
| + if i < 0 { |
| + t.Fatalf("no PutObjectInput in the pinned client") |
| + } |
| + block := text[i:] |
| + block = block[:strings.Index(block, "}")] |
| + t.Logf("FACT R7-1a: the pinned PutObjectInput fields: %s", strings.Join(strings.Fields(strings.ReplaceAll(block, "\n", " ")), " ")) |
| + for _, cond := range []string{"IfNoneMatch", "IfMatch"} { |
| + if strings.Contains(block, cond) { |
| + t.Fatalf("the pinned client's PutObject is conditional (%s): the writer-epoch premise changed, review the contract", cond) |
| + } |
| + } |
| + if !strings.Contains(text, `fmt.Sprintf("%04x/%s", level, filename)`) || !strings.Contains(text, "ltx.FormatFilename(minTXID, maxTXID)") { |
| + t.Fatalf("the pinned client's key is no longer prefix/level/TXID range") |
| } |
| + t.Logf("FACT R7-1b: the key is c.Path + \"/\" + %%04x level + ltx.FormatFilename(min, max): two writers of one generation share it") |
| } |