refactor(replication): split four oversized hot-path functions into focused helpers

Pure-move decomposition of the four oversized functions flagged by the
replication compatibility review (P1-18), unblocking migration milestone
M2 which requires resyncer moves to stay mechanical:

- resync_bucket (522 lines -> 61-line step sequence): leader lock,
  target resolution, walk/collector/worker spawning, and dispatch loop
  extracted into focused helpers; pure decision helpers (DTO builders,
  HEAD-result classification) separated from IO orchestration.
- replicate_all (411 lines -> 113-line main body): initial target-info
  seeding, read/stat option builders, skip-path notes, target HEAD
  action resolution, and the multipart/single-put payload transport
  extracted as private free functions.
- start_mrf_processor (306 lines -> 46-line spawn body): recovery guard,
  ledger load, per-entry replay (delete/object/metadata), and retained
  entry resolution extracted; retry bookkeeping semantics preserved
  exactly (inner continue-paths push inside helpers, outer Missed push
  stays in the loop).
- apply_iam_item (255 lines -> match dispatch skeleton): one helper per
  IAM item type.

No behavior change: log texts, error paths, event emissions, and metric
counts are byte-identical; existing tests unchanged and green (238
ecstore replication/mrf/resync + 232 rustfs site-replication).

crates/checksums/src/base64.rs

@@ -11,7 +11,6 @@
 // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 // See the License for the specific language governing permissions and
 // limitations under the License.
-#![allow(dead_code)]
 
 use base64_simd::STANDARD;
 

crates/checksums/src/http.rs

@@ -38,7 +38,10 @@ pub const XXHASH_3_HEADER_NAME: &str = "x-amz-checksum-xxhash3";
 pub const XXHASH_64_HEADER_NAME: &str = "x-amz-checksum-xxhash64";
 pub const XXHASH_128_HEADER_NAME: &str = "x-amz-checksum-xxhash128";
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 wire name, resolved by header_name() below and asserted by this crate's tests (backlog#1823)"
+)]
 pub(crate) static MD5_HEADER_NAME: &str = "content-md5";
 
 pub const CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER: [&str; 5] =

crates/checksums/src/lib.rs

@@ -476,13 +476,19 @@ impl Checksum for Xxhash64 {
     }
 }
 
-#[allow(dead_code)]
 #[derive(Debug, Default)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
+)]
 struct Md5 {
     hasher: md5::Md5,
 }
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "Content-MD5 is not a ChecksumAlgorithm variant and has no arm in into_impl: S3 carries it as its own header, separate from the x-amz-checksum-* family. This impl exists so the two paths share the Checksum trait, and is asserted by this crate's tests (backlog#1823)"
+)]
 impl Md5 {
     fn update(&mut self, bytes: &[u8]) {
         use md5::Digest;

crates/ecstore/src/bucket/bucket_target_sys.rs

@@ -82,7 +82,6 @@ use tracing::warn;
 use url::Url;
 use uuid::Uuid;
 
-const DEFAULT_HEALTH_CHECK_RELOAD_DURATION: Duration = Duration::from_secs(30 * 60);
 const MAX_CONCURRENT_TARGET_HEALTH_CHECKS: usize = 16;
 const REDACTED_CREDENTIAL: &str = "<redacted>";
 

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -126,11 +126,23 @@ const EVENT_LIFECYCLE_EXPIRED_DETECTED: &str = "lifecycle_expired_detected";
 const EVENT_LIFECYCLE_NOT_ENQUEUED: &str = "lifecycle_not_enqueued";
 const EVENT_LIFECYCLE_DELETE_DISPATCHED: &str = "lifecycle_delete_dispatched";
 const EVENT_LIFECYCLE_DELETE_COMPLETED: &str = "lifecycle_delete_completed";
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 const EVENT_LIFECYCLE_TIER_AUDIT: &str = "lifecycle_tier_audit";
 const EVENT_LIFECYCLE_TIER_OPERATION_FAILED: &str = "lifecycle_tier_operation_failed";
 const EVENT_LIFECYCLE_DELETE_FAILED: &str = "lifecycle_delete_failed";
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub type TimeFn = Arc<dyn Fn() -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub type TraceFn =
     Arc<dyn Fn(String, HashMap<String, String>) -> Pin<Box<dyn Future<Output = ()> + Send>> + Send + Sync + 'static>;
 pub type ExpiryOpType = Box<dyn ExpiryOp + Send + Sync + 'static>;
@@ -140,9 +152,21 @@ static TIER_FREE_VERSION_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 static MANUAL_TRANSITION_JOB_RECOVERY_STARTED: OnceLock<()> = OnceLock::new();
 
 pub const AMZ_OBJECT_TAGGING: &str = "X-Amz-Tagging";
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub const AMZ_TAG_COUNT: &str = "x-amz-tagging-count";
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub const AMZ_TAG_DIRECTIVE: &str = "X-Amz-Tagging-Directive";
 pub const AMZ_ENCRYPTION_AES: &str = "AES256";
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub const AMZ_ENCRYPTION_KMS: &str = "aws:kms";
 
 pub const ERR_INVALID_STORAGECLASS: &str = "invalid tier.";
@@ -280,6 +304,10 @@ impl LifecycleSys {
         }
     }
 
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+    )]
     pub fn trace(oi: &ObjectInfo) -> TraceFn {
         let bucket = oi.bucket.clone();
         let name = oi.name.clone();
@@ -570,6 +598,10 @@ async fn delete_free_version_remote_object(
     Ok(())
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 async fn delete_free_version_remote_object_then<T, F, Fut>(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
@@ -2868,6 +2900,10 @@ fn stale_upload_default_due(initiated: OffsetDateTime, default_expiry: StdDurati
     initiated + time::Duration::seconds(default_expiry.as_secs() as i64)
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 async fn stale_upload_current_size(set: &Arc<SetDisks>, metadata: &HashMap<String, String>, upload_dir: &str) -> Option<usize> {
     stale_upload_current_size_with_opts(set, metadata, upload_dir, false).await
 }
@@ -3352,6 +3388,10 @@ pub async fn validate_transition_tier(lc: &BucketLifecycleConfiguration) -> Resu
     Ok(())
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 fn mark_delete_opts_skip_decommissioned_on_remote_success(opts: &mut ObjectOptions, remote_delete_succeeded: bool) {
     if remote_delete_succeeded {
         opts.skip_decommissioned = true;
@@ -4339,6 +4379,10 @@ pub async fn expire_transitioned_object(
     Ok(dobj)
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub fn gen_transition_objname(bucket: &str) -> Result<String, Error> {
     let us = Uuid::new_v4().to_string();
     let mut hasher = Sha256::new();
@@ -4373,6 +4417,10 @@ pub async fn transition_object(api: Arc<ECStore>, oi: &ObjectInfo, lae: LcAuditE
     result
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     let tier = _tier.to_string();
     Arc::new(move || {
@@ -4391,6 +4439,10 @@ pub fn audit_tier_actions(_tier: &str, bytes: i64) -> TimeFn {
     })
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn get_transitioned_object_reader(
     bucket: &str,
     object: &str,
@@ -5145,6 +5197,10 @@ async fn lifecycle_delete_config_snapshot(api: &ECStore, oi: &ObjectInfo) -> Res
     ReplicationObjectBridge::delete_request_config(api, &oi.bucket).await
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool {
     let mut success = false;
     match event.action {
@@ -7422,6 +7478,10 @@ mod tests {
     // process environment while `env::set_var`/`env::remove_var` is active.
     // SAFETY: keep this note adjacent to the allowance for the repository guard.
     #[allow(unsafe_code)]
+    #[allow(
+        dead_code,
+        reason = "transition-queue env fixture kept for tests that scope those vars; no test uses it today (backlog#1823)"
+    )]
     async fn with_transition_queue_env_async<F, Fut>(capacity: Option<&str>, timeout_ms: Option<&str>, test_fn: F)
     where
         F: FnOnce() -> Fut,

crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs

@@ -759,6 +759,10 @@ pub struct ManualTransitionWorkerResultRecord {
 }
 
 impl ManualTransitionWorkerResultRecord {
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+    )]
     pub fn new(job_id: Uuid, task_key: impl Into<String>, result: ManualTransitionWorkerResult) -> Self {
         Self::new_with_reason(job_id, task_key, result, None)
     }
@@ -1257,6 +1261,10 @@ pub(crate) async fn save_manual_transition_task_if_absent(
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn load_manual_transition_task_record(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1320,6 +1328,10 @@ async fn scan_manual_transition_task_journal(api: Arc<ECStore>, job_id: Uuid) ->
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn load_manual_transition_worker_result_stats(
     api: Arc<ECStore>,
     job_id: Uuid,
@@ -1455,6 +1467,10 @@ async fn scan_manual_transition_worker_result_journal(
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn reconcile_manual_transition_worker_results(
     api: Arc<ECStore>,
     job_id: Uuid,

crates/ecstore/src/bucket/lifecycle/replication_sink.rs

@@ -15,25 +15,35 @@
 use rustfs_common::metrics::IlmAction;
 
 use crate::bucket::lifecycle::lifecycle::ObjectOpts;
+use crate::bucket::replication::ReplicationLifecycleBridge;
 pub(crate) use crate::bucket::replication::ReplicationStatusType;
 #[cfg(test)]
 pub(crate) use crate::bucket::replication::VersionPurgeStatusType;
 pub(crate) use crate::bucket::replication::{
     DeleteReplicationConfigSnapshot, ReplicationObjectBridge, replication_state_to_filemeta,
 };
-use crate::bucket::replication::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
 use crate::storage_api_contracts::object::DeletedObject;
 
-pub(crate) type LifecycleReplicationConfig = ReplicationLifecycleConfig;
-
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) fn has_pending_version_purge(obj: &ObjectOpts) -> bool {
     obj.version_purge_status.is_pending()
 }
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) fn has_pending_object_replication(obj: &ObjectOpts) -> bool {
     replication_status_blocks_lifecycle(&obj.replication_status)
 }
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) fn has_pending_lifecycle_replication(obj: &ObjectOpts) -> bool {
     has_pending_object_replication(obj) || has_pending_version_purge(obj)
 }

crates/ecstore/src/bucket/lifecycle/tagging_boundary.rs

@@ -14,6 +14,10 @@
 
 use std::collections::HashMap;
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) fn decode_tags_to_map(tags: &str) -> HashMap<String, String> {
     crate::bucket::tagging::decode_tags_to_map(tags)
 }

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -331,6 +331,10 @@ where
     persist_tier_delete_journal_entry(api, &committed).await
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn abort_tier_delete_journal_entry<S>(api: Arc<S>, je: &Jentry) -> std::io::Result<()>
 where
     S: ObjectOperations<

crates/ecstore/src/bucket/lifecycle/tier_free_version_recovery.rs

@@ -148,6 +148,10 @@ struct RecoveryCursor {
     object: String,
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn recover_tier_free_versions(
     api: Arc<ECStore>,
     limit: usize,

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -385,6 +385,10 @@ impl ExpiryOp for Jentry {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     let result = delete_object_from_remote_tier_raw(obj_name, rv_id, tier_name).await;
     if let Err(err) = &result
@@ -395,6 +399,10 @@ pub async fn delete_object_from_remote_tier(obj_name: &str, rv_id: &str, tier_na
     result
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_name: &str) -> Result<(), std::io::Error> {
     #[cfg(test)]
     if let Some(result) = run_remote_tier_delete_test_hook(obj_name, rv_id, tier_name) {
@@ -405,6 +413,10 @@ async fn delete_object_from_remote_tier_raw(obj_name: &str, rv_id: &str, tier_na
     delete_object_from_remote_tier_raw_with_manager(obj_name, rv_id, tier_name, &tier_config_mgr).await
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 async fn delete_object_from_remote_tier_raw_with_manager(
     obj_name: &str,
     rv_id: &str,
@@ -485,6 +497,10 @@ pub enum RemoteTierDeleteOutcome {
     AlreadyRemoved,
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+)]
 pub async fn delete_object_from_remote_tier_idempotent(
     obj_name: &str,
     rv_id: &str,

crates/ecstore/src/bucket/lifecycle/transition_transaction.rs

@@ -50,8 +50,16 @@ pub type Result<T> = std::result::Result<T, TransitionTransactionError>;
 #[derive(Debug, thiserror::Error)]
 pub enum TransitionTransactionError {
     #[error("transition transaction already exists")]
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+    )]
     AlreadyExists,
     #[error("transition transaction is not found")]
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity tier/lifecycle entry point that this port never wired (backlog#1823)"
+    )]
     NotFound,
     #[error("transition transaction is corrupt: {0}")]
     Corrupt(&'static str),

crates/ecstore/src/bucket/metadata_sys.rs

@@ -60,12 +60,14 @@ struct ConfigWriteLockProbeState {
 static CONFIG_WRITE_LOCK_PROBES: std::sync::OnceLock<StdMutex<Vec<Arc<ConfigWriteLockProbeState>>>> = std::sync::OnceLock::new();
 
 #[cfg(any(test, feature = "test-util"))]
+#[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
 pub struct ConfigWriteLockProbe {
     state: Arc<ConfigWriteLockProbeState>,
 }
 
 #[cfg(any(test, feature = "test-util"))]
 impl ConfigWriteLockProbe {
+    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub fn install(bucket: &str) -> Self {
         let state = Arc::new(ConfigWriteLockProbeState {
             bucket: bucket.to_string(),
@@ -84,6 +86,7 @@ impl ConfigWriteLockProbe {
         Self { state }
     }
 
+    #[allow(dead_code, reason = "installed by tests behind `--features test-util` (backlog#1823)")]
     pub async fn wait_until_attempted(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
@@ -890,6 +893,10 @@ pub async fn get_bucket_policy_raw(bucket: &str) -> Result<(String, OffsetDateTi
     bucket_meta_sys.get_bucket_policy_raw(bucket).await
 }
 
+#[allow(
+    dead_code,
+    reason = "free-function facade over the live BucketMetadataSys::get_bucket_acl_config; no caller in this port (backlog#1823)"
+)]
 pub async fn get_bucket_acl_config(bucket: &str) -> Result<(String, OffsetDateTime)> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1104,6 +1111,10 @@ pub async fn get_config_from_disk(bucket: &str) -> Result<BucketMetadata> {
     bucket_meta_sys.get_config_from_disk(bucket).await
 }
 
+#[allow(
+    dead_code,
+    reason = "ambient-facade variant of the live created_at_in; no caller in this port (backlog#1823)"
+)]
 pub async fn created_at(bucket: &str) -> Result<OffsetDateTime> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let bucket_meta_sys = bucket_meta_sys_lock.read().await;
@@ -1617,6 +1628,7 @@ impl BucketMetadataSys {
     /// [`Self::update`], with the payload computed from the loaded metadata
     /// instead of supplied up front. Loads through this system's own store so
     /// the read and the persisted write target the same instance.
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn update_config_with<F>(&self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
     where
         F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
@@ -1721,6 +1733,7 @@ impl BucketMetadataSys {
     /// A miss is never published as an authoritative default, and a snapshot
     /// read before delete plus same-name recreation cannot replace the new
     /// generation.
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(crate) async fn reload_from_store(&self, bucket: &str) -> Result<()> {
         if is_meta_bucketname(bucket) {
             return Err(Error::other("errInvalidArgument"));

crates/ecstore/src/bucket/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: bucket subsystems still contain staged ECStore migration code.
-#![allow(dead_code)]
 
 pub mod bandwidth;
 pub mod bucket_target_sys;

crates/ecstore/src/bucket/object_lock/objectlock_sys.rs

@@ -136,6 +136,7 @@ pub fn add_years(dt: OffsetDateTime, years: i32) -> OffsetDateTime {
 
 /// Check if an object has legal hold enabled.
 /// Returns true if legal hold is ON.
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> bool {
     let lhold = objectlock::get_object_legalhold_meta(user_defined);
     matches!(lhold.status, Some(ref st) if st.as_str() == ObjectLockLegalHoldStatus::ON)
@@ -151,6 +152,7 @@ fn has_legal_hold(user_defined: &std::collections::HashMap<String, String>) -> b
 /// # Returns
 /// * `true` if the object is locked (cannot be deleted/modified)
 /// * `false` if the object is not locked
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_object_locked_by_metadata(user_defined: &std::collections::HashMap<String, String>, is_delete_marker: bool) -> bool {
     // Delete markers are never locked
     if is_delete_marker {

crates/ecstore/src/bucket/quota/mod.rs

@@ -193,6 +193,7 @@ pub enum QuotaError {
 }
 
 #[derive(Debug, Serialize)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub struct QuotaErrorResponse {
     #[serde(rename = "Code")]
     pub code: String,
@@ -208,6 +209,7 @@ pub struct QuotaErrorResponse {
 }
 
 impl QuotaErrorResponse {
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn new(quota_error: &QuotaError, request_id: &str, host_id: &str) -> Self {
         match quota_error {
             QuotaError::QuotaExceeded { .. } => Self {

crates/ecstore/src/bucket/quota/reservation.rs

@@ -899,6 +899,7 @@ async fn save_ledger_locked(
 }
 
 #[cfg(any(test, feature = "test-util"))]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn fail_next_quota_ledger_save_for_test() {
     FAIL_NEXT_LEDGER_SAVE.store(true, std::sync::atomic::Ordering::SeqCst);
 }

crates/ecstore/src/bucket/replication/mod.rs

@@ -60,7 +60,7 @@ pub use replication_filemeta_boundary::{
 pub(crate) use replication_filemeta_boundary::{
     replication_state_from_filemeta, replication_status_from_filemeta, version_purge_status_from_filemeta,
 };
-pub(crate) use replication_lifecycle_bridge::{ReplicationLifecycleBridge, ReplicationLifecycleConfig};
+pub(crate) use replication_lifecycle_bridge::ReplicationLifecycleBridge;
 pub(crate) use replication_migration_bridge::ReplicationMigrationBridge;
 pub use replication_object_bridge::ReplicationObjectBridge;
 pub use replication_object_config::{DeleteReplicationConfigSnapshot, ReplicationConfig};

crates/ecstore/src/bucket/replication/replication_config_store.rs

@@ -37,6 +37,10 @@ impl ReplicationConfigStore {
         com::read_config_limited(api, file, max_bytes).await
     }
 
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     pub(crate) async fn read_no_lock<S>(api: Arc<S>, file: &str) -> Result<Vec<u8>>
     where
         S: ReplicationObjectIO,

crates/ecstore/src/bucket/replication/replication_lifecycle_bridge.rs

@@ -24,15 +24,27 @@ use super::replication_storage_boundary::{
     DeletedObject, ObjectInfo, ObjectOptions, ObjectToDelete, deleted_object_for_replication,
 };
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) type ReplicationLifecycleConfig = ReplicationConfig;
 
 pub(crate) struct ReplicationLifecycleBridge;
 
 impl ReplicationLifecycleBridge {
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn new_config(config: ReplicationConfiguration) -> ReplicationLifecycleConfig {
         ReplicationConfig::new(Some(config), None)
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn has_pending_version_purge(
         config: &ReplicationLifecycleConfig,
         object_name: &str,
@@ -45,6 +57,10 @@ impl ReplicationLifecycleBridge {
                 .is_some_and(|config| config.has_active_rules(object_name, true))
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) async fn check_delete_replication(
         bucket: &str,
         object: &ObjectToDelete,
@@ -54,6 +70,10 @@ impl ReplicationLifecycleBridge {
         check_replicate_delete(bucket, object, source, opts, None).await
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn version_delete_replication_state(decision: &ReplicateDecision) -> ReplicationState {
         let pending_status = decision.pending_status();
         ReplicationState {

crates/ecstore/src/bucket/replication/replication_msgp_boundary.rs

@@ -19,17 +19,33 @@ use time::OffsetDateTime;
 use super::replication_error_boundary::Result;
 use crate::bucket::msgp_decode;
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) struct ReplicationMsgpCodec;
 
 impl ReplicationMsgpCodec {
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn read_ext8_time<R: Read>(rd: &mut R) -> Result<OffsetDateTime> {
         msgp_decode::read_msgp_ext8_time(rd)
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn skip_value<R: Read>(rd: &mut R) -> Result<()> {
         msgp_decode::skip_msgp_value(rd)
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn write_time<W: Write>(wr: &mut W, time: OffsetDateTime) -> Result<()> {
         msgp_decode::write_msgp_time(wr, time)
     }

crates/ecstore/src/bucket/replication/replication_object_bridge.rs

@@ -77,6 +77,10 @@ impl ReplicationObjectBridge {
         load_delete_request_config_in(ctx, bucket).await
     }
 
+    #[allow(
+        dead_code,
+        reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+    )]
     pub(crate) async fn delete_config_snapshot_in(
         ctx: &ReplicationInstanceContext,
         bucket: &str,

crates/ecstore/src/bucket/replication/replication_object_config.rs

@@ -231,6 +231,10 @@ pub(crate) async fn load_delete_replication_config(
     delete_snapshot_from_metadata(ReplicationMetadataStore::delete_metadata(bucket).await?)
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+)]
 pub(crate) async fn load_delete_replication_config_in(
     ctx: &ReplicationInstanceContext,
     bucket: &str,

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -217,6 +217,10 @@ impl DurableMrfBacklogTracker {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+)]
 fn durable_mrf_backlog_tracker_from_entries(entries: &[MrfReplicateEntry]) -> DurableMrfBacklogTracker {
     let mut tracker = DurableMrfBacklogTracker {
         available: true,
@@ -663,6 +667,368 @@ async fn acknowledge_mrf_recovery<S: ReplicationStorage>(
     Err(EcstoreError::PreconditionFailed)
 }
 
+/// Acquires the MRF recovery leader lock for the startup replay.
+/// Returns `None` (after logging) when the lock cannot be created or another
+/// node is already processing the backlog.
+async fn acquire_mrf_recovery_guard<S: ReplicationStorage>(storage: &Arc<S>) -> Option<rustfs_lock::NamespaceLockGuard> {
+    let recovery_lock = match storage
+        .new_ns_lock(
+            ReplicationMetadataStore::rustfs_meta_bucket(),
+            ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
+        )
+        .await
+    {
+        Ok(lock) => lock,
+        Err(error) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %error,
+                "Failed to create the MRF recovery leader lock"
+            );
+            return None;
+        }
+    };
+    match recovery_lock
+        .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
+        .await
+    {
+        Ok(guard) => Some(guard),
+        Err(_) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                "Another node is already processing the MRF recovery backlog"
+            );
+            None
+        }
+    }
+}
+
+/// Reads and decodes the on-disk MRF recovery file.
+/// Returns `None` when there is nothing to replay: missing file (publishes an
+/// empty available summary), read failure, or corrupt data (quarantined).
+async fn load_mrf_recovery_entries<S: ReplicationStorage>(storage: &Arc<S>) -> Option<Vec<MrfReplicateEntry>> {
+    let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
+        Ok(d) => d,
+        Err(EcstoreError::ConfigNotFound) => {
+            set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
+                available: true,
+                buckets: Vec::new(),
+            });
+            return None;
+        }
+        Err(e) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %e,
+                "Failed to load MRF recovery file"
+            );
+            return None;
+        }
+    };
+
+    match decode_mrf_file(&data) {
+        Ok(v) => Some(v),
+        Err(e) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %e,
+                "Failed to decode MRF recovery file — preserving corrupt data"
+            );
+            quarantine_mrf_file(storage, &data).await;
+            None
+        }
+    }
+}
+
+/// Replays one MRF recovery entry by operation kind.
+/// Returns `None` when the entry is skipped entirely (no admission outcome);
+/// entries that must be retried later are pushed onto `retry_entries`.
+async fn replay_mrf_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    match entry.op {
+        MrfOpKind::Delete => replay_mrf_delete_entry(entry, storage, retry_entries).await,
+        MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
+            replay_mrf_object_entry(entry, storage, retry_entries).await
+        }
+        MrfOpKind::Metadata => replay_mrf_metadata_entry(entry, storage, retry_entries).await,
+    }
+}
+
+/// Replays a delete-kind MRF entry: force-delete intents replay directly,
+/// stale force-delete generations are skipped, and plain deletes are
+/// reconstructed as heal deletes.
+async fn replay_mrf_delete_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    if should_replay_force_delete_intent(entry) {
+        let operation_id = entry.force_delete_id?;
+        let delete = force_delete_heal_replication_info(entry, operation_id);
+        if replicate_delete_with_outcome(delete, storage.clone()).await {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    } else if entry.force_delete_id.is_some() {
+        Some(ReplicationQueueAdmission::Skipped)
+    } else {
+        replay_mrf_reconstructed_delete(entry, storage, retry_entries).await
+    }
+}
+
+/// Pure DTO construction: heal replication info for a replayed force-delete intent.
+fn force_delete_heal_replication_info(entry: &MrfReplicateEntry, operation_id: uuid::Uuid) -> DeletedObjectReplicationInfo {
+    DeletedObjectReplicationInfo {
+        delete_object: ReplicationDeletedObject {
+            object_name: entry.object.clone(),
+            force_delete: true,
+            force_delete_id: Some(operation_id),
+            force_delete_target_arns: entry.target_arns.clone(),
+            force_delete_generation: entry.force_delete_generation,
+            ..Default::default()
+        },
+        bucket: entry.bucket.clone(),
+        op_type: ReplicationType::Heal,
+        event_type: REPLICATE_HEAL_DELETE.to_string(),
+        ..Default::default()
+    }
+}
+
+/// Reconstruct a heal delete and re-queue it.  We do NOT call
+/// get_object_info here because the delete-marker or version may
+/// already be absent from the local store — that is expected.
+async fn replay_mrf_reconstructed_delete<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
+    let oi = ObjectInfo {
+        bucket: entry.bucket.clone(),
+        name: entry.object.clone(),
+        version_id: entry.version_id,
+        delete_marker: entry.delete_marker,
+        ..Default::default()
+    };
+    let dsc = resolve_mrf_delete_replicate_decision(entry, &oi, versioned, retry_entries).await?;
+    let dv = reconstructed_heal_delete_info(entry, &oi, &dsc);
+    if replicate_delete_with_outcome(dv, storage.clone()).await {
+        Some(ReplicationQueueAdmission::Queued)
+    } else {
+        Some(ReplicationQueueAdmission::Missed)
+    }
+}
+
+/// The MRF entry does not persist the replication decision and the
+/// source object is gone, so re-derive the decision from the live
+/// bucket config (mirroring get_heal_replicate_object_info) and set
+/// it on the reconstructed delete. Without this the decision string
+/// is empty and the delete replicates to zero targets — a silent
+/// no-op that leaves replicas diverged (backlog#858 / #799 B9).
+async fn resolve_mrf_delete_replicate_decision(
+    entry: &MrfReplicateEntry,
+    oi: &ObjectInfo,
+    versioned: bool,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicateDecision> {
+    if entry.target_arns.is_empty() {
+        match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
+            Ok(None) => None,
+            Err(_) => {
+                retry_entries.push(entry.clone());
+                None
+            }
+            Ok(Some(_)) => match check_replicate_delete_strict(
+                &entry.bucket,
+                &ObjectToDelete {
+                    object_name: entry.object.clone(),
+                    version_id: entry.version_id,
+                    ..Default::default()
+                },
+                oi,
+                &ObjectOptions {
+                    versioned,
+                    ..Default::default()
+                },
+                None,
+            )
+            .await
+            {
+                Ok(dsc) => Some(dsc),
+                Err(_) => {
+                    retry_entries.push(entry.clone());
+                    None
+                }
+            },
+        }
+    } else {
+        Some(replicate_decision_for_admitted_targets(&entry.target_arns))
+    }
+}
+
+/// Pure DTO construction: reconstructed heal delete carrying the re-derived
+/// replication decision.
+fn reconstructed_heal_delete_info(
+    entry: &MrfReplicateEntry,
+    oi: &ObjectInfo,
+    dsc: &ReplicateDecision,
+) -> DeletedObjectReplicationInfo {
+    let mut rstate = oi.replication_state();
+    rstate.replicate_decision_str = dsc.to_string();
+
+    let delete_marker_mtime = entry
+        .delete_marker_mtime
+        .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
+
+    DeletedObjectReplicationInfo {
+        delete_object: ReplicationDeletedObject {
+            object_name: entry.object.clone(),
+            version_id: entry.version_id,
+            delete_marker_version_id: entry.delete_marker_version_id,
+            delete_marker: entry.delete_marker,
+            delete_marker_mtime,
+            force_delete: entry.force_delete,
+            replication_state: Some(rstate),
+            ..Default::default()
+        },
+        bucket: entry.bucket.clone(),
+        op_type: ReplicationType::Heal,
+        event_type: REPLICATE_HEAL_DELETE.to_string(),
+        ..Default::default()
+    }
+}
+
+/// Replays an Object/Heal/ExistingObject MRF entry against the live source object.
+async fn replay_mrf_object_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let opts = ObjectOptions {
+        version_id: entry.version_id.map(|u| u.to_string()),
+        ..Default::default()
+    };
+    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+        Ok(oi) => oi,
+        Err(e) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %entry.bucket,
+                object = %entry.object,
+                error = %e,
+                "MRF recovery: source object lookup failed"
+            );
+            if should_retry_mrf_source_lookup(&e) {
+                retry_entries.push(entry.clone());
+            }
+            return None;
+        }
+    };
+    if entry.target_arns.is_empty() {
+        // Legacy entries predate target admission persistence. They cannot
+        // be safely attributed, so retain the old live-config fallback.
+        Some(queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
+    } else {
+        let roi = admitted_mrf_replicate_object(oi, entry, entry.op.replication_type());
+        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    }
+}
+
+/// Replays a metadata-kind MRF entry against the live source object.
+async fn replay_mrf_metadata_entry<S: ReplicationStorage>(
+    entry: &MrfReplicateEntry,
+    storage: &Arc<S>,
+    retry_entries: &mut Vec<MrfReplicateEntry>,
+) -> Option<ReplicationQueueAdmission> {
+    let opts = ObjectOptions {
+        version_id: entry.version_id.map(|u| u.to_string()),
+        ..Default::default()
+    };
+    let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
+        Ok(oi) => oi,
+        Err(e) => {
+            debug!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                bucket = %entry.bucket,
+                object = %entry.object,
+                error = %e,
+                "MRF metadata recovery: source object lookup failed"
+            );
+            if should_retry_mrf_source_lookup(&e) {
+                retry_entries.push(entry.clone());
+            }
+            return None;
+        }
+    };
+    if entry.target_arns.is_empty() {
+        Some(queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await)
+    } else {
+        let roi = admitted_mrf_replicate_object(oi, entry, ReplicationType::Metadata);
+        if replicate_object_with_outcome(roi, storage.clone()).await.1 {
+            Some(ReplicationQueueAdmission::Queued)
+        } else {
+            Some(ReplicationQueueAdmission::Missed)
+        }
+    }
+}
+
+/// Pure DTO construction: replicate-object info for an entry with persisted
+/// admitted targets, carrying over the entry's retry count.
+fn admitted_mrf_replicate_object(oi: ObjectInfo, entry: &MrfReplicateEntry, op_type: ReplicationType) -> ReplicateObjectInfo {
+    let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
+    let mut roi = replicate_object_info_from_object_info(oi, dsc, op_type);
+    roi.retry_count = entry.retry_count.max(0) as u32;
+    roi
+}
+
+/// Acknowledges the replayed MRF prefix and returns the retained backlog.
+/// On acknowledgement failure the backlog is preserved for the next startup and
+/// re-read (falling back to the replayed snapshot) so the published summary stays accurate.
+async fn resolve_retained_mrf_entries<S: ReplicationStorage>(
+    storage: &Arc<S>,
+    recovery_guard: &rustfs_lock::NamespaceLockGuard,
+    entries: &[MrfReplicateEntry],
+    retry_entries: &[MrfReplicateEntry],
+) -> Vec<MrfReplicateEntry> {
+    match acknowledge_mrf_recovery(storage.clone(), recovery_guard, entries, retry_entries).await {
+        Ok(retained) => retained,
+        Err(error) => {
+            warn!(
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                error = %error,
+                "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
+            );
+            match read_mrf_entries(storage.clone()).await {
+                Ok(current) => current,
+                Err(read_error) => {
+                    warn!(
+                        component = LOG_COMPONENT_ECSTORE,
+                        subsystem = LOG_SUBSYSTEM_REPLICATION,
+                        error = %read_error,
+                        "Failed to refresh the MRF backlog after acknowledgement failure"
+                    );
+                    entries.to_vec()
+                }
+            }
+        }
+    }
+}
+
 #[derive(Debug, thiserror::Error)]
 #[error("replication resync {active_resync_id} is already active for {bucket}/{arn}")]
 struct ResyncActiveConflictError {
@@ -712,6 +1078,10 @@ pub struct ReplicationPool<S: ReplicationStorage> {
 
     // MRF worker lifecycle
     mrf_worker_cancellations: Mutex<Vec<CancellationToken>>,
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     mrf_stop_tx: Sender<()>,
 
     // Worker size tracking
@@ -940,6 +1310,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Resizes worker priority and counts
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     pub async fn resize_worker_priority(
         &self,
         pri: ReplicationPriority,
@@ -1180,6 +1554,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Queues an MRF save operation
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     async fn queue_mrf_save(&self, entry: MrfReplicateEntry) {
         let _ = self.queue_mrf_save_admission(entry, "mrf_worker").await;
     }
@@ -1205,71 +1583,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         let storage = self.storage.clone();
 
         let handle = tokio::spawn(async move {
-            let recovery_lock = match storage
-                .new_ns_lock(
-                    ReplicationMetadataStore::rustfs_meta_bucket(),
-                    ReplicationMetadataStore::MRF_REPLICATION_RECOVERY_LOCK,
-                )
-                .await
-            {
-                Ok(lock) => lock,
-                Err(error) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %error,
-                        "Failed to create the MRF recovery leader lock"
-                    );
-                    return;
-                }
-            };
-            let recovery_guard = match recovery_lock
-                .get_write_lock_quiet(ReplicationLockTiming::acquire_timeout())
-                .await
-            {
-                Ok(guard) => guard,
-                Err(_) => {
-                    debug!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        "Another node is already processing the MRF recovery backlog"
-                    );
-                    return;
-                }
+            let Some(recovery_guard) = acquire_mrf_recovery_guard(&storage).await else {
+                return;
             };
 
-            let data = match ReplicationConfigStore::read(storage.clone(), ReplicationMetadataStore::MRF_REPLICATION_FILE).await {
-                Ok(d) => d,
-                Err(EcstoreError::ConfigNotFound) => {
-                    set_durable_mrf_backlog_summary(DurableMrfBacklogSummary {
-                        available: true,
-                        buckets: Vec::new(),
-                    });
-                    return;
-                }
-                Err(e) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %e,
-                        "Failed to load MRF recovery file"
-                    );
-                    return;
-                }
-            };
-
-            let entries = match decode_mrf_file(&data) {
-                Ok(v) => v,
-                Err(e) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %e,
-                        "Failed to decode MRF recovery file — preserving corrupt data"
-                    );
-                    quarantine_mrf_file(&storage, &data).await;
-                    return;
-                }
+            let Some(entries) = load_mrf_recovery_entries(&storage).await else {
+                return;
             };
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&entries));
 
@@ -1278,187 +1597,8 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
             let mut retry_entries = Vec::new();
 
             for entry in entries.iter() {
-                let admission = match entry.op {
-                    MrfOpKind::Delete => {
-                        if should_replay_force_delete_intent(entry) {
-                            let Some(operation_id) = entry.force_delete_id else {
-                                continue;
-                            };
-                            let delete = DeletedObjectReplicationInfo {
-                                delete_object: ReplicationDeletedObject {
-                                    object_name: entry.object.clone(),
-                                    force_delete: true,
-                                    force_delete_id: Some(operation_id),
-                                    force_delete_target_arns: entry.target_arns.clone(),
-                                    force_delete_generation: entry.force_delete_generation,
-                                    ..Default::default()
-                                },
-                                bucket: entry.bucket.clone(),
-                                op_type: ReplicationType::Heal,
-                                event_type: REPLICATE_HEAL_DELETE.to_string(),
-                                ..Default::default()
-                            };
-                            if replicate_delete_with_outcome(delete, storage.clone()).await {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        } else if entry.force_delete_id.is_some() {
-                            ReplicationQueueAdmission::Skipped
-                        } else {
-                            // Reconstruct a heal delete and re-queue it.  We do NOT call
-                            // get_object_info here because the delete-marker or version may
-                            // already be absent from the local store — that is expected.
-                            //
-                            // The MRF entry does not persist the replication decision and the
-                            // source object is gone, so re-derive the decision from the live
-                            // bucket config (mirroring get_heal_replicate_object_info) and set
-                            // it on the reconstructed delete. Without this the decision string
-                            // is empty and the delete replicates to zero targets — a silent
-                            // no-op that leaves replicas diverged (backlog#858 / #799 B9).
-                            let versioned = ReplicationVersioningStore::prefix_enabled(&entry.bucket, &entry.object).await;
-                            let oi = ObjectInfo {
-                                bucket: entry.bucket.clone(),
-                                name: entry.object.clone(),
-                                version_id: entry.version_id,
-                                delete_marker: entry.delete_marker,
-                                ..Default::default()
-                            };
-                            let dsc = if entry.target_arns.is_empty() {
-                                match ReplicationMetadataStore::optional_replication_config(&entry.bucket).await {
-                                    Ok(None) => continue,
-                                    Err(_) => {
-                                        retry_entries.push(entry.clone());
-                                        continue;
-                                    }
-                                    Ok(Some(_)) => match check_replicate_delete_strict(
-                                        &entry.bucket,
-                                        &ObjectToDelete {
-                                            object_name: entry.object.clone(),
-                                            version_id: entry.version_id,
-                                            ..Default::default()
-                                        },
-                                        &oi,
-                                        &ObjectOptions {
-                                            versioned,
-                                            ..Default::default()
-                                        },
-                                        None,
-                                    )
-                                    .await
-                                    {
-                                        Ok(dsc) => dsc,
-                                        Err(_) => {
-                                            retry_entries.push(entry.clone());
-                                            continue;
-                                        }
-                                    },
-                                }
-                            } else {
-                                replicate_decision_for_admitted_targets(&entry.target_arns)
-                            };
-                            let mut rstate = oi.replication_state();
-                            rstate.replicate_decision_str = dsc.to_string();
-
-                            let delete_marker_mtime = entry
-                                .delete_marker_mtime
-                                .and_then(|nanos| OffsetDateTime::from_unix_timestamp_nanos(i128::from(nanos)).ok());
-
-                            let dv = DeletedObjectReplicationInfo {
-                                delete_object: ReplicationDeletedObject {
-                                    object_name: entry.object.clone(),
-                                    version_id: entry.version_id,
-                                    delete_marker_version_id: entry.delete_marker_version_id,
-                                    delete_marker: entry.delete_marker,
-                                    delete_marker_mtime,
-                                    force_delete: entry.force_delete,
-                                    replication_state: Some(rstate),
-                                    ..Default::default()
-                                },
-                                bucket: entry.bucket.clone(),
-                                op_type: ReplicationType::Heal,
-                                event_type: REPLICATE_HEAL_DELETE.to_string(),
-                                ..Default::default()
-                            };
-                            if replicate_delete_with_outcome(dv, storage.clone()).await {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
-                    MrfOpKind::Object | MrfOpKind::Heal | MrfOpKind::ExistingObject => {
-                        let opts = ObjectOptions {
-                            version_id: entry.version_id.map(|u| u.to_string()),
-                            ..Default::default()
-                        };
-                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-                            Ok(oi) => oi,
-                            Err(e) => {
-                                debug!(
-                                    component = LOG_COMPONENT_ECSTORE,
-                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                    bucket = %entry.bucket,
-                                    object = %entry.object,
-                                    error = %e,
-                                    "MRF recovery: source object lookup failed"
-                                );
-                                if should_retry_mrf_source_lookup(&e) {
-                                    retry_entries.push(entry.clone());
-                                }
-                                continue;
-                            }
-                        };
-                        if entry.target_arns.is_empty() {
-                            // Legacy entries predate target admission persistence. They cannot
-                            // be safely attributed, so retain the old live-config fallback.
-                            queue_replication_heal(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
-                        } else {
-                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-                            let mut roi = replicate_object_info_from_object_info(oi, dsc, entry.op.replication_type());
-                            roi.retry_count = entry.retry_count.max(0) as u32;
-                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
-                    MrfOpKind::Metadata => {
-                        let opts = ObjectOptions {
-                            version_id: entry.version_id.map(|u| u.to_string()),
-                            ..Default::default()
-                        };
-                        let oi = match storage.get_object_info(&entry.bucket, &entry.object, &opts).await {
-                            Ok(oi) => oi,
-                            Err(e) => {
-                                debug!(
-                                    component = LOG_COMPONENT_ECSTORE,
-                                    subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                    bucket = %entry.bucket,
-                                    object = %entry.object,
-                                    error = %e,
-                                    "MRF metadata recovery: source object lookup failed"
-                                );
-                                if should_retry_mrf_source_lookup(&e) {
-                                    retry_entries.push(entry.clone());
-                                }
-                                continue;
-                            }
-                        };
-                        if entry.target_arns.is_empty() {
-                            queue_replication_metadata(&entry.bucket, oi, entry.retry_count.max(0) as u32).await
-                        } else {
-                            let dsc = replicate_decision_for_admitted_targets(&entry.target_arns);
-                            let mut roi = replicate_object_info_from_object_info(oi, dsc, ReplicationType::Metadata);
-                            roi.retry_count = entry.retry_count.max(0) as u32;
-                            if replicate_object_with_outcome(roi, storage.clone()).await.1 {
-                                ReplicationQueueAdmission::Queued
-                            } else {
-                                ReplicationQueueAdmission::Missed
-                            }
-                        }
-                    }
+                let Some(admission) = replay_mrf_entry(entry, &storage, &mut retry_entries).await else {
+                    continue;
                 };
 
                 if admission == ReplicationQueueAdmission::Missed {
@@ -1468,29 +1608,7 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
                 }
             }
 
-            let retained = match acknowledge_mrf_recovery(storage.clone(), &recovery_guard, &entries, &retry_entries).await {
-                Ok(retained) => retained,
-                Err(error) => {
-                    warn!(
-                        component = LOG_COMPONENT_ECSTORE,
-                        subsystem = LOG_SUBSYSTEM_REPLICATION,
-                        error = %error,
-                        "Failed to acknowledge the MRF recovery prefix; preserving it for the next startup"
-                    );
-                    match read_mrf_entries(storage.clone()).await {
-                        Ok(current) => current,
-                        Err(read_error) => {
-                            warn!(
-                                component = LOG_COMPONENT_ECSTORE,
-                                subsystem = LOG_SUBSYSTEM_REPLICATION,
-                                error = %read_error,
-                                "Failed to refresh the MRF backlog after acknowledgement failure"
-                            );
-                            entries.clone()
-                        }
-                    }
-                }
-            };
+            let retained = resolve_retained_mrf_entries(&storage, &recovery_guard, &entries, &retry_entries).await;
             let retained_count = retained.len();
             set_durable_mrf_backlog_snapshot(durable_mrf_backlog_summary_from_entries(&retained));
 
@@ -1651,6 +1769,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling regular replication operations
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     async fn add_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1664,6 +1786,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling large object replication operations
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     async fn add_large_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1678,6 +1804,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Worker function for handling MRF (Most Recent Failures) operations
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     async fn add_mrf_worker(
         &self,
         mut rx: Receiver<ReplicationOperation>,
@@ -1691,6 +1821,10 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
     }
 
     /// Delete resync metadata from replication resync state in memory
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     pub async fn delete_resync_metadata(&self, bucket: &str) {
         let mut status_map = self.resyncer.status_map.write().await;
         status_map.remove(bucket);

crates/ecstore/src/bucket/replication/replication_resync_boundary.rs

@@ -21,11 +21,31 @@ pub(crate) use rustfs_replication::{
     should_count_head_proxy_failure,
 };
 
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) const RESYNC_META_FORMAT: u16 = rustfs_replication::resync::RESYNC_META_FORMAT;
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) const RESYNC_META_VERSION: u16 = rustfs_replication::resync::RESYNC_META_VERSION;
 pub(crate) const RESYNC_FILE_MAX_BYTES: usize = rustfs_replication::RESYNC_FILE_MAX_BYTES;
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) const WIRE_ZERO_TIME_UNIX: i64 = rustfs_replication::resync::WIRE_ZERO_TIME_UNIX;
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) const MRF_META_FORMAT: u16 = rustfs_replication::mrf::MRF_META_FORMAT;
+#[allow(
+    dead_code,
+    reason = "declared boundary surface for the ECStore replication split plan; no caller in this port (backlog#1823)"
+)]
 pub(crate) const MRF_META_VERSION: u16 = rustfs_replication::mrf::MRF_META_VERSION;
 
 fn map_replication_error(err: rustfs_replication::Error) -> Error {

crates/ecstore/src/bucket/replication/replication_state.rs

@@ -340,6 +340,10 @@ impl ReplicationStats {
     }
 
     /// Site replication update replica statistics
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity replication surface with no caller in this port (backlog#1823)"
+    )]
     fn sr_update_replica_stat(&self, size: i64) {
         self.sr_stats.replica_size.fetch_add(size, Ordering::Relaxed);
         self.sr_stats.replica_count.fetch_add(1, Ordering::Relaxed);

crates/ecstore/src/bucket/target/bucket_target.rs

@@ -59,6 +59,10 @@ impl fmt::Debug for Credentials {
 }
 
 #[derive(Debug, Deserialize, Serialize, Default, Clone)]
+#[allow(
+    dead_code,
+    reason = "MinIO-parity bucket-target service discriminator with no caller in this port (backlog#1823)"
+)]
 pub enum ServiceType {
     #[default]
     Replication,

crates/ecstore/src/bucket/utils.rs

@@ -73,23 +73,6 @@ pub fn check_valid_bucket_name_strict(bucket_name: &str) -> Result<()> {
     check_bucket_name_common(bucket_name, true)
 }
 
-pub fn check_valid_object_name_prefix(object_name: &str) -> Result<()> {
-    if object_name.len() > 1024 {
-        return Err(Error::other("Object name cannot be longer than 1024 characters"));
-    }
-    if !object_name.is_ascii() {
-        return Err(Error::other("Object name with non-UTF-8 strings are not supported"));
-    }
-    Ok(())
-}
-
-pub fn check_valid_object_name(object_name: &str) -> Result<()> {
-    if object_name.trim().is_empty() {
-        return Err(Error::other("Object name cannot be empty"));
-    }
-    check_valid_object_name_prefix(object_name)
-}
-
 pub fn deserialize<T>(input: &[u8]) -> xml::DeResult<T>
 where
     T: for<'xml> xml::Deserialize<'xml>,
@@ -100,6 +83,10 @@ where
     Ok(ans)
 }
 
+#[allow(
+    dead_code,
+    reason = "xml serialize helper with no caller in this port; the live sibling is deserialize (backlog#1823)"
+)]
 pub fn serialize_content<T: xml::SerializeContent>(val: &T) -> xml::SerResult<String> {
     let mut buf = Vec::with_capacity(256);
     {
@@ -186,15 +173,27 @@ pub fn is_valid_object_name(object: &str) -> bool {
 /// Client-facing reason attached to rejections of object keys that Win32/NTFS
 /// cannot represent as file paths (issue #3299). Deployments on Linux/macOS
 /// accept the full S3 key character set.
+#[allow(
+    dead_code,
+    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
+)]
 pub const WINDOWS_RESERVED_CHARACTERS_REASON: &str =
     "object key contains characters unsupported on Windows hosts (one of ':', '*', '?', '\"', '|', '<', '>')";
 
 /// Client-facing reason for path segments Windows can store but not address
 /// afterwards (issue #3449): trailing dot/space or reserved DOS device names.
+#[allow(
+    dead_code,
+    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
+)]
 pub const WINDOWS_RESERVED_SEGMENT_REASON: &str = "object key contains a path segment unsupported on Windows hosts (trailing dot or space, or a reserved device name such as NUL/CON/COM1)";
 
 /// Reserved DOS device names that shadow regular files on Windows, even when
 /// an extension is appended (e.g. `NUL.txt` resolves to the `NUL` device).
+#[allow(
+    dead_code,
+    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
+)]
 const WINDOWS_RESERVED_NAMES: &[&str] = &[
     "CON", "PRN", "AUX", "NUL", "COM1", "COM2", "COM3", "COM4", "COM5", "COM6", "COM7", "COM8", "COM9", "LPT1", "LPT2", "LPT3",
     "LPT4", "LPT5", "LPT6", "LPT7", "LPT8", "LPT9",
@@ -204,6 +203,10 @@ const WINDOWS_RESERVED_NAMES: &[&str] = &[
 /// the Win32 API cannot address afterwards (issue #3449): segments ending in a
 /// dot or a space, and reserved DOS device names — bare or with an extension
 /// (`NUL.txt`), matching classic Win32 path resolution semantics.
+#[allow(
+    dead_code,
+    reason = "live on Windows: callers sit inside the #[cfg(target_os = \"windows\")] block in check_object_name_for_length_and_slash (backlog#1823)"
+)]
 pub fn object_name_has_windows_incompatible_segment(object: &str) -> bool {
     object.split(['/', '\\']).any(|segment| {
         if segment.ends_with('.') || segment.ends_with(' ') {

crates/ecstore/src/bucket/versioning_sys.rs

@@ -90,6 +90,10 @@ impl BucketVersioningSys {
     /// caller's own instance context so a second in-process store never
     /// answers with the first instance's versioning state; falls back to the
     /// ambient system when the instance cell is not initialized.
+    #[allow(
+        dead_code,
+        reason = "instance-scoped seam (backlog#1052) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result<VersioningConfiguration> {
         if bucket == RUSTFS_META_BUCKET || bucket.starts_with(RUSTFS_META_BUCKET) {
             return Ok(VersioningConfiguration::default());

crates/ecstore/src/disk/disk_store.rs

@@ -637,14 +637,23 @@ impl Default for DiskOperationMetrics {
 }
 
 impl DiskOperationMetrics {
+    #[allow(
+        dead_code,
+        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
+    )]
     fn record_call(&mut self) {
         self.lifetime_calls.fetch_add(1, Ordering::Relaxed);
     }
 
+    #[allow(
+        dead_code,
+        reason = "internal metrics recorder reached only from record() below (backlog#1823)"
+    )]
     fn record_latency(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_latency_atomic(now_sec, elapsed);
     }
 
+    #[allow(dead_code, reason = "metrics roll-up with no caller in this port (backlog#1823)")]
     fn record(&mut self, now_sec: u64, elapsed: Duration) {
         self.record_call();
         self.record_latency(now_sec, elapsed);
@@ -770,6 +779,7 @@ impl DiskHealthTracker {
     }
 
     /// Set disk as faulty
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn set_faulty(&self) {
         self.status.store(DISK_HEALTH_FAULTY, Ordering::Release);
     }
@@ -850,6 +860,7 @@ impl DiskHealthTracker {
         became_offline
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn mark_offline(&self, endpoint: &Endpoint, reason: &'static str) -> bool {
         let current = self.runtime_state();
         if current == RuntimeDriveHealthState::Offline {
@@ -980,11 +991,13 @@ impl DiskHealthTracker {
     }
 
     /// Get waiting operations count
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn waiting_count(&self) -> u32 {
         self.waiting.load(Ordering::Relaxed)
     }
 
     /// Get last success timestamp
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub fn last_success(&self) -> i64 {
         self.last_success.load(Ordering::Acquire)
     }
@@ -1026,21 +1039,6 @@ impl Default for DiskHealthTracker {
     }
 }
 
-/// Health check context key for tracking disk operations
-#[derive(Debug, Clone)]
-struct HealthDiskCtxKey;
-
-#[derive(Debug)]
-struct HealthDiskCtxValue {
-    last_success: Arc<AtomicI64>,
-}
-
-impl HealthDiskCtxValue {
-    fn log_success(&self) {
-        self.last_success.store(current_unix_nanos(), Ordering::Relaxed);
-    }
-}
-
 /// LocalDiskWrapper wraps a DiskStore with health tracking capabilities.
 /// This is similar to Go's xlStorageDiskIDCheck.
 #[derive(Debug, Clone)]
@@ -1072,10 +1070,6 @@ impl LocalDiskWrapper {
         )
     }
 
-    pub(crate) fn new_with_health(disk: Arc<LocalDisk>, health_check: bool, health: Arc<DiskHealthTracker>) -> Self {
-        Self::new_with_health_and_metrics(disk, health_check, health, Arc::new(DiskHealthMetricEpoch::default()))
-    }
-
     pub(crate) fn new_with_reconnect_state(
         disk: Arc<LocalDisk>,
         health_check: bool,
@@ -1438,20 +1432,6 @@ impl LocalDiskWrapper {
         }
     }
 
-    async fn check_id(&self, want_id: Option<Uuid>) -> Result<()> {
-        if want_id.is_none() {
-            return Ok(());
-        }
-
-        let stored_disk_id = self.disk.get_disk_id().await?;
-
-        if stored_disk_id != want_id {
-            return Err(Error::other(format!("Disk ID mismatch wanted {want_id:?}, got {stored_disk_id:?}")));
-        }
-
-        Ok(())
-    }
-
     /// Check if disk ID is stale
     async fn check_disk_stale(&self) -> Result<()> {
         let Some(current_disk_id) = *self.disk_id.read().await else {

crates/ecstore/src/disk/error_conv.rs

@@ -48,6 +48,7 @@ pub fn to_volume_error(io_err: std::io::Error) -> std::io::Error {
     }
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn to_disk_error(io_err: std::io::Error) -> std::io::Error {
     match io_err.kind() {
         std::io::ErrorKind::NotFound => DiskError::DiskNotFound.into(),

crates/ecstore/src/disk/fs.rs

@@ -178,6 +178,7 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
     }
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn remove_all(path: impl AsRef<Path>) -> io::Result<()> {
     // Try remove_file first; fall back to remove_dir_all if it's a directory
     match fs::remove_file(path.as_ref()).await {

crates/ecstore/src/disk/local.rs

@@ -665,6 +665,7 @@ async fn remove_empty_directory_tree_under_mount_lease(
 }
 
 #[cfg(unix)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 async fn remove_empty_directory_tree_with(
     root: &Path,
     before_descend: impl FnMut(&Path) -> std::io::Result<()>,
@@ -1016,13 +1017,29 @@ fn record_direct_read_page_fault_delta(path: &'static str, stage: &'static str,
 /// When enabled, shard reads bypass the page cache using O_DIRECT flag.
 /// Requires aligned buffers (typically 512 bytes or 4096 bytes).
 /// Default: false (uses page cache via mmap/pread).
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE";
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE: bool = false;
 
 /// Minimum shard size threshold for O_DIRECT reads.
 /// Only shards larger than this threshold will use O_DIRECT.
 /// Default: 4MB.
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: &str = "RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD";
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 
 /// Enable O_DIRECT for erasure shard / multipart part data writes (Linux only).
@@ -1036,7 +1053,15 @@ const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD: usize = 4 * 1024 * 1024;
 /// EINVAL/EOPNOTSUPP (tmpfs, overlayfs, 9p, ...) latch the path off and fall
 /// back to buffered writes for the whole disk. Non-Linux always falls back.
 /// Default: false (buffered writes via the page cache, as before).
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: &str = "RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE";
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE: bool = false;
 const ENV_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: &str = "RUSTFS_OBJECT_MMAP_POPULATE_ENABLE";
 const DEFAULT_RUSTFS_OBJECT_MMAP_POPULATE_ENABLE: bool = false;
@@ -1095,12 +1120,14 @@ macro_rules! cached_read_env {
 
 cached_read_env! {
     /// Check if O_DIRECT reads are enabled.
+    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_read_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_ENABLE);
 }
 
 cached_read_env! {
     /// Check if O_DIRECT shard/part data writes are enabled.
+    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn is_direct_io_write_enabled() -> bool =
         rustfs_utils::get_env_bool(ENV_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_WRITE_ENABLE);
 }
@@ -1456,6 +1483,7 @@ pub(crate) fn effective_durability(volume: &str) -> DurabilityMode {
 
 cached_read_env! {
     /// Get the O_DIRECT read threshold size.
+    #[allow(dead_code, reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)")]
     fn get_direct_io_read_threshold() -> usize =
         rustfs_utils::get_env_usize(ENV_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD, DEFAULT_RUSTFS_OBJECT_DIRECT_IO_READ_THRESHOLD);
 }
@@ -1673,12 +1701,20 @@ impl DirectIoWriteState {
 /// Target staging size for O_DIRECT writes, rounded up to the DIO alignment.
 /// Bounds the per-writer aligned bounce buffer and batches many shard blocks
 /// into one positioned write to keep the syscall count low.
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 const DIRECT_WRITE_STAGING_BYTES: usize = 1024 * 1024;
 
 /// Aligned bounce-buffer capacity for a given DIO alignment: the target staging
 /// size rounded up to a whole multiple of `align` so the buffer address, every
 /// flushed batch length, and every write offset stay alignment-correct.
 /// Platform-independent (no O_DIRECT), so it is unit-tested on any host.
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 fn direct_write_staging_capacity(align: usize) -> usize {
     debug_assert!(align.is_power_of_two() && align >= 512);
     DIRECT_WRITE_STAGING_BYTES.div_ceil(align) * align
@@ -1687,6 +1723,10 @@ fn direct_write_staging_capacity(align: usize) -> usize {
 /// Split `filled` staged bytes into the alignment-sized prefix written with
 /// O_DIRECT and the sub-alignment tail written buffered. Platform-independent,
 /// so the tail-boundary math is unit-tested on any host.
+#[allow(
+    dead_code,
+    reason = "platform-conditional: production callers are inside #[cfg(target_os = \"linux\")] blocks, so this reads as dead on non-Linux hosts (backlog#1823)"
+)]
 fn direct_write_tail_split(filled: usize, align: usize) -> (usize, usize) {
     let aligned = filled - (filled % align);
     (aligned, filled - aligned)
@@ -2142,6 +2182,7 @@ fn set_delete_version_fail_after_data_staged(path: &str) {
 }
 
 #[cfg(test)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn set_delete_version_fail_after_commit(root: &Path, path: &str) {
     DELETE_VERSION_FAIL_AFTER_COMMIT
         .lock()
@@ -2447,6 +2488,10 @@ enum SyncMode {
     FileOnly,
 }
 
+#[allow(
+    dead_code,
+    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
+)]
 struct FileCacheReclaimWriter {
     inner: File,
     reclaim_len: usize,
@@ -2454,6 +2499,10 @@ struct FileCacheReclaimWriter {
     reclaimed: bool,
 }
 
+#[allow(
+    dead_code,
+    reason = "reclaim bookkeeping fields written by Drop but never read back (backlog#1823)"
+)]
 struct FileCacheReclaimReader {
     inner: File,
     reclaim_offset: u64,
@@ -2519,6 +2568,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for StallTimeoutReader<R> {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "reclaim metrics emitter reached only from the Linux-gated reclaim paths (backlog#1823)"
+)]
 fn record_file_cache_reclaim_success(kind: &'static str, reclaim_len: usize, started: std::time::Instant) {
     // Runs per read-stream page-cache reclaim window; skip the whole emission
     // (three metric-key constructions) when general metrics are disabled.
@@ -3071,6 +3124,7 @@ impl LocalIoBackend for StdBackend {
             use memmap2::MmapOptions;
             use std::time::{Duration as StdDuration, Instant as StdInstant};
 
+            #[allow(dead_code, reason = "mmap copy result slot kept beside the mapping it owns (backlog#1823)")]
             struct MmapCopyReadResult {
                 bytes: Bytes,
                 access_check_duration: StdDuration,
@@ -4704,6 +4758,10 @@ fn build_local_io_backend(root: PathBuf) -> Arc<dyn LocalIoBackend> {
     Arc::new(StdBackend::new(root))
 }
 
+#[allow(
+    dead_code,
+    reason = "path cache and cwd slots retained beside the disk root they derive from (backlog#1823)"
+)]
 pub struct LocalDisk {
     pub root: PathBuf,
     publication_root: os::PublicationRoot,
@@ -5490,6 +5548,7 @@ impl LocalDisk {
         Ok(Self::resolve_abs_path_from(&self.root, path.as_ref()))
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn io_resolve_abs_path(&self, path: impl AsRef<Path>) -> PathBuf {
         let path_ref = path.as_ref();
         let path_str = path_ref.to_string_lossy();
@@ -5567,15 +5626,24 @@ impl LocalDisk {
     }
 
     // Check if a path is valid
+    #[allow(
+        dead_code,
+        reason = "method wrapper over the live free function check_local_disk_valid_path; no caller in this port (backlog#1823)"
+    )]
     fn check_valid_path<P: AsRef<Path>>(&self, path: P) -> Result<()> {
         check_local_disk_valid_path(self.io_root(), path)
     }
 
+    #[allow(
+        dead_code,
+        reason = "method wrapper over the live free function reject_local_disk_symlink_components; no caller in this port (backlog#1823)"
+    )]
     fn reject_symlink_components(&self, path: &Path) -> Result<()> {
         reject_local_disk_symlink_components(self.io_root(), path)
     }
 
     // Batch path generation with single lock acquisition
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn get_object_paths_batch(&self, requests: &[(String, String)]) -> Result<Vec<PathBuf>> {
         let mut results = Vec::with_capacity(requests.len());
         let mut cache_misses = Vec::new();
@@ -6488,6 +6556,7 @@ impl LocalDisk {
         Ok(f)
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn open_file_read_only(&self, path: impl AsRef<Path>) -> Result<File> {
         let f = super::fs::open_file(path.as_ref(), O_RDONLY).await.map_err(to_file_error)?;
         Ok(f)

crates/ecstore/src/disk/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: disk abstractions still carry staged health and direct-I/O migration paths.
-#![allow(dead_code)]
 
 pub mod disk_store;
 pub mod endpoint;
@@ -1114,6 +1113,10 @@ pub struct DiskInfo {
 }
 
 #[derive(Clone, Debug, Default)]
+#[allow(
+    dead_code,
+    reason = "MinIO-parity disk info shape with no constructor in this port (backlog#1823)"
+)]
 pub struct Info {
     pub total: u64,
     pub free: u64,
@@ -1372,6 +1375,7 @@ pub fn conv_part_err_to_int(err: &Option<Error>) -> usize {
     }
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn has_part_err(part_errs: &[usize]) -> bool {
     part_errs.iter().any(|err| *err != CHECK_PART_SUCCESS)
 }

crates/ecstore/src/disk/os.rs

@@ -571,6 +571,10 @@ fn regular_files(dir: &Path) -> io::Result<Vec<PathBuf>> {
 
 /// Fdatasync every regular file directly inside `dir`, then fsync the directory
 /// itself.
+#[allow(
+    dead_code,
+    reason = "reached only through sync_dir_files, whose callers are tests (backlog#1823)"
+)]
 pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
     for entry in std::fs::read_dir(dir.as_ref())? {
         let entry = entry?;
@@ -583,6 +587,7 @@ pub fn sync_dir_files_std(dir: impl AsRef<Path>) -> io::Result<()> {
 
 /// Async wrapper around [`sync_dir_files_std`]. Large directories flush files
 /// concurrently, bounded both per directory and process-wide.
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub async fn sync_dir_files(dir: impl AsRef<Path>) -> io::Result<()> {
     sync_dir_files_with_limiter(dir, Arc::new(Semaphore::new(MAX_PARALLEL_FILE_SYNCS))).await
 }
@@ -1809,10 +1814,6 @@ impl RenameCommitGuard {
         })
     }
 
-    pub(crate) fn lock_destination_directory_for_path_access(&self, directory: &Path) -> io::Result<RenameDestinationPathGuard> {
-        self.destination_directory_guard(directory, false)
-    }
-
     pub(crate) fn create_destination_directory_for_path_access(
         &self,
         directory: &Path,
@@ -2858,13 +2859,6 @@ pub async fn os_mkdir_all(dir_path: impl AsRef<Path>, base_dir: impl AsRef<Path>
     Ok(())
 }
 
-/// Check if a file exists.
-/// Returns true if the file exists, false otherwise.
-#[tracing::instrument(level = "debug", skip_all)]
-pub fn file_exists(path: impl AsRef<Path>) -> bool {
-    std::fs::metadata(path.as_ref()).map(|_| true).unwrap_or(false)
-}
-
 /// Whether an [`io::Error`] means "the directory is not empty".
 ///
 /// POSIX lets `rmdir`/`rename` report a non-empty directory as either

crates/ecstore/src/io_support/bitrot.rs

@@ -704,6 +704,7 @@ pub(crate) async fn create_bitrot_reader_from_bytes_with_stage_metrics(
 }
 
 #[allow(clippy::too_many_arguments)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn create_deferred_bitrot_reader(
     inline_data: Option<Bytes>,
     disk: Option<DiskStore>,

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -180,11 +180,13 @@ pub(in crate::set_disk) enum GetCodecStreamingReaderBuildOutcome {
     Fallback(GetCodecStreamingFallbackReason),
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) struct MultipartCodecStreamingReader {
     pub(in crate::set_disk) readers: VecDeque<Box<dyn AsyncRead + Unpin + Send + Sync>>,
 }
 
 impl MultipartCodecStreamingReader {
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) fn new(readers: Vec<Box<dyn AsyncRead + Unpin + Send + Sync>>) -> Self {
         Self {
             readers: VecDeque::from(readers),
@@ -666,6 +668,60 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     parts_metadata: &[FileInfo],
     disks: &[Option<DiskStore>],
 ) -> Option<&'static str> {
+    if let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(candidate) {
+        return Some(reason);
+    }
+
+    let Ok(erasure) = coding::Erasure::try_new_with_options(
+        candidate.erasure.data_blocks,
+        candidate.erasure.parity_blocks,
+        candidate.erasure.block_size,
+        candidate.uses_legacy_checksum,
+    ) else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
+    };
+    let data_files =
+        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
+            disks.get(index).is_some_and(Option::is_some)
+        }) {
+            Ok(data_files) => data_files,
+            Err(reason) => return Some(reason),
+        };
+
+    let Some(part) = candidate.parts.first() else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
+    };
+    let Ok(object_size) = usize::try_from(candidate.size) else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
+    };
+    let checksum_info = candidate.erasure.get_checksum_info(part.number);
+    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
+        HashAlgorithm::HighwayHash256SLegacy
+    } else {
+        checksum_info.algorithm
+    };
+    let read_length = inline_erasure_shard_file_offset(
+        0,
+        object_size,
+        object_size,
+        candidate.erasure.block_size,
+        erasure.data_shards,
+        candidate.uses_legacy_checksum,
+    );
+    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
+    let Ok(mut readers) =
+        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
+    else {
+        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
+    };
+
+    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
+        Some(body) if body.len() == object_size => None,
+        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
+    }
+}
+
+fn data_read_early_stop_inline_candidate_miss_reason(candidate: &FileInfo) -> Option<&'static str> {
     // `inline_data` excludes remote objects; this diagnostic reports them separately.
     if !rustfs_utils::http::contains_key_str(&candidate.metadata, rustfs_utils::http::SUFFIX_INLINE_DATA) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_NOT_INLINE);
@@ -703,51 +759,7 @@ pub(in crate::set_disk) async fn data_read_early_stop_inline_body_miss_reason(
     if !can_try_inline_data_shards_direct(object_size, candidate.erasure.block_size) {
         return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_SIZE);
     }
-
-    let Ok(erasure) = coding::Erasure::try_new_with_options(
-        candidate.erasure.data_blocks,
-        candidate.erasure.parity_blocks,
-        candidate.erasure.block_size,
-        candidate.uses_legacy_checksum,
-    ) else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_GEOMETRY);
-    };
-    let data_files =
-        match collect_inline_data_shard_fileinfos_by_index_or_reason(parts_metadata, candidate, erasure.data_shards, |index| {
-            disks.get(index).is_some_and(Option::is_some)
-        }) {
-            Ok(data_files) => data_files,
-            Err(reason) => return Some(reason),
-        };
-
-    let Some(part) = candidate.parts.first() else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_PART_SHAPE);
-    };
-    let checksum_info = candidate.erasure.get_checksum_info(part.number);
-    let checksum_algo = if candidate.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
-        HashAlgorithm::HighwayHash256SLegacy
-    } else {
-        checksum_info.algorithm
-    };
-    let read_length = inline_erasure_shard_file_offset(
-        0,
-        object_size,
-        object_size,
-        candidate.erasure.block_size,
-        erasure.data_shards,
-        candidate.uses_legacy_checksum,
-    );
-    let shard_size = inline_erasure_shard_size(candidate.erasure.block_size, erasure.data_shards, candidate.uses_legacy_checksum);
-    let Ok(mut readers) =
-        build_inline_bitrot_readers_from_refs(&data_files, bucket, object, read_length, shard_size, &checksum_algo, false).await
-    else {
-        return Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY);
-    };
-
-    match try_read_inline_data_shards_direct(&mut readers, erasure.data_shards, read_length, object_size).await {
-        Some(body) if body.len() == object_size => None,
-        _ => Some(GET_METADATA_EARLY_STOP_REASON_DATA_READ_INLINE_BODY_VERIFY),
-    }
+    None
 }
 
 fn data_read_inline_missing_shards_are_pending(
@@ -1836,6 +1848,7 @@ pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum_all_shards(
 }
 
 #[allow(clippy::too_many_arguments)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn create_bitrot_readers_until_quorum(
     files: &[FileInfo],
     disks: &[Option<DiskStore>],
@@ -2126,6 +2139,7 @@ pub(in crate::set_disk) async fn create_data_block_bitrot_readers(
     setup
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(in crate::set_disk) async fn collect_read_multiple_results<F>(
     tasks: Vec<F>,
     read_quorum: usize,
@@ -2442,6 +2456,7 @@ impl SetDisks {
         let bucket: Arc<str> = Arc::from(bucket);
         let object: Arc<str> = Arc::from(object);
         let version_id: Arc<str> = Arc::from(version_id);
+        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let futures = disks.iter().enumerate().map(|(disk_index, disk)| {
             let disk = disk.clone();
             let task_opts = opts;
@@ -2449,10 +2464,14 @@ impl SetDisks {
             let bucket = bucket.clone();
             let object = object.clone();
             let version_id = version_id.clone();
+            let slowtail_fault = slowtail_fault.clone();
             tokio::spawn(async move {
                 let response_start = observe.then(Instant::now);
                 let result = if let Some(disk) = disk {
                     Self::record_read_version_call(&object, disk_index);
+                    if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(disk_index)) {
+                        tokio::time::sleep(delay).await;
+                    }
                     disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                         .await
                 } else {
@@ -2548,6 +2567,7 @@ impl SetDisks {
         let mut scheduled_count = 0usize;
         let mut force_full_wait = false;
         let mut final_miss_reason_override = None;
+        let slowtail_fault = get_metadata_slowtail_fault_request(bucket.as_ref(), object.as_ref(), read_data);
         let spawn_read_version =
             |join_set: &mut JoinSet<(usize, disk::error::Result<FileInfo>, Duration)>, index: usize, disk: Option<DiskStore>| {
                 let task_opts = opts;
@@ -2555,6 +2575,7 @@ impl SetDisks {
                 let bucket = bucket.clone();
                 let object = object.clone();
                 let version_id = version_id.clone();
+                let slowtail_fault = slowtail_fault.clone();
                 join_set.spawn(async move {
                     let response_start = Instant::now();
                     let result = if let Some(disk) = disk {
@@ -2563,6 +2584,9 @@ impl SetDisks {
                         Self::record_read_version_call(&object, index);
                         #[cfg(test)]
                         Self::read_version_fanout_barrier(&object, index).await;
+                        if let Some(delay) = slowtail_fault.as_ref().and_then(|fault| fault.delay_for_disk(index)) {
+                            tokio::time::sleep(delay).await;
+                        }
                         disk.read_version(&org_bucket, &bucket, &object, &version_id, &task_opts)
                             .await
                     } else {
@@ -2596,6 +2620,14 @@ impl SetDisks {
                     Ok(file_info) => {
                         observations.push(MetadataFanoutObservation::from_file_info(&file_info, elapsed));
                         accumulator.observe_file_info(&file_info);
+                        if bounded_fanout
+                            && read_data
+                            && !force_full_wait
+                            && let Some(reason) = data_read_early_stop_inline_candidate_miss_reason(&file_info)
+                        {
+                            force_full_wait = true;
+                            final_miss_reason_override.get_or_insert(reason);
+                        }
                         if let Some(slot) = ress.get_mut(index) {
                             *slot = file_info;
                         }
@@ -2955,6 +2987,7 @@ impl SetDisks {
         (meta_file_infos, errs)
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn read_multiple_files(
         disks: &[Option<DiskStore>],
         req: ReadMultipleReq,
@@ -2988,14 +3021,11 @@ impl SetDisks {
             });
         }
 
-        let (ress, errors) = match collect_read_multiple_results(futures, read_quorum).await {
+        let (ress, _errors) = match collect_read_multiple_results(futures, read_quorum).await {
             Ok(collected) => collected,
             Err(()) => return empty_quorum_result(),
         };
 
-        // debug!("ReadMultipleResp ress {:?}", ress);
-        // debug!("ReadMultipleResp errors {:?}", errors);
-
         let mut ret = Vec::with_capacity(req.files.len());
 
         for want in req.files.iter() {
@@ -3134,6 +3164,7 @@ pub(in crate::set_disk) struct RenameDataCommit {
     pub(in crate::set_disk) committed_file_info: FileInfo,
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 type RenameDataLegacyTuple = (
     Vec<Option<DiskStore>>,
     RenameConvergence,
@@ -3143,6 +3174,7 @@ type RenameDataLegacyTuple = (
 );
 
 impl RenameDataCommit {
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn into_legacy_tuple(self) -> RenameDataLegacyTuple {
         (
             self.online_disks,
@@ -3261,6 +3293,7 @@ impl SetDisks {
 
     #[tracing::instrument(level = "debug", skip(disks, file_infos))]
     #[allow(clippy::type_complexity)]
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn rename_data(
         disks: &[Option<DiskStore>],
         src_bucket: &str,
@@ -5073,6 +5106,7 @@ fn is_cleanup_not_found(e: &DiskError) -> bool {
 /// normalized to `DiskNotFound`: a panic is not a "disk absent" condition and
 /// must not be silently swallowed as an ignorable error (fixes the historical
 /// `Unexpected`/`DiskNotFound` misclassification).
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn map_cleanup_join_result(joined: std::result::Result<Option<DiskError>, tokio::task::JoinError>) -> Option<DiskError> {
     match joined {
         Ok(res) => res,
@@ -5297,6 +5331,7 @@ pub(in crate::set_disk) mod rename_fanout_barrier_phase {
     /// The per-disk old-data-dir cleanup phase of the commit fan-out.
     pub const CLEANUP: &str = "cleanup";
     /// The per-disk `read_version` phase of metadata read fan-out.
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub const READ_VERSION: &str = "read_version";
 }
 
@@ -5734,6 +5769,130 @@ mod tests {
         (dirs, disks)
     }
 
+    #[test]
+    fn metadata_slowtail_fault_delay_parses_and_filters_request() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some("bench-bucket")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            || {
+                assert_eq!(
+                    get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, true),
+                    Some(Duration::from_millis(25))
+                );
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 2, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("other-bucket", "objects/000001", 3, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "other/000001", 3, true).is_none());
+                assert!(get_metadata_slowtail_fault_delay("bench-bucket", "objects/000001", 3, false).is_none());
+            },
+        );
+    }
+
+    #[test]
+    fn metadata_slowtail_fault_delay_disables_invalid_disk_list() {
+        temp_env::with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("25")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("1,nope")),
+            ],
+            || {
+                assert!(get_metadata_slowtail_fault_delay("bucket", "object", 1, true).is_none());
+            },
+        );
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn metadata_slowtail_fault_delays_only_data_read_metadata_task() {
+        const DISKS: usize = 4;
+        let bucket = "metadata-slowtail-fault-bucket";
+        let object = "objects/metadata-slowtail-fault-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
+
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            async {
+                let read_without_data =
+                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", false, false, false, true, 2);
+                tokio::time::timeout(Duration::from_millis(100), read_without_data)
+                    .await
+                    .expect("non-data metadata fanout must not be delayed by the data-read slowtail hook")
+                    .expect("metadata fanout without read_data should resolve");
+
+                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
+                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
+                ));
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
+                        .await
+                        .is_err(),
+                    "data-read metadata fanout must wait for the injected slow read_version response"
+                );
+                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
+                    .await
+                    .expect("injected slowtail should eventually complete")
+                    .expect("data-read metadata fanout should resolve");
+                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
+                assert!(errs.iter().all(Option::is_none));
+                assert_eq!(diagnostics.total_responses(), DISKS);
+            },
+        )
+        .await;
+
+        drop(dirs);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn metadata_slowtail_fault_delays_early_stop_metadata_task() {
+        const DISKS: usize = 4;
+        let bucket = "metadata-slowtail-early-stop-bucket";
+        let object = "objects/metadata-slowtail-early-stop-object";
+        let (dirs, disks) = call_counter_local_disks(bucket, DISKS).await;
+        install_metadata_fanout_fileinfo(&disks, bucket, object, None).await;
+
+        temp_env::async_with_vars(
+            [
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
+                (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("true")),
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, Some("150")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS, Some("3")),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET, Some(bucket)),
+                (ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX, Some("objects/")),
+            ],
+            async {
+                let mut read_with_data = Box::pin(SetDisks::read_all_fileinfo_observed(
+                    &disks, bucket, bucket, object, "", true, false, false, true, 2,
+                ));
+                assert!(
+                    tokio::time::timeout(Duration::from_millis(40), &mut read_with_data)
+                        .await
+                        .is_err(),
+                    "early-stop metadata fanout must still wait for the injected slow response after fallback to full wait"
+                );
+                let (parts_metadata, errs, diagnostics) = tokio::time::timeout(Duration::from_secs(2), read_with_data)
+                    .await
+                    .expect("injected early-stop slowtail should eventually complete")
+                    .expect("early-stop metadata fanout should resolve");
+                assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
+                assert!(errs.iter().all(Option::is_none));
+                assert_eq!(diagnostics.total_responses(), DISKS);
+            },
+        )
+        .await;
+
+        drop(dirs);
+    }
+
     /// Demo / regression guard for the backlog#1325 per-disk call counters.
     ///
     /// The metadata fan-out issues each `read_version` inside its own
@@ -7081,7 +7240,7 @@ mod tests {
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn bounded_non_inline_data_get_hedges_then_waits_for_full_fanout() {
+    async fn bounded_non_inline_data_get_immediately_forces_full_fanout() {
         const DISKS: usize = 4;
         let bucket = "bounded-data-get-hedge-bucket";
         let object = "bounded-data-get-hedge-object";
@@ -7112,7 +7271,7 @@ mod tests {
                     }
                 })
                 .await
-                .expect("bounded data-read fanout should hedge by starting the spare disk");
+                .expect("bounded non-inline data-read fanout should immediately schedule the spare disk");
 
                 let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
                 assert!(
@@ -7128,7 +7287,7 @@ mod tests {
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "bounded data-read fanout should issue the paused disk plus one spare hedge"
+                    "bounded non-inline data-read fanout should issue the paused disk plus the remaining spare"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);
@@ -7155,16 +7314,42 @@ mod tests {
                 ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", None::<&str>),
             ],
             async {
+                let barrier = rename_fanout_barrier::arm(object, 2, rename_fanout_barrier::PHASE_READ_VERSION);
                 let calls = disk_call_counters::observe(object);
-                let (parts_metadata, errs, diagnostics) =
-                    SetDisks::read_all_fileinfo_observed(&disks, bucket, bucket, object, "", true, false, false, true, 2)
+                let disks_for_read = disks.clone();
+                let mut read = tokio::spawn(async move {
+                    SetDisks::read_all_fileinfo_observed(&disks_for_read, bucket, bucket, object, "", true, false, false, true, 2)
                         .await
-                        .expect("default data-read metadata should resolve");
+                });
+
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, barrier.wait_until_paused())
+                    .await
+                    .expect("default bounded non-inline read should schedule the paused metadata task");
+                tokio::time::timeout(BARRIER_PAUSE_GUARD, async {
+                    while calls.for_disk(disk_call_counters::KIND_READ_VERSION, 3) == 0 {
+                        tokio::task::yield_now().await;
+                    }
+                })
+                .await
+                .expect(
+                    "default bounded non-inline read should immediately force full fanout after the first non-inline response",
+                );
+
+                let pending = tokio::time::timeout(BARRIER_PAUSE_GUARD, &mut read).await;
+                assert!(
+                    pending.is_err(),
+                    "default non-inline data reads must not return before the paused metadata response"
+                );
+                barrier.release();
+                let (parts_metadata, errs, diagnostics) = read
+                    .await
+                    .expect("metadata read task should not panic")
+                    .expect("default data-read metadata should resolve");
 
                 assert_eq!(
                     calls.total(disk_call_counters::KIND_READ_VERSION),
                     DISKS as u64,
-                    "default non-inline GET data-read metadata must keep full fanout for read-failure tolerance"
+                    "default non-inline GET data-read metadata must keep full fanout without waiting for a quorum miss first"
                 );
                 assert_eq!(diagnostics.total_responses(), DISKS);
                 assert_eq!(parts_metadata.iter().filter(|fi| fi.name == object).count(), DISKS);

crates/ecstore/src/set_disk/ctx.rs

@@ -42,12 +42,20 @@ impl<'a> SetDisksCtx<'a> {
     }
 
     /// The borrowed core, for state not yet fronted by a typed accessor.
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn core(&self) -> &'a SetDisks {
         self.core
     }
 
     // --- Immutable topology / config (fixed after construction) ---
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn set_index(&self) -> usize {
         self.core.set_index
     }
@@ -56,14 +64,26 @@ impl<'a> SetDisksCtx<'a> {
         self.core.pool_index
     }
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn set_drive_count(&self) -> usize {
         self.core.set_drive_count
     }
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn default_parity_count(&self) -> usize {
         self.core.default_parity_count
     }
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn set_endpoints(&self) -> &'a [Endpoint] {
         &self.core.set_endpoints
     }
@@ -72,6 +92,10 @@ impl<'a> SetDisksCtx<'a> {
         &self.core.format
     }
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn locker_owner(&self) -> &'a str {
         &self.core.locker_owner
     }
@@ -84,6 +108,10 @@ impl<'a> SetDisksCtx<'a> {
 
     // --- Locker trio ---
 
+    #[allow(
+        dead_code,
+        reason = "SetDisks split seam (backlog#815) with no caller in this port (backlog#1823)"
+    )]
     pub(crate) fn lockers(&self) -> &'a [Arc<dyn LockClient>] {
         &self.core.lockers
     }

crates/ecstore/src/set_disk/mod.rs

@@ -39,7 +39,6 @@
 //! - `metadata.rs`, `replication.rs`, `shard_source.rs` — supporting helpers.
 
 // #730: SetDisks still hosts staged read/heal/write migration helpers.
-#![allow(dead_code)]
 #![allow(unused_imports)]
 #![allow(unused_variables)]
 
@@ -103,9 +102,7 @@ use crate::storage_api_contracts::{
 };
 use crate::store::utils::is_reserved_or_invalid_bucket;
 use crate::{
-    bucket::lifecycle::bucket_lifecycle_ops::{
-        LifecycleOps, gen_transition_objname, get_transitioned_object_reader_with_tier_manager, put_restore_opts,
-    },
+    bucket::lifecycle::bucket_lifecycle_ops::{LifecycleOps, get_transitioned_object_reader_with_tier_manager, put_restore_opts},
     cache_value::metacache_set::{ListPathRawOptions, list_path_raw},
     config::storageclass,
     disk::{
@@ -177,15 +174,14 @@ use std::future::Future;
 use std::hash::{BuildHasher, Hash, Hasher};
 use std::mem::{self};
 use std::pin::Pin;
-use std::sync::OnceLock;
 use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, OnceLock};
 use std::task::{Context, Poll};
 use std::time::{Instant, SystemTime, UNIX_EPOCH};
 use std::{
     collections::{HashMap, HashSet},
     io::{Cursor, Write},
     path::Path,
-    sync::Arc,
     time::Duration,
 };
 use time::OffsetDateTime;
@@ -624,7 +620,9 @@ fn adaptive_duplex_buffer_size(object_size: i64) -> usize {
 // Each flag has a corresponding `*_ROLLOUT_PCT` for percentage-based gradual rollout.
 // ============================================================================
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_ONLINE_TIMEOUT: Duration = Duration::from_secs(1);
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 const DISK_HEALTH_CACHE_TTL: Duration = Duration::from_millis(750);
 const GET_OBJECT_METADATA_CACHE_TTL: Duration = Duration::from_secs(2); // Increased from 250ms to 2s
 const DEFAULT_GET_OBJECT_METADATA_CACHE_MAX_ENTRIES: usize = 4096; // Increased from 1024 to 4096
@@ -698,7 +696,15 @@ const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_EAR
 // the env var to `false` to fall back to full-wait metadata fanout.
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE: bool = true;
 
+#[allow(
+    dead_code,
+    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
+)]
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT";
+#[allow(
+    dead_code,
+    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
+)]
 const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_ROLLOUT_PCT: u32 = 100;
 
 const ENV_RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_METADATA_VERSION_EARLY_STOP_ENABLE";
@@ -708,7 +714,12 @@ const ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: &str = "RUSTFS_GET_ME
 const DEFAULT_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE: bool = true;
 
 const ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: &str = "RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT";
-const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = false;
+const DEFAULT_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT: bool = true;
+
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET";
+const ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX: &str = "RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX";
 
 // --- Multipart Reader-Setup Prefetch Configuration (backlog#870) ---
 
@@ -910,6 +921,10 @@ mod prepared_get_object_metadata_tests {
             .expect("test should find an object whose initial fanout covers both data shards")
     }
 
+    #[allow(
+        dead_code,
+        reason = "test fixture no assertion in this module uses today; the live namesake lives in io_primitives tests (backlog#1823)"
+    )]
     fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
         *bounded_metadata_fanout_order(bucket, object, 4, 2)
             .get(3)
@@ -1110,7 +1125,10 @@ mod prepared_get_object_metadata_tests {
 
         assert_eq!(object_size, payload.len() as i64);
         assert_eq!(restored, payload);
-        assert_eq!(calls_total, 4, "default production GET should eagerly schedule the full metadata fanout");
+        assert_eq!(
+            calls_total, 4,
+            "default production inline GET should schedule the initial bounded quorum plus one hedge"
+        );
         assert_eq!(
             recorder.histogram_values(
                 "rustfs_io_get_object_metadata_fanout_scheduled",
@@ -1684,6 +1702,95 @@ fn is_get_metadata_early_stop_bounded_fanout_enabled() -> bool {
     }
 }
 
+#[derive(Debug)]
+struct GetMetadataSlowtailFaultConfig {
+    delay: Duration,
+    disks: Arc<[usize]>,
+    bucket: Option<String>,
+    object_prefix: Option<String>,
+}
+
+#[derive(Clone, Debug)]
+struct GetMetadataSlowtailFaultRequest {
+    delay: Duration,
+    disks: Arc<[usize]>,
+}
+
+impl GetMetadataSlowtailFaultRequest {
+    fn delay_for_disk(&self, disk_index: usize) -> Option<Duration> {
+        self.disks.contains(&disk_index).then_some(self.delay)
+    }
+}
+
+fn parse_get_metadata_slowtail_fault_disks(raw: &str) -> Option<Vec<usize>> {
+    let mut disks = Vec::new();
+    for item in raw.split(',').map(str::trim).filter(|item| !item.is_empty()) {
+        let Ok(index) = item.parse::<usize>() else {
+            return None;
+        };
+        if !disks.contains(&index) {
+            disks.push(index);
+        }
+    }
+    (!disks.is_empty()).then_some(disks)
+}
+
+fn load_get_metadata_slowtail_fault_config() -> Option<GetMetadataSlowtailFaultConfig> {
+    let delay_ms = rustfs_utils::get_env_u64(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DELAY_MS, 0);
+    if delay_ms == 0 {
+        return None;
+    }
+    let disks = parse_get_metadata_slowtail_fault_disks(&std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_DISKS).ok()?)?;
+    let bucket = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_BUCKET)
+        .ok()
+        .filter(|value| !value.is_empty());
+    let object_prefix = std::env::var(ENV_RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX)
+        .ok()
+        .filter(|value| !value.is_empty());
+    Some(GetMetadataSlowtailFaultConfig {
+        delay: Duration::from_millis(delay_ms),
+        disks: Arc::from(disks.into_boxed_slice()),
+        bucket,
+        object_prefix,
+    })
+}
+
+fn get_metadata_slowtail_fault_request(bucket: &str, object: &str, read_data: bool) -> Option<GetMetadataSlowtailFaultRequest> {
+    if !read_data {
+        return None;
+    }
+
+    #[cfg(test)]
+    let config = load_get_metadata_slowtail_fault_config();
+    #[cfg(test)]
+    let config = config.as_ref()?;
+    #[cfg(not(test))]
+    let config = ({
+        static CACHED: OnceLock<Option<GetMetadataSlowtailFaultConfig>> = OnceLock::new();
+        CACHED.get_or_init(load_get_metadata_slowtail_fault_config).as_ref()
+    })?;
+
+    if let Some(expected_bucket) = &config.bucket
+        && expected_bucket != bucket
+    {
+        return None;
+    }
+    if let Some(expected_prefix) = &config.object_prefix
+        && !object.starts_with(expected_prefix)
+    {
+        return None;
+    }
+    Some(GetMetadataSlowtailFaultRequest {
+        delay: config.delay,
+        disks: config.disks.clone(),
+    })
+}
+
+#[cfg(test)]
+fn get_metadata_slowtail_fault_delay(bucket: &str, object: &str, disk_index: usize, read_data: bool) -> Option<Duration> {
+    get_metadata_slowtail_fault_request(bucket, object, read_data)?.delay_for_disk(disk_index)
+}
+
 /// Check if multipart reads prefetch the next part's bitrot reader setup
 /// while the current part decodes (backlog#870).
 ///
@@ -1709,6 +1816,10 @@ fn is_multipart_reader_setup_prefetch_enabled() -> bool {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
+)]
 fn get_metadata_early_stop_rollout_pct() -> u32 {
     static CACHED: OnceLock<u32> = OnceLock::new();
     *CACHED.get_or_init(|| {
@@ -1748,6 +1859,10 @@ fn should_use_codec_streaming(config: GetCodecStreamingConfig, bucket: &str, obj
 }
 
 /// Should this specific request use metadata early-stop?
+#[allow(
+    dead_code,
+    reason = "percentage-rollout facet of the metadata early-stop switch; its predicate has no caller while the sibling enable flag is live (backlog#1823)"
+)]
 pub fn should_use_metadata_early_stop(bucket: &str, object: &str) -> bool {
     let base = is_get_metadata_early_stop_enabled();
     let pct = get_metadata_early_stop_rollout_pct();
@@ -2181,6 +2296,7 @@ fn classify_get_codec_streaming_object_class(
     GetCodecStreamingObjectClass::PlainSinglePart
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_small_object_direct_memory_eligible_with_threshold(
     range: &Option<HTTPRangeSpec>,
     object_info: &ObjectInfo,
@@ -2786,6 +2902,7 @@ pub struct SetDisks {
     /// Stable namespace shared by every object lock created for this set.
     set_lock_namespace: Arc<str>,
     pub format: FormatV3,
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     disk_health_cache: Arc<RwLock<Vec<Option<DiskHealthEntry>>>>,
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
@@ -3061,11 +3178,13 @@ struct GetObjectMetadataCacheEntry {
 
 #[derive(Clone, Debug)]
 struct DiskHealthEntry {
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     last_check: Instant,
     online: bool,
 }
 
 impl DiskHealthEntry {
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn cached_value(&self) -> Option<bool> {
         if self.last_check.elapsed() <= DISK_HEALTH_CACHE_TTL {
             Some(self.online)
@@ -3659,6 +3778,7 @@ fn multipart_put_large_batch_min_size_bytes() -> usize {
     })
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn classify_small_write_path(is_inline_buffer: bool, object_size: i64, block_size: usize) -> SmallWritePath {
     if should_use_inline_small_fast_path(is_inline_buffer, object_size, block_size) {
         SmallWritePath::Inline
@@ -4237,6 +4357,7 @@ fn check_object_lock_retention_update(bucket: &str, object: &str, obj_info: &Obj
 ///
 /// Fail closed: when bucket metadata cannot be resolved the check stays on, so
 /// object-lock protection is never skipped because of a metadata lookup miss.
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(crate) fn object_lock_delete_check_required(bucket_meta: Option<&crate::bucket::metadata::BucketMetadata>) -> bool {
     bucket_meta.is_none_or(|meta| meta.object_locking())
 }
@@ -4512,15 +4633,6 @@ impl Hash for ObjProps {
     }
 }
 
-#[derive(Default, Clone, Debug)]
-pub struct HealEntryResult {
-    pub bytes: usize,
-    pub success: bool,
-    pub skipped: bool,
-    pub entry_done: bool,
-    pub name: String,
-}
-
 fn is_object_dangling(
     meta_arr: &[FileInfo],
     errs: &[Option<DiskError>],
@@ -5297,6 +5409,7 @@ pub fn is_valid_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is a cold storage tier that requires special handling
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_cold_storage_class(storage_class: &str) -> bool {
     matches!(
         storage_class,
@@ -5305,6 +5418,7 @@ pub fn is_cold_storage_class(storage_class: &str) -> bool {
 }
 
 /// Returns true if the storage class is an infrequent access tier
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub fn is_infrequent_access_class(storage_class: &str) -> bool {
     matches!(
         storage_class,

crates/ecstore/src/set_disk/ops/heal.rs

@@ -453,7 +453,9 @@ impl SetDisks {
             ..Default::default()
         };
 
-        let write_lock_guard = if !opts.no_lock {
+        // Bound, not `_`: this guard must live to the end of the scope. A bare
+        // `_` would drop it here and release the namespace write lock.
+        let _write_lock_guard = if !opts.no_lock {
             let ns_lock = self.new_ns_lock(bucket, object).await?;
             Some(
                 ns_lock
@@ -996,7 +998,7 @@ impl SetDisks {
                                                 readers.push(None);
                                                 continue;
                                             }
-                                            Err(e) => {
+                                            Err(_e) => {
                                                 readers.push(None);
                                                 continue;
                                             }
@@ -1545,6 +1547,9 @@ impl SetDisks {
 
         for candidate in candidates.iter_mut().filter(|candidate| candidate.local_payload) {
             for (disk_index, disk) in disks.iter().enumerate() {
+                // Only the #[cfg(test)] fault-injection branch below reads this.
+                #[cfg(not(test))]
+                let _ = disk_index;
                 let Some(disk) = disk else {
                     return Ok(DanglingDeleteSafety::UnsafeToDelete);
                 };
@@ -1716,6 +1721,10 @@ impl SetDisks {
         Ok((result, None))
     }
 
+    #[allow(
+        dead_code,
+        reason = "lock-taking wrapper over the live heal_object_dir_locked; only comments reference it (backlog#1823)"
+    )]
     #[tracing::instrument(level = "trace", skip(self), fields(bucket = %bucket, object = %object))]
     pub(in crate::set_disk) async fn heal_object_dir(
         &self,

crates/ecstore/src/set_disk/ops/locking.rs

@@ -66,6 +66,7 @@ impl crate::storage_api_contracts::namespace::NamespaceLocking for SetDisks {
 }
 
 impl SetDisks {
+    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error(&self, bucket: &str, object: &str, mode: &str, err: &LockResult) -> String {
         match err {
             LockResult::Timeout => {
@@ -79,6 +80,7 @@ impl SetDisks {
         }
     }
 
+    #[allow(dead_code, reason = "lock diagnostics formatter with no caller in this port (backlog#1823)")]
     pub(in crate::set_disk) fn format_lock_error_from_error(
         &self,
         bucket: &str,
@@ -143,6 +145,7 @@ impl SetDisks {
         disks
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(in crate::set_disk) async fn get_online_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot.strict_online_candidates().into_iter().map(Some).collect::<Vec<_>>();
@@ -153,6 +156,10 @@ impl SetDisks {
         disks
     }
 
+    #[allow(
+        dead_code,
+        reason = "local-only sibling of the test-covered get_online_disks; no caller in this port (backlog#1823)"
+    )]
     pub(in crate::set_disk) async fn get_online_local_disks(&self) -> Vec<Option<DiskStore>> {
         let snapshot = self.drive_membership_snapshot().await;
         let mut disks = snapshot
@@ -432,6 +439,10 @@ impl SetDisks {
         Ok((disk, fm))
     }
 
+    #[allow(
+        dead_code,
+        reason = "MinIO-parity healing-disk accessor with no caller in this port (backlog#1823)"
+    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing(
         &self,
         incl_healing: bool,
@@ -440,6 +451,10 @@ impl SetDisks {
         Ok((new_disks, healing > 0))
     }
 
+    #[allow(
+        dead_code,
+        reason = "reached only from get_online_disk_with_healing, itself uncalled in this port (backlog#1823)"
+    )]
     pub(in crate::set_disk) async fn get_online_disk_with_healing_and_info(
         &self,
         incl_healing: bool,

crates/ecstore/src/set_disk/ops/multipart.rs

@@ -415,6 +415,7 @@ fn reduce_quorum_part_numbers(object_parts: Vec<Vec<String>>, read_quorum: usize
 /// never returned, but flips `is_truncated` to `true` and yields a
 /// `next_upload_id_marker` pointing at the last returned upload so the caller can
 /// resume paging.
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn paginate_upload_page(remaining: &[MultipartInfo], max_uploads: usize) -> (Vec<MultipartInfo>, bool, Option<String>) {
     let is_truncated = remaining.len() > max_uploads;
     let page: Vec<MultipartInfo> = remaining.iter().take(max_uploads).cloned().collect();
@@ -557,6 +558,7 @@ impl SetDisks {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     pub(super) async fn check_upload_id_exists(
         &self,
         bucket: &str,
@@ -1398,7 +1400,7 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
 
         let mut count = max_parts;
 
-        for (i, part) in object_parts.iter().enumerate() {
+        for part in object_parts.iter() {
             if let Some(err) = &part.error {
                 warn!("list_object_parts part error: {:?}", &err);
             }
@@ -2041,8 +2043,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                     && let Err(err) = checksum.add_part(&cs, ext_part.actual_size)
                 {
                     error!(
-                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}",
-                        p.part_num, bucket, object
+                        "complete_multipart_upload checksum add_part failed part_id={}, bucket={}, object={}, err={}",
+                        p.part_num, bucket, object, err
                     );
                     return Err(Error::InvalidPart(p.part_num, ext_part.etag.clone(), p.etag.clone().unwrap_or_default()));
                 }
@@ -2087,8 +2089,8 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 }
             } else if let Err(err) = wtcs.matches(&checksum_combined, uploaded_parts.len() as i32) {
                 error!(
-                    "complete_multipart_upload checksum matches failed want={}, got={}",
-                    wtcs.encoded, checksum.encoded
+                    "complete_multipart_upload checksum matches failed want={}, got={}, err={}",
+                    wtcs.encoded, checksum.encoded, err
                 );
                 return Err(Error::other(format!(
                     "complete_multipart_upload checksum matches failed want={}, got={}",

crates/ecstore/src/set_disk/ops/object.rs

@@ -3507,6 +3507,10 @@ struct TransitionUploadedSaveProbeState {
 }
 
 #[cfg(test)]
+#[allow(
+    dead_code,
+    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+)]
 struct TransitionUploadedSaveProbe {
     state: Arc<TransitionUploadedSaveProbeState>,
 }
@@ -3517,6 +3521,10 @@ static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Opti
 
 #[cfg(test)]
 impl TransitionUploadedSaveProbe {
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(TransitionUploadedSaveProbeState {
             bucket: bucket.to_string(),
@@ -3533,6 +3541,10 @@ impl TransitionUploadedSaveProbe {
         Self { state }
     }
 
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn attempts(&self) -> usize {
         self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
     }
@@ -3738,6 +3750,10 @@ struct TransitionCommitBarrierState {
 }
 
 #[cfg(test)]
+#[allow(
+    dead_code,
+    reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+)]
 struct TransitionCommitBarrier {
     state: Arc<TransitionCommitBarrierState>,
 }
@@ -3748,14 +3764,26 @@ static TRANSITION_COMMIT_BARRIER: std::sync::OnceLock<std::sync::Mutex<Option<Ar
 
 #[cfg(test)]
 impl TransitionCommitBarrier {
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn install_before_lock_lost_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLockLost)
     }
 
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn install(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::BeforeLeaseValidation)
     }
 
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn install_after_lease_check(bucket: &str, object: &str) -> Self {
         Self::install_at(bucket, object, TransitionCommitPause::AfterLeaseValidation)
     }
@@ -3778,12 +3806,20 @@ impl TransitionCommitBarrier {
         Self { state }
     }
 
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     async fn wait_until_paused(&self) {
         tokio::time::timeout(Duration::from_secs(30), self.state.arrived.notified())
             .await
             .expect("transition should reach the deterministic commit barrier");
     }
 
+    #[allow(
+        dead_code,
+        reason = "installed by set_disk tests behind `--features test-util` (backlog#1823)"
+    )]
     fn release(&self) {
         self.state.release.notify_one();
     }
@@ -5620,7 +5656,9 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         // TODO: Lifecycle
 
         let mut version_found = true;
-        let (mut goi, write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
+        // delete_object_version below derives its own majority quorum from the
+        // disk array, so the object-derived quorum here is unused.
+        let (mut goi, _write_quorum, gerr) = self.get_object_info_and_quorum(bucket, object, &opts).await;
         if let Some(err) = &gerr
             && goi.name.is_empty()
         {
@@ -6374,7 +6412,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         self.record_capacity_scope_if_needed(opts.capacity_scope_token, &disks);
 
         for disk in disks.iter() {
-            if let Some(disk) = disk {
+            if disk.is_some() {
                 continue;
             }
             let _ = self
@@ -10170,6 +10208,288 @@ mod transition_upload_integrity_tests {
         assert!(backend.contains(remote_object).await, "committed remote object should remain available");
     }
 
+    /// Compresses `plaintext` with the codec the PUT path uses, so the stored
+    /// bytes round-trip through the read path's decompressor.
+    async fn compress_for_storage(plaintext: &[u8]) -> Vec<u8> {
+        let mut reader = crate::io_support::rio::compression_reader(
+            Cursor::new(plaintext.to_vec()),
+            rustfs_utils::CompressionAlgorithm::default(),
+            false,
+        );
+        let mut compressed = Vec::new();
+        reader.read_to_end(&mut compressed).await.expect("plaintext should compress");
+        assert!(compressed.len() < plaintext.len(), "test payload must actually compress");
+        compressed
+    }
+
+    /// Writes a genuinely compressed object: stored data is `compressed`, and the
+    /// metadata marks it compressed with the plaintext length as its actual size,
+    /// exactly as the app-layer compress path records it.
+    async fn write_compressed_source(
+        set_disks: &Arc<SetDisks>,
+        disk_stores: &[DiskStore],
+        bucket: &str,
+        object: &str,
+        plaintext: &[u8],
+        compressed: &[u8],
+    ) -> ObjectInfo {
+        for disk in disk_stores {
+            disk.make_volume(bucket).await.expect("bucket volume should be created");
+        }
+        let mut user_defined = HashMap::new();
+        rustfs_utils::http::insert_str(
+            &mut user_defined,
+            rustfs_utils::http::SUFFIX_COMPRESSION,
+            crate::io_support::rio::compression_metadata_value(rustfs_utils::CompressionAlgorithm::default()),
+        );
+        rustfs_utils::http::insert_str(&mut user_defined, rustfs_utils::http::SUFFIX_ACTUAL_SIZE, plaintext.len().to_string());
+        let stream = crate::io_support::rio::HashReader::from_stream(
+            Cursor::new(compressed.to_vec()),
+            compressed.len() as i64,
+            plaintext.len() as i64,
+            None,
+            None,
+            false,
+        )
+        .expect("hash reader over compressed bytes");
+        let mut reader = PutObjReader::new(stream);
+        set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    no_lock: true,
+                    user_defined,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("compressed object should be written")
+    }
+
+    async fn read_transitioned(
+        set_disks: &Arc<SetDisks>,
+        bucket: &str,
+        object: &str,
+        range: Option<HTTPRangeSpec>,
+        opts: &ObjectOptions,
+    ) -> (Vec<u8>, i64) {
+        let mut reader = set_disks
+            .get_object_reader(bucket, object, range, HeaderMap::new(), opts)
+            .await
+            .expect("transitioned object reader should open");
+        let published_size = reader.object_info.size;
+        let mut body = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut body)
+            .await
+            .expect("transitioned body should drain");
+        (body, published_size)
+    }
+
+    /// Transition uploads the object's STORED bytes, so a tiered read has to
+    /// apply the same transform an erasure read would. #6107 routed this path
+    /// through `ReadPlan` to stop serving an encrypted object's ciphertext;
+    /// compression rides the same plan, and nothing pinned it (backlog#1851).
+    /// Without the transform this GET returns the compressed bytes under the
+    /// compressed size — silent corruption for every client of a compressed
+    /// object that ILM has moved to a warm tier.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn transitioned_compressed_object_get_returns_plaintext() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transitioned-compressed-get-bucket";
+        let object = "object.txt";
+        let plaintext = b"transitioned compressed objects must decompress on read ".repeat(20_000);
+        let compressed = compress_for_storage(&plaintext).await;
+        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
+
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let (local_body, local_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
+        assert_eq!(local_body, plaintext, "control: the pre-transition read must decompress");
+        assert_eq!(
+            local_size,
+            plaintext.len() as i64,
+            "control: the pre-transition read publishes the plaintext size"
+        );
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("transition should commit");
+
+        let put_versions = backend.put_versions().await;
+        assert_eq!(put_versions.len(), 1, "transition should upload one remote candidate");
+        let remote_bytes = backend
+            .bytes(&put_versions[0].0)
+            .await
+            .expect("remote candidate should be stored");
+        assert_eq!(
+            remote_bytes, compressed,
+            "transition uploads the stored representation; the read side is what has to decode it"
+        );
+
+        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
+        assert_eq!(body, plaintext, "a tiered read must return the object's content, not its stored bytes");
+        assert_eq!(
+            published_size,
+            plaintext.len() as i64,
+            "a tiered read must publish the plaintext size, not the compressed one"
+        );
+    }
+
+    /// A ranged tiered read is expressed in plaintext coordinates, so the plan
+    /// has to translate it into the remote copy's compressed extent and skip
+    /// into the decompressed stream — the same translation the erasure path does.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn transitioned_compressed_object_range_get_returns_plaintext_slice() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transitioned-compressed-range-bucket";
+        let object = "object.txt";
+        let plaintext = b"ranged reads of transitioned compressed objects must land in plaintext ".repeat(20_000);
+        let compressed = compress_for_storage(&plaintext).await;
+        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("transition should commit");
+
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        // Deliberately past the compressed size, so a range still measured in
+        // stored coordinates could not produce this slice.
+        let start = compressed.len() as i64 + 4096;
+        let end = start + 511;
+        let range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start,
+            end,
+        };
+        let (body, published_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
+
+        let expected = &plaintext[start as usize..=end as usize];
+        assert_eq!(body, expected, "a ranged tiered read must return that plaintext slice");
+        assert_eq!(published_size, expected.len() as i64, "a ranged tiered read publishes the slice length");
+    }
+
+    /// The restore copy-back re-writes the object under its original metadata,
+    /// which still says "compressed". It therefore has to keep receiving the
+    /// STORED bytes: `restore_request_active` holds it on the plan's `Plain`
+    /// branch, and decompressing there would write plaintext under compressed
+    /// metadata.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn restore_read_of_transitioned_compressed_object_keeps_stored_bytes() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transitioned-compressed-restore-bucket";
+        let object = "object.txt";
+        let plaintext = b"restore copy-back must keep the stored representation intact ".repeat(20_000);
+        let compressed = compress_for_storage(&plaintext).await;
+        let original = write_compressed_source(&set_disks, &disk_stores, bucket, object, &plaintext, &compressed).await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("transition should commit");
+
+        let oi = set_disks
+            .get_object_info(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("transitioned metadata should resolve");
+        let restore_opts = ObjectOptions {
+            no_lock: true,
+            part_number: Some(1),
+            transition: TransitionOptions {
+                restore_request: s3s::dto::RestoreRequest {
+                    days: Some(1),
+                    ..Default::default()
+                },
+                ..Default::default()
+            },
+            ..Default::default()
+        };
+        let mut reader = get_transitioned_object_reader_with_tier_manager(
+            bucket,
+            object,
+            &None,
+            &HeaderMap::new(),
+            &oi,
+            &restore_opts,
+            &set_disks.ctx.tier_config_mgr(),
+            set_disks.ctx.object_encryption_resolver(),
+        )
+        .await
+        .expect("restore read of the tiered copy should open");
+        let published_size = reader.object_info.size;
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("restore body should drain");
+
+        assert_eq!(body, compressed, "a restore read must copy the stored bytes back verbatim");
+        assert_eq!(
+            published_size,
+            compressed.len() as i64,
+            "a restore read must keep publishing the stored size"
+        );
+    }
+
+    /// Plain objects must keep streaming the remote bytes through untouched:
+    /// their plan is `Plain`, so the tiered read stays byte-identical.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn transitioned_plain_object_get_is_unchanged() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transitioned-plain-get-bucket";
+        let object = "object.bin";
+        let payload = b"plain transitioned objects must keep reading back byte-identical ".repeat(1024);
+        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
+
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("transition should commit");
+
+        let opts = ObjectOptions {
+            no_lock: true,
+            ..Default::default()
+        };
+        let (body, published_size) = read_transitioned(&set_disks, bucket, object, None, &opts).await;
+        assert_eq!(body, payload);
+        assert_eq!(published_size, payload.len() as i64);
+
+        let range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 100,
+            end: 611,
+        };
+        let (ranged_body, ranged_size) = read_transitioned(&set_disks, bucket, object, Some(range), &opts).await;
+        assert_eq!(ranged_body, &payload[100..=611]);
+        assert_eq!(ranged_size, payload.len() as i64, "a plain ranged read keeps publishing the object size");
+    }
+
     async fn corrupt_beyond_read_quorum(
         temp_dirs: &[tempfile::TempDir],
         bucket: &str,

crates/ecstore/src/set_disk/read.rs

@@ -116,6 +116,7 @@ impl SetDisks {
             .then_some(GET_METADATA_CACHE_REASON_DIST_ERASURE)
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     async fn cached_get_object_fileinfo(&self, bucket: &str, object: &str) -> Option<Arc<GetObjectMetadataCacheEntry>> {
         match self.lookup_cached_get_object_fileinfo(bucket, object).await {
             MetadataCacheLookup::Hit(entry) => Some(entry),
@@ -1826,6 +1827,7 @@ fn get_object_metadata_cache_request_bypass_reason(bucket: &str, opts: &ObjectOp
         .then_some(GET_METADATA_CACHE_REASON_META_BUCKET)
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn is_get_object_metadata_cache_request_eligible(bucket: &str, opts: &ObjectOptions, read_data: bool) -> bool {
     get_object_metadata_cache_request_bypass_reason(bucket, opts, read_data).is_none()
 }
@@ -3935,7 +3937,7 @@ mod tests {
     }
 
     #[test]
-    fn metadata_early_stop_bounded_fanout_defaults_to_disabled() {
+    fn metadata_early_stop_bounded_fanout_defaults_to_enabled() {
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_EARLY_STOP_ENABLE, Some("true")),
@@ -3944,20 +3946,20 @@ mod tests {
             ],
             || {
                 assert!(is_get_metadata_data_read_early_stop_enabled());
-                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
-        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true"))], || {
-            assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
+        temp_env::with_vars([(ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false"))], || {
+            assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
         });
         temp_env::with_vars(
             [
                 (ENV_RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE, Some("false")),
-                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("false")),
+                (ENV_RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT, Some("true")),
             ],
             || {
                 assert!(!is_get_metadata_data_read_early_stop_enabled());
-                assert!(!is_get_metadata_early_stop_bounded_fanout_enabled());
+                assert!(is_get_metadata_early_stop_bounded_fanout_enabled());
             },
         );
     }

crates/io-metrics/src/lib.rs

@@ -487,22 +487,21 @@ pub fn record_get_object_completion(total_duration_secs: f64, response_size_byte
 
 /// Record the streaming strategy chosen for a GetObject response body.
 #[inline(always)]
-pub fn record_get_object_stream_strategy(strategy: &str, buffer_size_bytes: usize, response_size_bytes: i64) {
+pub fn record_get_object_stream_strategy(strategy: &'static str, buffer_size_bytes: usize, response_size_bytes: i64) {
     if !get_stage_metrics_enabled() {
         return;
     }
-    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy.to_string()).increment(1);
-    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy.to_string())
-        .record(usize_to_f64(buffer_size_bytes));
-    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy.to_string())
+    counter!("rustfs_io_get_object_stream_strategy_total", "strategy" => strategy).increment(1);
+    histogram!("rustfs_io_get_object_stream_buffer_size_bytes", "strategy" => strategy).record(usize_to_f64(buffer_size_bytes));
+    histogram!("rustfs_io_get_object_stream_response_size_bytes", "strategy" => strategy)
         .record(i64_non_negative_to_f64(response_size_bytes));
 }
 
 /// Record the response-body handoff shape from a GetObject reader into the S3 streaming body.
 #[inline(always)]
 pub fn record_get_object_response_handoff(
-    strategy: &str,
-    buffer_source: &str,
+    strategy: &'static str,
+    buffer_source: &'static str,
     buffer_size_bytes: usize,
     response_size_bytes: i64,
     duration_secs: f64,
@@ -512,26 +511,26 @@ pub fn record_get_object_response_handoff(
     }
     counter!(
         "rustfs_io_get_object_response_handoff_total",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string()
+        "strategy" => strategy,
+        "buffer_source" => buffer_source
     )
     .increment(1);
     histogram!(
         "rustfs_io_get_object_response_handoff_buffer_size_bytes",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string()
+        "strategy" => strategy,
+        "buffer_source" => buffer_source
     )
     .record(usize_to_f64(buffer_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_response_size_bytes",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string()
+        "strategy" => strategy,
+        "buffer_source" => buffer_source
     )
     .record(i64_non_negative_to_f64(response_size_bytes));
     histogram!(
         "rustfs_io_get_object_response_handoff_duration_seconds",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string()
+        "strategy" => strategy,
+        "buffer_source" => buffer_source
     )
     .record(duration_secs);
     record_get_object_response_handoff_duration("s3_handler", duration_secs);
@@ -539,14 +538,18 @@ pub fn record_get_object_response_handoff(
 
 /// Record ReaderStream capacity chosen for GetObject handoff.
 #[inline(always)]
-pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source: &str, buffer_size_bytes: usize) {
+pub fn record_get_object_reader_stream_buffer_size(
+    strategy: &'static str,
+    buffer_source: &'static str,
+    buffer_size_bytes: usize,
+) {
     if !get_stage_metrics_enabled() {
         return;
     }
     histogram!(
         "rustfs_io_get_object_reader_stream_buffer_size_bytes",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string()
+        "strategy" => strategy,
+        "buffer_source" => buffer_source
     )
     .record(usize_to_f64(buffer_size_bytes));
 }
@@ -554,8 +557,8 @@ pub fn record_get_object_reader_stream_buffer_size(strategy: &str, buffer_source
 /// Record ReaderStream poll outcomes for GetObject handoff attribution.
 #[inline(always)]
 pub fn record_get_object_reader_stream_poll(
-    strategy: &str,
-    buffer_source: &str,
+    strategy: &'static str,
+    buffer_source: &'static str,
     outcome: &'static str,
     remaining_before: usize,
     bytes: usize,
@@ -567,36 +570,36 @@ pub fn record_get_object_reader_stream_poll(
     let bytes = u64::try_from(bytes).unwrap_or(u64::MAX);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_total",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string(),
+        "strategy" => strategy,
+        "buffer_source" => buffer_source,
         "outcome" => outcome
     )
     .increment(1);
     counter!(
         "rustfs_io_get_object_reader_stream_poll_bytes_total",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string(),
+        "strategy" => strategy,
+        "buffer_source" => buffer_source,
         "outcome" => outcome
     )
     .increment(bytes);
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_remaining_bytes",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string(),
+        "strategy" => strategy,
+        "buffer_source" => buffer_source,
         "outcome" => outcome
     )
     .record(usize_to_f64(remaining_before));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_bytes",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string(),
+        "strategy" => strategy,
+        "buffer_source" => buffer_source,
         "outcome" => outcome
     )
     .record(usize_to_f64(bytes as usize));
     histogram!(
         "rustfs_io_get_object_reader_stream_poll_duration_seconds",
-        "strategy" => strategy.to_string(),
-        "buffer_source" => buffer_source.to_string(),
+        "strategy" => strategy,
+        "buffer_source" => buffer_source,
         "outcome" => outcome
     )
     .record(duration_secs);

crates/keystone/src/client.rs

@@ -31,7 +31,10 @@ pub struct KeystoneClient {
     admin_password: Option<String>,
     admin_project: Option<String>,
     admin_domain: String,
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "TLS verification flag parsed from config; the reqwest client is built before it is consulted, so nothing reads it back (backlog#1823)"
+    )]
     verify_ssl: bool,
     /// Request timeout applied to the underlying HTTP client.
     timeout: std::time::Duration,

crates/keystone/src/identity.rs

@@ -20,7 +20,10 @@ use tracing::{debug, info};
 
 /// Maps Keystone identities to RustFS concepts
 pub struct KeystoneIdentityMapper {
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "keeps the Keystone client alive for the mapper's lifetime; the mapping paths do not call through it yet (backlog#1823)"
+    )]
     client: Arc<KeystoneClient>,
     role_policy_map: HashMap<String, String>,
     enable_tenant_prefix: bool,

crates/kms/src/api_types.rs

@@ -293,6 +293,15 @@ enum StrictVaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
+    Kubernetes {
+        role: String,
+        #[serde(default)]
+        mount: Option<String>,
+        #[serde(default)]
+        jwt_path: Option<std::path::PathBuf>,
+        #[serde(default)]
+        refresh_safety_window_secs: Option<u64>,
+    },
     TokenFile {
         path: std::path::PathBuf,
         #[serde(default)]
@@ -319,6 +328,17 @@ impl From<StrictVaultAuthMethod> for VaultAuthMethod {
                 mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_APPROLE_MOUNT.to_string()),
                 refresh_safety_window_secs,
             },
+            StrictVaultAuthMethod::Kubernetes {
+                role,
+                mount,
+                jwt_path,
+                refresh_safety_window_secs,
+            } => Self::Kubernetes {
+                role,
+                mount: mount.unwrap_or_else(|| crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()),
+                jwt_path: jwt_path.unwrap_or_else(|| std::path::PathBuf::from(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH)),
+                refresh_safety_window_secs,
+            },
             StrictVaultAuthMethod::TokenFile {
                 path,
                 poll_interval_secs,
@@ -499,6 +519,7 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
+                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -513,6 +534,7 @@ impl From<&KmsConfig> for KmsConfigSummary {
                 auth_method_type: match &vault_config.auth_method {
                     VaultAuthMethod::Token { .. } => "token".to_string(),
                     VaultAuthMethod::AppRole { .. } => "approle".to_string(),
+                    VaultAuthMethod::Kubernetes { .. } => "kubernetes".to_string(),
                     VaultAuthMethod::TokenFile { .. } => "token_file".to_string(),
                 },
                 has_stored_credentials: true,
@@ -901,6 +923,42 @@ mod tests {
         assert!(request.to_kms_config().validate().is_ok());
     }
 
+    /// The admin API reaches Kubernetes auth with the role alone; the mount and
+    /// the projected token path fall back to the cluster defaults, so a Tenant
+    /// manifest carries no credential and no cluster-specific paths.
+    #[test]
+    fn test_deserialize_vault_configure_request_accepts_kubernetes_auth() {
+        let raw = serde_json::json!({
+            "backend_type": "vault-transit",
+            "address": "https://vault.example.com:8200",
+            "mount_path": "rustfs",
+            "auth_method": { "Kubernetes": { "role": "rustfs" } }
+        });
+
+        let request: ConfigureKmsRequest = serde_json::from_value(raw).expect("kubernetes auth should deserialize");
+        let config = request.to_kms_config();
+        config.validate().expect("kubernetes auth must validate");
+
+        let vault = config.vault_transit_config().expect("vault transit backend config");
+        let VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } = &vault.auth_method
+        else {
+            panic!("expected Kubernetes auth, got {:?}", vault.auth_method);
+        };
+        assert_eq!(role, "rustfs");
+        assert_eq!(mount, crate::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
+        assert_eq!(jwt_path, std::path::Path::new(crate::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
+
+        let unknown_field = serde_json::json!({
+            "backend_type": "vault-transit",
+            "address": "https://vault.example.com:8200",
+            "auth_method": { "Kubernetes": { "role": "rustfs", "service_account": "rustfs" } }
+        });
+        serde_json::from_value::<ConfigureKmsRequest>(unknown_field)
+            .expect_err("an unknown auth field must be rejected rather than silently dropped");
+    }
+
     #[test]
     fn test_deserialize_aws_configure_request_accepts_type_aliases() {
         for backend_type in ["AWS", "AwsKms", "aws", "aws-kms", "aws_kms"] {

crates/kms/src/backends/vault.rs

@@ -550,6 +550,7 @@ impl VaultKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backends/vault_credentials.rs

@@ -326,6 +326,97 @@ impl fmt::Debug for AppRoleLogin {
     }
 }
 
+/// Token source for [`VaultAuthMethod::Kubernetes`]: exchanges the pod's
+/// projected ServiceAccount token for a lease-bound Vault token.
+///
+/// The JWT is re-read on every login because the kubelet rotates a projected
+/// token well inside the pod's lifetime; caching it would strand the source on
+/// an expired assertion once the current Vault token can no longer be renewed.
+///
+/// Unlike [`TokenFileSource`], the file mode is not checked: the kubelet owns
+/// the projected token and mounts it world-readable by default, so rejecting
+/// group/other bits would refuse every standard pod rather than catch a
+/// deployment error.
+pub(crate) struct KubernetesLogin {
+    /// Unauthenticated client used only for the login exchange.
+    login_client: VaultClient,
+    mount: String,
+    role: String,
+    jwt_path: PathBuf,
+}
+
+impl KubernetesLogin {
+    pub(crate) fn new(settings: &VaultConnectionSettings, mount: String, role: String, jwt_path: PathBuf) -> Result<Self> {
+        Ok(Self {
+            login_client: settings.build_login_client()?,
+            mount,
+            role,
+            jwt_path,
+        })
+    }
+
+    /// Read the ServiceAccount token for one login attempt.
+    ///
+    /// Mirrors [`AppRoleLogin::resolve_secret_id`]: a read failure is fatal for
+    /// the attempt but the refresh loop keeps retrying, so a token the kubelet
+    /// has not projected yet heals the source without a restart.
+    async fn resolve_jwt(&self) -> AttemptResult<SecretString> {
+        let mut raw = tokio::fs::read_to_string(&self.jwt_path)
+            .await
+            .map_err(|error| AttemptError {
+                class: ErrorClass::Fatal,
+                error: KmsError::configuration_error(format!(
+                    "Failed to read Kubernetes ServiceAccount token {}: {error}",
+                    self.jwt_path.display()
+                )),
+            })?;
+        let trimmed = raw.trim();
+        if trimmed.is_empty() {
+            raw.zeroize();
+            return Err(AttemptError {
+                class: ErrorClass::Fatal,
+                error: KmsError::configuration_error(format!(
+                    "Kubernetes ServiceAccount token {} is empty",
+                    self.jwt_path.display()
+                )),
+            });
+        }
+        let jwt = SecretString::new(trimmed.to_string());
+        raw.zeroize();
+        Ok(jwt)
+    }
+}
+
+#[async_trait]
+impl TokenSource for KubernetesLogin {
+    async fn acquire(&self) -> AttemptResult<TokenLease> {
+        let jwt = self.resolve_jwt().await?;
+        let auth = vaultrs::auth::kubernetes::login(&self.login_client, &self.mount, &self.role, jwt.expose())
+            .await
+            .map_err(|error| attempt_error("Kubernetes login", error))?;
+        Ok(TokenLease::from_auth(auth))
+    }
+
+    async fn renew(&self, client: &VaultClient) -> AttemptResult<TokenLease> {
+        let auth = vaultrs::token::renew_self(client, None)
+            .await
+            .map_err(|error| attempt_error("token renewal", error))?;
+        Ok(TokenLease::from_auth(auth))
+    }
+}
+
+impl fmt::Debug for KubernetesLogin {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        // The login client embeds Vault client settings and must stay out of
+        // Debug output; the role name is not a secret, and the JWT is never held.
+        f.debug_struct("KubernetesLogin")
+            .field("mount", &self.mount)
+            .field("role", &self.role)
+            .field("jwt_path", &self.jwt_path)
+            .finish_non_exhaustive()
+    }
+}
+
 /// Token source for [`VaultAuthMethod::TokenFile`]: reads an agent-managed
 /// token file (for example a Vault Agent auto-auth sink).
 ///
@@ -464,6 +555,9 @@ pub(crate) fn token_source_for(
             secret_id.clone(),
             secret_id_file.clone(),
         )?)),
+        VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } => Ok(Box::new(KubernetesLogin::new(settings, mount.clone(), role.clone(), jwt_path.clone())?)),
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -486,6 +580,9 @@ pub(crate) struct VaultConnectionSettings {
     pub(crate) namespace: Option<String>,
     /// Per-attempt HTTP timeout applied to the underlying reqwest client.
     pub(crate) attempt_timeout: Duration,
+    /// Whether to accept an unverified Vault server certificate. Gated on
+    /// `allow_insecure_dev_defaults` by `KmsConfig::validate`.
+    pub(crate) skip_tls_verify: bool,
 }
 
 impl VaultConnectionSettings {
@@ -499,6 +596,11 @@ impl VaultConnectionSettings {
         // operation-level retry policy.
         settings_builder.timeout(Some(self.attempt_timeout));
         settings_builder.token(token);
+        // Always set explicitly: left unset, vaultrs derives this from its own
+        // VAULT_SKIP_VERIFY variable, so a stray value in the environment would
+        // disable certificate verification behind the KMS configuration and its
+        // insecure-defaults gate.
+        settings_builder.verify(!self.skip_tls_verify);
 
         if let Some(namespace) = &self.namespace {
             settings_builder.namespace(Some(namespace.clone()));
@@ -551,6 +653,10 @@ impl VaultCredentialPolicy {
                 refresh_safety_window_secs: Some(secs),
                 ..
             }
+            | VaultAuthMethod::Kubernetes {
+                refresh_safety_window_secs: Some(secs),
+                ..
+            }
             | VaultAuthMethod::TokenFile {
                 refresh_safety_window_secs: Some(secs),
                 ..
@@ -584,15 +690,25 @@ pub(crate) struct VaultClientHandle {
 
 impl VaultClientHandle {
     /// Absolute expiry of this generation's token.
+    ///
+    /// `lease.ttl` is built from the `lease_duration` the Vault server sent, so
+    /// a value too large to add to `issued_at` would panic on the bare `+`. A
+    /// TTL that cannot be represented is indistinguishable from no expiry, so it
+    /// collapses to `None` — the same answer already given for the zero-lease
+    /// tokens Vault issues, which keeps the token in use and still fully
+    /// validated by Vault on every call.
     fn expires_at(&self) -> Option<Instant> {
-        self.lease.map(|lease| self.issued_at + lease.ttl)
+        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl))
     }
 
     /// When the renewal task should refresh this generation: half the TTL,
     /// leaving the second half as budget for retries before the fail-closed
     /// window is reached.
+    ///
+    /// Unrepresentable TTLs collapse to `None` as in [`Self::expires_at`],
+    /// leaving a token that never expires with nothing to renew.
     fn renew_at(&self) -> Option<Instant> {
-        self.lease.map(|lease| self.issued_at + lease.ttl / 2)
+        self.lease.and_then(|lease| self.issued_at.checked_add(lease.ttl / 2))
     }
 }
 
@@ -662,7 +778,7 @@ impl VaultCredentialProvider {
         let handle = self.current.load_full();
         if let Some(expires_at) = handle.expires_at() {
             let now = Instant::now();
-            if now + self.policy.safety_window >= expires_at {
+            if self.inside_safety_window(now, expires_at) {
                 return Err(KmsError::credentials_unavailable(format!(
                     "Vault token (generation {}) is within {:?} of expiry and has not been refreshed; refusing to use it",
                     handle.generation, self.policy.safety_window
@@ -672,6 +788,18 @@ impl VaultCredentialProvider {
         Ok(handle)
     }
 
+    /// Whether the token expiring at `expires_at` is close enough to refuse.
+    ///
+    /// `safety_window` reaches here from persisted configuration, so it is not
+    /// guaranteed to have passed this version's validation: a window too large
+    /// to add to the current instant would panic on the bare `+`. Such a window
+    /// means every token is always inside it, so saturating to "refuse" is both
+    /// the fail-closed answer and the one the arithmetic was reaching for.
+    fn inside_safety_window(&self, now: Instant, expires_at: Instant) -> bool {
+        now.checked_add(self.policy.safety_window)
+            .is_none_or(|deadline| deadline >= expires_at)
+    }
+
     /// Publish the credential gauges for the generation currently installed.
     ///
     /// The fail-closed gauge re-evaluates the very gate
@@ -683,7 +811,7 @@ impl VaultCredentialProvider {
         let fail_closed = match handle.expires_at() {
             Some(expires_at) => {
                 metrics::gauge!(METRIC_TOKEN_TTL_SECONDS).set(expires_at.saturating_duration_since(now).as_secs_f64());
-                now + self.policy.safety_window >= expires_at
+                self.inside_safety_window(now, expires_at)
             }
             // A generation without an expiry has no remaining TTL to report
             // and can never lapse, so it can never fail closed either.
@@ -860,7 +988,7 @@ impl Drop for CredentialTaskHandle {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::config::REDACTED_SECRET;
+    use crate::config::{DEFAULT_VAULT_KUBERNETES_MOUNT, REDACTED_SECRET};
     use std::sync::atomic::{AtomicBool, AtomicU32, Ordering};
 
     const TEST_TOKEN: &str = "vault-token-debug-leak-canary";
@@ -871,6 +999,7 @@ mod tests {
             address: "http://127.0.0.1:8200".to_string(),
             namespace: Some("team-namespace".to_string()),
             attempt_timeout: Duration::from_secs(30),
+            skip_tls_verify: false,
         }
     }
 
@@ -1057,6 +1186,143 @@ mod tests {
         assert!(format!("{source:?}").contains("AppRoleLogin"));
     }
 
+    #[tokio::test]
+    async fn test_kubernetes_auth_method_maps_to_login_source() {
+        let settings = test_settings();
+        let source = token_source_for(&VaultAuthMethod::kubernetes("rustfs".to_string()), &settings)
+            .expect("kubernetes auth must map to a login source");
+
+        assert!(format!("{source:?}").contains("KubernetesLogin"));
+    }
+
+    /// `refresh_safety_window_secs` is operator-supplied and reaches the request
+    /// path from persisted configuration, so the fail-closed comparison must
+    /// survive a window too large to add to the current instant. Before the
+    /// checked arithmetic this panicked with "overflow when adding duration to
+    /// instant" on the first request after a lease-bearing login.
+    #[tokio::test]
+    async fn test_current_refuses_rather_than_panics_on_an_unrepresentable_safety_window() {
+        let (provider, _state) = scripted_provider(
+            Duration::from_secs(60),
+            true,
+            test_policy(Duration::from_secs(u64::MAX), Duration::from_secs(5)),
+        )
+        .await;
+
+        let error = provider
+            .current()
+            .expect_err("a window wider than any lease must refuse the token");
+        assert!(
+            matches!(error, KmsError::CredentialsUnavailable { .. }),
+            "expected CredentialsUnavailable, got {error:?}"
+        );
+    }
+
+    /// `lease_duration` is a bare u64 straight off the Vault response and forms
+    /// the other side of the same comparison, so an absurd one must not panic
+    /// either. It is indistinguishable from a non-expiring token, which is how
+    /// the zero-lease case already behaves.
+    #[tokio::test]
+    async fn test_an_unrepresentable_lease_is_treated_as_non_expiring() {
+        let (provider, _state) = scripted_provider(
+            Duration::from_secs(u64::MAX),
+            true,
+            test_policy(Duration::from_secs(30), Duration::from_secs(5)),
+        )
+        .await;
+
+        provider
+            .current()
+            .expect("a token whose expiry cannot be represented must stay usable");
+    }
+
+    /// The configured flag has to reach the HTTP client, not just the config
+    /// struct: every generation (authenticated and login) builds its own client,
+    /// and a Vault with a self-signed certificate fails the handshake unless
+    /// each one carries the setting.
+    #[test]
+    fn test_skip_tls_verify_reaches_every_vault_client_generation() {
+        for skip_tls_verify in [false, true] {
+            let settings = VaultConnectionSettings {
+                address: "https://vault.example.com:8200".to_string(),
+                namespace: None,
+                attempt_timeout: Duration::from_secs(30),
+                skip_tls_verify,
+            };
+
+            let authenticated = settings.build_client(TEST_TOKEN).expect("authenticated client must build");
+            assert_eq!(authenticated.settings.verify, !skip_tls_verify);
+
+            let login = settings.build_login_client().expect("login client must build");
+            assert_eq!(login.settings.verify, !skip_tls_verify);
+        }
+    }
+
+    /// vaultrs derives `verify` from its own VAULT_SKIP_VERIFY variable when the
+    /// builder leaves it unset, which would disable certificate verification
+    /// without passing the KMS insecure-defaults gate.
+    #[test]
+    fn test_vaultrs_skip_verify_env_cannot_override_the_configured_setting() {
+        temp_env::with_var("VAULT_SKIP_VERIFY", Some("true"), || {
+            let client = test_settings().build_client(TEST_TOKEN).expect("client must build");
+            assert!(
+                client.settings.verify,
+                "a stray VAULT_SKIP_VERIFY must not disable verification behind the KMS configuration"
+            );
+        });
+    }
+
+    /// The projected token is read fresh per login attempt and trimmed, so a
+    /// kubelet rotation is picked up without a restart and a trailing newline
+    /// does not corrupt the assertion sent to Vault.
+    #[tokio::test]
+    async fn test_kubernetes_login_rereads_and_trims_the_service_account_token() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let path = dir.path().join("token");
+        tokio::fs::write(&path, "  first-jwt\n").await.expect("write token");
+
+        let login = KubernetesLogin::new(
+            &test_settings(),
+            DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
+            "rustfs".to_string(),
+            path.clone(),
+        )
+        .expect("login source must build");
+
+        assert_eq!(login.resolve_jwt().await.expect("first read").expose(), "first-jwt");
+
+        tokio::fs::write(&path, "rotated-jwt").await.expect("rotate token");
+        assert_eq!(
+            login.resolve_jwt().await.expect("second read").expose(),
+            "rotated-jwt",
+            "a rotated projected token must be picked up without a restart"
+        );
+    }
+
+    /// The ServiceAccount token is re-read per attempt, so an unreadable or
+    /// empty one fails that attempt without reaching Vault; the refresh loop
+    /// keeps retrying, which is what lets a late projection heal the source.
+    #[tokio::test]
+    async fn test_kubernetes_login_rejects_an_unusable_service_account_token() {
+        let dir = tempfile::tempdir().expect("temp dir");
+        let missing = dir.path().join("absent-token");
+        let empty = dir.path().join("empty-token");
+        tokio::fs::write(&empty, "  \n").await.expect("write empty token");
+
+        for (path, expected) in [(missing, "Failed to read"), (empty, "is empty")] {
+            let login =
+                KubernetesLogin::new(&test_settings(), DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(), "rustfs".to_string(), path)
+                    .expect("login source must build");
+
+            let error = login
+                .acquire()
+                .await
+                .expect_err("an unusable ServiceAccount token must fail the attempt");
+            assert!(matches!(error.class, ErrorClass::Fatal));
+            assert!(error.error.to_string().contains(expected), "got {}", error.error);
+        }
+    }
+
     #[tokio::test(start_paused = true)]
     async fn test_renewal_task_renews_at_half_ttl() {
         let (provider, state) = scripted_provider(

crates/kms/src/backends/vault_transit.rs

@@ -415,6 +415,7 @@ impl VaultTransitKmsClient {
             address: config.address.clone(),
             namespace: config.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            skip_tls_verify: config.tls.as_ref().is_some_and(|tls| tls.skip_verify),
         };
         let source = token_source_for(&config.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/backup/vault_restore.rs

@@ -450,6 +450,10 @@ impl VaultRestoreClient {
             address: target.address.clone(),
             namespace: target.namespace.clone(),
             attempt_timeout: kms_config.effective_timeout(),
+            // A restore target carries no TLS settings, so certificates are
+            // always verified: recovery is the last path that should accept an
+            // unauthenticated Vault.
+            skip_tls_verify: false,
         };
         let source = token_source_for(&target.auth_method, &settings)?;
         let policy = VaultCredentialPolicy::from_kms_config(

crates/kms/src/config.rs

@@ -25,6 +25,10 @@ use url::Url;
 
 pub const ENV_KMS_ALLOW_INSECURE_DEV_DEFAULTS: &str = "RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS";
 pub const ENV_KMS_ALLOW_IMMEDIATE_DELETION: &str = "RUSTFS_KMS_ALLOW_IMMEDIATE_DELETION";
+pub const ENV_KMS_VAULT_ADDRESS: &str = "RUSTFS_KMS_VAULT_ADDRESS";
+pub const ENV_KMS_VAULT_TOKEN: &str = "RUSTFS_KMS_VAULT_TOKEN";
+pub const ENV_KMS_VAULT_NAMESPACE: &str = "RUSTFS_KMS_VAULT_NAMESPACE";
+pub const ENV_KMS_VAULT_MOUNT_PATH: &str = "RUSTFS_KMS_VAULT_MOUNT_PATH";
 pub const ENV_KMS_VAULT_SKIP_TLS_VERIFY: &str = "RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT";
 pub const ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX: &str = "RUSTFS_KMS_VAULT_TRANSIT_METADATA_PREFIX";
@@ -35,6 +39,9 @@ pub const ENV_KMS_VAULT_APPROLE_SECRET_ID: &str = "RUSTFS_KMS_VAULT_APPROLE_SECR
 pub const ENV_KMS_VAULT_APPROLE_SECRET_ID_FILE: &str = "RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE";
 pub const ENV_KMS_VAULT_APPROLE_MOUNT: &str = "RUSTFS_KMS_VAULT_APPROLE_MOUNT";
 pub const ENV_KMS_VAULT_TOKEN_FILE: &str = "RUSTFS_KMS_VAULT_TOKEN_FILE";
+pub const ENV_KMS_VAULT_KUBERNETES_ROLE: &str = "RUSTFS_KMS_VAULT_KUBERNETES_ROLE";
+pub const ENV_KMS_VAULT_KUBERNETES_MOUNT: &str = "RUSTFS_KMS_VAULT_KUBERNETES_MOUNT";
+pub const ENV_KMS_VAULT_KUBERNETES_JWT_PATH: &str = "RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH";
 pub const ENV_KMS_AWS_REGION: &str = "RUSTFS_KMS_AWS_REGION";
 pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// Age in whole seconds beyond which a key is reported as due for rotation;
@@ -45,6 +52,9 @@ pub const ENV_KMS_ROTATION_MAX_WRAPS: &str = "RUSTFS_KMS_ROTATION_MAX_WRAPS";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT: &str = "secret";
 pub const DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX: &str = "rustfs/kms/transit-metadata";
 pub const DEFAULT_VAULT_APPROLE_MOUNT: &str = "approle";
+pub const DEFAULT_VAULT_KUBERNETES_MOUNT: &str = "kubernetes";
+/// Where the kubelet projects a pod's ServiceAccount token by default.
+pub const DEFAULT_VAULT_KUBERNETES_JWT_PATH: &str = "/var/run/secrets/kubernetes.io/serviceaccount/token";
 
 /// Upper bound applied to `KmsConfig::timeout` when deriving backend behavior.
 ///
@@ -84,6 +94,14 @@ fn default_vault_approle_mount() -> String {
     DEFAULT_VAULT_APPROLE_MOUNT.to_string()
 }
 
+fn default_vault_kubernetes_mount() -> String {
+    DEFAULT_VAULT_KUBERNETES_MOUNT.to_string()
+}
+
+fn default_vault_kubernetes_jwt_path() -> PathBuf {
+    PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH)
+}
+
 pub const KMS_CONFIG_REDACTION_RULES: &[RedactionRule] = &[
     RedactionRule::new("kms.local.master_key", RedactionLevel::Secret, "local backend key encryption material"),
     RedactionRule::new("kms.vault.token", RedactionLevel::Secret, "vault authentication token"),
@@ -490,6 +508,23 @@ pub enum VaultAuthMethod {
         #[serde(default)]
         refresh_safety_window_secs: Option<u64>,
     },
+    /// Kubernetes authentication: the pod's ServiceAccount token is exchanged
+    /// for a lease-bound Vault token that is renewed in the background.
+    Kubernetes {
+        /// Vault role bound to this ServiceAccount.
+        role: String,
+        /// Kubernetes auth engine mount path.
+        #[serde(default = "default_vault_kubernetes_mount")]
+        mount: String,
+        /// Projected ServiceAccount token to present. Re-read on every login so
+        /// a token the kubelet rotates is picked up without a restart.
+        #[serde(default = "default_vault_kubernetes_jwt_path")]
+        jwt_path: PathBuf,
+        /// Fail-closed margin in seconds, as on `AppRole`. Defaults to the
+        /// per-attempt timeout.
+        #[serde(default)]
+        refresh_safety_window_secs: Option<u64>,
+    },
     /// Agent-managed token file (for example a Vault Agent auto-auth sink):
     /// the token is read from `path` and re-read periodically so a token
     /// rotated by the agent is picked up without a restart.
@@ -520,6 +555,16 @@ impl VaultAuthMethod {
         }
     }
 
+    /// Kubernetes authentication with the default mount and projected token path.
+    pub fn kubernetes(role: String) -> Self {
+        Self::Kubernetes {
+            role,
+            mount: default_vault_kubernetes_mount(),
+            jwt_path: default_vault_kubernetes_jwt_path(),
+            refresh_safety_window_secs: None,
+        }
+    }
+
     /// Agent-managed token file with the default poll interval.
     pub fn token_file(path: PathBuf) -> Self {
         Self::TokenFile {
@@ -548,6 +593,20 @@ impl fmt::Debug for VaultAuthMethod {
                 .field("mount", mount)
                 .field("refresh_safety_window_secs", refresh_safety_window_secs)
                 .finish(),
+            // No redaction: the role and mount name a Vault binding, and the
+            // ServiceAccount token itself is never held on this type.
+            Self::Kubernetes {
+                role,
+                mount,
+                jwt_path,
+                refresh_safety_window_secs,
+            } => f
+                .debug_struct("Kubernetes")
+                .field("role", role)
+                .field("mount", mount)
+                .field("jwt_path", jwt_path)
+                .field("refresh_safety_window_secs", refresh_safety_window_secs)
+                .finish(),
             Self::TokenFile {
                 path,
                 poll_interval_secs,
@@ -1028,50 +1087,12 @@ impl KmsConfig {
                 });
             }
             KmsBackend::VaultKv2 => {
-                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                let auth_method = vault_auth_method_from_env()?;
-                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
-
-                let mount_path = match get_env_opt_str("RUSTFS_KMS_VAULT_MOUNT_PATH") {
-                    Some(path) => {
-                        tracing::warn!(
-                            "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
-                        );
-                        path
-                    }
-                    None => default_vault_kv2_mount_path(),
-                };
-
-                config.backend_config = BackendConfig::VaultKv2(Box::new(VaultConfig {
-                    address,
-                    auth_method,
-                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
-                    mount_path,
-                    kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
-                    key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
-                    tls: vault_tls_config(skip_tls_verify),
-                }));
+                config.backend_config =
+                    BackendConfig::VaultKv2(Box::new(vault_kv2_config_from_env(VaultCliOverrides::default())?));
             }
             KmsBackend::VaultTransit => {
-                let address = get_env_str("RUSTFS_KMS_VAULT_ADDRESS", "http://localhost:8200");
-                let auth_method = vault_auth_method_from_env()?;
-                let skip_tls_verify = get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false);
-
-                config.backend_config = BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
-                    address,
-                    auth_method,
-                    namespace: get_env_opt_str("RUSTFS_KMS_VAULT_NAMESPACE"),
-                    mount_path: get_env_str("RUSTFS_KMS_VAULT_MOUNT_PATH", "transit"),
-                    metadata_kv_mount: get_env_str(
-                        ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT,
-                        DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT,
-                    ),
-                    metadata_key_prefix: get_env_str(
-                        ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX,
-                        DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX,
-                    ),
-                    tls: vault_tls_config(skip_tls_verify),
-                }));
+                config.backend_config =
+                    BackendConfig::VaultTransit(Box::new(vault_transit_config_from_env(VaultCliOverrides::default())?));
             }
             KmsBackend::Static => {
                 // Read from file first, then fall back to direct env var
@@ -1202,6 +1223,78 @@ fn is_under_temp_dir(path: &Path) -> bool {
     path.starts_with(std::env::temp_dir())
 }
 
+/// Command-line values that take precedence over the matching environment
+/// variables when assembling a Vault backend configuration.
+///
+/// Every field has a `RUSTFS_KMS_VAULT_*` equivalent that the CLI layer already
+/// reads, so these are only set when the operator passed an explicit flag.
+///
+/// Deliberately not `Debug`: `token` holds the raw Vault token, and the
+/// redacting `Debug` impls elsewhere in this module exist because a derived one
+/// would print it. Denying the derive makes a future `{overrides:?}` a compile
+/// error instead of a leak.
+#[derive(Default, Clone, Copy)]
+pub struct VaultCliOverrides<'a> {
+    pub address: Option<&'a str>,
+    pub token: Option<&'a str>,
+    pub mount_path: Option<&'a str>,
+}
+
+/// Assemble the Vault KV2 backend configuration from the environment.
+///
+/// Shared by [`KmsConfig::from_env`] and the server's command-line startup path
+/// so both resolve the same auth method, namespace, TLS and mount settings.
+pub fn vault_kv2_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultConfig> {
+    let mount_path = match overrides
+        .mount_path
+        .map(str::to_string)
+        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_MOUNT_PATH))
+    {
+        Some(path) => {
+            tracing::warn!(
+                "RUSTFS_KMS_VAULT_MOUNT_PATH is deprecated for the Vault KV2 backend: it never calls the Transit engine and the value is stored but unused"
+            );
+            path
+        }
+        None => default_vault_kv2_mount_path(),
+    };
+
+    Ok(VaultConfig {
+        address: vault_address_from_env(overrides.address),
+        auth_method: vault_auth_method_from_env(overrides.token)?,
+        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
+        mount_path,
+        kv_mount: get_env_str("RUSTFS_KMS_VAULT_KV_MOUNT", "secret"),
+        key_path_prefix: get_env_str("RUSTFS_KMS_VAULT_KEY_PREFIX", "rustfs/kms/keys"),
+        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
+    })
+}
+
+/// Assemble the Vault Transit backend configuration from the environment.
+///
+/// Companion to [`vault_kv2_config_from_env`]; see there for why both entry
+/// points share it.
+pub fn vault_transit_config_from_env(overrides: VaultCliOverrides<'_>) -> Result<VaultTransitConfig> {
+    Ok(VaultTransitConfig {
+        address: vault_address_from_env(overrides.address),
+        auth_method: vault_auth_method_from_env(overrides.token)?,
+        namespace: get_env_opt_str(ENV_KMS_VAULT_NAMESPACE),
+        mount_path: overrides
+            .mount_path
+            .map(str::to_string)
+            .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_MOUNT_PATH, "transit")),
+        metadata_kv_mount: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT, DEFAULT_VAULT_TRANSIT_METADATA_KV_MOUNT),
+        metadata_key_prefix: get_env_str(ENV_KMS_VAULT_TRANSIT_METADATA_PREFIX, DEFAULT_VAULT_TRANSIT_METADATA_KEY_PREFIX),
+        tls: vault_tls_config(get_env_bool(ENV_KMS_VAULT_SKIP_TLS_VERIFY, false)),
+    })
+}
+
+fn vault_address_from_env(override_value: Option<&str>) -> String {
+    override_value
+        .map(str::to_string)
+        .unwrap_or_else(|| get_env_str(ENV_KMS_VAULT_ADDRESS, "http://localhost:8200"))
+}
+
 /// Resolve the Vault auth method from environment variables.
 ///
 /// Setting `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` selects AppRole authentication;
@@ -1209,27 +1302,59 @@ fn is_under_temp_dir(path: &Path) -> bool {
 /// (re-read on every login, mirroring the `RUSTFS_KMS_STATIC_SECRET_KEY_FILE`
 /// precedent) or inline from `RUSTFS_KMS_VAULT_APPROLE_SECRET_ID`, with the
 /// file taking precedence. Without a role id the legacy token flow applies.
-fn vault_auth_method_from_env() -> Result<VaultAuthMethod> {
+///
+/// `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` selects Kubernetes authentication, which
+/// presents the pod's projected ServiceAccount token.
+///
+/// `token_override` carries a token supplied on the command line; it stands in
+/// for `RUSTFS_KMS_VAULT_TOKEN` everywhere below, including the conflict checks,
+/// so a flag and the variable it mirrors select the same method.
+fn vault_auth_method_from_env(token_override: Option<&str>) -> Result<VaultAuthMethod> {
+    let token = token_override
+        .map(str::to_string)
+        .or_else(|| get_env_opt_str(ENV_KMS_VAULT_TOKEN));
+    let role_id = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID);
+    let kubernetes_role = get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_ROLE);
+
     if let Some(token_file) = get_env_opt_str(ENV_KMS_VAULT_TOKEN_FILE) {
         // A token file names one authoritative credential source; combining it
         // with another one would leave the effective identity ambiguous, so
         // that is a configuration error rather than a precedence rule.
-        if get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID).is_some() {
-            return Err(KmsError::configuration_error(format!(
-                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
-            )));
-        }
-        if get_env_opt_str("RUSTFS_KMS_VAULT_TOKEN").is_some() {
-            return Err(KmsError::configuration_error(format!(
-                "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with RUSTFS_KMS_VAULT_TOKEN; configure exactly one Vault auth method"
-            )));
+        for (name, configured) in [
+            (ENV_KMS_VAULT_APPROLE_ROLE_ID, role_id.is_some()),
+            (ENV_KMS_VAULT_KUBERNETES_ROLE, kubernetes_role.is_some()),
+            (ENV_KMS_VAULT_TOKEN, token.is_some()),
+        ] {
+            if configured {
+                return Err(KmsError::configuration_error(format!(
+                    "{ENV_KMS_VAULT_TOKEN_FILE} cannot be combined with {name}; configure exactly one Vault auth method"
+                )));
+            }
         }
         return Ok(VaultAuthMethod::token_file(PathBuf::from(token_file)));
     }
 
-    let Some(role_id) = get_env_opt_str(ENV_KMS_VAULT_APPROLE_ROLE_ID) else {
+    if let Some(role) = kubernetes_role {
+        // Unlike a leftover static token, a second login method is never a
+        // stale remnant: both were configured deliberately and neither can be
+        // ranked over the other.
+        if role_id.is_some() {
+            return Err(KmsError::configuration_error(format!(
+                "{ENV_KMS_VAULT_KUBERNETES_ROLE} cannot be combined with {ENV_KMS_VAULT_APPROLE_ROLE_ID}; configure exactly one Vault auth method"
+            )));
+        }
+        return Ok(VaultAuthMethod::Kubernetes {
+            role,
+            mount: get_env_str(ENV_KMS_VAULT_KUBERNETES_MOUNT, DEFAULT_VAULT_KUBERNETES_MOUNT),
+            jwt_path: get_env_opt_str(ENV_KMS_VAULT_KUBERNETES_JWT_PATH)
+                .map_or_else(default_vault_kubernetes_jwt_path, PathBuf::from),
+            refresh_safety_window_secs: None,
+        });
+    }
+
+    let Some(role_id) = role_id else {
         return Ok(VaultAuthMethod::Token {
-            token: get_env_str("RUSTFS_KMS_VAULT_TOKEN", "dev-token"),
+            token: token.unwrap_or_else(|| "dev-token".to_string()),
         });
     };
 
@@ -1273,6 +1398,22 @@ fn validate_vault_auth_method(backend_name: &str, auth_method: &VaultAuthMethod)
             }
             Ok(())
         }
+        VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } => {
+            if role.is_empty() {
+                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes role cannot be empty")));
+            }
+            if mount.is_empty() {
+                return Err(KmsError::configuration_error(format!("{backend_name} Kubernetes mount cannot be empty")));
+            }
+            if jwt_path.as_os_str().is_empty() {
+                return Err(KmsError::configuration_error(format!(
+                    "{backend_name} Kubernetes ServiceAccount token path cannot be empty"
+                )));
+            }
+            Ok(())
+        }
         VaultAuthMethod::TokenFile {
             path,
             poll_interval_secs,
@@ -1976,6 +2117,106 @@ mod tests {
             .expect("well-formed token file auth must validate");
     }
 
+    /// A Kubernetes role alone configures the method: the credential is the
+    /// pod's projected ServiceAccount token, so nothing secret is in the
+    /// environment and the mount and token path fall back to the cluster
+    /// defaults.
+    #[test]
+    fn test_from_env_selects_kubernetes() {
+        with_vars(
+            vec![
+                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
+                (ENV_KMS_VAULT_ADDRESS, Some("https://vault.example.com")),
+                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
+                (ENV_KMS_VAULT_KUBERNETES_MOUNT, None),
+                (ENV_KMS_VAULT_KUBERNETES_JWT_PATH, None),
+                (ENV_KMS_VAULT_TOKEN, None),
+                (ENV_KMS_VAULT_TOKEN_FILE, None),
+                (ENV_KMS_VAULT_APPROLE_ROLE_ID, None),
+            ],
+            || {
+                let config = KmsConfig::from_env().expect("kms config should load from env");
+                let vault = config.vault_transit_config().expect("vault transit backend config");
+                let VaultAuthMethod::Kubernetes {
+                    role,
+                    mount,
+                    jwt_path,
+                    refresh_safety_window_secs,
+                } = &vault.auth_method
+                else {
+                    panic!(
+                        "a kubernetes role in the environment must select Kubernetes auth, got {:?}",
+                        vault.auth_method
+                    );
+                };
+                assert_eq!(role, "rustfs");
+                assert_eq!(mount, DEFAULT_VAULT_KUBERNETES_MOUNT);
+                assert_eq!(jwt_path, Path::new(DEFAULT_VAULT_KUBERNETES_JWT_PATH));
+                assert_eq!(refresh_safety_window_secs, &None);
+            },
+        );
+    }
+
+    #[test]
+    fn test_from_env_kubernetes_is_mutually_exclusive_with_other_auth() {
+        with_vars(
+            vec![
+                ("RUSTFS_KMS_BACKEND", Some("vault-transit")),
+                (ENV_KMS_VAULT_KUBERNETES_ROLE, Some("rustfs")),
+                (ENV_KMS_VAULT_APPROLE_ROLE_ID, Some("env-role-id")),
+                (ENV_KMS_VAULT_TOKEN, None),
+                (ENV_KMS_VAULT_TOKEN_FILE, None),
+            ],
+            || {
+                let error = KmsConfig::from_env().expect_err("kubernetes combined with approle must be rejected");
+                assert!(error.to_string().contains(ENV_KMS_VAULT_KUBERNETES_ROLE));
+                assert!(error.to_string().contains(ENV_KMS_VAULT_APPROLE_ROLE_ID));
+            },
+        );
+    }
+
+    #[test]
+    fn test_validate_rejects_bad_kubernetes_settings() {
+        let vault_config = |auth_method: VaultAuthMethod| KmsConfig {
+            backend: KmsBackend::VaultTransit,
+            backend_config: BackendConfig::VaultTransit(Box::new(VaultTransitConfig {
+                address: "https://vault.example.com:8200".to_string(),
+                auth_method,
+                ..Default::default()
+            })),
+            ..Default::default()
+        };
+
+        let error = vault_config(VaultAuthMethod::kubernetes(String::new()))
+            .validate()
+            .expect_err("an empty kubernetes role must be rejected");
+        assert!(error.to_string().contains("role"), "got {error}");
+
+        let error = vault_config(VaultAuthMethod::Kubernetes {
+            role: "rustfs".to_string(),
+            mount: String::new(),
+            jwt_path: PathBuf::from(DEFAULT_VAULT_KUBERNETES_JWT_PATH),
+            refresh_safety_window_secs: None,
+        })
+        .validate()
+        .expect_err("an empty kubernetes mount must be rejected");
+        assert!(error.to_string().contains("mount"), "got {error}");
+
+        let error = vault_config(VaultAuthMethod::Kubernetes {
+            role: "rustfs".to_string(),
+            mount: DEFAULT_VAULT_KUBERNETES_MOUNT.to_string(),
+            jwt_path: PathBuf::new(),
+            refresh_safety_window_secs: None,
+        })
+        .validate()
+        .expect_err("an empty ServiceAccount token path must be rejected");
+        assert!(error.to_string().contains("token path"), "got {error}");
+
+        vault_config(VaultAuthMethod::kubernetes("rustfs".to_string()))
+            .validate()
+            .expect("well-formed kubernetes auth must validate");
+    }
+
     /// Every KV2 read, write and listing is routed through `kv_mount`, so an
     /// empty one names a path no Vault engine answers. The Transit backend
     /// already rejects its own empty mounts; this closes the same gap on the

crates/kms/src/encryption/dek.rs

@@ -18,8 +18,6 @@
 //! data encryption keys using master keys. It abstracts the encryption
 //! operations so that different backends can share the same encryption logic.
 
-#![allow(dead_code)] // Trait methods may be used by implementations
-
 use crate::error::{KmsError, Result};
 use crate::persisted_observability::{BoundedUnknownFieldName, UnknownFieldSummary};
 use async_trait::async_trait;

crates/madmin/src/site_replication.rs

@@ -258,7 +258,7 @@ pub struct SRLDAPUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRIAMUser {
     #[serde(rename = "accessKey", default)]
     pub access_key: String,
@@ -270,7 +270,7 @@ pub struct SRIAMUser {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRGroupInfo {
     #[serde(rename = "updateReq", default)]
     pub update_req: GroupAddRemove,
@@ -346,7 +346,7 @@ pub struct SRCredInfo {
     pub api_version: Option<String>,
 }
 
-#[derive(Debug, Serialize, Deserialize, Default)]
+#[derive(Debug, Clone, Serialize, Deserialize, Default)]
 pub struct SRIAMItem {
     #[serde(default)]
     pub r#type: String,

crates/notify/src/rules/config.rs

@@ -40,7 +40,10 @@ impl RuleEvents for RuleView {
 #[derive(Debug)]
 struct CompiledRules {
     // Keep RulesMap (can be used later if you want to make more complex judgments during the snapshot reading phase)
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "speculative retention: the comment above keeps it for richer snapshot-time judgements that no code performs yet (backlog#1823)"
+    )]
     rules_map: RulesMap,
     // for RulesContainer::iter_rules
     rule_views: Vec<RuleView>,

crates/notify/src/rules/rules_map.rs

@@ -187,7 +187,6 @@ impl RulesMap {
     /// # Parameters
     /// * `event_name` - The EventName from which to remove the rule.
     /// * `pattern` - The pattern of the rule to be removed.
-    #[allow(dead_code)]
     pub fn remove_rule(&mut self, event_name: &EventName, pattern: &str) {
         let mut remove_event = false;
 
@@ -209,7 +208,6 @@ impl RulesMap {
     ///
     /// # Parameters
     /// * `event_names` - A slice of EventNames to be removed.
-    #[allow(dead_code)]
     pub fn remove_rules(&mut self, event_names: &[EventName]) {
         for event_name in event_names {
             self.map.remove(event_name);
@@ -223,7 +221,6 @@ impl RulesMap {
     /// * `event_name` - The EventName to update.
     /// * `pattern` - The pattern of the rule to be updated.
     /// * `target_id` - The TargetID to be added.
-    #[allow(dead_code)]
     pub fn update_rule(&mut self, event_name: EventName, pattern: String, target_id: TargetID) {
         self.map.entry(event_name).or_default().add(pattern, target_id);
         self.total_events_mask |= event_name.mask(); // Update only the relevant bitmask

crates/notify/src/rules/target_id_set.rs

@@ -18,12 +18,6 @@ use rustfs_targets::arn::TargetID;
 /// TargetIDSet - A collection representation of TargetID.
 pub type TargetIdSet = HashSet<TargetID>;
 
-/// Provides a Go-like method for TargetIdSet (can be implemented as trait if needed)
-#[allow(dead_code)]
-pub(crate) fn new_target_id_set(target_ids: Vec<TargetID>) -> TargetIdSet {
-    target_ids.into_iter().collect()
-}
-
 // HashSet has built-in clone, union, difference and other operations.
 // But the Go version of the method returns a new Set, and the HashSet method is usually iterator or modify itself.
 // If you need to exactly match Go's API style, you can add wrapper functions.

crates/obs/src/metrics/config.rs

@@ -17,7 +17,6 @@ use std::time::Duration;
 /// Environment variable key for the global default metrics interval (seconds).
 pub const ENV_DEFAULT_METRICS_INTERVAL: &str = "RUSTFS_METRICS_DEFAULT_INTERVAL_SEC";
 /// Default interval for metrics collection if not specified otherwise.
-#[allow(dead_code)]
 pub const DEFAULT_METRICS_INTERVAL: Duration = Duration::from_secs(60);
 
 /// Environment variable key for cluster metrics interval (seconds).

crates/obs/src/metrics/report.rs

@@ -145,21 +145,18 @@ impl PrometheusMetric {
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_label(mut self, key: &'static str, value: impl Into<Cow<'static, str>>) -> Self {
         self.labels.push((key, value.into()));
         self
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_label_owned(mut self, key: &'static str, value: String) -> Self {
         self.labels.push((key, Cow::Owned(value)));
         self
     }
 
     #[inline]
-    #[allow(dead_code)]
     pub fn with_labels(mut self, labels: Vec<(&'static str, Cow<'static, str>)>) -> Self {
         self.labels = labels;
         self

crates/obs/src/metrics/schema/entry/descriptor.rs

@@ -16,7 +16,6 @@ use crate::{MetricName, MetricNamespace, MetricSubsystem, MetricType};
 use std::collections::HashSet;
 
 /// MetricDescriptor - Metric descriptors
-#[allow(dead_code)]
 #[derive(Debug, Clone)]
 pub struct MetricDescriptor {
     pub name: MetricName,
@@ -52,7 +51,6 @@ impl MetricDescriptor {
     }
 
     /// Get the full metric name in Prometheus style: <namespace>_<subsystem>_<name>
-    #[allow(dead_code)]
     pub fn get_full_metric_name(&self) -> String {
         let namespace = self.namespace.as_str();
         let formatted_subsystem = self.subsystem.as_str();
@@ -61,7 +59,6 @@ impl MetricDescriptor {
     }
 
     /// check whether the label is in the label set
-    #[allow(dead_code)]
     pub fn has_label(&mut self, label: &str) -> bool {
         self.get_label_set().contains(label)
     }

crates/obs/src/metrics/schema/entry/metric_name.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 /// The metric name is the individual name of the metric
-#[allow(dead_code)]
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub enum MetricName {
     // The generic metric name
@@ -443,7 +442,6 @@ pub enum MetricName {
 }
 
 impl MetricName {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         match self {
             Self::AuthTotal => "auth_total".to_string(),

crates/obs/src/metrics/schema/entry/metric_type.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 /// MetricType - Indicates the type of indicator
-#[allow(dead_code)]
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum MetricType {
     Counter,
@@ -23,7 +22,6 @@ pub enum MetricType {
 
 impl MetricType {
     /// convert the metric type to a string representation
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Counter => "counter",
@@ -34,7 +32,6 @@ impl MetricType {
 
     /// Convert the metric type to the Prometheus value type
     /// In a Rust implementation, this might return the corresponding Prometheus Rust client type
-    #[allow(dead_code)]
     pub fn as_prom(&self) -> &'static str {
         match self {
             Self::Counter => "counter.",

crates/obs/src/metrics/schema/entry/mod.rs

@@ -56,7 +56,6 @@ pub fn new_gauge_md(
 }
 
 /// create a new histogram indicator descriptor
-#[allow(dead_code)]
 pub fn new_histogram_md(
     name: impl Into<MetricName>,
     help: impl Into<String>,

crates/obs/src/metrics/schema/entry/namespace.rs

@@ -19,7 +19,6 @@ pub enum MetricNamespace {
 }
 
 impl MetricNamespace {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::RustFS => "rustfs",

crates/obs/src/metrics/schema/entry/path_utils.rs

@@ -14,7 +14,6 @@
 
 /// Format the path to the metric name format
 /// Replace '/' and '-' with '_'
-#[allow(dead_code)]
 pub fn format_path_to_metric_name(path: &str) -> String {
     path.trim_start_matches('/').replace(['/', '-'], "_")
 }

crates/obs/src/metrics/schema/entry/subsystem.rs

@@ -102,7 +102,6 @@ impl MetricSubsystem {
     }
 
     /// Get the formatted metric name format string
-    #[allow(dead_code)]
     pub fn as_str(&self) -> String {
         format_path_to_metric_name(self.path())
     }
@@ -151,7 +150,6 @@ impl MetricSubsystem {
     }
 
     /// A convenient way to create custom subsystems directly
-    #[allow(dead_code)]
     pub fn new(path: impl Into<String>) -> Self {
         Self::Custom(path.into())
     }
@@ -176,7 +174,6 @@ impl std::fmt::Display for MetricSubsystem {
     }
 }
 
-#[allow(dead_code)]
 pub mod subsystems {
     use super::MetricSubsystem;
 

crates/obs/src/telemetry/rolling.rs

@@ -38,7 +38,10 @@ pub enum Rotation {
     Minutely,
     Hourly,
     Daily,
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "constructed only by this file's rolling-appender tests; the lib target cannot see them (backlog#1823)"
+    )]
     Never,
 }
 

crates/policy/src/policy/function.rs

@@ -219,10 +219,6 @@ impl PartialEq for Functions {
     }
 }
 
-#[derive(Clone, Serialize, Deserialize)]
-#[allow(dead_code)]
-pub struct Value;
-
 #[cfg(test)]
 mod tests {
     use crate::policy::Functions;

crates/policy/src/policy/utils/wildcard.rs

@@ -12,7 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#[allow(dead_code)]
 pub fn is_simple_match<P, N>(pattern: P, name: N) -> bool
 where
     P: AsRef<str>,
@@ -29,7 +28,10 @@ where
     inner_match(pattern, name, false)
 }
 
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "prefix-matcher asserted by this file's tests; no production caller yet (backlog#1823)"
+)]
 pub fn is_match_as_pattern_prefix<P, N>(pattern: P, text: N) -> bool
 where
     P: AsRef<str>,

crates/protocols/src/swift/account.rs

@@ -38,7 +38,6 @@ use std::collections::HashMap;
 /// - Account format is invalid
 /// - Credentials don't contain project_id
 /// - Account project_id doesn't match credentials project_id
-#[allow(dead_code)] // Used by Swift implementation
 pub fn validate_account_access(account: &str, credentials: &Credentials) -> SwiftResult<String> {
     // Extract project_id from account (strip "AUTH_" prefix)
     let account_project_id = account
@@ -70,7 +69,6 @@ pub fn validate_account_access(account: &str, credentials: &Credentials) -> Swif
 ///
 /// Admin users (with "admin" or "reseller_admin" roles) can perform
 /// cross-tenant operations and administrative tasks.
-#[allow(dead_code)] // Used by Swift implementation
 pub fn is_admin_user(credentials: &Credentials) -> bool {
     credentials
         .claims

crates/protocols/src/swift/container.rs

@@ -144,7 +144,6 @@ impl ContainerMapper {
     /// - S3 bucket name compatible (only uses [a-z0-9-])
     /// - Deterministic mapping (same input always produces same bucket name)
     /// - Fixed-length prefix (16 hex chars = 8 bytes)
-    #[allow(dead_code)] // Used in: create/delete container operations
     pub fn swift_to_s3_bucket(&self, container: &str, project_id: &str) -> String {
         if self.config.tenant_prefix_enabled {
             let hash = self.hash_project_id(project_id);
@@ -216,7 +215,6 @@ pub fn bucket_info_to_container(info: &BucketInfo, mapper: &ContainerMapper, pro
 /// 2. Lists all S3 buckets
 /// 3. Filters to buckets belonging to this tenant (using tenant prefix)
 /// 4. Converts BucketInfo to Swift Container format
-#[allow(dead_code)] // Used by handler: list containers
 pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftResult<Vec<Container>> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -279,7 +277,6 @@ pub async fn list_containers(account: &str, credentials: &Credentials) -> SwiftR
 /// - Returns 201 Created on success
 /// - Returns 202 Accepted if container already exists
 /// - Returns 400 Bad Request for invalid container names
-#[allow(dead_code)] // Used by handler
 pub async fn create_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<bool> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -348,7 +345,6 @@ fn validate_container_name(container: &str) -> SwiftResult<()> {
 }
 
 /// Container metadata for HEAD response
-#[allow(dead_code)] // TODO: Remove once Swift API integration is complete
 #[derive(Debug, Clone)]
 pub struct ContainerMetadata {
     /// Number of objects in container
@@ -411,7 +407,6 @@ pub(crate) async fn get_container_custom_metadata(
 /// - HEAD /v1/{account}/{container} returns container metadata
 /// - Returns 204 No Content on success with headers
 /// - Returns 404 Not Found if container doesn't exist
-#[allow(dead_code)] // Used by handler
 pub async fn get_container_metadata(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<ContainerMetadata> {
     let (bucket_name, bucket_info, custom_metadata) = get_container_metadata_base(account, container, credentials).await?;
 
@@ -448,7 +443,6 @@ pub async fn get_container_metadata(account: &str, container: &str, credentials:
 /// - The update is additive: items the request does not name keep their stored
 ///   value, and removal is explicit, via `X-Remove-Container-Meta-{name}` or an
 ///   empty value
-#[allow(dead_code)] // Used by handler
 pub async fn update_container_metadata(
     account: &str,
     container: &str,
@@ -520,7 +514,6 @@ pub async fn update_container_metadata(
 /// - Returns 204 No Content on success
 /// - Returns 404 Not Found if container doesn't exist
 /// - Returns 409 Conflict if container is not empty
-#[allow(dead_code)] // Used by handler
 pub async fn delete_container(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access and extract project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -603,7 +596,6 @@ pub async fn delete_container(account: &str, container: &str, credentials: &Cred
 /// - Account validation fails
 /// - Container doesn't exist
 /// - Storage layer errors occur
-#[allow(dead_code)] // Handler integration: GET container
 pub async fn list_objects(
     account: &str,
     container: &str,
@@ -706,7 +698,6 @@ pub async fn list_objects(
 /// Versioning configuration is stored as an S3 bucket tag:
 /// - Tag key: `swift-versions-location`
 /// - Tag value: archive container name
-#[allow(dead_code)] // Used by handler
 pub async fn enable_versioning(
     account: &str,
     container: &str,
@@ -795,7 +786,6 @@ pub async fn enable_versioning(
 /// * `account` - Account identifier
 /// * `container` - Container name to disable versioning on
 /// * `credentials` - Keystone credentials
-#[allow(dead_code)] // Used by handler
 pub async fn disable_versioning(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -855,7 +845,6 @@ pub async fn disable_versioning(account: &str, container: &str, credentials: &Cr
 /// # Returns
 /// - Some(archive_container_name) if versioning is enabled
 /// - None if versioning is not enabled
-#[allow(dead_code)] // Used by handler and object.rs
 pub async fn get_versions_location(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<Option<String>> {
     // Validate account access
     let project_id = validate_account_access(account, credentials)?;
@@ -918,7 +907,6 @@ pub async fn get_versions_location(account: &str, container: &str, credentials:
 ///     &credentials
 /// ).await?;
 /// ```
-#[allow(dead_code)] // Used by handler
 pub async fn set_container_acl(
     account: &str,
     container: &str,
@@ -1022,7 +1010,6 @@ pub async fn set_container_acl(
 ///     println!("Container is publicly readable");
 /// }
 /// ```
-#[allow(dead_code)] // Used by handler
 pub async fn get_container_acl(
     account: &str,
     container: &str,
@@ -1083,7 +1070,6 @@ pub async fn get_container_acl(
 ///
 /// # Returns
 /// Ok(()) if ACLs were deleted successfully
-#[allow(dead_code)] // Used by handler
 pub async fn delete_container_acl(account: &str, container: &str, credentials: &Credentials) -> SwiftResult<()> {
     // Setting both ACLs to None removes them
     set_container_acl(account, container, None, None, credentials).await

crates/protocols/src/swift/errors.rs

@@ -20,7 +20,6 @@ use std::fmt;
 
 /// Swift-specific error type
 #[derive(Debug)]
-#[allow(dead_code)] // Error variants used by Swift implementation
 pub enum SwiftError {
     /// 400 Bad Request
     BadRequest(String),

crates/protocols/src/swift/object.rs

@@ -122,12 +122,10 @@ fn swift_user_metadata(headers: &HeaderMap) -> Option<HashMap<String, String>> {
 ///
 /// Handles URL encoding/decoding and path normalization for Swift object keys.
 /// Swift object names can contain any UTF-8 characters except null bytes.
-#[allow(dead_code)] // Used in: object operations
 pub struct ObjectKeyMapper;
 
 impl ObjectKeyMapper {
     /// Create a new object key mapper
-    #[allow(dead_code)] // Used in: object operations
     pub fn new() -> Self {
         Self
     }
@@ -140,7 +138,6 @@ impl ObjectKeyMapper {
     /// - Not contain null bytes
     /// - Not contain '..' path segments (directory traversal)
     /// - Not start with '/' (leading slash handled by routing)
-    #[allow(dead_code)] // Used in: object operations
     pub fn validate_object_name(object: &str) -> SwiftResult<()> {
         if object.is_empty() {
             return Err(SwiftError::BadRequest("Object name cannot be empty".to_string()));
@@ -183,7 +180,6 @@ impl ObjectKeyMapper {
     /// Example:
     /// - Swift: "photos/vacation/beach photo.jpg"
     /// - S3: "photos/vacation/beach photo.jpg"
-    #[allow(dead_code)] // Used in: object operations
     pub fn swift_to_s3_key(object: &str) -> SwiftResult<String> {
         Self::validate_object_name(object)?;
         Ok(object.to_string())
@@ -193,7 +189,6 @@ impl ObjectKeyMapper {
     ///
     /// This is essentially an identity transformation since we store
     /// Swift object names as-is in S3.
-    #[allow(dead_code)] // Used in: object operations
     pub fn s3_to_swift_name(key: &str) -> String {
         key.to_string()
     }
@@ -208,7 +203,6 @@ impl ObjectKeyMapper {
     /// - Object: "vacation/beach.jpg"
     /// - Bucket: "abc123:photos"
     /// - Key: "vacation/beach.jpg"
-    #[allow(dead_code)] // Used in: object operations
     pub fn build_s3_key(object: &str) -> SwiftResult<String> {
         Self::swift_to_s3_key(object)
     }
@@ -220,7 +214,6 @@ impl ObjectKeyMapper {
     ///
     /// Example URL: /v1/AUTH_abc/container/path%2Fto%2Ffile.txt
     /// Decoded: "path/to/file.txt"
-    #[allow(dead_code)] // Used in: object operations
     pub fn decode_object_from_url(encoded: &str) -> SwiftResult<String> {
         // Decode percent-encoding
         let decoded = urlencoding::decode(encoded).map_err(|e| SwiftError::BadRequest(format!("Invalid URL encoding: {}", e)))?;
@@ -233,7 +226,6 @@ impl ObjectKeyMapper {
     ///
     /// When constructing URLs (e.g., for redirect responses), we need to
     /// percent-encode object names.
-    #[allow(dead_code)] // Used in: object operations
     pub fn encode_object_for_url(object: &str) -> String {
         urlencoding::encode(object).to_string()
     }
@@ -241,7 +233,6 @@ impl ObjectKeyMapper {
     /// Check if object name represents a directory (pseudo-directory)
     ///
     /// In Swift, objects ending with '/' are treated as directory markers.
-    #[allow(dead_code)] // Used in: object operations
     pub fn is_directory_marker(object: &str) -> bool {
         object.ends_with('/')
     }
@@ -250,7 +241,6 @@ impl ObjectKeyMapper {
     ///
     /// Removes redundant slashes and normalizes the path while preserving
     /// trailing slashes for directory markers.
-    #[allow(dead_code)] // Used in: object operations
     pub fn normalize_path(object: &str) -> String {
         // Split by '/', filter out empty segments (except if it's the end)
         let has_trailing_slash = object.ends_with('/');
@@ -324,7 +314,6 @@ fn sanitize_storage_error<E: std::fmt::Display>(operation: &str, error: E) -> Sw
 /// # Returns
 /// * `Ok(etag)` - Object ETag on success
 /// * `Err(SwiftError)` - Error if validation fails or upload fails
-#[allow(dead_code)] // Handler integration: PUT object
 pub async fn put_object<R>(
     account: &str,
     container: &str,
@@ -445,7 +434,6 @@ where
 ///
 /// Similar to put_object, but allows directly specifying metadata instead of extracting from headers.
 /// This is used internally for storing SLO manifests and marker objects.
-#[allow(dead_code)] // Used by SLO implementation
 pub async fn put_object_with_metadata<R>(
     account: &str,
     container: &str,
@@ -549,7 +537,6 @@ where
 /// - `bytes=1000-1999` - Bytes 1000-1999
 /// - `bytes=1000-` - From byte 1000 to end
 /// - `bytes=-500` - Last 500 bytes
-#[allow(dead_code)] // Handler integration: GET object
 pub async fn get_object(
     account: &str,
     container: &str,
@@ -608,7 +595,6 @@ pub async fn get_object(
 /// # Returns
 /// * `Ok(object_info)` - Object metadata (ObjectInfo)
 /// * `Err(SwiftError)` - Error if validation fails or object not found
-#[allow(dead_code)] // Handler integration: HEAD object
 pub async fn head_object(
     account: &str,
     container: &str,
@@ -671,7 +657,6 @@ pub async fn head_object(
 /// # Returns
 /// * `Ok(())` - Object deleted successfully (or didn't exist)
 /// * `Err(SwiftError)` - Error if validation fails or deletion fails
-#[allow(dead_code)] // Handler integration: DELETE object
 pub async fn delete_object(account: &str, container: &str, object: &str, credentials: &Credentials) -> SwiftResult<()> {
     // 1. Validate account access and get project_id
     let project_id = validate_account_access(account, credentials)?;
@@ -732,7 +717,6 @@ pub async fn delete_object(account: &str, container: &str, object: &str, credent
 /// # Returns
 /// * `Ok(())` - Metadata updated successfully
 /// * `Err(SwiftError)` - Error if validation fails, object not found, or update fails
-#[allow(dead_code)] // Handler integration: POST object
 pub async fn update_object_metadata(
     account: &str,
     container: &str,
@@ -846,7 +830,6 @@ pub async fn update_object_metadata(
 /// # Handler Integration Note
 /// The current handler architecture needs to be updated to pass headers through
 /// to support COPY method and X-Copy-From header detection. See handler.rs for details.
-#[allow(dead_code)] // Handler integration: COPY object
 #[allow(clippy::too_many_arguments)] // Necessary for full copy functionality
 pub async fn copy_object(
     src_account: &str,
@@ -979,7 +962,6 @@ pub async fn copy_object(
 /// assert_eq!(container, "my-container");
 /// assert_eq!(object, "path/to/file.txt");
 /// ```
-#[allow(dead_code)] // Handler integration: COPY method
 pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, String)> {
     let destination = destination.trim_start_matches('/');
     let parts: Vec<&str> = destination.splitn(2, '/').collect();
@@ -1013,7 +995,6 @@ pub fn parse_destination_header(destination: &str) -> SwiftResult<(String, Strin
 /// # Returns
 /// * `Ok((container, object))` - Parsed container and object names
 /// * `Err(SwiftError)` - Error if format is invalid
-#[allow(dead_code)] // Handler integration: X-Copy-From
 pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)> {
     // Same parsing logic as Destination header
     parse_destination_header(copy_from)
@@ -1042,7 +1023,6 @@ pub fn parse_copy_from_header(copy_from: &str) -> SwiftResult<(String, String)>
 /// assert_eq!(range.start, 0);
 /// assert_eq!(range.end, 1023);
 /// ```
-#[allow(dead_code)] // Handler integration: Range header
 pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
     if !range_str.starts_with("bytes=") {
         return Err(SwiftError::BadRequest("Range header must start with 'bytes='".to_string()));
@@ -1124,7 +1104,6 @@ pub fn parse_range_header(range_str: &str) -> SwiftResult<HTTPRangeSpec> {
 /// let header = format_content_range(0, 1023, 5000);
 /// assert_eq!(header, "bytes 0-1023/5000");
 /// ```
-#[allow(dead_code)] // Handler integration: Range header
 pub fn format_content_range(start: i64, end: i64, total: i64) -> String {
     format!("bytes {}-{}/{}", start, end, total)
 }

crates/protocols/src/swift/router.rs

@@ -50,7 +50,6 @@ pub enum SwiftRoute {
 
 impl SwiftRoute {
     /// Get the account identifier from the route
-    #[allow(dead_code)] // Public API for future use
     pub fn account(&self) -> &str {
         match self {
             SwiftRoute::Account { account, .. } => account,
@@ -60,7 +59,6 @@ impl SwiftRoute {
     }
 
     /// Extract project_id from account string (removes AUTH_ prefix)
-    #[allow(dead_code)] // Public API for future use
     pub fn project_id(&self) -> Option<&str> {
         let account = self.account();
         ACCOUNT_PATTERN

crates/protocols/src/swift/types.rs

@@ -19,7 +19,6 @@ use std::collections::HashMap;
 
 /// Swift container metadata
 #[derive(Debug, Clone, Serialize, Deserialize)]
-#[allow(dead_code)] // Used in container listing operations
 pub struct Container {
     /// Container name
     pub name: String,
@@ -34,7 +33,6 @@ pub struct Container {
 
 /// Swift object metadata
 #[derive(Debug, Clone, Serialize, Deserialize)]
-#[allow(dead_code)] // Used in object listing operations
 pub struct Object {
     /// Object name (key)
     pub name: String,
@@ -50,7 +48,6 @@ pub struct Object {
 
 /// Swift metadata extracted from headers
 #[derive(Debug, Clone, Default)]
-#[allow(dead_code)] // Used by Swift implementation
 pub struct SwiftMetadata {
     /// Custom metadata key-value pairs (from X-Container-Meta-* or X-Object-Meta-*)
     pub metadata: HashMap<String, String>,

crates/rio-v2/tests/minio_fixture_lab/Dockerfile

@@ -6,9 +6,18 @@
 #
 # The MinIO release is pinned so the captured fixture format is reproducible;
 # this is the release the interop tests were validated against.
-FROM minio/minio:RELEASE.2025-09-07T16-13-09Z AS minio
+#
+# Both base images are build args so a network that cannot reach Docker Hub can
+# point them at a mirror carrying the same content — quay.io publishes the MinIO
+# releases, and public.ecr.aws mirrors the official Python images. CI keeps the
+# Docker Hub defaults. Override with:
+#   --build-arg MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
+#   --build-arg PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim
+ARG MINIO_IMAGE=minio/minio:RELEASE.2025-09-07T16-13-09Z
+ARG PYTHON_IMAGE=python:3.12-slim
+FROM ${MINIO_IMAGE} AS minio
 
-FROM python:3.12-slim
+FROM ${PYTHON_IMAGE}
 RUN apt-get update \
     && apt-get install -y --no-install-recommends openssl ca-certificates \
     && rm -rf /var/lib/apt/lists/*

crates/rio-v2/tests/minio_fixture_lab/README.md

@@ -22,6 +22,18 @@ Use the automated path when you want the lab to:
 - upload a predefined SSE fixture case
 - export the generated backend tree into the lab layout
 
+## Networks without Docker Hub access
+
+`capture_via_docker.sh` pulls its two base images from Docker Hub by default. Where that registry is unreachable, point the build at mirrors carrying the same content — quay.io publishes the MinIO releases and public.ecr.aws mirrors the official Python images:
+
+```bash
+MINIO_LAB_MINIO_IMAGE=quay.io/minio/minio:RELEASE.2025-09-07T16-13-09Z \
+MINIO_LAB_PYTHON_IMAGE=public.ecr.aws/docker/library/python:3.12-slim \
+./capture_via_docker.sh
+```
+
+Pin the MinIO tag to the same release the Dockerfile names; an unpinned `:latest` captures whatever format that day's build writes, which is not what the interop tests were validated against.
+
 ## Layout
 
 The default root is `artifacts/minio-fixture-lab`, which is already ignored by the repository.

crates/rio-v2/tests/minio_fixture_lab/capture_via_docker.sh

@@ -34,8 +34,19 @@ if [ "${cases[0]}" != "all" ]; then
   done
 fi
 
+# Base images are overridable so a network without Docker Hub access can point
+# them at a mirror (see the Dockerfile header). Unset by default, which keeps the
+# Dockerfile's Docker Hub defaults for CI.
+build_args=()
+if [ -n "${MINIO_LAB_MINIO_IMAGE:-}" ]; then
+  build_args+=(--build-arg "MINIO_IMAGE=${MINIO_LAB_MINIO_IMAGE}")
+fi
+if [ -n "${MINIO_LAB_PYTHON_IMAGE:-}" ]; then
+  build_args+=(--build-arg "PYTHON_IMAGE=${MINIO_LAB_PYTHON_IMAGE}")
+fi
+
 echo ">> building ${IMAGE}"
-docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${SCRIPT_DIR}"
+docker build -f "${SCRIPT_DIR}/Dockerfile" -t "${IMAGE}" "${build_args[@]}" "${SCRIPT_DIR}"
 
 echo ">> capturing fixtures into ${FIXTURE_REL}"
 docker run --rm -v "${REPO_ROOT}:/repo" "${IMAGE}" \

crates/scanner/src/scanner_folder.rs

@@ -4621,6 +4621,8 @@ mod tests {
         let _subscriber_guard = tracing::subscriber::set_default(subscriber);
 
         let (mut scanner, temp_dir) = build_test_scanner().await;
+        // Canonicalize for the "drive" field comparison (scanner resolves symlinks).
+        let canonical_temp_dir = std::fs::canonicalize(&temp_dir).unwrap_or_else(|_| temp_dir.clone());
         let _guard = TestGuard::new(60, 100, &mut scanner, temp_dir.clone());
 
         let object_dir = temp_dir.join("bucket").join("object");
@@ -4689,7 +4691,7 @@ mod tests {
         let fields = &events[0]["fields"];
         assert_eq!(fields["component"], LOG_COMPONENT_SCANNER);
         assert_eq!(fields["subsystem"], LOG_SUBSYSTEM_FOLDER);
-        assert_eq!(fields["drive"], temp_dir.to_string_lossy().as_ref());
+        assert_eq!(fields["drive"], canonical_temp_dir.to_string_lossy().as_ref());
         assert_eq!(fields["bucket"], "bucket");
         assert_eq!(fields["object"], "object");
         assert_eq!(fields["metadata_path"], metadata_path.to_string_lossy().as_ref());

crates/trusted-proxies/src/cloud/metadata/aws.rs

@@ -27,6 +27,10 @@ use crate::CloudMetadataFetcher;
 #[derive(Debug, Clone)]
 pub struct AwsMetadataFetcher {
     client: Client,
+    #[allow(
+        dead_code,
+        reason = "IMDS endpoint retained beside the client it configures; requests build their own URLs (backlog#1823)"
+    )]
     metadata_endpoint: String,
 }
 
@@ -46,55 +50,6 @@ impl AwsMetadataFetcher {
             metadata_endpoint: "http://169.254.169.254".to_string(),
         }
     }
-
-    /// Retrieves an IMDSv2 token for secure metadata access.
-    #[allow(dead_code)]
-    async fn get_metadata_token(&self) -> Result<String, AppError> {
-        let url = format!("{}/latest/api/token", self.metadata_endpoint);
-
-        match self
-            .client
-            .put(&url)
-            .header("X-aws-ec2-metadata-token-ttl-seconds", "21600")
-            .send()
-            .await
-        {
-            Ok(response) => {
-                if response.status().is_success() {
-                    let token = response
-                        .text()
-                        .await
-                        .map_err(|e| AppError::cloud(format!("Failed to read IMDSv2 token: {}", e)))?;
-                    Ok(token)
-                } else {
-                    debug!(
-                        event = "trusted_proxies.cloud_metadata",
-                        component = "trusted_proxies",
-                        subsystem = "aws_metadata",
-                        provider = "aws",
-                        operation = "imdsv2_token",
-                        result = "http_error",
-                        status = %response.status(),
-                        "trusted proxy cloud metadata request failed"
-                    );
-                    Err(AppError::cloud("Failed to obtain IMDSv2 token"))
-                }
-            }
-            Err(e) => {
-                debug!(
-                    event = "trusted_proxies.cloud_metadata",
-                    component = "trusted_proxies",
-                    subsystem = "aws_metadata",
-                    provider = "aws",
-                    operation = "imdsv2_token",
-                    result = "request_failed",
-                    error = %e,
-                    "trusted proxy cloud metadata request failed"
-                );
-                Err(AppError::cloud(format!("IMDSv2 request failed: {}", e)))
-            }
-        }
-    }
 }
 
 #[async_trait]

crates/trusted-proxies/src/config/env.rs

@@ -68,7 +68,6 @@ pub fn is_env_set(key: &str) -> bool {
 }
 
 /// Returns a list of all proxy-related environment variables and their current values.
-#[allow(dead_code)]
 pub fn get_all_proxy_env_vars() -> Vec<(String, String)> {
     let vars = [
         ENV_TRUSTED_PROXY_ENABLED,

crates/utils/src/io.rs

@@ -68,7 +68,6 @@ pub async fn read_full_or_eof<R: AsyncRead + Send + Sync + Unpin>(
 
 /// Read exactly buf.len() bytes into buf, or return an error if EOF is reached before any bytes are read.
 /// Like Go's io.ReadFull.
-#[allow(dead_code)]
 pub async fn read_full<R: AsyncRead + Send + Sync + Unpin>(reader: R, buf: &mut [u8]) -> std::io::Result<usize> {
     match read_full_or_eof(reader, buf).await? {
         Some(n) => Ok(n),

crates/utils/src/net.rs

@@ -431,7 +431,6 @@ pub fn parse_and_resolve_address(addr_str: &str) -> std::io::Result<SocketAddr>
     Ok(resolved_addr)
 }
 
-#[allow(dead_code)]
 pub fn bytes_stream<S, E>(stream: S, content_length: usize) -> impl Stream<Item = Result<Bytes, E>> + Send + 'static
 where
     S: Stream<Item = Result<Bytes, E>> + Send + 'static,

crates/utils/src/os/fs_type.rs

@@ -16,7 +16,6 @@
 ///
 /// The table follows Linux `include/uapi/linux/magic.h`; filesystem magic
 /// values without a stable Linux uapi source stay `UNKNOWN`.
-#[allow(dead_code)]
 pub(crate) fn get_fs_type(fs_type: u64) -> &'static str {
     // Magic numbers for various filesystems.
     match fs_type {

crates/utils/src/path.rs

@@ -70,7 +70,6 @@ pub fn is_dir_object(object: &str) -> bool {
 ///
 /// If the object name ends with `GLOBAL_DIR_SUFFIX`, it is replaced with a slash.
 /// Otherwise, the name is returned as is.
-#[allow(dead_code)]
 pub fn decode_dir_object(object: &str) -> String {
     if has_suffix(object, GLOBAL_DIR_SUFFIX) {
         format!("{}{}", object.trim_end_matches(GLOBAL_DIR_SUFFIX), SLASH_SEPARATOR)

docs/operations/kms-backend-security.md

@@ -2,7 +2,7 @@
 
 RustFS ships several KMS backends. They differ not only in deployment effort but in **where master key material lives and who can read it**. Pick a backend based on the confidentiality boundary you need, not on the name alone.
 
-For how the Vault backends authenticate (static token, AppRole, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
+For how the Vault backends authenticate (static token, AppRole, Kubernetes, Vault Agent token file) and how credential refresh and the fail-closed window behave, see the [Vault KMS authentication runbook](vault-kms-authentication.md). For what may be claimed about the cryptographic implementations themselves, see [Cryptographic compliance positioning](kms-cryptographic-compliance.md). For which RustFS identities may manage or use a given key, see [Per-key KMS authorization](kms-per-key-authorization.md). If you are migrating from MinIO, read [Migrating from MinIO: encrypted objects do not carry over](#migrating-from-minio-encrypted-objects-do-not-carry-over) first.
 
 ## Backend comparison
 

docs/operations/vault-kms-authentication.md

@@ -8,9 +8,12 @@ This runbook covers how the RustFS Vault KMS backends (KV2 and Transit) authenti
 | --- | --- | --- | --- | --- |
 | Static token | `Token` | Whatever the operator provisioned; RustFS never renews it | None | Development; short-lived experiments |
 | AppRole | `AppRole` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production without a Vault Agent sidecar |
+| Kubernetes | `Kubernetes` | Lease-bound token obtained by login; renewed by RustFS | Renew at half TTL, re-login on failure | Production on Kubernetes, with no credential to distribute |
 | Agent token file | `TokenFile` | Owned by Vault Agent; RustFS only re-reads the sink file | File re-read once per poll interval | Production with a Vault Agent (or equivalent) managing auth |
 
-Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID` or an explicit `RUSTFS_KMS_VAULT_TOKEN` is rejected at startup with a configuration error, because the effective identity would be ambiguous.
+Exactly one method must be configured. Setting `RUSTFS_KMS_VAULT_TOKEN_FILE` together with any other method, or `RUSTFS_KMS_VAULT_KUBERNETES_ROLE` together with `RUSTFS_KMS_VAULT_APPROLE_ROLE_ID`, is rejected at startup with a configuration error, because the effective identity would be ambiguous. A leftover `RUSTFS_KMS_VAULT_TOKEN` alongside a configured login method is tolerated and ignored, so a stale variable cannot silently downgrade the identity.
+
+All of these are read the same way whether the service is started with `RUSTFS_KMS_ENABLE=true` or configured later through `POST /rustfs/admin/v3/kms/configure`.
 
 The default `dev-token` fallback for `RUSTFS_KMS_VAULT_TOKEN` is rejected outside explicit development mode (`RUSTFS_KMS_ALLOW_INSECURE_DEV_DEFAULTS=true`), as are plain-HTTP Vault addresses and disabled TLS verification.
 
@@ -56,7 +59,44 @@ Deliver the SecretID out of band — a secrets-manager-mounted file, an init-con
 
 The secret_id file is re-read on every login attempt, so rotating the SecretID is a two-step operation with no restart: generate a new SecretID (`vault write -f auth/approle/role/rustfs-kms/secret-id`), atomically replace the file, then revoke the old SecretID accessor. The already-issued token keeps renewing; the new SecretID is only needed at the next full re-login.
 
-An empty or missing secret_id file fails the login attempt immediately (no Vault round trip) and is retried on the normal refresh cadence, so repairing the file heals the backend without a restart.
+An empty or missing secret_id file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a file missing at boot is recovered by restarting the process, not by an in-process retry. Once RustFS is running, the same failure is retried on the normal refresh cadence, so repairing the file mid-run heals the backend without a restart.
+
+## Kubernetes authentication
+
+On Kubernetes this is the method to prefer: the pod's own ServiceAccount is the identity, so there is no credential to distribute, rotate, or leak into a Secret.
+
+### Vault-side setup
+
+```shell
+vault auth enable kubernetes
+
+vault write auth/kubernetes/config \
+    kubernetes_host="https://$KUBERNETES_SERVICE_HOST:$KUBERNETES_SERVICE_PORT"
+
+vault write auth/kubernetes/role/rustfs \
+    bound_service_account_names=rustfs \
+    bound_service_account_namespaces=rustfs \
+    token_policies=rustfs-kms \
+    token_ttl=1h
+```
+
+As with AppRole, keep `token_ttl` comfortably above the RustFS per-attempt timeout (default 30s).
+
+### RustFS configuration
+
+```shell
+RUSTFS_KMS_BACKEND=vault-transit            # or "vault" for the KV2 backend
+RUSTFS_KMS_VAULT_ADDRESS=https://vault.vault.svc.cluster.local:8200
+RUSTFS_KMS_VAULT_KUBERNETES_ROLE=rustfs
+# Optional, defaults to "kubernetes":
+# RUSTFS_KMS_VAULT_KUBERNETES_MOUNT=kubernetes
+# Optional, defaults to the kubelet's projected token path:
+# RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH=/var/run/secrets/kubernetes.io/serviceaccount/token
+```
+
+RustFS logs in at startup and renews the token at half its TTL, falling back to a fresh login exactly as AppRole does. The ServiceAccount token is re-read from disk on every login rather than cached, so a projected token the kubelet rotates is picked up without a restart.
+
+A missing or empty token file fails the login attempt immediately (no Vault round trip). At startup the error is fatal — provider construction fails and the process exits — so a token projected late during a slow pod start is recovered by the pod restart loop, not by an in-process retry. Once RustFS is running, a token file that goes missing or turns empty is retried on the normal refresh cadence and heals the backend on its own.
 
 ## Vault Agent token file
 
@@ -101,13 +141,13 @@ If the agent stops refreshing the file that is fine — RustFS re-reads the same
 
 ## Fail-closed window
 
-For lease-bound credentials (AppRole tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
+For lease-bound credentials (AppRole and Kubernetes tokens, token files), `current()` refuses to hand out a token that is within the safety window of its expiry and has not been refreshed. Requests then fail with `KMS credentials unavailable: ...` instead of being sent with a token that could lapse mid-flight and fail unpredictably on the Vault side.
 
 - Default window: one per-attempt timeout (`RUSTFS_KMS_TIMEOUT_SECS`, default 30s) — a request issued now can legitimately stay in flight that long, so the token must outlive it.
-- Override: `refresh_safety_window_secs` on the `AppRole` or `TokenFile` auth configuration.
+- Override: `refresh_safety_window_secs` on the `AppRole`, `Kubernetes` or `TokenFile` auth configuration.
 - Static tokens never trip the window: they carry no lease and are assumed valid until Vault says otherwise.
 
-The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole) or two poll intervals (token file).
+The window is a symptom threshold, not the fault itself: by the time it trips, refresh has been failing for roughly half the token TTL (AppRole, Kubernetes) or two poll intervals (token file).
 
 ### Troubleshooting
 
@@ -117,6 +157,8 @@ The window is a symptom threshold, not the fault itself: by the time it trips, r
 | Renewal succeeded but re-login later fails | `Vault token renewal failed; falling back to a fresh login` followed by login errors | SecretID expired/revoked or AppRole role changed; rotate the secret_id file |
 | Token file mode error at startup or during polls | `has insecure permissions` in the error | Fix the sink `mode` (0600) and the file owner; the next poll heals the provider |
 | Token file missing/empty errors | `Failed to read Vault token file` / `token file ... is empty` | Vault Agent down or sink misconfigured; restart the agent, the next poll heals the provider |
-| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, token file |
+| Kubernetes login fails with a permission error | `Vault Kubernetes login failed` | The pod's ServiceAccount is not in the role's `bound_service_account_names`/`_namespaces`, or `auth/kubernetes/config` names the wrong API server |
+| Kubernetes ServiceAccount token errors | `Failed to read Kubernetes ServiceAccount token` / `ServiceAccount token ... is empty` | The token is not projected into the pod (check `automountServiceAccountToken` and the volume mount); the next refresh cycle heals the provider |
+| Startup fails immediately with a configuration error naming two env vars | — | Two auth methods configured at once; keep exactly one of token, AppRole, Kubernetes, token file |
 
 When diagnosing, confirm three clocks/lifetimes in order: the Vault token TTL (`vault token lookup` with the token's accessor), the RustFS refresh cadence (half TTL or the poll interval), and the fail-closed window. The renewal task logs every failed cycle, so a silent gap in warnings combined with `CredentialsUnavailable` errors points at the process clock or a paused runtime rather than Vault.

rustfs/src/admin/handlers/kms_backup.rs

@@ -286,6 +286,7 @@ fn auth_method_kind(auth: &VaultAuthMethod) -> String {
     match auth {
         VaultAuthMethod::Token { .. } => "token",
         VaultAuthMethod::AppRole { .. } => "approle",
+        VaultAuthMethod::Kubernetes { .. } => "kubernetes",
         VaultAuthMethod::TokenFile { .. } => "token-file",
     }
     .to_string()
@@ -484,7 +485,10 @@ fn business_trust_root_secrets(config: &KmsConfig) -> Vec<Zeroizing<String>> {
             secrets.push(Zeroizing::new(role_id.clone()));
             secrets.push(Zeroizing::new(secret_id.clone()));
         }
-        VaultAuthMethod::TokenFile { .. } => {}
+        // Kubernetes and TokenFile hold no inline plaintext credential: the
+        // ServiceAccount token and the agent-managed token live in files, and
+        // the role names a Vault binding rather than half a credential pair.
+        VaultAuthMethod::Kubernetes { .. } | VaultAuthMethod::TokenFile { .. } => {}
     };
 
     match &config.backend_config {

rustfs/src/admin/handlers/scanner.rs

@@ -16,8 +16,8 @@ use crate::admin::auth::validate_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::current_scanner_metrics_report;
 use crate::auth::{check_key_valid, get_session_token};
+use crate::module_switches::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
-use crate::startup_background::{ENV_SCANNER_ENABLED, scanner_enabled_from_env};
 use chrono::Utc;
 use http::{HeaderMap, HeaderValue};
 use hyper::{Method, StatusCode};

rustfs/src/admin/handlers/table_catalog/mod.rs

@@ -2995,9 +2995,9 @@ fn table_entry_from_create_table_request(
     let CreateTableRequest {
         name,
         location,
-        schema,
-        partition_spec,
-        write_order,
+        mut schema,
+        mut partition_spec,
+        mut write_order,
         stage_create,
         mut properties,
     } = request;
@@ -3031,6 +3031,9 @@ fn table_entry_from_create_table_request(
     let metadata_location =
         crate::table_catalog::default_table_metadata_file_path(namespace, &table, &next_metadata_file_name(1, &table_id));
 
+    crate::table_catalog::assign_fresh_create_schema_ids(&mut schema, partition_spec.as_mut(), write_order.as_mut())
+        .map_err(catalog_store_error)?;
+
     let entry = crate::table_catalog::TableEntry {
         version: crate::table_catalog::TABLE_CATALOG_ENTRY_VERSION,
         table_bucket: bucket.to_string(),

rustfs/src/admin/handlers/table_catalog/tests.rs

@@ -1873,6 +1873,66 @@ fn create_table_request_accepts_standard_iceberg_rest_shape() {
     assert_eq!(request.name, "events");
 }
 
+#[test]
+fn create_table_assigns_positive_ids_to_spark_schema() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 0,
+            "fields": [
+                {"id": 0, "name": "id", "required": false, "type": "long"},
+                {"id": 1, "name": "payload", "required": false, "type": "string"}
+            ]
+        },
+        "partition-spec": {"spec-id": 0, "fields": []},
+        "properties": {"owner": "spark"}
+    }))
+    .expect("Spark create table request should parse");
+
+    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect("catalog should assign positive field IDs");
+
+    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
+    assert_eq!(metadata["schemas"][0]["fields"][1]["id"], 2);
+    assert_eq!(metadata["last-column-id"], 2);
+}
+
+#[test]
+fn create_table_assigns_fresh_id_to_negative_temporary_field_id() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "identifier-field-ids": [-1],
+            "fields": [{"id": -1, "name": "id", "required": true, "type": "long"}]
+        },
+        "partition-spec": {
+            "fields": [{"source-id": -1, "name": "id", "transform": "identity"}]
+        },
+        "write-order": {
+            "fields": [{
+                "source-id": -1,
+                "transform": "identity",
+                "direction": "asc",
+                "null-order": "nulls-first"
+            }]
+        }
+    }))
+    .expect("create table request with a negative temporary field ID should parse");
+
+    let (_, metadata) = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect("catalog should replace the negative temporary field ID");
+
+    assert_eq!(metadata["schemas"][0]["fields"][0]["id"], 1);
+    assert_eq!(metadata["schemas"][0]["identifier-field-ids"], serde_json::json!([1]));
+    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["last-column-id"], 1);
+}
+
 #[test]
 fn create_table_request_honors_supported_format_version_property() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -1990,6 +2050,142 @@ fn catalog_assigns_read_only_schema_spec_and_sort_order_ids() {
     assert_eq!(updated["default-sort-order-id"], 0);
 }
 
+#[test]
+fn create_table_assigns_fresh_schema_field_ids_and_rewrites_references() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "schema-id": 41,
+            "identifier-field-ids": [0],
+            "fields": [
+                {"id": 0, "name": "id", "required": true, "type": "long"},
+                {
+                    "id": 10,
+                    "name": "details",
+                    "required": false,
+                    "type": {
+                        "type": "struct",
+                        "fields": [{"id": 11, "name": "category", "required": false, "type": "string"}]
+                    }
+                },
+                {
+                    "id": 20,
+                    "name": "tags",
+                    "required": false,
+                    "type": {
+                        "type": "list",
+                        "element-id": 21,
+                        "element-required": false,
+                        "element": "string"
+                    }
+                },
+                {
+                    "id": 30,
+                    "name": "attributes",
+                    "required": false,
+                    "type": {
+                        "type": "map",
+                        "key-id": 31,
+                        "key": "string",
+                        "value-id": 32,
+                        "value-required": false,
+                        "value": {
+                            "type": "struct",
+                            "fields": [{"id": 33, "name": "score", "required": false, "type": "int"}]
+                        }
+                    }
+                }
+            ]
+        },
+        "partition-spec": {
+            "spec-id": 42,
+            "fields": [{"source-id": 0, "name": "id", "transform": "identity"}]
+        },
+        "write-order": {
+            "order-id": 43,
+            "fields": [{
+                "source-id": 11,
+                "transform": "identity",
+                "direction": "asc",
+                "null-order": "nulls-first"
+            }]
+        }
+    }))
+    .expect("create table request should parse");
+
+    let (_, metadata) =
+        table_entry_from_create_table_request("warehouse", &namespace, request).expect("catalog should assign fresh field IDs");
+
+    let schema = &metadata["schemas"][0];
+    assert_eq!(schema["fields"][0]["id"], 1);
+    assert_eq!(schema["fields"][1]["id"], 2);
+    assert_eq!(schema["fields"][2]["id"], 3);
+    assert_eq!(schema["fields"][3]["id"], 4);
+    assert_eq!(schema["fields"][1]["type"]["fields"][0]["id"], 5);
+    assert_eq!(schema["fields"][2]["type"]["element-id"], 6);
+    assert_eq!(schema["fields"][3]["type"]["key-id"], 7);
+    assert_eq!(schema["fields"][3]["type"]["value-id"], 8);
+    assert_eq!(schema["fields"][3]["type"]["value"]["fields"][0]["id"], 9);
+    assert_eq!(schema["identifier-field-ids"], serde_json::json!([1]));
+    assert_eq!(metadata["last-column-id"], 9);
+    assert_eq!(metadata["partition-specs"][0]["fields"][0]["source-id"], 1);
+    assert_eq!(metadata["sort-orders"][0]["fields"][0]["source-id"], 5);
+}
+
+#[test]
+fn create_table_rejects_duplicate_temporary_schema_field_ids() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let request: CreateTableRequest = serde_json::from_value(serde_json::json!({
+        "name": "events",
+        "schema": {
+            "type": "struct",
+            "fields": [
+                {"id": 0, "name": "id", "required": false, "type": "long"},
+                {"id": 0, "name": "payload", "required": false, "type": "string"}
+            ]
+        }
+    }))
+    .expect("create table request should parse");
+
+    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect_err("duplicate temporary field IDs must be rejected");
+
+    assert_eq!(error.message(), Some("duplicate create schema field id 0"));
+}
+
+#[test]
+fn create_table_rejects_excessive_schema_nesting() {
+    let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
+    let mut field_type = serde_json::Value::from("long");
+    for element_id in 1..=crate::table_catalog::ICEBERG_MAX_SCHEMA_NESTING_DEPTH + 1 {
+        field_type = serde_json::json!({
+            "type": "list",
+            "element-id": element_id,
+            "element-required": false,
+            "element": field_type
+        });
+    }
+    let request = CreateTableRequest {
+        name: "events".to_string(),
+        location: None,
+        schema: serde_json::json!({
+            "type": "struct",
+            "fields": [{"id": 0, "name": "nested", "required": false, "type": field_type}]
+        }),
+        partition_spec: None,
+        write_order: None,
+        stage_create: false,
+        properties: BTreeMap::new(),
+    };
+
+    let error = table_entry_from_create_table_request("warehouse", &namespace, request)
+        .expect_err("excessively nested create schemas must be rejected");
+
+    assert_eq!(error.message(), Some("create schema exceeds the maximum nesting depth"));
+}
+
 #[test]
 fn standard_commit_binds_new_specs_and_sort_orders_to_current_schema() {
     let namespace = crate::table_catalog::Namespace::parse("analytics").expect("namespace should parse");
@@ -2247,7 +2443,11 @@ fn create_table_counts_collection_ids_in_last_column_id() {
     let (_, metadata) =
         table_entry_from_create_table_request("warehouse", &namespace, request).expect("table metadata should be created");
 
-    assert_eq!(metadata["last-column-id"], 9);
+    let schema = &metadata["schemas"][0];
+    assert_eq!(schema["fields"][0]["type"]["element-id"], 3);
+    assert_eq!(schema["fields"][1]["type"]["key-id"], 4);
+    assert_eq!(schema["fields"][1]["type"]["value-id"], 5);
+    assert_eq!(metadata["last-column-id"], 5);
 }
 
 #[test]

rustfs/src/app/object_usecase.rs

@@ -1074,7 +1074,7 @@ where
 }
 
 impl futures::Stream for MemoryTrackedBytesStream {
-    type Item = std::io::Result<Bytes>;
+    type Item = Result<Bytes, S3StdError>;
 
     fn poll_next(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         let this = self.get_mut();
@@ -1105,7 +1105,8 @@ impl futures::Stream for MemoryTrackedBytesStream {
             return Poll::Ready(Some(Err(std::io::Error::new(
                 std::io::ErrorKind::InvalidData,
                 format!("materialized GET body length mismatch: expected {}, got {}", this.expected, actual),
-            ))));
+            )
+            .into())));
         }
 
         let Some(bytes) = this.bytes.take() else {
@@ -1132,6 +1133,16 @@ impl futures::Stream for MemoryTrackedBytesStream {
     }
 }
 
+impl ByteStream for MemoryTrackedBytesStream {
+    fn remaining_length(&self) -> RemainingLength {
+        if self.emitted || self.bytes.is_none() {
+            RemainingLength::new_exact(0)
+        } else {
+            RemainingLength::new_exact(self.expected)
+        }
+    }
+}
+
 impl Drop for MemoryTrackedBytesStream {
     fn drop(&mut self) {
         if self.lifecycle.is_finished() {
@@ -4149,7 +4160,7 @@ impl DefaultObjectUsecase {
         let bytes_len = bytes.len();
         let guard = rustfs_io_metrics::track_get_object_buffered_bytes(bytes_len);
         let remaining = usize::try_from(response_content_length.max(0)).unwrap_or(usize::MAX);
-        let blob = StreamingBlob::wrap(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle));
+        let blob = StreamingBlob::new(MemoryTrackedBytesStream::new(bytes, remaining, source, guard, lifecycle));
         if let Some(handoff_start) = handoff_start {
             rustfs_io_metrics::record_get_object_response_handoff(
                 "single_chunk",
@@ -12882,7 +12893,10 @@ mod tests {
             .await
             .expect("mismatched memory body must yield an item")
             .expect_err("a short memory body must fail the stream instead of serving a truncated body");
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(
+            err.downcast_ref::<std::io::Error>().map(std::io::Error::kind),
+            Some(std::io::ErrorKind::InvalidData)
+        );
         assert!(stream.next().await.is_none(), "stream must terminate after the error");
     }
 
@@ -12901,7 +12915,22 @@ mod tests {
             .await
             .expect("mismatched memory body must yield an item")
             .expect_err("an over-long memory body must fail the stream instead of serving mismatched bytes");
-        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(
+            err.downcast_ref::<std::io::Error>().map(std::io::Error::kind),
+            Some(std::io::ErrorKind::InvalidData)
+        );
+    }
+
+    #[test]
+    fn memory_blob_preserves_exact_remaining_length() {
+        let blob = DefaultObjectUsecase::build_memory_bytes_blob(
+            Bytes::from_static(b"hello"),
+            5,
+            GET_MEMORY_BODY_SOURCE_BUFFERED_BODY,
+            GetObjectBodyLifecycle::disabled(),
+        );
+
+        assert_eq!(blob.remaining_length().exact(), Some(5));
     }
 
     #[tokio::test]

rustfs/src/init.rs

@@ -304,30 +304,37 @@ fn build_local_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
     Ok(kms_config)
 }
 
+/// Collect the Vault settings the command line owns.
+///
+/// Everything else — auth method, namespace, TLS, KV mount and metadata paths —
+/// is resolved from the environment by the KMS crate, so this path and
+/// [`rustfs_kms::config::KmsConfig::from_env`] cannot drift apart. The address
+/// stays required here so a missing one is still named instead of silently
+/// falling back to the crate's localhost default.
+fn vault_cli_overrides<'a>(
+    cfg: &'a config::Config,
+    backend_name: &str,
+) -> std::io::Result<rustfs_kms::config::VaultCliOverrides<'a>> {
+    let address = cfg
+        .kms_vault_address
+        .as_deref()
+        .ok_or_else(|| Error::other(format!("Vault address is required for {backend_name} backend")))?;
+
+    Ok(rustfs_kms::config::VaultCliOverrides {
+        address: Some(address),
+        token: cfg.kms_vault_token.as_deref(),
+        mount_path: cfg.kms_vault_mount_path.as_deref(),
+    })
+}
+
 /// Build KMS configuration for Vault backend
 fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let vault_address = cfg
-        .kms_vault_address
-        .as_ref()
-        .ok_or_else(|| Error::other("Vault address is required for vault backend"))?;
-    let vault_token = cfg
-        .kms_vault_token
-        .as_ref()
-        .ok_or_else(|| Error::other("Vault token is required for vault backend"))?;
+    let backend_config = rustfs_kms::config::vault_kv2_config_from_env(vault_cli_overrides(cfg, "vault")?)
+        .map_err(|e| Error::other(format!("Vault KMS configuration failed: {e}")))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultKv2,
-        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(rustfs_kms::config::VaultConfig {
-            address: vault_address.clone(),
-            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
-                token: vault_token.clone(),
-            },
-            namespace: None,
-            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
-            kv_mount: "secret".to_string(),
-            key_path_prefix: "rustfs/kms/keys".to_string(),
-            tls: None,
-        })),
+        backend_config: rustfs_kms::config::BackendConfig::VaultKv2(Box::new(backend_config)),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -344,26 +351,12 @@ fn build_vault_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::c
 
 /// Build KMS configuration for Vault Transit backend
 fn build_vault_transit_kms_config(cfg: &config::Config) -> std::io::Result<rustfs_kms::config::KmsConfig> {
-    let vault_address = cfg
-        .kms_vault_address
-        .as_ref()
-        .ok_or_else(|| Error::other("Vault address is required for vault-transit backend"))?;
-    let vault_token = cfg
-        .kms_vault_token
-        .as_ref()
-        .ok_or_else(|| Error::other("Vault token is required for vault-transit backend"))?;
+    let backend_config = rustfs_kms::config::vault_transit_config_from_env(vault_cli_overrides(cfg, "vault-transit")?)
+        .map_err(|e| Error::other(format!("Vault Transit KMS configuration failed: {e}")))?;
 
     let kms_config = rustfs_kms::config::KmsConfig {
         backend: rustfs_kms::config::KmsBackend::VaultTransit,
-        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(rustfs_kms::config::VaultTransitConfig {
-            address: vault_address.clone(),
-            auth_method: rustfs_kms::config::VaultAuthMethod::Token {
-                token: vault_token.clone(),
-            },
-            namespace: None,
-            mount_path: cfg.kms_vault_mount_path.clone().unwrap_or_else(|| "transit".to_string()),
-            ..rustfs_kms::config::VaultTransitConfig::default()
-        })),
+        backend_config: rustfs_kms::config::BackendConfig::VaultTransit(Box::new(backend_config)),
         allow_insecure_dev_defaults: cfg.kms_allow_insecure_dev_defaults,
         allow_immediate_deletion: rustfs_kms::config::allow_immediate_deletion_from_env(),
         default_key_id: cfg.kms_default_key_id.clone(),
@@ -1405,7 +1398,10 @@ pub async fn init_sftp_system() -> Result<Option<ShutdownHandle>, Box<dyn std::e
 
 #[cfg(test)]
 mod tests {
-    use super::{build_aws_kms_config, notification_config_to_event_rules, resolve_buffer_profile_config};
+    use super::{
+        build_aws_kms_config, build_vault_kms_config, build_vault_transit_kms_config, notification_config_to_event_rules,
+        resolve_buffer_profile_config,
+    };
     use crate::config::{BufferConfig, WorkloadProfile};
     use rustfs_config::KI_B;
     use rustfs_s3_types::EventName;
@@ -1499,6 +1495,151 @@ mod tests {
         assert!(err.to_string().contains("Invalid ARN"), "unexpected error: {err}");
     }
 
+    fn vault_kms_test_config(backend: &str) -> crate::config::Config {
+        let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-vault-kms".to_string()]);
+        config.kms_enable = true;
+        config.kms_backend = backend.to_string();
+        config.kms_vault_address = Some("https://vault.example.com:8200".to_string());
+        config
+    }
+
+    /// The Vault auth method and the settings the CLI has no flag for come from
+    /// the environment, so startup and `KmsConfig::from_env` cannot disagree.
+    /// Regression: startup used to hardcode token auth and require a token,
+    /// which made every non-token method unreachable through `RUSTFS_KMS_ENABLE`.
+    #[test]
+    fn build_vault_transit_kms_config_resolves_auth_and_mounts_from_env() {
+        let config = temp_env::with_vars(
+            [
+                ("RUSTFS_KMS_VAULT_TOKEN", None),
+                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
+                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", Some("env-role-id")),
+                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID", Some("env-secret-id")),
+                ("RUSTFS_KMS_VAULT_APPROLE_SECRET_ID_FILE", None),
+                ("RUSTFS_KMS_VAULT_NAMESPACE", Some("team-a")),
+                ("RUSTFS_KMS_VAULT_TRANSIT_METADATA_KV_MOUNT", Some("rustfs-kv")),
+            ],
+            || {
+                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
+                    .expect("vault transit KMS configuration should build")
+            },
+        );
+
+        let vault = config.vault_transit_config().expect("vault transit backend config");
+        let rustfs_kms::config::VaultAuthMethod::AppRole { role_id, secret_id, .. } = &vault.auth_method else {
+            panic!("approle in the environment must select AppRole auth, got {:?}", vault.auth_method);
+        };
+        assert_eq!(role_id, "env-role-id");
+        assert_eq!(secret_id, "env-secret-id");
+        assert_eq!(vault.namespace.as_deref(), Some("team-a"));
+        assert_eq!(vault.metadata_kv_mount, "rustfs-kv");
+    }
+
+    /// Kubernetes auth needs no credential in the environment at all: the role
+    /// selects it and the pod's projected ServiceAccount token supplies the rest.
+    #[test]
+    fn build_vault_transit_kms_config_selects_kubernetes_auth() {
+        let config = temp_env::with_vars(
+            [
+                ("RUSTFS_KMS_VAULT_TOKEN", None),
+                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
+                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_MOUNT", None),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_JWT_PATH", None),
+            ],
+            || {
+                build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
+                    .expect("vault transit KMS configuration should build")
+            },
+        );
+
+        let vault = config.vault_transit_config().expect("vault transit backend config");
+        let rustfs_kms::config::VaultAuthMethod::Kubernetes {
+            role, mount, jwt_path, ..
+        } = &vault.auth_method
+        else {
+            panic!(
+                "a kubernetes role in the environment must select Kubernetes auth, got {:?}",
+                vault.auth_method
+            );
+        };
+        assert_eq!(role, "rustfs");
+        assert_eq!(mount, rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_MOUNT);
+        assert_eq!(jwt_path, std::path::Path::new(rustfs_kms::config::DEFAULT_VAULT_KUBERNETES_JWT_PATH));
+    }
+
+    /// Two credential sources leave the effective identity ambiguous, so
+    /// startup refuses rather than picking one.
+    #[test]
+    fn build_vault_kms_config_refuses_two_auth_methods() {
+        temp_env::with_vars(
+            [
+                ("RUSTFS_KMS_VAULT_TOKEN", None),
+                ("RUSTFS_KMS_VAULT_TOKEN_FILE", Some("/run/vault-agent/token")),
+                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", Some("rustfs")),
+            ],
+            || {
+                let error = build_vault_kms_config(&vault_kms_test_config("vault"))
+                    .expect_err("two Vault auth methods must not start the server");
+                assert!(error.to_string().contains("exactly one"), "unexpected error: {error}");
+            },
+        );
+    }
+
+    /// The KV2 backend has its own builder, so the key-location settings have
+    /// to be proven separately from the Transit one: pointing at the wrong KV
+    /// mount or prefix makes existing keys look absent.
+    #[test]
+    fn build_vault_kms_config_resolves_kv_mount_and_prefix_from_env() {
+        let config = temp_env::with_vars(
+            [
+                ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
+                ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
+                ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
+                ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
+                ("RUSTFS_KMS_VAULT_KV_MOUNT", Some("rustfs-kv")),
+                ("RUSTFS_KMS_VAULT_KEY_PREFIX", Some("tenant/keys")),
+            ],
+            || build_vault_kms_config(&vault_kms_test_config("vault")).expect("vault KV2 KMS configuration should build"),
+        );
+
+        let vault = config.vault_config().expect("vault kv2 backend config");
+        assert_eq!(vault.kv_mount, "rustfs-kv");
+        assert_eq!(vault.key_path_prefix, "tenant/keys");
+    }
+
+    /// Skipping TLS verification was silently dropped on this path before, so
+    /// an operator who asked for it still got a verified connection. Now that it
+    /// is honoured it must fail closed without the development opt-in, rather
+    /// than quietly downgrading the Vault connection.
+    #[test]
+    fn build_vault_transit_kms_config_refuses_skip_tls_verify_without_opt_in() {
+        let vars = [
+            ("RUSTFS_KMS_VAULT_TOKEN", Some("a-real-token")),
+            ("RUSTFS_KMS_VAULT_TOKEN_FILE", None),
+            ("RUSTFS_KMS_VAULT_APPROLE_ROLE_ID", None),
+            ("RUSTFS_KMS_VAULT_KUBERNETES_ROLE", None),
+            ("RUSTFS_KMS_VAULT_SKIP_TLS_VERIFY", Some("true")),
+        ];
+
+        temp_env::with_vars(vars, || {
+            let error = build_vault_transit_kms_config(&vault_kms_test_config("vault-transit"))
+                .expect_err("skipping TLS verification must not start the server");
+            assert!(error.to_string().contains("TLS"), "unexpected error: {error}");
+        });
+
+        temp_env::with_vars(vars, || {
+            let mut cfg = vault_kms_test_config("vault-transit");
+            cfg.kms_allow_insecure_dev_defaults = true;
+            let config = build_vault_transit_kms_config(&cfg).expect("the development opt-in should accept skip-verify");
+            let vault = config.vault_transit_config().expect("vault transit backend config");
+            assert!(vault.tls.as_ref().is_some_and(|tls| tls.skip_verify));
+        });
+    }
+
     fn aws_kms_test_config() -> crate::config::Config {
         let mut config = crate::config::Config::new("127.0.0.1:9000", vec!["/tmp/rustfs-aws-kms".to_string()]);
         config.kms_enable = true;

rustfs/src/lib.rs

@@ -88,6 +88,7 @@ pub mod inspect;
 pub(crate) mod kms_deletion_gate;
 pub mod license;
 pub mod memory_observability;
+pub mod module_switches;
 pub mod profiling;
 #[cfg(any(feature = "ftps", feature = "webdav", feature = "sftp"))]
 pub mod protocols;

rustfs/src/module_switches.rs

@@ -0,0 +1,68 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Layer-neutral module switches (backlog#1834).
+//!
+//! Whether the scanner, heal, audit and notify modules are on is read from the
+//! infra layer (storage helpers, node-service RPC) and from the interface layer
+//! (admin handlers), but the switches used to live in `startup_background`
+//! (composition) and `server` (interface). Every lower-layer read was therefore
+//! an upward edge that had to be baselined by the layer-dependency guard.
+//!
+//! The env-derived scanner/heal predicates and the audit/notify state cells now
+//! live here, at the bottom of the layer order, so those reads are ordinary
+//! downward edges. Resolving the audit/notify state still needs server-side
+//! configuration, so `server::refresh_audit_module_enabled` and its notify twin
+//! keep that logic and publish the result through the setters below.
+
+use rustfs_utils::get_env_bool_with_aliases;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+pub(crate) const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED";
+pub(crate) const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER";
+pub(crate) const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED";
+pub(crate) const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL";
+
+static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
+static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
+
+/// Whether the data scanner is enabled, defaulting to on.
+pub(crate) fn scanner_enabled_from_env() -> bool {
+    get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true)
+}
+
+/// Whether background heal is enabled, defaulting to on.
+pub(crate) fn heal_enabled_from_env() -> bool {
+    get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true)
+}
+
+/// Last published audit-module state.
+pub fn is_audit_module_enabled() -> bool {
+    AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
+}
+
+/// Publish the audit-module state resolved by `server::refresh_audit_module_enabled`.
+pub(crate) fn set_audit_module_enabled(enabled: bool) {
+    AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
+}
+
+/// Last published notify-module state.
+pub fn is_notify_module_enabled() -> bool {
+    NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
+}
+
+/// Publish the notify-module state resolved by `server::refresh_notify_module_enabled`.
+pub(crate) fn set_notify_module_enabled(enabled: bool) {
+    NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
+}

rustfs/src/server/audit.rs

@@ -19,11 +19,8 @@ use super::{
 use crate::runtime_sources::AppContext;
 use rustfs_audit::{AuditError, AuditResult, audit_system, init_audit_system, system::AuditSystemState};
 use std::collections::HashSet;
-use std::sync::atomic::{AtomicBool, Ordering};
 use tracing::{info, warn};
 
-static AUDIT_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_AUDIT_ENABLE);
-
 fn server_config_from_context() -> Option<rustfs_config::server_config::Config> {
     runtime_sources::current_server_config()
 }
@@ -37,13 +34,11 @@ fn server_config_for_context(context: Option<&AppContext>) -> Option<rustfs_conf
 
 pub fn refresh_audit_module_enabled() -> bool {
     let enabled = resolve_audit_module_state().enabled;
-    AUDIT_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
+    crate::module_switches::set_audit_module_enabled(enabled);
     enabled
 }
 
-pub fn is_audit_module_enabled() -> bool {
-    AUDIT_MODULE_ENABLED.load(Ordering::Relaxed)
-}
+pub use crate::module_switches::is_audit_module_enabled;
 
 fn has_any_persisted_audit_targets(config: &rustfs_config::server_config::Config) -> bool {
     for &subsystem in rustfs_config::audit::AUDIT_SUB_SYSTEMS {

rustfs/src/server/event.rs

@@ -34,7 +34,6 @@ use tokio::time::{Instant, MissedTickBehavior};
 use tokio_util::sync::CancellationToken;
 use tracing::{info, instrument, warn};
 
-static NOTIFY_MODULE_ENABLED: AtomicBool = AtomicBool::new(rustfs_config::DEFAULT_NOTIFY_ENABLE);
 static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false);
 static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false);
 static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new();
@@ -70,13 +69,11 @@ fn should_reconcile_bucket_notification_rules(runtime_changed: bool, notify_enab
 
 pub fn refresh_notify_module_enabled() -> bool {
     let enabled = resolve_notify_module_state().enabled;
-    NOTIFY_MODULE_ENABLED.store(enabled, Ordering::Relaxed);
+    crate::module_switches::set_notify_module_enabled(enabled);
     enabled
 }
 
-pub fn is_notify_module_enabled() -> bool {
-    NOTIFY_MODULE_ENABLED.load(Ordering::Relaxed)
-}
+pub use crate::module_switches::is_notify_module_enabled;
 
 pub(crate) use crate::shared_types::convert_ecstore_object_info;
 
@@ -171,7 +168,7 @@ pub(crate) async fn reconcile_event_notifier_from_store(
         let transition_system = system.clone();
         let transition_store = store.clone();
         let transition = with_refreshed_notify_module_state_from(store.clone(), move |resolution| async move {
-            NOTIFY_MODULE_ENABLED.store(resolution.enabled, Ordering::Relaxed);
+            crate::module_switches::set_notify_module_enabled(resolution.enabled);
             let read_store = transition_store.clone();
             let config_system = transition_system.clone();
             with_server_config_read_lock(transition_store, move || async move {