From eb99de102042888509726a665ea52c95d312f2f4 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 07:33:20 -0700 Subject: [PATCH 1/6] =?UTF-8?q?bench:=20search=20harness=20=E2=80=94=20HNS?= =?UTF-8?q?W=20recall=20vs=20speed,=20end-to-end=20hybrid=5Fsearch=20laten?= =?UTF-8?q?cy?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit New clawhdf5-bench binary `search_harness`, the measurement baseline for the search hot-path work. On deterministic clustered 384-dim data it reports HNSW build time and, per ef, recall@10 against an exact scan, QPS and p50/p99; and for HDF5Memory: ingest, checkpoint, open, first-query-after-open and steady-state hybrid_search latency at 1K/10K (and 100K with --full). Optional JSON output for tracking. Baseline recorded in BENCHMARKS.md. It shows two problems: HNSW recall@10 does not respond to ef and falls from 0.87 (1K) to 0.31 (100K) on clustered data, and end-to-end hybrid_search is ~1000x slower than its vector stage because every query rebuilds BM25 and rewrites the .h5 file. Co-Authored-By: Claude Fable 5.1 --- BENCHMARKS.md | 66 +++ crates/clawhdf5-bench/Cargo.toml | 5 + .../clawhdf5-bench/src/bin/search_harness.rs | 381 ++++++++++++++++++ 3 files changed, 452 insertions(+) create mode 100644 crates/clawhdf5-bench/src/bin/search_harness.rs diff --git a/BENCHMARKS.md b/BENCHMARKS.md index d7a0ed9..1f6b21e 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -28,6 +28,72 @@ --- +## Search harness baseline (v2.3.0) + +Produced by `cargo run --release -p clawhdf5-bench --bin search_harness -- --full` +on deterministic **clustered** synthetic data (384-dim, unit-normalised; points = +cluster centre + noise — uniform random vectors are nearly equidistant in high +dimension and say nothing about embeddings). Recall is measured against an exact +brute-force scan, 200 queries. This is the *before* picture for the search +hot-path work; every change to that path should be justified by a re-run. + +Two things stand out: + +* **HNSW recall does not respond to `ef`** and degrades sharply with size + (0.87 → 0.67 → 0.31 recall@10 at 1K / 10K / 100K). Latency plateaus at the same + point, i.e. the search exhausts the nodes it can reach: on clustered data the + graph is poorly connected. The index selects neighbours by plain top-M + distance rather than the HNSW paper's diversity heuristic. +* **End-to-end `hybrid_search` is ~1000x slower than its vector stage** (49 ms + vs ~0.03 ms at 10K; 884 ms at 100K). Each query rebuilds the BM25 index from + scratch and rewrites the whole `.h5` file. The first query after `open()` + additionally rebuilds the HNSW index (10.5 s at 100K). + +### HNSW, N = 1000, dim = 384, M = 16, ef_construction = 64 + +build: 72.4 ms (13818 vectors/s) · exact scan: 3854 QPS, p50 258 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.8710 | 59484 | 16 | 31 | +| 32 | 0.8730 | 46302 | 21 | 25 | +| 64 | 0.8730 | 31683 | 31 | 44 | +| 128 | 0.8730 | 24715 | 40 | 49 | +| 256 | 0.8730 | 24788 | 40 | 50 | + +### HNSW, N = 10000, dim = 384, M = 16, ef_construction = 64 + +build: 802.5 ms (12461 vectors/s) · exact scan: 418 QPS, p50 2363 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.6695 | 44031 | 19 | 51 | +| 32 | 0.6705 | 45066 | 22 | 30 | +| 64 | 0.6705 | 32746 | 30 | 41 | +| 128 | 0.6705 | 27542 | 36 | 51 | +| 256 | 0.6705 | 27754 | 36 | 49 | + +### HNSW, N = 100000, dim = 384, M = 16, ef_construction = 64 + +build: 9752.6 ms (10254 vectors/s) · exact scan: 40 QPS, p50 24648 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.3085 | 18046 | 57 | 84 | +| 32 | 0.3110 | 21621 | 43 | 75 | +| 64 | 0.3130 | 20015 | 49 | 70 | +| 128 | 0.3135 | 15822 | 63 | 99 | +| 256 | 0.3135 | 15308 | 66 | 124 | + +### End to end: `HDF5Memory::hybrid_search` (k = 10, weights 0.7 / 0.3) + +| N | ingest ms | checkpoint ms | open ms | first query ms | p50 ms | p99 ms | QPS | +|---:|---:|---:|---:|---:|---:|---:|---:| +| 1000 | 11 | 3.9 | 0.9 | 68.1 | 5.48 | 5.57 | 182.5 | +| 10000 | 114 | 32.2 | 10.9 | 845.0 | 48.56 | 78.65 | 19.8 | +| 100000 | 1486 | 713.0 | 354.5 | 10486.5 | 883.51 | 975.23 | 1.1 | +wrote /tmp/claude-1000/-home-osobh-projects-clawhdf5/422f755e-dd25-4c35-8613-5439087e3aaa/scratchpad/baseline_full.json + ## Vector Search Latency Brute-force cosine similarity over 384-dimensional embeddings (OpenAI text-embedding-3-small size). diff --git a/crates/clawhdf5-bench/Cargo.toml b/crates/clawhdf5-bench/Cargo.toml index 5cf059d..7f8ffbf 100644 --- a/crates/clawhdf5-bench/Cargo.toml +++ b/crates/clawhdf5-bench/Cargo.toml @@ -13,6 +13,10 @@ path = "src/bin/longmemeval_bench.rs" name = "memory_arena" path = "src/bin/memory_arena.rs" +[[bin]] +name = "search_harness" +path = "src/bin/search_harness.rs" + [[bin]] name = "footprint_bench" path = "src/bin/footprint_bench.rs" @@ -48,6 +52,7 @@ harness = false [dependencies] clawhdf5-agent = { path = "../clawhdf5-agent" } +clawhdf5-ann = { path = "../clawhdf5-ann" } clawhdf5-io = { path = "../clawhdf5-io" } mpi = { version = "0.8", optional = true } serde = { workspace = true } diff --git a/crates/clawhdf5-bench/src/bin/search_harness.rs b/crates/clawhdf5-bench/src/bin/search_harness.rs new file mode 100644 index 0000000..0db6931 --- /dev/null +++ b/crates/clawhdf5-bench/src/bin/search_harness.rs @@ -0,0 +1,381 @@ +//! Search measurement harness: recall vs. speed for the HNSW index, and +//! end-to-end `hybrid_search` latency as the store grows. +//! +//! Every search-path change should be justified by a before/after run of this +//! binary. It reports, for deterministic synthetic data: +//! +//! * **ANN** — index build time, and for each `ef`: recall@10 against an exact +//! brute-force scan, queries/second, and p50/p99 latency. +//! * **End to end** — `HDF5Memory`: ingest time, checkpoint time, `open()` +//! time, the first query after open (which pays for any index rebuild), and +//! steady-state `hybrid_search` p50/p99 at each store size. +//! +//! Data is *clustered* (points = cluster centre + noise, unit-normalised), not +//! uniform: uniform random high-dimensional vectors are nearly equidistant, +//! which makes recall numbers meaningless and is nothing like embeddings. +//! +//! ```text +//! cargo run --release -p clawhdf5-bench --bin search_harness # 1K, 10K +//! cargo run --release -p clawhdf5-bench --bin search_harness -- --full # + 100K +//! cargo run --release -p clawhdf5-bench --bin search_harness -- --json out.json +//! ``` + +use std::time::{Duration, Instant}; + +use clawhdf5_agent::{AgentMemory, HDF5Memory, MemoryConfig, MemoryEntry}; +use clawhdf5_ann::{DistanceMetric, HnswIndex}; + +const DIM: usize = 384; +const K: usize = 10; +const N_QUERIES: usize = 200; +const HNSW_M: usize = 16; +const HNSW_EF_CONSTRUCTION: usize = 64; +const EF_VALUES: [usize; 5] = [16, 32, 64, 128, 256]; + +// --------------------------------------------------------------------------- +// Deterministic data +// --------------------------------------------------------------------------- + +struct Rng(u64); + +impl Rng { + fn next_u64(&mut self) -> u64 { + self.0 = self.0.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = self.0; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + z ^ (z >> 31) + } + + /// Uniform in [0, 1). + fn unit(&mut self) -> f32 { + (self.next_u64() >> 40) as f32 / (1u64 << 24) as f32 + } + + /// Approximately standard normal (sum of uniforms). + fn gauss(&mut self) -> f32 { + let sum: f32 = (0..6).map(|_| self.unit()).sum(); + (sum - 3.0) * std::f32::consts::SQRT_2 + } + + fn below(&mut self, n: usize) -> usize { + (self.next_u64() % n as u64) as usize + } +} + +fn normalize(v: &mut [f32]) { + let norm = v.iter().map(|x| x * x).sum::().sqrt(); + if norm > 0.0 { + v.iter_mut().for_each(|x| *x /= norm); + } +} + +struct Dataset { + vectors: Vec>, + queries: Vec>, + /// Cluster id of each vector (used to give records topical text). + cluster_of: Vec, + query_cluster: Vec, +} + +fn make_dataset(n: usize, seed: u64) -> Dataset { + let mut rng = Rng(seed); + let n_clusters = (n / 100).clamp(8, 512); + let centres: Vec> = (0..n_clusters) + .map(|_| { + let mut c: Vec = (0..DIM).map(|_| rng.gauss()).collect(); + normalize(&mut c); + c + }) + .collect(); + let point = |rng: &mut Rng, cluster: usize| { + // Noise comparable to the centre's per-dimension magnitude, so + // clusters overlap and the nearest neighbours are non-trivial. + let scale = 0.6 / (DIM as f32).sqrt(); + let mut v: Vec = centres[cluster] + .iter() + .map(|c| c + rng.gauss() * scale) + .collect(); + normalize(&mut v); + v + }; + let mut vectors = Vec::with_capacity(n); + let mut cluster_of = Vec::with_capacity(n); + for _ in 0..n { + let c = rng.below(n_clusters); + vectors.push(point(&mut rng, c)); + cluster_of.push(c); + } + let mut queries = Vec::with_capacity(N_QUERIES); + let mut query_cluster = Vec::with_capacity(N_QUERIES); + for _ in 0..N_QUERIES { + let c = rng.below(n_clusters); + queries.push(point(&mut rng, c)); + query_cluster.push(c); + } + Dataset { + vectors, + queries, + cluster_of, + query_cluster, + } +} + +const WORDS: &[&str] = &[ + "deploy", "latency", "cache", "schema", "index", "vector", "memory", "agent", "kernel", + "buffer", "socket", "thread", "tensor", "gradient", "ledger", "invoice", "meeting", "roadmap", + "customer", "contract", "sensor", "orbit", "protein", "genome", "harbor", "bridge", "engine", + "battery", "harvest", "weather", "museum", "recipe", +]; + +/// Text whose vocabulary is biased by cluster, so keyword and vector signals +/// agree the way they do for real embedded text. +fn text_for(cluster: usize, i: usize, rng: &mut Rng) -> String { + let topic = [ + WORDS[cluster % WORDS.len()], + WORDS[(cluster / 7 + 3) % WORDS.len()], + ]; + let mut words = Vec::with_capacity(14); + for j in 0..14 { + if j % 3 == 0 { + words.push(topic[j / 3 % 2]); + } else { + words.push(WORDS[rng.below(WORDS.len())]); + } + } + format!("record {i}: {}", words.join(" ")) +} + +// --------------------------------------------------------------------------- +// Measurement helpers +// --------------------------------------------------------------------------- + +fn exact_top_k(vectors: &[Vec], query: &[f32], k: usize) -> Vec { + // Vectors are unit length, so cosine order == dot-product order. + let mut scored: Vec<(usize, f32)> = vectors + .iter() + .enumerate() + .map(|(i, v)| (i, v.iter().zip(query).map(|(a, b)| a * b).sum())) + .collect(); + scored.sort_by(|a, b| b.1.total_cmp(&a.1).then(a.0.cmp(&b.0))); + scored.truncate(k); + scored.into_iter().map(|(i, _)| i).collect() +} + +struct Latency { + p50: Duration, + p99: Duration, + qps: f64, +} + +fn summarize(mut samples: Vec) -> Latency { + samples.sort(); + let total: Duration = samples.iter().sum(); + let at = |q: f64| samples[((samples.len() - 1) as f64 * q).round() as usize]; + Latency { + p50: at(0.50), + p99: at(0.99), + qps: samples.len() as f64 / total.as_secs_f64(), + } +} + +fn micros(d: Duration) -> f64 { + d.as_secs_f64() * 1e6 +} + +fn millis(d: Duration) -> f64 { + d.as_secs_f64() * 1e3 +} + +// --------------------------------------------------------------------------- +// ANN: recall vs speed +// --------------------------------------------------------------------------- + +fn bench_ann(n: usize, json: &mut Vec) { + let data = make_dataset(n, 0xA11CE ^ n as u64); + let truth: Vec> = data + .queries + .iter() + .map(|q| exact_top_k(&data.vectors, q, K)) + .collect(); + + let started = Instant::now(); + let index = HnswIndex::build_with_metric( + &data.vectors, + HNSW_M, + HNSW_EF_CONSTRUCTION, + DistanceMetric::Cosine, + ); + let build = started.elapsed(); + + // Exact scan baseline, for scale. + let exact = summarize( + data.queries + .iter() + .map(|q| { + let t = Instant::now(); + std::hint::black_box(exact_top_k(&data.vectors, q, K)); + t.elapsed() + }) + .collect(), + ); + + println!( + "\n### HNSW, N = {n}, dim = {DIM}, M = {HNSW_M}, ef_construction = {HNSW_EF_CONSTRUCTION}\n" + ); + println!( + "build: {:.1} ms ({:.0} vectors/s) · exact scan: {:.0} QPS, p50 {:.0} µs\n", + millis(build), + n as f64 / build.as_secs_f64(), + exact.qps, + micros(exact.p50) + ); + println!("| ef | recall@{K} | QPS | p50 µs | p99 µs |"); + println!("|---:|---:|---:|---:|---:|"); + for ef in EF_VALUES { + let mut hits = 0usize; + let mut samples = Vec::with_capacity(data.queries.len()); + for (q, want) in data.queries.iter().zip(&truth) { + let t = Instant::now(); + let got = index.search(q, K, ef); + samples.push(t.elapsed()); + hits += got.iter().filter(|(id, _)| want.contains(id)).count(); + } + let recall = hits as f64 / (K * data.queries.len()) as f64; + let lat = summarize(samples); + println!( + "| {ef} | {recall:.4} | {:.0} | {:.0} | {:.0} |", + lat.qps, + micros(lat.p50), + micros(lat.p99) + ); + json.push(serde_json::json!({ + "bench": "hnsw", "n": n, "ef": ef, "recall_at_10": recall, + "qps": lat.qps, "p50_us": micros(lat.p50), "p99_us": micros(lat.p99), + "build_ms": millis(build), + })); + } +} + +// --------------------------------------------------------------------------- +// End to end: HDF5Memory::hybrid_search +// --------------------------------------------------------------------------- + +fn bench_end_to_end(n: usize, json: &mut Vec) { + let data = make_dataset(n, 0xE2E ^ n as u64); + let dir = tempfile::TempDir::new().unwrap(); + let path = dir.path().join("store.h5"); + let mut rng = Rng(7); + + let entries: Vec = data + .vectors + .iter() + .enumerate() + .map(|(i, v)| MemoryEntry { + chunk: text_for(data.cluster_of[i], i, &mut rng), + embedding: v.clone(), + source_channel: "bench".into(), + timestamp: i as f64, + session_id: format!("s{}", i % 50), + tags: format!("t{i}"), + }) + .collect(); + let query_texts: Vec = data + .query_cluster + .iter() + .enumerate() + .map(|(i, c)| text_for(*c, i, &mut rng)) + .collect(); + + let mut mem = HDF5Memory::create(MemoryConfig::new(path.clone(), "bench", DIM)).unwrap(); + let t = Instant::now(); + mem.save_batch(entries).unwrap(); + let ingest = t.elapsed(); + let t = Instant::now(); + mem.flush_wal().unwrap(); + let checkpoint = t.elapsed(); + drop(mem); + + let t = Instant::now(); + let mut mem = HDF5Memory::open(&path).unwrap(); + let open = t.elapsed(); + + // The first query after open pays for whatever is rebuilt lazily. + let t = Instant::now(); + std::hint::black_box(mem.hybrid_search(&data.queries[0], &query_texts[0], 0.7, 0.3, K)); + let first_query = t.elapsed(); + + // Fewer steady-state samples at large N: each query is currently O(N). + let samples_wanted = if n >= 100_000 { 20 } else { N_QUERIES.min(100) }; + let steady = summarize( + (0..samples_wanted) + .map(|i| { + let t = Instant::now(); + std::hint::black_box(mem.hybrid_search( + &data.queries[i % N_QUERIES], + &query_texts[i % N_QUERIES], + 0.7, + 0.3, + K, + )); + t.elapsed() + }) + .collect(), + ); + + println!( + "| {n} | {:.0} | {:.1} | {:.1} | {:.1} | {:.2} | {:.2} | {:.1} |", + millis(ingest), + millis(checkpoint), + millis(open), + millis(first_query), + millis(steady.p50), + millis(steady.p99), + steady.qps + ); + json.push(serde_json::json!({ + "bench": "hybrid_search", "n": n, + "ingest_ms": millis(ingest), "checkpoint_ms": millis(checkpoint), + "open_ms": millis(open), "first_query_ms": millis(first_query), + "p50_ms": millis(steady.p50), "p99_ms": millis(steady.p99), "qps": steady.qps, + })); +} + +fn main() { + let args: Vec = std::env::args().skip(1).collect(); + let full = args.iter().any(|a| a == "--full"); + let json_path = args + .iter() + .position(|a| a == "--json") + .and_then(|i| args.get(i + 1)) + .cloned(); + let sizes: &[usize] = if full { + &[1_000, 10_000, 100_000] + } else { + &[1_000, 10_000] + }; + + if cfg!(debug_assertions) { + eprintln!("warning: debug build — numbers are meaningless. Use --release."); + } + + let mut json = Vec::new(); + println!("## Search harness"); + for &n in sizes { + bench_ann(n, &mut json); + } + + println!("\n### End to end: `HDF5Memory::hybrid_search` (k = {K}, weights 0.7 / 0.3)\n"); + println!( + "| N | ingest ms | checkpoint ms | open ms | first query ms | p50 ms | p99 ms | QPS |" + ); + println!("|---:|---:|---:|---:|---:|---:|---:|---:|"); + for &n in sizes { + bench_end_to_end(n, &mut json); + } + + if let Some(path) = json_path { + std::fs::write(&path, serde_json::to_string_pretty(&json).unwrap()).unwrap(); + eprintln!("wrote {path}"); + } +} From 65d219c40920fa423b4978753c95e28f01c3a8e6 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 07:43:41 -0700 Subject: [PATCH 2/6] =?UTF-8?q?fix(ann):=20HNSW=20neighbour-selection=20he?= =?UTF-8?q?uristic=20=E2=80=94=20recall=200.31=20->=200.98=20at=20100K?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Neighbours were selected as the plain closest-M, for both a new node's links and back-link pruning. On clustered data every link of a node inside a tight cluster then goes to that same cluster, so clusters become islands a search entering elsewhere can never reach: recall@10 was 0.87 / 0.67 / 0.31 at 1K / 10K / 100K (384-dim) and flat in ef. Uniform random data — all the existing tests used — does not show it. Implement the HNSW paper's Algorithm 4 with keepPrunedConnections: accept a candidate only if it is closer to the node than to every neighbour already accepted, then fill spare slots with the closest rejected ones. Recall@10 at ef=64 is now 1.00 / 1.00 / 0.98 and rises with ef; uniform data improves slightly. Build is ~3.5x slower at 10K (extra distance evaluations), to be recovered by the distance-kernel work. The needless rayon fan-out over <=33 distances in prune_connections is gone. Tests: a clustered-data recall test for bulk build and incremental insert (scores 0.43 with the old selection), and a unit test of the selection rule. Harness gains --uniform and --ann-only; before/after in BENCHMARKS.md. Co-Authored-By: Claude Fable 5.1 --- CHANGELOG.md | 16 ++ crates/clawhdf5-ann/src/hnsw.rs | 167 ++++++++++++++++-- .../clawhdf5-bench/src/bin/search_harness.rs | 27 +++ 3 files changed, 194 insertions(+), 16 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 07adce5..cc08329 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,5 +1,21 @@ # Changelog +## Unreleased + +### Search +- `clawhdf5-ann`: **HNSW recall fix.** Neighbours were chosen as the plain + closest-M, which on clustered data (what embeddings look like) turns each + cluster into an island: recall@10 was 0.87 / 0.67 / 0.31 at 1K / 10K / 100K + vectors and did not improve with `ef`. The index now uses the HNSW paper's + diversity heuristic (Algorithm 4 with kept pruned connections) when linking a + new node and when pruning back-links: recall@10 at `ef = 64` is 1.00 / 1.00 / + 0.98 and responds to `ef`. Builds are slower (~3.5x at 10K). Existing + persisted indexes keep their old graph until rebuilt; the agent rebuilds its + index from the cache, so stores pick this up automatically. +- `clawhdf5-bench`: new `search_harness` binary — HNSW recall@10 / QPS / latency + per `ef` against an exact scan, and end-to-end `hybrid_search` timings, on + deterministic clustered (or `--uniform`) data. Baseline in `BENCHMARKS.md`. + ## v2.3.0 (2026-09-19) ### Upgrade Notes diff --git a/crates/clawhdf5-ann/src/hnsw.rs b/crates/clawhdf5-ann/src/hnsw.rs index c8ae49e..5755f2b 100644 --- a/crates/clawhdf5-ann/src/hnsw.rs +++ b/crates/clawhdf5-ann/src/hnsw.rs @@ -252,8 +252,9 @@ impl HnswIndex { metric, ); - // Select up to m closest neighbors - let selected: Vec = neighbors.iter().take(max_conn).map(|c| c.id).collect(); + let scored: Vec<(usize, f32)> = + neighbors.iter().map(|c| (c.id, c.distance)).collect(); + let selected = select_neighbors(vectors, &scored, max_conn, metric); // Add bidirectional connections graph[layer][i] = selected.clone(); @@ -382,7 +383,8 @@ impl HnswIndex { self.ef_construction, self.metric, ); - let selected: Vec = neighbors.iter().take(max_conn).map(|c| c.id).collect(); + let scored: Vec<(usize, f32)> = neighbors.iter().map(|c| (c.id, c.distance)).collect(); + let selected = select_neighbors(&self.vectors, &scored, max_conn, self.metric); self.graph[layer][id] = selected.clone(); for &neighbor in &selected { self.graph[layer][neighbor].push(id); @@ -828,7 +830,52 @@ fn search_layer( result } -/// Prune connections for a node to keep only the closest `max_conn` neighbors. +/// Choose up to `max_conn` neighbours for a node from `candidates` (sorted by +/// ascending distance to that node) — the HNSW paper's Algorithm 4 with +/// `keepPrunedConnections`. +/// +/// Taking the plain `max_conn` closest is what breaks the graph on clustered +/// data: every link of a node inside a tight cluster goes to that same cluster, +/// so clusters become islands that a search entering elsewhere can never +/// reach, however large `ef` is. Instead a candidate is accepted only if it is +/// closer to the node than to every neighbour already accepted, which spreads +/// links across directions and keeps the long edges that join clusters. Any +/// remaining slots are then filled with the closest rejected candidates, so a +/// node is never left under-connected. +fn select_neighbors( + vectors: &[Vec], + candidates: &[(usize, f32)], + max_conn: usize, + metric: DistanceMetric, +) -> Vec { + if candidates.len() <= max_conn { + return candidates.iter().map(|&(id, _)| id).collect(); + } + let mut selected: Vec = Vec::with_capacity(max_conn); + let mut rejected: Vec = Vec::new(); + for &(id, dist_to_node) in candidates { + if selected.len() >= max_conn { + break; + } + let diverse = selected + .iter() + .all(|&s| compute_distance(&vectors[id], &vectors[s], metric) > dist_to_node); + if diverse { + selected.push(id); + } else { + rejected.push(id); + } + } + for id in rejected { + if selected.len() >= max_conn { + break; + } + selected.push(id); + } + selected +} + +/// Trim `node`'s neighbour list back to `max_conn` with [`select_neighbors`]. fn prune_connections( vectors: &[Vec], neighbors: &mut Vec, @@ -839,22 +886,12 @@ fn prune_connections( if neighbors.len() <= max_conn { return; } - #[cfg(feature = "parallel")] - let mut scored: Vec<(usize, f32)> = { - use rayon::prelude::*; - neighbors - .par_iter() - .map(|&n| (n, compute_distance(&vectors[node], &vectors[n], metric))) - .collect() - }; - #[cfg(not(feature = "parallel"))] let mut scored: Vec<(usize, f32)> = neighbors .iter() .map(|&n| (n, compute_distance(&vectors[node], &vectors[n], metric))) .collect(); - scored.sort_by(|a, b| a.1.partial_cmp(&b.1).unwrap_or(std::cmp::Ordering::Equal)); - scored.truncate(max_conn); - *neighbors = scored.into_iter().map(|(id, _)| id).collect(); + scored.sort_by(|a, b| a.1.total_cmp(&b.1).then(a.0.cmp(&b.0))); + *neighbors = select_neighbors(vectors, &scored, max_conn, metric); } // --------------------------------------------------------------------------- @@ -1043,6 +1080,104 @@ fn get_attr_string(attrs: &[(String, AttrValue)], name: &str) -> Result Vec> { + let mut state = seed; + let mut next = move || { + state = state.wrapping_add(0x9E37_79B9_7F4A_7C15); + let mut z = state; + z = (z ^ (z >> 30)).wrapping_mul(0xBF58_476D_1CE4_E5B9); + z = (z ^ (z >> 27)).wrapping_mul(0x94D0_49BB_1331_11EB); + ((z ^ (z >> 31)) >> 40) as f32 / (1u64 << 24) as f32 - 0.5 + }; + let centres: Vec> = (0..clusters) + .map(|_| (0..dim).map(|_| next() * 10.0).collect()) + .collect(); + (0..n) + .map(|i| { + centres[i % clusters] + .iter() + .map(|c| c + next() * 0.5) + .collect() + }) + .collect() + } + + fn recall_at_10( + index: &HnswIndex, + vectors: &[Vec], + queries: &[Vec], + ef: usize, + ) -> f64 { + let mut hits = 0; + for q in queries { + let mut exact: Vec<(usize, f32)> = vectors + .iter() + .enumerate() + .map(|(i, v)| (i, compute_distance(q, v, DistanceMetric::L2))) + .collect(); + exact.sort_by(|a, b| a.1.total_cmp(&b.1)); + let want: Vec = exact[..10].iter().map(|e| e.0).collect(); + hits += index + .search(q, 10, ef) + .iter() + .filter(|(id, _)| want.contains(id)) + .count(); + } + hits as f64 / (10 * queries.len()) as f64 + } + + #[test] + fn clustered_data_keeps_high_recall() { + // Data and queries come from the same clusters: one draw, split. + let mut vectors = clustered(3060, 24, 30, 1); + let queries = vectors.split_off(3000); + let built = HnswIndex::build_with_metric(&vectors, 8, 40, DistanceMetric::L2); + let recall = recall_at_10(&built, &vectors, &queries, 64); + assert!(recall >= 0.95, "bulk build recall@10 = {recall}"); + + // Incremental inserts go through the same neighbour selection. + let mut incremental = HnswIndex::new(8, 40, DistanceMetric::L2); + for v in &vectors { + incremental.insert(v.clone()); + } + let recall = recall_at_10(&incremental, &vectors, &queries, 64); + assert!(recall >= 0.95, "incremental recall@10 = {recall}"); + } + + #[test] + fn select_neighbors_prefers_diverse_directions_and_fills_up() { + // Node at the origin. Three candidates bunched together on the right, + // one on the left. With room for two, plain closest-M would take two + // from the bunch and lose the only link leftwards. + let vectors = vec![ + vec![0.0, 0.0], // 0: the node + vec![1.0, 0.0], // 1 + vec![1.1, 0.0], // 2 + vec![1.2, 0.0], // 3 + vec![-2.0, 0.0], // 4 + ]; + let scored: Vec<(usize, f32)> = (1..5) + .map(|i| { + ( + i, + compute_distance(&vectors[0], &vectors[i], DistanceMetric::L2), + ) + }) + .collect(); + assert_eq!( + select_neighbors(&vectors, &scored, 2, DistanceMetric::L2), + [1, 4] + ); + // Spare capacity is filled with the closest rejected candidates. + assert_eq!( + select_neighbors(&vectors, &scored, 3, DistanceMetric::L2), + [1, 4, 2] + ); + } + fn make_random_vectors(n: usize, dim: usize, seed: u64) -> Vec> { let mut vectors = Vec::with_capacity(n); let mut state = seed; diff --git a/crates/clawhdf5-bench/src/bin/search_harness.rs b/crates/clawhdf5-bench/src/bin/search_harness.rs index 0db6931..d6e5d19 100644 --- a/crates/clawhdf5-bench/src/bin/search_harness.rs +++ b/crates/clawhdf5-bench/src/bin/search_harness.rs @@ -18,6 +18,7 @@ //! cargo run --release -p clawhdf5-bench --bin search_harness # 1K, 10K //! cargo run --release -p clawhdf5-bench --bin search_harness -- --full # + 100K //! cargo run --release -p clawhdf5-bench --bin search_harness -- --json out.json +//! cargo run --release -p clawhdf5-bench --bin search_harness -- --ann-only --uniform //! ``` use std::time::{Duration, Instant}; @@ -78,8 +79,26 @@ struct Dataset { query_cluster: Vec, } +/// `--uniform`: isotropic random unit vectors instead of clusters. Not a +/// realistic workload, but a useful second distribution — a recall problem +/// that appears only on clustered data points at graph connectivity. +static UNIFORM: std::sync::atomic::AtomicBool = std::sync::atomic::AtomicBool::new(false); + fn make_dataset(n: usize, seed: u64) -> Dataset { let mut rng = Rng(seed); + if UNIFORM.load(std::sync::atomic::Ordering::Relaxed) { + let random_unit = |rng: &mut Rng| { + let mut v: Vec = (0..DIM).map(|_| rng.gauss()).collect(); + normalize(&mut v); + v + }; + return Dataset { + vectors: (0..n).map(|_| random_unit(&mut rng)).collect(), + queries: (0..N_QUERIES).map(|_| random_unit(&mut rng)).collect(), + cluster_of: vec![0; n], + query_cluster: vec![0; N_QUERIES], + }; + } let n_clusters = (n / 100).clamp(8, 512); let centres: Vec> = (0..n_clusters) .map(|_| { @@ -344,6 +363,11 @@ fn bench_end_to_end(n: usize, json: &mut Vec) { fn main() { let args: Vec = std::env::args().skip(1).collect(); let full = args.iter().any(|a| a == "--full"); + let ann_only = args.iter().any(|a| a == "--ann-only"); + if args.iter().any(|a| a == "--uniform") { + UNIFORM.store(true, std::sync::atomic::Ordering::Relaxed); + println!("(uniform random data)"); + } let json_path = args .iter() .position(|a| a == "--json") @@ -365,6 +389,9 @@ fn main() { bench_ann(n, &mut json); } + if ann_only { + return; + } println!("\n### End to end: `HDF5Memory::hybrid_search` (k = {K}, weights 0.7 / 0.3)\n"); println!( "| N | ingest ms | checkpoint ms | open ms | first query ms | p50 ms | p99 ms | QPS |" From 61424d141808b4f71a78ac7cd19982f5165b5a8c Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 07:43:52 -0700 Subject: [PATCH 3/6] docs(bench): record search harness numbers after the HNSW heuristic Co-Authored-By: Claude Fable 5.1 --- BENCHMARKS.md | 45 +++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 45 insertions(+) diff --git a/BENCHMARKS.md b/BENCHMARKS.md index 1f6b21e..c1fe299 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -94,6 +94,51 @@ build: 9752.6 ms (10254 vectors/s) · exact scan: 40 QPS, p50 24648 µs | 100000 | 1486 | 713.0 | 354.5 | 10486.5 | 883.51 | 975.23 | 1.1 | wrote /tmp/claude-1000/-home-osobh-projects-clawhdf5/422f755e-dd25-4c35-8613-5439087e3aaa/scratchpad/baseline_full.json +### After: HNSW neighbour-selection heuristic + +Same harness, same data, after replacing closest-M neighbour selection with the +HNSW paper's diversity heuristic (Algorithm 4, keeping pruned connections) for +both new links and back-link pruning. Recall@10 at `ef = 64`: **0.87 → 1.00** +(1K), **0.67 → 1.00** (10K), **0.31 → 0.98** (100K), and it now rises with +`ef` as it should. The cost is a slower build (extra distance evaluations per +insert: ~3.5x at 10K); the distance-kernel work that follows targets that. + +### HNSW, N = 1000, dim = 384, M = 16, ef_construction = 64 + +build: 221.3 ms (4519 vectors/s) · exact scan: 3851 QPS, p50 258 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.9990 | 54760 | 18 | 29 | +| 32 | 1.0000 | 40422 | 24 | 44 | +| 64 | 1.0000 | 27744 | 36 | 51 | +| 128 | 1.0000 | 13164 | 74 | 106 | +| 256 | 1.0000 | 6879 | 144 | 175 | + +### HNSW, N = 10000, dim = 384, M = 16, ef_construction = 64 + +build: 2733.5 ms (3658 vectors/s) · exact scan: 423 QPS, p50 2362 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.9975 | 31321 | 27 | 61 | +| 32 | 1.0000 | 32427 | 29 | 48 | +| 64 | 1.0000 | 22738 | 42 | 62 | +| 128 | 1.0000 | 10055 | 99 | 129 | +| 256 | 1.0000 | 4649 | 214 | 266 | + +### HNSW, N = 100000, dim = 384, M = 16, ef_construction = 64 + +build: 36472.8 ms (2742 vectors/s) · exact scan: 40 QPS, p50 24644 µs + +| ef | recall@10 | QPS | p50 µs | p99 µs | +|---:|---:|---:|---:|---:| +| 16 | 0.9235 | 11394 | 82 | 194 | +| 32 | 0.9675 | 12788 | 73 | 161 | +| 64 | 0.9840 | 10406 | 91 | 186 | +| 128 | 0.9990 | 7633 | 126 | 248 | +| 256 | 0.9990 | 2823 | 352 | 510 | + ## Vector Search Latency Brute-force cosine similarity over 384-dimensional embeddings (OpenAI text-embedding-3-small size). From 2bfbb7fb4b6238b93b99a603262d2b9a545af5d8 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 07:51:21 -0700 Subject: [PATCH 4/6] perf(agent): persistent incremental BM25 index; no store rewrite per query hybrid_search rebuilt the BM25 index from scratch (re-tokenising every record) and rewrote the whole .h5 file on every single query, so a query cost O(store size) in both CPU and disk I/O. Steady-state p50 per the search harness: 5.5 -> 0.24 ms (1K), 49 -> 2.1 ms (10K), 884 -> 23 ms (100K). - BM25Index is incremental: add_document / remove_document keep it exactly equivalent to a fresh build over the same live documents (property test: 60 random op sequences compared against BM25Index::build after every step). IDF moves to query time since it depends on the live document count. Top-k uses a bounded heap, ties break by doc id (results were HashMap-ordered), and the "WAND" code that computed a bound and then discarded it is removed. - HDF5Memory keeps one index for its lifetime, built lazily. Appends are picked up by ensure_bm25_fresh whatever path added them; delete and in-place update report themselves; compaction drops the index. A test drives every mutation and compares against a fresh build. - A query no longer calls flush(). Activation boosts are marked dirty and persisted by the next checkpoint, including a best-effort one on drop so a search-only session keeps them (approved behaviour change). Activation weights are capped at 16.0; they previously grew without bound. The archived mission branch's BM25 cache was reviewed and not used: it was invalidated by every write, so interleaved save/search still rebuilt per query, and it changed the default fusion weights. Co-Authored-By: Claude Fable 5.1 --- BENCHMARKS.md | 20 ++ CHANGELOG.md | 14 ++ crates/clawhdf5-agent/src/bm25.rs | 330 ++++++++++++++++++---------- crates/clawhdf5-agent/src/lib.rs | 156 +++++++++++++ crates/clawhdf5-agent/src/search.rs | 25 ++- 5 files changed, 427 insertions(+), 118 deletions(-) diff --git a/BENCHMARKS.md b/BENCHMARKS.md index c1fe299..2701517 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -139,6 +139,26 @@ build: 36472.8 ms (2742 vectors/s) · exact scan: 40 QPS, p50 24644 µs | 128 | 0.9990 | 7633 | 126 | 248 | | 256 | 0.9990 | 2823 | 352 | 510 | +### After: persistent keyword index, no store rewrite per query + +`hybrid_search` used to rebuild the BM25 index from scratch (re-tokenising every +record) and rewrite the whole `.h5` file on **every query**. The index is now +kept for the life of the store and updated incrementally, and activation boosts +are persisted by the next checkpoint instead of inside the query. Steady-state +p50: **5.5 → 0.24 ms** (1K), **49 → 2.1 ms** (10K), **884 → 23 ms** (100K). + +The first query after `open()` is slower than before (it pays for the better — +slower — HNSW build plus the one-off keyword index build); persisting the HNSW +index removes that. + +### End to end: `HDF5Memory::hybrid_search` (k = 10, weights 0.7 / 0.3) + +| N | ingest ms | checkpoint ms | open ms | first query ms | p50 ms | p99 ms | QPS | +|---:|---:|---:|---:|---:|---:|---:|---:| +| 1000 | 11 | 3.8 | 0.9 | 195.9 | 0.24 | 0.27 | 4130.4 | +| 10000 | 104 | 31.1 | 10.9 | 2627.1 | 2.09 | 2.11 | 479.5 | +| 100000 | 1436 | 684.7 | 278.0 | 36308.1 | 22.90 | 25.46 | 43.5 | + ## Vector Search Latency Brute-force cosine similarity over 384-dimensional embeddings (OpenAI text-embedding-3-small size). diff --git a/CHANGELOG.md b/CHANGELOG.md index cc08329..e5f16ec 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -12,6 +12,20 @@ 0.98 and responds to `ef`. Builds are slower (~3.5x at 10K). Existing persisted indexes keep their old graph until rebuilt; the agent rebuilds its index from the cache, so stores pick this up automatically. +- `clawhdf5-agent`: **`hybrid_search` is 23-39x faster in steady state** (p50 + 5.5 -> 0.24 ms at 1K records, 49 -> 2.1 ms at 10K, 884 -> 23 ms at 100K). + Every query used to rebuild the BM25 index from scratch and rewrite the whole + `.h5` file. The keyword index now lives for the life of the store and is + updated incrementally (add / remove / in-place update, exactly equivalent to + a fresh build - property-tested), and a query no longer writes the store. + **Behaviour change:** Hebbian activation boosts are persisted by the next + checkpoint (any flushing write, `flush_wal`, or drop) rather than + immediately; a crash in between forgets only the boosts since the last + checkpoint. Activation weights are now capped (16.0) - they grew without + bound. +- `clawhdf5-agent`: BM25 results are deterministic (ties break by record id), + top-k uses a bounded heap, and the "WAND early termination" that computed a + bound and then ignored it is gone. IDF is computed per query. - `clawhdf5-bench`: new `search_harness` binary — HNSW recall@10 / QPS / latency per `ef` against an exact scan, and end-to-end `hybrid_search` timings, on deterministic clustered (or `--uniform`) data. Baseline in `BENCHMARKS.md`. diff --git a/crates/clawhdf5-agent/src/bm25.rs b/crates/clawhdf5-agent/src/bm25.rs index 258df33..1aff2d7 100644 --- a/crates/clawhdf5-agent/src/bm25.rs +++ b/crates/clawhdf5-agent/src/bm25.rs @@ -3,10 +3,15 @@ //! Provides a standard BM25 (Okapi BM25) implementation with an in-memory //! inverted index. Tombstoned documents are excluded from indexing and search. //! -//! Optimizations: -//! - Cached IDF scores (don't recompute per query) -//! - Sorted posting lists by doc_id for cache-friendly access -//! - Block-Max WAND early termination +//! The index is **incremental**: [`BM25Index::add_document`] and +//! [`BM25Index::remove_document`] keep it exactly equivalent to one built from +//! scratch over the same live documents, so a store can maintain one index for +//! its lifetime instead of re-tokenising the whole corpus per query. To make +//! that possible IDF is computed at query time (it depends on the live +//! document count) rather than cached at build time. +//! +//! - Posting lists sorted by doc id +//! - Bounded-heap top-k; results ordered by score, then doc id (deterministic) use std::cmp::Reverse; use std::collections::{BinaryHeap, HashMap}; @@ -41,10 +46,11 @@ const DEFAULT_B: f32 = 0.75; pub struct BM25Index { /// Inverted index: token -> sorted list of (doc_id, term_frequency). inverted: HashMap>, - /// Cached IDF scores per token. - idf_cache: HashMap, /// Number of tokens in each document (0 for tombstoned docs). doc_lengths: Vec, + /// Sum of `doc_lengths` over live documents (keeps `avg_dl` exact under + /// incremental updates). + total_length: u64, /// Average document length across non-tombstoned docs. avg_dl: f32, /// Number of non-tombstoned documents. @@ -60,8 +66,8 @@ impl BM25Index { pub fn build(documents: &[String], tombstones: &[u8]) -> Self { let mut index = Self { inverted: HashMap::new(), - idf_cache: HashMap::new(), doc_lengths: vec![0; documents.len()], + total_length: 0, avg_dl: 0.0, num_docs: 0, k1: DEFAULT_K1, @@ -81,103 +87,135 @@ impl BM25Index { return Vec::new(); } - let tokens = tokenize(query); - if tokens.is_empty() { - return Vec::new(); - } - - // Collect posting lists and cached IDF scores for query tokens - type QueryTerm<'a> = (&'a str, f32, &'a [(usize, u32)]); - let mut query_terms: Vec> = Vec::new(); - for token in &tokens { - if let (Some(postings), Some(&idf)) = ( - self.inverted.get(token.as_str()), - self.idf_cache.get(token.as_str()), - ) { - query_terms.push((token, idf, postings)); - } - } - - if query_terms.is_empty() { - return Vec::new(); - } - - // Accumulate BM25 scores per document using WAND-style scoring + // Term-at-a-time accumulation. IDF is computed here rather than cached + // at build time: it depends on the live document count, which changes + // with every incremental add/remove, and costs one `ln` per query term. let mut scores: HashMap = HashMap::new(); - - // Compute maximum possible contribution per term for WAND - let max_tf_score: Vec = query_terms - .iter() - .map(|(_, idf, _)| { - // Upper bound: max TF contribution when tf is high and dl is short - let max_tf_num = 10.0 * (self.k1 + 1.0); - let max_tf_den = 10.0 + self.k1 * (1.0 - self.b); - idf * max_tf_num / max_tf_den - }) - .collect(); - - let total_max_contribution: f32 = max_tf_score.iter().sum(); - - // Threshold for WAND early termination. `top_k_heap` is a min-heap of - // size k (worst-of-the-top-k at the head) so it can be maintained in - // O(log k) per update instead of re-sorting the whole buffer. - let mut threshold = 0.0f32; - let mut top_k_heap: BinaryHeap> = BinaryHeap::with_capacity(k); - - for (term_idx, (_, idf, postings)) in query_terms.iter().enumerate() { - for &(doc_id, freq) in *postings { + for token in tokenize(query) { + let Some(postings) = self.inverted.get(token.as_str()) else { + continue; + }; + let df = postings.len() as f32; + let idf = ((self.num_docs as f32 - df + 0.5) / (df + 0.5) + 1.0).ln(); + for &(doc_id, freq) in postings { let dl = self.doc_lengths[doc_id] as f32; let freq_f = freq as f32; let tf = (freq_f * (self.k1 + 1.0)) / (freq_f + self.k1 * (1.0 - self.b + self.b * dl / self.avg_dl)); - let contribution = idf * tf; - - let entry = scores.entry(doc_id).or_insert(0.0); - *entry += contribution; - - // WAND check: if this doc's current partial score + remaining - // max terms can't beat threshold, we can skip (but we still - // accumulate since we process term-at-a-time) - if term_idx == query_terms.len() - 1 { - // Last term: check if this doc beats threshold - let final_score = *entry; - if top_k_heap.len() >= k { - if final_score > threshold { - // Replace the current worst-of-top-k. - top_k_heap.pop(); - top_k_heap.push(Reverse(HeapScore(final_score))); - threshold = top_k_heap.peek().map(|Reverse(s)| s.0).unwrap_or(0.0); - } - } else { - top_k_heap.push(Reverse(HeapScore(final_score))); - if top_k_heap.len() == k { - threshold = top_k_heap.peek().map(|Reverse(s)| s.0).unwrap_or(0.0); - } - } - } - } - // After processing each term, check if remaining terms can - // possibly produce results above threshold - let remaining_max: f32 = max_tf_score[term_idx + 1..].iter().sum(); - if remaining_max < threshold && total_max_contribution > 0.0 { - // Early termination: remaining terms can't produce new top-k - // entries on their own. But existing partial scores may still - // be updated, so we continue (WAND is approximate here). - let _ = remaining_max; // hint to compiler + *scores.entry(doc_id).or_insert(0.0) += idf * tf; } } - let mut results: Vec<(usize, f32)> = scores.into_iter().collect(); - results.sort_by(|a, b| b.1.partial_cmp(&a.1).unwrap_or(std::cmp::Ordering::Equal)); - results.truncate(k); + // Top-k with a bounded min-heap: O(matches * log k) instead of sorting + // every match. Ties break towards the lower doc id so results are + // deterministic (the accumulator is a HashMap). + let mut heap: BinaryHeap)>> = + BinaryHeap::with_capacity(k + 1); + for (doc_id, score) in scores { + heap.push(Reverse((HeapScore(score), Reverse(doc_id)))); + if heap.len() > k { + heap.pop(); + } + } + let mut results: Vec<(usize, f32)> = heap + .into_iter() + .map(|Reverse((HeapScore(score), Reverse(doc_id)))| (doc_id, score)) + .collect(); + results.sort_by(|a, b| b.1.total_cmp(&a.1).then(a.0.cmp(&b.0))); results } + /// Number of document slots (live or not) the index covers. Ids are + /// positions in the document list it mirrors. + pub fn len(&self) -> usize { + self.doc_lengths.len() + } + + /// `true` when the index covers no document slots. + pub fn is_empty(&self) -> bool { + self.doc_lengths.is_empty() + } + + /// Index `text` as document `doc_id`, which must be the next free id + /// (`self.len()`) or an existing slot that is currently empty (removed or + /// tombstoned). After any sequence of `add_document` / `remove_document` + /// calls the index scores exactly as one freshly built from the same live + /// documents. + pub fn add_document(&mut self, doc_id: usize, text: &str) { + if doc_id >= self.doc_lengths.len() { + self.doc_lengths.resize(doc_id + 1, 0); + } + debug_assert_eq!(self.doc_lengths[doc_id], 0, "slot {doc_id} is occupied"); + + let tokens = tokenize(text); + let mut term_freqs: HashMap<&str, u32> = HashMap::new(); + for token in &tokens { + *term_freqs.entry(token).or_insert(0) += 1; + } + for (token, freq) in term_freqs { + let postings = self.inverted.entry(token.to_string()).or_default(); + // Posting lists stay sorted by doc id; appends are the common case. + match postings.last() { + Some(&(last, _)) if last >= doc_id => { + let at = postings.partition_point(|&(id, _)| id < doc_id); + postings.insert(at, (doc_id, freq)); + } + _ => postings.push((doc_id, freq)), + } + } + self.doc_lengths[doc_id] = tokens.len() as u32; + self.total_length += tokens.len() as u64; + self.num_docs += 1; + self.refresh_avg_dl(); + } + + /// Extend the index to cover `len` document slots, leaving new ones empty. + /// Used for slots that hold no live document (tombstoned records). + pub fn pad_to(&mut self, len: usize) { + if len > self.doc_lengths.len() { + self.doc_lengths.resize(len, 0); + } + } + + /// Remove document `doc_id`, whose indexed text was `text`. The text is + /// needed to find its postings; pass exactly what was added. + pub fn remove_document(&mut self, doc_id: usize, text: &str) { + let tokens = tokenize(text); + let mut seen: std::collections::HashSet<&str> = std::collections::HashSet::new(); + for token in &tokens { + if !seen.insert(token) { + continue; + } + if let Some(postings) = self.inverted.get_mut(token.as_str()) { + if let Ok(at) = postings.binary_search_by_key(&doc_id, |&(id, _)| id) { + postings.remove(at); + } + if postings.is_empty() { + self.inverted.remove(token.as_str()); + } + } + } + if let Some(len) = self.doc_lengths.get_mut(doc_id) { + self.total_length = self.total_length.saturating_sub(u64::from(*len)); + *len = 0; + } + self.num_docs = self.num_docs.saturating_sub(1); + self.refresh_avg_dl(); + } + + fn refresh_avg_dl(&mut self) { + self.avg_dl = if self.num_docs > 0 { + self.total_length as f32 / self.num_docs as f32 + } else { + 0.0 + }; + } + /// Rebuild the index from scratch (e.g., after compaction). pub fn rebuild(&mut self, documents: &[String], tombstones: &[u8]) { self.inverted.clear(); - self.idf_cache.clear(); self.doc_lengths = vec![0; documents.len()]; + self.total_length = 0; self.avg_dl = 0.0; self.num_docs = 0; self.index_documents(documents, tombstones); @@ -214,23 +252,13 @@ impl BM25Index { } self.num_docs = count; - self.avg_dl = if count > 0 { - total_length as f32 / count as f32 - } else { - 0.0 - }; + self.total_length = total_length; + self.refresh_avg_dl(); // Sort posting lists by doc_id for cache-friendly access for postings in self.inverted.values_mut() { postings.sort_by_key(|&(doc_id, _)| doc_id); } - - // Pre-compute and cache IDF scores - for (token, postings) in &self.inverted { - let df = postings.len() as f32; - let idf = ((self.num_docs as f32 - df + 0.5) / (df + 0.5) + 1.0).ln(); - self.idf_cache.insert(token.clone(), idf); - } } } @@ -386,24 +414,21 @@ mod tests { } #[test] - fn cached_idf_consistent_with_computed() { + fn score_matches_the_bm25_formula() { let docs = vec![ "rust programming".to_string(), "rust systems".to_string(), "python scripting".to_string(), ]; - let tombstones = vec![0, 0, 0]; - let index = BM25Index::build(&docs, &tombstones); + let index = BM25Index::build(&docs, &[0, 0, 0]); - // IDF for "rust" (appears in 2 of 3 docs) - let idf_rust = index.idf_cache.get("rust").unwrap(); - let expected_idf = ((3.0f32 - 2.0 + 0.5) / (2.0 + 0.5) + 1.0).ln(); - assert!( - (idf_rust - expected_idf).abs() < 1e-6, - "cached IDF mismatch: {} vs {}", - idf_rust, - expected_idf - ); + // "python": df = 1 of N = 3. Every doc has the average length (2) and + // tf = 1, so the tf factor is exactly 1 and the score is the IDF. + let results = index.search("python", 3); + let expected_idf = ((3.0f32 - 1.0 + 0.5) / (1.0 + 0.5) + 1.0).ln(); + assert_eq!(results.len(), 1); + assert_eq!(results[0].0, 2); + assert!((results[0].1 - expected_idf).abs() < 1e-6, "{results:?}"); } #[test] @@ -467,4 +492,83 @@ mod tests { ); } } + /// Documents drawn from a small vocabulary so terms collide heavily. + fn random_doc(state: &mut u64) -> String { + const VOCAB: &[&str] = &[ + "alpha", "beta", "gamma", "delta", "eps", "zeta", "eta", "x1", + ]; + let mut next = || { + *state = state + .wrapping_mul(6364136223846793005) + .wrapping_add(1442695040888963407); + (*state >> 33) as usize + }; + let len = 1 + next() % 9; + (0..len) + .map(|_| VOCAB[next() % VOCAB.len()]) + .collect::>() + .join(" ") + } + + #[test] + fn incremental_updates_match_a_fresh_build_exactly() { + for seed in 0..60u64 { + let mut state = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; + let mut docs: Vec = Vec::new(); + let mut tombstones: Vec = Vec::new(); + let mut index = BM25Index::build(&docs, &tombstones); + + for step in 0..80 { + state = state.wrapping_mul(6364136223846793005).wrapping_add(1); + let live: Vec = (0..docs.len()).filter(|&i| tombstones[i] == 0).collect(); + match (state >> 40) % 4 { + 0 if !live.is_empty() => { + // delete + let id = live[(state >> 20) as usize % live.len()]; + index.remove_document(id, &docs[id]); + tombstones[id] = 1; + } + 1 if !live.is_empty() => { + // update in place + let id = live[(state >> 20) as usize % live.len()]; + let new_text = random_doc(&mut state); + index.remove_document(id, &docs[id]); + index.add_document(id, &new_text); + docs[id] = new_text; + } + _ => { + let text = random_doc(&mut state); + index.add_document(docs.len(), &text); + docs.push(text); + tombstones.push(0); + } + } + + let fresh = BM25Index::build(&docs, &tombstones); + for query in ["alpha", "beta gamma", "x1 zeta alpha delta", "missing"] { + let got = index.search(query, 5); + let want = fresh.search(query, 5); + assert_eq!(got.len(), want.len(), "seed {seed} step {step} {query:?}"); + for (g, w) in got.iter().zip(&want) { + assert_eq!( + g.0, w.0, + "seed {seed} step {step} {query:?}: {got:?} vs {want:?}" + ); + assert!( + (g.1 - w.1).abs() < 1e-5, + "seed {seed} step {step} {query:?}" + ); + } + } + } + } + } + + #[test] + fn ties_break_towards_the_lower_doc_id() { + let docs: Vec = (0..6).map(|_| "same text".to_string()).collect(); + let index = BM25Index::build(&docs, &[0; 6]); + let ids: Vec = index.search("same", 3).into_iter().map(|r| r.0).collect(); + assert_eq!(ids, [0, 1, 2]); + } } diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index f07c755..ad624f0 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -205,6 +205,12 @@ pub trait AgentMemory { fn get_session_summary(&self, session_id: &str) -> Result>; } +/// Ceiling for a record's Hebbian activation weight. Each hit adds +/// `hebbian_boost` and the fused score is scaled by `sqrt(weight)`, so without +/// a cap a frequently returned record's advantage grows without limit and it +/// eventually outranks better matches purely on popularity. +pub(crate) const MAX_ACTIVATION_WEIGHT: f32 = 16.0; + /// Most anomaly alerts kept between `take_anomaly_alerts` calls. const MAX_PENDING_ALERTS: usize = 1024; @@ -247,6 +253,14 @@ pub struct HDF5Memory { /// via [`HDF5Memory::take_anomaly_alerts`]. Saves are never blocked on /// these — surfacing is opt-in for callers that want to act on them. anomaly_alerts: Vec, + /// Keyword index over `cache.chunks`, kept for the life of the store and + /// updated incrementally — it used to be rebuilt from scratch, re-tokenising + /// every record, on every single query. Built lazily on first use; see + /// [`HDF5Memory::ensure_bm25_fresh`] for how it stays in sync. + bm25: Option, + /// Activation weights changed since the last checkpoint (searches boost + /// the records they return). Cleared by `flush`. + activations_dirty: bool, /// Opened with [`HDF5Memory::open_read_only`]: nothing may reach the disk. read_only: bool, /// A WAL that `open()` could not read and moved aside; see @@ -299,6 +313,8 @@ impl HDF5Memory { provenance: provenance::ProvenanceStore::new(), anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()), anomaly_alerts: Vec::new(), + bm25: None, + activations_dirty: false, read_only: false, quarantined_wal: None, _lock: Some(lock), @@ -434,12 +450,58 @@ impl HDF5Memory { provenance: provenance::ProvenanceStore::new(), anomaly: anomaly::WriteAnomalyDetector::new(anomaly::AnomalyConfig::default()), anomaly_alerts: Vec::new(), + bm25: None, + activations_dirty: false, read_only, quarantined_wal, _lock: lock, }) } + /// Bring the keyword index in line with the cache and return it. + /// + /// Appends need no hook: records the index hasn't seen yet (whatever path + /// added them) are indexed here, in order. Changes that keep the length the + /// same are reported explicitly — [`Self::bm25_on_delete`] and + /// [`Self::bm25_on_update`] — and anything that renumbers records + /// (compaction) drops the index so it is rebuilt. + pub(crate) fn ensure_bm25_fresh(&mut self) -> &bm25::BM25Index { + let n = self.cache.chunks.len(); + let bm25 = match self.bm25.take() { + Some(index) if index.len() <= n => { + let mut index = index; + for id in index.len()..n { + if self.cache.tombstones[id] == 0 { + index.add_document(id, &self.cache.chunks[id]); + } + } + index.pad_to(n); + index + } + _ => bm25::BM25Index::build(&self.cache.chunks, &self.cache.tombstones), + }; + self.bm25.insert(bm25) + } + + /// Record `id` was tombstoned; its text is still in the cache. + fn bm25_on_delete(&mut self, id: usize) { + if let Some(index) = self.bm25.as_mut() + && id < index.len() + { + index.remove_document(id, &self.cache.chunks[id]); + } + } + + /// Record `id`'s text changed from `old_text` to what the cache holds now. + fn bm25_on_update(&mut self, id: usize, old_text: &str) { + if let Some(index) = self.bm25.as_mut() + && id < index.len() + { + index.remove_document(id, old_text); + index.add_document(id, &self.cache.chunks[id]); + } + } + /// Flush current state to disk and truncate the WAL. /// /// Every code path that persists the full cache to the .h5 file must @@ -466,6 +528,7 @@ impl HDF5Memory { if let Some(ref mut w) = self.wal { w.truncate()?; } + self.activations_dirty = false; Ok(()) } @@ -777,6 +840,7 @@ impl HDF5Memory { &entry.session_id, entry.timestamp, ); + let old_text = std::mem::take(&mut self.cache.chunks[existing_idx]); self.cache.update( existing_idx, entry.chunk, @@ -785,6 +849,7 @@ impl HDF5Memory { entry.timestamp, entry.session_id, ); + self.bm25_on_update(existing_idx, &old_text); // In-place embedding change: the index node is stale, force rebuild. self.hnsw_mark_dirty(); let needs_flush = self @@ -876,6 +941,7 @@ impl AgentMemory for HDF5Memory { ))); } self.hnsw_on_delete(id); + self.bm25_on_delete(id); self.flush()?; // Auto-compact if threshold exceeded @@ -893,6 +959,7 @@ impl AgentMemory for HDF5Memory { if removed > 0 { // Record ids are cache indices, which compaction just renumbered. self.provenance.remap(&index_map); + self.bm25 = None; // Compaction renumbers cache indices; rebuild the index to match. self.hnsw_mark_dirty(); self.flush()?; @@ -1172,6 +1239,18 @@ impl HDF5Memory { // --- Tests --- +impl Drop for HDF5Memory { + /// Best-effort checkpoint of activation weights that only searches have + /// touched. Everything else is already durable through the WAL or an + /// earlier checkpoint; without this a search-only session would forget + /// every boost it made. + fn drop(&mut self) { + if self.activations_dirty && !self.read_only { + let _ = self.flush(); + } + } +} + #[cfg(test)] mod tests { use super::*; @@ -1595,6 +1674,83 @@ mod tests { assert_eq!(restored.cache.chunks, ["checkpointed", "wal-only"]); } + #[test] + fn keyword_index_stays_in_sync_through_every_mutation() { + let dir = TempDir::new().unwrap(); + let mut config = make_config(&dir); + config.compact_threshold = 0.0; // compact only when asked + let mut mem = HDF5Memory::create(config).unwrap(); + let check = |mem: &mut HDF5Memory, what: &str| { + let fresh = bm25::BM25Index::build(&mem.cache.chunks, &mem.cache.tombstones); + let n = mem.cache.len(); + for query in ["apple", "banana cherry", "date", "nothing"] { + let kept = mem.ensure_bm25_fresh().search(query, n); + assert_eq!(kept, fresh.search(query, n), "{what}: {query:?}"); + } + }; + let tagged = |chunk: &str, tag: &str| { + let mut e = make_entry(chunk, &[1.0, 0.0, 0.0, 0.0]); + e.tags = tag.into(); + e + }; + + check(&mut mem, "empty"); + mem.save(tagged("apple banana", "a")).unwrap(); + mem.save(tagged("banana cherry cherry", "b")).unwrap(); + check(&mut mem, "after saves"); + mem.save_batch(vec![tagged("date apple", "c"), tagged("cherry", "d")]) + .unwrap(); + check(&mut mem, "after save_batch"); + mem.save_or_update(tagged("date date date", "a")).unwrap(); + check(&mut mem, "after in-place update"); + mem.delete(1).unwrap(); + check(&mut mem, "after delete"); + mem.save(tagged("apple cherry", "e")).unwrap(); + check(&mut mem, "after save following a delete"); + mem.compact().unwrap(); + check(&mut mem, "after compact"); + mem.hybrid_search(&[1.0, 0.0, 0.0, 0.0], "apple", 0.5, 0.5, 3); + check(&mut mem, "after a search"); + } + + #[test] + fn search_does_not_write_the_store_but_boosts_persist_on_drop() { + let dir = TempDir::new().unwrap(); + let config = make_config(&dir); + let path = config.path.clone(); + let mut mem = HDF5Memory::create(config).unwrap(); + mem.save(make_entry("findable", &[1.0, 0.0, 0.0, 0.0])) + .unwrap(); + let before = std::fs::read(&path).unwrap(); + + for _ in 0..3 { + mem.hybrid_search(&[1.0, 0.0, 0.0, 0.0], "findable", 1.0, 0.0, 1); + } + assert_eq!( + std::fs::read(&path).unwrap(), + before, + "a query must not rewrite the store" + ); + let boosted = mem.cache.activation_weights[0]; + assert!(boosted > 1.0); + drop(mem); + + let reopened = HDF5Memory::open(&path).unwrap(); + assert_eq!(reopened.cache.activation_weights[0], boosted); + } + + #[test] + fn activation_weight_is_capped() { + let dir = TempDir::new().unwrap(); + let mut mem = HDF5Memory::create(make_config(&dir)).unwrap(); + mem.save(make_entry("popular", &[1.0, 0.0, 0.0, 0.0])) + .unwrap(); + for _ in 0..500 { + mem.hybrid_search(&[1.0, 0.0, 0.0, 0.0], "popular", 1.0, 0.0, 1); + } + assert_eq!(mem.cache.activation_weights[0], MAX_ACTIVATION_WEIGHT); + } + #[test] fn store_has_a_single_writer() { let dir = TempDir::new().unwrap(); diff --git a/crates/clawhdf5-agent/src/search.rs b/crates/clawhdf5-agent/src/search.rs index b66f2f2..30b679a 100644 --- a/crates/clawhdf5-agent/src/search.rs +++ b/crates/clawhdf5-agent/src/search.rs @@ -4,7 +4,7 @@ use std::path::Path; use crate::bm25; use crate::hybrid; -use crate::{HDF5Memory, MemoryError, Result, SearchResult}; +use crate::{HDF5Memory, MAX_ACTIVATION_WEIGHT, MemoryError, Result, SearchResult}; impl HDF5Memory { /// Vector + keyword scoring stage of [`HDF5Memory::hybrid_search`]. @@ -90,7 +90,11 @@ impl HDF5Memory { keyword_weight: f32, k: usize, ) -> Vec { - let bm25 = bm25::BM25Index::build(&self.cache.chunks, &self.cache.tombstones); + // The keyword index lives for the life of the store and is updated + // incrementally. Take it out for the duration of the call so the + // vector stage can borrow `self` mutably, then put it back. + self.ensure_bm25_fresh(); + let bm25 = self.bm25.take().expect("ensure_bm25_fresh leaves an index"); let scored = self.vector_keyword_search( query_embedding, query_text, @@ -132,15 +136,26 @@ impl HDF5Memory { .map(|r| r.index) .collect(); self.apply_hebbian_boost(&hit_indices); - self.flush().ok(); + self.bm25 = Some(bm25); results } + /// Reinforce the records a query returned. The new weights are persisted by + /// the next checkpoint (any write that flushes, `flush_wal`, or drop) — not + /// by rewriting the whole store inside the query, which is what made + /// `hybrid_search` cost O(store size) in disk I/O. They are a ranking hint, + /// not user data: a crash before the next checkpoint only forgets the + /// boosts since the last one. fn apply_hebbian_boost(&mut self, hit_indices: &[usize]) { - for &idx in hit_indices { - self.cache.activation_weights[idx] += self.config.hebbian_boost; + if hit_indices.is_empty() || self.config.hebbian_boost == 0.0 { + return; } + for &idx in hit_indices { + let w = &mut self.cache.activation_weights[idx]; + *w = (*w + self.config.hebbian_boost).min(MAX_ACTIVATION_WEIGHT); + } + self.activations_dirty = true; } /// Get the chunk text for a memory entry by index. From 0ee698accda090efc3725158235566f3f06706a9 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 08:00:57 -0700 Subject: [PATCH 5/6] perf(agent): persist the vector index graph; incremental catch-up MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit open() marked the HNSW index dirty, so the first search of every session rebuilt it from scratch — 36 s at 100K records with the (better, slower) heuristic build. First query after open is now 1.7 / 15 / 159 ms at 1K / 10K / 100K; what remains is the one-off keyword index build. - clawhdf5-ann: HnswIndex::graph_to_bytes / from_graph_bytes serialize the graph only (levels, tombstones, adjacency as u32, CRC32). The existing HDF5 serializer embeds a full copy of every vector, which would double a store that already holds them. Loading validates everything — counts, levels vs layer count, connection limits, every neighbour id and the layer it must exist on — so a damaged graph, or a hostile one with a valid checksum, is an error rather than an out-of-bounds walk during search. - clawhdf5-agent: each checkpoint writes the graph to .h5.ann (synced, atomic, before the .h5) and records a fresh generation id in /meta. open() loads the sidecar only if its generation matches that checkpoint; missing, stale, damaged or mismatched sidecars are ignored and the index rebuilt. Records appended through WAL replay join the loaded index incrementally; a replayed Update or Tombstone invalidates it. snapshot() copies it. Only an index that exactly mirrors the cache is saved; otherwise a stale sidecar is removed. - ensure_hnsw_fresh inserts records appended since the last sync instead of rebuilding, so save_batch no longer marks the whole index dirty. - CheckpointMeta { wal_applied, ann_generation } with *_with_meta build/write/ read functions; the *_with_mark ones delegate. - Harness reports the one-off cold index build separately from the first query after a reopen. Co-Authored-By: Claude Fable 5.1 --- BENCHMARKS.md | 17 + CHANGELOG.md | 11 + crates/clawhdf5-agent/src/lib.rs | 302 +++++++++++++++++- crates/clawhdf5-agent/src/schema.rs | 50 +++ crates/clawhdf5-agent/src/storage.rs | 36 ++- crates/clawhdf5-ann/src/hnsw.rs | 224 +++++++++++++ .../clawhdf5-bench/src/bin/search_harness.rs | 21 +- 7 files changed, 641 insertions(+), 20 deletions(-) diff --git a/BENCHMARKS.md b/BENCHMARKS.md index 2701517..0a80afe 100644 --- a/BENCHMARKS.md +++ b/BENCHMARKS.md @@ -159,6 +159,23 @@ index removes that. | 10000 | 104 | 31.1 | 10.9 | 2627.1 | 2.09 | 2.11 | 479.5 | | 100000 | 1436 | 684.7 | 278.0 | 36308.1 | 22.90 | 25.46 | 43.5 | +### After: vector index persisted with the checkpoint + +The HNSW graph (not the vectors, which the store already holds) is saved to +`.h5.ann` at each checkpoint and reloaded by `open()`, tied to that +checkpoint by a generation id. The index is now built once per store (the *cold +index build* column — the first query ever), not once per session. First query +after `open()`: **196 → 1.7 ms** (1K), **2627 → 15 ms** (10K), +**36308 → 159 ms** (100K); what remains is the one-off keyword index build. +Batch saves no longer force a full rebuild either: appended records join the +index incrementally. + +| N | ingest ms | cold index build ms | checkpoint ms | open ms | first query after open ms | p50 ms | p99 ms | QPS | +|---:|---:|---:|---:|---:|---:|---:|---:|---:| +| 1000 | 14 | 220 | 6.1 | 1.2 | 1.7 | 0.24 | 0.27 | 4049.9 | +| 10000 | 120 | 2916 | 33.1 | 14.0 | 15.4 | 2.15 | 3.30 | 421.2 | +| 100000 | 1591 | 40515 | 747.3 | 324.7 | 158.9 | 23.07 | 30.42 | 41.3 | + ## Vector Search Latency Brute-force cosine similarity over 384-dimensional embeddings (OpenAI text-embedding-3-small size). diff --git a/CHANGELOG.md b/CHANGELOG.md index e5f16ec..28a74fa 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -23,6 +23,17 @@ immediately; a crash in between forgets only the boosts since the last checkpoint. Activation weights are now capped (16.0) - they grew without bound. +- `clawhdf5-agent`: **the vector index is persisted**, so `open()` no longer + rebuilds it on the first search (first query after open: 2627 -> 15 ms at 10K + records, 36 s -> 159 ms at 100K). The HNSW graph — not the vectors, which the + store already holds — is written to `.h5.ann` at each checkpoint and + tied to it by a generation id in `/meta`; a missing, stale, damaged or + structurally invalid sidecar is ignored and the index rebuilt. Records + replayed from the WAL join the loaded index incrementally; a replayed update + or delete invalidates it. `snapshot()` copies it. Batch saves no longer force + a full index rebuild. +- `clawhdf5-ann`: `HnswIndex::graph_to_bytes` / `from_graph_bytes` — graph-only + serialization (checksummed, every neighbour id and level validated on load). - `clawhdf5-agent`: BM25 results are deterministic (ties break by record id), top-k uses a bounded heap, and the "WAND early termination" that computed a bound and then ignored it is gone. IDF is computed per query. diff --git a/crates/clawhdf5-agent/src/lib.rs b/crates/clawhdf5-agent/src/lib.rs index ad624f0..746152e 100644 --- a/crates/clawhdf5-agent/src/lib.rs +++ b/crates/clawhdf5-agent/src/lib.rs @@ -390,8 +390,13 @@ impl HDF5Memory { } else { Some(store_lock::StoreLock::acquire(path)?) }; - let ((config, mut cache, sessions, knowledge), wal_applied) = - storage::read_from_disk_with_mark(path)?; + let ((config, mut cache, sessions, knowledge), checkpoint) = + storage::read_from_disk_with_meta(path)?; + let wal_applied = checkpoint.wal_applied; + let n_checkpoint = cache.len(); + // Set if WAL replay did anything other than append records; the saved + // vector index then no longer describes the first `n_checkpoint` ones. + let mut replay_only_appended = true; // Replay WAL if present let wal_path = path.with_extension("h5.wal"); @@ -408,6 +413,9 @@ impl HDF5Memory { && let Ok(entries) = wal::WalFile::read_entries_for_migration(&wal_path, wal_applied) { + replay_only_appended &= entries + .iter() + .all(|e| e.entry_type == wal::WalEntryType::Save); wal::replay_into_cache(&entries, &mut cache); } None @@ -419,6 +427,9 @@ impl HDF5Memory { // in case the process died between writing the .h5 and // truncating the WAL. let entries = wal::WalFile::read_entries_for_migration(&wal_path, wal_applied)?; + replay_only_appended &= entries + .iter() + .all(|e| e.entry_type == wal::WalEntryType::Save); wal::replay_into_cache(&entries, &mut cache); Some(wal::WalFile::open(&wal_path)?) } else if config.wal_enabled { @@ -427,6 +438,25 @@ impl HDF5Memory { None }; + #[cfg(feature = "hnsw")] + let loaded_index = if replay_only_appended { + Self::load_vector_index(path, checkpoint.ann_generation, &cache, n_checkpoint) + } else { + None + }; + #[cfg(not(feature = "hnsw"))] + let _ = ( + n_checkpoint, + replay_only_appended, + checkpoint.ann_generation, + ); + #[cfg(feature = "hnsw")] + let synced_len = if loaded_index.is_some() { + cache.len() + } else { + 0 + }; + Ok(Self { config, cache, @@ -435,14 +465,14 @@ impl HDF5Memory { wal, strategy: None, ephemeral: None, - // Existing data is loaded from disk + WAL replay; mark the index - // dirty so it is (re)built from the cache on the first search. + // Reuse the vector index saved with the checkpoint when there is + // one; otherwise mark it dirty so the first search builds it. #[cfg(feature = "hnsw")] - hnsw: None, + hnsw_dirty: loaded_index.is_none(), #[cfg(feature = "hnsw")] - hnsw_dirty: true, + hnsw_synced_len: synced_len, #[cfg(feature = "hnsw")] - hnsw_synced_len: 0, + hnsw: loaded_index, // No on-disk provenance ledger exists yet (see CLAUDE.md), so // there's no historical hash to verify loaded records against — // the store starts empty and is populated as records are @@ -458,6 +488,99 @@ impl HDF5Memory { }) } + /// Where the vector index graph is kept between sessions. + #[cfg_attr(not(feature = "hnsw"), allow(dead_code))] + fn vector_index_path(store: &Path) -> PathBuf { + store.with_extension("h5.ann") + } + + /// Save the vector index graph next to the store, returning the generation + /// id the checkpoint must record for it. Only an index that exactly mirrors + /// the cache is saved; otherwise any stale sidecar is removed and `None` + /// returned, and the next session rebuilds. Failures are not errors — the + /// sidecar is a cache of derived data. + #[cfg(feature = "hnsw")] + fn persist_vector_index(&self) -> Option { + let path = Self::vector_index_path(&self.config.path); + let index = match self.hnsw.as_ref() { + Some(index) + if !self.hnsw_dirty + && self.hnsw_synced_len == self.cache.embeddings.len() + && index.len() == self.cache.embeddings.len() => + { + index + } + _ => { + let _ = std::fs::remove_file(&path); + return None; + } + }; + static COUNTER: std::sync::atomic::AtomicU64 = std::sync::atomic::AtomicU64::new(0); + let nanos = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos() as u64); + let generation = nanos + ^ (u64::from(std::process::id()) << 32) + ^ COUNTER.fetch_add(1, std::sync::atomic::Ordering::Relaxed); + + let mut bytes = generation.to_le_bytes().to_vec(); + bytes.extend_from_slice(&index.graph_to_bytes()); + let tmp = path.with_extension("ann.tmp"); + let written = + storage::write_synced(&tmp, &bytes).and_then(|()| storage::rename_synced(&tmp, &path)); + match written { + Ok(()) => Some(generation), + Err(_) => { + let _ = std::fs::remove_file(&tmp); + let _ = std::fs::remove_file(&path); + None + } + } + } + + #[cfg(not(feature = "hnsw"))] + fn persist_vector_index(&self) -> Option { + None + } + + /// Load the vector index saved with the checkpoint identified by + /// `generation`, covering the first `n_checkpoint` records of `cache`. + /// Anything unexpected — no sidecar, another generation, a damaged or + /// mismatched graph — yields `None` and the index is rebuilt on demand. + #[cfg(feature = "hnsw")] + fn load_vector_index( + store: &Path, + generation: Option, + cache: &MemoryCache, + n_checkpoint: usize, + ) -> Option { + let generation = generation?; + let bytes = std::fs::read(Self::vector_index_path(store)).ok()?; + let (stamp, graph) = bytes.split_at_checked(8)?; + if u64::from_le_bytes(stamp.try_into().ok()?) != generation { + return None; + } + let vectors = cache.embeddings.get(..n_checkpoint)?.to_vec(); + let mut index = HnswIndex::from_graph_bytes(graph, vectors).ok()?; + if index.dimension() != cache.embedding_dim { + return None; + } + // Records appended since (replayed from the WAL) join incrementally. + for id in n_checkpoint..cache.embeddings.len() { + if cache.embeddings[id].len() != index.dimension() + || index.insert(cache.embeddings[id].clone()) != id + { + return None; + } + } + for (id, &t) in cache.tombstones.iter().enumerate() { + if t != 0 { + index.mark_deleted(id); + } + } + Some(index) + } + /// Bring the keyword index in line with the cache and return it. /// /// Appends need no hook: records the index hasn't seen yet (whatever path @@ -517,13 +640,19 @@ impl HDF5Memory { // Record which WAL prefix this checkpoint contains, so a crash before // the truncate below can't replay those entries a second time. let wal_applied = self.wal.as_ref().map(|w| w.mark()); - storage::write_to_disk_with_mark( + // Written before the .h5 so a crash in between leaves a sidecar whose + // generation matches no checkpoint (ignored), never the reverse. + let ann_generation = self.persist_vector_index(); + storage::write_to_disk_with_meta( &self.config.path, &self.config, &self.cache, &self.sessions, &self.knowledge, - wal_applied, + &schema::CheckpointMeta { + wal_applied, + ann_generation, + }, )?; if let Some(ref mut w) = self.wal { w.truncate()?; @@ -686,6 +815,30 @@ impl HDF5Memory { #[cfg(feature = "hnsw")] fn ensure_hnsw_fresh(&mut self) { let n = self.cache.embeddings.len(); + // Records appended since the index was last in sync (a batch save, or + // any path that pushes to the cache without a hook) are inserted + // incrementally rather than triggering a rebuild of the whole graph. + if !self.hnsw_dirty + && self.hnsw_synced_len < n + && let Some(index) = self.hnsw.as_mut() + && index.len() == self.hnsw_synced_len + { + let dim = index.dimension(); + let appended = (self.hnsw_synced_len..n).all(|id| { + self.cache.embeddings[id].len() == dim + && index.insert(self.cache.embeddings[id].clone()) == id + }); + if appended { + for id in self.hnsw_synced_len..n { + if self.cache.tombstones[id] != 0 { + index.mark_deleted(id); + } + } + self.hnsw_synced_len = n; + } else { + self.hnsw_dirty = true; + } + } if self.hnsw_dirty || self.hnsw_synced_len != n { self.hnsw = self.build_hnsw_from_cache(); self.hnsw_synced_len = n; @@ -928,8 +1081,8 @@ impl AgentMemory for HDF5Memory { ); indices.push(idx); } - // Batch inserts rebuild the index once rather than node-by-node. - self.hnsw_mark_dirty(); + // The vector and keyword indexes pick the new records up + // incrementally the next time they are needed. self.flush()?; Ok(indices) } @@ -987,6 +1140,13 @@ impl AgentMemory for HDF5Memory { if self.wal.as_ref().is_some_and(|w| !w.is_empty()) && wal_path.exists() { storage::snapshot_file(&wal_path, &snapshot.with_extension("h5.wal"))?; } + // The saved vector index belongs to the checkpoint just copied (its + // generation id is in that .h5), so it is valid for the snapshot too. + // Best effort: without it the snapshot simply rebuilds on first search. + let ann_path = Self::vector_index_path(&self.config.path); + if ann_path.exists() { + let _ = storage::snapshot_file(&ann_path, &Self::vector_index_path(&snapshot)); + } Ok(snapshot) } @@ -1674,6 +1834,126 @@ mod tests { assert_eq!(restored.cache.chunks, ["checkpointed", "wal-only"]); } + /// A store with `n` records spread over a few directions, WAL on. + #[cfg(feature = "hnsw")] + fn indexed_store(dir: &TempDir, n: usize) -> (HDF5Memory, PathBuf) { + let mut config = make_config(dir); + config.wal_enabled = true; + config.wal_max_entries = 10_000; + config.compact_threshold = 0.0; + let path = config.path.clone(); + let mut mem = HDF5Memory::create(config).unwrap(); + for i in 0..n { + let a = i as f32 * 0.37; + mem.save(make_entry( + &format!("rec{i}"), + &[a.cos(), a.sin(), (a * 0.5).cos(), 0.1], + )) + .unwrap(); + } + (mem, path) + } + + #[cfg(feature = "hnsw")] + fn top_ids(mem: &mut HDF5Memory, q: &[f32]) -> Vec { + mem.hybrid_search(q, "", 1.0, 0.0, 5) + .into_iter() + .map(|r| r.index) + .collect() + } + + #[cfg(feature = "hnsw")] + #[test] + fn vector_index_is_reloaded_not_rebuilt() { + let dir = TempDir::new().unwrap(); + let (mut mem, path) = indexed_store(&dir, 60); + let q = [0.3f32.cos(), 0.3f32.sin(), 0.9, 0.1]; + let expected = top_ids(&mut mem, &q); // builds the index + mem.flush_wal().unwrap(); // checkpoint + sidecar + drop(mem); + assert!(HDF5Memory::vector_index_path(&path).exists()); + + let mut reopened = HDF5Memory::open(&path).unwrap(); + assert!(!reopened.hnsw_dirty, "index should come from the sidecar"); + assert_eq!(reopened.hnsw.as_ref().unwrap().len(), 60); + assert_eq!(top_ids(&mut reopened, &q), expected); + } + + #[cfg(feature = "hnsw")] + #[test] + fn records_appended_after_the_checkpoint_join_the_loaded_index() { + let dir = TempDir::new().unwrap(); + let (mut mem, path) = indexed_store(&dir, 40); + top_ids(&mut mem, &[1.0, 0.0, 0.0, 0.0]); + mem.flush_wal().unwrap(); + // Only in the WAL when the process "dies". + mem.save(make_entry("late", &[0.0, 0.0, 0.0, 1.0])).unwrap(); + drop(mem); + + let mut reopened = HDF5Memory::open(&path).unwrap(); + assert!(!reopened.hnsw_dirty); + assert_eq!(reopened.hnsw.as_ref().unwrap().len(), 41); + assert_eq!(top_ids(&mut reopened, &[0.0, 0.0, 0.0, 1.0])[0], 40); + } + + #[cfg(feature = "hnsw")] + #[test] + fn replayed_update_invalidates_the_saved_index() { + let dir = TempDir::new().unwrap(); + let (mut mem, path) = indexed_store(&dir, 40); + top_ids(&mut mem, &[1.0, 0.0, 0.0, 0.0]); + mem.flush_wal().unwrap(); + // An in-place update after the checkpoint changes record 0's vector; + // the saved graph was built over the old one. + let mut moved = make_entry("rec0 moved", &[0.0, 0.0, 0.0, 1.0]); + moved.tags = mem.cache.tags[0].clone(); + mem.save_or_update(moved).unwrap(); + let expected = top_ids(&mut mem, &[0.0, 0.0, 0.0, 1.0]); + std::mem::forget(mem); // die without the drop-time checkpoint + + let mut reopened = HDF5Memory::open_read_only(&path).unwrap(); + assert!(reopened.hnsw_dirty, "saved index must not be reused"); + assert_eq!(top_ids(&mut reopened, &[0.0, 0.0, 0.0, 1.0]), expected); + } + + #[cfg(feature = "hnsw")] + #[test] + fn stale_or_damaged_index_sidecar_is_ignored() { + let dir = TempDir::new().unwrap(); + let (mut mem, path) = indexed_store(&dir, 40); + let q = [1.0, 0.0, 0.0, 0.0]; + top_ids(&mut mem, &q); // builds the index + mem.flush_wal().unwrap(); + let ann = HDF5Memory::vector_index_path(&path); + let first_sidecar = std::fs::read(&ann).unwrap(); + // A second checkpoint gets a new generation. + mem.save(make_entry("more", &[0.5, 0.5, 0.0, 0.0])).unwrap(); + top_ids(&mut mem, &q); + mem.flush_wal().unwrap(); + let expected_after = top_ids(&mut mem, &q); + drop(mem); + + // Sidecar from the earlier checkpoint: wrong generation. + std::fs::write(&ann, &first_sidecar).unwrap(); + let mut reopened = HDF5Memory::open_read_only(&path).unwrap(); + assert!(reopened.hnsw_dirty); + assert_eq!(top_ids(&mut reopened, &q), expected_after); + drop(reopened); + + // Right generation, damaged graph. + let mut mem = HDF5Memory::open(&path).unwrap(); + top_ids(&mut mem, &q); + mem.flush_wal().unwrap(); + drop(mem); + let mut bytes = std::fs::read(&ann).unwrap(); + let mid = bytes.len() / 2; + bytes[mid] ^= 0xFF; + std::fs::write(&ann, &bytes).unwrap(); + let mut reopened = HDF5Memory::open_read_only(&path).unwrap(); + assert!(reopened.hnsw_dirty); + assert_eq!(top_ids(&mut reopened, &q), expected_after); + } + #[test] fn keyword_index_stays_in_sync_through_every_mutation() { let dir = TempDir::new().unwrap(); diff --git a/crates/clawhdf5-agent/src/schema.rs b/crates/clawhdf5-agent/src/schema.rs index 5376d13..c7cb92a 100644 --- a/crates/clawhdf5-agent/src/schema.rs +++ b/crates/clawhdf5-agent/src/schema.rs @@ -22,6 +22,7 @@ pub const ZEROCLAW_VERSION: &str = "0.8.0"; /// the checkpoint was taken with an empty WAL. const WAL_APPLIED_LEN_ATTR: &str = "wal_applied_len"; const WAL_APPLIED_CRC_ATTR: &str = "wal_applied_crc"; +const ANN_GENERATION_ATTR: &str = "ann_generation"; /// Build a complete HDF5 file from the in-memory state. pub fn build_hdf5_file( @@ -43,6 +44,34 @@ pub fn build_hdf5_file_with_mark( knowledge: &KnowledgeCache, wal_applied: Option, ) -> Result, MemoryError> { + let meta = CheckpointMeta { + wal_applied, + ann_generation: None, + }; + build_hdf5_file_with_meta(config, cache, sessions, knowledge, &meta) +} + +/// Bookkeeping a checkpoint records in `/meta` beside the store's contents. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct CheckpointMeta { + /// The WAL prefix this checkpoint already contains; see [`WalMark`]. + pub wal_applied: Option, + /// Identifies the vector-index sidecar (`.h5.ann`) written with this + /// checkpoint. A sidecar is loaded only if it carries the same value, so + /// one left over from another checkpoint can never be attached to records + /// it wasn't built from. + pub ann_generation: Option, +} + +/// [`build_hdf5_file`] with checkpoint bookkeeping. +pub fn build_hdf5_file_with_meta( + config: &MemoryConfig, + cache: &MemoryCache, + sessions: &SessionCache, + knowledge: &KnowledgeCache, + checkpoint: &CheckpointMeta, +) -> Result, MemoryError> { + let wal_applied = checkpoint.wal_applied; let mut builder = clawhdf5::FileBuilder::new(); // /meta group with schema attributes @@ -83,6 +112,11 @@ pub fn build_hdf5_file_with_mark( meta.set_attr(WAL_APPLIED_LEN_ATTR, AttrValue::I64(mark.len as i64)); meta.set_attr(WAL_APPLIED_CRC_ATTR, AttrValue::I64(i64::from(mark.crc))); } + if let Some(generation) = checkpoint.ann_generation { + // Stored as the i64 with the same bits; attributes have no u64 scalar + // round trip through every reader. + meta.set_attr(ANN_GENERATION_ATTR, AttrValue::I64(generation as i64)); + } // Need at least one dataset in the group for it to be a proper group meta.create_dataset("_marker").with_u8_data(&[1]).compact(); let finished_meta = meta.finish(); @@ -386,6 +420,22 @@ pub fn read_wal_mark(file: &clawhdf5::File) -> Option { Some(WalMark { len, crc }) } +/// Read the checkpoint bookkeeping from `/meta`. +pub fn read_checkpoint_meta(file: &clawhdf5::File) -> CheckpointMeta { + let ann_generation = file + .group("meta") + .ok() + .and_then(|g| g.attrs().ok()) + .and_then(|attrs| match attrs.get(ANN_GENERATION_ATTR) { + Some(AttrValue::I64(v)) => Some(*v as u64), + _ => None, + }); + CheckpointMeta { + wal_applied: read_wal_mark(file), + ann_generation, + } +} + pub fn validate_and_load( file: &clawhdf5::File, ) -> Result<(MemoryConfig, MemoryCache, SessionCache, KnowledgeCache), MemoryError> { diff --git a/crates/clawhdf5-agent/src/storage.rs b/crates/clawhdf5-agent/src/storage.rs index c9000a1..533bad3 100644 --- a/crates/clawhdf5-agent/src/storage.rs +++ b/crates/clawhdf5-agent/src/storage.rs @@ -34,7 +34,23 @@ pub fn write_to_disk_with_mark( knowledge: &KnowledgeCache, wal_applied: Option, ) -> Result<(), MemoryError> { - let bytes = schema::build_hdf5_file_with_mark(config, cache, sessions, knowledge, wal_applied)?; + let meta = schema::CheckpointMeta { + wal_applied, + ann_generation: None, + }; + write_to_disk_with_meta(path, config, cache, sessions, knowledge, &meta) +} + +/// [`write_to_disk`] with full checkpoint bookkeeping. +pub fn write_to_disk_with_meta( + path: &Path, + config: &MemoryConfig, + cache: &MemoryCache, + sessions: &SessionCache, + knowledge: &KnowledgeCache, + checkpoint: &schema::CheckpointMeta, +) -> Result<(), MemoryError> { + let bytes = schema::build_hdf5_file_with_meta(config, cache, sessions, knowledge, checkpoint)?; if bytes.is_empty() { return Err(MemoryError::Hdf5("build_hdf5_file produced 0 bytes".into())); @@ -47,7 +63,7 @@ pub fn write_to_disk_with_mark( } /// Write `bytes` to `path` and flush them to stable storage. -fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), MemoryError> { +pub(crate) fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), MemoryError> { use std::io::Write; let mut f = std::fs::File::create(path).map_err(MemoryError::Io)?; f.write_all(bytes).map_err(MemoryError::Io)?; @@ -62,7 +78,7 @@ fn write_synced(path: &Path, bytes: &[u8]) -> Result<(), MemoryError> { /// This is per-checkpoint/snapshot cost only (each is already a full file /// write). Individual WAL appends are deliberately not synced — see the /// durability notes in the crate docs. -fn rename_synced(from: &Path, to: &Path) -> Result<(), MemoryError> { +pub(crate) fn rename_synced(from: &Path, to: &Path) -> Result<(), MemoryError> { std::fs::rename(from, to).map_err(MemoryError::Io)?; #[cfg(unix)] if let Some(dir) = to.parent() { @@ -113,6 +129,20 @@ pub fn read_from_disk_with_mark(path: &Path) -> Result<(StoreState, Option Result<(StoreState, schema::CheckpointMeta), MemoryError> { + let mmap = clawhdf5_io::MmapReader::open(path).map_err(MemoryError::Io)?; + mmap.advise_willneed(0, mmap.len()); + let file = clawhdf5::File::from_bytes(mmap.as_bytes().to_vec()) + .map_err(|e| MemoryError::Hdf5(format!("cannot open {}: {e}", path.display())))?; + let (mut config, cache, sessions, knowledge) = schema::validate_and_load(&file)?; + config.path = path.to_path_buf(); + let meta = schema::read_checkpoint_meta(&file); + Ok(((config, cache, sessions, knowledge), meta)) +} + /// Copy an HDF5 file atomically to a destination. pub fn snapshot_file(src: &Path, dest: &Path) -> Result { let dest_file = if dest.is_dir() { diff --git a/crates/clawhdf5-ann/src/hnsw.rs b/crates/clawhdf5-ann/src/hnsw.rs index 5755f2b..2644add 100644 --- a/crates/clawhdf5-ann/src/hnsw.rs +++ b/crates/clawhdf5-ann/src/hnsw.rs @@ -134,6 +134,9 @@ impl Ord for FarCandidate { } } +/// Magic for [`HnswIndex::graph_to_bytes`]. +const GRAPH_MAGIC: &[u8; 4] = b"CHG1"; + /// On-disk format version for the serialized HNSW index. /// /// - Version 1: original layout (`vectors`, `graph_layer_*`, `config`), no @@ -690,6 +693,160 @@ impl HnswIndex { }) } + /// Serialize the **graph only** — levels, tombstones and adjacency, not the + /// vectors — for a caller that already stores the vectors elsewhere (the + /// agent's record cache). [`HnswIndex::to_hdf5_bytes`] writes a complete, + /// self-contained index including a full copy of every vector, which would + /// double such a store's size. Reattach with + /// [`HnswIndex::from_graph_bytes`]. + /// + /// Layout (little endian): magic `CHG1`, then u32 fields `n`, `m`, + /// `m_max0`, `ef_construction`, `entry_point`, `num_layers`, `metric`; + /// `n` level bytes; `n` tombstone bytes; per layer, per node that exists on + /// that layer: u32 neighbour count + u32 ids; trailing CRC32 of all of it. + pub fn graph_to_bytes(&self) -> Vec { + let n = self.vectors.len(); + let mut out = Vec::with_capacity(32 + n * 2 + n * self.m_max0 * 4); + out.extend_from_slice(GRAPH_MAGIC); + for field in [ + n, + self.m, + self.m_max0, + self.ef_construction, + self.entry_point, + self.graph.len(), + match self.metric { + DistanceMetric::L2 => 0, + DistanceMetric::Cosine => 1, + }, + ] { + out.extend_from_slice(&(field as u32).to_le_bytes()); + } + out.extend(self.node_levels.iter().map(|&l| l.min(255) as u8)); + out.extend(self.deleted.iter().map(|&d| u8::from(d))); + for (layer, adjacency) in self.graph.iter().enumerate() { + for (node, neighbors) in adjacency.iter().enumerate() { + if self.node_levels[node] < layer { + continue; // node does not exist on this layer + } + out.extend_from_slice(&(neighbors.len() as u32).to_le_bytes()); + for &id in neighbors { + out.extend_from_slice(&(id as u32).to_le_bytes()); + } + } + } + let crc = clawhdf5_format::checksum::crc32(&out); + out.extend_from_slice(&crc.to_le_bytes()); + out + } + + /// Rebuild an index from [`HnswIndex::graph_to_bytes`] output and the + /// vectors it was built over (same order). Every structural claim in + /// `bytes` is validated — a corrupt or mismatched graph is an error, never + /// an index that panics or walks out of bounds during a search. + pub fn from_graph_bytes(bytes: &[u8], vectors: Vec>) -> Result { + let bad = |what: &str| FormatError::SerializationError(format!("HNSW graph: {what}")); + let body_len = bytes + .len() + .checked_sub(4) + .filter(|&l| l >= GRAPH_MAGIC.len() + 7 * 4) + .ok_or_else(|| bad("truncated"))?; + let (body, crc_bytes) = bytes.split_at(body_len); + if &body[..4] != GRAPH_MAGIC { + return Err(bad("bad magic")); + } + let stored_crc = + u32::from_le_bytes([crc_bytes[0], crc_bytes[1], crc_bytes[2], crc_bytes[3]]); + if clawhdf5_format::checksum::crc32(body) != stored_crc { + return Err(bad("checksum mismatch")); + } + + let mut pos = 4; + let next_u32 = |pos: &mut usize| -> Result { + let b = body.get(*pos..*pos + 4).ok_or_else(|| bad("truncated"))?; + *pos += 4; + Ok(u32::from_le_bytes([b[0], b[1], b[2], b[3]]) as usize) + }; + let n = next_u32(&mut pos)?; + let m = next_u32(&mut pos)?; + let m_max0 = next_u32(&mut pos)?; + let ef_construction = next_u32(&mut pos)?; + let entry_point = next_u32(&mut pos)?; + let num_layers = next_u32(&mut pos)?; + let metric = match next_u32(&mut pos)? { + 0 => DistanceMetric::L2, + 1 => DistanceMetric::Cosine, + _ => return Err(bad("unknown metric")), + }; + if n != vectors.len() { + return Err(bad("vector count does not match the graph")); + } + if n == 0 || entry_point >= n || m < 2 || num_layers == 0 || num_layers > 256 { + return Err(bad("invalid header")); + } + let dim = vectors[0].len(); + if vectors.iter().any(|v| v.len() != dim) { + return Err(bad("vectors have mixed dimensions")); + } + + let levels = body.get(pos..pos + n).ok_or_else(|| bad("truncated"))?; + pos += n; + let node_levels: Vec = levels.iter().map(|&l| l as usize).collect(); + if node_levels.iter().any(|&l| l >= num_layers) + || node_levels[entry_point] + 1 != num_layers + { + return Err(bad("levels inconsistent with layer count")); + } + let deleted: Vec = body + .get(pos..pos + n) + .ok_or_else(|| bad("truncated"))? + .iter() + .map(|&d| d != 0) + .collect(); + pos += n; + + let mut graph: Vec>> = Vec::with_capacity(num_layers); + for layer in 0..num_layers { + let max_conn = if layer == 0 { m_max0 } else { m }; + let mut adjacency = vec![Vec::new(); n]; + for (node, slot) in adjacency.iter_mut().enumerate() { + if node_levels[node] < layer { + continue; + } + let count = next_u32(&mut pos)?; + if count > max_conn { + return Err(bad("neighbour list exceeds the connection limit")); + } + let mut neighbors = Vec::with_capacity(count); + for _ in 0..count { + let id = next_u32(&mut pos)?; + // A neighbour must exist, and exist on this layer. + if id >= n || node_levels[id] < layer { + return Err(bad("neighbour id out of range for its layer")); + } + neighbors.push(id); + } + *slot = neighbors; + } + graph.push(adjacency); + } + if pos != body.len() { + return Err(bad("trailing bytes")); + } + + Ok(Self { + vectors, + graph, + deleted, + entry_point, + m, + m_max0, + ef_construction, + node_levels, + metric, + }) + } + /// Returns the number of vectors in the index. pub fn len(&self) -> usize { self.vectors.len() @@ -1147,6 +1304,73 @@ mod tests { assert!(recall >= 0.95, "incremental recall@10 = {recall}"); } + #[test] + fn graph_bytes_round_trip_gives_identical_searches() { + let mut vectors = clustered(1260, 16, 12, 9); + let queries = vectors.split_off(1200); + let mut index = HnswIndex::build_with_metric(&vectors, 8, 40, DistanceMetric::L2); + index.mark_deleted(3); + index.mark_deleted(700); + + let bytes = index.graph_to_bytes(); + // The graph is a small fraction of the vectors it indexes... not + // necessarily at dim 16, but it must not embed them. + assert!(bytes.len() < 1200 * (16 * 2 + 2) * 4); + let restored = HnswIndex::from_graph_bytes(&bytes, vectors.clone()).unwrap(); + assert_eq!(restored.deleted_count(), 2); + for q in &queries { + assert_eq!(restored.search(q, 10, 50), index.search(q, 10, 50)); + } + // A restored index keeps working incrementally. + let mut restored = restored; + let id = restored.insert(queries[0].clone()); + assert_eq!(restored.search(&queries[0], 1, 50)[0].0, id); + } + + #[test] + fn damaged_or_mismatched_graph_bytes_are_errors() { + let vectors = clustered(300, 8, 6, 4); + let index = HnswIndex::build_with_metric(&vectors, 6, 30, DistanceMetric::Cosine); + let bytes = index.graph_to_bytes(); + + // Wrong vector set. + assert!(HnswIndex::from_graph_bytes(&bytes, vectors[..299].to_vec()).is_err()); + // Every truncation. + for len in 0..bytes.len() { + assert!( + HnswIndex::from_graph_bytes(&bytes[..len], vectors.clone()).is_err(), + "truncated to {len}" + ); + } + // A flipped bit anywhere. + for i in (0..bytes.len()).step_by(7) { + let mut damaged = bytes.clone(); + damaged[i] ^= 0x10; + assert!( + HnswIndex::from_graph_bytes(&damaged, vectors.clone()).is_err(), + "bit flip at {i}" + ); + } + } + + #[test] + fn structurally_invalid_graph_with_a_valid_checksum_is_rejected() { + // The CRC only proves the bytes are what was written; a hostile or + // buggy writer can checksum nonsense. Out-of-range neighbour ids must + // still be caught, or search would index out of bounds. + let vectors = clustered(50, 4, 3, 5); + let index = HnswIndex::build_with_metric(&vectors, 4, 20, DistanceMetric::L2); + let mut bytes = index.graph_to_bytes(); + let body_len = bytes.len() - 4; + // First neighbour id of node 0 on layer 0 sits right after the header, + // levels, tombstones and node 0's count. + let at = 4 + 7 * 4 + 50 + 50 + 4; + bytes[at..at + 4].copy_from_slice(&9999u32.to_le_bytes()); + let crc = clawhdf5_format::checksum::crc32(&bytes[..body_len]); + bytes[body_len..].copy_from_slice(&crc.to_le_bytes()); + assert!(HnswIndex::from_graph_bytes(&bytes, vectors).is_err()); + } + #[test] fn select_neighbors_prefers_diverse_directions_and_fills_up() { // Node at the origin. Three candidates bunched together on the right, diff --git a/crates/clawhdf5-bench/src/bin/search_harness.rs b/crates/clawhdf5-bench/src/bin/search_harness.rs index d6e5d19..4e50c75 100644 --- a/crates/clawhdf5-bench/src/bin/search_harness.rs +++ b/crates/clawhdf5-bench/src/bin/search_harness.rs @@ -7,8 +7,8 @@ //! * **ANN** — index build time, and for each `ef`: recall@10 against an exact //! brute-force scan, queries/second, and p50/p99 latency. //! * **End to end** — `HDF5Memory`: ingest time, checkpoint time, `open()` -//! time, the first query after open (which pays for any index rebuild), and -//! steady-state `hybrid_search` p50/p99 at each store size. +//! time, the one-off cold index build (first query ever), the first query +//! after a reopen, and steady-state `hybrid_search` p50/p99 at each size. //! //! Data is *clustered* (points = cluster centre + noise, unit-normalised), not //! uniform: uniform random high-dimensional vectors are nearly equidistant, @@ -310,6 +310,13 @@ fn bench_end_to_end(n: usize, json: &mut Vec) { let t = Instant::now(); mem.save_batch(entries).unwrap(); let ingest = t.elapsed(); + // The very first query builds the vector and keyword indexes from + // scratch. It happens once per store, not once per session: the checkpoint + // below saves the vector index, so a later `open()` reloads it. + let t = Instant::now(); + std::hint::black_box(mem.hybrid_search(&data.queries[1], &query_texts[1], 0.7, 0.3, K)); + let cold_build = t.elapsed(); + let t = Instant::now(); mem.flush_wal().unwrap(); let checkpoint = t.elapsed(); @@ -343,8 +350,9 @@ fn bench_end_to_end(n: usize, json: &mut Vec) { ); println!( - "| {n} | {:.0} | {:.1} | {:.1} | {:.1} | {:.2} | {:.2} | {:.1} |", + "| {n} | {:.0} | {:.0} | {:.1} | {:.1} | {:.1} | {:.2} | {:.2} | {:.1} |", millis(ingest), + millis(cold_build), millis(checkpoint), millis(open), millis(first_query), @@ -354,7 +362,8 @@ fn bench_end_to_end(n: usize, json: &mut Vec) { ); json.push(serde_json::json!({ "bench": "hybrid_search", "n": n, - "ingest_ms": millis(ingest), "checkpoint_ms": millis(checkpoint), + "ingest_ms": millis(ingest), "cold_index_build_ms": millis(cold_build), + "checkpoint_ms": millis(checkpoint), "open_ms": millis(open), "first_query_ms": millis(first_query), "p50_ms": millis(steady.p50), "p99_ms": millis(steady.p99), "qps": steady.qps, })); @@ -394,9 +403,9 @@ fn main() { } println!("\n### End to end: `HDF5Memory::hybrid_search` (k = {K}, weights 0.7 / 0.3)\n"); println!( - "| N | ingest ms | checkpoint ms | open ms | first query ms | p50 ms | p99 ms | QPS |" + "| N | ingest ms | cold index build ms | checkpoint ms | open ms | first query after open ms | p50 ms | p99 ms | QPS |" ); - println!("|---:|---:|---:|---:|---:|---:|---:|---:|"); + println!("|---:|---:|---:|---:|---:|---:|---:|---:|---:|"); for &n in sizes { bench_end_to_end(n, &mut json); } From 39bf2bebf497df1762ef053c7ef259debce185b0 Mon Sep 17 00:00:00 2001 From: osobh Date: Sat, 19 Sep 2026 08:01:06 -0700 Subject: [PATCH 6/6] =?UTF-8?q?docs:=20CLAUDE.md=20=E2=80=94=20search=20pa?= =?UTF-8?q?th=20after=20the=20hot-path=20work?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-Authored-By: Claude Fable 5.1 --- CLAUDE.md | 10 ++++++++++ 1 file changed, 10 insertions(+) diff --git a/CLAUDE.md b/CLAUDE.md index 041cfa3..2a47ad5 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,6 +33,16 @@ Cargo workspace with 16 crates under `crates/` (plus `libaec-sys`, an internal F the approximate `clawhdf5-ann` index for the vector stage (the index mirrors the cache and self-heals on drift). Build the agent with `--no-default-features --features float16` to force the exact linear cosine scan. + The index uses the HNSW paper's diversity heuristic for neighbour selection + (plain closest-M capped recall on clustered data: 0.31 recall@10 at 100K). Its + graph is saved to `.h5.ann` at each checkpoint and reloaded by `open()` + (tied to the checkpoint by a generation id; stale/damaged sidecars are + ignored and the index rebuilt). `hybrid_search` keeps one incremental BM25 + index for the life of the store and never writes the store: Hebbian + activation boosts are persisted by the next checkpoint (or on drop), not per + query. Measure any search-path change with + `cargo run --release -p clawhdf5-bench --bin search_harness` (baselines in + `BENCHMARKS.md`). - WAL (write-ahead log) for crash-safe persistence, with a chained CRC32 trailer per entry (each entry's CRC folds in the previous entry's CRC) so a corrupted, reordered, duplicated, or spliced entry stops replay cleanly