From 908b3fda54ca6e8ba3889f1f2b68192421c8a71b Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 10:27:21 +0530 Subject: [PATCH 1/9] docs: design spec for incremental checkpoint index rebuild (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- ...026-09-26-incremental-checkpoint-design.md | 170 ++++++++++++++++++ 1 file changed, 170 insertions(+) create mode 100644 docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md diff --git a/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md b/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md new file mode 100644 index 000000000..a758b23f6 --- /dev/null +++ b/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md @@ -0,0 +1,170 @@ +# Incremental checkpoint index rebuild (#315) — Design + +**Issue:** #315 — `checkpoint()` is O(graph size) and flat in dirty bytes +**Milestone:** v2.3.0 (`main`, file format v7 — no format change) +**Follow-up:** true O(delta) checkpoints move to #374 on the `v3` branch (see §7) + +## 1. Problem + +Every `checkpoint()` streams all four covering indexes (EAVT/AEVT/AVET/VAET) off +disk, decodes every entry, merges in the pending entries, re-serialises everything +and bulk-builds four new B+trees. Cost scales with total graph size, not with the +number of dirty facts. Downstream (temporal_reasoning#241) this was ~half of +at-scale ingestion wall clock. The default `wal_checkpoint_threshold` (1000) makes +every default user pay `N_checkpoints × graph_size`. + +Reproduced locally (50k facts, 13 MB, checkpoint after **one** new fact, mean of 20): +**~80 ms**. + +## 2. Where the time goes + +Phase timing of `PersistentFactStorage::save` (tmpfs, same harness): + +| phase | ms | +|---|---:| +| merge + serialise + `build_btree` (Step D) | ~58 | +| `stream_all_entries` × 4 (Step A) | ~19 | +| full-file CRC re-read | ~3.5 | +| pack facts + fsyncs | <1 (a few ms on NVMe) | + +The cost is CPU in decode/re-encode of every index entry, not I/O. + +## 3. Why true O(delta) is out of scope here + +In format v7, fact pages must be contiguous at pages `1..=fact_page_count` +(`read_all_with_refs`, `CommittedFactLoaderImpl::first_fact_page`), and indexes live +directly after them. New fact pages therefore always overwrite the start of the old +index region, so every checkpoint must relocate the index. Avoiding that needs +copy-on-write B+trees plus free-space management — a format change, and the same +change #374 (save() is not crash-atomic) requires. That work belongs on `v3`. + +Dropping the unconditional `force_dirty()` (issue option 2) does not help: with one +dirty fact `dirty` is already true, and the zero-dirty case is already short-circuited +by the `wal_entry_count == 0 && !pfs.is_dirty()` guard. + +## 4. Design + +### 4.1 Leaf reuse — `src/storage/btree_v6.rs` + +`btree_v6.rs` is the B+tree for both v6 and v7 (v7 changed only the header). The +node layout is unchanged by this design, so v2.x readers are unaffected. + +New function: + +```rust +pub fn rebuild_btree_incremental( + old_leaves: Vec>>, // raw leaf pages of the old tree, in next_leaf order + pending: Vec<(K, FactRef)>, // sorted + backend: &mut dyn StorageBackend, + cache: &PageCache, + start_page_id: u64, +) -> Result<(u64, u64)> // (root_page_id, next_free_page_id) +where K: Serialize + for<'de> Deserialize<'de> + Ord +``` + +Algorithm: + +1. For each old leaf decode **only the first entry's key** → `first_key[i]`. +2. Route each pending entry to leaf `i` where `first_key[i] ≤ key < first_key[i+1]`; + keys below `first_key[0]` go to leaf 0. +3. Leaf with no pending entries → **raw copy**: write the 4 KB page verbatim to the + next page id, patching only bytes 4..12 (`next_leaf`). No decode/encode. +4. Leaf with pending entries → decode all its entries, merge with its pending + entries, serialise, and repack using `build_btree`'s 75% fill rule (may emit + more than one leaf). +5. Leaves are written in order with `next_leaf = pid + 1` (0 for the last), so there + is no second patch pass. +6. Collect `(page_id, first_key_bytes)` per new leaf (postcard of the first key) and + build internal levels bottom-up via a helper **extracted from `build_btree`'s + phase 2** (`build_internal_levels`), shared by both paths. + +`build_btree` remains the path for the first save, for an empty or missing old tree +(root 0 or a single empty leaf), and for load-time index rebuild / migrations. + +Collecting raw leaves: `collect_leaf_pages(root, backend, cache) -> Vec>>` +walks `find_leftmost_leaf` + the `next_leaf` chain, validating page type as +`stream_all_entries` does. + +Leaf fill: untouched leaves keep their existing fill; touched leaves are repacked at +75%. The tree shape differs from a fresh bulk build but is a valid B+tree for +`range_scan` / `stream_all_entries` / `find_leaf_for_key`. No merging of +under-filled leaves (YAGNI). + +### 4.2 `save()` ordering — `src/storage/persistent_facts.rs` + +Step A reads the raw leaf pages of all four old trees into memory **before** any +write (new fact pages overwrite the start of the old index region). This replaces +the decoded `Vec<(K, FactRef)>` streams; memory drops (raw pages ≤ decoded entries ++ serialised copies held today). + +Step D calls `rebuild_btree_incremental` per index when the old root is non-zero, +otherwise `build_btree` as today. + +Crash semantics are unchanged: the WAL is only deleted after the header write, and a +checksum mismatch on open rebuilds indexes from fact pages (#370 path). #374 remains +the real fix for atomicity. + +### 4.3 Checksum prefix cache + +`PersistentFactStorage` gains `fact_prefix_crc: Option` — CRC32 +state after hashing fact pages `1..=committed_fact_pages`. + +- **Set** in `load()`: the existing full-checksum pass is split into fact pages then + index pages; the hasher is cloned between the two segments. Only set when the + new-style full checksum matched. +- **Set** at the end of `save()`: prefix + the new fact pages (hashed from the + in-memory page buffers). +- **Cleared (None)** on any other path that rewrites the header/indexes (index + rebuild in `load()`, legacy/v5 migrations). +- **Use** in `save()`: clone the prefix, update with the new fact pages from memory, + then read back and hash only the index pages. `None` → today's full pass. + +The stored `index_checksum` value is identical to today's (CRC over pages +`1..page_count`), so older v2.x readers verify it unchanged. + +### 4.4 Docs and cleanup + +- `Minigraf::checkpoint` rustdoc: cost is O(graph size) page copies plus O(delta) + decode; checkpoint is **not** a durability boundary (the WAL is crash-durable); + batching trades only reopen latency (WAL replay). +- Same note on `OpenOptions::wal_checkpoint_threshold`. +- Trim the obsolete "same-process double-opens" rationale in `do_checkpoint`'s + guard comment (now refused outright since #304/#314). +- At release: `CHANGELOG.md`, `ROADMAP.md`, `docs/BENCHMARKS.md`, + `docs/TEST_COVERAGE.md`, `CLAUDE.md` test count. + +## 5. Testing + +No new dependencies; deterministic seeded loops. + +- **Equivalence (unit, `btree_v6.rs`):** for random committed sets and random + pending batches (including batches that force leaf splits, keys below the first + leaf, keys after the last leaf, and many pending entries in one leaf), + `stream_all_entries` of the incremental tree equals the merged sorted set. +- **Structure:** `next_leaf` chain is strictly ascending across leaves; `range_scan` + over random bounds matches a filter over the expected set. +- **Multi-checkpoint (`persistent_facts.rs`):** many save cycles on a file backend; + after each, all four indexes stream equal to a from-scratch build over all facts. +- **Checksum:** after incremental saves `header.index_checksum == + compute_page_checksum(1, page_count-1)`; reopening does not take the rebuild path. + Prefix cache is `None` after a rebuild/migration path and a subsequent save still + writes a correct checksum. +- **Regression:** full existing suite. +- **Benchmark:** new `checkpoint/after_1_fact` group (10k / 100k facts) in + `benches/minigraf_bench.rs`; before/after numbers in `docs/BENCHMARKS.md`. + +Test assert messages follow the CodeQL rule (no `{:?}` of UUID-bearing types). + +## 6. Success criteria + +- 50k facts, checkpoint after 1 fact: ~80 ms → **≤ 15 ms** on the §1 harness. +- No file-format or public-API change; files written by the new code open without + index rebuild on v2.0.1. +- All existing tests pass. + +## 7. Follow-up + +Comment on #374 (v3.0.0): copy-on-write B+tree pages with a free list make save() +crash-atomic **and** give O(delta · log n) checkpoints; #315's remaining O(size) +page-copy term is owned there. #315 is closed by this work as mitigated, not fully +resolved (no closing keyword on #374). From 0dbfbee9320eabeaf763a074949571b5d16419c7 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 11:39:35 +0530 Subject: [PATCH 2/9] docs: implementation plan for incremental checkpoint (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- .../2026-09-26-incremental-checkpoint.md | 1076 +++++++++++++++++ 1 file changed, 1076 insertions(+) create mode 100644 docs/superpowers/plans/2026-09-26-incremental-checkpoint.md diff --git a/docs/superpowers/plans/2026-09-26-incremental-checkpoint.md b/docs/superpowers/plans/2026-09-26-incremental-checkpoint.md new file mode 100644 index 000000000..88eca25c6 --- /dev/null +++ b/docs/superpowers/plans/2026-09-26-incremental-checkpoint.md @@ -0,0 +1,1076 @@ +# Incremental Checkpoint Index Rebuild Implementation Plan + +> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. + +**Goal:** Make `checkpoint()` decode/re-encode only the index leaves the delta touches, and stop re-reading unchanged fact pages for the file checksum, without changing file format v7. + +**Architecture:** `save()` snapshots each old B+tree's raw leaf pages, then `rebuild_btree_incremental` copies untouched leaves verbatim (patching only `next_leaf`), repacks only leaves that receive pending entries, and rebuilds internal levels with a helper extracted from `build_btree`. `PersistentFactStorage` caches the CRC32 state over its (immutable) fact pages so each save hashes only new fact pages plus the index region. + +**Tech Stack:** Rust (MSRV 1.89), postcard, crc32fast, criterion. + +**Spec:** `docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md` + +## Global Constraints + +- File format stays v7: node layout, header layout, and `index_checksum` value (CRC32 over pages `1..page_count`) are unchanged; files must open on v2.0.1 without the index-rebuild path. +- No public API change; no new dependencies (dev or runtime). +- Test assert messages must never format `Result`, `Fact`, `Value`, `EdnValue`, `EavtKey`, `FactRef` or anything containing a `Uuid` with `{:?}` (CodeQL `rust/cleartext-logging`). Use plain string messages. +- Match the surrounding clippy style: `btree_v6.rs` functions that index/do arithmetic carry `#[allow(clippy::arithmetic_side_effects)]` / `#[allow(clippy::indexing_slicing)]`; errors use `err_coded!`/`bail_coded!` with `ErrorCode::Int049` for B+tree invariant violations. +- `cargo test --release` does not work (panic=abort profile); use `cargo test` and `cargo bench`. +- Pre-push hook runs `cargo fmt --check`, `cargo clippy`, `cargo test`; run `cargo fmt` and `cargo clippy --all-targets -- -D warnings` before each commit. +- Commit messages end with: + ``` + Co-Authored-By: Claude Opus 5.5 + Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy + ``` +- Success target: 50k-fact graph, checkpoint after 1 fact ≤ 15 ms (was ~80 ms). + +## Review Focus + +1. **New index region overlaps the old one** — `save()` writes new fact pages over the start of the old index, and the new tree starts where the old tree's leaves were. Expect: entries identical to a from-scratch build. Pinned by `test_incremental_new_tree_overlaps_old_pages` (Task 3). +2. **Repeated checkpoints on an incrementally built tree** — the second, third… checkpoint consumes a tree whose leaves were raw-copied/split by the previous one. Expect: still exact. Pinned by `test_incremental_repeated_rounds` (Task 3) and `test_incremental_saves_keep_indexes_and_checksum_exact` (Task 4). +3. **Pending keys outside the old key range** (below the first leaf's first key, above the last leaf's last key). Expect: land in first/last leaf, range scans find them. Pinned by `test_incremental_pending_outside_old_range` (Task 3). +4. **Stale or wrong checksum-prefix cache** (after a failed save, a rebuild-on-open, or a page-count mismatch). Expect: silent fallback to a full pass and a correct stored checksum. Pinned by `test_prefix_cache_page_count_mismatch_falls_back` and `test_prefix_cache_absent_after_rebuild_on_open` (Task 4). +5. **Many pending entries into one leaf** (hot key range, e.g. all new facts for one entity). Expect: leaf splits into several ≤75%-fill leaves, no overflow. Pinned by `test_incremental_many_pending_into_one_leaf_splits` (Task 3). + +--- + +### Task 1: Benchmark `checkpoint/after_1_fact` and record the baseline + +**Files:** +- Modify: `benches/minigraf_bench.rs` (new function after `bench_checkpoint`, ~line 416; register in `criterion_group!` at ~line 1778) + +**Interfaces:** +- Consumes: `helpers::populate_file_no_checkpoint(n, &path)`, `helpers::open_file_no_checkpoint(&path) -> Arc` +- Produces: criterion group `checkpoint/after_1_fact` with ids `10k`, `100k` + +- [ ] **Step 1: Add the bench function** after `bench_checkpoint`: + +```rust +// ── checkpoint/after_1_fact (#315) ──────────────────────────────────────────── + +/// Checkpoint cost when only one fact is dirty, on an already-checkpointed graph. +/// Before #315 this was flat in dirty bytes and proportional to graph size. +fn bench_checkpoint_after_1_fact(c: &mut Criterion) { + use criterion::BatchSize; + use tempfile::NamedTempFile; + + let mut group = c.benchmark_group("checkpoint/after_1_fact"); + group.sample_size(20); + for &(label, n) in &[("10k", 10_000usize), ("100k", 100_000)] { + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + helpers::populate_file_no_checkpoint(n, &path); + let db = helpers::open_file_no_checkpoint(&path); + db.checkpoint().unwrap(); + let mut i = 0u64; + group.bench_function(BenchmarkId::from_parameter(label), |b| { + b.iter_batched( + || { + i += 1; + db.execute(&format!("(transact [[:ck{i} :val {i}]])")) + .unwrap(); + }, + |()| db.checkpoint().unwrap(), + BatchSize::SmallInput, + ); + }); + } + group.finish(); +} +``` + +- [ ] **Step 2: Register it** — add `bench_checkpoint_after_1_fact,` to the `criterion_group!(benches, ...)` list directly after `bench_checkpoint,`. + +- [ ] **Step 3: Record the baseline** + +Run: `cargo bench --bench minigraf_bench -- checkpoint/after_1_fact` +Expected: completes; note the two `time:` medians (≈ tens of ms at 100k) in the task report — they are the "Before" column for Task 5. + +- [ ] **Step 4: Commit** + +```bash +cargo fmt && git add benches/minigraf_bench.rs +git commit -m "bench: add checkpoint/after_1_fact group (#315)" +``` + +--- + +### Task 2: Extract leaf encoding, fill rule and internal-level build from `build_btree` + +Pure refactor; existing `btree_v6` tests are the safety net. + +**Files:** +- Modify: `src/storage/btree_v6.rs:62-420` + +**Interfaces:** +- Produces (module-private, used by Task 3): + - `fn encode_leaf_page(entries: &[Vec], next_leaf: u64) -> Result>` + - `fn leaf_overflows(n_entries: usize, data_bytes: usize, entry_len: usize) -> bool` + - `fn build_internal_levels(leaf_infos: Vec<(u64, Vec)>, backend: &mut dyn StorageBackend, cache: &PageCache, next_page: u64) -> Result<(u64, u64)>` — returns `(root_page_id, next_free_page_id)` + +- [ ] **Step 1: Split `write_leaf_page`.** Rename the current body (everything up to but excluding `backend.write_page`) into `encode_leaf_page` returning the page, and make `write_leaf_page` a wrapper: + +```rust +/// Encode a leaf page: fixed header, slot directory, entries written end-to-start. +/// +/// `entries`: each element is the postcard-serialised `(K, FactRef)` bytes for +/// one index entry, in sort order. +#[allow(clippy::arithmetic_side_effects, clippy::indexing_slicing)] +fn encode_leaf_page(entries: &[Vec], next_leaf: u64) -> Result> { + // ... existing body of write_leaf_page from `let entry_count = ...` + // through the slot-directory loop, unchanged ... + Ok(page) +} + +/// Write a single leaf page and insert it into the cache. +fn write_leaf_page( + backend: &mut dyn StorageBackend, + cache: &PageCache, + page_id: u64, + entries: &[Vec], + next_leaf: u64, +) -> Result<()> { + let page = encode_leaf_page(entries, next_leaf)?; + backend.write_page(page_id, &page)?; + cache.put_dirty(page_id, page); + Ok(()) +} +``` + +- [ ] **Step 2: Extract the fill rule.** Add below `write_internal_page`: + +```rust +/// True when adding an entry of `entry_len` bytes to a leaf that already holds +/// `n_entries` entries totalling `data_bytes` would exceed the fill threshold. +#[allow(clippy::arithmetic_side_effects)] +fn leaf_overflows(n_entries: usize, data_bytes: usize, entry_len: usize) -> bool { + LEAF_HEADER_SIZE + (n_entries + 1) * SLOT_SIZE + data_bytes + entry_len > PAGE_FILL_BYTES +} +``` + +and in `build_btree` phase 1 replace + +```rust + let projected = LEAF_HEADER_SIZE + + (cur_entries.len() + 1) * SLOT_SIZE + + cur_data_bytes + + entry_bytes.len(); + + if projected > PAGE_FILL_BYTES && !cur_entries.is_empty() { +``` + +with + +```rust + if !cur_entries.is_empty() + && leaf_overflows(cur_entries.len(), cur_data_bytes, entry_bytes.len()) + { +``` + +- [ ] **Step 3: Extract phase 2.** Move everything in `build_btree` from `// Single leaf: it is the root` to the end of the function into: + +```rust +/// Build internal levels bottom-up over `leaf_infos` (`(page_id, first_key_bytes)` +/// per leaf, in key order), writing nodes from `next_page` onward. +/// +/// Returns `(root_page_id, next_free_page_id)`. A single leaf is its own root. +#[allow(clippy::arithmetic_side_effects, clippy::indexing_slicing)] +fn build_internal_levels( + leaf_infos: Vec<(u64, Vec)>, + backend: &mut dyn StorageBackend, + cache: &PageCache, + next_page: u64, +) -> Result<(u64, u64)> { + if leaf_infos.is_empty() { + bail_coded!( + ErrorCode::Int049, + "build_internal_levels: no leaves".to_string() + ); + } + let mut next_page = next_page; + let mut current_level = leaf_infos; + loop { + // ... existing phase-2 loop body, unchanged (it already returns + // `(current_level[0].0, next_page)` when current_level.len() == 1) ... + } +} +``` + +Delete the now-redundant "Single leaf: it is the root" block (the loop handles `len() == 1`). End `build_btree` with: + +```rust + build_internal_levels(leaf_infos, backend, cache, next_page) +``` + +- [ ] **Step 4: Run the B+tree and storage tests** + +Run: `cargo test --lib storage::` +Expected: all PASS (no behaviour change). + +- [ ] **Step 5: Commit** + +```bash +cargo fmt && cargo clippy --all-targets -- -D warnings +git add src/storage/btree_v6.rs +git commit -m "refactor(btree_v6): extract leaf encoding, fill rule, internal-level build (#315)" +``` + +--- + +### Task 3: `collect_leaf_pages` and `rebuild_btree_incremental` + +**Files:** +- Modify: `src/storage/btree_v6.rs` (new section after `merge_sorted_vecs`; tests in the existing `mod tests`) + +**Interfaces:** +- Consumes (Task 2): `encode_leaf_page`, `leaf_overflows`, `build_internal_levels`; existing `build_btree`, `btree_entries`, `merge_sorted_vecs`, `read_leaf_entries`, `find_leftmost_leaf`, `read_u16_at`, `read_u64_at`. +- Produces (used by Task 4): + - `pub fn collect_leaf_pages(root_page_id: u64, backend: &dyn StorageBackend, cache: &PageCache) -> Result>>>` + - `pub fn rebuild_btree_incremental(old_leaves: Vec>>, pending: Vec<(K, FactRef)>, backend: &mut dyn StorageBackend, cache: &PageCache, start_page_id: u64) -> Result<(u64, u64)> where K: Serialize + for<'de> Deserialize<'de> + Ord` — `pending` must be sorted; returns `(root_page_id, next_free_page_id)`. + +- [ ] **Step 1: Write the failing tests** in `mod tests` (add `use std::sync::Arc;` if not already imported there): + +```rust + /// Deterministic xorshift64 so randomized tests are reproducible without deps. + struct Rng(u64); + impl Rng { + fn next(&mut self) -> u64 { + let mut x = self.0; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.0 = x; + x + } + } + + /// `n` EAVT entries with random entities; `counter` keeps keys unique. + fn random_eavt(rng: &mut Rng, n: usize, counter: &mut u64) -> Vec<(EavtKey, FactRef)> { + let mut v: Vec<(EavtKey, FactRef)> = (0..n) + .map(|_| { + *counter += 1; + let entity = (u128::from(rng.next()) << 64) | u128::from(*counter); + make_eavt(entity, ":attr", *counter) + }) + .collect(); + v.sort(); + v + } + + fn sorted_union( + a: &[(EavtKey, FactRef)], + b: &[(EavtKey, FactRef)], + ) -> Vec<(EavtKey, FactRef)> { + let mut v: Vec<_> = a.iter().chain(b.iter()).cloned().collect(); + v.sort(); + v + } + + /// Build `committed` at page 1, then rebuild with `pending` starting at `start`. + fn build_then_incremental( + committed: &[(EavtKey, FactRef)], + pending: &[(EavtKey, FactRef)], + start: u64, + ) -> (MemoryBackend, PageCache, u64, u64) { + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let ser = btree_entries(committed.iter().cloned()).unwrap(); + let (old_root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let leaves = collect_leaf_pages(old_root, &backend, &cache).unwrap(); + let (root, next_free) = + rebuild_btree_incremental(leaves, pending.to_vec(), &mut backend, &cache, start) + .unwrap(); + (backend, cache, root, next_free) + } + + /// Stream equality, a strictly ascending next_leaf chain of non-empty leaves, + /// and range_scan agreement on random bounds. + fn assert_tree_exact( + root: u64, + backend: &MemoryBackend, + cache: &PageCache, + expected: &[(EavtKey, FactRef)], + rng: &mut Rng, + ) { + let got: Vec<(EavtKey, FactRef)> = stream_all_entries(root, backend, cache).unwrap(); + assert_eq!(got.len(), expected.len(), "entry count differs"); + assert!(got == expected, "streamed entries differ from expected"); + + let leaves = collect_leaf_pages(root, backend, cache).unwrap(); + if expected.is_empty() { + assert_eq!(leaves.len(), 1, "empty tree is one empty leaf"); + } else { + for page in &leaves { + assert!(read_u16_at(&page[..], 2).unwrap() > 0, "empty leaf in tree"); + } + } + + for _ in 0..20 { + if expected.is_empty() { + break; + } + let a = (rng.next() as usize) % expected.len(); + let b = (rng.next() as usize) % expected.len(); + let (lo, hi) = (a.min(b), a.max(b)); + let start = &expected[lo].0; + let end = &expected[hi].0; + let want: Vec = expected[lo..hi].iter().map(|(_, r)| *r).collect(); + let refs = range_scan(root, start, Some(end), backend, cache).unwrap(); + assert!(refs == want, "range_scan differs from expected slice"); + } + } + + #[test] + fn test_incremental_no_pending_copies_leaves_verbatim() { + let mut rng = Rng(0x315); + let mut ctr = 0; + let committed = random_eavt(&mut rng, 2000, &mut ctr); + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let ser = btree_entries(committed.iter().cloned()).unwrap(); + let (old_root, old_next) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let old_leaves = collect_leaf_pages(old_root, &backend, &cache).unwrap(); + let (root, _) = rebuild_btree_incremental( + old_leaves.clone(), + Vec::new(), + &mut backend, + &cache, + old_next, + ) + .unwrap(); + let new_leaves = collect_leaf_pages(root, &backend, &cache).unwrap(); + assert_eq!(new_leaves.len(), old_leaves.len(), "leaf count changed"); + for (o, n) in old_leaves.iter().zip(new_leaves.iter()) { + assert!(o[..4] == n[..4], "leaf header prefix changed"); + assert!(o[12..] == n[12..], "leaf body changed"); + } + assert_tree_exact(root, &backend, &cache, &committed, &mut rng); + } + + #[test] + fn test_incremental_matches_merge_random() { + let mut rng = Rng(0xC0FFEE); + let mut ctr = 0; + for round in 0..40 { + let committed = random_eavt(&mut rng, (rng.next() % 3000) as usize, &mut ctr); + let pending = random_eavt(&mut rng, (rng.next() % 200) as usize, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + let expected = sorted_union(&committed, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + let _ = round; + } + } + + #[test] + fn test_incremental_pending_outside_old_range() { + let mut rng = Rng(7); + let committed: Vec<_> = (1000u128..3000).map(|n| make_eavt(n, ":a", n as u64)).collect(); + let pending = vec![ + make_eavt(1, ":a", 1), + make_eavt(2, ":a", 2), + make_eavt(9_000, ":a", 9_000), + make_eavt(9_001, ":a", 9_001), + ]; + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + assert_tree_exact(root, &backend, &cache, &sorted_union(&committed, &pending), &mut rng); + } + + #[test] + fn test_incremental_many_pending_into_one_leaf_splits() { + let mut rng = Rng(11); + // Committed entities are 0, 10_000, 20_000, ...: pending 0 < e < 10_000 all route to leaf 0. + let committed: Vec<_> = (0u128..2000) + .map(|n| make_eavt(n * 10_000, ":a", n as u64 + 1)) + .collect(); + let pending: Vec<_> = (1u128..3000) + .map(|n| make_eavt(n, ":a", 100_000 + n as u64)) + .collect(); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + assert_tree_exact(root, &backend, &cache, &sorted_union(&committed, &pending), &mut rng); + } + + #[test] + fn test_incremental_empty_old_tree_falls_back_to_bulk_build() { + let mut rng = Rng(13); + let mut ctr = 0; + let pending = random_eavt(&mut rng, 500, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&[], &pending, 5000); + assert_tree_exact(root, &backend, &cache, &pending, &mut rng); + // Nothing old, nothing new: still a valid single empty leaf. + let (backend, cache, root, next) = build_then_incremental(&[], &[], 5000); + assert_eq!(next, root + 1, "empty result is one page"); + assert_tree_exact(root, &backend, &cache, &[], &mut rng); + } + + #[test] + fn test_incremental_new_tree_overlaps_old_pages() { + // save() writes the new tree over the old tree's pages; old leaves are + // snapshotted first, so the result must still be exact. + let mut rng = Rng(17); + let mut ctr = 0; + let committed = random_eavt(&mut rng, 3000, &mut ctr); + let pending = random_eavt(&mut rng, 100, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 3); + assert_tree_exact(root, &backend, &cache, &sorted_union(&committed, &pending), &mut rng); + } + + #[test] + fn test_incremental_repeated_rounds() { + // Each round consumes the previous round's incrementally built tree, + // writing the new tree at the same start page like save() does. + let mut rng = Rng(19); + let mut ctr = 0; + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let mut expected = random_eavt(&mut rng, 1500, &mut ctr); + let ser = btree_entries(expected.iter().cloned()).unwrap(); + let (mut root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + for round in 0..30u64 { + let pending = random_eavt(&mut rng, 1 + (rng.next() % 150) as usize, &mut ctr); + let leaves = collect_leaf_pages(root, &backend, &cache).unwrap(); + let start = 1 + round % 3; + let (r, _) = + rebuild_btree_incremental(leaves, pending.clone(), &mut backend, &cache, start) + .unwrap(); + root = r; + expected = sorted_union(&expected, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + } +``` + +- [ ] **Step 2: Run to verify failure** + +Run: `cargo test --lib btree_v6::tests::test_incremental` +Expected: compile error — `collect_leaf_pages` / `rebuild_btree_incremental` not found. + +- [ ] **Step 3: Implement.** Add after `merge_sorted_vecs`: + +```rust +// ─── Incremental rebuild (#315) ─────────────────────────────────────────────── + +/// Collect the raw leaf pages of the B+tree at `root_page_id`, in key order. +/// +/// `save()` calls this before writing anything: new fact pages overwrite the +/// start of the old index region, so the old leaves must be snapshotted first. +pub fn collect_leaf_pages( + root_page_id: u64, + backend: &dyn StorageBackend, + cache: &PageCache, +) -> Result>>> { + let mut leaves = Vec::new(); + let mut leaf_id = find_leftmost_leaf(root_page_id, backend, cache)?; + loop { + let page = cache.get_or_load(leaf_id, backend)?; + if page.first().copied() != Some(PAGE_TYPE_LEAF) { + bail_coded!( + ErrorCode::Int049, + format!("collect_leaf_pages: expected leaf page at page_id={leaf_id}") + ); + } + let next_leaf = read_u64_at(&page[..], 4)?; + leaves.push(page); + if next_leaf == 0 { + break; + } + leaf_id = next_leaf; + } + Ok(leaves) +} + +/// Decode the key of a leaf's first entry; `None` for an empty leaf. +#[allow(clippy::arithmetic_side_effects)] +fn leaf_first_key(page: &[u8]) -> Result> +where + K: for<'de> Deserialize<'de>, +{ + if read_u16_at(page, 2)? == 0 { + return Ok(None); + } + let offset = read_u16_at(page, LEAF_HEADER_SIZE)? as usize; + let length = read_u16_at(page, LEAF_HEADER_SIZE + 2)? as usize; + let slice = page + .get(offset..offset.saturating_add(length)) + .ok_or_else(|| { + err_coded!( + ErrorCode::Int049, + format!("first entry out of bounds: offset={offset} len={length}") + ) + })?; + let (key, _): (K, FactRef) = postcard::from_bytes(slice)?; + Ok(Some(key)) +} + +/// Writes leaf pages at consecutive page ids, linking each to the next. +/// +/// One page is held back so its `next_leaf` can be set to the following page +/// id, or to 0 once it is known to be the last leaf. +struct LeafEmitter<'a> { + backend: &'a mut dyn StorageBackend, + cache: &'a PageCache, + next_page: u64, + held: Option<(u64, Vec)>, + infos: Vec<(u64, Vec)>, +} + +impl LeafEmitter<'_> { + #[allow(clippy::arithmetic_side_effects)] + fn push(&mut self, page: Vec, first_key_bytes: Vec) -> Result<()> { + let page_id = self.next_page; + self.next_page += 1; + self.release_held(page_id)?; + self.held = Some((page_id, page)); + self.infos.push((page_id, first_key_bytes)); + Ok(()) + } + + fn release_held(&mut self, next_leaf: u64) -> Result<()> { + if let Some((page_id, mut page)) = self.held.take() { + page.get_mut(4..12) + .ok_or_else(|| { + err_coded!( + ErrorCode::Int049, + "page too small to write next_leaf".to_string() + ) + })? + .copy_from_slice(&next_leaf.to_le_bytes()); + self.backend.write_page(page_id, &page)?; + self.cache.put_dirty(page_id, page); + } + Ok(()) + } + + /// Write the last held leaf and return `(leaf_infos, next_free_page_id)`. + fn finish(mut self) -> Result<(Vec<(u64, Vec)>, u64)> { + self.release_held(0)?; + Ok((self.infos, self.next_page)) + } +} + +/// Serialise sorted entries and emit them as leaves using `build_btree`'s fill rule. +#[allow(clippy::arithmetic_side_effects)] +fn emit_packed( + emitter: &mut LeafEmitter<'_>, + entries: impl Iterator, +) -> Result<()> { + let mut cur: Vec> = Vec::new(); + let mut cur_bytes = 0usize; + let mut cur_first: Option> = None; + for (key, fact_ref) in entries { + let entry = postcard::to_allocvec(&(&key, &fact_ref))?; + if !cur.is_empty() && leaf_overflows(cur.len(), cur_bytes, entry.len()) { + let first = cur_first.take().ok_or_else(|| { + err_coded!(ErrorCode::Int049, "BUG: leaf without first key".to_string()) + })?; + emitter.push(encode_leaf_page(&cur, 0)?, first)?; + cur.clear(); + cur_bytes = 0; + } + if cur_first.is_none() { + cur_first = Some(postcard::to_allocvec(&key)?); + } + cur_bytes += entry.len(); + cur.push(entry); + } + if let Some(first) = cur_first { + emitter.push(encode_leaf_page(&cur, 0)?, first)?; + } + Ok(()) +} + +/// Rebuild a B+tree from its old leaves plus sorted `pending` entries. +/// +/// Leaves that receive no pending entry are copied verbatim (only `next_leaf` +/// is patched); only leaves that do are decoded, merged and repacked. Internal +/// levels are rebuilt from scratch. Pending keys route to leaf `i` when +/// `first_key[i] <= key < first_key[i + 1]`; keys below the first leaf's first +/// key go to leaf 0. With no non-empty old leaves this is a plain bulk build. +/// +/// `old_leaves` must be snapshotted (see [`collect_leaf_pages`]) before any +/// page in `start_page_id..` is written. Returns `(root_page_id, next_free_page_id)`. +pub fn rebuild_btree_incremental( + old_leaves: Vec>>, + pending: Vec<(K, FactRef)>, + backend: &mut dyn StorageBackend, + cache: &PageCache, + start_page_id: u64, +) -> Result<(u64, u64)> +where + K: Serialize + for<'de> Deserialize<'de> + Ord, +{ + let mut leaves: Vec<(Arc>, K)> = Vec::with_capacity(old_leaves.len()); + for page in old_leaves { + if let Some(first_key) = leaf_first_key::(&page[..])? { + leaves.push((page, first_key)); + } + } + if leaves.is_empty() { + return build_btree( + btree_entries(pending.into_iter())?.into_iter(), + backend, + cache, + start_page_id, + ); + } + + let mut emitter = LeafEmitter { + backend: &mut *backend, + cache, + next_page: start_page_id, + held: None, + infos: Vec::new(), + }; + let mut pending = pending.into_iter().peekable(); + for (i, (page, first_key)) in leaves.iter().enumerate() { + let upper = leaves.get(i.saturating_add(1)).map(|(_, k)| k); + let mut batch = Vec::new(); + while let Some((key, _)) = pending.peek() { + if upper.is_some_and(|u| key >= u) { + break; + } + if let Some(entry) = pending.next() { + batch.push(entry); + } + } + if batch.is_empty() { + emitter.push((**page).clone(), postcard::to_allocvec(first_key)?)?; + } else { + let old = read_leaf_entries::(&page[..])?; + emit_packed(&mut emitter, merge_sorted_vecs(old, batch))?; + } + } + let (leaf_infos, next_page) = emitter.finish()?; + build_internal_levels(leaf_infos, backend, cache, next_page) +} +``` + +- [ ] **Step 4: Run the tests** + +Run: `cargo test --lib btree_v6::` +Expected: all PASS, including the seven `test_incremental_*`. + +- [ ] **Step 5: Commit** + +```bash +cargo fmt && cargo clippy --all-targets -- -D warnings +git add src/storage/btree_v6.rs +git commit -m "feat(btree_v6): incremental rebuild that copies untouched leaves (#315)" +``` + +--- + +### Task 4: Wire into `save()` and cache the fact-page checksum prefix + +**Files:** +- Modify: `src/storage/persistent_facts.rs` — struct (~line 145), `new` (~line 166), `load` checksum block (~lines 277-301), `save` Steps A/D/checksum (~lines 842-970), `compute_page_checksum` (~line 1102), imports (lines 10-13), tests module + +**Interfaces:** +- Consumes (Task 3): `collect_leaf_pages`, `rebuild_btree_incremental`. +- Produces: field `fact_prefix_crc: Option<(u64, Hasher)>`; `fn hash_pages(backend: &dyn StorageBackend, hasher: &mut Hasher, first_page: u64, num_pages: u64) -> Result<()>`. + +- [ ] **Step 1: Write the failing tests** in the `#[cfg(test)] mod tests` of `persistent_facts.rs`: + +```rust + /// All four on-disk indexes equal a from-scratch derivation from the fact pages, + /// and the stored checksum equals a full pass over pages 1..page_count. + fn assert_indexes_and_checksum_exact( + pfs: &PersistentFactStorage, + ) { + let backend = pfs.backend.lock().unwrap(); + let header = FileHeader::from_bytes(&backend.read_page(0).unwrap()).unwrap(); + let full = compute_page_checksum(&*backend, 1, header.page_count - 1).unwrap(); + assert_eq!(header.index_checksum, full, "stored checksum != full pass"); + + let (facts, refs) = crate::storage::packed_pages::read_all_with_refs( + &*backend, + 1, + header.fact_page_count, + ) + .unwrap(); + let (eavt, aevt, avet, vaet) = build_sorted_index_entries(&facts, &refs); + let cache = &pfs.page_cache; + let got_eavt: Vec<(EavtKey, FactRef)> = + stream_all_entries(header.eavt_root_page, &*backend, cache).unwrap(); + let got_aevt: Vec<(AevtKey, FactRef)> = + stream_all_entries(header.aevt_root_page, &*backend, cache).unwrap(); + let got_avet: Vec<(AvetKey, FactRef)> = + stream_all_entries(header.avet_root_page, &*backend, cache).unwrap(); + let got_vaet: Vec<(VaetKey, FactRef)> = + stream_all_entries(header.vaet_root_page, &*backend, cache).unwrap(); + assert!(got_eavt == eavt, "EAVT differs from full derivation"); + assert!(got_aevt == aevt, "AEVT differs from full derivation"); + assert!(got_avet == avet, "AVET differs from full derivation"); + assert!(got_vaet == vaet, "VAET differs from full derivation"); + } + + /// One transact of `n` facts: new and reused entities, strings and refs (VAET). + fn transact_mixed( + pfs: &mut PersistentFactStorage, + entities: &mut Vec, + n: usize, + seed: u64, + ) { + let mut batch = Vec::new(); + for i in 0..n { + let k = seed.wrapping_mul(31).wrapping_add(i as u64); + let e = if entities.is_empty() || k % 3 == 0 { + let e = Uuid::new_v4(); + entities.push(e); + e + } else { + entities[(k as usize) % entities.len()] + }; + let value = if k % 4 == 0 { + Value::Ref(entities[(k as usize / 4) % entities.len()]) + } else { + Value::String(format!("v{k}")) + }; + batch.push((e, format!(":a{}", k % 5), value)); + } + pfs.storage().transact(batch, None).unwrap(); + pfs.mark_dirty(); + } + + #[test] + fn test_incremental_saves_keep_indexes_and_checksum_exact() { + let mut pfs = PersistentFactStorage::new(MemoryBackend::new(), 256).unwrap(); + let mut entities = Vec::new(); + for round in 0..25u64 { + let n = 1 + ((round * 37) % 60) as usize; + transact_mixed(&mut pfs, &mut entities, n, round); + pfs.save().unwrap(); + assert!(pfs.fact_prefix_crc.is_some(), "prefix cached after save"); + assert_indexes_and_checksum_exact(&pfs); + } + } + + #[test] + fn test_reopen_after_incremental_saves_takes_no_rebuild() { + use crate::storage::backend::FileBackend; + use tempfile::NamedTempFile; + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + let mut entities = Vec::new(); + { + let mut pfs = + PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + for round in 0..10u64 { + transact_mixed(&mut pfs, &mut entities, 40, round); + pfs.save().unwrap(); + } + } + let mut pfs = PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + // Set only when the full checksum matched on load, i.e. no rebuild. + assert!(pfs.fact_prefix_crc.is_some(), "full checksum must match on reopen"); + transact_mixed(&mut pfs, &mut entities, 40, 99); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } + + #[test] + fn test_prefix_cache_page_count_mismatch_falls_back() { + let mut pfs = PersistentFactStorage::new(MemoryBackend::new(), 256).unwrap(); + let mut entities = Vec::new(); + transact_mixed(&mut pfs, &mut entities, 50, 1); + pfs.save().unwrap(); + pfs.fact_prefix_crc = Some((999, Hasher::new())); + transact_mixed(&mut pfs, &mut entities, 50, 2); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } + + #[test] + fn test_prefix_cache_absent_after_rebuild_on_open() { + use crate::storage::backend::FileBackend; + use tempfile::NamedTempFile; + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + let mut entities = Vec::new(); + { + let mut pfs = + PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + for round in 0..3u64 { + transact_mixed(&mut pfs, &mut entities, 30, round); + pfs.save().unwrap(); + } + } + // Corrupt index_checksum (re-sealing the header) to force the rebuild path. + { + let mut backend = FileBackend::open(&path).unwrap(); + let mut page = backend.read_page(0).unwrap(); + page[64] ^= 0xFF; + let cs = compute_header_checksum_from_bytes(&page); + page[80..84].copy_from_slice(&cs.to_le_bytes()); + backend.write_page(0, &page).unwrap(); + backend.sync().unwrap(); + } + let mut pfs = PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + assert!(pfs.fact_prefix_crc.is_none(), "no prefix after rebuild-on-open"); + transact_mixed(&mut pfs, &mut entities, 30, 7); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } +``` + +Add any missing test-module imports (`FileHeader`, `StorageBackend`, `MemoryBackend`, `Value`, `Uuid`, `Hasher`, `EavtKey`/`AevtKey`/`AvetKey`/`VaetKey`, `FactRef`, `stream_all_entries`) — most are already imported via `use super::*;`; the compiler will list the rest. + +- [ ] **Step 2: Run to verify failure** + +Run: `cargo test --lib persistent_facts::tests::test_ -- incremental prefix reopen_after` +Expected: compile error — no field `fact_prefix_crc`. + +- [ ] **Step 3: Add the field.** In `PersistentFactStorage`: + +```rust + committed_fact_pages: Arc, + /// CRC32 state after hashing fact pages `1..=n` (`n` is the first element). + /// Fact pages are never rewritten, so `save()` extends this instead of + /// re-reading them; `None` (or a stale `n`) falls back to a full pass (#315). + fact_prefix_crc: Option<(u64, Hasher)>, +``` + +and `fact_prefix_crc: None,` in the constructor literal in `new`. + +- [ ] **Step 4: Add `hash_pages`** and re-express `compute_page_checksum` with it: + +```rust +/// Feed pages `first_page..first_page + num_pages` into `hasher`. +fn hash_pages( + backend: &dyn StorageBackend, + hasher: &mut Hasher, + first_page: u64, + num_pages: u64, +) -> Result<()> { + for i in 0..num_pages { + let page_id = first_page + .checked_add(i) + .ok_or_else(|| err_coded!(ErrorCode::Stg021))?; + hasher.update(&backend.read_page(page_id)?); + } + Ok(()) +} + +fn compute_page_checksum( + backend: &dyn StorageBackend, + first_page: u64, + num_pages: u64, +) -> Result { + let mut hasher = Hasher::new(); + hash_pages(backend, &mut hasher, first_page, num_pages)?; + Ok(hasher.finalize()) +} +``` + +- [ ] **Step 5: Capture the prefix in `load()`.** In the `needs_rebuild` block replace + +```rust + let full_checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + if full_checksum == stored { + false // new-style checksum matches: facts + indexes verified +``` + +with + +```rust + // Hash fact pages, keep that state for save() (#315), then the index pages. + let (full_checksum, fact_prefix) = if total_data_pages >= num_fact_pages { + let mut hasher = Hasher::new(); + hash_pages(&*backend, &mut hasher, 1, num_fact_pages)?; + let prefix = hasher.clone(); + let index_start = num_fact_pages + .checked_add(1) + .ok_or_else(|| err_coded!(ErrorCode::Stg021))?; + hash_pages( + &*backend, + &mut hasher, + index_start, + total_data_pages - num_fact_pages, + )?; + (hasher.finalize(), Some(prefix)) + } else { + (compute_page_checksum(&*backend, 1, total_data_pages)?, None) + }; + if full_checksum == stored { + self.fact_prefix_crc = fact_prefix.map(|p| (num_fact_pages, p)); + false // new-style checksum matches: facts + indexes verified +``` + +(The subtraction is guarded by the `>=` check; add `#[allow(clippy::arithmetic_side_effects)]` only if clippy asks.) + +- [ ] **Step 6: Step A of `save()`** — replace the four `committed_* = stream_all_entries(...)` blocks with raw-leaf snapshots: + +```rust + // Snapshot old B+tree leaves BEFORE writing new pages that may overlap them. + let old_leaves = |root: u64| -> Result>>> { + if root == 0 { + Ok(Vec::new()) + } else { + collect_leaf_pages(root, &*backend, &self.page_cache) + } + }; + let old_eavt = old_leaves(curr_header.eavt_root_page)?; + let old_aevt = old_leaves(curr_header.aevt_root_page)?; + let old_avet = old_leaves(curr_header.avet_root_page)?; + let old_vaet = old_leaves(curr_header.vaet_root_page)?; +``` + +(If the closure's borrow of `backend` conflicts with the later `&mut *backend`, it ends at its last use; if the borrow checker still objects, inline the four `if` expressions.) + +- [ ] **Step 7: Step D of `save()`** — replace the four `*_ser` / `build_btree` pairs with: + +```rust + let (eavt_root, next1) = rebuild_btree_incremental( + old_eavt, + pending_eavt, + &mut *backend, + &self.page_cache, + index_start, + )?; + let (aevt_root, next2) = + rebuild_btree_incremental(old_aevt, pending_aevt, &mut *backend, &self.page_cache, next1)?; + let (avet_root, next3) = + rebuild_btree_incremental(old_avet, pending_avet, &mut *backend, &self.page_cache, next2)?; + let (vaet_root, next4) = + rebuild_btree_incremental(old_vaet, pending_vaet, &mut *backend, &self.page_cache, next3)?; +``` + +- [ ] **Step 8: Checksum in `save()`** — replace + +```rust + let total_data_pages = next4.saturating_sub(1); + let checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; +``` + +with + +```rust + // Fact pages 1..=old count are unchanged since the last save/load, so extend + // the cached CRC state instead of re-reading them (#315). Taking the cache + // means a failed save leaves None and the next save does a full pass. + let mut hasher = match self.fact_prefix_crc.take() { + Some((n, prefix)) if n == old_fact_page_count => { + let mut h = prefix; + hash_pages(&*backend, &mut h, new_fact_start, new_pages_len)?; + h + } + _ => { + let mut h = Hasher::new(); + hash_pages(&*backend, &mut h, 1, new_total_fact_pages)?; + h + } + }; + let new_prefix = hasher.clone(); + hash_pages( + &*backend, + &mut hasher, + index_start, + next4.saturating_sub(index_start), + )?; + let checksum = hasher.finalize(); +``` + +and after `self.committed_fact_pages.store(new_total_fact_pages, ...)` add: + +```rust + self.fact_prefix_crc = Some((new_total_fact_pages, new_prefix)); +``` + +- [ ] **Step 9: Fix imports** — add `collect_leaf_pages, rebuild_btree_incremental` to the `use crate::storage::btree_v6::{...}` list; drop names clippy reports unused in non-test code (e.g. `merge_sorted_vecs`; `stream_all_entries` if only tests use it — then import it inside `mod tests`). + +- [ ] **Step 10: Run the tests** + +Run: `cargo test --lib persistent_facts::` then `cargo test` +Expected: all PASS; total = baseline 1168 + 7 (Task 3) + 4 (this task) = 1179 passing, 8 ignored. + +- [ ] **Step 11: Measure** — `cargo bench --bench minigraf_bench -- checkpoint/after_1_fact`. Expected: substantially below the Task 1 baseline. Also rerun the scratch harness if still present (50k facts target ≤ 15 ms). Record numbers in the task report. + +- [ ] **Step 12: Commit** + +```bash +cargo fmt && cargo clippy --all-targets -- -D warnings +git add src/storage/persistent_facts.rs +git commit -m "perf: checkpoint copies untouched index leaves and skips re-hashing fact pages (#315)" +``` + +--- + +### Task 5: Docs, comment trim, doc sync + +**Files:** +- Modify: `src/db.rs` (`OpenOptions::wal_checkpoint_threshold` field doc ~line 112; `Minigraf::checkpoint` doc ~line 684; `do_checkpoint` guard comment ~lines 730-742) +- Modify: `docs/BENCHMARKS.md`, `CHANGELOG.md`, `CLAUDE.md`, `README.md`, `docs/TEST_COVERAGE.md` + +- [ ] **Step 1: `Minigraf::checkpoint` rustdoc** — replace the first paragraph with: + +```rust + /// Manually trigger a checkpoint: flush all in-memory facts to the main file + /// and delete the WAL sidecar. + /// + /// A checkpoint is **not** a durability boundary: committed writes are already + /// crash-durable in the `.wal` sidecar and are replayed on the next open. + /// Checkpointing less often only makes that replay (reopen) slower. + /// + /// Cost: every checkpoint rewrites the four covering indexes after the fact + /// pages, so it copies pages in proportion to the total index size; only the + /// index leaves that receive new entries are decoded and re-encoded. Prefer + /// checkpointing on a size or time budget over once per logical unit of work. + /// + /// No-op for in-memory databases. +``` + +- [ ] **Step 2: `wal_checkpoint_threshold` field doc** — append: + +```rust + /// + /// A checkpoint's cost grows with the total size of the database (see + /// [`Minigraf::checkpoint`]), while durability does not depend on it: the WAL + /// is crash-durable. Raise this for write-heavy workloads on large graphs. +``` + +- [ ] **Step 3: Trim the `do_checkpoint` guard comment** — replace the paragraph starting "The kernel file lock taken by FileBackend::open…" through "…concurrent writers." with: + +```rust + // Kernel file locking refuses a second writer (another process, or a + // second handle in this one — #304/#314). The guard still matters where + // the filesystem cannot lock and the caller set `allow_unlocked`. +``` + +- [ ] **Step 4: `docs/BENCHMARKS.md`** — add after the "Point Query vs. Version-Chain Depth (#323)" section: + +```markdown +### Checkpoint After One Dirty Fact (#315) + +**Date**: · **Command**: `cargo bench --bench minigraf_bench -- checkpoint/after_1_fact` · same host as above. + +An already-checkpointed file database of `n` facts receives one new fact, then `checkpoint()` runs. "Before" is v2.0.1 behaviour (every index entry decoded and re-encoded); "after" copies untouched index leaves verbatim. + +| Facts | Before | After | +|---:|---:|---:| +| 10k | | | +| 100k | | | + +Cost still grows with graph size (index pages are copied every checkpoint); O(delta) checkpoints need copy-on-write pages and are tracked with #374 for v3.0.0. +``` + +Fill `<…>` with the recorded numbers — no placeholders may remain. + +- [ ] **Step 5: `CHANGELOG.md`** — under `## Unreleased` → `### Performance` (create headings if absent), add: + +```markdown +- **Checkpoints no longer re-encode the whole index (#315).** `checkpoint()` used to decode and re-serialise every entry of all four covering indexes and re-read every page for the file checksum, so checkpointing after one new fact cost as much as after thousands. Index leaves that receive no new entries are now copied verbatim, and unchanged fact pages are not re-hashed. On a 100k-fact file, checkpoint after one fact drops from to . Cost still scales with graph size in page copies; the file format is unchanged. The `checkpoint()` docs now state that a checkpoint is not a durability boundary. +``` + +- [ ] **Step 6: Test counts** — run `cargo test 2>&1 | grep "^test result"` and sum; update `CLAUDE.md` (`**N tests passing** (P passing, 8 ignored…)`), `README.md` (`cargo test # run N tests`), `docs/TEST_COVERAGE.md` (`**Verified**` date, `**Result**` line, and one sentence appended to the suite summary: "Since v2.0.1, incremental B+tree rebuild tests check that checkpoints which copy untouched leaves produce exactly the indexes and checksum a full rebuild would (#315)."). + +- [ ] **Step 7: Verify and commit** + +Run: `cargo test` and `cargo doc --no-deps 2>&1 | grep -i warning` (expect none new) + +```bash +cargo fmt && cargo clippy --all-targets -- -D warnings +git add src/db.rs docs/BENCHMARKS.md CHANGELOG.md CLAUDE.md README.md docs/TEST_COVERAGE.md +git commit -m "docs: checkpoint cost and durability semantics; #315 benchmarks and doc sync" +``` + +- [ ] **Step 8 (at PR time, with the user's go-ahead):** comment on #374 that copy-on-write pages there also own #315's remaining O(size) page-copy term. The PR body uses "Mitigates #315" wording chosen with the user, never a closing keyword for #374. From 9bca11628276fb45776b6c524f77f63f18678405 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 11:45:28 +0530 Subject: [PATCH 3/9] bench: add checkpoint/after_1_fact group (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- benches/minigraf_bench.rs | 33 +++++++++++++++++++++++++++++++++ 1 file changed, 33 insertions(+) diff --git a/benches/minigraf_bench.rs b/benches/minigraf_bench.rs index d4b901a7c..1b70d4ed0 100644 --- a/benches/minigraf_bench.rs +++ b/benches/minigraf_bench.rs @@ -415,6 +415,38 @@ fn bench_checkpoint(c: &mut Criterion) { group.finish(); } +// ── checkpoint/after_1_fact (#315) ──────────────────────────────────────────── + +/// Checkpoint cost when only one fact is dirty, on an already-checkpointed graph. +/// Before #315 this was flat in dirty bytes and proportional to graph size. +fn bench_checkpoint_after_1_fact(c: &mut Criterion) { + use criterion::BatchSize; + use tempfile::NamedTempFile; + + let mut group = c.benchmark_group("checkpoint/after_1_fact"); + group.sample_size(20); + for &(label, n) in &[("10k", 10_000usize), ("100k", 100_000)] { + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + helpers::populate_file_no_checkpoint(n, &path); + let db = helpers::open_file_no_checkpoint(&path); + db.checkpoint().unwrap(); + let mut i = 0u64; + group.bench_function(BenchmarkId::from_parameter(label), |b| { + b.iter_batched( + || { + i += 1; + db.execute(&format!("(transact [[:ck{i} :val {i}]])")) + .unwrap(); + }, + |()| db.checkpoint().unwrap(), + BatchSize::SmallInput, + ); + }); + } + group.finish(); +} + // ── Task 10: concurrent/ ───────────────────────────────────────────────────── fn bench_concurrent(c: &mut Criterion) { @@ -1793,6 +1825,7 @@ criterion_group!( bench_query_extras, bench_open, bench_checkpoint, + bench_checkpoint_after_1_fact, bench_concurrent, bench_concurrent_file, bench_concurrent_btree_scan, From f8a38ca0a12ca97bde15c7d0d916bf88b624ad07 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 11:47:01 +0530 Subject: [PATCH 4/9] bench: run one dirty fact per checkpoint iteration (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- benches/minigraf_bench.rs | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/benches/minigraf_bench.rs b/benches/minigraf_bench.rs index 1b70d4ed0..2de35d982 100644 --- a/benches/minigraf_bench.rs +++ b/benches/minigraf_bench.rs @@ -439,8 +439,10 @@ fn bench_checkpoint_after_1_fact(c: &mut Criterion) { db.execute(&format!("(transact [[:ck{i} :val {i}]])")) .unwrap(); }, + // One dirty fact per checkpoint: batched setups would leave + // every checkpoint after the first with nothing to do. |()| db.checkpoint().unwrap(), - BatchSize::SmallInput, + BatchSize::PerIteration, ); }); } From 96e929d5d137e857f2d332026bd31cf1282e1728 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 11:47:43 +0530 Subject: [PATCH 5/9] refactor(btree_v6): extract leaf encoding, fill rule, internal-level build (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- src/storage/btree_v6.rs | 75 ++++++++++++++++++++++++----------------- 1 file changed, 44 insertions(+), 31 deletions(-) diff --git a/src/storage/btree_v6.rs b/src/storage/btree_v6.rs index 5c5a9b9e1..c2a2c82c6 100644 --- a/src/storage/btree_v6.rs +++ b/src/storage/btree_v6.rs @@ -61,18 +61,12 @@ fn read_u64_at(page: &[u8], offset: usize) -> Result { // ─── Low-level page writers ─────────────────────────────────────────────────── -/// Write a single leaf page and insert it into the cache. +/// Encode a leaf page: fixed header, slot directory, entries written end-to-start. /// /// `entries`: each element is the postcard-serialised `(K, FactRef)` bytes for -/// one index entry, in sort order. Written end-to-start in the page. +/// one index entry, in sort order. #[allow(clippy::arithmetic_side_effects, clippy::indexing_slicing)] -fn write_leaf_page( - backend: &mut dyn StorageBackend, - cache: &PageCache, - page_id: u64, - entries: &[Vec], - next_leaf: u64, -) -> Result<()> { +fn encode_leaf_page(entries: &[Vec], next_leaf: u64) -> Result> { let entry_count = u16::try_from(entries.len()).map_err(|_| { err_coded!( ErrorCode::Int049, @@ -108,7 +102,18 @@ fn write_leaf_page( page[slot_off..slot_off + 2].copy_from_slice(&write_pos_u16.to_le_bytes()); page[slot_off + 2..slot_off + 4].copy_from_slice(&entry_len_u16.to_le_bytes()); } + Ok(page) +} +/// Write a single leaf page and insert it into the cache. +fn write_leaf_page( + backend: &mut dyn StorageBackend, + cache: &PageCache, + page_id: u64, + entries: &[Vec], + next_leaf: u64, +) -> Result<()> { + let page = encode_leaf_page(entries, next_leaf)?; backend.write_page(page_id, &page)?; cache.put_dirty(page_id, page); Ok(()) @@ -188,6 +193,13 @@ fn write_internal_page( Ok(()) } +/// True when adding an entry of `entry_len` bytes to a leaf that already holds +/// `n_entries` entries totalling `data_bytes` would exceed the fill threshold. +#[allow(clippy::arithmetic_side_effects)] +fn leaf_overflows(n_entries: usize, data_bytes: usize, entry_len: usize) -> bool { + LEAF_HEADER_SIZE + (n_entries + 1) * SLOT_SIZE + data_bytes + entry_len > PAGE_FILL_BYTES +} + // ─── build_btree ────────────────────────────────────────────────────────────── /// Serialize `(key, fact_ref)` pairs into the byte format expected by [`build_btree`]. @@ -235,12 +247,9 @@ pub fn build_btree( let mut next_page = start_page_id; for (entry_bytes, key_bytes) in sorted_entries { - let projected = LEAF_HEADER_SIZE - + (cur_entries.len() + 1) * SLOT_SIZE - + cur_data_bytes - + entry_bytes.len(); - - if projected > PAGE_FILL_BYTES && !cur_entries.is_empty() { + if !cur_entries.is_empty() + && leaf_overflows(cur_entries.len(), cur_data_bytes, entry_bytes.len()) + { write_leaf_page(backend, cache, next_page, &cur_entries, 0)?; let first_key = cur_first_key.take().ok_or_else(|| { err_coded!( @@ -310,23 +319,27 @@ pub fn build_btree( cache.put_dirty(pid, page); } - // Single leaf: it is the root - if leaf_infos.len() == 1 { - return Ok(( - leaf_infos - .first() - .ok_or_else(|| { - err_coded!( - ErrorCode::Int049, - "leaf_infos unexpectedly empty".to_string() - ) - })? - .0, - next_page, - )); - } + build_internal_levels(leaf_infos, backend, cache, next_page) +} - // ── Phase 2: build internal levels bottom-up ────────────────────────────── +/// Build internal levels bottom-up over `leaf_infos` (`(page_id, first_key_bytes)` +/// per leaf, in key order), writing nodes from `next_page` onward. +/// +/// Returns `(root_page_id, next_free_page_id)`. A single leaf is its own root. +#[allow(clippy::arithmetic_side_effects, clippy::indexing_slicing)] +fn build_internal_levels( + leaf_infos: Vec<(u64, Vec)>, + backend: &mut dyn StorageBackend, + cache: &PageCache, + next_page: u64, +) -> Result<(u64, u64)> { + if leaf_infos.is_empty() { + bail_coded!( + ErrorCode::Int049, + "build_internal_levels: no leaves".to_string() + ); + } + let mut next_page = next_page; let mut current_level = leaf_infos; loop { From 0eb0dc3d7cfe030bdad18a238481f01112c45351 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 11:49:15 +0530 Subject: [PATCH 6/9] feat(btree_v6): incremental rebuild that copies untouched leaves (#315) Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- src/storage/btree_v6.rs | 408 ++++++++++++++++++++++++++++++++++++++++ 1 file changed, 408 insertions(+) diff --git a/src/storage/btree_v6.rs b/src/storage/btree_v6.rs index c2a2c82c6..7b33ef0b2 100644 --- a/src/storage/btree_v6.rs +++ b/src/storage/btree_v6.rs @@ -450,6 +450,202 @@ pub fn merge_sorted_vecs(a: Vec, b: Vec) -> impl Iterator Result>>> { + let mut leaves = Vec::new(); + let mut leaf_id = find_leftmost_leaf(root_page_id, backend, cache)?; + loop { + let page = cache.get_or_load(leaf_id, backend)?; + if page.first().copied() != Some(PAGE_TYPE_LEAF) { + bail_coded!( + ErrorCode::Int049, + format!("collect_leaf_pages: expected leaf page at page_id={leaf_id}") + ); + } + let next_leaf = read_u64_at(&page[..], 4)?; + leaves.push(page); + if next_leaf == 0 { + break; + } + leaf_id = next_leaf; + } + Ok(leaves) +} + +/// Decode the key of a leaf's first entry; `None` for an empty leaf. +#[allow(clippy::arithmetic_side_effects)] +fn leaf_first_key(page: &[u8]) -> Result> +where + K: for<'de> Deserialize<'de>, +{ + if read_u16_at(page, 2)? == 0 { + return Ok(None); + } + let offset = read_u16_at(page, LEAF_HEADER_SIZE)? as usize; + let length = read_u16_at(page, LEAF_HEADER_SIZE + 2)? as usize; + let slice = page + .get(offset..offset.saturating_add(length)) + .ok_or_else(|| { + err_coded!( + ErrorCode::Int049, + format!("first entry out of bounds: offset={offset} len={length}") + ) + })?; + let (key, _): (K, FactRef) = postcard::from_bytes(slice)?; + Ok(Some(key)) +} + +/// Writes leaf pages at consecutive page ids, linking each to the next. +/// +/// One page is held back so its `next_leaf` can be set to the following page +/// id, or to 0 once it is known to be the last leaf. +struct LeafEmitter<'a> { + backend: &'a mut dyn StorageBackend, + cache: &'a PageCache, + next_page: u64, + held: Option<(u64, Vec)>, + infos: Vec<(u64, Vec)>, +} + +impl LeafEmitter<'_> { + #[allow(clippy::arithmetic_side_effects)] + fn push(&mut self, page: Vec, first_key_bytes: Vec) -> Result<()> { + let page_id = self.next_page; + self.next_page += 1; + self.release_held(page_id)?; + self.held = Some((page_id, page)); + self.infos.push((page_id, first_key_bytes)); + Ok(()) + } + + fn release_held(&mut self, next_leaf: u64) -> Result<()> { + if let Some((page_id, mut page)) = self.held.take() { + page.get_mut(4..12) + .ok_or_else(|| { + err_coded!( + ErrorCode::Int049, + "page too small to write next_leaf".to_string() + ) + })? + .copy_from_slice(&next_leaf.to_le_bytes()); + self.backend.write_page(page_id, &page)?; + self.cache.put_dirty(page_id, page); + } + Ok(()) + } + + /// Write the last held leaf and return `(leaf_infos, next_free_page_id)`. + fn finish(mut self) -> Result<(Vec<(u64, Vec)>, u64)> { + self.release_held(0)?; + Ok((self.infos, self.next_page)) + } +} + +/// Serialise sorted entries and emit them as leaves using `build_btree`'s fill rule. +#[allow(clippy::arithmetic_side_effects)] +fn emit_packed( + emitter: &mut LeafEmitter<'_>, + entries: impl Iterator, +) -> Result<()> { + let mut cur: Vec> = Vec::new(); + let mut cur_bytes = 0usize; + let mut cur_first: Option> = None; + for (key, fact_ref) in entries { + let entry = postcard::to_allocvec(&(&key, &fact_ref))?; + if !cur.is_empty() && leaf_overflows(cur.len(), cur_bytes, entry.len()) { + let first = cur_first.take().ok_or_else(|| { + err_coded!(ErrorCode::Int049, "BUG: leaf without first key".to_string()) + })?; + emitter.push(encode_leaf_page(&cur, 0)?, first)?; + cur.clear(); + cur_bytes = 0; + } + if cur_first.is_none() { + cur_first = Some(postcard::to_allocvec(&key)?); + } + cur_bytes += entry.len(); + cur.push(entry); + } + if let Some(first) = cur_first { + emitter.push(encode_leaf_page(&cur, 0)?, first)?; + } + Ok(()) +} + +/// Rebuild a B+tree from its old leaves plus sorted `pending` entries. +/// +/// Leaves that receive no pending entry are copied verbatim (only `next_leaf` +/// is patched); only leaves that do are decoded, merged and repacked. Internal +/// levels are rebuilt from scratch. Pending keys route to leaf `i` when +/// `first_key[i] <= key < first_key[i + 1]`; keys below the first leaf's first +/// key go to leaf 0. With no non-empty old leaves this is a plain bulk build. +/// +/// `old_leaves` must be snapshotted (see [`collect_leaf_pages`]) before any +/// page in `start_page_id..` is written. Returns `(root_page_id, next_free_page_id)`. +pub fn rebuild_btree_incremental( + old_leaves: Vec>>, + pending: Vec<(K, FactRef)>, + backend: &mut dyn StorageBackend, + cache: &PageCache, + start_page_id: u64, +) -> Result<(u64, u64)> +where + K: Serialize + for<'de> Deserialize<'de> + Ord, +{ + let mut leaves: Vec<(Arc>, K)> = Vec::with_capacity(old_leaves.len()); + for page in old_leaves { + if let Some(first_key) = leaf_first_key::(&page[..])? { + leaves.push((page, first_key)); + } + } + if leaves.is_empty() { + return build_btree( + btree_entries(pending.into_iter())?.into_iter(), + backend, + cache, + start_page_id, + ); + } + + let mut emitter = LeafEmitter { + backend: &mut *backend, + cache, + next_page: start_page_id, + held: None, + infos: Vec::new(), + }; + let mut pending = pending.into_iter().peekable(); + for (i, (page, first_key)) in leaves.iter().enumerate() { + let upper = leaves.get(i.saturating_add(1)).map(|(_, k)| k); + let mut batch = Vec::new(); + while let Some((key, _)) = pending.peek() { + if upper.is_some_and(|u| key >= u) { + break; + } + if let Some(entry) = pending.next() { + batch.push(entry); + } + } + if batch.is_empty() { + emitter.push((**page).clone(), postcard::to_allocvec(first_key)?)?; + } else { + let old = read_leaf_entries::(&page[..])?; + emit_packed(&mut emitter, merge_sorted_vecs(old, batch))?; + } + } + let (leaf_infos, next_page) = emitter.finish()?; + build_internal_levels(leaf_infos, backend, cache, next_page) +} + // ─── Leaf traversal helpers ─────────────────────────────────────────────────── /// Traverse internal nodes from `root` to find the leftmost (first) leaf page. @@ -839,6 +1035,218 @@ mod tests { ) } + /// Deterministic xorshift64 so randomized tests are reproducible without deps. + struct Rng(u64); + impl Rng { + fn next(&mut self) -> u64 { + let mut x = self.0; + x ^= x << 13; + x ^= x >> 7; + x ^= x << 17; + self.0 = x; + x + } + } + + /// `n` EAVT entries with random entities; `counter` keeps keys unique. + fn random_eavt(rng: &mut Rng, n: usize, counter: &mut u64) -> Vec<(EavtKey, FactRef)> { + let mut v: Vec<(EavtKey, FactRef)> = (0..n) + .map(|_| { + *counter += 1; + let entity = (u128::from(rng.next()) << 64) | u128::from(*counter); + make_eavt(entity, ":attr", *counter) + }) + .collect(); + v.sort(); + v + } + + fn sorted_union(a: &[(EavtKey, FactRef)], b: &[(EavtKey, FactRef)]) -> Vec<(EavtKey, FactRef)> { + let mut v: Vec<_> = a.iter().chain(b.iter()).cloned().collect(); + v.sort(); + v + } + + /// Build `committed` at page 1, then rebuild with `pending` starting at `start`. + fn build_then_incremental( + committed: &[(EavtKey, FactRef)], + pending: &[(EavtKey, FactRef)], + start: u64, + ) -> (MemoryBackend, PageCache, u64, u64) { + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let ser = btree_entries(committed.iter().cloned()).unwrap(); + let (old_root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let leaves = collect_leaf_pages(old_root, &backend, &cache).unwrap(); + let (root, next_free) = + rebuild_btree_incremental(leaves, pending.to_vec(), &mut backend, &cache, start) + .unwrap(); + (backend, cache, root, next_free) + } + + /// Stream equality, a next_leaf chain of non-empty leaves, and range_scan + /// agreement on random bounds. + fn assert_tree_exact( + root: u64, + backend: &MemoryBackend, + cache: &PageCache, + expected: &[(EavtKey, FactRef)], + rng: &mut Rng, + ) { + let got: Vec<(EavtKey, FactRef)> = stream_all_entries(root, backend, cache).unwrap(); + assert_eq!(got.len(), expected.len(), "entry count differs"); + assert!(got == expected, "streamed entries differ from expected"); + + let leaves = collect_leaf_pages(root, backend, cache).unwrap(); + if expected.is_empty() { + assert_eq!(leaves.len(), 1, "empty tree is one empty leaf"); + } else { + for page in &leaves { + assert!(read_u16_at(&page[..], 2).unwrap() > 0, "empty leaf in tree"); + } + } + + for _ in 0..20 { + if expected.is_empty() { + break; + } + let a = (rng.next() as usize) % expected.len(); + let b = (rng.next() as usize) % expected.len(); + let (lo, hi) = (a.min(b), a.max(b)); + let start = &expected[lo].0; + let end = &expected[hi].0; + let want: Vec = expected[lo..hi].iter().map(|(_, r)| *r).collect(); + let refs = range_scan(root, start, Some(end), backend, cache).unwrap(); + assert!(refs == want, "range_scan differs from expected slice"); + } + } + + #[test] + fn test_incremental_no_pending_copies_leaves_verbatim() { + let mut rng = Rng(0x315); + let mut ctr = 0; + let committed = random_eavt(&mut rng, 2000, &mut ctr); + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let ser = btree_entries(committed.iter().cloned()).unwrap(); + let (old_root, old_next) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let old_leaves = collect_leaf_pages(old_root, &backend, &cache).unwrap(); + let (root, _) = rebuild_btree_incremental( + old_leaves.clone(), + Vec::<(EavtKey, FactRef)>::new(), + &mut backend, + &cache, + old_next, + ) + .unwrap(); + let new_leaves = collect_leaf_pages(root, &backend, &cache).unwrap(); + assert_eq!(new_leaves.len(), old_leaves.len(), "leaf count changed"); + for (o, n) in old_leaves.iter().zip(new_leaves.iter()) { + assert!(o[..4] == n[..4], "leaf header prefix changed"); + assert!(o[12..] == n[12..], "leaf body changed"); + } + assert_tree_exact(root, &backend, &cache, &committed, &mut rng); + } + + #[test] + fn test_incremental_matches_merge_random() { + let mut rng = Rng(0xC0FFEE); + let mut ctr = 0; + for _ in 0..40 { + let n_committed = (rng.next() % 3000) as usize; + let committed = random_eavt(&mut rng, n_committed, &mut ctr); + let n_pending = (rng.next() % 200) as usize; + let pending = random_eavt(&mut rng, n_pending, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + let expected = sorted_union(&committed, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + } + + #[test] + fn test_incremental_pending_outside_old_range() { + let mut rng = Rng(7); + let committed: Vec<_> = (1000u128..3000) + .map(|n| make_eavt(n, ":a", n as u64)) + .collect(); + let pending = vec![ + make_eavt(1, ":a", 1), + make_eavt(2, ":a", 2), + make_eavt(9_000, ":a", 9_000), + make_eavt(9_001, ":a", 9_001), + ]; + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + let expected = sorted_union(&committed, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + + #[test] + fn test_incremental_many_pending_into_one_leaf_splits() { + let mut rng = Rng(11); + // Committed entities are 0, 10_000, 20_000, …: every pending 0 < e < 10_000 + // routes to leaf 0, which must split into several leaves. + let committed: Vec<_> = (0u128..2000) + .map(|n| make_eavt(n * 10_000, ":a", n as u64 + 1)) + .collect(); + let pending: Vec<_> = (1u128..3000) + .map(|n| make_eavt(n, ":a", 100_000 + n as u64)) + .collect(); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 5000); + let expected = sorted_union(&committed, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + + #[test] + fn test_incremental_empty_old_tree_falls_back_to_bulk_build() { + let mut rng = Rng(13); + let mut ctr = 0; + let pending = random_eavt(&mut rng, 500, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&[], &pending, 5000); + assert_tree_exact(root, &backend, &cache, &pending, &mut rng); + // Nothing old, nothing new: still a valid single empty leaf. + let (backend, cache, root, next) = build_then_incremental(&[], &[], 5000); + assert_eq!(next, root + 1, "empty result is one page"); + assert_tree_exact(root, &backend, &cache, &[], &mut rng); + } + + #[test] + fn test_incremental_new_tree_overlaps_old_pages() { + // save() writes the new tree over the old tree's pages; old leaves are + // snapshotted first, so the result must still be exact. + let mut rng = Rng(17); + let mut ctr = 0; + let committed = random_eavt(&mut rng, 3000, &mut ctr); + let pending = random_eavt(&mut rng, 100, &mut ctr); + let (backend, cache, root, _) = build_then_incremental(&committed, &pending, 3); + let expected = sorted_union(&committed, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + + #[test] + fn test_incremental_repeated_rounds() { + // Each round consumes the previous round's incrementally built tree, + // writing the new tree over the old pages like save() does. + let mut rng = Rng(19); + let mut ctr = 0; + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let mut expected = random_eavt(&mut rng, 1500, &mut ctr); + let ser = btree_entries(expected.iter().cloned()).unwrap(); + let (mut root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + for round in 0..30u64 { + let n_pending = 1 + (rng.next() % 150) as usize; + let pending = random_eavt(&mut rng, n_pending, &mut ctr); + let leaves = collect_leaf_pages(root, &backend, &cache).unwrap(); + let start = 1 + round % 3; + let (r, _) = + rebuild_btree_incremental(leaves, pending.clone(), &mut backend, &cache, start) + .unwrap(); + root = r; + expected = sorted_union(&expected, &pending); + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + } + } + #[test] fn test_read_u16_at_oob_rejected() { let page = vec![0u8; 4]; From 20723a0acf5c8c8a8aa86f4b0140c667f7c590bb Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 12:01:06 +0530 Subject: [PATCH 7/9] perf: checkpoint copies untouched index leaves and skips re-hashing fact pages (#315) save() snapshots each index's raw leaf pages and rebuilds via rebuild_btree_incremental: leaves no pending entry touches are copied verbatim, touched leaves split balanced so repeated checkpoints do not fragment the tree. The CRC32 state over unchanged fact pages is cached so each save hashes only new fact pages and the index region. File format and stored checksum are unchanged. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- ...026-09-26-incremental-checkpoint-design.md | 17 +- src/storage/btree_v6.rs | 73 +++- src/storage/persistent_facts.rs | 336 +++++++++++++++--- 3 files changed, 354 insertions(+), 72 deletions(-) diff --git a/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md b/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md index a758b23f6..5c6bf4fa0 100644 --- a/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md +++ b/docs/superpowers/specs/2026-09-26-incremental-checkpoint-design.md @@ -70,8 +70,12 @@ Algorithm: 3. Leaf with no pending entries → **raw copy**: write the 4 KB page verbatim to the next page id, patching only bytes 4..12 (`next_leaf`). No decode/encode. 4. Leaf with pending entries → decode all its entries, merge with its pending - entries, serialise, and repack using `build_btree`'s 75% fill rule (may emit - more than one leaf). + entries and serialise. If the result fits in one physical page it stays one + leaf; otherwise it splits into pages of at most `PAGE_FILL_BYTES`, balanced by + size (classic B+tree split). *Revised during implementation:* repacking with + `build_btree`'s "fill to 75%, then start a new leaf" rule left a near-empty + leaf on every checkpoint that inserted into a full leaf (43 → 596 leaves after + 600 single-fact checkpoints), so cost drifted upward with checkpoint count. 5. Leaves are written in order with `next_leaf = pid + 1` (0 for the last), so there is no second patch pass. 6. Collect `(page_id, first_key_bytes)` per new leaf (postcard of the first key) and @@ -85,10 +89,11 @@ Collecting raw leaves: `collect_leaf_pages(root, backend, cache) -> Vec { } /// Write the last held leaf and return `(leaf_infos, next_free_page_id)`. + #[allow(clippy::type_complexity)] fn finish(mut self) -> Result<(Vec<(u64, Vec)>, u64)> { self.release_held(0)?; Ok((self.infos, self.next_page)) } } -/// Serialise sorted entries and emit them as leaves using `build_btree`'s fill rule. +/// Serialise sorted entries and emit them as leaves. +/// +/// Entries that fit in one physical page stay in one leaf. Otherwise they are +/// split into pages of at most `PAGE_FILL_BYTES`, balanced by size, so a leaf that +/// overflows by one entry splits roughly in half. Splitting "full page + remainder" +/// instead left a near-empty leaf behind on every checkpoint that inserted into a +/// full leaf, growing the index without bound (#315). #[allow(clippy::arithmetic_side_effects)] fn emit_packed( emitter: &mut LeafEmitter<'_>, entries: impl Iterator, ) -> Result<()> { - let mut cur: Vec> = Vec::new(); - let mut cur_bytes = 0usize; - let mut cur_first: Option> = None; + // Each entry costs its bytes plus one slot-directory entry. + let mut serialised: Vec<(Vec, K)> = Vec::new(); + let mut total = 0usize; for (key, fact_ref) in entries { let entry = postcard::to_allocvec(&(&key, &fact_ref))?; - if !cur.is_empty() && leaf_overflows(cur.len(), cur_bytes, entry.len()) { + total += entry.len() + SLOT_SIZE; + serialised.push((entry, key)); + } + if serialised.is_empty() { + return Ok(()); + } + let target = if LEAF_HEADER_SIZE + total <= PAGE_SIZE { + total + } else { + let usable = PAGE_FILL_BYTES - LEAF_HEADER_SIZE; + total.div_ceil(total.div_ceil(usable)) + }; + + let mut cur: Vec> = Vec::new(); + let mut cur_size = 0usize; + let mut cur_first: Option> = None; + for (entry, key) in serialised { + let size = entry.len() + SLOT_SIZE; + if !cur.is_empty() && cur_size + size > target { let first = cur_first.take().ok_or_else(|| { err_coded!(ErrorCode::Int049, "BUG: leaf without first key".to_string()) })?; emitter.push(encode_leaf_page(&cur, 0)?, first)?; cur.clear(); - cur_bytes = 0; + cur_size = 0; } if cur_first.is_none() { cur_first = Some(postcard::to_allocvec(&key)?); } - cur_bytes += entry.len(); + cur_size += size; cur.push(entry); } if let Some(first) = cur_first { @@ -780,6 +805,10 @@ where // ─── stream_all_entries ─────────────────────────────────────────────────────── /// Stream all `(K, FactRef)` entries from a B+tree in sorted order. +/// +/// Test-only since #315: `save()` snapshots raw leaves via [`collect_leaf_pages`] +/// instead of decoding every entry. +#[cfg(test)] pub fn stream_all_entries( root_page_id: u64, backend: &dyn StorageBackend, @@ -1222,6 +1251,36 @@ mod tests { assert_tree_exact(root, &backend, &cache, &expected, &mut rng); } + #[test] + fn test_incremental_single_inserts_do_not_fragment_leaves() { + // One random insert per round into a bulk-built (75%-full) tree, like one + // checkpoint per new fact. Splitting "75% + remainder" left a near-empty + // leaf per insert; leaves must stay reasonably full instead. + let mut rng = Rng(23); + let mut ctr = 0; + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let mut expected = random_eavt(&mut rng, 3000, &mut ctr); + let ser = btree_entries(expected.iter().cloned()).unwrap(); + let (mut root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let bulk_leaves = collect_leaf_pages(root, &backend, &cache).unwrap().len(); + for _ in 0..600 { + let pending = random_eavt(&mut rng, 1, &mut ctr); + let leaves = collect_leaf_pages(root, &backend, &cache).unwrap(); + root = rebuild_btree_incremental(leaves, pending.clone(), &mut backend, &cache, 1) + .unwrap() + .0; + expected = sorted_union(&expected, &pending); + } + assert_tree_exact(root, &backend, &cache, &expected, &mut rng); + // 20% more entries; at >= ~37% fill that is at most ~2.5x the bulk leaf count. + let leaves = collect_leaf_pages(root, &backend, &cache).unwrap().len(); + assert!( + leaves <= bulk_leaves * 5 / 2, + "leaf count grew from {bulk_leaves} to {leaves}" + ); + } + #[test] fn test_incremental_repeated_rounds() { // Each round consumes the previous round's incrementally built tree, diff --git a/src/storage/persistent_facts.rs b/src/storage/persistent_facts.rs index 2a057513f..d7147375d 100644 --- a/src/storage/persistent_facts.rs +++ b/src/storage/persistent_facts.rs @@ -8,8 +8,8 @@ use crate::graph::types::Fact; use crate::storage::FACT_PAGE_FORMAT_PACKED; use crate::storage::btree::{read_aevt_index, read_avet_index, read_eavt_index, read_vaet_index}; use crate::storage::btree_v6::{ - MutexStorageBackend, OnDiskIndexReader, btree_entries, build_btree, merge_sorted_vecs, - stream_all_entries, + MutexStorageBackend, OnDiskIndexReader, btree_entries, build_btree, collect_leaf_pages, + rebuild_btree_incremental, }; use crate::storage::cache::PageCache; use crate::storage::index::{AevtKey, AvetKey, EavtKey, FactRef, VaetKey, encode_value}; @@ -149,6 +149,10 @@ pub struct PersistentFactStorage { dirty: bool, last_checkpointed_tx_count: u64, committed_fact_pages: Arc, + /// CRC32 state after hashing fact pages `1..=n` (`n` is the first element). + /// Fact pages are never rewritten, so `save()` extends this instead of + /// re-reading them; `None` (or a stale `n`) falls back to a full pass (#315). + fact_prefix_crc: Option<(u64, Hasher)>, } impl PersistentFactStorage { @@ -170,6 +174,7 @@ impl PersistentFactStorage { dirty: false, last_checkpointed_tx_count: 0, committed_fact_pages, + fact_prefix_crc: None, }; // Try to load existing data. @@ -288,8 +293,26 @@ impl PersistentFactStorage { let stored = header.index_checksum; // Total data pages: pages 1 through page_count-1 (everything except header) let total_data_pages = header.page_count.saturating_sub(1); - let full_checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + // Hash fact pages, keep that state for save() (#315), then the index pages. + let (full_checksum, fact_prefix) = if total_data_pages >= num_fact_pages { + let mut hasher = Hasher::new(); + hash_pages(&*backend, &mut hasher, 1, num_fact_pages)?; + let prefix = hasher.clone(); + let index_start = num_fact_pages + .checked_add(1) + .ok_or_else(|| err_coded!(ErrorCode::Stg021))?; + hash_pages( + &*backend, + &mut hasher, + index_start, + total_data_pages.saturating_sub(num_fact_pages), + )?; + (hasher.finalize(), Some(prefix)) + } else { + (compute_page_checksum(&*backend, 1, total_data_pages)?, None) + }; if full_checksum == stored { + self.fact_prefix_crc = fact_prefix.map(|p| (num_fact_pages, p)); false // new-style checksum matches: facts + indexes verified } else { // Fall back: old files stored checksum over fact pages only @@ -863,27 +886,18 @@ impl PersistentFactStorage { Err(e) => bail_coded!(ErrorCode::Stg010, e), }; - // Stream committed B+tree entries BEFORE writing new pages that may overlap - let committed_eavt: Vec<(EavtKey, FactRef)> = if curr_header.eavt_root_page != 0 { - stream_all_entries(curr_header.eavt_root_page, &*backend, &self.page_cache)? - } else { - Vec::new() - }; - let committed_aevt: Vec<(AevtKey, FactRef)> = if curr_header.aevt_root_page != 0 { - stream_all_entries(curr_header.aevt_root_page, &*backend, &self.page_cache)? - } else { - Vec::new() - }; - let committed_avet: Vec<(AvetKey, FactRef)> = if curr_header.avet_root_page != 0 { - stream_all_entries(curr_header.avet_root_page, &*backend, &self.page_cache)? - } else { - Vec::new() - }; - let committed_vaet: Vec<(VaetKey, FactRef)> = if curr_header.vaet_root_page != 0 { - stream_all_entries(curr_header.vaet_root_page, &*backend, &self.page_cache)? - } else { - Vec::new() + // Snapshot old B+tree leaves BEFORE writing new pages that may overlap them. + let old_leaves = |root: u64| -> Result>>> { + if root == 0 { + Ok(Vec::new()) + } else { + collect_leaf_pages(root, &*backend, &self.page_cache) + } }; + let old_eavt = old_leaves(curr_header.eavt_root_page)?; + let old_aevt = old_leaves(curr_header.aevt_root_page)?; + let old_avet = old_leaves(curr_header.avet_root_page)?; + let old_vaet = old_leaves(curr_header.vaet_root_page)?; // Invalidate cached pages that will be overwritten (old index pages) self.page_cache.invalidate_from(new_fact_start); @@ -917,41 +931,35 @@ impl PersistentFactStorage { .checked_add(new_total_fact_pages) .ok_or_else(|| err_coded!(ErrorCode::Stg017))?; - let eavt_ser = if !committed_eavt.is_empty() { - btree_entries(merge_sorted_vecs(committed_eavt, pending_eavt))? - } else { - btree_entries(pending_eavt.into_iter())? - }; - let (eavt_root, next1) = build_btree( - eavt_ser.into_iter(), + // Copy untouched leaves, repack only leaves that receive pending entries (#315). + let (eavt_root, next1) = rebuild_btree_incremental( + old_eavt, + pending_eavt, &mut *backend, &self.page_cache, index_start, )?; - - let aevt_ser = if !committed_aevt.is_empty() { - btree_entries(merge_sorted_vecs(committed_aevt, pending_aevt))? - } else { - btree_entries(pending_aevt.into_iter())? - }; - let (aevt_root, next2) = - build_btree(aevt_ser.into_iter(), &mut *backend, &self.page_cache, next1)?; - - let avet_ser = if !committed_avet.is_empty() { - btree_entries(merge_sorted_vecs(committed_avet, pending_avet))? - } else { - btree_entries(pending_avet.into_iter())? - }; - let (avet_root, next3) = - build_btree(avet_ser.into_iter(), &mut *backend, &self.page_cache, next2)?; - - let vaet_ser = if !committed_vaet.is_empty() { - btree_entries(merge_sorted_vecs(committed_vaet, pending_vaet))? - } else { - btree_entries(pending_vaet.into_iter())? - }; - let (vaet_root, next4) = - build_btree(vaet_ser.into_iter(), &mut *backend, &self.page_cache, next3)?; + let (aevt_root, next2) = rebuild_btree_incremental( + old_aevt, + pending_aevt, + &mut *backend, + &self.page_cache, + next1, + )?; + let (avet_root, next3) = rebuild_btree_incremental( + old_avet, + pending_avet, + &mut *backend, + &self.page_cache, + next2, + )?; + let (vaet_root, next4) = rebuild_btree_incremental( + old_vaet, + pending_vaet, + &mut *backend, + &self.page_cache, + next3, + )?; // Sync index pages to disk before writing the header. // The header update is the atomic commit point: once it's durable, @@ -961,8 +969,29 @@ impl PersistentFactStorage { // CRC32 over ALL data pages (facts + indexes), excluding page 0 (header). // This detects corruption in both fact pages and B+tree index pages. - let total_data_pages = next4.saturating_sub(1); - let checksum = compute_page_checksum(&*backend, 1, total_data_pages)?; + // Fact pages 1..=old count are unchanged since the last save/load, so extend + // the cached CRC state instead of re-reading them (#315). Taking the cache + // means a failed save leaves None and the next save does a full pass. + let mut hasher = match self.fact_prefix_crc.take() { + Some((n, prefix)) if n == old_fact_page_count => { + let mut h = prefix; + hash_pages(&*backend, &mut h, new_fact_start, new_pages_len)?; + h + } + _ => { + let mut h = Hasher::new(); + hash_pages(&*backend, &mut h, 1, new_total_fact_pages)?; + h + } + }; + let new_prefix = hasher.clone(); + hash_pages( + &*backend, + &mut hasher, + index_start, + next4.saturating_sub(index_start), + )?; + let checksum = hasher.finalize(); // ── Step E: write header (last write = crash-safe boundary) ───────────── let mut header = FileHeader::new(); // version=7 @@ -991,6 +1020,7 @@ impl PersistentFactStorage { self.committed_fact_pages .store(new_total_fact_pages, Ordering::SeqCst); + self.fact_prefix_crc = Some((new_total_fact_pages, new_prefix)); self.last_checkpointed_tx_count = self.storage.current_tx_count(); self.dirty = false; @@ -1105,14 +1135,24 @@ fn compute_page_checksum( num_pages: u64, ) -> Result { let mut hasher = Hasher::new(); + hash_pages(backend, &mut hasher, first_page, num_pages)?; + Ok(hasher.finalize()) +} + +/// Feed pages `first_page..first_page + num_pages` into `hasher`. +fn hash_pages( + backend: &dyn StorageBackend, + hasher: &mut Hasher, + first_page: u64, + num_pages: u64, +) -> Result<()> { for i in 0..num_pages { let page_id = first_page .checked_add(i) .ok_or_else(|| err_coded!(ErrorCode::Stg021))?; - let page = backend.read_page(page_id)?; - hasher.update(&page); + hasher.update(&backend.read_page(page_id)?); } - Ok(hasher.finalize()) + Ok(()) } /// Compute CRC32 checksum over header bytes 0-79 (header_checksum field zeroed). @@ -1265,9 +1305,187 @@ mod tests { use super::*; use crate::graph::types::Value; use crate::storage::backend::MemoryBackend; + use crate::storage::btree_v6::stream_all_entries; use std::io::Write; use uuid::Uuid; + /// All four on-disk indexes equal a from-scratch derivation from the fact pages, + /// and the stored checksum equals a full pass over pages 1..page_count. + fn assert_indexes_and_checksum_exact( + pfs: &PersistentFactStorage, + ) { + let backend = pfs.backend.lock().unwrap(); + let header = FileHeader::from_bytes(&backend.read_page(0).unwrap()).unwrap(); + let full = compute_page_checksum(&*backend, 1, header.page_count - 1).unwrap(); + assert_eq!(header.index_checksum, full, "stored checksum != full pass"); + + let (facts, refs) = + crate::storage::packed_pages::read_all_with_refs(&*backend, 1, header.fact_page_count) + .unwrap(); + let (eavt, aevt, avet, vaet) = build_sorted_index_entries(&facts, &refs); + let cache = &pfs.page_cache; + let got_eavt: Vec<(EavtKey, FactRef)> = + stream_all_entries(header.eavt_root_page, &*backend, cache).unwrap(); + let got_aevt: Vec<(AevtKey, FactRef)> = + stream_all_entries(header.aevt_root_page, &*backend, cache).unwrap(); + let got_avet: Vec<(AvetKey, FactRef)> = + stream_all_entries(header.avet_root_page, &*backend, cache).unwrap(); + let got_vaet: Vec<(VaetKey, FactRef)> = + stream_all_entries(header.vaet_root_page, &*backend, cache).unwrap(); + assert!( + got_eavt.is_sorted_by(|a, b| a.0 <= b.0), + "EAVT not in key order" + ); + assert!( + got_aevt.is_sorted_by(|a, b| a.0 <= b.0), + "AEVT not in key order" + ); + assert!( + got_avet.is_sorted_by(|a, b| a.0 <= b.0), + "AVET not in key order" + ); + assert!( + got_vaet.is_sorted_by(|a, b| a.0 <= b.0), + "VAET not in key order" + ); + // One transaction can write the same (entity, attribute) twice, giving equal + // v7 keys (#371); their relative order is unspecified, so compare as sets. + fn by_key_then_ref(mut v: Vec<(K, FactRef)>) -> Vec<(K, FactRef)> { + v.sort(); + v + } + let (eavt, aevt, avet, vaet) = ( + by_key_then_ref(eavt), + by_key_then_ref(aevt), + by_key_then_ref(avet), + by_key_then_ref(vaet), + ); + let (got_eavt, got_aevt, got_avet, got_vaet) = ( + by_key_then_ref(got_eavt), + by_key_then_ref(got_aevt), + by_key_then_ref(got_avet), + by_key_then_ref(got_vaet), + ); + assert!(got_eavt == eavt, "EAVT differs from full derivation"); + assert!(got_aevt == aevt, "AEVT differs from full derivation"); + assert!(got_avet == avet, "AVET differs from full derivation"); + assert!(got_vaet == vaet, "VAET differs from full derivation"); + } + + /// One transact of `n` facts: new and reused entities, strings and refs (VAET). + fn transact_mixed( + pfs: &mut PersistentFactStorage, + entities: &mut Vec, + n: usize, + seed: u64, + ) { + let mut batch = Vec::new(); + for i in 0..n { + let k = seed.wrapping_mul(31).wrapping_add(i as u64); + let e = if entities.is_empty() || k % 3 == 0 { + let e = Uuid::new_v4(); + entities.push(e); + e + } else { + entities[(k as usize) % entities.len()] + }; + let value = if k % 4 == 0 { + Value::Ref(entities[(k as usize / 4) % entities.len()]) + } else { + Value::String(format!("v{k}")) + }; + batch.push((e, format!(":a{}", k % 5), value)); + } + pfs.storage().transact(batch, None).unwrap(); + pfs.mark_dirty(); + } + + #[test] + fn test_incremental_saves_keep_indexes_and_checksum_exact() { + let mut pfs = PersistentFactStorage::new(MemoryBackend::new(), 256).unwrap(); + let mut entities = Vec::new(); + for round in 0..25u64 { + let n = 1 + ((round * 37) % 60) as usize; + transact_mixed(&mut pfs, &mut entities, n, round); + pfs.save().unwrap(); + assert!(pfs.fact_prefix_crc.is_some(), "prefix cached after save"); + assert_indexes_and_checksum_exact(&pfs); + } + } + + #[test] + fn test_reopen_after_incremental_saves_takes_no_rebuild() { + use crate::storage::backend::FileBackend; + use tempfile::NamedTempFile; + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + let mut entities = Vec::new(); + { + let mut pfs = + PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + for round in 0..10u64 { + transact_mixed(&mut pfs, &mut entities, 40, round); + pfs.save().unwrap(); + } + } + let mut pfs = PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + // Set only when the full checksum matched on load, i.e. no rebuild. + assert!( + pfs.fact_prefix_crc.is_some(), + "full checksum must match on reopen" + ); + transact_mixed(&mut pfs, &mut entities, 40, 99); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } + + #[test] + fn test_prefix_cache_page_count_mismatch_falls_back() { + let mut pfs = PersistentFactStorage::new(MemoryBackend::new(), 256).unwrap(); + let mut entities = Vec::new(); + transact_mixed(&mut pfs, &mut entities, 50, 1); + pfs.save().unwrap(); + pfs.fact_prefix_crc = Some((999, Hasher::new())); + transact_mixed(&mut pfs, &mut entities, 50, 2); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } + + #[test] + fn test_prefix_cache_absent_after_rebuild_on_open() { + use crate::storage::backend::FileBackend; + use tempfile::NamedTempFile; + let tmp = NamedTempFile::new().unwrap(); + let path = tmp.path().to_str().unwrap().to_string(); + let mut entities = Vec::new(); + { + let mut pfs = + PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + for round in 0..3u64 { + transact_mixed(&mut pfs, &mut entities, 30, round); + pfs.save().unwrap(); + } + } + // Corrupt index_checksum (re-sealing the header) to force the rebuild path. + { + let mut backend = FileBackend::open(&path).unwrap(); + let mut page = backend.read_page(0).unwrap(); + page[64] ^= 0xFF; + let cs = compute_header_checksum_from_bytes(&page); + page[80..84].copy_from_slice(&cs.to_le_bytes()); + backend.write_page(0, &page).unwrap(); + backend.sync().unwrap(); + } + let mut pfs = PersistentFactStorage::new(FileBackend::open(&path).unwrap(), 256).unwrap(); + assert!( + pfs.fact_prefix_crc.is_none(), + "no prefix after rebuild-on-open" + ); + transact_mixed(&mut pfs, &mut entities, 30, 7); + pfs.save().unwrap(); + assert_indexes_and_checksum_exact(&pfs); + } + #[test] fn test_page_cache_capacity_reflects_constructed_value() { let zero_cap = PersistentFactStorage::new(MemoryBackend::new(), 0).unwrap(); From 7ebe1dab4db62f9ebc5d77b77e3e49eda87b8dba Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 12:01:50 +0530 Subject: [PATCH 8/9] docs: checkpoint cost and durability semantics; #315 benchmarks and doc sync Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- CHANGELOG.md | 3 +++ CLAUDE.md | 4 ++-- README.md | 2 +- docs/BENCHMARKS.md | 13 +++++++++++++ docs/TEST_COVERAGE.md | 4 ++-- src/db.rs | 24 +++++++++++++++++------- 6 files changed, 38 insertions(+), 12 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 4cb3be676..8df72dd5e 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Bound-entity point queries no longer pay for other attributes' history (#323).** `[:e :attr ?v]` now range-scans only `(e, :attr)` in the EAVT index instead of every record the entity has ever written. Reading a rarely changed attribute of a heavily rewritten entity no longer slows down as that entity's history grows: with 2,000 retract/reassert cycles on another attribute of the same entity, it drops from 4.58 ms to 19.0 µs. Reading the heavily rewritten attribute itself, and attribute scans (`[?e :attr ?v]`), are about 1.35–1.4× faster (4.64 ms → 3.36 ms), from cheaper net-assert grouping and removing a redundant dedup pass. The file format is unchanged. - **Attribute scans are bounded to exactly the queried attribute (#381).** `[?e :attr ?v]` computed the end of its AEVT range by incrementing the attribute's last byte. For attributes whose last byte is `0x7F` or `0xBF` — including about 1 in 64 non-ASCII characters, such as `:丿` or `:ÿ` — the result was not valid UTF-8, so the scan ran to the end of the whole AEVT index and read every later fact before filtering. It also read facts for prefix siblings (`:ab`, `:a/b` when scanning `:a`). The range now ends at `attribute + "\0"`, which covers exactly one attribute. Scanning 100 `:丿` facts in a checkpointed file with 40,000 facts on neighbouring attributes drops from 45.4 ms to 251 µs. Results are unchanged. +- **Checkpoints no longer re-encode the whole index (#315).** `checkpoint()` decoded and re-serialised every entry of all four covering indexes and re-read every page of the file for its checksum, so checkpointing after one new fact cost almost as much as after thousands. Index leaves that receive no new entries are now copied verbatim, only the leaves that do are decoded and re-split, and unchanged fact pages are no longer re-hashed. Checkpoint after one new fact drops from 246 ms to 25.7 ms on a 100k-fact file and from 23.9 ms to 3.9 ms at 10k. Cost still grows with graph size, because the index pages are copied on every checkpoint; the file format is unchanged. The `checkpoint()` and `wal_checkpoint_threshold` docs now state that a checkpoint is not a durability boundary: the WAL is already crash-durable. + ### Fixed - **Bound-entity queries could drop rows when one `WriteTransaction` wrote the same attribute in several valid-time windows (#323).** Both facts share a `tx_count`, and the selective lookup path de-duplicated on `(entity, attribute, tx_count, asserted)`, so one window's value was silently lost, while the same query via a full scan returned both. That de-duplication has been removed, and bound-entity queries now always match a full scan. @@ -19,6 +21,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ### Notes - On v2.x, reading a heavily rewritten attribute still costs time proportional to its history, because v7 index keys carry neither the value nor the assert/retract flag; the structural fix needs the v8 keys and is tracked for v3.0.0 in #379. +- Checkpoints on v2.x still copy every index page, so their cost grows with graph size. Copy-on-write index pages on the v3 branch, which also make `save()` crash-atomic, are needed for checkpoints proportional to the change alone (#374). ## v2.0.1 — 2026-09-25 diff --git a/CLAUDE.md b/CLAUDE.md index 78c9a8578..aab3b4dab 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -108,7 +108,7 @@ cargo run < demos/demo_negation.txt - `backend/file.rs`: Single `.graph` file backend (4KB pages, cross-platform) - `backend/memory.rs`: In-memory backend for testing - `index.rs`: EAVT / AEVT / AVET / VAET key types, `FactRef`, `encode_value` - - `btree_v6.rs`: On-disk B+tree (`build_btree`, `OnDiskIndexReader`, `MutexStorageBackend`) + - `btree_v6.rs`: On-disk B+tree (`build_btree`, `rebuild_btree_incremental`, `OnDiskIndexReader`, `MutexStorageBackend`) - `btree.rs`: Legacy v5 B+tree (migration only) - `cache.rs`: LRU page cache (`PageCache`, default 256 pages) - `packed_pages.rs`: Packed fact pages (~25 facts/4KB page), `MAX_FACT_BYTES` @@ -167,7 +167,7 @@ Auto-migrates v1/v2/v3/v4/v5/v6 → v7 on open/checkpoint. ## Test Coverage -**1176 tests passing** (1168 passing, 8 ignored; unit + integration + doc). +**1188 tests passing** (1180 passing, 8 ignored; unit + integration + doc). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/README.md b/README.md index 874758f15..4fa6eac65 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit ```bash cargo run # interactive Datalog REPL -cargo test # run 1176 tests +cargo test # run 1188 tests cargo run < demos/demo_recursive.txt # recursive rules demo ``` diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index 8b13948d9..e6ba817e1 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -50,6 +50,19 @@ One entity's `:hash` is retracted and reasserted `depth` times (exactly one live The churned attribute still scales with its own history on v2.x; see #379 for the v3.0.0 fix. +### Checkpoint After One Dirty Fact (#315) + +**Date**: 2026-09-26 · **Command**: `cargo bench --bench minigraf_bench -- checkpoint/after_1_fact` · same host as above; file on tmpfs, so fsync cost is excluded. + +An already-checkpointed file database of `n` facts receives one new fact, then `checkpoint()` runs; every iteration has exactly one dirty fact. "Before" is v2.0.1, which decoded and re-encoded every index entry. "After" copies index leaves that receive no new entries. + +| Facts | Before | After | +|---:|---:|---:| +| 10k | 23.9 ms | 3.94 ms | +| 100k | 246 ms | 25.7 ms | + +Cost still grows with graph size, because index pages are copied on every checkpoint. Checkpoints proportional to the change alone need copy-on-write pages, tracked with #374 for v3.0.0. + ## Measurement Method Criterion warms up each benchmark, collects ten samples for this query group, and estimates per-call latency from repeated iterations. The reported confidence intervals describe measurement uncertainty on this host; they are not cross-machine performance guarantees. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index beb7cb9c0..ef93bb001 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,9 +4,9 @@ **Verified**: 2026-09-26 with `cargo test` -**Result**: 1,168 passing tests and 8 ignored tests (1,176 total) +**Result**: 1,180 passing tests and 8 ignored tests (1,188 total) -The v2.0.1 suite covers the public database API, the Datalog engine, storage and recovery paths, and the v2.0.0 locking and structured-error-code changes, plus v2.0.1's regression test for rebuilding indexes on open after several checkpoints (#370). Since v2.0.1, `tests/point_query_history_test.rs` and new storage unit tests check that attribute-narrowed bound-entity lookups return exactly what a full scan returns, and that `net_asserted_facts` matches its previous implementation on randomized histories (#323). Attribute-scan tests check that the AEVT range covers exactly the queried attribute, including non-ASCII names such as `:丿` and prefix siblings such as `:ab` for `:a` (#381). +The v2.0.1 suite covers the public database API, the Datalog engine, storage and recovery paths, and the v2.0.0 locking and structured-error-code changes, plus v2.0.1's regression test for rebuilding indexes on open after several checkpoints (#370). Since v2.0.1, `tests/point_query_history_test.rs` and new storage unit tests check that attribute-narrowed bound-entity lookups return exactly what a full scan returns, and that `net_asserted_facts` matches its previous implementation on randomized histories (#323). Attribute-scan tests check that the AEVT range covers exactly the queried attribute, including non-ASCII names such as `:丿` and prefix siblings such as `:ab` for `:a` (#381). Incremental B+tree rebuild tests check that checkpoints which copy untouched index leaves produce exactly the entries and file checksum a full rebuild would, including over repeated checkpoints, and that single-fact checkpoints do not fragment the index (#315). ### Covered Areas diff --git a/src/db.rs b/src/db.rs index 4d0a4c4b6..eb1dc6548 100644 --- a/src/db.rs +++ b/src/db.rs @@ -113,6 +113,10 @@ pub struct OpenOptions { /// /// Defaults to 1000. Lower values mean more frequent checkpoints (smaller WAL, /// more I/O). Higher values mean less frequent checkpoints (larger WAL, less I/O). + /// + /// A checkpoint's cost grows with the total size of the database (see + /// [`Minigraf::checkpoint`]), while durability does not depend on it: the WAL + /// is crash-durable. Raise this for write-heavy workloads on large graphs. pub wal_checkpoint_threshold: usize, /// Number of pages to hold in the LRU page cache. Default: 256 (= 1MB at 4KB pages). /// @@ -684,6 +688,15 @@ impl Minigraf { /// Manually trigger a checkpoint: flush all in-memory facts to the main file /// and delete the WAL sidecar. /// + /// A checkpoint is **not** a durability boundary: committed writes are already + /// crash-durable in the `.wal` sidecar and are replayed on the next open. + /// Checkpointing less often only makes that replay (reopen) slower. + /// + /// Cost: every checkpoint rewrites the four covering indexes after the fact + /// pages, so it copies pages in proportion to the total index size; only the + /// index leaves that receive new entries are decoded and re-encoded. Prefer + /// checkpointing on a size or time budget over once per logical unit of work. + /// /// No-op for in-memory databases. /// /// # Errors @@ -733,13 +746,10 @@ impl Minigraf { // replayed WAL entries on open (crash-recovery path). `pfs.is_dirty()` // catches any facts marked dirty via the normal write path. // - // The kernel file lock taken by FileBackend::open covers both a - // second process and a second handle in this one, since flock - // and OFD locks attach to the open file description. This guard - // remains for environments where the filesystem cannot lock at all — - // for example, NFSv3 without lockd or some FUSE mounts — where the - // caller set `allow_unlocked` to accept that Minigraf cannot detect - // concurrent writers. + // Kernel file locking refuses a second writer (another process, or a + // second handle in this one — #304/#314). The guard still matters where + // the filesystem cannot lock (e.g. NFSv3 without lockd, some FUSE + // mounts) and the caller set `allow_unlocked`. if *wal_entry_count == 0 && !pfs.is_dirty() { return Ok(()); } From f2f9f69f583af4d47be8d770f080d8a841e90fa7 Mon Sep 17 00:00:00 2001 From: Aditya Mukhopadhyay Date: Sat, 26 Sep 2026 12:21:56 +0530 Subject: [PATCH 9/9] fix: reject cyclic leaf chains and damaged leaves before copying them (#315) Untouched leaves are now copied verbatim and sealed under a fresh file checksum, so collect_leaf_pages bounds-checks each leaf's slot directory and caps the chain at the file's page count instead of looping. Co-Authored-By: Claude Opus 5.5 Claude-Session: https://claude.ai/code/session_01BedF9ZmqVZLkvDwn91fuKy --- CHANGELOG.md | 2 +- CLAUDE.md | 2 +- README.md | 2 +- docs/BENCHMARKS.md | 4 +- docs/TEST_COVERAGE.md | 4 +- src/storage/btree_v6.rs | 99 +++++++++++++++++++++++++++++++++++++++++ 6 files changed, 106 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8df72dd5e..d479cb9ca 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,7 +12,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 - **Bound-entity point queries no longer pay for other attributes' history (#323).** `[:e :attr ?v]` now range-scans only `(e, :attr)` in the EAVT index instead of every record the entity has ever written. Reading a rarely changed attribute of a heavily rewritten entity no longer slows down as that entity's history grows: with 2,000 retract/reassert cycles on another attribute of the same entity, it drops from 4.58 ms to 19.0 µs. Reading the heavily rewritten attribute itself, and attribute scans (`[?e :attr ?v]`), are about 1.35–1.4× faster (4.64 ms → 3.36 ms), from cheaper net-assert grouping and removing a redundant dedup pass. The file format is unchanged. - **Attribute scans are bounded to exactly the queried attribute (#381).** `[?e :attr ?v]` computed the end of its AEVT range by incrementing the attribute's last byte. For attributes whose last byte is `0x7F` or `0xBF` — including about 1 in 64 non-ASCII characters, such as `:丿` or `:ÿ` — the result was not valid UTF-8, so the scan ran to the end of the whole AEVT index and read every later fact before filtering. It also read facts for prefix siblings (`:ab`, `:a/b` when scanning `:a`). The range now ends at `attribute + "\0"`, which covers exactly one attribute. Scanning 100 `:丿` facts in a checkpointed file with 40,000 facts on neighbouring attributes drops from 45.4 ms to 251 µs. Results are unchanged. -- **Checkpoints no longer re-encode the whole index (#315).** `checkpoint()` decoded and re-serialised every entry of all four covering indexes and re-read every page of the file for its checksum, so checkpointing after one new fact cost almost as much as after thousands. Index leaves that receive no new entries are now copied verbatim, only the leaves that do are decoded and re-split, and unchanged fact pages are no longer re-hashed. Checkpoint after one new fact drops from 246 ms to 25.7 ms on a 100k-fact file and from 23.9 ms to 3.9 ms at 10k. Cost still grows with graph size, because the index pages are copied on every checkpoint; the file format is unchanged. The `checkpoint()` and `wal_checkpoint_threshold` docs now state that a checkpoint is not a durability boundary: the WAL is already crash-durable. +- **Checkpoints no longer re-encode the whole index (#315).** `checkpoint()` decoded and re-serialised every entry of all four covering indexes and re-read every page of the file for its checksum, so checkpointing after one new fact cost almost as much as after thousands. Index leaves that receive no new entries are now copied verbatim, only the leaves that do are decoded and re-split, and unchanged fact pages are no longer re-hashed. Leaves are still bounds-checked and the leaf chain is checked for cycles, so a damaged index fails the checkpoint instead of being copied forward. Checkpoint after one new fact drops from 246 ms to 26.3 ms on a 100k-fact file and from 23.9 ms to 4.0 ms at 10k. Cost still grows with graph size, because the index pages are copied on every checkpoint; the file format is unchanged. The `checkpoint()` and `wal_checkpoint_threshold` docs now state that a checkpoint is not a durability boundary: the WAL is already crash-durable. ### Fixed diff --git a/CLAUDE.md b/CLAUDE.md index aab3b4dab..ff620b95b 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -167,7 +167,7 @@ Auto-migrates v1/v2/v3/v4/v5/v6 → v7 on open/checkpoint. ## Test Coverage -**1188 tests passing** (1180 passing, 8 ignored; unit + integration + doc). +**1190 tests passing** (1182 passing, 8 ignored; unit + integration + doc). See `docs/TEST_COVERAGE.md` for the full per-file breakdown. **Testing conventions** — see the Testing Conventions section below before writing any tests. diff --git a/README.md b/README.md index 4fa6eac65..aad4e86be 100644 --- a/README.md +++ b/README.md @@ -92,7 +92,7 @@ let r2 = pq.execute(&[("tx", BindValue::TxCount(2)), ("entity", BindValue::Entit ```bash cargo run # interactive Datalog REPL -cargo test # run 1188 tests +cargo test # run 1190 tests cargo run < demos/demo_recursive.txt # recursive rules demo ``` diff --git a/docs/BENCHMARKS.md b/docs/BENCHMARKS.md index e6ba817e1..d67494cba 100644 --- a/docs/BENCHMARKS.md +++ b/docs/BENCHMARKS.md @@ -58,8 +58,8 @@ An already-checkpointed file database of `n` facts receives one new fact, then ` | Facts | Before | After | |---:|---:|---:| -| 10k | 23.9 ms | 3.94 ms | -| 100k | 246 ms | 25.7 ms | +| 10k | 23.9 ms | 4.01 ms | +| 100k | 246 ms | 26.3 ms | Cost still grows with graph size, because index pages are copied on every checkpoint. Checkpoints proportional to the change alone need copy-on-write pages, tracked with #374 for v3.0.0. diff --git a/docs/TEST_COVERAGE.md b/docs/TEST_COVERAGE.md index ef93bb001..5850fcb4e 100644 --- a/docs/TEST_COVERAGE.md +++ b/docs/TEST_COVERAGE.md @@ -4,9 +4,9 @@ **Verified**: 2026-09-26 with `cargo test` -**Result**: 1,180 passing tests and 8 ignored tests (1,188 total) +**Result**: 1,182 passing tests and 8 ignored tests (1,190 total) -The v2.0.1 suite covers the public database API, the Datalog engine, storage and recovery paths, and the v2.0.0 locking and structured-error-code changes, plus v2.0.1's regression test for rebuilding indexes on open after several checkpoints (#370). Since v2.0.1, `tests/point_query_history_test.rs` and new storage unit tests check that attribute-narrowed bound-entity lookups return exactly what a full scan returns, and that `net_asserted_facts` matches its previous implementation on randomized histories (#323). Attribute-scan tests check that the AEVT range covers exactly the queried attribute, including non-ASCII names such as `:丿` and prefix siblings such as `:ab` for `:a` (#381). Incremental B+tree rebuild tests check that checkpoints which copy untouched index leaves produce exactly the entries and file checksum a full rebuild would, including over repeated checkpoints, and that single-fact checkpoints do not fragment the index (#315). +The v2.0.1 suite covers the public database API, the Datalog engine, storage and recovery paths, and the v2.0.0 locking and structured-error-code changes, plus v2.0.1's regression test for rebuilding indexes on open after several checkpoints (#370). Since v2.0.1, `tests/point_query_history_test.rs` and new storage unit tests check that attribute-narrowed bound-entity lookups return exactly what a full scan returns, and that `net_asserted_facts` matches its previous implementation on randomized histories (#323). Attribute-scan tests check that the AEVT range covers exactly the queried attribute, including non-ASCII names such as `:丿` and prefix siblings such as `:ab` for `:a` (#381). Incremental B+tree rebuild tests check that checkpoints which copy untouched index leaves produce exactly the entries and file checksum a full rebuild would, including over repeated checkpoints, that single-fact checkpoints do not fragment the index, and that a cyclic leaf chain or damaged slot directory is rejected rather than copied (#315). ### Covered Areas diff --git a/src/storage/btree_v6.rs b/src/storage/btree_v6.rs index 1e08ac122..684d0ca9a 100644 --- a/src/storage/btree_v6.rs +++ b/src/storage/btree_v6.rs @@ -461,6 +461,8 @@ pub fn collect_leaf_pages( backend: &dyn StorageBackend, cache: &PageCache, ) -> Result>>> { + // A chain longer than the file has pages must contain a cycle. + let max_leaves = backend.page_count()?; let mut leaves = Vec::new(); let mut leaf_id = find_leftmost_leaf(root_page_id, backend, cache)?; loop { @@ -471,6 +473,13 @@ pub fn collect_leaf_pages( format!("collect_leaf_pages: expected leaf page at page_id={leaf_id}") ); } + if u64::try_from(leaves.len()).map_or(true, |n| n >= max_leaves) { + bail_coded!( + ErrorCode::Int049, + format!("collect_leaf_pages: leaf chain longer than the file at page_id={leaf_id}") + ); + } + validate_leaf_slots(&page[..], leaf_id)?; let next_leaf = read_u64_at(&page[..], 4)?; leaves.push(page); if next_leaf == 0 { @@ -481,6 +490,34 @@ pub fn collect_leaf_pages( Ok(leaves) } +/// Check a leaf's slot directory stays inside the page without decoding entries. +/// +/// Untouched leaves are copied verbatim and sealed under a fresh file checksum, +/// so damage must be caught here instead of being carried forward. +#[allow(clippy::arithmetic_side_effects)] +fn validate_leaf_slots(page: &[u8], page_id: u64) -> Result<()> { + let count = read_u16_at(page, 2)? as usize; + let data_start = LEAF_HEADER_SIZE + count * SLOT_SIZE; + if data_start > page.len() { + bail_coded!( + ErrorCode::Int049, + format!("leaf page_id={page_id}: slot directory overflows the page") + ); + } + for i in 0..count { + let slot_off = LEAF_HEADER_SIZE + i * SLOT_SIZE; + let offset = read_u16_at(page, slot_off)? as usize; + let length = read_u16_at(page, slot_off + 2)? as usize; + if offset < data_start || offset + length > page.len() { + bail_coded!( + ErrorCode::Int049, + format!("leaf page_id={page_id}: slot {i} out of bounds") + ); + } + } + Ok(()) +} + /// Decode the key of a leaf's first entry; `None` for an empty leaf. #[allow(clippy::arithmetic_side_effects)] fn leaf_first_key(page: &[u8]) -> Result> @@ -1150,6 +1187,68 @@ mod tests { } } + /// Build a multi-leaf tree and return (backend, cache, root, leaf page ids). + fn multi_leaf_tree() -> (MemoryBackend, PageCache, u64, Vec) { + let mut backend = MemoryBackend::new(); + let cache = PageCache::new(4096); + let entries: Vec<_> = (0u128..500).map(|n| make_eavt(n, ":a", n as u64)).collect(); + let ser = btree_entries(entries.into_iter()).unwrap(); + let (root, _) = build_btree(ser.into_iter(), &mut backend, &cache, 1).unwrap(); + let mut ids = vec![find_leftmost_leaf(root, &backend, &cache).unwrap()]; + loop { + let page = cache.get_or_load(*ids.last().unwrap(), &backend).unwrap(); + let next = read_u64_at(&page[..], 4).unwrap(); + if next == 0 { + break; + } + ids.push(next); + } + assert!(ids.len() >= 3, "fixture needs several leaves"); + (backend, cache, root, ids) + } + + fn overwrite_page(backend: &mut MemoryBackend, cache: &PageCache, id: u64, page: Vec) { + backend.write_page(id, &page).unwrap(); + cache.put_dirty(id, page); + } + + #[test] + fn test_collect_leaf_pages_rejects_cyclic_chain() { + // A corrupt next_leaf pointing back must be an error, not an endless loop. + let (mut backend, cache, root, ids) = multi_leaf_tree(); + let mut page = cache.get_or_load(ids[1], &backend).unwrap().to_vec(); + page[4..12].copy_from_slice(&ids[0].to_le_bytes()); + overwrite_page(&mut backend, &cache, ids[1], page); + assert!( + collect_leaf_pages(root, &backend, &cache).is_err(), + "cyclic leaf chain must be rejected" + ); + } + + #[test] + fn test_collect_leaf_pages_rejects_corrupt_slot_directory() { + // Untouched leaves are copied without decoding, so a damaged slot must be + // caught here rather than sealed under a fresh checksum. + let (mut backend, cache, root, ids) = multi_leaf_tree(); + let mut page = cache.get_or_load(ids[1], &backend).unwrap().to_vec(); + let bad_offset = u16::try_from(PAGE_SIZE - 2).unwrap(); + page[LEAF_HEADER_SIZE..LEAF_HEADER_SIZE + 2].copy_from_slice(&bad_offset.to_le_bytes()); + overwrite_page(&mut backend, &cache, ids[1], page); + assert!( + collect_leaf_pages(root, &backend, &cache).is_err(), + "slot pointing past the page end must be rejected" + ); + + let (mut backend, cache, root, ids) = multi_leaf_tree(); + let mut page = cache.get_or_load(ids[2], &backend).unwrap().to_vec(); + page[2..4].copy_from_slice(&u16::MAX.to_le_bytes()); + overwrite_page(&mut backend, &cache, ids[2], page); + assert!( + collect_leaf_pages(root, &backend, &cache).is_err(), + "entry count overflowing the page must be rejected" + ); + } + #[test] fn test_incremental_no_pending_copies_leaves_verbatim() { let mut rng = Rng(0x315);