fix(storage): address pending metadata and health gaps (#4380)

crates/ecstore/src/disk/local.rs

@@ -36,7 +36,7 @@ use crate::erasure::coding::bitrot_verify;
 use crate::runtime::sources as runtime_sources;
 use bytes::Bytes;
 use metrics::counter;
-use parking_lot::RwLock as ParkingLotRwLock;
+use parking_lot::{Mutex as ParkingLotMutex, RwLock as ParkingLotRwLock};
 use rustfs_filemeta::{
     Cache, FileInfo, FileInfoOpts, FileMeta, MetaCacheEntry, MetacacheWriter, ObjectPartInfo, Opts, RawFileInfo, UpdateFn,
     get_file_info, read_xl_meta_no_data,
@@ -1461,6 +1461,7 @@ pub struct LocalDisk {
     pub major: u64,
     pub minor: u64,
     pub nrrequests: u64,
+    scan_locks: Arc<ParkingLotMutex<HashSet<(String, String)>>>,
     // Performance optimization fields
     path_cache: Arc<ParkingLotRwLock<HashMap<String, PathBuf>>>,
     current_dir: Arc<OnceLock<PathBuf>>,
@@ -1698,6 +1699,7 @@ impl LocalDisk {
             minor: Default::default(),
             major: Default::default(),
             nrrequests: Default::default(),
+            scan_locks: Arc::new(ParkingLotMutex::new(HashSet::new())),
             // // format_legacy,
             // format_file_info: Mutex::new(format_meta),
             // format_data: Mutex::new(format_data),
@@ -2060,6 +2062,20 @@ impl LocalDisk {
         reject_local_disk_symlink_components(&self.root, path)
     }
 
+    fn try_acquire_scan_lock(&self, opts: &WalkDirOptions) -> Result<LocalScanLockGuard> {
+        let key = local_disk_scan_lock_key(&opts.bucket, &opts.base_dir, opts.filter_prefix.as_deref());
+        let mut scan_locks = self.scan_locks.lock();
+
+        if !scan_locks.insert(key.clone()) {
+            return Err(DiskError::DiskOngoingReq);
+        }
+
+        Ok(LocalScanLockGuard {
+            scan_locks: Arc::clone(&self.scan_locks),
+            key,
+        })
+    }
+
     // Batch path generation with single lock acquisition
     fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
         let mut results = Vec::with_capacity(requests.len());
@@ -3015,6 +3031,34 @@ impl Drop for ScanGuard {
     }
 }
 
+struct LocalScanLockGuard {
+    scan_locks: Arc<ParkingLotMutex<HashSet<(String, String)>>>,
+    key: (String, String),
+}
+
+impl Drop for LocalScanLockGuard {
+    fn drop(&mut self) {
+        self.scan_locks.lock().remove(&self.key);
+    }
+}
+
+fn local_disk_scan_lock_key(bucket: &str, base_dir: &str, filter_prefix: Option<&str>) -> (String, String) {
+    let mut prefix = base_dir.trim_matches('/').to_owned();
+    if let Some(filter_prefix) = filter_prefix
+        .map(|prefix| prefix.trim_matches('/'))
+        .filter(|prefix| !prefix.is_empty())
+    {
+        if prefix.is_empty() {
+            prefix.push_str(filter_prefix);
+        } else {
+            prefix.push_str(SLASH_SEPARATOR);
+            prefix.push_str(filter_prefix);
+        }
+    }
+
+    (bucket.to_owned(), prefix)
+}
+
 fn is_root_path(path: impl AsRef<Path>) -> bool {
     path.as_ref().components().count() == 1 && path.as_ref().has_root()
 }
@@ -3850,6 +3894,8 @@ impl DiskAPI for LocalDisk {
             return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
         }
 
+        let _scan_lock = self.try_acquire_scan_lock(&opts)?;
+
         let mut wr = wr;
 
         let mut out = MetacacheWriter::new(&mut wr);
@@ -4828,7 +4874,7 @@ mod test {
     use std::io;
     use std::pin::Pin;
     use std::task::{Context, Poll};
-    use tokio::io::{AsyncReadExt, ReadBuf};
+    use tokio::io::{AsyncReadExt, AsyncWrite, ReadBuf};
 
     fn test_file_info(name: &str, version_id: Uuid, data_dir: Option<Uuid>, data: Option<Bytes>) -> FileInfo {
         let size = data
@@ -4859,6 +4905,82 @@ mod test {
         }
     }
 
+    struct BlockingScanWriter {
+        entered_tx: Option<tokio::sync::oneshot::Sender<()>>,
+    }
+
+    impl AsyncWrite for BlockingScanWriter {
+        fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
+            if let Some(tx) = self.entered_tx.take() {
+                let _ = tx.send(());
+            }
+            Poll::Pending
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Pending
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Pending
+        }
+    }
+
+    #[tokio::test]
+    async fn test_local_disk_scan_rejects_concurrent_same_prefix_and_releases_on_cancel() {
+        use tempfile::tempdir;
+
+        let dir = tempdir().expect("temp dir should be created");
+        let bucket = "test-bucket";
+        let object_dir = dir.path().join(bucket).join("prefix/object");
+        fs::create_dir_all(&object_dir).await.expect("object dir should be created");
+        fs::write(object_dir.join(STORAGE_FORMAT_FILE), b"meta")
+            .await
+            .expect("object metadata should be written");
+
+        let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+        let opts = WalkDirOptions {
+            bucket: bucket.to_string(),
+            base_dir: "prefix/".to_string(),
+            recursive: true,
+            ..Default::default()
+        };
+
+        let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
+        let first_disk = Arc::clone(&disk);
+        let first_opts = opts.clone();
+        let mut blocking_writer = BlockingScanWriter {
+            entered_tx: Some(entered_tx),
+        };
+        let first_scan = tokio::spawn(async move { first_disk.walk_dir(first_opts, &mut blocking_writer).await });
+
+        entered_rx.await.expect("first scan should enter write path");
+
+        let mut second_writer = tokio::io::sink();
+        let second_scan = disk.walk_dir(opts.clone(), &mut second_writer).await;
+        assert!(
+            matches!(second_scan, Err(DiskError::DiskOngoingReq)),
+            "concurrent scan of same bucket and prefix must be rejected, got {second_scan:?}"
+        );
+
+        first_scan.abort();
+        assert!(
+            first_scan
+                .await
+                .expect_err("first scan task should be cancelled")
+                .is_cancelled(),
+            "aborting the blocked scan should cancel the task"
+        );
+
+        let mut after_cancel_writer = tokio::io::sink();
+        let after_cancel = disk.walk_dir(opts, &mut after_cancel_writer).await;
+        assert!(
+            after_cancel.is_ok(),
+            "cancelled scan must release the bucket/prefix lock, got {after_cancel:?}"
+        );
+    }
+
     #[tokio::test]
     async fn test_rename_data_writes_old_metadata_backup_before_non_inline_undo() {
         use tempfile::tempdir;