diff --git a/crates/ruvector-core/src/index.rs b/crates/ruvector-core/src/index.rs index eadb730be6..e073f0c9c1 100644 --- a/crates/ruvector-core/src/index.rs +++ b/crates/ruvector-core/src/index.rs @@ -23,6 +23,17 @@ pub trait VectorIndex: Send + Sync { /// Search for k nearest neighbors fn search(&self, query: &[f32], k: usize) -> Result>; + /// Search for k nearest neighbors with a caller-supplied candidate-list + /// width. Indexes without a breadth parameter (e.g. brute force) ignore it. + fn search_with_ef( + &self, + query: &[f32], + k: usize, + _ef_search: usize, + ) -> Result> { + self.search(query, k) + } + /// Remove a vector from the index fn remove(&mut self, id: &VectorId) -> Result; diff --git a/crates/ruvector-core/src/index/hnsw.rs b/crates/ruvector-core/src/index/hnsw.rs index 4a63cad82a..b53b932651 100644 --- a/crates/ruvector-core/src/index/hnsw.rs +++ b/crates/ruvector-core/src/index/hnsw.rs @@ -401,7 +401,16 @@ impl VectorIndex for HnswIndex { fn search(&self, query: &[f32], k: usize) -> Result> { // Use configured ef_search - self.search_with_ef(query, k, self.config.ef_search) + HnswIndex::search_with_ef(self, query, k, self.config.ef_search) + } + + fn search_with_ef( + &self, + query: &[f32], + k: usize, + ef_search: usize, + ) -> Result> { + HnswIndex::search_with_ef(self, query, k, ef_search) } fn remove(&mut self, id: &VectorId) -> Result { @@ -539,6 +548,114 @@ mod tests { Ok(()) } + /// Deterministic unit vector, mirroring the generator used by the + /// `hnsw_completeness_test` integration tests. + fn seeded_unit_vector(seed: u64, dims: usize) -> Vec { + let mut x = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; + let mut v = Vec::with_capacity(dims); + for _ in 0..dims { + x ^= x >> 12; + x ^= x << 25; + x ^= x >> 27; + let bits = x.wrapping_mul(0x2545_F491_4F6C_DD1D); + v.push((bits >> 40) as f32 / 8_388_608.0 - 1.0); + } + normalize_vector(&v) + } + + /// Walk the layer-0 graph from the entry point and return the set of + /// external ids reachable, plus the total number of points in the graph. + /// + /// The entry point is the first point stored in the highest occupied + /// layer: `check_entry_point` only replaces the entry point on a strictly + /// greater level, and `points_by_layer` preserves insertion order. + fn layer0_reachable(index: &HnswIndex) -> (std::collections::HashSet, usize) { + use std::collections::HashSet; + + let inner = index.inner.read(); + let indexation = inner.hnsw.get_point_indexation(); + let max_level = indexation.get_max_level_observed() as usize; + let total = indexation.get_nb_point(); + + let entry = indexation + .get_layer_iterator(max_level) + .next() + .expect("entry point"); + + // Map point id -> (origin id, layer-0 neighbour point ids) for the + // whole graph so the walk can follow edges without re-locking. + let mut adjacency = std::collections::HashMap::new(); + for level in 0..=max_level { + for point in indexation.get_layer_iterator(level) { + let neighbours = point.get_neighborhood_id(); + let layer0: Vec<_> = neighbours + .first() + .map(|n| n.iter().map(|nb| nb.p_id).collect()) + .unwrap_or_default(); + adjacency.insert(point.get_point_id(), (point.get_origin_id(), layer0)); + } + } + + let mut reachable = HashSet::new(); + let mut stack = vec![entry.get_point_id()]; + let mut visited = HashSet::new(); + while let Some(p_id) = stack.pop() { + if !visited.insert(p_id) { + continue; + } + if let Some((origin, neighbours)) = adjacency.get(&p_id) { + reachable.insert(*origin); + stack.extend(neighbours.iter().copied()); + } + } + + (reachable, total) + } + + /// Every inserted point must be reachable from the entry point by + /// following layer-0 edges. A point with no layer-0 in-edge is an orphan: + /// no `efSearch` can recover it, because search widens the frontier but + /// never reaches a node nothing points at (issue #773). + #[test] + fn test_hnsw_layer0_reachability_invariant() -> Result<()> { + let config = HnswConfig { + m: 16, + ef_construction: 100, + ef_search: 100, + max_elements: 1_000, + }; + + // The failure is level-assignment dependent and hits a fraction of a + // percent of small graphs, so the sweep has to be wide enough that a + // regression cannot slip through on luck. + for trial in 0..1_500u64 { + for &rows in &[2usize, 3, 5, 9] { + let mut index = HnswIndex::new(64, DistanceMetric::Cosine, config.clone())?; + for i in 0..rows { + index.add( + format!("m{i}"), + seeded_unit_vector(trial * 1_000 + i as u64, 64), + )?; + } + + let (reachable, total) = layer0_reachable(&index); + assert_eq!(total, rows, "trial {trial}: point count drifted"); + assert_eq!( + reachable.len(), + rows, + "trial {trial} rows={rows}: only {} of {rows} points are reachable \ + from the entry point on layer 0 (orphaned origin ids: {:?})", + reachable.len(), + (0..rows) + .filter(|i| !reachable.contains(i)) + .collect::>() + ); + } + } + + Ok(()) + } + #[test] fn test_dimension_mismatch() -> Result<()> { let config = HnswConfig::default(); diff --git a/crates/ruvector-core/src/vector_db.rs b/crates/ruvector-core/src/vector_db.rs index 5e594b99fe..160f6b9710 100644 --- a/crates/ruvector-core/src/vector_db.rs +++ b/crates/ruvector-core/src/vector_db.rs @@ -187,7 +187,13 @@ impl VectorDB { /// Search for similar vectors pub fn search(&self, query: SearchQuery) -> Result> { let index = self.index.read(); - let mut results = index.search(&query.vector, query.k)?; + // Honour a per-query efSearch override; without this the field was + // accepted and silently discarded, so callers widening the search to + // chase missing rows saw no change at all (issue #773). + let mut results = match query.ef_search { + Some(ef_search) => index.search_with_ef(&query.vector, query.k, ef_search)?, + None => index.search(&query.vector, query.k)?, + }; // Enrich results with full data if needed for result in &mut results { diff --git a/crates/ruvector-core/tests/hnsw_completeness_test.rs b/crates/ruvector-core/tests/hnsw_completeness_test.rs new file mode 100644 index 0000000000..f2556f9f42 --- /dev/null +++ b/crates/ruvector-core/tests/hnsw_completeness_test.rs @@ -0,0 +1,286 @@ +//! Regression tests for issue #773: `search()` silently omitted stored rows. +//! +//! A point inserted at HNSW level >= 1 never received reciprocal edges on the +//! lower layers, so once it stopped being the graph entry point nothing on +//! layer 0 pointed at it and no amount of `efSearch` could reach it again. +//! These tests pin the invariant that every stored id is retrievable. + +#![cfg(feature = "hnsw")] + +use ruvector_core::index::hnsw::HnswIndex; +use ruvector_core::index::VectorIndex; +use ruvector_core::types::{DistanceMetric, HnswConfig}; + +/// Deterministic unit-vector generator (xorshift64*, seeded per vector). +fn unit_vector(seed: u64, dims: usize) -> Vec { + let mut x = seed.wrapping_mul(0x9E37_79B9_7F4A_7C15) | 1; + let mut v = Vec::with_capacity(dims); + for _ in 0..dims { + x ^= x >> 12; + x ^= x << 25; + x ^= x >> 27; + let bits = x.wrapping_mul(0x2545_F491_4F6C_DD1D); + v.push((bits >> 40) as f32 / 8_388_608.0 - 1.0); + } + let norm = v.iter().map(|a| a * a).sum::().sqrt(); + if norm > 0.0 { + for a in &mut v { + *a /= norm; + } + } + v +} + +fn build_index(trial: u64, rows: usize, dims: usize) -> HnswIndex { + let config = HnswConfig { + m: 16, + ef_construction: 100, + ef_search: 100, + max_elements: 1_000, + }; + let mut index = HnswIndex::new(dims, DistanceMetric::Cosine, config).expect("index"); + for i in 0..rows { + index + .add(format!("m{i}"), unit_vector(trial * 1_000 + i as u64, dims)) + .expect("add"); + } + index +} + +/// Every stored id must come back when `k` is at least the row count. +/// +/// This is the exact shape of the reported reproduction: 3 rows of 384-d +/// cosine vectors, queried with `k` and `efSearch` far above the row count. +#[test] +fn search_returns_every_stored_row() { + const DIMS: usize = 384; + const ROWS: usize = 3; + const TRIALS: u64 = 200; + + let mut short_trials = Vec::new(); + + for trial in 0..TRIALS { + let index = build_index(trial, ROWS, DIMS); + // Query with the first row's own vector, as the reproduction does. + let query = unit_vector(trial * 1_000, DIMS); + let hits = index.search_with_ef(&query, 64, 256).expect("search"); + + let mut seen: Vec<_> = hits.iter().map(|h| h.id.as_str()).collect(); + seen.sort_unstable(); + seen.dedup(); + if seen.len() != ROWS { + short_trials.push((trial, seen.len())); + } + } + + assert!( + short_trials.is_empty(), + "search() omitted stored rows in {}/{} trials (trial, returned): {:?}", + short_trials.len(), + TRIALS, + &short_trials[..short_trials.len().min(10)] + ); +} + +/// Probe each row with its own vector — the detection recipe from issue #773. +/// A nearest-neighbour search for a stored vector must return the row holding +/// that exact vector. +#[test] +fn every_row_finds_itself() { + const DIMS: usize = 128; + const ROWS: usize = 24; + + for trial in 0..40u64 { + let index = build_index(trial, ROWS, DIMS); + for i in 0..ROWS { + let id = format!("m{i}"); + let query = unit_vector(trial * 1_000 + i as u64, DIMS); + let hits = index + .search_with_ef(&query, ROWS, 4 * ROWS) + .expect("search"); + assert!( + hits.iter().any(|h| h.id == id), + "trial {trial}: probing {id} with its own vector did not return it \ + (returned {} of {ROWS} rows)", + hits.len() + ); + } + } +} + +/// End-to-end shape of the report in issue #773: a `VectorDB` holding three +/// 384-d cosine rows, queried with `k` and `efSearch` far above the row count, +/// must return every row. +#[test] +#[cfg(feature = "storage")] +fn vector_db_search_returns_every_stored_row() { + use ruvector_core::types::{DbOptions, SearchQuery, VectorEntry}; + use ruvector_core::VectorDB; + + const DIMS: usize = 384; + const ROWS: usize = 3; + // Opening a store per trial is the expensive part, so this test only has + // to prove the end-to-end path; the statistical net is the much cheaper + // index-level sweep above. Seeds 24 and 25 both failed before the fix. + const TRIALS: u64 = 30; + + let dir = tempfile::tempdir().expect("tempdir"); + let mut short_trials = Vec::new(); + + for trial in 0..TRIALS { + let options = DbOptions { + dimensions: DIMS, + distance_metric: DistanceMetric::Cosine, + storage_path: dir + .path() + .join(format!("t{trial}.db")) + .to_string_lossy() + .into_owned(), + hnsw_config: Some(HnswConfig { + m: 16, + ef_construction: 100, + ef_search: 100, + max_elements: 1_000, + }), + quantization: None, + }; + let db = VectorDB::new(options).expect("db"); + + for i in 0..ROWS { + db.insert(VectorEntry { + id: Some(format!("m{i}")), + vector: unit_vector(trial * 1_000 + i as u64, DIMS), + metadata: None, + }) + .expect("insert"); + } + + let hits = db + .search(SearchQuery { + vector: unit_vector(trial * 1_000, DIMS), + k: 64, + filter: None, + ef_search: Some(256), + }) + .expect("search"); + + if hits.len() != db.len().expect("len") { + short_trials.push((trial, hits.len())); + } + } + + assert!( + short_trials.is_empty(), + "VectorDB::search omitted stored rows in {}/{TRIALS} trials (trial, returned): {:?}", + short_trials.len(), + short_trials + ); +} + +/// `add_batch` switches to `parallel_insert_slice` at 10 000 entries, a code +/// path with different interleaving of the reciprocal-edge update. Probe every +/// id with its own vector to confirm none was left orphaned. Ignored by default +/// because building a 12 000-point graph is slow in a debug build. +#[test] +#[ignore = "builds a 12k-point graph to exercise the parallel insert path"] +fn parallel_batch_insert_leaves_no_orphans() { + const DIMS: usize = 64; + const ROWS: usize = 12_000; + + let config = HnswConfig { + m: 16, + ef_construction: 200, + ef_search: 200, + max_elements: 20_000, + }; + let mut index = HnswIndex::new(DIMS, DistanceMetric::Cosine, config).expect("index"); + let entries: Vec<_> = (0..ROWS) + .map(|i| (format!("m{i}"), unit_vector(i as u64, DIMS))) + .collect(); + index.add_batch(entries.clone()).expect("add_batch"); + + let missing: Vec<_> = entries + .iter() + .filter(|(id, vector)| { + let hits = index.search_with_ef(vector, 10, 200).expect("search"); + !hits.iter().any(|h| &h.id == id) + }) + .map(|(id, _)| id.clone()) + .collect(); + + assert!( + missing.is_empty(), + "{} of {ROWS} rows could not find themselves after a parallel batch insert: {:?}", + missing.len(), + &missing[..missing.len().min(10)] + ); +} + +/// Wide scan used to quantify the omission rate. Ignored by default because it +/// builds tens of thousands of graphs; run with +/// `cargo test -p ruvector-core --test hnsw_completeness_test -- --ignored --nocapture`. +#[test] +#[ignore = "long-running rate measurement, not a pass/fail regression guard"] +fn omission_rate_scan() { + const DIMS: usize = 384; + const TRIALS: u64 = 5_000; + + let mut total = 0usize; + let mut short = 0usize; + for &rows in &[2usize, 3, 4, 5, 8] { + let mut short_for_rows = 0usize; + for trial in 0..TRIALS { + let index = build_index(trial, rows, DIMS); + let query = unit_vector(trial * 1_000, DIMS); + let hits = index.search_with_ef(&query, 64, 256).expect("search"); + let mut ids: Vec<_> = hits.iter().map(|h| h.id.as_str()).collect(); + ids.sort_unstable(); + ids.dedup(); + if ids.len() != rows { + short_for_rows += 1; + } + total += 1; + } + short += short_for_rows; + println!( + "rows={rows}: {short_for_rows}/{TRIALS} trials returned an incomplete result set \ + ({:.2}%)", + short_for_rows as f64 * 100.0 / TRIALS as f64 + ); + } + println!( + "overall: {short}/{total} ({:.2}%)", + short as f64 * 100.0 / total as f64 + ); +} + +/// Randomized sweep over sizes and seeds so a future regression surfaces even +/// if it only shows up at a different scale. +#[test] +fn search_is_complete_across_sizes_and_seeds() { + for &dims in &[16usize, 64, 384] { + for &rows in &[2usize, 3, 5, 17, 40] { + for trial in 0..12u64 { + let seed = trial * 7 + dims as u64 + rows as u64 * 31; + let index = build_index(seed, rows, dims); + + for probe in 0..rows { + let query = unit_vector(seed * 1_000 + probe as u64, dims); + let hits = index + .search_with_ef(&query, rows, (4 * rows).max(64)) + .expect("search"); + let mut ids: Vec<_> = hits.iter().map(|h| h.id.as_str()).collect(); + ids.sort_unstable(); + ids.dedup(); + assert_eq!( + ids.len(), + rows, + "dims={dims} rows={rows} seed={seed} probe={probe}: \ + search returned {} distinct ids, expected {rows} ({ids:?})", + ids.len() + ); + } + } + } + } +} diff --git a/crates/rvf/rvf-runtime/src/filter.rs b/crates/rvf/rvf-runtime/src/filter.rs index 4145970015..104d0f9c21 100644 --- a/crates/rvf/rvf-runtime/src/filter.rs +++ b/crates/rvf/rvf-runtime/src/filter.rs @@ -60,29 +60,106 @@ impl FilterValue { } } +/// The `META_SEG` schema type a stored value declares for its field, or `None` +/// for values that constrain no type of their own. +/// +/// Mirrors the mapping `build_metadata_schema` applies when it encodes a +/// generation; the discriminant is `rvf_types::metadata::MetadataType as u8`. +fn declared_type(value: &MetadataValue) -> Option { + use rvf_types::metadata::MetadataType; + Some(match value { + MetadataValue::Null | MetadataValue::DeleteField => return None, + MetadataValue::String(_) => MetadataType::String as u8, + MetadataValue::Bytes(_) => MetadataType::Bytes as u8, + MetadataValue::I64(_) => MetadataType::I64 as u8, + MetadataValue::U64(_) => MetadataType::U64 as u8, + MetadataValue::F64(_) => MetadataType::F64 as u8, + MetadataValue::Bool(_) => MetadataType::Bool as u8, + }) +} + /// In-memory metadata store for filter evaluation. /// Maps vector IDs to their complete durable metadata record. #[derive(Clone)] pub(crate) struct MetadataStore { entries: std::collections::BTreeMap>, + /// How many live records declare each field id with each value type. + /// + /// A `META_SEG` declares one schema entry per field id, so a field carried + /// by two types at once cannot be encoded as a full snapshot. Maintaining + /// the counts as records are written makes that state detectable in + /// `O(fields)` at ingest time instead of requiring a full rescan of every + /// record on every commit (issue #772). + field_types: std::collections::BTreeMap>, } impl MetadataStore { pub(crate) fn new() -> Self { Self { entries: std::collections::BTreeMap::new(), + field_types: std::collections::BTreeMap::new(), } } + /// Count `value` towards its field's live type set. + fn declare(&mut self, field_id: u16, value: &MetadataValue) { + let Some(value_type) = declared_type(value) else { + return; + }; + *self + .field_types + .entry(field_id) + .or_default() + .entry(value_type) + .or_insert(0) += 1; + } + + /// Discount a value that is no longer live, dropping the field's entry + /// once nothing declares it any more. + fn undeclare(&mut self, field_id: u16, value: &MetadataValue) { + let Some(value_type) = declared_type(value) else { + return; + }; + let Some(counts) = self.field_types.get_mut(&field_id) else { + return; + }; + if let Some(count) = counts.get_mut(&value_type) { + *count = count.saturating_sub(1); + if *count == 0 { + counts.remove(&value_type); + } + } + if counts.is_empty() { + self.field_types.remove(&field_id); + } + } + + /// The lowest field id that two live records give different value types. + /// + /// `None` means the live records can be encoded as one full snapshot. + pub(crate) fn conflicting_field(&self) -> Option { + self.field_types + .iter() + .find(|(_, types)| types.len() > 1) + .map(|(&field_id, _)| field_id) + } + /// Add metadata for a vector. `fields` are (field_id, value) pairs. pub(crate) fn insert(&mut self, vector_id: u64, fields: Vec<(u16, MetadataValue)>) { - let record = self.entries.entry(vector_id).or_default(); + self.entries.entry(vector_id).or_default(); for (field_id, value) in fields { - if matches!(value, MetadataValue::DeleteField) { - record.remove(&field_id); - } else { - record.insert(field_id, value); + let previous = { + let record = self.entries.entry(vector_id).or_default(); + if matches!(value, MetadataValue::DeleteField) { + record.remove(&field_id) + } else { + record.insert(field_id, value.clone()) + } + }; + if let Some(previous) = previous { + self.undeclare(field_id, &previous); } + self.declare(field_id, &value); } } @@ -124,7 +201,13 @@ impl MetadataStore { /// Drop every record whose vector identifier fails `keep`. pub(crate) fn retain_ids(&mut self, keep: impl Fn(u64) -> bool) { - self.entries.retain(|&vector_id, _| keep(vector_id)); + let dropped: Vec = self + .entries + .keys() + .copied() + .filter(|&vector_id| !keep(vector_id)) + .collect(); + self.remove_ids(&dropped); } pub(crate) fn decoded_size(&self, vector_id: u64) -> usize { @@ -144,7 +227,12 @@ impl MetadataStore { /// Remove all metadata for the given vector IDs. pub(crate) fn remove_ids(&mut self, ids: &[u64]) { for id in ids { - self.entries.remove(id); + let Some(record) = self.entries.remove(id) else { + continue; + }; + for (field_id, value) in record { + self.undeclare(field_id, &value); + } } } diff --git a/crates/rvf/rvf-runtime/src/store.rs b/crates/rvf/rvf-runtime/src/store.rs index b83502338f..da17cba18a 100644 --- a/crates/rvf/rvf-runtime/src/store.rs +++ b/crates/rvf/rvf-runtime/src/store.rs @@ -523,10 +523,26 @@ impl RvfStore { let old_metadata = self.metadata.clone(); let old_chain = self.metadata_chain; let old_directory = self.segment_dir.clone(); + let old_deletion_bitmap = self.deletion_bitmap.clone(); let old_epoch = self.epoch; let old_witness_hash = self.last_witness_hash; - let candidate_metadata = self.build_ingest_metadata(&metadata, &valid_ids)?; + // Writing an identifier makes it live again: the tombstone described + // the vector this batch is replacing, not the identifier itself. It is + // cleared before the record is built and before the generation is + // encoded, so the new vector is queryable, carries the metadata this + // batch supplies rather than the record the delete removed, and + // survives the next `compact()` instead of being reclaimed as dead + // (issue #748). + self.deletion_bitmap.clear_ids(&valid_ids); + + let candidate_metadata = match self.build_ingest_metadata(&metadata, &valid_ids) { + Ok(candidate) => candidate, + Err(error) => { + self.deletion_bitmap = old_deletion_bitmap; + return Err(error); + } + }; let has_metadata = !matches!(metadata, IngestMetadata::None); let writer = self @@ -607,6 +623,7 @@ impl RvfStore { self.metadata_chain = old_chain; self.vectors = old_vectors; self.segment_dir = old_directory; + self.deletion_bitmap = old_deletion_bitmap; *self.index.lock().unwrap_or_else(|e| e.into_inner()) = None; *self.rabitq.lock().unwrap_or_else(|e| e.into_inner()) = None; return Err(error); @@ -618,6 +635,7 @@ impl RvfStore { self.metadata = old_metadata; self.metadata_chain = old_chain; self.segment_dir = old_directory; + self.deletion_bitmap = old_deletion_bitmap; *self.index.lock().unwrap_or_else(|e| e.into_inner()) = None; *self.rabitq.lock().unwrap_or_else(|e| e.into_inner()) = None; return Err(err(ErrorCode::FsyncFailed)); @@ -633,6 +651,7 @@ impl RvfStore { self.metadata = old_metadata; self.metadata_chain = old_chain; self.segment_dir = old_directory; + self.deletion_bitmap = old_deletion_bitmap; self.epoch = old_epoch; self.last_witness_hash = old_witness_hash; *self.index.lock().unwrap_or_else(|e| e.into_inner()) = None; @@ -646,6 +665,7 @@ impl RvfStore { self.metadata = old_metadata; self.metadata_chain = old_chain; self.segment_dir = old_directory; + self.deletion_bitmap = old_deletion_bitmap; self.epoch = old_epoch; self.last_witness_hash = old_witness_hash; *self.index.lock().unwrap_or_else(|e| e.into_inner()) = None; @@ -1828,8 +1848,12 @@ impl RvfStore { self.seg_writer = Some(seg_writer); self.last_compaction_time = now_secs(); - // Reset witness chain after compaction (the file has been rewritten). - self.last_witness_hash = [0u8; 32]; + // Compaction rewrites the file but preserves every witness segment + // byte-for-byte, so the chain the compacted file records is the one + // that was there before. Re-derive the tip from it rather than + // restarting at genesis, or the in-memory tip disagrees with what a + // reopen of the same file reconstructs (issue #747). + self.restore_witness_chain(); // Append a witness entry recording this compact operation. if self.options.witness.witness_compact { @@ -2761,6 +2785,17 @@ impl RvfStore { } } } + // A `META_SEG` declares one schema entry per field id, so live records + // that give one field two different value types have no encodable full + // snapshot. Accepting such a batch acknowledges a write that fails + // later, at whichever unrelated commit happens to materialize a + // snapshot; reject it here so the error lands on the call that caused + // it (issue #772). A batch that *resolves* an existing conflict -- + // rewriting the odd record with the agreed type, or deleting it -- is + // still accepted, so a store already in this state can be repaired. + if !matches!(metadata, IngestMetadata::None) && candidate.conflicting_field().is_some() { + return Err(err(ErrorCode::InvalidMetadata)); + } Ok(candidate) } @@ -3179,6 +3214,36 @@ impl RvfStore { Ok(()) } + /// Restore the witness chain tip from the newest `WITNESS_SEG` on file. + /// + /// The bytes [`Self::append_witness`] hashes are the segment payload + /// verbatim, so the tip is the digest of the newest witness payload and no + /// replay of the whole log is needed. Without this a reopen restarts the + /// chain at genesis and the next entry links to zeros instead of to the + /// entry that precedes it on disk, which breaks external verification of + /// the chain across a close (issue #747). + /// + /// A witness segment that cannot be read leaves the chain at genesis + /// rather than failing the open: the witness log is an audit trail beside + /// the committed vector and metadata state, not part of it. + fn restore_witness_chain(&mut self) { + let newest = self + .segment_dir + .iter() + .rev() + .find(|&&(_, _, _, segment_type)| segment_type == SegmentType::Witness as u8) + .map(|&(_, offset, _, _)| offset); + let payload = newest.and_then(|offset| { + let mut reader = BufReader::new(&self.file); + read_path::read_segment_payload(&mut reader, offset) + .ok() + .map(|(_, payload)| payload) + }); + self.last_witness_hash = payload + .map(|payload| simple_shake256_256(&payload)) + .unwrap_or([0u8; 32]); + } + fn boot(&mut self) -> Result<(), RvfError> { let own_path = fs::canonicalize(&self.path).map_err(|_| err(ErrorCode::InvalidManifest))?; let mut ancestry = HashSet::from([own_path]); @@ -3255,6 +3320,7 @@ impl RvfStore { self.restore_metadata()?; self.restore_cow_state(ancestry)?; + self.restore_witness_chain(); // Load the most recently persisted HNSW index, if any. A stale or // corrupt INDEX_SEG is ignored; the index is then rebuilt from @@ -3890,8 +3956,10 @@ impl RvfStore { /// consecutive valid deltas that follow it. Replay therefore serves that /// longest complete prefix and reports what it had to drop through /// [`Self::metadata_recovery`], rather than refusing to open because one - /// byte of one delta went bad. Only a chain with no readable snapshot at - /// all is an error, since then nothing committed can be reconstructed. + /// byte of one delta went bad. That holds for a damaged snapshot too: the + /// walk continues past it to the next readable one however far back that + /// is. Only a chain with no readable snapshot at all is an error, since + /// then nothing committed can be reconstructed. fn restore_metadata(&mut self) -> Result<(), RvfError> { let meta_offsets: Vec = self .segment_dir @@ -3903,9 +3971,13 @@ impl RvfStore { return Ok(()); } - // Walk newest-first and stop at the first full snapshot: that is the - // newest one, and everything older than it is superseded history. - // Generations that cannot be read or decoded are collected as gaps. + // Walk newest-first and stop at the first *readable* full snapshot: + // that is the newest one still usable as a base, and everything older + // than it is superseded history. Generations that cannot be read or + // decoded are collected as gaps and walked past -- a damaged snapshot + // must not end the search while an intact older one is still reachable. + // Only the decoded-byte ceiling bounds the walk, because finding out + // whether a generation is a snapshot requires decoding it. let mut decoded_bytes = 0usize; let mut newest_first: Vec<(u64, usize, MetadataSegment)> = Vec::new(); let mut visited: HashSet = HashSet::new(); @@ -3913,9 +3985,6 @@ impl RvfStore { let mut found_snapshot = false; for &offset in meta_offsets.iter().rev() { - if newest_first.len() > rvf_types::metadata::MAX_META_DELTAS { - return Err(err(ErrorCode::MetadataReplayLimitExceeded)); - } let payload = { let mut reader = BufReader::new(&self.file); read_path::read_segment_payload(&mut reader, offset) @@ -3982,6 +4051,16 @@ impl RvfStore { // Everything after the break is unreachable too, so it is dropped as // well and must be counted -- the CLI reports this number verbatim. damaged = damaged.saturating_add(remaining.count() as u64); + // The replay budget bounds the work an open actually performs, so it + // is enforced on the chain that was applied rather than on the + // generations the scan walked past. Scanning stops at the newest + // *readable* snapshot, and damage can push that arbitrarily far back + // while leaving a short applied chain; charging the scan would refuse + // to open a file whose committed state replays in a handful of steps + // (issue #770). + if applied.len().saturating_sub(1) > rvf_types::metadata::MAX_META_DELTAS { + return Err(err(ErrorCode::MetadataReplayLimitExceeded)); + } let latest = applied .last() .map(|segment| segment.generation) diff --git a/crates/rvf/tests/rvf-integration/tests/e2e_store_lifecycle.rs b/crates/rvf/tests/rvf-integration/tests/e2e_store_lifecycle.rs index a971e956a3..dde00cea43 100644 --- a/crates/rvf/tests/rvf-integration/tests/e2e_store_lifecycle.rs +++ b/crates/rvf/tests/rvf-integration/tests/e2e_store_lifecycle.rs @@ -533,3 +533,71 @@ fn lifecycle_dimension_mismatch_rejected() { store.close().unwrap(); } + +/// Durable configuration must survive a reopen. `open()` reconstructs the +/// store from the file alone, so anything it fails to restore silently changes +/// how the same bytes behave: a metric reset to L2 re-ranks every query, and a +/// witness chain restarted at genesis breaks external verification across a +/// close (issue #747). +#[test] +fn reopen_preserves_metric_and_witness_chain() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("reopen_options.rvf"); + + // Under cosine the aligned-but-distant vector 1 wins; under L2 the nearby + // but off-axis vector 2 does. The ranking is therefore evidence of which + // metric the reopened store is actually using. + let aligned = [10.0f32, 0.0, 0.0]; + let nearby = [0.9f32, 0.4, 0.0]; + let query = [1.0f32, 0.0, 0.0]; + + let options = RvfOptions { + dimension: 3, + metric: DistanceMetric::Cosine, + ..Default::default() + }; + + let mut store = RvfStore::create(&path, options).unwrap(); + store + .ingest_batch(&[&aligned, &nearby], &[1, 2], None) + .unwrap(); + let ranking: Vec = store + .query(&query, 2, &QueryOptions::default()) + .unwrap() + .iter() + .map(|hit| hit.id) + .collect(); + assert_eq!( + ranking, + vec![1, 2], + "cosine must prefer the aligned vector; the test is meaningless otherwise" + ); + let witness = *store.last_witness_hash(); + assert_ne!( + witness, [0u8; 32], + "the ingest must have extended the witness chain" + ); + store.close().unwrap(); + + let reopened = RvfStore::open(&path).unwrap(); + assert_eq!( + reopened.metric(), + DistanceMetric::Cosine, + "the configured metric must be restored from the manifest" + ); + let reopened_ranking: Vec = reopened + .query(&query, 2, &QueryOptions::default()) + .unwrap() + .iter() + .map(|hit| hit.id) + .collect(); + assert_eq!( + reopened_ranking, ranking, + "a reopen must rank identically, not fall back to L2" + ); + assert_eq!( + *reopened.last_witness_hash(), + witness, + "a reopen must continue the witness chain rather than restart it at genesis" + ); +} diff --git a/crates/rvf/tests/rvf-integration/tests/metadata_crash_safety.rs b/crates/rvf/tests/rvf-integration/tests/metadata_crash_safety.rs index 1a4832738d..b732c82ff3 100644 --- a/crates/rvf/tests/rvf-integration/tests/metadata_crash_safety.rs +++ b/crates/rvf/tests/rvf-integration/tests/metadata_crash_safety.rs @@ -424,6 +424,77 @@ fn delete_metadata_commit_is_atomic_under_crash_injection() { assert_atomic_metadata_commit("delete", &all_ids, &[], &bytes_a, &bytes_b, &probe); } +/// The runtime's segment content hash: an IEEE CRC32 of the payload rotated +/// into four little-endian lanes (`rvf-runtime::hashing`). Rewriting a payload +/// in place means repairing this, or the reader rejects the segment as damaged. +fn legacy_content_hash(data: &[u8]) -> [u8; 16] { + let mut crc = 0xFFFF_FFFFu32; + for &byte in data { + crc ^= byte as u32; + for _ in 0..8 { + let mask = (crc & 1).wrapping_neg(); + crc = (crc >> 1) ^ (0xEDB8_8320 & mask); + } + } + let crc = !crc; + let mut hash = [0u8; 16]; + for lane in 0..4 { + hash[lane * 4..(lane + 1) * 4] + .copy_from_slice(&crc.rotate_left(lane as u32 * 8).to_le_bytes()); + } + hash +} + +/// Renumber the newest `META_SEG` of `path` to generation `u64::MAX`, so the +/// next metadata commit has no generation number left to allocate. +/// +/// The payload is decoded and re-encoded rather than patched byte-wise, so it +/// stays a segment the reader accepts on its own terms; the generation is a +/// fixed-width field, so the replacement is the same length and the +/// append-only offsets the manifest records still hold. +fn exhaust_metadata_generation(path: &std::path::Path) { + let mut bytes = fs::read(path).unwrap(); + let mut meta: Option<(usize, usize)> = None; + let mut offset = 0usize; + while offset + SEGMENT_HEADER_SIZE <= bytes.len() { + if u32::from_le_bytes(bytes[offset..offset + 4].try_into().unwrap()) != SEGMENT_MAGIC { + break; + } + let payload_len = + u64::from_le_bytes(bytes[offset + 0x10..offset + 0x18].try_into().unwrap()) as usize; + let end = match offset.checked_add(SEGMENT_HEADER_SIZE + payload_len) { + Some(end) if end <= bytes.len() => end, + _ => break, + }; + if bytes[offset + 0x05] == rvf_types::SegmentType::Meta as u8 { + meta = Some((offset + SEGMENT_HEADER_SIZE, payload_len)); + } + offset = end; + } + let (payload_start, payload_len) = meta.expect("the store must have written a META_SEG"); + + let mut segment = rvf_types::metadata::MetadataSegment::decode( + &bytes[payload_start..payload_start + payload_len], + ) + .expect("the newest META_SEG must decode"); + assert!( + segment.full_snapshot, + "the newest generation must be the child's first full snapshot" + ); + segment.generation = u64::MAX; + let payload = segment.encode().unwrap(); + assert_eq!( + payload.len(), + payload_len, + "renumbering must not change the encoded length" + ); + + bytes[payload_start..payload_start + payload_len].copy_from_slice(&payload); + let header = payload_start - SEGMENT_HEADER_SIZE; + bytes[header + 0x28..header + 0x38].copy_from_slice(&legacy_content_hash(&payload)); + fs::write(path, &bytes).unwrap(); +} + /// A `delete` that fails must leave no trace in memory. It tombstones vectors /// and prunes the membership filter before writing its metadata generation, so /// a rollback that restored only the metadata would leave the store claiming @@ -431,11 +502,13 @@ fn delete_metadata_commit_is_atomic_under_crash_injection() { /// a manifest whose deleted set contradicts the committed metadata, which open /// rejects outright. /// -/// The failure is injected through the encoder: a COW child inherits its -/// parent's records in memory and commits them as one full snapshot on its -/// first metadata write, and a snapshot that declares one field id with two -/// different value types is not encodable. The parent itself never hits this, -/// because each of its generations touches only one of the records. +/// The failure is injected at the head of `write_metadata_generation`, which +/// numbers the next generation with a checked increment: a chain whose newest +/// generation is `u64::MAX` cannot be extended, so the commit fails after the +/// tombstone and the membership prune have already been applied. Rewriting the +/// generation number on disk is a self-contained way to reach that state -- +/// unlike an encoder-level failure it does not require the store to be left in +/// a shape a write should have rejected in the first place (issue #772). #[test] fn failed_delete_leaves_no_uncommitted_tombstones() { let dir = TempDir::new().unwrap(); @@ -444,27 +517,36 @@ fn failed_delete_leaves_no_uncommitted_tombstones() { let dim: u16 = 2; let mut parent = RvfStore::create(&parent_path, make_options(dim)).unwrap(); - for (id, value) in [ - (1u64, MetadataValue::U64(1)), - (2, MetadataValue::String("two".into())), - (3, MetadataValue::U64(3)), - ] { + for id in [1u64, 2, 3] { parent .ingest_batch_with_metadata( &[&[id as f32, 1.0]], &[id], &[VectorMetadata { vector_id: id, - fields: vec![MetadataEntry { field_id: 1, value }], + fields: vec![MetadataEntry { + field_id: 1, + value: MetadataValue::U64(id), + }], delete_record: false, }], ) .unwrap(); } + // The child commits the inherited records as one full snapshot on its + // first metadata write; that is the generation whose number is rewritten. let mut child = parent.branch(&child_path).unwrap(); - // Deleting vector 3 leaves records 1 (u64) and 2 (string) to be written as - // the child's first full snapshot, which cannot be encoded. + child + .set_file_metadata("app.owner".into(), MetadataValue::String("catalog".into())) + .unwrap(); + child.close().unwrap(); + parent.close().unwrap(); + exhaust_metadata_generation(&child_path); + + let mut child = RvfStore::open(&child_path).unwrap(); + // Vector 3 is inherited from the parent, so deleting it prunes the + // membership filter as well as tombstoning the record. let failure = child.delete(&[3]).expect_err("delete must fail to commit"); // A later commit that writes no metadata generation still publishes a @@ -473,7 +555,6 @@ fn failed_delete_leaves_no_uncommitted_tombstones() { .ingest_batch(&[&[4.0, 1.0]], &[4], None) .unwrap_or_else(|e| panic!("[{failure:?}] a later ingest must still commit: {e:?}")); child.close().unwrap(); - parent.close().unwrap(); let reopened = RvfStore::open_readonly(&child_path).unwrap_or_else(|e| { panic!("[{failure:?}] a rolled-back delete must not brick the artifact: {e:?}") @@ -537,3 +618,143 @@ fn file_metadata_commit_is_atomic_under_crash_injection() { &probe, ); } + +/// A mutation that fails after its metadata generation is on disk must restore +/// the segment directory it started from, not merely drop the entries it +/// appended. +/// +/// `write_metadata_generation` both prunes and appends when it re-anchors a +/// recovered chain: it drops the generations replay could not reach and adds +/// the snapshot that supersedes them. The directory therefore ends up *shorter* +/// than it started, so a rollback that only trimmed back to the original length +/// would keep every one of those edits -- publishing the uncommitted generation +/// to the next commit, which then serves it as authoritative and loses whatever +/// the failed mutation had removed (issue #771). +/// +/// The failure is injected by removing the parent file of a COW child: the +/// child's manifest write has to re-derive the relative parent reference, which +/// is the first thing after the metadata generation that can fail on demand. +#[test] +fn failed_mutation_restores_the_pruned_segment_directory() { + for case in ["delete", "set_file_metadata"] { + let dir = TempDir::new().unwrap(); + let parent_path = dir.path().join("rollback_parent.rvf"); + let child_path = dir.path().join("rollback_child.rvf"); + let dim: u16 = 2; + + let mut parent = RvfStore::create(&parent_path, make_options(dim)).unwrap(); + parent + .ingest_batch_with_metadata( + &[&[1.0, 0.0]], + &[1], + &[VectorMetadata { + vector_id: 1, + fields: vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("parent".into()), + }], + delete_record: false, + }], + ) + .unwrap(); + + // Four child generations: a full snapshot of the inherited state plus + // three deltas, so damaging the third strands two of them. + let mut child = parent.branch(&child_path).unwrap(); + for id in [11u64, 12, 13, 14] { + child + .ingest_batch_with_metadata( + &[&[id as f32, 1.0]], + &[id], + &[VectorMetadata { + vector_id: id, + fields: vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String(format!("child-{id}")), + }], + delete_record: false, + }], + ) + .unwrap(); + } + child.close().unwrap(); + parent.close().unwrap(); + let parent_bytes = fs::read(&parent_path).unwrap(); + + // Damage the third generation so the child opens on a truncated chain + // with orphans for the next metadata write to prune. + let mut bytes = fs::read(&child_path).unwrap(); + let mut meta_offsets = Vec::new(); + let mut offset = 0usize; + while offset + SEGMENT_HEADER_SIZE <= bytes.len() { + if u32::from_le_bytes(bytes[offset..offset + 4].try_into().unwrap()) != SEGMENT_MAGIC { + break; + } + let payload_len = + u64::from_le_bytes(bytes[offset + 0x10..offset + 0x18].try_into().unwrap()) + as usize; + let end = match offset.checked_add(SEGMENT_HEADER_SIZE + payload_len) { + Some(end) if end <= bytes.len() => end, + _ => break, + }; + if bytes[offset + 0x05] == rvf_types::SegmentType::Meta as u8 { + meta_offsets.push((offset + SEGMENT_HEADER_SIZE, payload_len)); + } + offset = end; + } + assert_eq!(meta_offsets.len(), 4, "[{case}]"); + let (start, len) = meta_offsets[2]; + bytes[start + len / 2] ^= 0xFF; + fs::write(&child_path, &bytes).unwrap(); + + let mut child = RvfStore::open(&child_path).unwrap(); + assert_eq!( + child.metadata_recovery().dropped_generations, + 2, + "[{case}] generations 3 and 4 must be the orphans the next write prunes" + ); + + // Removing the parent makes the child's manifest write fail, while the + // metadata generation before it still commits to the file. + fs::remove_file(&parent_path).unwrap(); + let failure = match case { + "delete" => child.delete(&[11]).err(), + _ => child + .set_file_metadata("app.owner".into(), MetadataValue::String("catalog".into())) + .err(), + }; + let failure = failure.unwrap_or_else(|| panic!("[{case}] the mutation must fail")); + fs::write(&parent_path, &parent_bytes).unwrap(); + + // A later commit that writes no metadata generation publishes whatever + // segment directory the rollback left behind. + child + .ingest_batch(&[&[99.0, 1.0]], &[99], None) + .unwrap_or_else(|e| panic!("[{case}] a later ingest must still commit: {e:?}")); + child.close().unwrap(); + + let reopened = RvfStore::open_readonly(&child_path).unwrap_or_else(|e| { + panic!("[{case}] a rolled-back mutation must not brick the artifact: {e:?}") + }); + assert_eq!( + reopened.get_metadata(11).unwrap(), + vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("child-11".into()), + }], + "[{case}] [{failure:?}] the record the failed delete removed must survive" + ); + assert_eq!( + reopened.get_metadata(12).unwrap(), + vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("child-12".into()), + }], + "[{case}] [{failure:?}] the recovered chain must still be the served state" + ); + assert!( + reopened.get_file_metadata("app.owner").is_none(), + "[{case}] [{failure:?}] the failed set_file_metadata must not become visible" + ); + } +} diff --git a/crates/rvf/tests/rvf-integration/tests/metadata_durability.rs b/crates/rvf/tests/rvf-integration/tests/metadata_durability.rs index abba138165..e8d91fbb5a 100644 --- a/crates/rvf/tests/rvf-integration/tests/metadata_durability.rs +++ b/crates/rvf/tests/rvf-integration/tests/metadata_durability.rs @@ -492,3 +492,145 @@ fn cow_child_search_hits_reflect_overrides_and_tombstones() { let reopened_child = RvfStore::open_readonly(&child_path).unwrap(); assert_child_hits(&reopened_child); } + +/// A soft delete tombstones the vector an identifier currently holds, not the +/// identifier itself. Writing the identifier again therefore makes it live +/// with the new vector and the new record: the query must return it, its +/// metadata must be what the re-ingest supplied rather than what the delete +/// removed, a reopen must agree, and `compact()` must not reclaim it as dead +/// (issue #748). +#[test] +fn re_ingesting_a_deleted_id_makes_it_live_again() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("undelete_by_reingest.rvf"); + let before = [1.0f32, 0.0]; + let after = [0.0f32, 1.0]; + + let mut store = RvfStore::create(&path, make_options(2)).unwrap(); + store + .ingest_batch_with_metadata( + &[&before], + &[42], + &[VectorMetadata { + vector_id: 42, + fields: vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("before-delete".into()), + }], + delete_record: false, + }], + ) + .unwrap(); + store.delete(&[42]).unwrap(); + assert!( + store.get_metadata(42).is_none(), + "the delete must remove the record while the tombstone stands" + ); + + store + .ingest_batch_with_metadata( + &[&after], + &[42], + &[VectorMetadata { + vector_id: 42, + fields: vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("after-reingest".into()), + }], + delete_record: false, + }], + ) + .unwrap(); + + let expected = vec![MetadataEntry { + field_id: 1, + value: MetadataValue::String("after-reingest".into()), + }]; + let hits = store.query(&after, 1, &QueryOptions::default()).unwrap(); + assert_eq!( + hits.iter().map(|hit| hit.id).collect::>(), + vec![42], + "the re-ingested vector must be queryable" + ); + assert_eq!(store.get_metadata(42).unwrap(), expected); + + // Compaction reclaims what is still tombstoned at compact time, so the + // cleared identifier must survive it with its new payload intact. + store.compact().unwrap(); + let hits = store.query(&after, 1, &QueryOptions::default()).unwrap(); + assert_eq!(hits.iter().map(|hit| hit.id).collect::>(), vec![42]); + store.close().unwrap(); + + let reopened = RvfStore::open_readonly(&path).unwrap(); + assert_eq!( + reopened.get_metadata(42).unwrap(), + expected, + "the reopened store must serve the re-ingested record, not the deleted one" + ); + let hits = reopened.query(&after, 1, &QueryOptions::default()).unwrap(); + assert_eq!(hits.iter().map(|hit| hit.id).collect::>(), vec![42]); +} + +/// A `META_SEG` declares one schema entry per field id, so live records that +/// give one field two different value types have no encodable full snapshot. +/// Accepting such a batch acknowledges a write that fails later, at whichever +/// unrelated commit happens to materialize a snapshot, so it is rejected on +/// the call that introduces it (issue #772). +#[test] +fn conflicting_field_types_are_rejected_at_ingest() { + let dir = TempDir::new().unwrap(); + let path = dir.path().join("field_type_conflict.rvf"); + + fn typed(vector_id: u64, value: MetadataValue) -> VectorMetadata { + VectorMetadata { + vector_id, + fields: vec![MetadataEntry { field_id: 1, value }], + delete_record: false, + } + } + + let mut store = RvfStore::create(&path, make_options(2)).unwrap(); + store + .ingest_batch_with_metadata(&[&[1.0, 0.0]], &[1], &[typed(1, MetadataValue::U64(1))]) + .unwrap(); + + // A second record typing field 1 as a string cannot share a snapshot with + // the first, so the ingest that would introduce it fails. + let error = store + .ingest_batch_with_metadata( + &[&[0.0, 1.0]], + &[2], + &[typed(2, MetadataValue::String("two".into()))], + ) + .expect_err("a conflicting value type must be rejected at ingest"); + assert!( + format!("{error:?}").contains("InvalidMetadata"), + "expected InvalidMetadata, got {error:?}" + ); + + // The rejected batch left nothing behind, and the agreed type still works. + assert!(store.get_metadata(2).is_none()); + store + .ingest_batch_with_metadata(&[&[0.0, 1.0]], &[2], &[typed(2, MetadataValue::U64(2))]) + .unwrap(); + store.close().unwrap(); + + // The store stayed snapshot-encodable throughout, which is the property + // the rejection exists to preserve. + let mut reopened = RvfStore::open(&path).unwrap(); + reopened.compact().unwrap(); + assert_eq!( + reopened.get_metadata(1).unwrap(), + vec![MetadataEntry { + field_id: 1, + value: MetadataValue::U64(1), + }] + ); + assert_eq!( + reopened.get_metadata(2).unwrap(), + vec![MetadataEntry { + field_id: 1, + value: MetadataValue::U64(2), + }] + ); +} diff --git a/crates/rvf/tests/rvf-integration/tests/metadata_growth.rs b/crates/rvf/tests/rvf-integration/tests/metadata_growth.rs index 44b86b21fd..9ae0db7721 100644 --- a/crates/rvf/tests/rvf-integration/tests/metadata_growth.rs +++ b/crates/rvf/tests/rvf-integration/tests/metadata_growth.rs @@ -793,3 +793,178 @@ fn manifest_only_write_after_truncated_recovery_preserves_the_artifact() { } } } + +/// A damaged snapshot must not end the newest-first walk. +/// +/// The replay budget bounds the chain that is actually *applied*, not the +/// generations the scan had to walk past to find a readable base. Damage can +/// push the newest usable snapshot arbitrarily far back while still leaving a +/// short applied chain, so charging the scan refuses to open a file whose +/// committed state replays in a handful of steps -- and because the failure is +/// in `open` itself, `compact()` cannot repair it either (issue #770). +#[test] +fn walk_past_two_corrupt_snapshots_reaches_the_next_readable_one() { + use rvf_types::metadata::MetadataSegment; + + // Full snapshots land at generations 1/34/67/100, i.e. segment indices + // 0/33/66/99. Corrupting the newest two leaves the walk to reach index 33 + // after scanning 97 generations -- past the 64-delta replay budget. + const COMMITS: u64 = 130; + const CORRUPTED: [usize; 2] = [99, 66]; + const SURVIVING_SNAPSHOT: usize = 33; + // Generation `g` commits record `g - 1`, so the snapshot at generation 34 + // carries records 0..=33 and the deltas after it reach record 65 before + // the damaged generation 67 truncates the chain. + const LAST_SERVED_RECORD: u64 = 65; + + let dir = TempDir::new().unwrap(); + let path = dir.path().join("two_corrupt_snapshots.rvf"); + commit_metadata_history(&path, COMMITS); + + let mut bytes = std::fs::read(&path).unwrap(); + let segments = meta_segments(&bytes); + assert_eq!(segments.len() as u64, COMMITS); + for index in CORRUPTED.into_iter().chain([SURVIVING_SNAPSHOT]) { + let (start, len) = segments[index]; + assert!( + MetadataSegment::decode(&bytes[start..start + len]) + .unwrap() + .full_snapshot, + "segment {index} must be a full snapshot for this test to mean anything" + ); + } + for index in CORRUPTED { + let (start, len) = segments[index]; + bytes[start + len / 2] ^= 0xFF; + } + std::fs::write(&path, &bytes).unwrap(); + + for label in ["open_readonly", "open"] { + let store = if label == "open" { + RvfStore::open(&path) + } else { + RvfStore::open_readonly(&path) + } + .unwrap_or_else(|e| { + panic!("[{label}] two damaged snapshots must not make the artifact unopenable: {e:?}") + }); + + assert_eq!( + store.metadata_recovery().generation, + LAST_SERVED_RECORD + 1, + "[{label}] the served state must stop at the generation before the second damage" + ); + for i in 0..=LAST_SERVED_RECORD { + assert_eq!( + store.get_metadata(i).unwrap(), + expected_record(i), + "[{label}] record {i} is reachable from the surviving snapshot" + ); + } + for i in LAST_SERVED_RECORD + 1..COMMITS { + assert!( + store.get_metadata(i).is_none(), + "[{label}] record {i} came from a dropped generation" + ); + } + } + + // Recovery must converge: the next write re-anchors the chain, and + // `compact()` is reachable again because open no longer fails. + let mut store = RvfStore::open(&path).unwrap(); + store + .ingest_batch_with_metadata(&[&[99.0, 1.0]], &[900], &[record(900, "after-recovery")]) + .unwrap(); + store.compact().unwrap(); + store.close().unwrap(); + + let reopened = RvfStore::open_readonly(&path).unwrap(); + assert_eq!(reopened.metadata_recovery().dropped_generations, 0); + assert_eq!( + reopened.get_metadata(900).unwrap(), + vec![record_entry("after-recovery")] + ); + assert_eq!(reopened.get_metadata(0).unwrap(), expected_record(0)); +} + +/// `dropped_generations` is the number the CLI prints verbatim, so it must be +/// the real count and not merely non-zero: every generation the applied chain +/// no longer reaches is counted once (issue #771). +/// +/// An eight-generation chain damaged at generation 5 loses exactly four: the +/// damaged generation itself, and generations 6, 7 and 8, which are stranded +/// behind the gap even though each of them decodes. +#[test] +fn dropped_generation_count_is_exact() { + const COMMITS: u64 = 8; + const CORRUPTED: usize = 4; + + let dir = TempDir::new().unwrap(); + let path = dir.path().join("exact_dropped_count.rvf"); + commit_metadata_history(&path, COMMITS); + + let mut bytes = std::fs::read(&path).unwrap(); + let segments = meta_segments(&bytes); + assert_eq!(segments.len() as u64, COMMITS); + let (start, len) = segments[CORRUPTED]; + bytes[start + len / 2] ^= 0xFF; + std::fs::write(&path, &bytes).unwrap(); + + let reopened = RvfStore::open_readonly(&path).unwrap(); + let recovery = reopened.metadata_recovery(); + assert_eq!( + recovery.dropped_generations, 4, + "generations 5, 6, 7 and 8 are all unreachable" + ); + assert_eq!(recovery.generation, CORRUPTED as u64); + assert_eq!(recovery.dropped_records, 0); +} + +/// The re-anchoring snapshot must also *prune* the generations the recovered +/// chain no longer reaches. Convergence alone does not prove it -- the orphans +/// sit below the new snapshot, so a walk that stops at the newest snapshot +/// never reads them either way -- but leaving them in the segment directory +/// means the artifact keeps reporting and carrying segments nothing can use +/// (issue #771). +#[test] +fn re_anchoring_write_prunes_the_orphaned_generations() { + const COMMITS: u64 = 6; + const CORRUPTED: usize = 3; + // Generations 4, 5 and 6 are stranded behind the damage. + const ORPHANS: u32 = 3; + + let dir = TempDir::new().unwrap(); + let path = dir.path().join("re_anchor_prune.rvf"); + commit_metadata_history(&path, COMMITS); + + let mut bytes = std::fs::read(&path).unwrap(); + let (start, len) = meta_segments(&bytes)[CORRUPTED]; + bytes[start + len / 2] ^= 0xFF; + std::fs::write(&path, &bytes).unwrap(); + + let mut store = RvfStore::open(&path).unwrap(); + assert_eq!( + store.metadata_recovery().dropped_generations, + ORPHANS as u64 + ); + let before = store.status().total_segments; + + // One ingest appends four segments -- VEC, META, WITNESS, MANIFEST -- and + // its META is the re-anchoring full snapshot that supersedes the orphans. + store + .ingest_batch_with_metadata(&[&[42.0, 1.0]], &[42], &[record(42, "re-anchor")]) + .unwrap(); + assert_eq!( + store.status().total_segments, + before + 4 - ORPHANS, + "the re-anchoring commit must drop the orphaned META entries" + ); + store.close().unwrap(); + + // The pruning is durable: the reopened directory does not carry them back. + // A manifest's directory does not list the manifest segment carrying it, + // so the reopened count is one below the in-memory count at commit time. + let reopened = RvfStore::open_readonly(&path).unwrap(); + assert_eq!(reopened.status().total_segments, before + 4 - ORPHANS - 1); + assert_eq!(reopened.metadata_recovery().dropped_generations, 0); +} diff --git a/patches/hnsw_rs/src/hnsw.rs b/patches/hnsw_rs/src/hnsw.rs index 444f33744a..53249fb5e9 100644 --- a/patches/hnsw_rs/src/hnsw.rs +++ b/patches/hnsw_rs/src/hnsw.rs @@ -930,11 +930,17 @@ impl<'b, T: Clone + Send + Sync, D: Distance + Send + Sync> Hnsw<'b, T, D> { // we will store positive distances in this one let mut return_points = BinaryHeap::>>::with_capacity(skiplist_size); // - if self.layer_indexed_points.points_by_layer.read()[layer as usize].is_empty() { - // at the beginning we can have nothing in layer - trace!("search layer {:?}, empty layer", layer); - return return_points; - } + // NOTE: there used to be an early return here when + // `points_by_layer[layer]` was empty. That list only records points + // whose *maximum* level is `layer`, but a point of level L takes part + // in every layer 0..=L, so the list being empty does not mean the layer + // is. Bailing out made insert link the new point to nothing at that + // layer -- e.g. two points both entering at level 1 got no layer-0 + // edges at all, leaving them unreachable to any later layer-0 search + // regardless of efSearch (issue #773). The caller always supplies a + // live entry point, and `Point::neighbours` is allocated for every + // layer, so searching an unpopulated layer is safe: it simply returns + // the entry point, which is exactly the link we want. if entry_point.p_id.1 < 0 { trace!("search layer negative point id : {:?}", entry_point.p_id); return return_points; @@ -1240,9 +1246,15 @@ impl<'b, T: Clone + Send + Sync, D: Distance + Send + Sync> Hnsw<'b, T, D> { let q_point = &q.point_ref; let mut q_point_neighbours = q_point.neighbours.write(); let n_to_add = PointWithOrder::::new(&Arc::clone(&new_point), q.dist_to_ref); - // must be sure that we add a point at the correct level. See the comment to search_layer! - // this ensures that reverse updating do not add problems. - let l_n = n_to_add.point_ref.p_id.0 as usize; + // The reciprocal edge must land on the same layer as the forward + // edge (Malkov Alg. 1: "add bidirectional connections ... at layer + // lc"). Using new_point's own level here instead meant a point + // entering at level >= 1 got its layer-0 back-edges filed under + // that upper level, leaving it with no in-edge on layer 0. Once it + // stopped being the entry point nothing on layer 0 pointed at it, + // so search dropped it permanently and no efSearch could recover + // it. See issue #773. + let l_n = l as usize; let already = q_point_neighbours[l_n] .iter() .position(|old| old.point_ref.p_id == new_point.p_id);