[LiveReload] Preload chunked vectors (#10225)

* add InMemoryBitvecFlags::live_preload

* impl live_preload for ReadOnlyChunkedVectors

* heal-resilient reload

* no panic, service error

lib/segment/src/common/flags/in_memory_bitvec_flags.rs

@@ -91,6 +91,26 @@ impl InMemoryBitvecFlags {
         Ok(())
     }
 
+    /// Stage the two files [`Self::reload_appended`] may open. A no-op for
+    /// [`Self::from_bitvec`] flags, mirroring the reload.
+    pub fn live_preload(&self, fs: &impl CachedReadFs) -> OperationResult<()> {
+        let Some(directory) = &self.directory else {
+            return Ok(());
+        };
+
+        fs.schedule_prefetch(
+            &status_file(directory),
+            Some(bitslice_open_options(Populate::No)),
+            None,
+        )?;
+        fs.schedule_prefetch(
+            &directory.join(FLAGS_FILE),
+            Some(bitslice_open_options(Populate::No)),
+            None,
+        )?;
+        Ok(())
+    }
+
     /// Logical flag count from the status file; the flags file itself is padded
     /// past this.
     fn persisted_len<S: UniversalRead>(

lib/segment/src/vector_storage/chunked_vectors/config.rs

@@ -1,7 +1,8 @@
 use std::path::{Path, PathBuf};
 
 use common::universal_io::{
-    UniversalIoError, UniversalReadFs, UniversalWriteFileOps, read_json_via, read_whole_via,
+    UioResult, UniversalIoError, UniversalReadFs, UniversalWriteFileOps, read_json_via,
+    read_whole_via,
 };
 use serde::{Deserialize, Serialize};
 
@@ -133,7 +134,7 @@ pub(super) fn load_config<Fs: UniversalReadFs>(
 pub(super) fn read_status_len<Fs: UniversalReadFs>(
     fs: &Fs,
     status_file: &Path,
-) -> OperationResult<usize> {
+) -> UioResult<usize> {
     let needed = std::mem::size_of::<usize>();
     let len = read_whole_via(fs, status_file, |bytes| {
         let head = bytes.get(..needed).ok_or_else(|| {

lib/segment/src/vector_storage/chunked_vectors/read_only/live_reload.rs

@@ -1,9 +1,11 @@
 use common::counter::hardware_counter::HardwareCounterCell;
 use common::sorted_slice::SortedSlice;
 use common::types::PointOffsetType;
-use common::universal_io::{UniversalRead, UniversalReadFs};
+use common::universal_io::{
+    CachedReadFs, OkUnchanged, TypedStorage, UniversalRead, UniversalReadFs,
+};
 
-use super::super::chunks::read_chunks_from;
+use super::super::chunks::{chunk_name, chunk_open_options, list_chunk_files, read_chunks_from};
 use super::super::config::{read_status_len, status_file};
 use super::ReadOnlyChunkedVectors;
 use crate::common::live_reload::LiveReload;
@@ -12,18 +14,41 @@ use crate::common::operation_error::OperationResult;
 impl<T: bytemuck::Pod + Send, S: UniversalRead> LiveReload for ReadOnlyChunkedVectors<T, S> {
     type File = S;
 
+    fn live_preload<Fs: CachedReadFs<File = S>>(&self, fs: &Fs) -> OperationResult<()> {
+        // Status is the change signal, let reload skip reloading if this didn't change.
+        fs.reschedule_prefetch(&status_file(&self.directory), None, None)?;
+
+        let num_files = list_chunk_files(fs, &self.directory)?.len();
+
+        // `len` marks max committed vector. First chunk that can have changed:
+        // the one the next append lands in.
+        let last_chunk = self.config.get_chunk_index(self.len);
+
+        let fresh_from = if last_chunk < self.chunks.len().min(num_files) {
+            fs.reschedule_prefetch(
+                &chunk_name(&self.directory, last_chunk),
+                Some(chunk_open_options(self.advice, self.populate, false)),
+                None,
+            )?;
+            last_chunk + 1
+        } else {
+            last_chunk
+        };
+
+        // Prefetch the rest of the chunks the reload may open.
+        for chunk_id in fresh_from..num_files {
+            fs.schedule_prefetch(
+                &chunk_name(&self.directory, chunk_id),
+                Some(chunk_open_options(self.advice, self.populate, false)),
+                None,
+            )?;
+        }
+        Ok(())
+    }
+
     /// Refresh the chunks that can have gained vectors since the last load; a
-    /// no-op when the length is unchanged (the status file is read fresh, so
-    /// it is a reliable change signal).
-    ///
-    /// Chunk files are preallocated to full size, so appended vectors are
-    /// in-place writes *within* the existing file length — which a held
-    /// handle's cached blocks never reflect on caching backends (a block
-    /// fetched earlier extends past the old tail into then-unwritten space).
-    /// So, mirroring `Pages::live_reload`, the last held chunk — the only one
-    /// that can have gained vectors — is dropped and re-opened fresh,
-    /// alongside adopting newly created chunk files. Deletions/new points are
-    /// tracked by callers, so they are unused here.
+    /// no-op when the length is unchanged (the status file is saved last, so
+    /// it's a reliable change signal).
     fn live_reload<Fs: UniversalReadFs<File = S>>(
         &mut self,
         fs: &Fs,
@@ -31,21 +56,53 @@ impl<T: bytemuck::Pod + Send, S: UniversalRead> LiveReload for ReadOnlyChunkedVe
         _new_points: &SortedSlice<'_, PointOffsetType>,
         _hw_counter: &HardwareCounterCell,
     ) -> OperationResult<()> {
-        let new_len = read_status_len(fs, &status_file(&self.directory))?;
+        let Some(new_len) = read_status_len(fs, &status_file(&self.directory)).ok_unchanged()?
+        else {
+            return Ok(());
+        };
+
+        // Same len is also no-op
         if new_len == self.len {
             return Ok(());
         }
+        if new_len < self.len {
+            return Err(
+                crate::common::operation_error::OperationError::service_error(
+                    "live_reload only supports appends",
+                ),
+            );
+        }
+
+        // First chunk that can have changed: the one committed by `len`
+        let last_chunk = self.config.get_chunk_index(self.len);
+
+        let fresh_from = if last_chunk < self.chunks.len() {
+            if let Some(fresh_chunk) = TypedStorage::open(
+                fs,
+                &chunk_name(&self.directory, last_chunk),
+                chunk_open_options(self.advice, self.populate, false),
+                Default::default(),
+            )
+            .ok_unchanged()?
+            {
+                // Fresh handle for the watermark chunk
+                self.chunks[last_chunk] = fresh_chunk;
+            }
+            last_chunk + 1
+        } else {
+            last_chunk
+        };
 
-        let reload_from = self.chunks.len().saturating_sub(1);
         let new_chunks = read_chunks_from(
             fs,
             &self.directory,
-            reload_from,
+            fresh_from,
             self.advice,
             self.populate,
             false,
         )?;
-        self.chunks.truncate(reload_from);
+
+        self.chunks.truncate(fresh_from);
         self.chunks.extend(new_chunks);
         self.len = new_len;
         Ok(())

lib/segment/src/vector_storage/chunked_vectors/read_only/mod.rs

@@ -90,6 +90,157 @@ mod tests {
         assert_eq!(got.as_ref(), make_vec(100, DIM).as_slice());
     }
 
+    /// Preload must stage every file the reload opens: after the preload the
+    /// backing files are deleted, so the reload can only succeed from the
+    /// prefetch pool (parked mmap handles keep reading deleted files on unix).
+    #[cfg(unix)]
+    #[test]
+    fn live_preload_then_reload_sees_appended_vectors() {
+        use common::universal_io::{CachedFs, CachedReadFs};
+
+        const DIM: usize = 32;
+        let dir = Builder::new().prefix("chunked_preload").tempdir().unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..100, DIM, &hw);
+
+        let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+            &MmapFs,
+            dir.path(),
+            DIM,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap();
+        assert_eq!(reader.len(), 100);
+
+        append_range(&mut writer, 100, 100..250, DIM, &hw);
+        drop(writer);
+
+        let mut cached_fs = CachedFs::new(MmapFs, dir.path()).unwrap();
+        cached_fs.cache_file_info().unwrap();
+        LiveReload::live_preload(&reader, &cached_fs).unwrap();
+
+        for file in fs_err::read_dir(dir.path()).unwrap() {
+            fs_err::remove_file(file.unwrap().path()).unwrap();
+        }
+
+        let empty = SortedSlice::new(&[]).unwrap();
+        reader.live_reload(&cached_fs, &empty, &empty, &hw).unwrap();
+
+        assert_eq!(reader.len(), 250);
+        let got = reader.get::<Random>(100).unwrap();
+        assert_eq!(got.as_ref(), make_vec(100, DIM).as_slice());
+    }
+
+    /// Growth starting exactly at a chunk boundary leaves the last held chunk
+    /// untouched while the length changed: it is fully committed, so preload
+    /// and reload skip it, keeping the current handle and adopting only the
+    /// new chunk.
+    #[test]
+    fn live_preload_unchanged_last_chunk_keeps_handle() {
+        use common::universal_io::{CachedFs, CachedReadFs};
+
+        const DIM: usize = 32; // 4096 vectors per test chunk
+        let dir = Builder::new().prefix("chunked_boundary").tempdir().unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..4096, DIM, &hw);
+
+        let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+            &MmapFs,
+            dir.path(),
+            DIM,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap();
+        assert_eq!(reader.len(), 4096);
+
+        let empty = SortedSlice::new(&[]).unwrap();
+        let mut cached_fs = CachedFs::new(MmapFs, dir.path()).unwrap();
+
+        // First cycle: no previous snapshot, staging parks fresh handles.
+        cached_fs.cache_file_info().unwrap();
+        LiveReload::live_preload(&reader, &cached_fs).unwrap();
+        reader.live_reload(&cached_fs, &empty, &empty, &hw).unwrap();
+
+        // Growth lands entirely in a new chunk; chunk 0 stays untouched.
+        append_range(&mut writer, 4096, 4096..4196, DIM, &hw);
+
+        // Second cycle: chunk 0 is rescheduled and unchanged -> sentinel.
+        cached_fs.cache_file_info().unwrap();
+        LiveReload::live_preload(&reader, &cached_fs).unwrap();
+        reader.live_reload(&cached_fs, &empty, &empty, &hw).unwrap();
+
+        assert_eq!(reader.len(), 4196);
+        for offset in [0, 4095, 4096, 4195] {
+            assert_eq!(
+                reader.get::<Random>(offset).unwrap().as_ref(),
+                make_vec(offset, DIM).as_slice(),
+                "vector {offset} mismatch after reload",
+            );
+        }
+    }
+
+    /// Writer recovery (`ensure_chunk_lengths`) can remove uncommitted
+    /// trailing chunk files: a reader that opened while such a file existed
+    /// must drop its handle on reload instead of re-opening the deleted file,
+    /// and refresh the chunk holding the watermark rather than the last held.
+    #[cfg(unix)] // recovery deletes a chunk file the reader holds mapped
+    #[test]
+    fn live_reload_drops_chunks_removed_by_writer_recovery() {
+        use common::universal_io::{CachedFs, CachedReadFs};
+
+        const DIM: usize = 32;
+        let dir = Builder::new().prefix("chunked_shrink").tempdir().unwrap();
+        let hw = HardwareCounterCell::disposable();
+
+        let mut writer =
+            UpdateOnlyChunkedVectors::<f32, MmapFile>::open(MmapFs, dir.path(), DIM).unwrap();
+        append_range(&mut writer, 0, 0..100, DIM, &hw);
+
+        // Crash leftover: a chunk file past the committed watermark.
+        fs_err::write(chunk_name(dir.path(), 1), vec![7u8; 128]).unwrap();
+
+        let mut reader = ReadOnlyChunkedVectors::<f32, MmapFile>::open(
+            &MmapFs,
+            dir.path(),
+            DIM,
+            AdviceSetting::Global,
+            Populate::No,
+        )
+        .unwrap();
+        assert_eq!(reader.len(), 100);
+        assert_eq!(reader.chunks.len(), 2, "leftover chunk is listed and held");
+
+        // The next batch trusts the watermark: it removes the leftover chunk,
+        // then lands in chunk 0.
+        append_range(&mut writer, 100, 100..150, DIM, &hw);
+        drop(writer);
+
+        let mut cached_fs = CachedFs::new(MmapFs, dir.path()).unwrap();
+        cached_fs.cache_file_info().unwrap();
+        LiveReload::live_preload(&reader, &cached_fs).unwrap();
+
+        let empty = SortedSlice::new(&[]).unwrap();
+        reader.live_reload(&cached_fs, &empty, &empty, &hw).unwrap();
+
+        assert_eq!(reader.len(), 150);
+        assert_eq!(reader.chunks.len(), 1, "removed trailing chunk is dropped");
+        for offset in [0, 99, 100, 149] {
+            assert_eq!(
+                reader.get::<Random>(offset).unwrap().as_ref(),
+                make_vec(offset, DIM).as_slice(),
+                "vector {offset} mismatch after reload",
+            );
+        }
+    }
+
     /// Case-5 regression of the live-reload staleness audit: a reader over a
     /// caching backend that fetched a block straddling the old tail (any read
     /// near the tail pulls a 16KiB block covering space appended into later)

lib/segment/src/vector_storage/chunked_vectors/update_only/mod.rs

@@ -72,7 +72,7 @@ where
     /// Needed where a storage's rows are not indexed by point slot — the
     /// multivector ones — so a batch knows where the row space ends.
     pub fn stored_len(&self) -> OperationResult<usize> {
-        read_status_len(&self.fs, &status_file(&self.directory))
+        Ok(read_status_len(&self.fs, &status_file(&self.directory))?)
     }
 
     /// How many more vectors fit in the chunk that `key` falls in.