fix(ecstore): reject incomplete listing usage refreshes (#4698)

Co-authored-by: Henry Guo <marshawcoco@users.noreply.github.com>

crates/ecstore/src/cache_value/metacache_set.rs

@@ -24,9 +24,10 @@ use std::{
     io::ErrorKind,
     pin::Pin,
     sync::{Arc, OnceLock},
+    task::{Context, Poll},
     time::Duration,
 };
-use tokio::io::AsyncRead;
+use tokio::io::{AsyncRead, AsyncWrite};
 use tokio::spawn;
 use tokio::sync::Mutex as TokioMutex;
 use tokio::time::timeout;
@@ -102,6 +103,56 @@ fn duration_millis(duration: Duration) -> u64 {
     u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
 }
 
+fn classify_listing_quorum_failure(errors: &[DiskError]) -> DiskError {
+    if errors.contains(&DiskError::Timeout) {
+        return DiskError::Timeout;
+    }
+    if let Some(first) = errors.first()
+        && !matches!(first, DiskError::Io(_))
+        && errors.iter().all(|err| err == first)
+    {
+        return first.clone();
+    }
+    DiskError::ErasureReadQuorum
+}
+
+struct PublishedBytesWriter<W> {
+    inner: W,
+    published: bool,
+}
+
+impl<W> PublishedBytesWriter<W> {
+    fn new(inner: W) -> Self {
+        Self { inner, published: false }
+    }
+
+    fn has_published(&self) -> bool {
+        self.published
+    }
+}
+
+impl<W: AsyncWrite + Unpin> AsyncWrite for PublishedBytesWriter<W> {
+    fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<std::io::Result<usize>> {
+        match Pin::new(&mut self.inner).poll_write(cx, buf) {
+            Poll::Ready(Ok(written)) => {
+                if written != 0 && !self.published {
+                    self.published = true;
+                }
+                Poll::Ready(Ok(written))
+            }
+            other => other,
+        }
+    }
+
+    fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_flush(cx)
+    }
+
+    fn poll_shutdown(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+        Pin::new(&mut self.inner).poll_shutdown(cx)
+    }
+}
+
 #[cfg(test)]
 #[derive(Clone)]
 pub(crate) enum TestReaderBehavior {
@@ -223,12 +274,12 @@ async fn list_path_raw_inner(
         let (rd, wr) = tokio::io::duplex(64);
         readers.push(MetacacheReader::new(rd));
         jobs.push(spawn(async move {
+            let mut wr = PublishedBytesWriter::new(wr);
             #[cfg(test)]
             let test_primary_error = if let Some(behavior) = opts_clone.test_reader_behaviors.get(disk_idx).cloned() {
                 match behavior {
                     TestReaderBehavior::Eof => return Ok(()),
                     TestReaderBehavior::Entries(entries) => {
-                        let mut wr = wr;
                         let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
                         out.write(&entries).await.expect("test entries should be written");
                         out.close().await.expect("test entries should close");
@@ -250,20 +301,17 @@ async fn list_path_raw_inner(
                     }
                     TestReaderBehavior::PrimaryErrorThenFallback(err) => Some(err),
                     TestReaderBehavior::PartialThenTimeout(entries) => {
-                        let mut wr = wr;
                         let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
                         let err = DiskError::Timeout;
-                        record_producer_error(&producer_errs_clone, disk_idx, &err);
                         let _ = out.write(&entries).await;
                         drop(out);
-                        return Err(err);
+                        Some(err)
                     }
                 }
             } else {
                 None
             };
 
-            let mut wr = wr;
             let wakl_opts = WalkDirOptions {
                 bucket: opts_clone.bucket.clone(),
                 base_dir: opts_clone.path.clone(),
@@ -283,6 +331,10 @@ async fn list_path_raw_inner(
             let mut last_err = None;
             #[cfg(test)]
             if let Some(err) = test_primary_error {
+                if wr.has_published() {
+                    record_producer_error(&producer_errs_clone, disk_idx, &err);
+                    return Err(err);
+                }
                 last_err = Some(err);
                 need_fallback = true;
             }
@@ -333,6 +385,10 @@ async fn list_path_raw_inner(
                                 "Metacache walk_dir failed"
                             );
                         }
+                        if wr.has_published() {
+                            record_producer_error(&producer_errs_clone, disk_idx, &err);
+                            return Err(err);
+                        }
                         last_err = Some(err);
                         need_fallback = true;
                     }
@@ -371,12 +427,18 @@ async fn list_path_raw_inner(
                             last_err = Some(err);
                             continue;
                         }
-                        TestReaderBehavior::Stall
+                        TestReaderBehavior::Stall | TestReaderBehavior::IgnoreCancel => {
-                        | TestReaderBehavior::IgnoreCancel
-                        | TestReaderBehavior::PartialThenTimeout(_) => {
                             last_err = Some(DiskError::Timeout);
                             continue;
                         }
+                        TestReaderBehavior::PartialThenTimeout(entries) => {
+                            let mut out = rustfs_filemeta::MetacacheWriter::new(&mut wr);
+                            let err = DiskError::Timeout;
+                            let _ = out.write(&entries).await;
+                            drop(out);
+                            record_producer_error(&producer_errs_clone, disk_idx, &err);
+                            return Err(err);
+                        }
                     }
                 }
 
@@ -489,6 +551,10 @@ async fn list_path_raw_inner(
                                 "Metacache fallback walk_dir failed"
                             );
                         }
+                        if wr.has_published() {
+                            record_producer_error(&producer_errs_clone, disk_idx, &err);
+                            return Err(err);
+                        }
                         last_err = Some(err);
                     }
                 }
@@ -693,15 +759,6 @@ async fn list_path_raw_inner(
                 if let Some(finished_fn) = opts.finished.as_ref() {
                     finished_fn(&errs).await;
                 }
-                if errs.iter().flatten().any(|err| *err == DiskError::Timeout) {
-                    return Err(DiskError::Timeout);
-                }
-                let mut err_iter = errs.iter().flatten();
-                if let Some(err) = err_iter.next()
-                    && err_iter.next().is_none()
-                {
-                    return Err(err.clone());
-                }
                 let mut combined_err = Vec::new();
                 errs.iter().zip(opts.disks.iter()).for_each(|(err, disk)| match (err, disk) {
                     (Some(err), Some(disk)) => {
@@ -723,7 +780,8 @@ async fn list_path_raw_inner(
                     error = %combined_err.join(", "),
                     "Metacache listing quorum failed"
                 );
-                return Err(DiskError::other(combined_err.join(", ")));
+                let failures = errs.iter().flatten().cloned().collect::<Vec<_>>();
+                return Err(classify_listing_quorum_failure(&failures));
             }
 
             // Break if all at EOF or error.
@@ -872,7 +930,7 @@ async fn list_path_raw_inner(
     }
 
     if job_errs.len() > max_disk_failures {
-        return Err(job_errs.remove(0));
+        return Err(classify_listing_quorum_failure(&job_errs));
     }
 
     // warn!("list_path_raw: done");
@@ -923,6 +981,44 @@ mod tests {
         assert_eq!(err, DiskError::ErasureReadQuorum);
     }
 
+    #[tokio::test]
+    async fn list_path_raw_returns_typed_quorum_error_for_multiple_drive_failures() {
+        let err = list_path_raw(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![None, None, None],
+                min_disks: 2,
+                test_reader_behaviors: vec![
+                    TestReaderBehavior::ProducerError(DiskError::other("/sensitive/disk-a")),
+                    TestReaderBehavior::ProducerError(DiskError::other("http://internal-node/disk-b")),
+                    TestReaderBehavior::Eof,
+                ],
+                ..Default::default()
+            },
+        )
+        .await
+        .expect_err("multiple producer failures beyond tolerance should fail listing quorum");
+
+        assert_eq!(err, DiskError::ErasureReadQuorum);
+        assert!(!err.to_string().contains("sensitive"));
+        assert!(!err.to_string().contains("internal-node"));
+    }
+
+    #[test]
+    fn listing_quorum_failure_never_returns_raw_io_details() {
+        let err = classify_listing_quorum_failure(&[DiskError::other("/sensitive/disk-a")]);
+
+        assert_eq!(err, DiskError::ErasureReadQuorum);
+        assert!(!err.to_string().contains("sensitive"));
+
+        let access_denied = classify_listing_quorum_failure(&[DiskError::FileAccessDenied, DiskError::FileAccessDenied]);
+        assert_eq!(access_denied, DiskError::FileAccessDenied);
+
+        let mixed_timeout =
+            classify_listing_quorum_failure(&[DiskError::Timeout, DiskError::FileAccessDenied, DiskError::ErasureReadQuorum]);
+        assert_eq!(mixed_timeout, DiskError::Timeout);
+    }
+
     #[test]
     fn missing_path_error_classification_excludes_actionable_failures() {
         assert!(is_missing_path_error(&DiskError::FileNotFound));
@@ -1067,6 +1163,27 @@ mod tests {
         assert_eq!(err, DiskError::Timeout);
     }
 
+    #[tokio::test]
+    async fn list_path_raw_prefers_timeout_for_mixed_errors_beyond_quorum_budget() {
+        let err = list_path_raw(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![None, None, None],
+                min_disks: 2,
+                test_reader_behaviors: vec![
+                    TestReaderBehavior::ProducerError(DiskError::FileAccessDenied),
+                    TestReaderBehavior::ProducerError(DiskError::Timeout),
+                    TestReaderBehavior::Eof,
+                ],
+                ..Default::default()
+            },
+        )
+        .await
+        .expect_err("mixed failures beyond the tolerated budget should fail listing");
+
+        assert_eq!(err, DiskError::Timeout);
+    }
+
     #[tokio::test]
     async fn list_path_raw_tolerates_stalled_reader_after_quorum_eof() {
         list_path_raw(
@@ -1211,8 +1328,9 @@ mod tests {
     async fn list_path_raw_returns_timeout_when_producer_fails_after_partial_entry() {
         let seen = Arc::new(Mutex::new(Vec::new()));
         let seen_clone = seen.clone();
+        let claim_tracker = FallbackClaimTracker::default();
 
-        let err = list_path_raw(
+        let err = list_path_raw_with_claim_tracker(
             CancellationToken::new(),
             ListPathRawOptions {
                 disks: vec![None],
@@ -1223,6 +1341,7 @@ mod tests {
                     cached: None,
                     reusable: false,
                 }])],
+                test_fallback_reader_behaviors: vec![TestReaderBehavior::Entries(vec![fallback_test_entry()])],
                 agreed: Some(Box::new(move |entry: MetaCacheEntry| {
                     let seen = seen_clone.clone();
                     Box::pin(async move {
@@ -1231,12 +1350,59 @@ mod tests {
                 })),
                 ..Default::default()
             },
+            claim_tracker.clone(),
         )
         .await
         .expect_err("producer timeout after partial output must fail the listing");
 
         assert_eq!(err, DiskError::Timeout);
         assert_eq!(seen.lock().expect("seen mutex poisoned").as_slice(), &["bucket/object".to_string()]);
+        assert!(
+            claim_tracker.claimed_keys().await.is_empty(),
+            "fallback must not append a second metacache stream after partial output"
+        );
+    }
+
+    #[tokio::test]
+    async fn list_path_raw_does_not_try_second_fallback_after_partial_fallback_output() {
+        let seen = Arc::new(Mutex::new(Vec::new()));
+        let seen_clone = seen.clone();
+        let claim_tracker = FallbackClaimTracker::default();
+
+        let err = list_path_raw_with_claim_tracker(
+            CancellationToken::new(),
+            ListPathRawOptions {
+                disks: vec![None],
+                min_disks: 1,
+                test_reader_behaviors: vec![TestReaderBehavior::PrimaryErrorThenFallback(DiskError::DiskNotFound)],
+                test_fallback_reader_behaviors: vec![
+                    TestReaderBehavior::PartialThenTimeout(vec![MetaCacheEntry {
+                        name: "bucket/partial-fallback".to_string(),
+                        metadata: vec![1, 2, 3],
+                        cached: None,
+                        reusable: false,
+                    }]),
+                    TestReaderBehavior::Entries(vec![fallback_test_entry()]),
+                ],
+                agreed: Some(Box::new(move |entry: MetaCacheEntry| {
+                    let seen = seen_clone.clone();
+                    Box::pin(async move {
+                        seen.lock().expect("seen mutex poisoned").push(entry.name);
+                    })
+                })),
+                ..Default::default()
+            },
+            claim_tracker.clone(),
+        )
+        .await
+        .expect_err("a fallback timeout after partial output must fail the listing");
+
+        assert_eq!(err, DiskError::Timeout);
+        assert_eq!(
+            seen.lock().expect("seen mutex poisoned").as_slice(),
+            &["bucket/partial-fallback".to_string()]
+        );
+        assert_eq!(claim_tracker.claimed_keys().await.len(), 1, "a second fallback must not be claimed");
     }
 
     #[tokio::test]

crates/ecstore/src/cluster/rpc/http_auth.rs

@@ -69,6 +69,13 @@ fn signature_payload(url: &str, method: &Method, timestamp: i64) -> String {
     format!("{url}|{method}|{timestamp}")
 }
 
+fn redacted_rpc_path(url: &str) -> String {
+    url.parse::<Uri>()
+        .ok()
+        .map(|uri| uri.path().to_string())
+        .unwrap_or_else(|| "<invalid-rpc-url>".to_string())
+}
+
 /// Generate HMAC-SHA256 signature for the given data
 fn generate_signature(secret: &str, url: &str, method: &Method, timestamp: i64) -> String {
     let data = signature_payload(url, method, timestamp);
@@ -146,12 +153,13 @@ pub fn verify_rpc_signature(url: &str, method: &Method, headers: &HeaderMap) ->
     let secret = get_shared_secret()?;
 
     if !verify_signature(&secret, url, method, timestamp, signature) {
+        let rpc_path = redacted_rpc_path(url);
         error!(
-            "verify_rpc_signature: Invalid signature: url {}, method {}, timestamp {}, signature_len {}",
+            rpc_path = %rpc_path,
-            url,
+            method = %method,
-            method,
             timestamp,
-            signature.len()
+            signature_len = signature.len(),
+            "verify_rpc_signature: Invalid signature"
         );
 
         return Err(std::io::Error::other("Invalid signature"));
@@ -386,10 +394,31 @@ mod tests {
         assert!(!verify_signature(secret, url, &method, timestamp, "invalid-signature"));
     }
 
+    #[test]
+    fn walk_dir_capability_is_covered_by_the_signature() {
+        let secret = "test-secret";
+        let signed_url = concat!(
+            "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a&walk_dir_stream_completion=error-v1",
+            "&walk_dir_body_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+        );
+        let downgraded_url = "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a";
+        let tampered_body_digest = concat!(
+            "http://node1:9000/rustfs/rpc/walk_dir?disk=disk-a&walk_dir_stream_completion=error-v1",
+            "&walk_dir_body_sha256=aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"
+        );
+        let method = Method::GET;
+        let timestamp = 1_640_995_200;
+        let signature = generate_signature(secret, signed_url, &method, timestamp);
+
+        assert!(verify_signature(secret, signed_url, &method, timestamp, &signature));
+        assert!(!verify_signature(secret, downgraded_url, &method, timestamp, &signature));
+        assert!(!verify_signature(secret, tampered_body_digest, &method, timestamp, &signature));
+    }
+
     #[test]
     fn test_invalid_signature_log_contract_excludes_secrets() {
         ensure_test_rpc_secret();
-        let url = "http://example.com/api/test";
+        let url = "http://example.com/api/test?disk=/sensitive/path&token=private";
         let method = Method::GET;
         let timestamp = OffsetDateTime::now_utc().unix_timestamp();
         let secret = get_shared_secret().expect("test RPC secret should resolve");
@@ -417,6 +446,8 @@ mod tests {
         assert!(!captured.contains(&secret));
         assert!(!captured.contains(&expected_signature));
         assert!(!captured.contains(invalid_signature));
+        assert!(!captured.contains("sensitive"));
+        assert!(!captured.contains("private"));
     }
 
     #[test]

crates/ecstore/src/cluster/rpc/internode_data_transport.rs

@@ -15,6 +15,9 @@
 use crate::cluster::rpc::build_auth_headers;
 use crate::disk::error::{Error, Result};
 use crate::disk::{FileReader, FileWriter};
+use crate::storage_api_contracts::internode::{
+    WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
+};
 use async_trait::async_trait;
 use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE};
 use rustfs_config::{
@@ -22,6 +25,7 @@ use rustfs_config::{
     KNOWN_INTERNODE_DATA_TRANSPORT_BACKENDS,
 };
 use rustfs_rio::{HttpReader, HttpWriter};
+use sha2::{Digest, Sha256};
 use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 
@@ -180,7 +184,17 @@ fn build_put_file_stream_url(request: &WriteStreamRequest) -> String {
 }
 
 fn build_walk_dir_url(request: &WalkDirStreamRequest) -> String {
-    format!("{}{}?disk={}", request.endpoint, WALK_DIR_PATH, urlencoding::encode(&request.disk))
+    let body_sha256 = hex_simd::encode_to_string(Sha256::digest(&request.body), hex_simd::AsciiCase::Lower);
+    format!(
+        "{}{}?disk={}&{}={}&{}={}",
+        request.endpoint,
+        WALK_DIR_PATH,
+        urlencoding::encode(&request.disk),
+        WALK_DIR_STREAM_COMPLETION_QUERY,
+        WALK_DIR_STREAM_COMPLETION_V1,
+        WALK_DIR_BODY_SHA256_QUERY,
+        body_sha256
+    )
 }
 
 fn json_headers() -> HeaderMap {
@@ -300,7 +314,11 @@ mod tests {
 
         assert_eq!(
             url,
-            "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0"
+            concat!(
+                "http://node1:9000/rustfs/rpc/walk_dir?disk=http%3A%2F%2Fnode1%3A9000%2Fdata%2Frustfs0",
+                "&walk_dir_stream_completion=error-v1",
+                "&walk_dir_body_sha256=e3b0c44298fc1c149afbf4c8996fb92427ae41e4649b934ca495991b7852b855"
+            )
         );
     }
 

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -827,14 +827,13 @@ impl RemoteDisk {
             .expect("operation should succeed")
             .as_nanos() as i64;
         self.health.last_started.store(now, std::sync::atomic::Ordering::Relaxed);
-        self.health.increment_waiting();
+        let _waiting_guard = self.health.waiting_guard();
 
         if timeout_duration == Duration::ZERO {
             let operation_result = operation().await;
             if operation_result.is_ok() {
                 self.health.log_success();
             }
-            self.health.decrement_waiting();
             self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action)
                 .await;
             return operation_result;
@@ -845,18 +844,16 @@ impl RemoteDisk {
 
         match result {
             Ok(operation_result) => {
-                // Log success and decrement waiting counter
+                // Log success; the waiting guard balances every exit path.
                 if operation_result.is_ok() {
                     self.health.log_success();
                 }
-                self.health.decrement_waiting();
                 self.handle_network_like_error(op, timeout_duration, &operation_result, failure_health_action)
                     .await;
                 operation_result
             }
             Err(_) => {
                 // Timeout occurred, mark disk as potentially faulty
-                self.health.decrement_waiting();
                 counter!(
                     "rustfs_drive_op_timeout_total",
                     "endpoint" => self.endpoint.to_string(),
@@ -3314,6 +3311,33 @@ mod tests {
             .expect("operation should succeed")
     }
 
+    #[tokio::test]
+    async fn remote_disk_health_wrapper_balances_task_cancellation() {
+        let disk = Arc::new(new_remote_disk_with_transport(Arc::new(RecordingInternodeDataTransport::default())).await);
+        let task_disk = Arc::clone(&disk);
+        let task = tokio::spawn(async move {
+            task_disk
+                .execute_with_timeout_for_op(
+                    "cancellation-test",
+                    || async { std::future::pending::<Result<()>>().await },
+                    Duration::ZERO,
+                )
+                .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while disk.health.waiting_count() == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("operation should enter remote disk health tracking");
+        task.abort();
+        let _ = task.await;
+
+        assert_eq!(disk.health.waiting_count(), 0);
+    }
+
     #[derive(Debug)]
     struct PartialThenErrorReader {
         cursor: Cursor<Vec<u8>>,

crates/ecstore/src/data_usage/mod.rs

@@ -19,7 +19,7 @@ pub mod local_snapshot;
 
 use crate::storage_api_contracts::{
     bucket::{BucketOperations as _, BucketOptions},
-    list::ListOperations as _,
+    list::{ListOperations as _, StorageListObjectVersionsInfo},
     object::ObjectIO as _,
 };
 use crate::{
@@ -27,8 +27,9 @@ use crate::{
     config::com::read_config,
     disk::DiskAPI,
     error::{Error, classify_system_path_failure_reason},
+    object_api::ObjectInfo,
     runtime::sources as runtime_sources,
-    store::ECStore,
+    store::{ECStore, list_objects::list_marker_key},
 };
 pub use local_snapshot::{LocalUsageSnapshot, read_snapshot as read_local_snapshot, snapshot_path};
 use rustfs_data_usage::{
@@ -39,6 +40,7 @@ use rustfs_io_metrics::record_system_path_failure;
 use rustfs_utils::path::SLASH_SEPARATOR;
 use std::{
     collections::{HashMap, HashSet, hash_map::Entry},
+    future::Future,
     sync::{Arc, LazyLock, OnceLock},
     time::{Duration, SystemTime},
 };
@@ -53,6 +55,7 @@ const DATA_COMPRESSION_TOTAL_NAME: &str = ".compression.json";
 const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin";
 pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin";
 const DATA_USAGE_CACHE_TTL_SECS: u64 = 30;
+const LIVE_BUCKET_USAGE_MAX_ENTRIES: u64 = 1024;
 
 #[derive(Debug, Clone)]
 struct CachedBucketUsage {
@@ -67,9 +70,11 @@ struct CachedBucketUsage {
 
 type UsageMemoryCache = Arc<RwLock<HashMap<String, CachedBucketUsage>>>;
 type CacheUpdating = Arc<RwLock<bool>>;
+type LiveBucketUsageCache = moka::future::Cache<String, BucketUsageInfo>;
 
 static USAGE_MEMORY_CACHE: OnceLock<UsageMemoryCache> = OnceLock::new();
 static USAGE_CACHE_UPDATING: OnceLock<CacheUpdating> = OnceLock::new();
+static LIVE_BUCKET_USAGE_CACHE: OnceLock<LiveBucketUsageCache> = OnceLock::new();
 
 /// Deferred persist thresholds for compression totals: persist after this many
 /// operations recorded, but no more often than the min interval.
@@ -110,6 +115,14 @@ fn cache_updating() -> &'static CacheUpdating {
     USAGE_CACHE_UPDATING.get_or_init(|| Arc::new(RwLock::new(false)))
 }
 
+fn live_bucket_usage_cache() -> &'static LiveBucketUsageCache {
+    LIVE_BUCKET_USAGE_CACHE.get_or_init(|| {
+        moka::future::Cache::builder()
+            .max_capacity(LIVE_BUCKET_USAGE_MAX_ENTRIES)
+            .build()
+    })
+}
+
 // Data usage storage paths
 lazy_static::lazy_static! {
     pub static ref DATA_USAGE_BUCKET: String = format!("{}{}{}",
@@ -208,6 +221,7 @@ async fn clear_bucket_usage_memory(bucket: &str) {
 }
 
 pub async fn remove_bucket_usage_from_backend(store: Arc<ECStore>, bucket: &str) -> Result<(), Error> {
+    live_bucket_usage_cache().invalidate(bucket).await;
     clear_bucket_usage_memory(bucket).await;
 
     let data_usage_info = load_data_usage_from_backend(store.clone()).await?;
@@ -442,79 +456,203 @@ pub async fn aggregate_local_snapshots(store: Arc<ECStore>) -> Result<(Vec<DiskU
 }
 
 /// Calculate accurate bucket usage statistics by enumerating objects through the object layer.
+#[derive(Default)]
+struct BucketUsageAccumulator {
+    current_object_name: Option<String>,
+    // FileMeta caps versions per object, so replay detection remains bounded.
+    current_object_versions: HashSet<Option<[u8; 16]>>,
+    current_live_versions: u64,
+    objects_count: u64,
+    versions_count: u64,
+    total_size: u64,
+    delete_markers: u64,
+    size_histogram: SizeHistogram,
+    versions_histogram: VersionsHistogram,
+}
+
+impl BucketUsageAccumulator {
+    fn record(&mut self, bucket: &str, object: &ObjectInfo) -> Result<(), Error> {
+        if object.is_dir {
+            return Ok(());
+        }
+
+        if self
+            .current_object_name
+            .as_deref()
+            .is_some_and(|current_name| object.name.as_str() != current_name)
+        {
+            self.finish_current_object();
+        }
+
+        record_version_listing_entry(
+            bucket,
+            &mut self.current_object_name,
+            &mut self.current_object_versions,
+            &object.name,
+            object.version_id.as_ref().map(|version_id| version_id.as_bytes()),
+        )?;
+
+        if object.delete_marker {
+            self.delete_markers = self.delete_markers.saturating_add(1);
+            return Ok(());
+        }
+
+        let object_size = object.size.max(0) as u64;
+        self.current_live_versions = self.current_live_versions.saturating_add(1);
+        self.size_histogram.add(object_size);
+        self.total_size = self.total_size.saturating_add(object_size);
+        self.versions_count = self.versions_count.saturating_add(1);
+        Ok(())
+    }
+
+    fn finish_current_object(&mut self) {
+        if self.current_live_versions > 0 {
+            self.objects_count = self.objects_count.saturating_add(1);
+            self.versions_histogram.add(self.current_live_versions);
+        }
+        self.current_live_versions = 0;
+    }
+
+    fn finish(mut self) -> BucketUsageInfo {
+        self.finish_current_object();
+        BucketUsageInfo {
+            size: self.total_size,
+            objects_count: self.objects_count,
+            versions_count: self.versions_count,
+            delete_markers_count: self.delete_markers,
+            object_size_histogram: self.size_histogram.to_map(),
+            object_versions_histogram: self.versions_histogram.to_map(),
+            ..Default::default()
+        }
+    }
+}
+
+type UsageVersionPage = StorageListObjectVersionsInfo<ObjectInfo>;
+
 pub async fn compute_bucket_usage(store: Arc<ECStore>, bucket_name: &str) -> Result<BucketUsageInfo, Error> {
+    let bucket = bucket_name.to_string();
+    compute_bucket_usage_with_pages(bucket_name, move |marker, version_marker| {
+        let store = Arc::clone(&store);
+        let bucket = bucket.clone();
+        async move {
+            store
+                .list_object_versions(&bucket, "", marker, version_marker, None, 1000)
+                .await
+        }
+    })
+    .await
+}
+
+async fn compute_bucket_usage_with_pages<F, Fut>(bucket_name: &str, mut fetch_page: F) -> Result<BucketUsageInfo, Error>
+where
+    F: FnMut(Option<String>, Option<String>) -> Fut,
+    Fut: Future<Output = Result<UsageVersionPage, Error>>,
+{
     let mut marker: Option<String> = None;
     let mut version_marker: Option<String> = None;
-    let mut object_names: HashSet<String> = HashSet::new();
+    let mut usage = BucketUsageAccumulator::default();
-    let mut object_versions: HashMap<String, u64> = HashMap::new();
-    let mut versions_count: u64 = 0;
-    let mut total_size: u64 = 0;
-    let mut delete_markers: u64 = 0;
-    let mut size_histogram = SizeHistogram::default();
 
     loop {
-        let result = store
+        let result = fetch_page(marker.clone(), version_marker.clone()).await?;
-            .clone()
-            .list_object_versions(
-                bucket_name,
-                "", // prefix
-                marker.clone(),
-                version_marker.clone(),
-                None, // delimiter
-                1000, // max_keys
-            )
-            .await?;
 
+        let page_entries = result.objects.len();
         for object in result.objects.iter() {
-            if object.is_dir {
+            usage.record(bucket_name, object)?;
-                continue;
-            }
-
-            if object.delete_marker {
-                delete_markers = delete_markers.saturating_add(1);
-                continue;
-            }
-
-            let object_size = object.size.max(0) as u64;
-            object_names.insert(object.name.clone());
-            *object_versions.entry(object.name.clone()).or_insert(0) += 1;
-            size_histogram.add(object_size);
-            total_size = total_size.saturating_add(object_size);
-            versions_count = versions_count.saturating_add(1);
         }
 
         if !result.is_truncated {
             break;
         }
+        ensure_truncated_version_page_has_entries(bucket_name, page_entries)?;
 
-        marker = result.next_marker.clone();
+        advance_version_listing_cursor(
-        version_marker = result.next_version_idmarker.clone();
+            bucket_name,
-        if marker.is_none() {
+            &mut marker,
-            info!(
+            &mut version_marker,
-                "Bucket {} version listing marked truncated but no marker returned; stopping early",
+            result.next_marker,
-                bucket_name
+            result.next_version_idmarker,
-            );
+        )?;
-            break;
+    }
+
+    Ok(usage.finish())
+}
+
+fn ensure_truncated_version_page_has_entries(bucket: &str, page_entries: usize) -> Result<(), Error> {
+    if page_entries == 0 {
+        return Err(Error::other(format!("bucket {bucket} version listing returned an empty truncated page")));
+    }
+    Ok(())
+}
+
+fn record_version_listing_entry(
+    bucket: &str,
+    current_object_name: &mut Option<String>,
+    current_object_versions: &mut HashSet<Option<[u8; 16]>>,
+    object_name: &str,
+    version_id: Option<&[u8; 16]>,
+) -> Result<(), Error> {
+    match current_object_name.as_deref() {
+        Some(current_name) if object_name < current_name => {
+            return Err(Error::other(format!("bucket {bucket} version listing returned an out-of-order object")));
         }
+        Some(current_name) if object_name > current_name => {
+            current_object_versions.clear();
+            *current_object_name = Some(object_name.to_string());
+        }
+        None => *current_object_name = Some(object_name.to_string()),
+        Some(_) => {}
     }
 
-    let objects_count = object_names.len() as u64;
+    if !current_object_versions.insert(version_id.copied()) {
-    let mut versions_histogram = VersionsHistogram::default();
+        return Err(Error::other(format!(
-    for version_count in object_versions.values() {
+            "bucket {bucket} version listing returned a repeated object version"
-        versions_histogram.add(*version_count);
+        )));
     }
+    Ok(())
+}
 
-    let usage = BucketUsageInfo {
+fn advance_version_listing_cursor(
-        size: total_size,
+    bucket: &str,
-        objects_count,
+    marker: &mut Option<String>,
-        versions_count,
+    version_marker: &mut Option<String>,
-        delete_markers_count: delete_markers,
+    next_marker: Option<String>,
-        object_size_histogram: size_histogram.to_map(),
+    next_version_marker: Option<String>,
-        object_versions_histogram: versions_histogram.to_map(),
+) -> Result<(), Error> {
-        ..Default::default()
+    let next_marker = next_marker
-    };
+        .filter(|next_marker| !next_marker.is_empty())
+        .ok_or_else(|| Error::other(format!("bucket {bucket} version listing was truncated without a key marker")))?;
+    let next_version_marker = next_version_marker
+        .filter(|next_version_marker| !next_version_marker.is_empty())
+        .ok_or_else(|| Error::other(format!("bucket {bucket} version listing was truncated without a version marker")))?;
+    let current_key_marker = marker.as_deref().map(list_marker_key);
+    let next_key_marker = list_marker_key(&next_marker);
+    if current_key_marker == Some(next_key_marker) && version_marker.as_deref() == Some(next_version_marker.as_str()) {
+        return Err(Error::other(format!(
+            "bucket {bucket} version listing returned a repeated continuation marker"
+        )));
+    }
+    if current_key_marker.is_some_and(|marker| next_key_marker < marker) {
+        return Err(Error::other(format!("bucket {bucket} version listing returned a regressing key marker")));
+    }
+    *marker = Some(next_marker);
+    *version_marker = Some(next_version_marker);
+    Ok(())
+}
 
-    Ok(usage)
+async fn coalesce_live_bucket_usage<F>(bucket: String, init: F) -> Result<BucketUsageInfo, Error>
+where
+    F: Future<Output = Result<BucketUsageInfo, Error>> + Send + 'static,
+{
+    let result = live_bucket_usage_cache().try_get_with(bucket.clone(), init).await;
+    live_bucket_usage_cache().invalidate(&bucket).await;
+    result.map_err(|err| Error::other(err.to_string()))
+}
+
+fn apply_live_bucket_usage_to_response(data_usage_info: &mut DataUsageInfo, bucket: &str, usage: &BucketUsageInfo) {
+    data_usage_info.bucket_sizes.insert(bucket.to_string(), usage.size);
+    data_usage_info.buckets_usage.insert(bucket.to_string(), usage.clone());
+    set_buckets_count_from_usage(data_usage_info);
+    data_usage_info.calculate_totals();
 }
 
 pub async fn refresh_bucket_usage_from_object_layer(
@@ -522,13 +660,12 @@ pub async fn refresh_bucket_usage_from_object_layer(
     data_usage_info: &mut DataUsageInfo,
     bucket: &str,
 ) -> Result<BucketUsageInfo, Error> {
-    let refresh_started_at = SystemTime::now();
+    let bucket_name = bucket.to_string();
-    let usage = compute_bucket_usage(store, bucket).await?;
+    let usage =
-    replace_bucket_usage_memory_from_authoritative(bucket, usage.clone(), refresh_started_at).await;
+        coalesce_live_bucket_usage(bucket_name.clone(), async move { compute_bucket_usage(store, &bucket_name).await }).await?;
-    data_usage_info.bucket_sizes.insert(bucket.to_string(), usage.size);
+    // Request-time listings are not linearizable with writes on other nodes.
-    data_usage_info.buckets_usage.insert(bucket.to_string(), usage.clone());
+    // Keep the live result response-local instead of promoting it into the quota cache.
-    set_buckets_count_from_usage(data_usage_info);
+    apply_live_bucket_usage_to_response(data_usage_info, bucket, &usage);
-    data_usage_info.calculate_totals();
     Ok(usage)
 }
 
@@ -546,45 +683,26 @@ pub async fn refresh_versioned_bucket_usage_from_object_layer(store: Arc<ECStore
             Vec::new()
         }
     };
-    let buckets = bucket_names_for_versioned_refresh(data_usage_info, listed_bucket_names);
+    let mut buckets = data_usage_info.buckets_usage.keys().cloned().collect::<HashSet<String>>();
-    let mut changed = false;
+    buckets.extend(listed_bucket_names.into_iter().filter(|bucket| !bucket.is_empty()));
+    let mut buckets = buckets.into_iter().collect::<Vec<_>>();
+    buckets.sort();
 
     for bucket in buckets {
         let Ok(versioning) = BucketVersioningSys::get(&bucket).await else {
             continue;
         };
-
         if !versioning.enabled() && !versioning.suspended() {
             continue;
         }
-
         if let Err(err) = refresh_bucket_usage_from_object_layer(store.clone(), data_usage_info, &bucket).await {
             debug!(
                 bucket = %bucket,
                 error = %err,
                 "failed to refresh versioned bucket usage from object layer"
             );
-            continue;
         }
-        changed = true;
     }
-
-    if changed {
-        set_buckets_count_from_usage(data_usage_info);
-        data_usage_info.calculate_totals();
-    }
-}
-
-fn bucket_names_for_versioned_refresh(
-    data_usage_info: &DataUsageInfo,
-    listed_bucket_names: impl IntoIterator<Item = String>,
-) -> Vec<String> {
-    let mut buckets = data_usage_info.buckets_usage.keys().cloned().collect::<HashSet<String>>();
-    buckets.extend(listed_bucket_names.into_iter().filter(|bucket| !bucket.is_empty()));
-
-    let mut buckets = buckets.into_iter().collect::<Vec<_>>();
-    buckets.sort();
-    buckets
 }
 
 async fn ensure_bucket_usage_cached(bucket: &str) {
@@ -629,6 +747,7 @@ fn bucket_usage_counts_match(left: &BucketUsageInfo, right: &BucketUsageInfo) ->
         && left.delete_markers_count == right.delete_markers_count
 }
 
+#[cfg(test)]
 async fn replace_bucket_usage_memory_from_authoritative(bucket: &str, usage: BucketUsageInfo, refresh_started_at: SystemTime) {
     let mut cache = memory_cache().write().await;
     if let Some(existing) = cache.get(bucket)
@@ -1253,6 +1372,216 @@ mod tests {
         info
     }
 
+    #[tokio::test]
+    async fn compute_usage_preserves_same_object_across_1000_entry_page_boundary() {
+        let first_page = UsageVersionPage {
+            is_truncated: true,
+            next_marker: Some("object-a".to_string()),
+            next_version_idmarker: Some(uuid::Uuid::from_u128(1000).to_string()),
+            objects: (1..=1000_u128)
+                .map(|version| ObjectInfo {
+                    name: "object-a".to_string(),
+                    size: 1,
+                    version_id: Some(uuid::Uuid::from_u128(version)),
+                    ..Default::default()
+                })
+                .collect(),
+            ..Default::default()
+        };
+        let second_page = UsageVersionPage {
+            objects: vec![
+                ObjectInfo {
+                    name: "object-a".to_string(),
+                    size: 1,
+                    version_id: Some(uuid::Uuid::from_u128(1001)),
+                    ..Default::default()
+                },
+                ObjectInfo {
+                    name: "object-a".to_string(),
+                    version_id: Some(uuid::Uuid::from_u128(1002)),
+                    delete_marker: true,
+                    ..Default::default()
+                },
+                ObjectInfo {
+                    name: "object-b".to_string(),
+                    size: 2,
+                    version_id: Some(uuid::Uuid::from_u128(1003)),
+                    ..Default::default()
+                },
+            ],
+            ..Default::default()
+        };
+        let pages = Arc::new(std::sync::Mutex::new(std::collections::VecDeque::from([first_page, second_page])));
+        let fetch_pages = Arc::clone(&pages);
+
+        let usage = compute_bucket_usage_with_pages("bucket-a", move |marker, version_marker| {
+            let page = fetch_pages
+                .lock()
+                .expect("page queue lock should not be poisoned")
+                .pop_front()
+                .expect("pagination must not request an unexpected page");
+            if page.is_truncated {
+                assert_eq!((marker, version_marker), (None, None));
+            } else {
+                let expected_version_marker = uuid::Uuid::from_u128(1000).to_string();
+                assert_eq!(marker.as_deref(), Some("object-a"));
+                assert_eq!(version_marker.as_deref(), Some(expected_version_marker.as_str()));
+            }
+            async move { Ok(page) }
+        })
+        .await
+        .expect("two-page usage aggregation should succeed");
+
+        assert!(pages.lock().expect("page queue lock should not be poisoned").is_empty());
+        assert_eq!(usage.objects_count, 2);
+        assert_eq!(usage.versions_count, 1002);
+        assert_eq!(usage.delete_markers_count, 1);
+        assert_eq!(usage.size, 1003);
+        assert_eq!(usage.object_versions_histogram.get("SINGLE_VERSION"), Some(&1));
+        assert_eq!(usage.object_versions_histogram.get("BETWEEN_1000_AND_10000"), Some(&1));
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn live_bucket_usage_refreshes_are_coalesced_only_while_in_flight() {
+        const BUCKET: &str = "coalesced-live-usage-test";
+        live_bucket_usage_cache().invalidate(BUCKET).await;
+        let calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let mut tasks = Vec::new();
+
+        for _ in 0..8 {
+            let calls = Arc::clone(&calls);
+            tasks.push(tokio::spawn(coalesce_live_bucket_usage(BUCKET.to_string(), async move {
+                calls.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                tokio::time::sleep(Duration::from_millis(25)).await;
+                Ok(BucketUsageInfo {
+                    objects_count: 7,
+                    size: 42,
+                    ..Default::default()
+                })
+            })));
+        }
+
+        for task in tasks {
+            let usage = task
+                .await
+                .expect("coalesced refresh task should not panic")
+                .expect("coalesced refresh should succeed");
+            assert_eq!((usage.objects_count, usage.size), (7, 42));
+        }
+        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 1);
+
+        let calls_after_batch = Arc::clone(&calls);
+        coalesce_live_bucket_usage(BUCKET.to_string(), async move {
+            calls_after_batch.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            Ok(BucketUsageInfo::default())
+        })
+        .await
+        .expect("a later refresh should run after the in-flight entry is removed");
+        assert_eq!(calls.load(std::sync::atomic::Ordering::SeqCst), 2);
+        live_bucket_usage_cache().invalidate(BUCKET).await;
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn live_usage_updates_response_without_replacing_quota_memory() {
+        clear_usage_memory_cache_for_test().await;
+        let persisted = data_usage_info_for_test("bucket-a", 100, 1_000, SystemTime::now());
+        replace_bucket_usage_memory_from_info(&persisted).await;
+        let mut response = persisted.clone();
+
+        apply_live_bucket_usage_to_response(&mut response, "bucket-a", &BucketUsageInfo::default());
+
+        assert_eq!(response.buckets_usage.get("bucket-a").map(|usage| usage.objects_count), Some(0));
+        assert_eq!(get_bucket_usage_memory("bucket-a").await, Some(1_000));
+    }
+
+    #[test]
+    fn version_listing_cursor_rejects_incomplete_or_non_advancing_pages() {
+        let mut marker = Some("object-a".to_string());
+        let mut version_marker = Some("version-a".to_string());
+
+        assert!(
+            advance_version_listing_cursor("bucket-a", &mut marker, &mut version_marker, None, Some("version-b".to_string()),)
+                .is_err()
+        );
+        assert!(
+            advance_version_listing_cursor("bucket-a", &mut marker, &mut version_marker, Some("object-a".to_string()), None,)
+                .is_err()
+        );
+        assert!(
+            advance_version_listing_cursor(
+                "bucket-a",
+                &mut marker,
+                &mut version_marker,
+                Some("object-a".to_string()),
+                Some("version-a".to_string()),
+            )
+            .is_err()
+        );
+        assert!(ensure_truncated_version_page_has_entries("bucket-a", 0).is_err());
+        ensure_truncated_version_page_has_entries("bucket-a", 1).expect("a truncated page with an entry can advance");
+
+        let mut tagged_marker = Some("object-a[rustfs_cache:v2,id:old]".to_string());
+        let mut tagged_version_marker = Some("version-a".to_string());
+        assert!(
+            advance_version_listing_cursor(
+                "bucket-a",
+                &mut tagged_marker,
+                &mut tagged_version_marker,
+                Some("object-a[rustfs_cache:v2,id:new]".to_string()),
+                Some("version-a".to_string()),
+            )
+            .is_err(),
+            "different cache tags for the same logical marker must not count as progress"
+        );
+    }
+
+    #[test]
+    fn version_listing_rejects_replayed_and_out_of_order_entries() {
+        let version_a = uuid::Uuid::from_u128(1);
+        let version_b = uuid::Uuid::from_u128(2);
+        let mut current_object = None;
+        let mut current_versions = HashSet::new();
+
+        record_version_listing_entry(
+            "bucket-a",
+            &mut current_object,
+            &mut current_versions,
+            "object-b",
+            Some(version_a.as_bytes()),
+        )
+        .expect("first version should be accepted");
+        assert!(
+            record_version_listing_entry(
+                "bucket-a",
+                &mut current_object,
+                &mut current_versions,
+                "object-b",
+                Some(version_a.as_bytes()),
+            )
+            .is_err()
+        );
+        record_version_listing_entry(
+            "bucket-a",
+            &mut current_object,
+            &mut current_versions,
+            "object-c",
+            Some(version_b.as_bytes()),
+        )
+        .expect("a lexicographically later object should reset version history");
+        assert!(
+            record_version_listing_entry(
+                "bucket-a",
+                &mut current_object,
+                &mut current_versions,
+                "object-a",
+                Some(version_a.as_bytes()),
+            )
+            .is_err()
+        );
+    }
+
     fn aggregate_for_test(
         inputs: Vec<(DiskUsageStatus, Result<Option<LocalUsageSnapshot>, Error>)>,
     ) -> (Vec<DiskUsageStatus>, DataUsageInfo) {
@@ -1574,24 +1903,6 @@ mod tests {
         );
     }
 
-    #[test]
-    fn versioned_refresh_bucket_names_include_live_buckets_without_usage_snapshot() {
-        let mut persisted = DataUsageInfo::default();
-        persisted.buckets_usage.insert(
-            "bucket-a".to_string(),
-            BucketUsageInfo {
-                objects_count: 1,
-                versions_count: 1,
-                size: 10,
-                ..Default::default()
-            },
-        );
-
-        let buckets = bucket_names_for_versioned_refresh(&persisted, vec!["bucket-b".to_string(), "bucket-a".to_string()]);
-
-        assert_eq!(buckets, vec!["bucket-a".to_string(), "bucket-b".to_string()]);
-    }
-
     #[tokio::test]
     #[serial]
     async fn authoritative_versioned_refresh_replaces_stale_dirty_memory() {

crates/ecstore/src/disk/disk_store.rs

@@ -283,6 +283,16 @@ pub struct DiskHealthTracker {
     pub last_capacity_probe_unix_secs: AtomicI64,
 }
 
+pub(crate) struct DiskHealthWaitingGuard<'a> {
+    health: &'a DiskHealthTracker,
+}
+
+impl Drop for DiskHealthWaitingGuard<'_> {
+    fn drop(&mut self) {
+        self.health.decrement_waiting();
+    }
+}
+
 impl DiskHealthTracker {
     /// Create a new disk health tracker
     pub fn new() -> Self {
@@ -537,6 +547,11 @@ impl DiskHealthTracker {
         self.waiting.fetch_add(1, Ordering::Relaxed);
     }
 
+    pub(crate) fn waiting_guard(&self) -> DiskHealthWaitingGuard<'_> {
+        self.increment_waiting();
+        DiskHealthWaitingGuard { health: self }
+    }
+
     /// Decrement waiting operations counter
     pub fn decrement_waiting(&self) {
         self.waiting.fetch_sub(1, Ordering::Relaxed);
@@ -1038,11 +1053,10 @@ impl LocalDiskWrapper {
             .unwrap()
             .as_nanos() as i64;
         self.health.last_started.store(now, Ordering::Relaxed);
-        self.health.increment_waiting();
+        let _waiting_guard = self.health.waiting_guard();
 
         if timeout_duration == Duration::ZERO {
             let result = operation().await;
-            self.health.decrement_waiting();
             if result.is_ok() {
                 self.health.record_operation_success(&self.endpoint(), "operation_success");
             }
@@ -1053,16 +1067,14 @@ impl LocalDiskWrapper {
 
         match result {
             Ok(operation_result) => {
-                // Log success and decrement waiting counter
+                // Log success; the waiting guard balances every exit path.
                 if operation_result.is_ok() {
                     self.health.record_operation_success(&self.endpoint(), "operation_success");
                 }
-                self.health.decrement_waiting();
                 operation_result
             }
             Err(_) => {
-                // Timeout occurred, mark disk as potentially faulty and decrement waiting counter
+                // Timeout occurred, mark disk as potentially faulty.
-                self.health.decrement_waiting();
                 if timeout_health_action == TimeoutHealthAction::MarkFailure
                     && self.health.mark_failure(&self.endpoint(), "operation_timeout")
                 {
@@ -1468,6 +1480,43 @@ mod tests {
 
     struct PendingWriter;
 
+    #[test]
+    fn disk_health_waiting_guard_balances_cancellation() {
+        let health = DiskHealthTracker::new();
+        {
+            let _guard = health.waiting_guard();
+            assert_eq!(health.waiting_count(), 1);
+        }
+        assert_eq!(health.waiting_count(), 0);
+    }
+
+    #[tokio::test]
+    async fn local_disk_health_wrapper_balances_task_cancellation() {
+        let dir = tempfile::tempdir().expect("temp dir should be created");
+        let endpoint =
+            Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8")).expect("endpoint should parse");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+        let wrapper = Arc::new(LocalDiskWrapper::new(disk, false));
+        let task_wrapper = Arc::clone(&wrapper);
+        let task = tokio::spawn(async move {
+            task_wrapper
+                .track_disk_health(|| async { std::future::pending::<Result<()>>().await }, Duration::ZERO)
+                .await
+        });
+
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while wrapper.health.waiting_count() == 0 {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("operation should enter disk health tracking");
+        task.abort();
+        let _ = task.await;
+
+        assert_eq!(wrapper.health.waiting_count(), 0);
+    }
+
     impl AsyncWrite for PendingWriter {
         fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
             Poll::Pending

crates/ecstore/src/object_api/types.rs

@@ -892,6 +892,25 @@ mod tests {
         assert_eq!(versions[1].version_id, Some(last_version));
     }
 
+    #[test]
+    fn versions_after_marker_preserves_stale_marker_compatibility() {
+        let existing_version =
+            Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("existing version UUID should parse");
+        let deleted_marker = Uuid::parse_str("aaaaaaaa-bbbb-cccc-dddd-eeeeeeeeeeee").expect("delete marker UUID should parse");
+        let file_infos = rustfs_filemeta::FileInfoVersions {
+            versions: vec![FileInfo {
+                version_id: Some(existing_version),
+                ..Default::default()
+            }],
+            ..Default::default()
+        };
+
+        let versions = versions_after_marker(&file_infos, VersionMarker::Version(deleted_marker));
+
+        assert_eq!(versions.len(), 1);
+        assert_eq!(versions[0].version_id, Some(existing_version));
+    }
+
     #[tokio::test]
     async fn versions_listing_applies_version_marker_only_to_first_entry() {
         let metadata = rustfs_filemeta::test_data::create_real_xlmeta().expect("test metadata should be valid");

crates/ecstore/src/storage_api_contracts/mod.rs

@@ -21,6 +21,12 @@ pub(crate) mod heal {
     pub(crate) use rustfs_storage_api::HealOperations;
 }
 
+pub(crate) mod internode {
+    pub(crate) use rustfs_storage_api::{
+        WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_STREAM_COMPLETION_QUERY, WALK_DIR_STREAM_COMPLETION_V1,
+    };
+}
+
 pub(crate) mod lifecycle {
     pub use rustfs_storage_api::{ExpirationOptions, TransitionedObject};
 }

crates/ecstore/src/store/list_objects.rs

@@ -271,6 +271,7 @@ pub struct ListPathOptions {
 
 const MARKER_TAG_VERSION: &str = "v2";
 const LEGACY_MARKER_TAG_VERSIONS: &[&str] = &["v1", MARKER_TAG_VERSION];
+const LIST_CACHE_MARKER_PREFIX: &str = "[rustfs_cache:";
 const LIST_CURSOR_SOURCE_WALKER: &str = "walker";
 const LIST_CURSOR_SOURCE_INDEX_KEY_ONLY: &str = "index_key_only";
 const LIST_CURSOR_SOURCE_INDEX_VERIFIED_PAGE: &str = "index_verified_page";
@@ -1831,7 +1832,7 @@ impl ListContinuationV2 {
     fn encode_marker(&self, marker: &str) -> String {
         let mut marker_tag = String::with_capacity(marker.len() + 64);
         marker_tag.push_str(marker);
-        marker_tag.push_str("[rustfs_cache:");
+        marker_tag.push_str(LIST_CACHE_MARKER_PREFIX);
         marker_tag.push_str(self.version);
         match &self.id {
             Some(id) => {
@@ -1928,6 +1929,13 @@ impl ListContinuationV2 {
     }
 }
 
+pub(crate) fn list_marker_key(marker: &str) -> &str {
+    marker
+        .rfind(LIST_CACHE_MARKER_PREFIX)
+        .filter(|start_idx| marker[*start_idx..].ends_with(']'))
+        .map_or(marker, |start_idx| &marker[..start_idx])
+}
+
 fn normalize_list_quorum(value: &str) -> &'static str {
     let value = value.trim();
     if value.eq_ignore_ascii_case("disk") {
@@ -3085,7 +3093,7 @@ impl ListPathOptions {
         let Some(marker) = self.marker.clone() else {
             return;
         };
-        let Some(start_idx) = marker.rfind("[rustfs_cache:") else {
+        let Some(start_idx) = marker.rfind(LIST_CACHE_MARKER_PREFIX) else {
             return;
         };
         let Some(end_offset) = marker[start_idx..].rfind(']') else {

crates/rio/src/http_reader.rs

@@ -156,7 +156,7 @@ impl std::fmt::Display for InternodeHttpRequestContext {
     }
 }
 
-#[derive(Debug, thiserror::Error)]
+#[derive(thiserror::Error)]
 #[error("{kind}: {context}")]
 pub struct InternodeHttpError {
     kind: InternodeHttpErrorKind,
@@ -165,6 +165,16 @@ pub struct InternodeHttpError {
     source: Option<Box<dyn std::error::Error + Send + Sync>>,
 }
 
+impl std::fmt::Debug for InternodeHttpError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("InternodeHttpError")
+            .field("kind", &self.kind)
+            .field("context", &self.context)
+            .field("source_present", &self.source.is_some())
+            .finish()
+    }
+}
+
 impl InternodeHttpError {
     pub fn kind(&self) -> InternodeHttpErrorKind {
         self.kind
@@ -605,11 +615,8 @@ async fn get_http_client(url: &str) -> io::Result<Client> {
 fn internode_request_context(method: &Method, url: &str, operation: Option<&'static str>) -> InternodeHttpRequestContext {
     let target = reqwest::Url::parse(url)
         .ok()
-        .map(|parsed| match parsed.query() {
+        .map(|parsed| parsed.path().to_string())
-            Some(query) => format!("{}?{query}", parsed.path()),
+        .unwrap_or_else(|| "<invalid-internode-url>".to_string());
-            None => parsed.path().to_string(),
-        })
-        .unwrap_or_else(|| url.to_string());
     InternodeHttpRequestContext {
         method: method.to_string(),
         target,
@@ -723,6 +730,12 @@ fn internode_reqwest_error(method: &Method, url: &str, operation: Option<&'stati
     InternodeHttpError::with_source(classified, context, err).into_io_error()
 }
 
+fn internode_reqwest_body_error(method: &Method, url: &str, operation: Option<&'static str>, err: reqwest::Error) -> io::Error {
+    let context = internode_request_context(method, url, operation);
+    let classified = classify_transport_error(&err, err.is_timeout(), err.is_connect(), true);
+    InternodeHttpError::with_source(classified, context, err).into_io_error()
+}
+
 fn internode_classified_error(
     method: &Method,
     url: &str,
@@ -823,8 +836,9 @@ impl HttpReader {
         let stream_error_method = method.clone();
         let stream = resp.bytes_stream().map_err(move |e| {
             record_internode_error(track_internode_metrics, internode_operation);
-            record_internode_classified_error(track_internode_metrics, internode_operation, classify_reqwest_error(&e));
+            let classified = classify_transport_error(&e, e.is_timeout(), e.is_connect(), true);
-            internode_reqwest_error(&stream_error_method, &stream_error_url, internode_operation, e)
+            record_internode_classified_error(track_internode_metrics, internode_operation, classified);
+            internode_reqwest_body_error(&stream_error_method, &stream_error_url, internode_operation, e)
         });
 
         Ok(Self {
@@ -1653,6 +1667,16 @@ mod tests {
         (StatusCode::OK, Body::from_stream(body_stream))
     }
 
+    async fn get_failing_stream(State(state): State<TestState>) -> impl IntoResponse {
+        state.get_count.fetch_add(1, Ordering::SeqCst);
+        let body_stream =
+            stream::once(async { Ok::<Bytes, io::Error>(Bytes::from_static(b"partial")) }).chain(stream::once(async {
+                tokio::time::sleep(Duration::from_millis(25)).await;
+                Err(io::Error::other("stream failed"))
+            }));
+        (StatusCode::OK, Body::from_stream(body_stream))
+    }
+
     async fn reject_head(State(state): State<TestState>) -> impl IntoResponse {
         state.head_count.fetch_add(1, Ordering::SeqCst);
         StatusCode::METHOD_NOT_ALLOWED
@@ -1684,6 +1708,7 @@ mod tests {
             .route("/reject-put", get(get_stream).put(reject_put))
             .route("/stall", get(get_stalling_stream))
             .route("/delayed-first", get(get_delayed_first_chunk))
+            .route("/fail-after-partial", get(get_failing_stream))
             .with_state(state);
 
         let handle = tokio::spawn(async move {
@@ -1918,6 +1943,37 @@ mod tests {
         assert!(source.context().target().contains("/stream"));
     }
 
+    #[tokio::test]
+    async fn http_reader_surfaces_body_error_after_partial_data() {
+        let state = TestState::default();
+        let Some((base_url, handle)) = start_test_server(state).await else {
+            return;
+        };
+        let url = base_url.replace("/stream", "/fail-after-partial");
+        let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None)
+            .await
+            .expect("reader should accept the successful response headers");
+        let mut partial = [0_u8; 7];
+
+        reader
+            .read_exact(&mut partial)
+            .await
+            .expect("partial response bytes should arrive before the terminal error");
+        let err = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("terminal body errors must not become clean EOF");
+
+        assert_eq!(&partial, b"partial");
+        let source = err
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<InternodeHttpError>())
+            .expect("body error should retain internode classification");
+        assert_eq!(source.kind(), InternodeHttpErrorKind::BodyStreamAborted);
+
+        handle.abort();
+    }
+
     #[test]
     fn classify_http_status_marks_retryable_gateway_errors() {
         let unavailable = classify_http_status(reqwest::StatusCode::SERVICE_UNAVAILABLE);
@@ -1961,7 +2017,36 @@ mod tests {
             InternodeHttpErrorKind::HttpStatus(reqwest::StatusCode::SERVICE_UNAVAILABLE)
         );
         assert_eq!(source.context().method(), "PUT");
-        assert!(source.context().target().contains(PUT_FILE_STREAM_PATH));
+        assert_eq!(source.context().target(), PUT_FILE_STREAM_PATH);
+        assert!(!err.to_string().contains("disk-a"));
+    }
+
+    #[test]
+    fn internode_request_context_redacts_malformed_targets() {
+        let context =
+            internode_request_context(&Method::GET, "not a url?disk=/sensitive/path", Some(INTERNODE_OPERATION_READ_FILE_STREAM));
+
+        assert_eq!(context.target(), "<invalid-internode-url>");
+        assert!(!context.to_string().contains("sensitive"));
+    }
+
+    #[test]
+    fn internode_http_error_debug_redacts_source_details() {
+        let context = InternodeHttpRequestContext {
+            method: "GET".to_string(),
+            target: WALK_DIR_PATH.to_string(),
+            operation: Some(INTERNODE_OPERATION_WALK_DIR),
+        };
+        let error = InternodeHttpError::with_source(
+            InternodeHttpErrorKind::BodyStreamAborted,
+            context,
+            io::Error::other("http://node/rustfs/rpc/walk_dir?disk=/sensitive/path&token=private"),
+        );
+        let debug = format!("{error:?}");
+
+        assert!(debug.contains("source_present: true"));
+        assert!(!debug.contains("sensitive"));
+        assert!(!debug.contains("private"));
     }
 
     #[test]

crates/storage-api/src/lib.rs

@@ -14,6 +14,10 @@
 
 //! Storage API contracts for RustFS.
 
+pub const WALK_DIR_STREAM_COMPLETION_QUERY: &str = "walk_dir_stream_completion";
+pub const WALK_DIR_STREAM_COMPLETION_V1: &str = "error-v1";
+pub const WALK_DIR_BODY_SHA256_QUERY: &str = "walk_dir_body_sha256";
+
 pub mod admin;
 pub mod bucket;
 pub mod capability;

docs/architecture/compat-cleanup-register.md

@@ -12,6 +12,8 @@ for later deletion.
 
 ## Open Items
 
+- `#4648` walk-dir stream completion capability: old clients can append fallback output to an already-used metacache writer after a terminal body error, so servers emit terminal walk errors only to clients that sign the `walk_dir_stream_completion=error-v1` query capability and its request-body digest. Remove the legacy clean-EOF path after the minimum supported RustFS peer version always advertises this capability.
+
 ## Review Checklist
 
 Before completing a PR that adds wrappers, re-exports, fallbacks, legacy action

rustfs/src/admin/handlers/account_info.rs

@@ -20,7 +20,7 @@ use crate::admin::storage_api::contract::admin::StorageAdminApi;
 use crate::admin::storage_api::contract::bucket::{BucketOperations, BucketOptions};
 use crate::admin::storage_api::data_usage::{
     apply_bucket_usage_memory_overlay, load_data_usage_from_backend, refresh_bucket_usage_from_object_layer,
-    refresh_versioned_bucket_usage_from_object_layer, replace_bucket_usage_memory_from_info,
+    replace_bucket_usage_memory_from_info,
 };
 use crate::admin::storage_api::metadata_sys;
 use crate::auth::get_condition_values;
@@ -258,7 +258,6 @@ impl Operation for AccountInfoHandler {
         let mut data_usage_info = load_data_usage_from_backend(store.clone())
             .await
             .map_err(|e| S3Error::with_message(S3ErrorCode::InternalError, e.to_string()))?;
-        refresh_versioned_bucket_usage_from_object_layer(store.clone(), &mut data_usage_info).await;
         replace_bucket_usage_memory_from_info(&data_usage_info).await;
         apply_bucket_usage_memory_overlay(&mut data_usage_info).await;
 

rustfs/src/admin/storage_api.rs

@@ -476,14 +476,6 @@ pub(crate) mod data_usage {
         crate::storage::storage_api::ecstore_data_usage::load_data_usage_from_backend(store).await
     }
 
-    pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer(
-        store: Arc<crate::storage::storage_api::ECStore>,
-        data_usage_info: &mut rustfs_data_usage::DataUsageInfo,
-    ) {
-        crate::storage::storage_api::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info)
-            .await;
-    }
-
     pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) {
         crate::storage::storage_api::ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await;
     }

rustfs/src/app/admin_usecase.rs

@@ -22,7 +22,7 @@ use super::storage_api::admin_usecase::contract::StorageAdminApi;
 use super::storage_api::admin_usecase::contract::bucket::{BucketOperations, BucketOptions};
 use super::storage_api::admin_usecase::data_usage::{
     apply_bucket_usage_memory_overlay, load_data_usage_from_backend, refresh_bucket_usage_from_object_layer,
-    refresh_versioned_bucket_usage_from_object_layer, replace_bucket_usage_memory_from_info,
+    replace_bucket_usage_memory_from_info,
 };
 use super::storage_api::admin_usecase::{ECStore, EndpointServerPools};
 use crate::app::runtime_sources::{
@@ -249,7 +249,6 @@ impl DefaultAdminUsecase {
             error!("load_data_usage_from_backend failed {:?}", e);
             Self::app_error(S3ErrorCode::InternalError, "load_data_usage_from_backend failed")
         })?;
-        refresh_versioned_bucket_usage_from_object_layer(store.clone(), &mut info).await;
         replace_bucket_usage_memory_from_info(&info).await;
         apply_bucket_usage_memory_overlay(&mut info).await;
         Self::refresh_live_bucket_usage_for_data_usage_info(store.clone(), &mut info).await;

rustfs/src/app/storage_api.rs

@@ -70,14 +70,6 @@ pub(crate) mod data_usage {
         .await
     }
 
-    pub(crate) async fn refresh_versioned_bucket_usage_from_object_layer(
-        store: Arc<crate::storage::storage_api::ECStore>,
-        data_usage_info: &mut rustfs_data_usage::DataUsageInfo,
-    ) {
-        crate::storage::storage_api::ecstore_data_usage::refresh_versioned_bucket_usage_from_object_layer(store, data_usage_info)
-            .await;
-    }
-
     pub(crate) async fn replace_bucket_usage_memory_from_info(data_usage_info: &rustfs_data_usage::DataUsageInfo) {
         crate::storage::storage_api::ecstore_data_usage::replace_bucket_usage_memory_from_info(data_usage_info).await;
     }

rustfs/src/storage/rpc/http_service.rs

@@ -14,12 +14,15 @@
 
 use crate::server::RPC_PREFIX;
 use crate::storage::request_context::spawn_traced;
+#[cfg(test)]
+use crate::storage::storage_api::rpc_consumer::http_service::WALK_DIR_BODY_SHA256_QUERY;
 use crate::storage::storage_api::rpc_consumer::http_service::{
-    DEFAULT_READ_BUFFER_SIZE, StorageDiskRpcExt as _, WalkDirOptions, find_local_disk_by_ref, verify_rpc_signature,
+    DEFAULT_READ_BUFFER_SIZE, StorageDiskRpcExt as _, WALK_DIR_STREAM_COMPLETION_V1, WalkDirOptions, find_local_disk_by_ref,
+    verify_rpc_signature,
 };
 use crate::storage::storage_api::runtime_sources_consumer::runtime_sources;
 use bytes::{Bytes, BytesMut};
-use futures_util::TryStreamExt;
+use futures_util::{Stream, StreamExt, TryStreamExt, stream};
 use http::{HeaderMap, Method, Request, Response, StatusCode, Uri};
 use http_body_util::{BodyExt, Limited};
 use hyper::body::Incoming;
@@ -33,11 +36,13 @@ use s3s::Body;
 use s3s::dto::StreamingBlob;
 use serde::de::DeserializeOwned;
 use serde_urlencoded::from_bytes;
+use sha2::{Digest, Sha256};
 use std::future::Future;
 use std::pin::Pin;
 use std::task::{Context, Poll};
 use std::time::Instant;
 use tokio::io::{self, AsyncWriteExt};
+use tokio::sync::oneshot;
 use tokio_util::io::ReaderStream;
 use tower::Service;
 use tracing::{error, warn};
@@ -238,6 +243,32 @@ struct ReadFileQuery {
 #[derive(Debug, Default, serde::Deserialize)]
 struct WalkDirQuery {
     disk: String,
+    walk_dir_stream_completion: Option<String>,
+    walk_dir_body_sha256: Option<String>,
+}
+
+fn supports_walk_dir_stream_completion(query: &WalkDirQuery) -> bool {
+    query.walk_dir_stream_completion.as_deref() == Some(WALK_DIR_STREAM_COMPLETION_V1)
+}
+
+fn verify_walk_dir_body_digest(query: &WalkDirQuery, body: &[u8]) -> bool {
+    if !supports_walk_dir_stream_completion(query) {
+        return true;
+    }
+
+    let Some(expected) = query.walk_dir_body_sha256.as_deref() else {
+        return false;
+    };
+    let actual = hex_simd::encode_to_string(Sha256::digest(body), hex_simd::AsciiCase::Lower);
+    expected == actual
+}
+
+fn validate_walk_dir_completion_request(query: &WalkDirQuery, body: &[u8]) -> Option<bool> {
+    let propagate_completion_errors = supports_walk_dir_stream_completion(query);
+    if !verify_walk_dir_body_digest(query, body) {
+        return None;
+    }
+    Some(propagate_completion_errors)
 }
 
 #[derive(Debug, Default, serde::Deserialize)]
@@ -450,7 +481,6 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
         Ok(query) => query,
         Err(response) => return *response,
     };
-
     let Some(disk) = find_local_disk_by_ref(&query.disk).await else {
         warn!(
             event = EVENT_RPC_REQUEST_REJECTED,
@@ -485,6 +515,26 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
             return response_with_status(StatusCode::PAYLOAD_TOO_LARGE, message);
         }
     };
+    // RUSTFS_COMPAT_TODO(#4648): old clients retry terminal stream failures on an already-used writer.
+    // Remove after every supported peer version advertises walk-dir stream completion v1.
+    let propagate_completion_errors = match validate_walk_dir_completion_request(&query, &body) {
+        Some(propagate_completion_errors) => propagate_completion_errors,
+        None => {
+            warn!(
+                event = EVENT_RPC_REQUEST_REJECTED,
+                component = LOG_COMPONENT_INTERNODE_RPC,
+                subsystem = LOG_SUBSYSTEM_DIRECTORY_WALK,
+                operation = INTERNODE_OPERATION_WALK_DIR,
+                result = "rejected",
+                status_code = StatusCode::FORBIDDEN.as_u16(),
+                rpc_path = WALK_DIR_PATH,
+                method = %Method::GET,
+                reason = "request_body_digest_mismatch",
+                "internode rpc request rejected"
+            );
+            return response_with_status(StatusCode::FORBIDDEN, "invalid request body digest");
+        }
+    };
 
     let args: WalkDirOptions = match serde_json::from_slice(&body) {
         Ok(args) => args,
@@ -518,9 +568,8 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
     let log_limit = args.limit;
     let log_disk_id = args.disk_id.clone();
     let log_skip_total_timeout = args.skip_total_timeout;
-    let (rd, mut wd) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
+    let body = walk_dir_response_body(propagate_completion_errors, move |mut writer| async move {
-    spawn_traced(async move {
+        disk.walk_dir(args, &mut writer).await.map_err(|e| {
-        if let Err(e) = disk.walk_dir(args, &mut wd).await {
             warn!(
                 event = EVENT_RPC_BACKGROUND_TASK_FAILED,
                 component = LOG_COMPONENT_INTERNODE_RPC,
@@ -540,13 +589,38 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
                 error = %e,
                 "internode rpc background task failed"
             );
-        }
+            io::Error::other("remote walk_dir failed")
+        })
     });
 
     runtime_sources::current_internode_metrics()
         .record_incoming_request_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP);
+
+    Response::builder()
+        .status(StatusCode::OK)
+        .body(body)
+        .expect("failed to build walk dir response")
+}
+
+fn walk_dir_response_body<F, Fut>(propagate_completion_errors: bool, producer: F) -> Body
+where
+    F: FnOnce(tokio::io::DuplexStream) -> Fut + Send + 'static,
+    Fut: Future<Output = io::Result<()>> + Send + 'static,
+{
+    let (reader, writer) = tokio::io::duplex(DEFAULT_READ_BUFFER_SIZE);
+    let (mut completion_tx, completion_rx) = oneshot::channel();
+    spawn_traced(async move {
+        tokio::select! {
+            biased;
+            result = producer(writer) => {
+                let _ = completion_tx.send(result);
+            }
+            _ = completion_tx.closed() => {}
+        }
+    });
+
     let metrics = runtime_sources::current_internode_metrics();
-    let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
+    let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| {
         metrics.record_sent_bytes_for_operation_and_backend(
             INTERNODE_OPERATION_WALK_DIR,
             INTERNODE_TRANSPORT_BACKEND_TCP_HTTP,
@@ -554,11 +628,32 @@ async fn handle_walk_dir(req: Request<Incoming>) -> Response<Body> {
         );
         bytes
     });
+    let stream = append_walk_dir_completion(stream, completion_rx, propagate_completion_errors);
 
-    Response::builder()
+    Body::from(StreamingBlob::wrap(stream))
-        .status(StatusCode::OK)
+}
-        .body(Body::from(StreamingBlob::wrap(stream)))
+
-        .expect("failed to build walk dir response")
+fn append_walk_dir_completion<S>(
+    stream: S,
+    completion_rx: oneshot::Receiver<io::Result<()>>,
+    propagate_completion_errors: bool,
+) -> impl Stream<Item = io::Result<Bytes>>
+where
+    S: Stream<Item = io::Result<Bytes>>,
+{
+    stream.chain(
+        stream::once(async move {
+            match completion_rx.await {
+                Ok(Ok(())) => None,
+                Ok(Err(err)) if propagate_completion_errors => Some(Err(err)),
+                Err(err) if propagate_completion_errors => {
+                    Some(Err(io::Error::other(format!("remote walk_dir task ended without a result: {err}"))))
+                }
+                Ok(Err(_)) | Err(_) => None,
+            }
+        })
+        .filter_map(std::future::ready),
+    )
 }
 
 async fn handle_put_file(req: Request<Incoming>) -> Response<Body> {
@@ -734,20 +829,34 @@ fn put_file_stage_error_message(stage: &str, query: &PutFileQuery, err: &dyn std
 mod tests {
     use super::{
         LOG_SUBSYSTEM_DIRECTORY_WALK, LOG_SUBSYSTEM_FILE_TRANSFER, LOG_SUBSYSTEM_ROUTING, PUT_FILE_STREAM_PATH, PutFileQuery,
-        READ_FILE_STREAM_PATH, WALK_DIR_PATH, internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path,
+        READ_FILE_STREAM_PATH, WALK_DIR_BODY_SHA256_QUERY, WALK_DIR_PATH, WalkDirQuery, append_walk_dir_completion,
-        put_body_size_mismatch, put_file_stage_error_message, read_file_body_stream, verify_internode_rpc_signature,
+        internode_http_operation, internode_rpc_subsystem, is_internode_rpc_path, put_body_size_mismatch,
-        write_body_chunks_to_writer,
+        put_file_stage_error_message, read_file_body_stream, supports_walk_dir_stream_completion,
+        validate_walk_dir_completion_request, verify_internode_rpc_signature, verify_walk_dir_body_digest,
+        walk_dir_response_body, write_body_chunks_to_writer,
     };
     use bytes::Bytes;
     use http::{HeaderMap, Method, StatusCode, Uri};
+    use http_body_util::BodyExt;
     use rustfs_io_metrics::internode_metrics::{
         INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR,
     };
+    use sha2::Digest as _;
     use tokio::io;
     use tokio::io::{AsyncReadExt, AsyncWriteExt};
     use tokio_stream::StreamExt;
     use tokio_stream::iter;
 
+    struct DropNotifier(Option<tokio::sync::oneshot::Sender<()>>);
+
+    impl Drop for DropNotifier {
+        fn drop(&mut self) {
+            if let Some(sender) = self.0.take() {
+                let _ = sender.send(());
+            }
+        }
+    }
+
     #[test]
     fn internode_rpc_path_matches_rpc_prefix() {
         assert!(is_internode_rpc_path("/rustfs/rpc/read_file_stream"));
@@ -852,6 +961,130 @@ mod tests {
         assert_eq!(out, b"hello world");
     }
 
+    #[tokio::test]
+    async fn walk_dir_body_surfaces_background_failure_after_data() {
+        let body = walk_dir_response_body(true, |mut writer| async move {
+            writer.write_all(b"partial walk data").await?;
+            Err(io::Error::other("remote walk_dir failed"))
+        });
+        let err = BodyExt::collect(body)
+            .await
+            .expect_err("failed completion must fail body collection");
+
+        assert!(err.to_string().contains("remote walk_dir failed"));
+    }
+
+    #[tokio::test]
+    async fn walk_dir_body_preserves_data_after_success() {
+        let body = walk_dir_response_body(true, |mut writer| async move {
+            writer.write_all(b"complete walk data").await?;
+            Ok(())
+        });
+        let bytes = BodyExt::collect(body)
+            .await
+            .expect("successful completion should preserve the body")
+            .to_bytes();
+
+        assert_eq!(bytes, Bytes::from_static(b"complete walk data"));
+    }
+
+    #[tokio::test]
+    async fn walk_dir_completion_stream_surfaces_cancelled_producer() {
+        let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
+        drop(completion_tx);
+        let stream = iter([Ok::<Bytes, io::Error>(Bytes::from_static(b"partial walk data"))]);
+        let body = s3s::Body::from(s3s::dto::StreamingBlob::wrap(append_walk_dir_completion(stream, completion_rx, true)));
+
+        let err = BodyExt::collect(body)
+            .await
+            .expect_err("a cancelled producer must fail body collection");
+
+        assert!(err.to_string().contains("ended without a result"));
+    }
+
+    #[tokio::test]
+    async fn dropping_walk_dir_body_cancels_blocked_producer() {
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
+        let body = walk_dir_response_body(true, move |_writer| async move {
+            let _drop_notifier = DropNotifier(Some(dropped_tx));
+            let _ = started_tx.send(());
+            std::future::pending::<io::Result<()>>().await
+        });
+
+        started_rx.await.expect("walk producer should start");
+        drop(body);
+
+        tokio::time::timeout(std::time::Duration::from_secs(1), dropped_rx)
+            .await
+            .expect("dropping the response body should cancel the walk producer")
+            .expect("drop notifier should send a cancellation signal");
+    }
+
+    #[tokio::test]
+    async fn legacy_walk_dir_client_keeps_clean_eof_compatibility() {
+        let body = walk_dir_response_body(false, |mut writer| async move {
+            writer.write_all(b"legacy partial data").await?;
+            Err(io::Error::other("remote walk_dir failed"))
+        });
+
+        let bytes = BodyExt::collect(body)
+            .await
+            .expect("legacy clients must retain clean EOF until they advertise stream completion support")
+            .to_bytes();
+
+        assert_eq!(bytes, Bytes::from_static(b"legacy partial data"));
+    }
+
+    #[tokio::test]
+    async fn legacy_walk_dir_client_keeps_clean_eof_when_producer_is_cancelled() {
+        let (completion_tx, completion_rx) = tokio::sync::oneshot::channel();
+        drop(completion_tx);
+        let stream = iter([Ok::<Bytes, io::Error>(Bytes::from_static(b"legacy partial data"))]);
+        let body = s3s::Body::from(s3s::dto::StreamingBlob::wrap(append_walk_dir_completion(stream, completion_rx, false)));
+
+        let bytes = BodyExt::collect(body)
+            .await
+            .expect("legacy clients must retain clean EOF after producer cancellation")
+            .to_bytes();
+
+        assert_eq!(bytes, Bytes::from_static(b"legacy partial data"));
+    }
+
+    #[test]
+    fn walk_dir_completion_requires_the_exact_signed_query_capability() {
+        let legacy: WalkDirQuery = serde_urlencoded::from_str("disk=disk-a").expect("legacy query should parse");
+        let unknown: WalkDirQuery =
+            serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v2").expect("unknown query should parse");
+        let capable: WalkDirQuery =
+            serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v1").expect("capable query should parse");
+
+        assert!(!supports_walk_dir_stream_completion(&legacy));
+        assert!(!supports_walk_dir_stream_completion(&unknown));
+        assert!(supports_walk_dir_stream_completion(&capable));
+    }
+
+    #[test]
+    fn walk_dir_completion_requires_matching_signed_body_digest() {
+        let body = br#"{"bucket":"bucket-a"}"#;
+        let digest = hex_simd::encode_to_string(sha2::Sha256::digest(body), hex_simd::AsciiCase::Lower);
+        let capable: WalkDirQuery = serde_urlencoded::from_str(&format!(
+            "disk=disk-a&walk_dir_stream_completion=error-v1&{WALK_DIR_BODY_SHA256_QUERY}={digest}"
+        ))
+        .expect("capable query should parse");
+        let missing: WalkDirQuery =
+            serde_urlencoded::from_str("disk=disk-a&walk_dir_stream_completion=error-v1").expect("query should parse");
+        let legacy: WalkDirQuery = serde_urlencoded::from_str("disk=disk-a").expect("legacy query should parse");
+
+        assert!(verify_walk_dir_body_digest(&capable, body));
+        assert!(!verify_walk_dir_body_digest(&capable, b"tampered"));
+        assert!(!verify_walk_dir_body_digest(&missing, body));
+        assert!(verify_walk_dir_body_digest(&legacy, b"legacy body"));
+        assert_eq!(validate_walk_dir_completion_request(&capable, body), Some(true));
+        assert_eq!(validate_walk_dir_completion_request(&legacy, b"legacy body"), Some(false));
+        assert_eq!(validate_walk_dir_completion_request(&capable, b"tampered"), None);
+    }
+
     #[tokio::test]
     async fn read_file_body_stream_keeps_full_stream_when_length_is_zero() {
         let (reader, mut writer) = tokio::io::duplex(64);

rustfs/src/storage/storage_api.rs

@@ -188,6 +188,9 @@ pub(crate) mod rpc_consumer {
 
     pub(crate) mod http_service {
         pub(crate) const DEFAULT_READ_BUFFER_SIZE: usize = super::super::DEFAULT_READ_BUFFER_SIZE;
+        #[cfg(test)]
+        pub(crate) use super::super::storage_contracts::WALK_DIR_BODY_SHA256_QUERY;
+        pub(crate) use super::super::storage_contracts::WALK_DIR_STREAM_COMPLETION_V1;
         pub(crate) use super::super::{StorageDiskRpcExt, WalkDirOptions, find_local_disk_by_ref, verify_rpc_signature};
     }
 
@@ -361,8 +364,8 @@ pub(crate) mod ecstore_data_usage {
         apply_bucket_usage_memory_overlay, init_compression_total_memory_from_backend, load_data_usage_from_backend,
         record_bucket_delete_marker_memory, record_bucket_object_delete_memory, record_bucket_object_version_write_memory,
         record_bucket_object_write_memory, record_bucket_object_write_unknown_previous_memory,
-        refresh_bucket_usage_from_object_layer, refresh_versioned_bucket_usage_from_object_layer,
+        refresh_bucket_usage_from_object_layer, remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info,
-        remove_bucket_usage_from_backend, replace_bucket_usage_memory_from_info, store_compression_total_in_backend,
+        store_compression_total_in_backend,
     };
 }