[CachedFs] Unchanged file open is no-op (#10050)

* add `UniversalIoError::UnchangedOpen`

* add & impl `CachedReadFs::reschedule_prefetch`

* clippy

* add `OkUnchanged` helper

* propagate scheduling errors immediately

* drop lock before reacquiring it

* clear prefetched_files on new snapshot

* only avoid prefetch on full FileInfo equality

* `UnchangedOpen` maps to `Cancelled`

* match-all match

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

@@ -20,6 +20,21 @@ pub struct FileInfo {
     pub last_modified: Option<std::time::SystemTime>,
 }
 
+impl FileInfo {
+    /// Return true if both `FileInfo` have all data and it's equal.
+    pub fn full_eq(&self, other: &FileInfo) -> bool {
+        let Self {
+            size,
+            last_modified,
+        } = self;
+
+        size == &other.size
+            && last_modified
+                .zip(other.last_modified)
+                .is_some_and(|(this, other)| this == other)
+    }
+}
+
 /// Read-only filesystem wrapper that snapshots the file listing and serves
 /// opens from explicitly prefetched handles. The only [`CachedReadFs`]
 /// implementation.
@@ -50,7 +65,9 @@ pub struct CachedFs<Fs: UniversalReadFs> {
     /// `None` until [`CachedFs::cache_file_info`] takes the listing
     /// snapshot; the wrapper forwards to `fs` until then.
     files_info: Option<HashMap<PathBuf, FileInfo>>,
-    files_prefetched: Arc<Mutex<HashMap<PathBuf, Fs::File>>>,
+    /// Previous listing snapshot.
+    previous_files_info: Option<HashMap<PathBuf, FileInfo>>,
+    files_prefetched: Arc<Mutex<HashMap<PathBuf, Option<Fs::File>>>>,
 }
 
 /// Manual impl: `derive(Clone)` would add a spurious `Fs::File: Clone`
@@ -62,12 +79,14 @@ impl<Fs: UniversalReadFs> Clone for CachedFs<Fs> {
             fs,
             prefix_path,
             files_info,
+            previous_files_info,
             files_prefetched,
         } = self;
         Self {
             fs: fs.clone(),
             prefix_path: prefix_path.clone(),
             files_info: files_info.clone(),
+            previous_files_info: previous_files_info.clone(),
             files_prefetched: files_prefetched.clone(),
         }
     }
@@ -79,6 +98,7 @@ impl<Fs: UniversalReadFs> CachedFs<Fs> {
             fs,
             prefix_path: prefix_path.to_path_buf(),
             files_info: None,
+            previous_files_info: None,
             files_prefetched: Arc::new(Mutex::new(HashMap::new())),
         })
     }
@@ -98,6 +118,11 @@ impl<Fs: UniversalReadFs> CachedFs<Fs> {
         self.files_info.as_ref()?.get(path)
     }
 
+    /// Previous file info from the snapshot; `None` before [`CachedFs::cache_file_info`] is called twice.
+    pub fn previous_file_info(&self, path: &Path) -> Option<&FileInfo> {
+        self.previous_files_info.as_ref()?.get(path)
+    }
+
     /// Files matching `prefix_path` in the snapshot; empty before
     /// [`CachedFs::cache_file_info`].
     ///
@@ -137,6 +162,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.
     fn cache_file_info(&mut self) -> UioResult<()> {
         // List all files
         let list = self.fs.list_files(&self.prefix_path)?;
@@ -158,7 +184,8 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
             )
             .collect();
 
-        self.files_info = Some(files_info);
+        self.previous_files_info = self.files_info.replace(files_info);
+        self.files_prefetched.lock().clear();
 
         Ok(())
     }
@@ -188,11 +215,39 @@ impl<Fs: UniversalReadFs> CachedReadFs for CachedFs<Fs> {
         }
 
         let file = self.fs.open(path, open_options, open_extra)?;
-        files_prefetched.insert(path.to_path_buf(), file);
+        files_prefetched.insert(path.to_path_buf(), Some(file));
 
         Ok(())
     }
 
+    fn reschedule_prefetch(
+        &self,
+        path: &Path,
+        open_arguments: Option<OpenOptions>,
+        open_extra: Option<Fs::OpenExtra>,
+    ) -> UioResult<()> {
+        {
+            let mut files_prefetched = self.files_prefetched.lock();
+
+            if files_prefetched.contains_key(path) {
+                return Ok(());
+            }
+
+            // Check if their file info is complete and didn't change.
+            if self
+                .previous_file_info(path)
+                .zip(self.file_info(path))
+                .is_some_and(|(previous, current)| previous.full_eq(current))
+            {
+                files_prefetched.insert(path.to_path_buf(), None);
+                return Ok(());
+            }
+        }
+
+        // Otherwise schedule normally
+        self.schedule_prefetch(path, open_arguments, open_extra)
+    }
+
     fn cached_file_info(&self, path: &Path) -> Option<FileInfo> {
         self.files_info.as_ref()?.get(path).cloned()
     }
@@ -235,12 +290,14 @@ impl<Fs: UniversalReadFs> Debug for CachedFs<Fs> {
             fs,
             prefix_path,
             files_info,
+            previous_files_info,
             files_prefetched,
         } = self;
         f.debug_struct("CachedReadFs")
             .field("fs", fs)
             .field("prefix_path", prefix_path)
             .field("files_info", files_info)
+            .field("previous_files_info", previous_files_info)
             .field("files_prefetched", &*files_prefetched.lock())
             .finish()
     }
@@ -270,7 +327,13 @@ impl<Fs: UniversalReadFs> UniversalReadFs for CachedFs<Fs> {
         }
 
         if let Some(file) = self.files_prefetched.lock().remove(path) {
-            return Ok(file);
+            return match file {
+                Some(file) => Ok(file),
+                None => Err(UniversalIoError::UnchangedOpen {
+                    path: path.to_owned(),
+                    since: self.file_info(path).and_then(|info| info.last_modified),
+                }),
+            };
         }
 
         // With a snapshot, unlisted paths fail locally — probing for

lib/common/common/src/universal_io/error.rs

@@ -47,6 +47,7 @@ impl IsNotFound for UniversalIoError {
             | Self::OutOfBounds { .. }
             | Self::InvalidFileIndex { .. }
             | Self::Uninitialized { .. }
+            | Self::UnchangedOpen { .. }
             | Self::QueueIsFull
             | Self::AppendOffsetConflict { .. }
             | Self::S3(_)
@@ -56,6 +57,26 @@ impl IsNotFound for UniversalIoError {
     }
 }
 
+pub trait OkUnchanged {
+    type Ok;
+    type Error;
+
+    fn ok_unchanged(self) -> Result<Option<Self::Ok>, Self::Error>;
+}
+
+impl<T> OkUnchanged for Result<T, UniversalIoError> {
+    type Ok = T;
+    type Error = UniversalIoError;
+
+    fn ok_unchanged(self) -> Result<Option<T>, UniversalIoError> {
+        match self {
+            Ok(t) => Ok(Some(t)),
+            Err(UniversalIoError::UnchangedOpen { .. }) => Ok(None),
+            Err(err) => Err(err),
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 pub enum UniversalIoError {
     #[error(transparent)]
@@ -81,6 +102,12 @@ pub enum UniversalIoError {
     #[error("path {path} not found")]
     NotFound { path: PathBuf },
 
+    #[error("Did not open a new instance of {path} because it didn't change since {since:?}")]
+    UnchangedOpen {
+        path: PathBuf,
+        since: Option<std::time::SystemTime>,
+    },
+
     #[error("elements range {start}..{end} is out of bounds, file contains {elements} elements")]
     OutOfBounds {
         start: u64,
@@ -131,6 +158,7 @@ impl UniversalIoError {
             | Self::QueueIsFull
             | Self::S3(_)
             | Self::S3Config { .. }
+            | Self::UnchangedOpen { .. }
             | Self::TaskPanicked(_) => false,
         }
     }

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

@@ -23,7 +23,7 @@ pub mod conformance;
 mod tests;
 
 pub use self::cached_fs::{CachedFs, CachedReadFsContext};
-pub use self::error::{IsNotFound, OkNotFound, UniversalIoError};
+pub use self::error::{IsNotFound, OkNotFound, OkUnchanged, UniversalIoError};
 #[cfg(target_os = "linux")]
 pub use self::io_uring::{IoUringFile, IoUringFs, IoUringOpenExtra, is_io_uring_supported};
 pub use self::mmap::{MmapFile, MmapFs};

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

@@ -170,6 +170,17 @@ pub trait CachedReadFs: UniversalReadFs {
         open_extra: Option<Self::OpenExtra>,
     ) -> UioResult<()>;
 
+    /// Schedule a prefetch for a file that has been opened already.
+    ///
+    /// This will force `Self::open` to return `UnchangedOpen` error if the file
+    /// did not change its `FileInfo` in between snapshots.
+    fn reschedule_prefetch(
+        &self,
+        path: &Path,
+        open_arguments: Option<OpenOptions>,
+        open_extra: Option<Self::OpenExtra>,
+    ) -> UioResult<()>;
+
     /// Return the file info from the current snapshot.
     fn cached_file_info(&self, path: &Path) -> Option<FileInfo>;
 }

lib/segment/src/common/operation_error.rs

@@ -229,6 +229,9 @@ impl From<UniversalIoError> for OperationError {
             UniversalIoError::Mmap(err) => Self::from(err),
 
             UniversalIoError::NotFound { path } => Self::FileNotFound { path },
+            UniversalIoError::UnchangedOpen { .. } => Self::Cancelled {
+                description: err.to_string(),
+            },
 
             UniversalIoError::Bincode(_)
             | UniversalIoError::BytemuckCast(_)
@@ -361,6 +364,7 @@ impl From<BlobstoreError> for OperationError {
                 | UniversalIoError::OutOfBounds { .. }
                 | UniversalIoError::InvalidFileIndex { .. }
                 | UniversalIoError::Uninitialized { .. }
+                | UniversalIoError::UnchangedOpen { .. }
                 | UniversalIoError::QueueIsFull
                 | UniversalIoError::AppendOffsetConflict { .. }
                 | UniversalIoError::S3(_)

src/tonic/api/storage_read_api/helpers.rs

@@ -262,6 +262,7 @@ pub fn io_error_to_status(e: UniversalIoError) -> Status {
         UniversalIoError::Uninitialized { description } => {
             Status::internal(format!("Uninitialized: {description}"))
         }
+        UniversalIoError::UnchangedOpen { .. } => Status::internal(e.to_string()),
         UniversalIoError::Bincode(e) => Status::internal(format!("Bincode error: {e}")),
         UniversalIoError::BytemuckCast(e) => Status::internal(format!("Bytemuck cast error: {e}")),
         UniversalIoError::ZerocopySize(e) => Status::internal(e),