diff --git a/Cargo.lock b/Cargo.lock index 416cbd5c4..0a6595954 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -10027,6 +10027,7 @@ dependencies = [ "rustfs-data-usage", "rustfs-ecstore", "rustfs-filemeta", + "rustfs-lock", "rustfs-storage-api", "rustfs-utils", "s3s", diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index 127604853..9b5aa3342 100644 --- a/crates/scanner/Cargo.toml +++ b/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 } diff --git a/crates/scanner/src/remote_scanner.rs b/crates/scanner/src/remote_scanner.rs index 0b44ad6bc..bb8236c5b 100644 --- a/crates/scanner/src/remote_scanner.rs +++ b/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) -> 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, source); - 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(); diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 95c5fb194..cddea3785 100644 --- a/crates/scanner/src/scanner.rs +++ b/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(¤t)); + assert_ne!(scanner_activity_snapshot_digest(&previous), scanner_activity_snapshot_digest(¤t)); } #[test] diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 68d44e54b..a0f0c2a66 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/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}; @@ -953,6 +954,76 @@ 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 { + 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(scan: Pin<&mut F>) where F: Future, @@ -1811,24 +1882,7 @@ async fn persist_and_publish_cache_snapshot( cache_cycle_floor: &AtomicU64, ) -> Option { let source = cache_snapshot.info.source?; - let lock_resource = scanner_cache_lock_resource(DATA_USAGE_CACHE_NAME, source); - 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!( @@ -1837,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" ); @@ -3095,31 +3149,34 @@ impl ScannerIOCache for SetDisks { }; // Lock order: scanner leader fence -> set-scoped 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, source); - 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 { + // 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", @@ -3128,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" ); @@ -3908,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); diff --git a/crates/storage-api/src/lib.rs b/crates/storage-api/src/lib.rs index 1747b3424..d4a73c10a 100644 --- a/crates/storage-api/src/lib.rs +++ b/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)] diff --git a/docs/architecture/compat-cleanup-register.md b/docs/architecture/compat-cleanup-register.md index 1bdc236a3..776773cb2 100644 --- a/docs/architecture/compat-cleanup-register.md +++ b/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.