From 38a1b8a4ee098d76087e073667e7b247abb86a55 Mon Sep 17 00:00:00 2001 From: Arnaud Gourlay Date: Fri, 12 Jun 2026 15:08:18 +0200 Subject: [PATCH] Enable multivector in model test, repair stale multi-dense vectors length on reload The multi-dense appendable storage splits each point across two independently flushed chunked stores (offsets and vectors). Flushers snapshot status.len at creation time but msync chunk data at execution time, so a re-upsert that grows a point's row count in that window can land its relocated offsets entry and rows durably while the vectors store's recorded length stays stale. On reload, reads of such entries are rejected and a WAL-replayed append reuses the referenced rows, overwriting another point's data. Fix: reload-time repair when opening the storage. Scan live offset entries and widen the recorded vectors length to cover the maximum referenced row end (rows are always msynced before the entries that reference them, so only the length can lag). Also enable the "m" multivector in the Collection model test, which is what surfaces this bug (seed 1 panics at the restart at op 1089 without the fix; seeds 1-7 and both harness smoke tests pass with it). Co-Authored-By: Claude Fable 5 --- lib/collection/src/model_testing/mod.rs | 3 +- lib/collection/src/model_testing/op/mod.rs | 3 - .../vector_storage/chunked_vectors/write.rs | 16 +++ ...endable_mmap_multi_dense_vector_storage.rs | 134 +++++++++++++++++- 4 files changed, 150 insertions(+), 6 deletions(-) diff --git a/lib/collection/src/model_testing/mod.rs b/lib/collection/src/model_testing/mod.rs index 1008461ac3..68c851165e 100644 --- a/lib/collection/src/model_testing/mod.rs +++ b/lib/collection/src/model_testing/mod.rs @@ -58,8 +58,7 @@ pub(super) const ALL_CANDIDATES: &[VectorCandidate] = &[ ]; /// Names present in the collection schema at fixture time. -// "m" (multivector) is temporarily disabled while its reload divergence is unresolved. -pub(super) const INITIAL_ACTIVE: &[&str] = &["a", "b", "i", "s"]; +pub(super) const INITIAL_ACTIVE: &[&str] = &["a", "b", "i", "s", "m"]; /// Dense vector name configured with HNSW `inline_storage` + scalar quantization in the fixture. pub(super) const INLINE_STORAGE_VECTOR: &str = "i"; diff --git a/lib/collection/src/model_testing/op/mod.rs b/lib/collection/src/model_testing/op/mod.rs index 9b6746ef21..6508e48369 100644 --- a/lib/collection/src/model_testing/op/mod.rs +++ b/lib/collection/src/model_testing/op/mod.rs @@ -428,12 +428,9 @@ impl Op { 22 => { // CreateVectorName: pick a candidate currently absent from the active set. // If all candidates are active, fall back to a regular upsert. - // Multivector (`MultiDense`) candidates are excluded while their reload - // divergence is unresolved — see the note on `INITIAL_ACTIVE`. let inactive: Vec<&'static super::VectorCandidate> = ALL_CANDIDATES .iter() .filter(|c| !active.contains(c.name)) - .filter(|c| !matches!(c.kind, VectorKind::MultiDense(_))) .collect(); if inactive.is_empty() { return upsert_fallback(rng, active, id_pool); diff --git a/lib/segment/src/vector_storage/chunked_vectors/write.rs b/lib/segment/src/vector_storage/chunked_vectors/write.rs index ae85e2c342..e5acb118a0 100644 --- a/lib/segment/src/vector_storage/chunked_vectors/write.rs +++ b/lib/segment/src/vector_storage/chunked_vectors/write.rs @@ -230,6 +230,22 @@ where Ok(new_id) } + /// Widen the recorded length to at least `len` without writing any vector data. + /// + /// Reload-time repair hook: flushers snapshot `status.len` at creation time but msync + /// chunk data at execution time, so a crash can leave durable rows (and durable + /// references to them in a sibling store) beyond the recorded length. The data is + /// already on disk; only the length lagged. Clamped to the physically existing chunk + /// space so a corrupt caller-supplied value can't make reads address missing chunks. + pub fn ensure_len_at_least(&mut self, len: usize) { + let physical_capacity = self.inner.chunks.len() * self.inner.config.chunk_size_vectors; + let len = len.min(physical_capacity); + if len > self.status.len { + self.status.len = len; + self.inner.len = len; + } + } + pub fn flusher(&self) -> Flusher { Box::new({ let status_flusher = self.status.flusher(); diff --git a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs index 22e586c8d7..2b37d2f7ee 100644 --- a/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs +++ b/lib/segment/src/vector_storage/multi_dense/appendable_mmap_multi_dense_vector_storage.rs @@ -573,8 +573,9 @@ pub fn open_appendable_memmap_multi_vector_storage_impl( + vectors: &mut ChunkedVectors, + offsets: &ChunkedVectors, +) { + let mut max_row_end = 0; + for key in 0..offsets.len() { + let Some(entry) = offsets.get::(key as VectorOffsetType) else { + continue; + }; + let &[ + MultivectorMmapOffset { + offset, + count: _, + // The reserved region is `offset..offset + capacity`; `count` only shrinks + // within it on in-place re-upserts, and the original append recorded + // `offset + capacity` as the length contribution. + capacity, + }, + ] = entry.as_ref() + else { + unreachable!("multi-vector offsets are stored as vectors of length 1"); + }; + max_row_end = max_row_end.max(offset as usize + capacity as usize); + } + if max_row_end > vectors.len() { + log::warn!( + "Multi-vector offsets reference rows beyond the recorded vectors length \ + ({} > {}); widening it (stale length snapshot from an interrupted flush)", + max_row_end, + vectors.len(), + ); + vectors.ensure_len_at_least(max_row_end); + } +} + /// Find files related to this dense vector storage #[cfg(test)] pub(crate) fn find_storage_files(vector_storage_path: &Path) -> OperationResult> { @@ -675,4 +722,89 @@ mod tests { "find_storage_files must find same files that storage reports", ); } + + /// Reproduces the on-disk state left by an interrupted flush: rows and the offsets + /// entries referencing them are durable, but the vectors store's recorded length + /// (`status.dat`) is stale. Without `repair_vectors_len`, reopening makes those + /// points unreadable and lets appends overwrite their rows. + #[test] + fn repair_stale_vectors_len_on_reload() { + const POINT_COUNT: PointOffsetType = 100; + const DIM: usize = 4; + + let dir = Builder::new().prefix("storage_dir").tempdir().unwrap(); + let mut rng = StdRng::seed_from_u64(RAND_SEED); + let hw_counter = HardwareCounterCell::disposable(); + + let mut expected = Vec::new(); + { + let mut storage = + open_appendable_memmap_multi_vector_storage_impl::( + dir.path(), + DIM, + Distance::Dot, + MultiVectorConfig::default(), + AdviceSetting::Global, + false, + ) + .unwrap(); + for internal_id in 0..POINT_COUNT { + let size = rng.random_range(2..=4); + let vectors = std::iter::repeat_with(|| { + std::iter::repeat_with(|| rng.random_range(-1.0..1.0)) + .take(DIM) + .collect() + }) + .take(size) + .collect::>>(); + let multivec = MultiDenseVectorInternal::try_from(vectors).unwrap(); + storage + .insert_vector(internal_id, VectorRef::from(&multivec), &hw_counter) + .unwrap(); + expected.push(multivec); + } + storage.flusher()().unwrap(); + } + + // Rewind the recorded length to 1 row, leaving the row data and the offsets + // entries untouched: the exact skew an interrupted flush persists. + let status_file = ChunkedVectorsRead::::status_file( + &dir.path().join(VECTORS_DIR_PATH), + ); + let mut status_bytes = fs::read(&status_file).unwrap(); + let true_len = usize::from_ne_bytes( + status_bytes[..std::mem::size_of::()] + .try_into() + .unwrap(), + ); + assert!(true_len > 1, "fixture must have appended multiple rows"); + status_bytes[..std::mem::size_of::()].copy_from_slice(&1usize.to_ne_bytes()); + fs::write(&status_file, &status_bytes).unwrap(); + + let storage = open_appendable_memmap_multi_vector_storage_impl::( + dir.path(), + DIM, + Distance::Dot, + MultiVectorConfig::default(), + AdviceSetting::Global, + false, + ) + .unwrap(); + + assert_eq!( + storage.vectors.len(), + true_len, + "repair must restore the full recorded length", + ); + for (internal_id, multivec) in expected.iter().enumerate() { + let read = storage + .get_multi_opt::(internal_id as PointOffsetType) + .unwrap_or_else(|| panic!("point {internal_id} unreadable after reload")); + assert_eq!( + read.as_vec_ref().flattened_vectors, + multivec.flattened_vectors.as_slice(), + "point {internal_id} content diverged after reload", + ); + } + } }