fix(storage): harden scanner and recovery edge cases (#5521)

* fix(ecstore): handle benign listing and GET disconnects

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(scanner): scope cache locks by set

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(kms): stabilize Vault transport retry coverage

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(scanner): fence scoped cache locks by protocol

Co-Authored-By: heihutu <heihutu@gmail.com>

* test(ecstore): stabilize topology DNS fallback coverage

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix(kms): remove stale local export test import

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

crates/ecstore/src/erasure/coding/decode.rs

@@ -1645,15 +1645,28 @@ impl Erasure {
             }
             Err(e) => {
                 record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_EMIT, emit_stage_start);
-                error!(
-                    block_offset,
-                    block_length,
-                    bytes_written = *written,
-                    stage = GET_STAGE_EMIT,
-                    reason = classify_io_error(&e).as_str(),
-                    error = ?e,
-                    "Erasure decode failed to emit reconstructed data"
-                );
+                let reason = classify_io_error(&e);
+                if reason == GetObjectFailureReason::DownstreamClosed {
+                    debug!(
+                        block_offset,
+                        block_length,
+                        bytes_written = *written,
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Erasure decode stopped after downstream closed"
+                    );
+                } else {
+                    error!(
+                        block_offset,
+                        block_length,
+                        bytes_written = *written,
+                        stage = GET_STAGE_EMIT,
+                        reason = reason.as_str(),
+                        error = ?e,
+                        "Erasure decode failed to emit reconstructed data"
+                    );
+                }
                 *ret_err = Some(e);
                 return StripeFlow::Stop;
             }
@@ -1945,7 +1958,7 @@ mod tests {
     use std::io::Cursor;
     use std::pin::Pin;
     use std::sync::{
-        Arc,
+        Arc, Mutex,
         atomic::{AtomicUsize, Ordering},
     };
     use std::task::{Context, Poll};
@@ -2120,6 +2133,59 @@ mod tests {
         }
     }
 
+    struct DownstreamClosedWriter;
+
+    impl AsyncWrite for DownstreamClosedWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<io::Result<usize>> {
+            Poll::Ready(Err(crate::diagnostics::get::mark_get_object_downstream_closed(io::Error::new(
+                ErrorKind::BrokenPipe,
+                "injected downstream close",
+            ))))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[derive(Clone, Default)]
+    struct CapturedLogs(Arc<Mutex<Vec<u8>>>);
+
+    struct CapturedLogWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl CapturedLogs {
+        fn contents(&self) -> String {
+            String::from_utf8(self.0.lock().expect("captured logs mutex should not be poisoned").clone())
+                .expect("captured logs should be valid UTF-8")
+        }
+    }
+
+    impl std::io::Write for CapturedLogWriter {
+        fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
+            self.0
+                .lock()
+                .expect("captured logs mutex should not be poisoned")
+                .extend_from_slice(buf);
+            Ok(buf.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for CapturedLogs {
+        type Writer = CapturedLogWriter;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            CapturedLogWriter(Arc::clone(&self.0))
+        }
+    }
+
     #[test]
     fn parallel_reader_constructor_variants_preserve_read_cost_and_verification_flags() {
         let erasure = Erasure::new(2, 1, 64);
@@ -2215,6 +2281,47 @@ mod tests {
         assert_eq!(err.to_string(), "injected emit failure");
     }
 
+    #[tokio::test(flavor = "current_thread")]
+    async fn erasure_decode_logs_reconstructed_downstream_close_at_debug() {
+        let logs = CapturedLogs::default();
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::DEBUG)
+            .with_writer(logs.clone())
+            .with_ansi(false)
+            .without_time()
+            .finish();
+        let _guard = tracing::subscriber::set_default(subscriber);
+
+        let erasure = Erasure::new(2, 1, 64);
+        let data: Vec<u8> = (0..64).collect();
+        let shard_size = erasure.shard_size();
+        let encoded = erasure.encode_data(&data).expect("test data should encode");
+        let readers = vec![
+            None,
+            Some(BitrotReader::new(
+                Cursor::new(encoded[1].to_vec()),
+                shard_size,
+                HashAlgorithm::None,
+                false,
+            )),
+            Some(BitrotReader::new(
+                Cursor::new(encoded[2].to_vec()),
+                shard_size,
+                HashAlgorithm::None,
+                false,
+            )),
+        ];
+
+        let mut writer = DownstreamClosedWriter;
+        let (written, err) = erasure.decode(&mut writer, readers, 0, data.len(), data.len()).await;
+
+        assert_eq!(written, 0);
+        assert_eq!(err.expect("downstream close must still terminate the GET").kind(), ErrorKind::BrokenPipe);
+        let captured = logs.contents();
+        assert!(captured.contains("Erasure decode stopped after downstream closed"));
+        assert!(!captured.contains("Erasure decode failed to emit reconstructed data"));
+    }
+
     #[tokio::test]
     async fn test_erasure_decode_rejects_reader_count_and_range_overflow() {
         let erasure = Erasure::new(2, 1, 64);

crates/ecstore/src/layout/endpoints.rs

@@ -22,6 +22,8 @@ use rustfs_config::{
     ENV_STARTUP_TOPOLOGY_WAIT_MODE, ENV_STARTUP_TOPOLOGY_WAIT_TIMEOUT, ENV_UNSAFE_BYPASS_DISK_CHECK,
 };
 use rustfs_utils::{XHost, check_local_server_addr, get_env_opt_str, get_host_ip, is_local_host};
+#[cfg(test)]
+use std::sync::{LazyLock, Mutex};
 use std::{
     collections::{BTreeMap, BTreeSet, HashMap, HashSet, hash_map::Entry},
     future::Future,
@@ -598,6 +600,58 @@ const DNS_RETRY_JITTER_PERCENT: u64 = 20;
 /// wait does not flood the log with one line per backoff tick.
 const TOPOLOGY_WARN_THROTTLE: Duration = Duration::from_secs(30);
 
+#[cfg(test)]
+static FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS: LazyLock<Mutex<HashSet<String>>> = LazyLock::new(|| Mutex::new(HashSet::new()));
+
+#[cfg(test)]
+struct LocalHostResolutionTimeoutGuard {
+    hosts: Vec<String>,
+}
+
+#[cfg(test)]
+impl Drop for LocalHostResolutionTimeoutGuard {
+    fn drop(&mut self) {
+        let mut forced_hosts = FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+            .lock()
+            .expect("local-host test resolver mutex poisoned");
+        for host in &self.hosts {
+            forced_hosts.remove(host);
+        }
+    }
+}
+
+#[cfg(test)]
+fn force_local_host_resolution_timeout_for_test(hosts: &[&str]) -> LocalHostResolutionTimeoutGuard {
+    let hosts = hosts.iter().map(|host| (*host).to_string()).collect::<Vec<_>>();
+    let mut forced_hosts = FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+        .lock()
+        .expect("local-host test resolver mutex poisoned");
+    forced_hosts.extend(hosts.iter().cloned());
+    LocalHostResolutionTimeoutGuard { hosts }
+}
+
+#[cfg(test)]
+fn local_host_resolution_timeout_forced(host: &Host<&str>) -> bool {
+    let host = match host {
+        Host::Domain(domain) => (*domain).to_string(),
+        Host::Ipv4(ip) => ip.to_string(),
+        Host::Ipv6(ip) => ip.to_string(),
+    };
+    FORCED_LOCAL_HOST_RESOLUTION_TIMEOUTS
+        .lock()
+        .expect("local-host test resolver mutex poisoned")
+        .contains(&host)
+}
+
+fn endpoint_is_local_host(host: Host<&str>, port: u16, local_port: u16) -> Result<bool> {
+    #[cfg(test)]
+    if local_host_resolution_timeout_forced(&host) {
+        return Err(Error::new(ErrorKind::TimedOut, "resolver timeout"));
+    }
+
+    is_local_host(host, port, local_port)
+}
+
 struct DnsRetryDeadline {
     started: Instant,
     timeout: Duration,
@@ -694,7 +748,7 @@ async fn resolve_local_host_with_retry(
     retry_dns_operation(
         || {
             let host = host.clone();
-            async move { is_local_host(host, port, local_port) }
+            async move { endpoint_is_local_host(host, port, local_port) }
         },
         async_sleep,
         dns_retry_deadline,
@@ -2231,6 +2285,9 @@ mod test {
     #[serial]
     #[tokio::test]
     async fn create_server_endpoints_bounds_kubernetes_alias_dns_fallback() {
+        let _resolution_timeout =
+            force_local_host_resolution_timeout_for_test(&["unrelated-0.example.invalid", "unrelated-1.example.invalid"]);
+
         async_with_vars(
             [
                 (ENV_LOCAL_ENDPOINT_HOST, None),

crates/ecstore/src/store/list_objects.rs

@@ -4861,10 +4861,7 @@ async fn poll_merge_head(rx: &CancellationToken, in_channels: &mut [Receiver<Met
 
 async fn send_or_cancel(rx: &CancellationToken, out_channel: &Sender<MetaCacheEntry>, entry: MetaCacheEntry) -> Result<bool> {
     tokio::select! {
-        result = out_channel.send(entry) => {
-            result.map_err(Error::other)?;
-            Ok(true)
-        }
+        result = out_channel.send(entry) => Ok(result.is_ok()),
         _ = rx.cancelled() => Ok(false),
     }
 }
@@ -9858,6 +9855,28 @@ mod test {
         assert_eq!(results, vec!["obj-a", "obj-b"]);
     }
 
+    #[tokio::test]
+    async fn merge_entry_channels_treats_dropped_output_receiver_as_completion() {
+        let (tx_a, rx_a) = mpsc::channel(4);
+        let (tx_b, rx_b) = mpsc::channel(4);
+        let (out_tx, out_rx) = mpsc::channel(1);
+
+        tx_a.send(test_meta_entry("obj-a")).await.unwrap();
+        tx_b.send(test_meta_entry("obj-b")).await.unwrap();
+        drop(tx_a);
+        drop(tx_b);
+        drop(out_rx);
+
+        let result = timeout(
+            Duration::from_secs(1),
+            merge_entry_channels(CancellationToken::new(), vec![rx_a, rx_b], out_tx, 1),
+        )
+        .await
+        .expect("merge should stop promptly when its consumer disconnects");
+
+        assert!(result.is_ok(), "consumer disconnect must not surface as a merge worker error");
+    }
+
     #[test]
     fn walk_ascending_versions_contract_reverses_newest_first_metadata() {
         // Documents the invariant the walk `versions_sort` handling relies on:

crates/kms/src/backends/scripted_vault.rs

@@ -26,16 +26,16 @@ use std::sync::{Arc, Mutex};
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
 use tokio::net::{TcpListener, TcpStream};
 
-/// One canned HTTP response.
-pub(crate) struct ScriptedResponse {
-    status: u16,
-    body: String,
+/// One scripted connection outcome.
+pub(crate) enum ScriptedResponse {
+    Http { status: u16, body: String },
+    Close,
 }
 
 impl ScriptedResponse {
     /// A 200 response carrying `data` inside the standard Vault envelope.
     pub(crate) fn ok(data: serde_json::Value) -> Self {
-        Self {
+        Self::Http {
             status: 200,
             body: serde_json::json!({
                 "request_id": "scripted",
@@ -50,11 +50,16 @@ impl ScriptedResponse {
 
     /// An error response in Vault's `{"errors": [...]}` format.
     pub(crate) fn error(status: u16, message: &str) -> Self {
-        Self {
+        Self::Http {
             status,
             body: serde_json::json!({ "errors": [message] }).to_string(),
         }
     }
+
+    /// Close the connection after consuming a request without sending an HTTP response.
+    pub(crate) fn close() -> Self {
+        Self::Close
+    }
 }
 
 /// A scripted stand-in Vault listening on a loopback port.
@@ -88,14 +93,14 @@ impl ScriptedVault {
                 let response = responses
                     .next()
                     .unwrap_or_else(|| ScriptedResponse::error(599, "scripted vault: script exhausted"));
-                let payload = format!(
-                    "HTTP/1.1 {} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{}",
-                    response.status,
-                    response.body.len(),
-                    response.body
-                );
-                let _ = stream.write_all(payload.as_bytes()).await;
-                let _ = stream.shutdown().await;
+                if let ScriptedResponse::Http { status, body } = response {
+                    let payload = format!(
+                        "HTTP/1.1 {status} Scripted\r\ncontent-type: application/json\r\ncontent-length: {}\r\nconnection: close\r\n\r\n{body}",
+                        body.len(),
+                    );
+                    let _ = stream.write_all(payload.as_bytes()).await;
+                    let _ = stream.shutdown().await;
+                }
             }
         });
 

crates/kms/src/backends/vault.rs

@@ -1512,6 +1512,8 @@ mod tests {
     use crate::backends::scripted_vault::{ScriptedResponse, ScriptedVault};
     use crate::config::{VaultAuthMethod, VaultConfig};
 
+    const SCRIPTED_RETRY_ATTEMPTS: u32 = 3;
+
     /// Vault + KMS config pair pointing at a scripted loopback Vault.
     fn scripted_configs(address: &str) -> (VaultConfig, KmsConfig) {
         let vault_config = VaultConfig {
@@ -1527,7 +1529,7 @@ mod tests {
         };
         let kms_config = KmsConfig {
             timeout: Duration::from_secs(5),
-            retry_attempts: 3,
+            retry_attempts: SCRIPTED_RETRY_ATTEMPTS,
             ..KmsConfig::default()
         };
         (vault_config, kms_config)
@@ -1596,6 +1598,31 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn wired_read_retries_closed_connections_within_budget() {
+        let (vault, client) = scripted_client(vec![ScriptedResponse::close(), ScriptedResponse::close()]).await;
+
+        let error = client
+            .get_key_data("wired-key")
+            .await
+            .expect_err("closed connections must exhaust the retry budget");
+        assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
+
+        let requests = vault.requests();
+        let expected_requests = usize::try_from(SCRIPTED_RETRY_ATTEMPTS).expect("retry attempts must fit usize");
+        assert_eq!(
+            requests.len(),
+            expected_requests,
+            "all budgeted retry attempts must reach Vault: {requests:?}"
+        );
+        assert!(
+            requests
+                .iter()
+                .all(|line| line == "GET /v1/secret/data/rustfs/kms/keys/wired-key"),
+            "all attempts must hit the same read endpoint: {requests:?}"
+        );
+    }
+
     #[tokio::test]
     async fn wired_read_does_not_retry_permission_errors() {
         let (vault, client) = scripted_client(vec![ScriptedResponse::error(403, "permission denied")]).await;

crates/kms/tests/vault_fault_injection.rs

@@ -14,8 +14,8 @@
 
 //! Fault-injection matrix for the Vault backend operation policy.
 //!
-//! Offline cases run against locally injected transport faults (a closed
-//! port, a listener that never responds) — deterministic, no external
+//! Offline cases run against locally injected transport faults (a listener
+//! that never responds) — deterministic, no external
 //! dependencies. Real-Vault cases are `#[ignore]`d and need a dev Vault
 //! (default `http://127.0.0.1:8200`, override with `RUSTFS_KMS_VAULT_ADDR`).
 //!
@@ -118,44 +118,6 @@ fn counter_value(snapshot: &[MetricEntry], name: &str, labels: &[(&str, &str)])
         .sum()
 }
 
-/// Connection refused: connection-class failures are retried up to the
-/// configured budget, then surface as a backend error.
-#[test]
-fn connection_refused_is_retried_within_budget() {
-    // Reserve a loopback port and release it so nothing is listening there.
-    let listener = std::net::TcpListener::bind("127.0.0.1:0").expect("reserve a loopback port");
-    let address = format!("http://{}", listener.local_addr().expect("reserved port addr"));
-    drop(listener);
-
-    let snapshot = record_metrics(|| {
-        Box::pin(async move {
-            let client = VaultKmsBackend::new(kms_config(vault_config(&address, "unused"), Duration::from_secs(2), 2))
-                .await
-                .expect("client construction performs no network calls");
-            let error = KmsBackendTrait::describe_key(&client, describe_key_request("fault-injection-refused"))
-                .await
-                .expect_err("a refused connection must fail the operation");
-            assert!(matches!(error, KmsError::BackendError { .. }), "got {error:?}");
-        })
-    });
-
-    assert_eq!(
-        counter_value(&snapshot, ATTEMPT_FAILURES_TOTAL, &[("error_class", "retryable_conn")]),
-        2,
-        "both budgeted attempts must observe the refused connection"
-    );
-    assert_eq!(counter_value(&snapshot, OPERATIONS_TOTAL, &[("outcome", "budget_exhausted")]), 1);
-    // The static-token login records its own success; the Vault read must not.
-    assert_eq!(
-        counter_value(
-            &snapshot,
-            OPERATIONS_TOTAL,
-            &[("operation", "vault_kv2_read_key"), ("outcome", "success")]
-        ),
-        0
-    );
-}
-
 /// Stalled connection: a server that accepts but never responds is cut off by
 /// the per-attempt timeout (either the policy timer or the equally sized HTTP
 /// client timeout, whichever fires first) instead of hanging forever.

crates/scanner/Cargo.toml

@@ -88,6 +88,7 @@ rmp-serde = { workspace = true }
 hmac = { workspace = true }
 sha2 = { workspace = true }
 rustfs-filemeta = { workspace = true }
+rustfs-lock.workspace = true
 tokio-util = { workspace = true, features = ["io", "compat", "rt"] }
 rustfs-ecstore = { workspace = true }
 rustfs-storage-api = { workspace = true }

crates/scanner/src/remote_scanner.rs

@@ -12,17 +12,18 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(test)]
+use crate::RUSTFS_META_BUCKET;
 use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig};
 use crate::scanner_io::{
-    DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, cache_root_entry_info, current_cache_root_or_prepare,
-    scanner_cache_lock_resource, scanner_cache_lock_timeout, scanner_set_disk_inventory,
+    DataUsageCacheScanState, ScannerDiskScanOutcome, ScannerIODisk, acquire_scanner_cache_locks, cache_root_entry_info,
+    current_cache_root_or_prepare, scanner_set_disk_inventory,
 };
 use crate::storage_api::owner::NS_SCANNER_PROTOCOL_VERSION;
-use crate::storage_api::scan::NamespaceLocking as _;
 use crate::{
     DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_CACHE_NAME, DataUsageCache, DataUsageCachePrepareOutcome, DataUsageCacheSource,
-    DataUsageEntryInfo, DataUsageScanPlanDigest, Disk, EcstoreError, RUSTFS_META_BUCKET, ScannerDiskExt as _, ScannerError,
-    ScannerObjectIO, StorageError, is_reserved_or_invalid_bucket, read_config, resolve_scanner_object_store_handle,
+    DataUsageEntryInfo, DataUsageScanPlanDigest, Disk, EcstoreError, ScannerDiskExt as _, ScannerError, ScannerObjectIO,
+    StorageError, is_reserved_or_invalid_bucket, read_config, resolve_scanner_object_store_handle,
 };
 use hmac::{Hmac, KeyInit, Mac};
 use rustfs_common::heal_channel::HealScanMode;
@@ -63,6 +64,7 @@ const NS_SCANNER_DISK_HEALTH_TIMEOUT: Duration = Duration::from_secs(5);
 const NS_SCANNER_VALIDATED_CYCLE_TTL: Duration = Duration::from_secs(1);
 const NS_SCANNER_MAX_REPLAY_SESSIONS: usize = 65_536;
 const NS_SCANNER_MAX_ERROR_CHARS: usize = 4096;
+const NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX: &str = "retry_bucket:";
 const NS_SCANNER_FRAME_AUTH_DOMAIN: &[u8] = b"rustfs-ns-scanner-frame-v3";
 
 pub const NS_SCANNER_MAX_REQUEST_BODY_SIZE: usize = 16 * 1024;
@@ -317,6 +319,13 @@ impl RemoteScannerServerError {
         }
     }
 
+    fn retry_bucket(message: impl Into<String>) -> Self {
+        Self {
+            scope: RemoteScannerErrorScope::Bucket,
+            error: ScannerError::Other(format!("{}{}", NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX, message.into())),
+        }
+    }
+
     fn into_frame(self) -> RemoteScannerErrorFrame {
         RemoteScannerErrorFrame {
             scope: self.scope,
@@ -336,6 +345,7 @@ struct RemoteScannerStreamError {
     error: StorageError,
     progress_fully_reported: bool,
     retire_worker: bool,
+    retry_bucket: bool,
 }
 
 impl RemoteScannerStreamError {
@@ -344,6 +354,7 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: false,
             retire_worker: true,
+            retry_bucket: false,
         }
     }
 
@@ -352,6 +363,7 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: true,
             retire_worker: true,
+            retry_bucket: false,
         }
     }
 
@@ -360,6 +372,16 @@ impl RemoteScannerStreamError {
             error,
             progress_fully_reported: true,
             retire_worker: false,
+            retry_bucket: false,
+        }
+    }
+
+    fn retry_bucket(error: StorageError) -> Self {
+        Self {
+            error,
+            progress_fully_reported: true,
+            retire_worker: false,
+            retry_bucket: true,
         }
     }
 
@@ -951,16 +973,14 @@ async fn scan_and_persist_local_bucket(
             ))
         })?;
     let cache_name = path_join_buf(&[&bucket, DATA_USAGE_CACHE_NAME]);
-    let lock_resource = scanner_cache_lock_resource(&cache_name);
-    let ns_lock = set
-        .new_ns_lock(RUSTFS_META_BUCKET, &lock_resource)
-        .await
-        .map_err(|err| RemoteScannerServerError::worker(format!("remote namespace scanner cache lock creation failed: {err}")))?;
-    let guard = ns_lock
-        .get_write_lock_quiet(scanner_cache_lock_timeout())
+    let guard = acquire_scanner_cache_locks(set.as_ref(), &cache_name, source)
         .await
         .map_err(|err| {
-            RemoteScannerServerError::worker(format!("remote namespace scanner cache lock acquisition failed: {err}"))
+            if err.is_contention() {
+                RemoteScannerServerError::retry_bucket(format!("remote namespace scanner cache lock contention: {err}"))
+            } else {
+                RemoteScannerServerError::worker(format!("remote namespace scanner cache lock acquisition failed: {err}"))
+            }
         })?;
     let mut cache = DataUsageCache::default();
     let revisions = cache.load_with_revisions(set.clone(), &cache_name).await.map_err(|err| {
@@ -1216,7 +1236,9 @@ fn finish_remote_scanner_stream(
             if !error.progress_fully_reported {
                 budget.cancel_after_unreported_remote_progress();
             }
-            let failure = if error.retire_worker {
+            let failure = if error.retry_bucket {
+                RemoteScannerFailure::retry_bucket(error.error)
+            } else if error.retire_worker {
                 RemoteScannerFailure::transport(error.error)
             } else {
                 RemoteScannerFailure::bucket(error.error)
@@ -1374,9 +1396,16 @@ where
                 return Ok(RemoteScannerOutcome::CycleAhead(required_cycle));
             }
             RemoteScannerFrameResult::Error(error_frame) => {
+                let retry_bucket = error_frame.message.starts_with(NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX);
+                let message = error_frame
+                    .message
+                    .strip_prefix(NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX)
+                    .map(str::trim_start)
+                    .unwrap_or(error_frame.message.as_str());
                 let error =
-                    StorageError::other(format!("remote namespace scanner failed: {}", limit_error_message(error_frame.message)));
+                    StorageError::other(format!("remote namespace scanner failed: {}", limit_error_message(message.to_string())));
                 return Err(match error_frame.scope {
+                    RemoteScannerErrorScope::Bucket if retry_bucket => RemoteScannerStreamError::retry_bucket(error),
                     RemoteScannerErrorScope::Bucket => RemoteScannerStreamError::bucket(error),
                     RemoteScannerErrorScope::Worker => RemoteScannerStreamError::reconciled(error),
                 });
@@ -2491,6 +2520,42 @@ mod tests {
         assert!(!budget.budget_elapsed());
     }
 
+    #[tokio::test]
+    async fn retry_bucket_error_terminal_frame_requeues_bucket_work() {
+        let request_id = Uuid::new_v4();
+        let writer_auth = FrameAuthenticator::for_test(request_id);
+        let reader_auth = FrameAuthenticator::for_test(request_id);
+        let (mut writer, reader) = tokio::io::duplex(4096);
+        tokio::spawn(async move {
+            let mut sequence = 0;
+            write_frame(
+                &mut writer,
+                &writer_auth,
+                &mut sequence,
+                &RemoteScannerFrame::terminal(
+                    RemoteScannerProgress::default(),
+                    RemoteScannerFrameResult::Error(RemoteScannerErrorFrame {
+                        scope: RemoteScannerErrorScope::Bucket,
+                        message: format!("{NS_SCANNER_RETRY_BUCKET_ERROR_PREFIX} cache lock contention"),
+                    }),
+                ),
+            )
+            .await
+            .expect("retry-bucket error terminal frame should write");
+        });
+
+        let parent = CancellationToken::new();
+        let budget = ScannerCycleBudget::new(&parent, ScannerCycleBudgetConfig::default());
+        let stream_result =
+            consume_remote_scanner_stream(reader, parent, budget.clone(), "bucket", TEST_SOURCE, TEST_PLAN_DIGEST, reader_auth)
+                .await;
+        let error = finish_remote_scanner_stream(stream_result, budget.as_ref()).expect_err("retry-bucket frame must fail");
+
+        assert!(error.retry_bucket_work());
+        assert!(!error.retire_worker());
+        assert!(error.to_string().contains("cache lock contention"));
+    }
+
     #[tokio::test]
     async fn worker_error_terminal_frame_retires_worker_without_cancelling_reported_budget() {
         let request_id = Uuid::new_v4();

crates/scanner/src/scanner.rs

@@ -949,7 +949,7 @@ pub(crate) async fn probe_scanner_activity(storeapi: &ECStore, distributed: bool
                 }
                 SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION => {
                     return Err(format!(
-                        "scanner activity peer {host} cannot acknowledge distributed dirty usage with protocol {}",
+                        "scanner activity peer {host} cannot safely share scanner cache locks with protocol {}",
                         SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION
                     ));
                 }
@@ -7114,9 +7114,17 @@ mod tests {
                 ..scanner_node_activity("epoch-a", 7, 3)
             },
         )]);
+        let previous = BTreeMap::from([(
+            "node-2".to_string(),
+            ScannerNodeActivity {
+                protocol_version: SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION,
+                ..scanner_node_activity("epoch-a", 7, 3)
+            },
+        )]);
         let current = BTreeMap::from([("node-2".to_string(), scanner_node_activity("epoch-a", 7, 3))]);
 
         assert_ne!(scanner_activity_snapshot_digest(&legacy), scanner_activity_snapshot_digest(&current));
+        assert_ne!(scanner_activity_snapshot_digest(&previous), scanner_activity_snapshot_digest(&current));
     }
 
     #[test]

crates/scanner/src/scanner_io.rs

@@ -30,6 +30,7 @@ use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, e
 use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS};
 use rustfs_data_usage::{BucketTargetUsageInfo, BucketUsageInfo};
 use rustfs_filemeta::FileMeta;
+use rustfs_lock::{LockError, NamespaceLockGuard};
 use rustfs_utils::path::path_join_buf;
 use s3s::dto::{BucketLifecycleConfiguration, ObjectLockConfiguration, ObjectLockEnabled, ReplicationConfiguration};
 use sha2::{Digest as _, Sha256};
@@ -944,14 +945,85 @@ fn checked_bucket_usage_info(entry: &DataUsageEntry) -> Option<BucketUsageInfo>
     Some(usage)
 }
 
-pub(crate) fn scanner_cache_lock_resource(cache_name: &str) -> String {
-    path_join_buf(&[crate::BUCKET_META_PREFIX, cache_name, SCANNER_CACHE_LOCK_SUFFIX])
+pub(crate) fn scanner_cache_lock_resource(cache_name: &str, source: DataUsageCacheSource) -> String {
+    let lock_name = format!("{SCANNER_CACHE_LOCK_SUFFIX}.pool-{}.set-{}", source.pool_index, source.set_index);
+    path_join_buf(&[crate::BUCKET_META_PREFIX, cache_name, &lock_name])
 }
 
 pub(crate) fn scanner_cache_lock_timeout() -> Duration {
     Duration::from_secs(rustfs_utils::get_env_u64("RUSTFS_LOCK_ACQUIRE_TIMEOUT", 5))
 }
 
+#[derive(Debug)]
+pub(crate) struct ScannerCacheLockGuards {
+    scoped: NamespaceLockGuard,
+}
+
+impl ScannerCacheLockGuards {
+    pub(crate) fn is_lock_lost(&self) -> bool {
+        self.scoped.is_lock_lost()
+    }
+}
+
+#[derive(Debug)]
+pub(crate) enum ScannerCacheLockError {
+    Create { resource: String, source: Error },
+    Acquire { resource: String, source: LockError },
+}
+
+impl ScannerCacheLockError {
+    pub(crate) fn state(&self) -> &'static str {
+        match self {
+            Self::Create { .. } => "lock_create_failed",
+            Self::Acquire { .. } => "lock_acquire_failed",
+        }
+    }
+
+    pub(crate) fn is_contention(&self) -> bool {
+        matches!(
+            self,
+            Self::Acquire {
+                source: LockError::Timeout { .. } | LockError::AlreadyLocked { .. },
+                ..
+            }
+        )
+    }
+}
+
+impl std::fmt::Display for ScannerCacheLockError {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Create { resource, source } => write!(formatter, "create scanner cache lock {resource}: {source}"),
+            Self::Acquire { resource, source } => write!(formatter, "acquire scanner cache lock {resource}: {source}"),
+        }
+    }
+}
+
+pub(crate) async fn acquire_scanner_cache_locks(
+    store: &SetDisks,
+    cache_name: &str,
+    source: DataUsageCacheSource,
+) -> std::result::Result<ScannerCacheLockGuards, ScannerCacheLockError> {
+    let timeout = scanner_cache_lock_timeout();
+    let scoped_resource = scanner_cache_lock_resource(cache_name, source);
+    let scoped_lock = store
+        .new_ns_lock(RUSTFS_META_BUCKET, &scoped_resource)
+        .await
+        .map_err(|source| ScannerCacheLockError::Create {
+            resource: scoped_resource.clone(),
+            source,
+        })?;
+    let scoped = scoped_lock
+        .get_write_lock_quiet(timeout)
+        .await
+        .map_err(|source| ScannerCacheLockError::Acquire {
+            resource: scoped_resource,
+            source,
+        })?;
+
+    Ok(ScannerCacheLockGuards { scoped })
+}
+
 async fn await_scanner_disk_shutdown<F>(scan: Pin<&mut F>)
 where
     F: Future,
@@ -1697,6 +1769,19 @@ mod publish_gate_tests {
         assert!(!cache_snapshot_is_current(&cache, "photos", source, 11, 0, second));
     }
 
+    #[test]
+    fn scanner_cache_lock_resource_is_scoped_to_cache_source() {
+        let cache_name = "photos/.usage-cache.bin";
+        let first_source = DataUsageCacheSource::new(0, 1);
+        let same_source = DataUsageCacheSource::new(0, 1);
+        let other_source = DataUsageCacheSource::new(1, 0);
+
+        let first = scanner_cache_lock_resource(cache_name, first_source);
+        assert_eq!(first, scanner_cache_lock_resource(cache_name, same_source));
+        assert_ne!(first, scanner_cache_lock_resource(cache_name, other_source));
+        assert!(first.ends_with(".scanner-cycle.lock.pool-0.set-1"));
+    }
+
     #[test]
     fn count_budget_serializes_set_and_disk_work() {
         assert_eq!(scanner_budgeted_concurrency_limit(8, true), 1);
@@ -1797,24 +1882,7 @@ async fn persist_and_publish_cache_snapshot(
     cache_cycle_floor: &AtomicU64,
 ) -> Option<SystemTime> {
     let source = cache_snapshot.info.source?;
-    let lock_resource = scanner_cache_lock_resource(DATA_USAGE_CACHE_NAME);
-    let ns_lock = match store.new_ns_lock(RUSTFS_META_BUCKET, &lock_resource).await {
-        Ok(lock) => lock,
-        Err(err) => {
-            error!(
-                target: "rustfs::scanner::io",
-                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                component = LOG_COMPONENT_SCANNER,
-                subsystem = LOG_SUBSYSTEM_IO,
-                cache_name = DATA_USAGE_CACHE_NAME,
-                state = "lock_create_failed",
-                error = %err,
-                "Scanner cache snapshot lock creation failed"
-            );
-            return None;
-        }
-    };
-    let guard = match ns_lock.get_write_lock_quiet(scanner_cache_lock_timeout()).await {
+    let guard = match acquire_scanner_cache_locks(store.as_ref(), DATA_USAGE_CACHE_NAME, source).await {
         Ok(guard) => guard,
         Err(err) => {
             error!(
@@ -1823,7 +1891,7 @@ async fn persist_and_publish_cache_snapshot(
                 component = LOG_COMPONENT_SCANNER,
                 subsystem = LOG_SUBSYSTEM_IO,
                 cache_name = DATA_USAGE_CACHE_NAME,
-                state = "lock_acquire_failed",
+                state = err.state(),
                 error = %err,
                 "Scanner cache snapshot lock acquisition failed"
             );
@@ -3080,32 +3148,35 @@ impl ScannerIOCache for SetDisks {
                         None
                     };
 
-                    // Lock order: scanner leader fence -> per-bucket cache lock ->
-                    // cache object read/write. The cache lock stays outermost for
-                    // local and rolling-upgrade workers so leader failover cannot
-                    // execute the same bucket concurrently.
-                    let lock_resource = scanner_cache_lock_resource(&cache_name);
-                    let ns_lock = match store_clone_clone.new_ns_lock(RUSTFS_META_BUCKET, &lock_resource).await {
-                        Ok(lock) => lock,
-                        Err(e) => {
-                            record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
-                            error!(
-                                target: "rustfs::scanner::io",
-                                event = EVENT_SCANNER_CACHE_PERSIST_STATE,
-                                component = LOG_COMPONENT_SCANNER,
-                                subsystem = LOG_SUBSYSTEM_IO,
-                                bucket = %bucket.name,
-                                cache_name = %cache_name,
-                                state = "lock_create_failed",
-                                error = %e,
-                                "Scanner bucket cache lock creation failed"
-                            );
-                            continue;
-                        }
-                    };
-                    let cache_guard = match ns_lock.get_write_lock_quiet(scanner_cache_lock_timeout()).await {
+                    // Lock order: scanner leader fence -> set-scoped per-bucket cache lock ->
+                    // cache object read/write.
+                    let cache_guard = match acquire_scanner_cache_locks(store_clone_clone.as_ref(), &cache_name, source).await {
                         Ok(guard) => guard,
                         Err(e) => {
+                            if e.is_contention() {
+                                if requeue_bucket_work(&bucket_tx_clone, &bucket, &mut work_guard).await {
+                                    increment_disk_bucket_scans_queued(
+                                        &queued_disk_bucket_scans_clone,
+                                        &pool_label_clone,
+                                        &set_label_clone,
+                                    );
+                                } else {
+                                    record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
+                                }
+                                debug!(
+                                    target: "rustfs::scanner::io",
+                                    event = EVENT_SCANNER_CACHE_PERSIST_STATE,
+                                    component = LOG_COMPONENT_SCANNER,
+                                    subsystem = LOG_SUBSYSTEM_IO,
+                                    bucket = %bucket.name,
+                                    cache_name = %cache_name,
+                                    state = "lock_contention_requeued",
+                                    error = %e,
+                                    "Scanner bucket cache lock contention requeued bucket work"
+                                );
+                                break;
+                            }
+
                             record_failed_dirty_bucket(&failed_dirty_buckets_clone, &bucket.name).await;
                             error!(
                                 target: "rustfs::scanner::io",
@@ -3114,7 +3185,7 @@ impl ScannerIOCache for SetDisks {
                                 subsystem = LOG_SUBSYSTEM_IO,
                                 bucket = %bucket.name,
                                 cache_name = %cache_name,
-                                state = "lock_acquire_failed",
+                                state = e.state(),
                                 error = %e,
                                 "Scanner bucket cache lock acquisition failed"
                             );
@@ -3894,6 +3965,52 @@ mod tests {
         (temp_dir, store)
     }
 
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cache_locks_block_same_source_workers() {
+        let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+        let set = &store.pools[0].disk_set[0];
+        let source = DataUsageCacheSource::new(0, 0);
+        let cache_name = "photos/.usage-cache.bin";
+
+        let guards = acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+            .await
+            .expect("scanner cache locks should be acquired");
+        let scoped_lock = set
+            .new_ns_lock(RUSTFS_META_BUCKET, &scanner_cache_lock_resource(cache_name, source))
+            .await
+            .expect("scoped scanner cache lock should be created");
+        let scoped_err = scoped_lock
+            .get_write_lock_quiet(Duration::from_millis(100))
+            .await
+            .expect_err("same-source workers must be blocked while scanner cache lock is held");
+        assert!(matches!(scoped_err, LockError::Timeout { .. } | LockError::AlreadyLocked { .. }));
+
+        drop(guards);
+        acquire_scanner_cache_locks(set.as_ref(), cache_name, source)
+            .await
+            .expect("scanner cache locks should be released when guards drop");
+    }
+
+    #[tokio::test]
+    #[serial]
+    async fn scanner_cache_locks_allow_cross_source_workers() {
+        let (_temp_dir, store) = setup_two_pool_scanner_store().await;
+        let first_set = &store.pools[0].disk_set[0];
+        let second_set = &store.pools[1].disk_set[0];
+        let cache_name = "photos/.usage-cache.bin";
+
+        let first = acquire_scanner_cache_locks(first_set.as_ref(), cache_name, DataUsageCacheSource::new(0, 0))
+            .await
+            .expect("first source scanner cache locks should be acquired");
+        let second = acquire_scanner_cache_locks(second_set.as_ref(), cache_name, DataUsageCacheSource::new(1, 0))
+            .await
+            .expect("different source scanner cache locks should not contend");
+
+        assert!(!first.is_lock_lost());
+        assert!(!second.is_lock_lost());
+    }
+
     #[tokio::test]
     async fn data_usage_publish_fails_when_receiver_is_closed() {
         let (updates, receiver) = mpsc::channel(1);

crates/storage-api/src/lib.rs

@@ -28,8 +28,8 @@ pub const NS_SCANNER_SESSION_SEQUENCE_QUERY: &str = "ns_scanner_session_sequence
 pub const NS_SCANNER_PROTOCOL_VERSION_QUERY: &str = "ns_scanner_protocol";
 pub const NS_SCANNER_PROTOCOL_VERSION: u16 = 3;
 pub const SCANNER_ACTIVITY_LEGACY_PROTOCOL_VERSION: u32 = 0;
-pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 4;
-pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 5;
+pub const SCANNER_ACTIVITY_PREVIOUS_PROTOCOL_VERSION: u32 = 5;
+pub const SCANNER_ACTIVITY_PROTOCOL_VERSION: u32 = 6;
 
 #[derive(Debug, serde::Deserialize, serde::Serialize)]
 #[serde(deny_unknown_fields)]