[LiveReload] Prepare segment preload (#10221)

* genericize live_reload fs parameters

* impl live_preload for ReadOnlySegment

* split edge refresh into preload and apply passes

* only rotate file infos after successful reload

lib/common/common/src/universal_io/cached_fs/mod.rs

@@ -175,7 +175,7 @@ impl<Fs: UniversalReadFs> CachedFs<Fs> {
 }
 
 impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
-    /// Take a LIST snapshot of the filesystem and replace existing cached data.
+    /// Take a LIST snapshot of the filesystem and drop prefetched files.
     fn cache_file_info(&mut self) -> UioResult<()> {
         // List all files
         let list = self.fs.list_files(&self.prefix_path)?;
@@ -199,12 +199,17 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
             )
             .collect();
 
-        self.previous_files_info = self.files_info.replace(files_info);
+        self.files_info = Some(files_info);
         self.files_prefetched.lock().clear();
 
         Ok(())
     }
 
+    fn rotate_cache_file_info(&mut self) {
+        self.previous_files_info = self.files_info.take();
+        self.files_prefetched.lock().clear();
+    }
+
     fn schedule_prefetch(
         &self,
         path: &Path,

lib/common/common/src/universal_io/traits/file_ops.rs

@@ -160,6 +160,9 @@ pub trait CachedReadFs: UniversalReadFs {
     /// fail with `NotFound` without touching the underlying filesystem.
     fn cache_file_info(&mut self) -> UioResult<()>;
 
+    /// Rotate the cache file info, keeping it as the previous snapshot.
+    fn rotate_cache_file_info(&mut self);
+
     /// Open `path` in the background and park the handle in the prefetch
     /// pool, to be consumed by a later [`UniversalReadFs::open`] of the same
     /// path. Idempotent per path while the handle is unconsumed.

lib/edge/src/read_only/lifecycle.rs

@@ -106,6 +106,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlyEdgeShard<S> {
             enumerator: Box::new(enumerator),
             search_pool,
             load_profile,
+            refresh_lock: Default::default(),
         };
         shard.refresh()?;
 

lib/edge/src/read_only/mod.rs

@@ -24,7 +24,7 @@ pub(crate) mod tests;
 use std::path::{Path, PathBuf};
 use std::sync::Arc;
 
-use parking_lot::RwLock;
+use parking_lot::{Mutex, RwLock};
 use segment::data_types::load_profile::LoadProfile;
 use segment::index::UniversalReadExt;
 
@@ -61,6 +61,10 @@ pub struct ReadOnlyEdgeShard<S: UniversalReadExt + 'static> {
     /// won't touch are parked cold instead of warmed per the segment configs. Kept so segments a
     /// later [`refresh`](Self::refresh) discovers load with the same placement.
     load_profile: Option<LoadProfile>,
+    /// Serializes [`refresh`](Self::refresh)es: concurrent refreshes would
+    /// duplicate the listing and load work, and could clear each other's staged
+    /// prefetches between a segment's preload and apply.
+    refresh_lock: Mutex<()>,
 }
 
 impl<S: UniversalReadExt + 'static> ReadOnlyEdgeShard<S> {

lib/edge/src/read_only/refresh.rs

@@ -4,6 +4,7 @@ use std::sync::Arc;
 use common::counter::hardware_counter::HardwareCounterCell;
 use common::universal_io::IsNotFound as _;
 use parking_lot::RwLock;
+use rayon::prelude::*;
 use segment::common::operation_error::{OperationError, OperationResult};
 use segment::index::UniversalReadExt;
 use segment::segment::read_only::ReadOnlySegment;
@@ -51,6 +52,8 @@ impl<S: UniversalReadExt + 'static> ReadOnlyEdgeShard<S> {
     where
         S::Fs: Send + Sync + Clone + 'static,
     {
+        let _refresh_guard = self.refresh_lock.lock();
+
         // A benign mid-attempt segment removal re-runs the attempt against the fresh manifest;
         // bound the re-runs so a leader churning segments faster than the follower converges
         // cannot spin this loop forever.
@@ -129,7 +132,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlyEdgeShard<S> {
 
             survivor_uuids
                 .into_iter()
-                .filter_map(|uuid| holder.segment_arc(&uuid).map(|segment| (uuid, segment)))
+                .filter_map(|uuid| Some((uuid, holder.segment_arc(&uuid)?)))
                 .collect()
         };
 
@@ -156,15 +159,25 @@ impl<S: UniversalReadExt + 'static> ReadOnlyEdgeShard<S> {
             *self.config.write() = Arc::new(derived);
         }
 
-        // 4. Live-reload survivors to fold in the leader's flushed in-place appends and deletes.
-        //    Newly-added segments are already current, so they are skipped. Survivors are
-        //    independent, so a failure does not stop the others from reloading; a failed segment
-        //    keeps serving its pre-refresh state and its unapplied delta is retained in
-        //    `pending_reload`, so a later reload replays the union and nothing is lost.
+        // 4. Live-reload survivors to assimilate new appends and deletes from data.
+        // Done in 2 steps: preload -> reload, so that we can avoid locking when prefetching.
+        self.search_pool.install(|| {
+            survivors.par_iter().for_each(|(uuid, segment)| {
+                let _ = segment.read().live_preload().inspect_err(|err| {
+                    log::warn!("live_preload of segment {uuid} failed: {err}");
+                });
+            });
+        });
+
         let mut not_found: Vec<(Uuid, OperationError)> = Vec::new();
         let mut first_hard_error: Option<OperationError> = None;
+
+        // TODO(uio): currently this locks each segment one at a time. Once we
+        // do `live_preload` async we'll have a clear signal of finishing
+        // prefetches to start locking. By then, we can also make this section a
+        // rayon par_iter, and take care of hw_counter not being thread-safe.
         for (uuid, segment) in survivors {
-            match segment.write().live_reload(&self.fs, hw_counter) {
+            match segment.write().live_reload(hw_counter) {
                 Ok(()) => {}
                 // An essential file is gone; whether that is benign (the leader removed the
                 // segment while we reloaded it) is decided against a re-read manifest below.

lib/segment/src/common/live_reload.rs

@@ -31,7 +31,6 @@ use crate::common::operation_error::OperationResult;
 pub(crate) trait LiveReload {
     type File: UniversalRead;
 
-    #[expect(dead_code)]
     fn live_preload<Fs: CachedReadFs<File = Self::File>>(
         &self,
         cached_fs: &Fs,

lib/segment/src/id_tracker/id_tracker_base/read_only_tracker_enum.rs

@@ -46,9 +46,8 @@ impl<S: UniversalRead> ReadOnlyIdTrackerEnum<S> {
     ///
     /// The attempts are sequential for now; they are independent and can be
     /// issued concurrently later (the slow-path being remote opens).
-    /// `raw_fs` is the canonical backend for the appendable tracker, which
-    /// stores a filesystem handle to re-open the append-only files on later
-    /// reloads — a caching wrapper's snapshot would go stale.
+    /// `raw_fs` is the canonical backend for the appendable tracker's
+    /// bootstrap opens, which bypass any prefetch pool.
     pub fn detect_and_load(
         fs: &impl UniversalReadFs<File = S>,
         raw_fs: &S::Fs,
@@ -68,15 +67,24 @@ impl<S: UniversalRead> ReadOnlyIdTrackerEnum<S> {
         )?))
     }
 
+    /// Stage everything the next [`Self::live_reload`] needs. Shared access.
+    pub fn live_preload(&self, fs: &impl CachedReadFs<File = S>) -> OperationResult<()> {
+        // todo(uio): dispatch per variant as the trackers gain live_preload
+        let _ = fs;
+        Ok(())
+    }
+
     /// Reload externally-applied changes, dispatching to the active variant.
     ///
     /// `fs` refreshes storages that mutate in place (the immutable and disk
-    /// trackers' deleted bitmaps) by opening fresh handles; the appendable
-    /// tracker keeps its own raw fs handle instead (see
-    /// [`Self::detect_and_load`]).
-    pub fn live_reload(&mut self, fs: &S::Fs) -> OperationResult<LiveReloadResult> {
+    /// trackers' deleted bitmaps) by opening fresh handles, and serves the
+    /// appendable tracker's lazy file opens.
+    pub fn live_reload<Fs: UniversalReadFs<File = S>>(
+        &mut self,
+        fs: &Fs,
+    ) -> OperationResult<LiveReloadResult> {
         match self {
-            Self::Appendable(id_tracker) => id_tracker.live_reload(),
+            Self::Appendable(id_tracker) => id_tracker.live_reload(fs),
             Self::Immutable(id_tracker) => id_tracker.live_reload(fs),
             Self::DiskResident(id_tracker) => id_tracker.live_reload(fs),
         }

lib/segment/src/id_tracker/mutable_id_tracker/read_only/lifecycle.rs

@@ -59,14 +59,11 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
         segment_path: impl Into<PathBuf>,
         deferred_internal_id: Option<PointOffsetType>,
     ) -> OperationResult<Self> {
-        // The tracker keeps a filesystem handle to re-open the append-only
-        // files on later reloads, so it must retain the *raw* backend — a
-        // caching wrapper's snapshot goes stale as soon as the writer
-        // appends. The bootstrap below consequently opens through the raw
-        // fs too, bypassing any prefetch pool.
+        // The bootstrap below opens through the raw fs passed here, bypassing
+        // any prefetch pool. Later reloads open through the fs their caller
+        // provides instead (typically a caching wrapper with a fresh snapshot).
         let mut tracker = Self {
             segment_path: segment_path.into(),
-            fs: fs.clone(),
             internal_to_version: Vec::new(),
             mappings: PointMappings::new(
                 Default::default(),
@@ -85,7 +82,7 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
 
         // Load the existing data the same way a live-reload consumes appended data. The reported
         // delta (the whole committed set as inserts) is irrelevant for an initial open.
-        tracker.live_reload()?;
+        tracker.live_reload(fs)?;
 
         #[cfg(debug_assertions)]
         tracker.mappings.assert_mappings();
@@ -128,7 +125,10 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
     /// second round-trip on object storage. Lazy backends (e.g. S3) touch the object only on the
     /// first read, so a missing object can instead surface as `NotFound` from a later `len`/`read`
     /// — `live_reload` tolerates that case too.
-    pub(super) fn try_open(fs: &S::Fs, path: &Path) -> OperationResult<Option<S>> {
+    pub(super) fn try_open(
+        fs: &impl UniversalReadFs<File = S>,
+        path: &Path,
+    ) -> OperationResult<Option<S>> {
         let options = Self::open_options();
         Ok(fs.open(path, options, Default::default()).ok_not_found()?)
     }

lib/segment/src/id_tracker/mutable_id_tracker/read_only/live_reload.rs

@@ -2,7 +2,7 @@ use std::io::Cursor;
 
 use common::generic_consts::Sequential;
 use common::types::PointOffsetType;
-use common::universal_io::{OkNotFound, ReadRange, UniversalRead};
+use common::universal_io::{OkNotFound, ReadRange, UniversalRead, UniversalReadFs};
 
 use super::ReadOnlyAppendableIdTracker;
 use crate::common::operation_error::OperationResult;
@@ -62,7 +62,8 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
     /// Consume mapping and version changes appended to storage since the last reload.
     ///
     /// File handles are refreshed via [`UniversalRead::reopen`] so data appended by the writer
-    /// becomes visible. Both result lists are sorted ascending.
+    /// becomes visible; not-yet-opened files are opened lazily through `fs` (a caching wrapper's
+    /// prefetch pool serves these opens when staged). Both result lists are sorted ascending.
     ///
     /// The writer flushes mappings before data before versions, so a point's version appears last
     /// and marks it as fully committed. Inserts are therefore driven by the versions file: an
@@ -71,14 +72,17 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
     /// withheld (its data may be partial) and reported on a later reload once its version lands.
     /// Deletes are driven by the mapping and need no version, a deleted point's version is
     /// considered gone.
-    pub fn live_reload(&mut self) -> OperationResult<LiveReloadResult> {
+    pub fn live_reload(
+        &mut self,
+        fs: &impl UniversalReadFs<File = S>,
+    ) -> OperationResult<LiveReloadResult> {
         // Append versions flushed since the last reload (mappings are flushed before versions).
         // `committed` is the exclusive offset bound for which versions exist, i.e. the commit mark.
-        let committed = self.reload_versions()? as PointOffsetType;
+        let committed = self.reload_versions(fs)? as PointOffsetType;
 
         // Consume new mapping changes. Inserts are buffered until committed (their version exists);
         // deletes act on the committed mapping immediately, or cancel a still-pending insert.
-        let changes = self.read_new_mapping_changes()?;
+        let changes = self.read_new_mapping_changes(fs)?;
 
         for change in &changes {
             log::trace!(target: "live-reload", "Read mapping in {:?} change: {:?}", self.segment_path, change);
@@ -131,11 +135,14 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
     ///
     /// The read stops at the last fully-readable entry; a partial trailing entry is left in place
     /// so it can be consumed on a later reload once the writer flushed it completely.
-    fn read_new_mapping_changes(&mut self) -> OperationResult<Vec<MappingChange>> {
+    fn read_new_mapping_changes(
+        &mut self,
+        fs: &impl UniversalReadFs<File = S>,
+    ) -> OperationResult<Vec<MappingChange>> {
         // The mappings file is absent until the writer flushes the first point; open it lazily once
         // it appears. Until then there is nothing to read.
         if self.mappings_file.is_none() {
-            self.mappings_file = Self::try_open(&self.fs, &mappings_path(&self.segment_path))?;
+            self.mappings_file = Self::try_open(fs, &mappings_path(&self.segment_path))?;
         }
         let Some(file) = self.mappings_file.as_mut() else {
             return Ok(Vec::new());
@@ -185,11 +192,11 @@ impl<S: UniversalRead> ReadOnlyAppendableIdTracker<S> {
     /// slot, so [`internal_version`](crate::id_tracker::IdTrackerRead::internal_version) returns
     /// `None` for it (it is never given a fake version) until its version is appended here. We do
     /// not read versions for deleted points, a deleted point's version is considered gone.
-    fn reload_versions(&mut self) -> OperationResult<usize> {
+    fn reload_versions(&mut self, fs: &impl UniversalReadFs<File = S>) -> OperationResult<usize> {
         // The versions file is absent until the writer flushes the first point; open it lazily once
         // it appears. Until then no version is committed.
         if self.versions_file.is_none() {
-            self.versions_file = Self::try_open(&self.fs, &versions_path(&self.segment_path))?;
+            self.versions_file = Self::try_open(fs, &versions_path(&self.segment_path))?;
         }
         let Some(versions_file) = self.versions_file.as_mut() else {
             return Ok(self.internal_to_version.len());

lib/segment/src/id_tracker/mutable_id_tracker/read_only/mod.rs

@@ -28,8 +28,8 @@ use crate::types::{PointIdType, SeqNumberType};
 /// The mappings and versions files may be absent: the writer only creates them once it flushes the
 /// first point (an empty file is never written), exactly as
 /// [`MutableIdTracker::open`](crate::id_tracker::mutable_id_tracker::MutableIdTracker::open)
-/// tolerates. A missing file is treated as an empty storage; the retained [`Self::fs`] lets the
-/// handle be opened lazily once the file appears.
+/// tolerates. A missing file is treated as an empty storage; the handle is opened lazily through
+/// the fs passed to [`Self::live_reload`] once the file appears.
 ///
 /// The mapping only ever contains *committed* points. The writer flushes mappings before data
 /// before versions, so a point is fully written only once its version is present. An insert read
@@ -37,9 +37,6 @@ use crate::types::{PointIdType, SeqNumberType};
 /// flushed, and only then linked into [`Self::mappings`].
 pub struct ReadOnlyAppendableIdTracker<S: UniversalRead> {
     segment_path: PathBuf,
-    /// Filesystem handle, retained so the mappings/versions files can be opened lazily once the
-    /// writer creates them (they are absent while empty).
-    fs: S::Fs,
     internal_to_version: Vec<SeqNumberType>,
     mappings: PointMappings,
 

lib/segment/src/id_tracker/mutable_id_tracker/read_only/tests.rs

@@ -135,7 +135,7 @@ fn test_live_reload_reports_inserts_and_deletes() {
 
     // A reload with no new changes reports nothing
     assert_eq!(
-        read_only.live_reload().unwrap(),
+        read_only.live_reload(&MmapFs).unwrap(),
         LiveReloadResult::default(),
     );
 
@@ -145,7 +145,7 @@ fn test_live_reload_reports_inserts_and_deletes() {
     mutable.drop(200.into()).unwrap();
     flush(&mutable);
 
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result.inserted, vec![3, 4]);
     assert_eq!(result.deleted, vec![1]);
     assert_in_sync(&read_only, &mutable);
@@ -173,7 +173,7 @@ fn test_live_reload_insert_then_delete_within_batch() {
     mutable.drop(200.into()).unwrap();
     flush(&mutable);
 
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result.inserted, Vec::<PointOffsetType>::new());
     assert_eq!(result.deleted, Vec::<PointOffsetType>::new());
     assert_in_sync(&read_only, &mutable);
@@ -210,7 +210,7 @@ fn test_live_reload_upsert_relinks_to_new_offset() {
     mutable.set_internal_version(1, 20).unwrap();
     flush(&mutable);
 
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result.inserted, vec![1]);
     assert_eq!(result.deleted, vec![0]);
     assert_eq!(
@@ -244,7 +244,7 @@ fn test_live_reload_withholds_insert_until_version_present() {
 
     // The version is not flushed yet, so the point is withheld from the result and, crucially, is
     // not present in the mapping at all (its data may be partially written).
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result, LiveReloadResult::default());
     assert_eq!(
         read_only
@@ -257,7 +257,7 @@ fn test_live_reload_withholds_insert_until_version_present() {
     // the insert, links it into the mapping, and reconciles the version.
     mutable.versions_flusher()().unwrap();
 
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result.inserted, vec![1]);
     assert_eq!(result.deleted, Vec::<PointOffsetType>::new());
     assert_eq!(
@@ -300,7 +300,7 @@ fn test_live_reload_ignores_partial_trailing_mapping_entry() {
     }
 
     // The partial entry is ignored and we don't advance past it.
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result, LiveReloadResult::default());
     assert_eq!(
         read_only.mappings_read_to, complete_len,
@@ -316,7 +316,7 @@ fn test_live_reload_ignores_partial_trailing_mapping_entry() {
         insert(&mut mutable, 300.into(), 2, 12);
         flush(&mutable);
 
-        let result = read_only.live_reload().unwrap();
+        let result = read_only.live_reload(&MmapFs).unwrap();
         assert_eq!(result.inserted, vec![2]);
         assert_eq!(result.deleted, Vec::<PointOffsetType>::new());
         assert_in_sync(&read_only, &mutable);
@@ -377,7 +377,7 @@ fn test_live_reload_withholds_partially_written_version() {
     }
 
     // Only part of the version is written, so the point is withheld.
-    let result = read_only.live_reload().unwrap();
+    let result = read_only.live_reload(&MmapFs).unwrap();
     assert_eq!(result.inserted, Vec::<PointOffsetType>::new());
     assert_eq!(read_only.internal_version(2), None);
 
@@ -389,7 +389,7 @@ fn test_live_reload_withholds_partially_written_version() {
     {
         mutable.versions_flusher()().unwrap();
 
-        let result = read_only.live_reload().unwrap();
+        let result = read_only.live_reload(&MmapFs).unwrap();
         assert_eq!(result.inserted, vec![2]);
         assert_eq!(
             read_only.internal_id_with_behavior(

lib/segment/src/segment/read_only/config_reload.rs

@@ -248,6 +248,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             payload_index,
             payload_storage: _,
             pending_reload: _,
+            reload_fs: _,
             segment_type,
             segment_config,
         } = self;

lib/segment/src/segment/read_only/lifecycle.rs

@@ -8,6 +8,7 @@ use common::types::PointOffsetType;
 use common::universal_io::{
     CachedFs, CachedReadFs, OkNotFound, Populate, UniversalReadFs, read_json_via,
 };
+use parking_lot::Mutex;
 use uuid::Uuid;
 
 use super::{ReadOnlySegment, ReadOnlyVectorData};
@@ -106,7 +107,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         let (segment_config, payload_config) =
             Self::first_preopen(&cached_fs, segment_path, load_profile)?;
         Self::open_via(
-            &cached_fs,
+            cached_fs,
             fs,
             segment_path,
             segment_config,
@@ -195,12 +196,12 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
     /// Read-only mirror of `load_segment`: assembles every read-only component
     /// from `fs` (id tracker, payload storage+index, per-vector storage/index). No writes.
     ///
-    /// `fs` is any filesystem producing `S`-typed handles — in production the
-    /// per-segment [`CachedReadFs`], whose opens are served from its prefetch
-    /// pool. `raw_fs` is the canonical backend, for the components that store
-    /// a filesystem handle to read files after this open (the appendable id
-    /// tracker's re-opens, the HNSW index's deferred graph load): a caching
-    /// wrapper's snapshot would go stale.
+    /// `fs` is the per-segment caching filesystem, whose opens are served from
+    /// its prefetch pool; the segment retains it for live reloads (see
+    /// [`ReadOnlySegment::reload_fs`]). `raw_fs` is the canonical backend, for
+    /// the components that store a filesystem handle to read files after this
+    /// open (the appendable id tracker's re-opens, the HNSW index's deferred
+    /// graph load): a caching wrapper's snapshot would go stale.
     ///
     /// `config` and `payload_config` are the ones
     /// [`first_preopen`](Self::first_preopen) already parsed off `fs`, and
@@ -208,7 +209,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
     /// here must make the same placement decisions the prefetches did.
     #[allow(clippy::too_many_arguments)]
     pub(crate) fn open_via(
-        fs: &impl CachedReadFs<File = S>,
+        fs: CachedFs<S::Fs>,
         raw_fs: &S::Fs,
         segment_path: &Path,
         config: SegmentConfig,
@@ -217,7 +218,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         deferred_internal_id: Option<PointOffsetType>,
         load_profile: Option<&LoadProfile>,
     ) -> OperationResult<Self> {
-        if SegmentVersion::load_universal(fs, segment_path)?.is_none() {
+        if SegmentVersion::load_universal(&fs, segment_path)?.is_none() {
             // `FileNotFound`, not a service error: the version file is written last, so
             // its absence means the segment vanished mid-open (or was never completed) —
             // a follower resolves that against the segment manifest.
@@ -233,7 +234,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             .and_then(|profile| profile.payload_storage_placement())
             .unwrap_or_else(|| payload_populate(&config));
         let payload_storage = Arc::new(AtomicRefCell::new(ReadOnlyPayloadStorage::open(
-            fs,
+            &fs,
             segment_path.to_path_buf(),
             payload_storage_populate,
         )?));
@@ -241,7 +242,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         // Detect the persisted format by attempting each format's open (no
         // per-file `exists` round-trips — important for object-storage backends).
         let id_tracker = Arc::new(AtomicRefCell::new(ReadOnlyIdTrackerEnum::detect_and_load(
-            fs,
+            &fs,
             raw_fs,
             segment_path,
             deferred_internal_id,
@@ -256,7 +257,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             let path = get_vector_storage_path(segment_path, vector_name);
             let storage_populate =
                 load_profile.and_then(|profile| profile.vector_storage_placement(vector_name));
-            let storage = VectorStorageReadEnum::open(fs, vector_config, &path, storage_populate)?
+            let storage = VectorStorageReadEnum::open(&fs, vector_config, &path, storage_populate)?
                 .ok_or_else(|| {
                     OperationError::service_error(format!(
                         "Read-only dense vector storage '{vector_name}' was not found, or is corrupted.",
@@ -268,7 +269,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             let path = get_vector_storage_path(segment_path, vector_name);
             let storage =
                 VectorStorageReadEnum::Sparse(Box::new(ReadOnlySparseVectorStorage::open(
-                    fs,
+                    &fs,
                     &path,
                     sparse_storage_populate(sparse_vector_config),
                 )?));
@@ -276,7 +277,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         }
 
         let payload_index = Arc::new(AtomicRefCell::new(ReadOnlyStructPayloadIndex::open(
-            fs,
+            &fs,
             payload_storage.clone(),
             id_tracker.clone(),
             vector_storages.clone(),
@@ -289,7 +290,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         for (vector_name, vector_config) in &config.vector_data {
             let vector_storage = vector_storages.remove(vector_name).unwrap();
             let data = ReadOnlyVectorData::open_dense(
-                fs,
+                &fs,
                 raw_fs,
                 segment_path,
                 vector_name,
@@ -305,7 +306,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
         for (vector_name, sparse_vector_config) in &config.sparse_vector_data {
             let vector_storage = vector_storages.remove(vector_name).unwrap();
             let data = ReadOnlyVectorData::open_sparse(
-                fs,
+                &fs,
                 segment_path,
                 vector_name,
                 sparse_vector_config,
@@ -331,6 +332,7 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             payload_index,
             payload_storage,
             pending_reload: AtomicRefCell::new(Default::default()),
+            reload_fs: Mutex::new(fs),
             segment_type,
             segment_config: config,
         })

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

@@ -1,6 +1,7 @@
 use common::counter::hardware_counter::HardwareCounterCell;
 use common::sorted_slice::SortedSlice;
 use common::types::PointOffsetType;
+use common::universal_io::{CachedReadFs, UniversalReadFs};
 
 use super::{ReadOnlySegment, ReadOnlyVectorData};
 use crate::common::live_reload::LiveReload;
@@ -9,18 +10,50 @@ use crate::id_tracker::mutable_id_tracker::read_only::LiveReloadResult;
 use crate::index::UniversalReadExt;
 
 impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
+    /// Stage every component's next [`Self::live_reload`] under shared access:
+    /// re-snapshot the retained caching filesystem's listing, then schedule
+    /// every fetch the reload will need. Fetches go in flight as scheduled.
+    pub fn live_preload(&self) -> OperationResult<()> {
+        let Self {
+            uuid: _,
+            segment_path: _,
+            id_tracker,
+            vector_data,
+            payload_index,
+            payload_storage,
+            pending_reload: _,
+            reload_fs,
+            segment_type: _,
+            segment_config: _,
+        } = self;
+
+        let mut reload_fs = reload_fs.lock();
+        // perf: one LIST per segment per refresh; could be a single shard-prefix
+        // LIST partitioned into the per-segment snapshots.
+        reload_fs.cache_file_info()?;
+        let fs = &*reload_fs;
+
+        id_tracker.borrow().live_preload(fs)?;
+        payload_storage.borrow().live_preload(fs)?;
+        payload_index.borrow().live_preload(fs)?;
+        for vector_data in vector_data.values() {
+            vector_data.live_preload(fs)?;
+        }
+        Ok(())
+    }
+
     /// Refresh every component to the current on-disk state (id-tracker delta → all components).
     ///
+    /// Must follow a [`Self::live_preload`]: opens resolve against (and consume)
+    /// what it staged, and files that appeared since its listing snapshot are
+    /// not visible.
+    ///
     /// Draining the id-tracker advances its internal state and cannot be replayed,
     /// so the delta is accumulated into `pending_reload` and only cleared once
     /// every component has reloaded successfully. If a component fails mid-way the
     /// delta is retained, and a later reload folds in the tracker's new changes and
     /// replays the union — no component is left drifting on a partial reload.
-    pub fn live_reload(
-        &mut self,
-        fs: &S::Fs,
-        hw_counter: &HardwareCounterCell,
-    ) -> OperationResult<()> {
+    pub fn live_reload(&mut self, hw_counter: &HardwareCounterCell) -> OperationResult<()> {
         let Self {
             uuid: _,
             segment_path: _,
@@ -29,10 +62,13 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             payload_index,
             payload_storage,
             pending_reload,
+            reload_fs,
             segment_type: _,
             segment_config: _,
         } = self;
 
+        let fs = &mut *reload_fs.get_mut();
+
         // Drain the tracker delta and fold it into whatever a previous reload left
         // unapplied. This must happen before any component reload can fail, so the
         // accumulated delta survives an error and is replayed on the next call.
@@ -61,23 +97,41 @@ impl<S: UniversalReadExt + 'static> ReadOnlySegment<S> {
             }
         }
 
-        // Every component is now in sync; discard the applied delta.
+        // Every component is now in sync; discard the applied delta and rotate
+        // file info.
         *pending = LiveReloadResult::default();
+        fs.rotate_cache_file_info();
 
         Ok(())
     }
 }
 
 impl<S: UniversalReadExt + 'static> ReadOnlyVectorData<S> {
+    /// Stage this vector's next [`Self::live_reload`]. Shared access only.
+    fn live_preload(&self, fs: &impl CachedReadFs<File = S>) -> OperationResult<()> {
+        let Self {
+            vector_index,
+            vector_storage,
+            quantized_vectors,
+        } = self;
+
+        vector_storage.borrow().live_preload(fs)?;
+        vector_index.borrow().live_preload(fs)?;
+        if let Some(quantized_vectors) = quantized_vectors.borrow().as_ref() {
+            quantized_vectors.live_preload(fs)?;
+        }
+        Ok(())
+    }
+
     /// Refresh this vector's storage, index and quantized vectors to the current
     /// on-disk state.
     ///
     /// `Self` is destructured so that every component is covered: adding a field
     /// without reloading it won't compile. Each component mutates through its own
     /// `Arc<AtomicRefCell<_>>`, so `&self` is enough — no `&mut` is needed.
-    fn live_reload(
+    fn live_reload<Fs: UniversalReadFs<File = S>>(
         &self,
-        fs: &S::Fs,
+        fs: &Fs,
         deleted: &SortedSlice<'_, PointOffsetType>,
         inserted: &SortedSlice<'_, PointOffsetType>,
         hw_counter: &HardwareCounterCell,

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

@@ -4,6 +4,8 @@ use std::path::PathBuf;
 use std::sync::Arc;
 
 use atomic_refcell::{AtomicRef, AtomicRefCell};
+use common::universal_io::CachedFs;
+use parking_lot::Mutex;
 use uuid::Uuid;
 
 use crate::id_tracker::mutable_id_tracker::read_only::LiveReloadResult;
@@ -48,6 +50,12 @@ pub struct ReadOnlySegment<S: UniversalReadExt + 'static> {
     /// can't drift out of sync. Empty once everything is in sync.
     pub pending_reload: AtomicRefCell<LiveReloadResult>,
 
+    /// Caching filesystem retained from open: [`live_preload`](ReadOnlySegment::live_preload)
+    /// re-snapshots its listing and stages into its prefetch pool, [`live_reload`](ReadOnlySegment::live_reload)
+    /// consumes what was staged. Retaining it gives unchanged-detection a previous
+    /// listing to compare against — the open-time one for the first reload.
+    pub(crate) reload_fs: Mutex<CachedFs<S::Fs>>,
+
     /// Shows what kind of indexes and storages are used in this segment
     pub segment_type: SegmentType,
     pub segment_config: SegmentConfig,

lib/segment/src/segment/read_only/tests.rs

@@ -480,9 +480,7 @@ fn read_only_segment_sparse_mutable_ram_matches_mutable() {
     mutable.flush(true).unwrap();
 
     // A live reload folds the delta into the rebuilt sparse index.
-    read_only
-        .live_reload(&MmapFs, &hw)
-        .expect("read-only live reload");
+    preload_then_reload(&mut read_only, &hw).expect("read-only live reload");
     assert_eq!(read_only.available_point_count(), NUM_POINTS + 20 - 2);
 
     assert_sparse_search_matches(&mutable, &read_only, &query);
@@ -491,6 +489,16 @@ fn read_only_segment_sparse_mutable_ram_matches_mutable() {
     assert_sparse_search_matches(&mutable, &read_only, &fresh_dim_query);
 }
 
+/// Full refresh cycle as the edge shard drives it: staged preload (shared
+/// access), then the exclusive apply.
+fn preload_then_reload(
+    segment: &mut ReadOnlySegment<MmapFile>,
+    hw_counter: &HardwareCounterCell,
+) -> crate::common::operation_error::OperationResult<()> {
+    segment.live_preload()?;
+    segment.live_reload(hw_counter)
+}
+
 /// Drive `config_reload_diff` + `apply_config_reload`: toggle the on-disk
 /// payload config (the `num` field index) while its index files stay in place,
 /// and assert the read-only segment loads/drops the field index accordingly and
@@ -595,15 +603,17 @@ fn vanished_segment_classifies_not_found() {
     assert!(err.is_not_found(), "expected not-found, got: {err}");
 
     // Live-reload of a segment whose directory vanished after open (the leader
-    // removed it): the first component reopen hits the missing file.
+    // removed it): the first component reopen hits the missing file. The failed
+    // preload mutates nothing — the segment keeps serving its pre-refresh state.
     let mut read_only =
         ReadOnlySegment::<MmapFile>::open(&MmapFs, &segment_path, segment_uuid, None, None)
             .expect("read-only open");
+    let points_before = read_only.available_point_count();
     fs_err::remove_dir_all(&segment_path).unwrap();
-    let err = read_only
-        .live_reload(&MmapFs, &HardwareCounterCell::disposable())
+    let err = preload_then_reload(&mut read_only, &HardwareCounterCell::disposable())
         .expect_err("live_reload over a removed directory must fail");
     assert!(err.is_not_found(), "expected not-found, got: {err}");
+    assert_eq!(read_only.available_point_count(), points_before);
 }
 
 /// A load profile that never scores a vector defers its HNSW graph load: the