Merge branch 'main' into houseme/fix-backlog-1578-issue-5506

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

Cargo.lock

@@ -10027,6 +10027,7 @@ dependencies = [
  "rustfs-data-usage",
  "rustfs-ecstore",
  "rustfs-filemeta",
+ "rustfs-lock",
  "rustfs-storage-api",
  "rustfs-utils",
  "s3s",

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)]

docs/architecture/compat-cleanup-register.md

@@ -17,7 +17,7 @@ for later deletion.
 - `rustfs-5416-kubernetes-alias-dns` Kubernetes endpoint identity fallback: deployments created before explicit local endpoint identity may use resolvable aliases that do not match the Pod hostname. An implicit auto-mode zero match retains legacy DNS locality with a bounded deadline, while ambiguous matches and invalid explicit anchors still fail closed. Remove the fallback after every supported direct-upgrade chart and deployment manifest provides a canonical RUSTFS_LOCAL_ENDPOINT_HOST for domain-based distributed topologies.
 - `rustfs-5416-zero-retry-delay` startup retry-delay validation: releases before bounded topology convergence accept RUSTFS_STARTUP_TOPOLOGY_RETRY_MAX_DELAY values of 0 or 0ms. New servers replace those values with the safe nonzero default so a direct upgrade neither fails startup nor enters a busy loop. Reject zero after the minimum supported direct-upgrade release validates or rewrites this setting before rollout.
 - `scanner-usage-v2` persisted scanner usage migration: pre-v2 scanners write `.usage.json`, so upgraded clusters read that primary/backup pair only while `.usage.v2.json` is absent and continue removing deleted buckets from legacy copies that still exist. The additive usage_snapshot_complete field in `.usage.v2.json` must remain optional while mixed-version clusters are supported; a missing field means the snapshot is not authoritative. Remove the legacy object fallback and cleanup only after every supported direct-upgrade source writes `.usage.v2.json`.
-- `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Current protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while the distributed scanner publishes usage only after every peer returns authenticated protocol v5 state. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, and remove the protocol-v4 activity codec after every supported peer implements protocol v5; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
+- `ns-scanner-rpc-v3` namespace scanner capability and activity handshake: old peers and legacy internode transports lack the authenticated startup-epoch handshake. The oldest peers send an empty activity request and receive a field-empty protocol-0 response. Protocol v4 binds the challenge and response topology but cannot authenticate distributed dirty-usage state. Protocol v5 binds the request version, acknowledgement target and generation, and the response dirty-usage state, but predates set-scoped scanner cache locks. Current protocol v6 additionally fences scanner cache lock-domain changes, so distributed scanner cycles publish usage only after every peer reports protocol v6 state. Servers retain protocol-0 and protocol-v4 codecs for rolling upgrades, while protocol-v5 peers are treated as previous-version peers that cannot safely participate in the new cache lock domain. Scanner selection treats HTTP 404/405/426 and the legacy MethodNotAllowed default as an explicit lack of remote scanner v3 support and assigns those disks to coordinator-driven workers; transient capability failures remain incomplete and do not activate the fallback. Remove the coordinator fallback after the minimum supported RustFS peer version implements namespace scanner protocol v3, remove protocol-0 activity requests and responses after every supported peer implements authenticated scanner activity protocol v4, remove the protocol-v4 activity codec after every supported peer implements protocol v5, and remove protocol-v5 previous-version rejection after every supported peer implements protocol v6; future protocol revisions must keep the same dual-version server/codec window before changing the advertised version.
 - `#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.
 - `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
 - `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.