ci: gate rio-v2 full-suite jobs on schedule; document lifecycle (#6036)

Co-authored-by: cxymds <cxymds@gmail.com>
Co-authored-by: houseme <housemecn@gmail.com>

.github/workflows/cache-warm.yml

@@ -182,7 +182,12 @@ jobs:
             echo '```'
           } >> "$GITHUB_STEP_SUMMARY"
 
-  # Readers: test-and-lint-rio-v2, build-rustfs-debug-binary-rio-v2.
+  # Readers: test-and-lint-rio-v2 (per-PR), build-rustfs-debug-binary-rio-v2
+  # (weekly schedule / manual dispatch only — dormant rio-v2 variant, see
+  # rustfs/backlog#1835 and docs/architecture/minio-file-format-compat.md).
+  # The second build below stays despite the reduced cadence: it warms the
+  # rio-v2,e2e-test-hooks feature resolution the scheduled build restores,
+  # which keeps that lane inside its 30-minute timeout.
   warm-ci-feat-rio:
     name: Warm ci-feat-rio
     runs-on: sm-standard-4

.github/workflows/ci.yml

@@ -533,7 +533,12 @@ jobs:
 
   build-rustfs-debug-binary-rio-v2:
     name: Build RustFS Debug Binary (rio-v2)
-    if: github.event_name != 'pull_request' || github.event.action != 'closed'
+    # Dormant rio-v2 variant (rustfs/backlog#1835): the feature ships in no
+    # default build, so this full-suite lane runs only on the weekly schedule
+    # and manual dispatch. Per-PR cfg-seam coverage stays with
+    # test-and-lint-rio-v2. Lifecycle and the promote-or-delete condition:
+    # docs/architecture/minio-file-format-compat.md ("rio-v2 variant lifecycle").
+    if: github.event_name == 'schedule' || github.event_name == 'workflow_dispatch'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 30
@@ -824,6 +829,9 @@ jobs:
 
   e2e-tests-rio-v2:
     name: End-to-End Tests (rio-v2)
+    # Inherits the schedule/dispatch-only gate through needs: on every other
+    # event build-rustfs-debug-binary-rio-v2 is skipped, so this job skips
+    # with it (see the dormant-variant comment on that job).
     needs: [ build-rustfs-debug-binary-rio-v2 ]
     runs-on: sm-standard-2
     timeout-minutes: 30

ARCHITECTURE.md

@@ -101,7 +101,10 @@ refactors.
 
 The `rustfs` binary crate composes these libraries into the running server.
 `ecstore` remains the storage engine at the architectural center; its internal
-module split is tracked under `docs/architecture/`.
+module split is tracked under `docs/architecture/`. `rio-v2` is the
+feature-gated MinIO on-disk format compatibility I/O layer; it ships in no
+default build (lifecycle:
+[docs/architecture/minio-file-format-compat.md](docs/architecture/minio-file-format-compat.md)).
 
 ## Architecture Invariants
 

Cargo.toml

@@ -41,7 +41,7 @@ members = [
     "crates/protocols", # Protocol implementations (FTPS, SFTP, etc.)
     "crates/protos", # Protocol buffer definitions
     "crates/rio", # Rust I/O utilities and abstractions
-    "crates/rio-v2", # Next-generation Rust I/O compatibility layer
+    "crates/rio-v2", # MinIO on-disk format compatibility I/O layer (feature-gated, ships in no default build)
     "crates/replication", # Replication contracts and wire formats
     "crates/concurrency", # Concurrency management for RustFS - timeout, locking, backpressure, and I/O scheduling
     "crates/s3-types", # S3 event type definitions

crates/config/src/constants/object.rs

@@ -137,6 +137,22 @@ pub const DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED: bool = false;
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE);
 const _: () = assert!(!DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED);
 
+/// Request the object-transaction fencing contract used by storage-owned
+/// cleanup receipts and lock-window optimizations.
+///
+/// This is fail-closed: enabling the writer without a live fleet proof rejects
+/// the commit rather than silently using a legacy-safe path.
+pub const ENV_OBJECT_TRANSACTION_FENCING_WRITE: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE";
+pub const DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE: bool = false;
+
+/// Operator-attested confirmation that every serving node understands the
+/// object transaction fencing contract.
+pub const ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: &str = "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED";
+pub const DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED: bool = false;
+
+const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_WRITE);
+const _: () = assert!(!DEFAULT_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED);
+
 /// Request preserving legacy per-part checksum metadata during data movement.
 ///
 /// This remains ineffective until
@@ -673,4 +689,13 @@ mod remote_version_state_tests {
             "RUSTFS_DATA_MOVEMENT_PART_CHECKSUMS_FLEET_CONFIRMED"
         );
     }
+
+    #[test]
+    fn object_transaction_fencing_gate_uses_stable_environment_names() {
+        assert_eq!(super::ENV_OBJECT_TRANSACTION_FENCING_WRITE, "RUSTFS_OBJECT_TRANSACTION_FENCING_WRITE");
+        assert_eq!(
+            super::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED,
+            "RUSTFS_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED"
+        );
+    }
 }

crates/ecstore/src/config/com.rs

@@ -46,7 +46,6 @@ use rustfs_config::{
     SCANNER_SUB_SYS,
 };
 use rustfs_filemeta::FileInfo;
-use rustfs_utils::path::SLASH_SEPARATOR;
 use serde_json::{Map, Value};
 use std::collections::{HashMap, HashSet};
 use std::sync::LazyLock;
@@ -200,8 +199,6 @@ pub const STORAGE_CLASS_SUB_SYS: &str = "storage_class";
 
 pub const COMMA_SEPARATED_LISTS: &[&str] = &[rustfs_config::oidc::OIDC_SCOPES, rustfs_config::oidc::OIDC_OTHER_AUDIENCES];
 
-static CONFIG_BUCKET: LazyLock<String> = LazyLock::new(|| format!("{RUSTFS_META_BUCKET}{SLASH_SEPARATOR}{CONFIG_PREFIX}"));
-
 type ServerConfigDecryptFn = crate::bucket::migration::LegacyBlobDecryptFn;
 
 static SERVER_CONFIG_DECRYPT_FN: LazyLock<RwLock<Option<ServerConfigDecryptFn>>> = LazyLock::new(|| RwLock::new(None));

crates/ecstore/src/config/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: configuration migration keeps legacy subsystem definitions available behind this module.
-#![allow(dead_code)]
 
 mod audit;
 pub mod com;

crates/ecstore/src/config/storageclass.rs

@@ -101,6 +101,7 @@ const DEFAULT_RRS_STORAGE_CLASS: &str = "EC:1";
 const ZERO_SET_DRIVE_COUNT_ERROR: &str = "set drive count must be greater than zero";
 
 pub static DEFAULT_INLINE_BLOCK: usize = 128 * 1024;
+const DEFAULT_INLINE_OBJECT_BUDGET: usize = 2 * DEFAULT_INLINE_BLOCK;
 
 pub static DEFAULT_KVS: LazyLock<KVS> = LazyLock::new(|| {
     let kvs = vec![
@@ -150,6 +151,8 @@ pub struct Config {
     optimize: Option<String>,
     inline_block: usize,
     initialized: bool,
+    #[serde(default, skip_serializing_if = "std::ops::Not::not")]
+    inline_block_explicit: bool,
     #[serde(skip)]
     standard_parities: Vec<PoolParity>,
     #[serde(skip)]
@@ -186,6 +189,10 @@ impl Config {
     /// A topology-bound lookup fails closed for unknown drive counts and for
     /// deserialized legacy configurations that have no pool topology. Legacy
     /// callers retain scalar compatibility through [`Self::get_parity_for_sc`].
+    #[allow(
+        dead_code,
+        reason = "per-set parity resolution asserted by this file's tests (backlog#1823)"
+    )]
     pub(crate) fn parity_for_sc(&self, sc: &str, drives_per_set: usize) -> Option<usize> {
         if !self.initialized {
             return None;
@@ -233,17 +240,19 @@ impl Config {
             .map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
     }
 
-    pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
-        if shard_size < 0 {
+    pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool {
+        if shard_size < 0 || data_shards == 0 {
             return false;
         }
 
         let shard_size = shard_size as usize;
-
-        let mut inline_block = DEFAULT_INLINE_BLOCK;
-        if self.initialized {
-            inline_block = self.inline_block;
-        }
+        // Keep the historical two-data-shard object budget while preventing
+        // wider EC layouts from multiplying the maximum inline object size.
+        let inline_block = if self.initialized && self.inline_block_explicit {
+            self.inline_block
+        } else {
+            (DEFAULT_INLINE_OBJECT_BUDGET / data_shards).min(DEFAULT_INLINE_BLOCK)
+        };
 
         if versioned {
             shard_size <= inline_block / 8
@@ -392,6 +401,7 @@ fn lookup_config_for_pools_with_env(
     }
 
     let optimize = overrides.optimize;
+    let inline_block_explicit = overrides.inline_block.is_some();
     let inline_block = if let Some(value) = overrides.inline_block {
         let block = value
             .parse::<bytesize::ByteSize>()
@@ -424,6 +434,7 @@ fn lookup_config_for_pools_with_env(
         optimize,
         inline_block,
         initialized: true,
+        inline_block_explicit,
         standard_parities,
         rrs_parities,
     })
@@ -541,22 +552,26 @@ mod tests {
     }
 
     #[test]
-    fn should_inline_preserves_exact_default_shard_boundaries() {
-        let config = Config::default();
+    fn should_inline_scales_default_threshold_by_data_shards() {
+        let config = lookup_config_for_pools_with_env(&KVS::new(), &[3, 12], no_env_overrides())
+            .expect("default inline policy should resolve for EC2+1 and EC8+4");
 
-        for (case, shard_size, versioned, expected) in [
-            ("unversioned below", 128 * 1024 - 1, false, true),
-            ("unversioned exact", 128 * 1024, false, true),
-            ("unversioned above", 128 * 1024 + 1, false, false),
-            ("versioned below", 16 * 1024 - 1, true, true),
-            ("versioned exact", 16 * 1024, true, true),
-            ("versioned above", 16 * 1024 + 1, true, false),
-            ("negative", -1, false, false),
+        for (case, shard_size, data_shards, versioned, expected) in [
+            ("EC2+1 unversioned exact", 128 * 1024, 2, false, true),
+            ("EC2+1 unversioned above", 128 * 1024 + 1, 2, false, false),
+            ("EC2+1 versioned exact", 16 * 1024, 2, true, true),
+            ("EC2+1 versioned above", 16 * 1024 + 1, 2, true, false),
+            ("EC8+4 unversioned exact", 32 * 1024, 8, false, true),
+            ("EC8+4 unversioned above", 32 * 1024 + 1, 8, false, false),
+            ("EC8+4 versioned exact", 4 * 1024, 8, true, true),
+            ("EC8+4 versioned above", 4 * 1024 + 1, 8, true, false),
+            ("negative", -1, 2, false, false),
+            ("zero data shards", 0, 0, false, false),
         ] {
             assert_eq!(
-                config.should_inline(shard_size, versioned),
+                config.should_inline(shard_size, data_shards, versioned),
                 expected,
-                "{case}: shard_size={shard_size}, versioned={versioned}"
+                "{case}: shard_size={shard_size}, data_shards={data_shards}, versioned={versioned}"
             );
         }
     }
@@ -577,13 +592,28 @@ mod tests {
             let shard_size = erasure.shard_file_size(object_size);
             assert_eq!(shard_size, expected_shard_size, "{case}: object_size={object_size}");
             assert_eq!(
-                config.should_inline(shard_size, versioned),
+                config.should_inline(shard_size, erasure.data_shards, versioned),
                 expected,
                 "{case}: object_size={object_size}, shard_size={shard_size}, versioned={versioned}"
             );
         }
     }
 
+    #[test]
+    fn explicit_inline_block_preserves_fixed_per_shard_rollback() {
+        let overrides = StorageClassEnvOverrides {
+            inline_block: Some("128KiB".to_string()),
+            ..Default::default()
+        };
+        let config = lookup_config_for_pools_with_env(&KVS::new(), &[12], overrides)
+            .expect("explicit inline block should resolve for EC8+4");
+
+        assert!(config.should_inline(128 * 1024, 8, false));
+        assert!(!config.should_inline(128 * 1024 + 1, 8, false));
+        assert!(config.should_inline(16 * 1024, 8, true));
+        assert!(!config.should_inline(16 * 1024 + 1, 8, true));
+    }
+
     #[test]
     fn write_capability_contract_only_accepts_implemented_layouts() {
         assert_eq!(SUPPORTED_WRITE_CLASSES, [STANDARD, RRS]);
@@ -777,6 +807,7 @@ mod tests {
         let encoded = serde_json::to_string(&cfg).expect("config should serialize");
         assert!(!encoded.contains("standard_parities"));
         assert!(!encoded.contains("rrs_parities"));
+        assert!(!encoded.contains("inline_block_explicit"));
 
         let decoded: Config = serde_json::from_str(&encoded).expect("legacy scalar config should deserialize");
         assert_eq!(decoded.get_parity_for_sc(STANDARD), Some(2));
@@ -786,6 +817,25 @@ mod tests {
         assert!(validate_parity(0, 0).is_err());
     }
 
+    #[test]
+    fn explicit_inline_block_survives_config_round_trip() {
+        let cfg = lookup_config_for_pools_with_env(
+            &KVS::new(),
+            &[12],
+            StorageClassEnvOverrides {
+                inline_block: Some("128KiB".to_string()),
+                ..Default::default()
+            },
+        )
+        .expect("explicit inline block should resolve");
+        assert!(cfg.should_inline(100 * 1024, 8, false));
+
+        let encoded = serde_json::to_string(&cfg).expect("config should serialize");
+        assert!(encoded.contains("\"inline_block_explicit\":true"));
+        let decoded: Config = serde_json::from_str(&encoded).expect("explicit inline config should deserialize");
+        assert!(decoded.should_inline(100 * 1024, 8, false));
+    }
+
     #[test]
     fn lookup_config_reads_rrs_from_class_rrs_key() {
         // Regression: kvs.get(RRS) used RRS="REDUCED_REDUNDANCY" instead of

crates/ecstore/src/core/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: pool coordination helpers are being migrated behind runtime owners.
-#![allow(dead_code)]
 
 pub(crate) mod pools;
 pub(crate) mod sets;

crates/ecstore/src/core/pools.rs

@@ -226,6 +226,7 @@ fn ensure_decommission_start_rebalance_meta_allowed(meta: Option<&RebalanceMeta>
     ensure_decommission_not_rebalancing(meta.is_some_and(is_rebalance_conflicting_with_decommission))
 }
 
+#[allow(dead_code, reason = "leader precondition asserted by this file's tests (backlog#1823)")]
 fn ensure_local_decommission_pool_leaders(endpoints: &EndpointServerPools, indices: &[usize]) -> Result<()> {
     for idx in indices {
         ensure_local_decommission_pool_leader(endpoints, *idx)?;
@@ -1058,11 +1059,19 @@ fn should_cleanup_decommission_source_entry(decommissioned: usize, total_version
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[allow(
+    dead_code,
+    reason = "terminal-state classification asserted by this file's tests (backlog#1823)"
+)]
 enum DecommissionTerminalState {
     Completed,
     Failed,
 }
 
+#[allow(
+    dead_code,
+    reason = "terminal-state classification asserted by this file's tests (backlog#1823)"
+)]
 fn classify_decommission_terminal_state(failed_items_present: bool) -> DecommissionTerminalState {
     if failed_items_present {
         DecommissionTerminalState::Failed

crates/ecstore/src/data_movement/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: data-movement migration keeps staged cleanup helpers until copy paths converge.
-#![allow(dead_code)]
 
 pub(crate) mod backpressure;
 
@@ -1019,6 +1018,10 @@ struct SourceCleanupDeleteBarrierState {
 }
 
 #[cfg(test)]
+#[allow(
+    dead_code,
+    reason = "installed by set_disk object tests behind `--features test-util` (backlog#1823)"
+)]
 pub(crate) struct SourceCleanupDeleteBarrier {
     state: Arc<SourceCleanupDeleteBarrierState>,
 }
@@ -1028,6 +1031,10 @@ static SOURCE_CLEANUP_DELETE_BARRIER: std::sync::OnceLock<std::sync::Mutex<Optio
     std::sync::OnceLock::new();
 
 #[cfg(test)]
+#[allow(
+    dead_code,
+    reason = "installed by set_disk object tests behind `--features test-util` (backlog#1823)"
+)]
 impl SourceCleanupDeleteBarrier {
     pub(crate) fn install(bucket: &str, object: &str) -> Self {
         let state = Arc::new(SourceCleanupDeleteBarrierState {
@@ -1166,6 +1173,7 @@ async fn find_data_movement_target_info(
     }
 }
 
+#[allow(dead_code, reason = "resume adjudication asserted by this file's tests (backlog#1823)")]
 fn resolve_data_movement_overwrite_resume_result(
     err: &Error,
     target_result: Result<Option<ObjectInfo>>,

crates/ecstore/src/diagnostics/admin_server_info.rs

@@ -653,6 +653,7 @@ fn reconcile_servers_with_endpoint_topology(
     (added, report)
 }
 
+#[allow(dead_code, reason = "exercised by this file's topology tests (backlog#1823)")]
 fn server_topology_completeness_report(
     servers: &[ServerProperties],
     endpoints: &EndpointServerPools,

crates/ecstore/src/diagnostics/get.rs

@@ -46,21 +46,49 @@ pub(crate) const GET_CODEC_STREAMING_OBJECT_CLASS_MULTIPART: &str = "multipart";
 pub(crate) const GET_STAGE_DECODE: &str = "decode";
 pub(crate) const GET_STAGE_EMIT: &str = "emit";
 pub(crate) const GET_STAGE_FILL: &str = "fill";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_FIRST_BYTE: &str = "first_byte";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_FIRST_METADATA_RESPONSE: &str = "first_metadata_response";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_FIRST_VALID_METADATA_RESPONSE: &str = "first_valid_metadata_response";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_FIRST_SHARD_READ: &str = "first_shard_read";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_FULL_BODY: &str = "full_body";
 pub(crate) const GET_STAGE_INLINE_PREPARE: &str = "inline_prepare";
 pub(crate) const GET_STAGE_LOCK_ACQUIRE: &str = "lock_acquire";
 pub(crate) const GET_STAGE_METADATA: &str = "metadata";
 pub(crate) const GET_STAGE_METADATA_CACHE_LOOKUP: &str = "metadata_cache_lookup";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_METADATA_FANOUT: &str = "metadata_fanout";
 pub(crate) const GET_STAGE_METADATA_RESOLVE: &str = "metadata_resolve";
 pub(crate) const GET_STAGE_OBJECT_INFO: &str = "object_info";
 pub(crate) const GET_STAGE_OUTPUT_LOCK_WAIT: &str = "output_lock_wait";
 pub(crate) const GET_STAGE_OUTPUT_POLL: &str = "output_poll";
 pub(crate) const GET_STAGE_PATH_DECISION: &str = "path_decision";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_QUORUM_REACHED: &str = "quorum_reached";
 pub(crate) const GET_STAGE_RANGE: &str = "range";
 pub(crate) const GET_STAGE_READER_SETUP: &str = "reader_setup";
@@ -84,12 +112,28 @@ pub(crate) const GET_STAGE_READER_STREAM_FIRST_READ: &str = "reader_stream_first
 pub(crate) const GET_STAGE_READER_TASK_BITROT_READER_INIT: &str = "reader_task_bitrot_reader_init";
 pub(crate) const GET_STAGE_READER_TASK_FILE_OPEN: &str = "reader_task_file_open";
 pub(crate) const GET_STAGE_READER_TASK_READER_CONSTRUCTION: &str = "reader_task_reader_construction";
+pub(crate) const GET_STAGE_READ_VERSION_DECODE: &str = "read_version_decode";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_CHECK: &str = "read_version_path_check";
+pub(crate) const GET_STAGE_READ_VERSION_PATH_RESOLVE: &str = "read_version_path_resolve";
+pub(crate) const GET_STAGE_READ_VERSION_XLMETA_READ: &str = "read_version_xlmeta_read";
 pub(crate) const GET_STAGE_RECONSTRUCT: &str = "reconstruct";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_RESPONSE_HANDOFF: &str = "response_handoff";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_SLOWEST_METADATA_RESPONSE: &str = "slowest_metadata_response";
 pub(crate) const GET_STAGE_STRIPE_READ: &str = "stripe_read";
 pub(crate) const GET_STAGE_STRIPE_READ_FIRST_SHARD: &str = "stripe_read_first_shard";
 pub(crate) const GET_STAGE_STRIPE_READ_QUORUM: &str = "stripe_read_quorum";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const GET_STAGE_BITROT_VERIFY: &str = "bitrot_verify";
 
 pub(crate) const GET_READER_BUFFER_OUTPUT: &str = "output";
@@ -155,8 +199,20 @@ pub(crate) const GET_METADATA_EARLY_STOP_REASON_VERSION_NOT_FOUND: &str = "versi
 pub(crate) const GET_METADATA_EARLY_STOP_REASON_VERSION_MATCH_QUORUM: &str = "version_match_quorum";
 
 /// Early-stop active state labels
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const EARLY_STOP_ACTIVE_HIT: &str = "hit";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const EARLY_STOP_ACTIVE_MISS: &str = "miss";
+#[allow(
+    dead_code,
+    reason = "GET stage vocabulary; value pinned by this file's tests, no writer yet (backlog#1823)"
+)]
 pub(crate) const EARLY_STOP_ACTIVE_DISABLED: &str = "disabled";
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
@@ -442,6 +498,10 @@ mod tests {
         assert_eq!(GET_STAGE_QUORUM_REACHED, "quorum_reached");
         assert_eq!(GET_STAGE_RANGE, "range");
         assert_eq!(GET_STAGE_READER_SETUP, "reader_setup");
+        assert_eq!(GET_STAGE_READ_VERSION_DECODE, "read_version_decode");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_CHECK, "read_version_path_check");
+        assert_eq!(GET_STAGE_READ_VERSION_PATH_RESOLVE, "read_version_path_resolve");
+        assert_eq!(GET_STAGE_READ_VERSION_XLMETA_READ, "read_version_xlmeta_read");
         assert_eq!(GET_STAGE_RECONSTRUCT, "reconstruct");
         assert_eq!(GET_STAGE_RESPONSE_HANDOFF, "response_handoff");
         assert_eq!(GET_STAGE_SLOWEST_METADATA_RESPONSE, "slowest_metadata_response");

crates/ecstore/src/diagnostics/mod.rs

@@ -13,8 +13,6 @@
 // limitations under the License.
 
 // #730: diagnostics constants are staged for request-path telemetry migration.
-#![allow(dead_code)]
 
 pub(crate) mod admin_server_info;
 pub(crate) mod get;
-pub(crate) mod pool;

crates/ecstore/src/diagnostics/pool.rs

@@ -1,30 +0,0 @@
-// 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.
-
-//! BytesPool metric label constants.
-//!
-//! These constants are used when recording pool acquisition and return
-//! metrics to avoid string allocations and ensure label consistency.
-
-/// BytesPool tier labels
-pub const POOL_TIER_SMALL: &str = "small";
-pub const POOL_TIER_MEDIUM: &str = "medium";
-pub const POOL_TIER_LARGE: &str = "large";
-pub const POOL_TIER_XLARGE: &str = "xlarge";
-
-/// BytesPool outcome labels
-pub const POOL_OUTCOME_HIT: &str = "hit";
-pub const POOL_OUTCOME_MISS: &str = "miss";
-pub const POOL_OUTCOME_RECYCLED: &str = "recycled";
-pub const POOL_OUTCOME_DROPPED: &str = "dropped";

crates/ecstore/src/disk/local.rs

@@ -15,6 +15,11 @@
 use crate::config::storageclass::DEFAULT_INLINE_BLOCK;
 use crate::crash_inject::{self, CrashPoint};
 use crate::data_usage::local_snapshot::ensure_data_usage_layout;
+use crate::diagnostics::get::{
+    GET_OBJECT_PATH_INTERNAL_META, GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_READ_VERSION_DECODE,
+    GET_STAGE_READ_VERSION_PATH_CHECK, GET_STAGE_READ_VERSION_PATH_RESOLVE, GET_STAGE_READ_VERSION_XLMETA_READ,
+    get_stage_timer_if_enabled, record_get_stage_duration_if_enabled,
+};
 #[cfg(test)]
 use crate::disk::HEALING_MARKER_PATH;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
@@ -9840,6 +9845,12 @@ impl DiskAPI for LocalDisk {
         opts: &ReadOptions,
     ) -> Result<FileInfo> {
         crate::hp_guard!("LocalDisk::read_version");
+        let stage_metrics_enabled = rustfs_io_metrics::get_stage_metrics_enabled();
+        let metrics_path = if stage_metrics_enabled && crate::bucket::utils::is_meta_bucketname(volume) {
+            GET_OBJECT_PATH_INTERNAL_META
+        } else {
+            GET_OBJECT_PATH_LEGACY_DUPLEX
+        };
         if !org_volume.is_empty() {
             let org_volume_path = self.io_get_bucket_path(org_volume)?;
             if !skip_access_checks(org_volume) {
@@ -9849,37 +9860,46 @@ impl DiskAPI for LocalDisk {
             }
         }
 
+        let path_resolve_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         let file_path = self.io_get_object_path(volume, path)?;
         let volume_dir = self.io_get_bucket_path(volume)?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_RESOLVE, path_resolve_start);
 
+        let path_check_start = get_stage_timer_if_enabled(stage_metrics_enabled);
         check_path_length(file_path.to_string_lossy().as_ref())?;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_PATH_CHECK, path_check_start);
 
         let read_data = opts.read_data;
 
-        let (data, _) = self
-            .read_raw(volume, volume_dir.clone(), file_path, read_data)
-            .await
-            .map_err(|e| {
-                if e == DiskError::FileNotFound && !version_id.is_empty() {
-                    DiskError::FileVersionNotFound
-                } else {
-                    e
-                }
-            })?;
+        let xlmeta_read_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let raw_read_result = self.read_raw(volume, volume_dir.clone(), file_path, read_data).await;
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_XLMETA_READ, xlmeta_read_start);
+        let (data, _) = raw_read_result.map_err(|e| {
+            if e == DiskError::FileNotFound && !version_id.is_empty() {
+                DiskError::FileVersionNotFound
+            } else {
+                e
+            }
+        })?;
 
-        let mut fi = get_file_info(
-            &data,
-            volume,
-            path,
-            version_id,
-            FileInfoOpts {
-                data: read_data,
-                include_free_versions: opts.incl_free_versions,
-                include_part_checksums: false,
-            },
-        )?;
-
-        fi.validate_for_metadata_read()?;
+        let decode_start = get_stage_timer_if_enabled(stage_metrics_enabled);
+        let file_info_result: Result<FileInfo> = (|| {
+            let fi = get_file_info(
+                &data,
+                volume,
+                path,
+                version_id,
+                FileInfoOpts {
+                    data: read_data,
+                    include_free_versions: opts.incl_free_versions,
+                    include_part_checksums: false,
+                },
+            )?;
+            fi.validate_for_metadata_read()?;
+            Ok(fi)
+        })();
+        record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READ_VERSION_DECODE, decode_start);
+        let mut fi = file_info_result?;
         if fi.is_canonical_delete_marker() {
             return Ok(fi);
         }
@@ -10562,6 +10582,108 @@ mod test {
         meta.marshal_msg().expect("test metadata should encode")
     }
 
+    #[test]
+    #[serial_test::serial]
+    fn read_version_records_local_metadata_stage_breakdown() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should be created");
+        let recorder = crate::test_metrics::CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint =
+                    Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
+                let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+                let bucket = "bucket";
+                let object = "stage-breakdown";
+                ensure_test_volume(&disk, bucket).await;
+
+                let object_dir = dir.path().join(bucket).join(object);
+                fs::create_dir_all(&object_dir)
+                    .await
+                    .expect("object directory should be created");
+                fs::write(
+                    object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(object, Uuid::new_v4(), None, Some(Bytes::from_static(b"inline")))),
+                )
+                .await
+                .expect("object metadata should be written");
+
+                disk.read_version(
+                    "",
+                    bucket,
+                    object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("read_version should succeed");
+
+                let meta_object = "stage-breakdown-meta";
+                let meta_object_dir = dir.path().join(RUSTFS_META_BUCKET).join(meta_object);
+                fs::create_dir_all(&meta_object_dir)
+                    .await
+                    .expect("internal metadata object directory should be created");
+                fs::write(
+                    meta_object_dir.join(STORAGE_FORMAT_FILE),
+                    test_meta(test_file_info(meta_object, Uuid::new_v4(), None, Some(Bytes::from_static(b"meta")))),
+                )
+                .await
+                .expect("internal metadata should be written");
+
+                disk.read_version(
+                    "",
+                    RUSTFS_META_BUCKET,
+                    meta_object,
+                    "",
+                    &ReadOptions {
+                        read_data: true,
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("internal metadata read_version should succeed");
+            });
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        for stage in [
+            GET_STAGE_READ_VERSION_PATH_RESOLVE,
+            GET_STAGE_READ_VERSION_PATH_CHECK,
+            GET_STAGE_READ_VERSION_XLMETA_READ,
+            GET_STAGE_READ_VERSION_DECODE,
+        ] {
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for user-bucket LocalDisk::read_version"
+            );
+            assert_eq!(
+                recorder
+                    .histogram_values(
+                        "rustfs_io_get_object_stage_duration_seconds",
+                        &[("path", GET_OBJECT_PATH_INTERNAL_META), ("stage", stage)]
+                    )
+                    .len(),
+                1,
+                "{stage} should be recorded once for internal-meta LocalDisk::read_version"
+            );
+        }
+    }
+
     #[test]
     fn inline_metadata_rollback_dir_avoids_real_data_dir_collision() {
         let target_version = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("version id should parse");

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

@@ -71,10 +71,16 @@ impl EncodedBlock {
 
 const MODERN_MAX_TOTAL_SHARDS: usize = <reed_solomon_erasure::galois_8::Field as reed_solomon_erasure::Field>::ORDER;
 const MODERN_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 64;
+const LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES: usize = 16;
+// Vec growth may retain twice the requested logical length. Keeping the logical
+// workspace at half the budget bounds each cached workspace's shard allocation to 1 MiB.
+const LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE: usize = 512 * 1024;
 
 type ModernReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<ReedSolomon>>>;
+type LegacyReedSolomonCache = RwLock<HashMap<(usize, usize), Arc<LegacyReedSolomonEncoder>>>;
 
 static MODERN_REED_SOLOMON_CACHE: OnceLock<ModernReedSolomonCache> = OnceLock::new();
+static LEGACY_REED_SOLOMON_CACHE: OnceLock<LegacyReedSolomonCache> = OnceLock::new();
 
 /// Errors returned when constructing an [`Erasure`] codec.
 #[derive(Debug, thiserror::Error)]
@@ -141,43 +147,61 @@ pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize {
 struct LegacyReedSolomonEncoder {
     data_shards: usize,
     parity_shards: usize,
-    encoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
-    decoder_cache: std::sync::RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
-}
-
-impl Clone for LegacyReedSolomonEncoder {
-    fn clone(&self) -> Self {
-        Self {
-            data_shards: self.data_shards,
-            parity_shards: self.parity_shards,
-            encoder_cache: std::sync::RwLock::new(None),
-            decoder_cache: std::sync::RwLock::new(None),
-        }
-    }
+    cache_workspaces: bool,
+    encoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
+    decoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
 }
 
 impl LegacyReedSolomonEncoder {
-    fn new(_data_shards: usize, _parity_shards: usize) -> io::Result<Self> {
+    fn new(data_shards: usize, parity_shards: usize) -> io::Result<Self> {
+        Self::with_workspace_cache(data_shards, parity_shards, false)
+    }
+
+    fn with_workspace_cache(data_shards: usize, parity_shards: usize, cache_workspaces: bool) -> io::Result<Self> {
         Ok(Self {
-            data_shards: _data_shards,
-            parity_shards: _parity_shards,
-            encoder_cache: std::sync::RwLock::new(None),
-            decoder_cache: std::sync::RwLock::new(None),
+            data_shards,
+            parity_shards,
+            cache_workspaces,
+            encoder_cache: RwLock::new(None),
+            decoder_cache: RwLock::new(None),
         })
     }
 
+    fn logical_shard_bytes_upper_bound(&self, shard_len: usize) -> Option<usize> {
+        let aligned_shard_len = shard_len.checked_add(63)?.checked_div(64)?.checked_mul(64)?;
+        let high_rate_decoder_work_count = self
+            .parity_shards
+            .checked_next_power_of_two()?
+            .checked_add(self.data_shards)?
+            .checked_next_power_of_two()?;
+        let low_rate_decoder_work_count = self
+            .data_shards
+            .checked_next_power_of_two()?
+            .checked_add(self.parity_shards)?
+            .checked_next_power_of_two()?;
+        aligned_shard_len.checked_mul(high_rate_decoder_work_count.max(low_rate_decoder_work_count))
+    }
+
+    fn should_cache_workspace(&self, shard_len: usize) -> bool {
+        self.cache_workspaces
+            && self
+                .logical_shard_bytes_upper_bound(shard_len)
+                .is_some_and(|bytes| bytes <= LEGACY_REED_SOLOMON_CACHE_MAX_LOGICAL_SHARD_BYTES_PER_WORKSPACE)
+    }
+
     fn encode(&self, shards: SmallVec<[&mut [u8]; 16]>) -> io::Result<()> {
         let mut shards_vec: Vec<&mut [u8]> = shards.into_vec();
         if shards_vec.is_empty() {
             return Ok(());
         }
         let shard_len = shards_vec[0].len();
+        let cached_encoder = self
+            .encoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?
+            .take();
         let mut encoder = {
-            let mut cache_guard = self
-                .encoder_cache
-                .write()
-                .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?;
-            match cache_guard.take() {
+            match cached_encoder {
                 Some(mut cached) => {
                     if cached.reset(self.data_shards, self.parity_shards, shard_len).is_err() {
                         reed_solomon_simd::ReedSolomonEncoder::new(self.data_shards, self.parity_shards, shard_len)
@@ -204,10 +228,15 @@ impl LegacyReedSolomonEncoder {
             }
         }
         drop(result);
-        *self
-            .encoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .encoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return encoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(encoder);
+            }
+        }
         Ok(())
     }
 
@@ -221,13 +250,13 @@ impl LegacyReedSolomonEncoder {
             .find_map(|s| s.as_ref().map(|v| v.len()))
             .ok_or_else(|| io::Error::other("No valid shards found for reconstruction"))?;
 
+        let cached_decoder = self
+            .decoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?
+            .take();
         let mut decoder = {
-            let mut cache_guard = self
-                .decoder_cache
-                .write()
-                .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?;
-
-            match cache_guard.take() {
+            match cached_decoder {
                 Some(mut cached_decoder) => {
                     if let Err(e) = cached_decoder.reset(self.data_shards, self.parity_shards, shard_len) {
                         warn!("Failed to reset SIMD decoder: {:?}, creating new one", e);
@@ -274,10 +303,15 @@ impl LegacyReedSolomonEncoder {
 
         drop(result);
 
-        *self
-            .decoder_cache
-            .write()
-            .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder);
+        if self.should_cache_workspace(shard_len) {
+            let mut cache = self
+                .decoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to return decoder to cache"))?;
+            if cache.is_none() {
+                *cache = Some(decoder);
+            }
+        }
 
         Ok(())
     }
@@ -435,6 +469,39 @@ fn cached_modern_reed_solomon(data_shards: usize, parity_shards: usize) -> Resul
     Ok(encoder)
 }
 
+fn cached_legacy_reed_solomon(data_shards: usize, parity_shards: usize) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
+    let cache = LEGACY_REED_SOLOMON_CACHE.get_or_init(|| RwLock::new(HashMap::new()));
+    cached_legacy_reed_solomon_in(cache, data_shards, parity_shards)
+}
+
+fn cached_legacy_reed_solomon_in(
+    cache: &LegacyReedSolomonCache,
+    data_shards: usize,
+    parity_shards: usize,
+) -> io::Result<Arc<LegacyReedSolomonEncoder>> {
+    let key = (data_shards, parity_shards);
+    if let Some(encoder) = cache
+        .read()
+        .unwrap_or_else(|poisoned| poisoned.into_inner())
+        .get(&key)
+        .cloned()
+    {
+        return Ok(encoder);
+    }
+
+    let mut cache = cache.write().unwrap_or_else(|poisoned| poisoned.into_inner());
+    if let Some(existing) = cache.get(&key) {
+        return Ok(Arc::clone(existing));
+    }
+    if cache.len() < LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
+        let encoder = Arc::new(LegacyReedSolomonEncoder::with_workspace_cache(data_shards, parity_shards, true)?);
+        cache.insert(key, Arc::clone(&encoder));
+        return Ok(encoder);
+    }
+    drop(cache);
+    Ok(Arc::new(LegacyReedSolomonEncoder::new(data_shards, parity_shards)?))
+}
+
 fn encode_parity_shards<F>(shards: &mut [Option<Vec<u8>>], data_shards: usize, parity_shards: usize, encode: F) -> io::Result<()>
 where
     F: FnOnce(SmallVec<[&mut [u8]; 16]>) -> io::Result<()>,
@@ -551,7 +618,7 @@ pub struct Erasure {
     pub data_shards: usize,
     pub parity_shards: usize,
     encoder: Option<ReedSolomonEncoder>,
-    legacy_encoder: Option<LegacyReedSolomonEncoder>,
+    legacy_encoder: Option<Arc<LegacyReedSolomonEncoder>>,
     pub block_size: usize,
     uses_legacy: bool,
     _id: Uuid,
@@ -687,7 +754,7 @@ impl Erasure {
 
         let legacy_encoder = if uses_legacy && parity_shards > 0 {
             Some(
-                LegacyReedSolomonEncoder::new(data_shards, parity_shards)
+                cached_legacy_reed_solomon(data_shards, parity_shards)
                     .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?,
             )
         } else {
@@ -1405,7 +1472,7 @@ mod tests {
         assert_eq!(cloned.block_size, legacy.block_size);
         assert!(cloned.uses_legacy);
 
-        let data = b"legacy clone should keep independent SIMD caches";
+        let data = b"legacy clone should preserve SIMD codec behavior";
         let encoded = cloned.encode_data(data).expect("legacy clone should encode");
         let mut shards = optional_shards(&encoded);
         shards[0] = None;
@@ -1413,6 +1480,93 @@ mod tests {
         assert_eq!(recover_data(&shards, cloned.data_shards, data.len()), data);
     }
 
+    #[test]
+    fn legacy_codecs_share_process_cache_across_erasure_instances() {
+        let first = Erasure::new_with_options(6, 3, 64, true)
+            .legacy_encoder
+            .expect("legacy codec should be initialized");
+        let second = Erasure::new_with_options(6, 3, 128, true)
+            .legacy_encoder
+            .expect("same legacy shard layout should be initialized");
+
+        assert!(Arc::ptr_eq(&first, &second));
+    }
+
+    #[test]
+    fn legacy_workspace_cache_rejects_oversize_buffers_and_isolates_layouts() {
+        let four_plus_two = Erasure::new_with_options(4, 2, 64, true)
+            .legacy_encoder
+            .expect("legacy codec should be initialized");
+        let four_plus_one = Erasure::new_with_options(4, 1, 64, true)
+            .legacy_encoder
+            .expect("distinct parity layout should be initialized");
+        let three_plus_two = Erasure::new_with_options(3, 2, 64, true)
+            .legacy_encoder
+            .expect("distinct data layout should be initialized");
+
+        assert!(!Arc::ptr_eq(&four_plus_two, &four_plus_one));
+        assert!(!Arc::ptr_eq(&four_plus_two, &three_plus_two));
+        assert_eq!(four_plus_two.logical_shard_bytes_upper_bound(64 * 1024), Some(512 * 1024));
+        assert!(four_plus_two.should_cache_workspace(64 * 1024));
+        assert!(!four_plus_two.should_cache_workspace(64 * 1024 + 1));
+
+        let nine_plus_seven =
+            LegacyReedSolomonEncoder::with_workspace_cache(9, 7, true).expect("9+7 legacy codec should construct");
+        assert_eq!(nine_plus_seven.logical_shard_bytes_upper_bound(16 * 1024), Some(512 * 1024));
+        assert!(nine_plus_seven.should_cache_workspace(16 * 1024));
+        assert!(!nine_plus_seven.should_cache_workspace(16 * 1024 + 1));
+
+        let uncached = LegacyReedSolomonEncoder::new(4, 2).expect("uncached legacy codec should construct");
+        assert!(!uncached.should_cache_workspace(64));
+    }
+
+    #[test]
+    fn saturated_legacy_codec_cache_does_not_retain_more_workspaces() {
+        let cache = RwLock::new(HashMap::new());
+        for parity_shards in 1..=LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES {
+            let cached =
+                cached_legacy_reed_solomon_in(&cache, 32, parity_shards).expect("cacheable legacy codec should construct");
+            assert!(cached.cache_workspaces);
+        }
+
+        let uncached =
+            cached_legacy_reed_solomon_in(&cache, 31, 1).expect("uncached legacy codec should construct after saturation");
+        assert!(!uncached.cache_workspaces);
+        assert_eq!(
+            cache.read().expect("cache lock should remain healthy").len(),
+            LEGACY_REED_SOLOMON_CACHE_MAX_ENTRIES
+        );
+    }
+
+    #[test]
+    fn concurrent_legacy_codecs_preserve_byte_exact_results() {
+        let barrier = Arc::new(std::sync::Barrier::new(2));
+        let payloads = [vec![0x35; 257], vec![0xca; 1025]];
+
+        std::thread::scope(|scope| {
+            let handles = payloads.each_ref().map(|payload| {
+                let barrier = Arc::clone(&barrier);
+                scope.spawn(move || {
+                    let erasure = Erasure::new_with_options(6, 3, 2048, true);
+                    barrier.wait();
+                    let encoded = erasure.encode_data(payload).expect("concurrent legacy encode should succeed");
+                    barrier.wait();
+
+                    let mut shards = optional_shards(&encoded);
+                    shards[0] = None;
+                    erasure
+                        .decode_data(&mut shards)
+                        .expect("concurrent legacy decode should reconstruct the missing shard");
+                    recover_data(&shards, erasure.data_shards, payload.len())
+                })
+            });
+
+            for (handle, payload) in handles.into_iter().zip(payloads.iter()) {
+                assert_eq!(handle.join().expect("concurrent legacy codec worker should not panic"), *payload);
+            }
+        });
+    }
+
     #[test]
     fn legacy_verify_reports_invalid_empty_valid_and_corrupt_parity_sets() {
         let legacy = LegacyReedSolomonEncoder::new(2, 2).expect("legacy encoder should construct");

crates/ecstore/src/error/mod.rs

@@ -13,13 +13,12 @@
 // limitations under the License.
 
 // #730: error taxonomy still exposes compatibility variants while callers move to contracts.
-#![allow(dead_code)]
 
 use crate::bucket::error::BucketMetadataError;
 use crate::disk::error::DiskError;
 use crate::storage_api_contracts::{error::StorageErrorCode, range::HTTPRangeError};
 use rustfs_utils::path::decode_dir_object;
-use s3s::{S3Error, S3ErrorCode};
+use s3s::S3ErrorCode;
 
 pub type Error = StorageError;
 pub type Result<T> = core::result::Result<T, Error>;
@@ -902,6 +901,7 @@ pub fn is_err_decommission_running(err: &Error) -> bool {
     matches!(err, &StorageError::DecommissionAlreadyRunning)
 }
 
+#[allow(dead_code, reason = "predicate asserted by this file's tests (backlog#1823)")]
 pub fn is_err_rebalance_running(err: &Error) -> bool {
     matches!(err, &StorageError::RebalanceAlreadyRunning)
 }
@@ -910,14 +910,11 @@ pub fn is_err_operation_canceled(err: &Error) -> bool {
     matches!(err, &StorageError::OperationCanceled)
 }
 
+#[allow(dead_code, reason = "predicate asserted by this file's tests (backlog#1823)")]
 pub fn is_err_not_initialized(err: &Error) -> bool {
     err.to_string().contains("errServerNotInitialized") || err.to_string().contains("ServerNotInitialized")
 }
 
-pub fn is_err_io(err: &Error) -> bool {
-    matches!(err, &StorageError::Io(_))
-}
-
 /// Strict "not found" predicate that only matches genuine object/version/volume
 /// absence errors: `FileNotFound`/`VolumeNotFound`/`FileVersionNotFound`/
 /// `ObjectNotFound`/`VersionNotFound`.
@@ -1078,21 +1075,9 @@ pub struct GenericError {
 
 #[derive(Debug, thiserror::Error, PartialEq, Eq)]
 pub enum ObjectApiError {
-    #[error("Operation timed out")]
-    OperationTimedOut,
-
-    #[error("etag of the object has changed")]
-    InvalidETag,
-
     #[error("BackendDown")]
     BackendDown(String),
 
-    #[error("Unsupported headers in Metadata")]
-    UnsupportedMetadata,
-
-    #[error("Method not allowed: {}/{}", .0.bucket, .0.object)]
-    MethodNotAllowed(GenericError),
-
     #[error("The operation is not valid for the current state of the object {}/{}({})", .0.bucket, .0.object, .0.version_id)]
     InvalidObjectState(GenericError),
 }
@@ -1175,30 +1160,6 @@ pub fn error_resp_to_object_err(err: ErrorResponse, params: Vec<&str>) -> std::i
     err
 }
 
-pub fn storage_to_object_err(err: Error, params: Vec<&str>) -> S3Error {
-    let storage_err = &err;
-    let mut bucket: String = "".to_string();
-    let mut object: String = "".to_string();
-    if !params.is_empty() {
-        bucket = params[0].to_string();
-    }
-    if params.len() >= 2 {
-        object = decode_dir_object(params[1]);
-    }
-    match storage_err {
-        StorageError::MethodNotAllowed => S3Error::with_message(
-            S3ErrorCode::MethodNotAllowed,
-            ObjectApiError::MethodNotAllowed(GenericError {
-                bucket,
-                object,
-                ..Default::default()
-            })
-            .to_string(),
-        ),
-        _ => s3s::S3Error::with_message(S3ErrorCode::Custom("err".into()), err.to_string()),
-    }
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

crates/ecstore/src/event/mod.rs

@@ -13,8 +13,6 @@
 // limitations under the License.
 
 // #730: event target types are retained for notification owner migration.
-#![allow(dead_code)]
 
 pub mod name;
-pub mod targetid;
 pub mod targetlist;

crates/ecstore/src/event/targetid.rs

@@ -1,25 +0,0 @@
-#![allow(clippy::all)]
-// 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.
-
-pub struct TargetID {
-    id: String,
-    name: String,
-}
-
-impl TargetID {
-    fn to_string(&self) -> String {
-        format!("{}:{}", self.id, self.name)
-    }
-}

crates/ecstore/src/event/targetlist.rs

@@ -12,18 +12,16 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-use crate::event::targetid::TargetID;
 use std::sync::atomic::AtomicI64;
 
+/// Placeholder notification target list held by `EventNotifier`.
+///
+/// The working notification stack lives in `rustfs-notify` / `rustfs-targets`;
+/// this type never grew past its counter. `total_events` is read by the
+/// notifier's log line but nothing increments it, so that field reports zero.
 #[derive(Default)]
 pub struct TargetList {
-    pub current_send_calls: AtomicI64,
     pub total_events: AtomicI64,
-    pub events_skipped: AtomicI64,
-    pub events_errors_total: AtomicI64,
-    //pub targets: HashMap<TargetID, Target>,
-    //pub queue:   AsyncEvent,
-    //pub targetStats: HashMap<TargetID, TargetStat>,
 }
 
 impl TargetList {
@@ -31,14 +29,3 @@ impl TargetList {
         TargetList::default()
     }
 }
-
-struct TargetStat {
-    current_send_calls: i64,
-    total_events: i64,
-    failed_events: i64,
-}
-
-struct TargetIDResult {
-    id: TargetID,
-    err: std::io::Error,
-}

crates/ecstore/src/io_support/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: I/O backend selection keeps test-only and staged rio helpers scoped here.
-#![allow(dead_code)]
 
 pub(crate) mod bitrot;
 pub(crate) mod compress;

crates/ecstore/src/io_support/rio.rs

@@ -25,9 +25,20 @@ use tokio::io::AsyncRead;
 
 #[cfg(feature = "rio-v2")]
 const MINIO_S2_COMPRESSION_SCHEME: &str = "klauspost/compress/s2";
+// The S2 padding multiple rio-v2 pads compressed streams to before
+// encryption. Only the padding test asserts it today, so the lib target sees
+// it as unused (backlog#1823).
 #[cfg(feature = "rio-v2")]
+#[allow(dead_code, reason = "on-disk contract asserted by the rio-v2 padding test (backlog#1823)")]
 const ENCRYPTED_S2_PADDING_MULTIPLE: usize = 256;
 
+/// Which rio implementation this build compiled in. Only the feature-seam
+/// guard test in lib.rs reads it, so the lib target sees it as unused
+/// (backlog#1823).
+#[allow(
+    dead_code,
+    reason = "asserted by the rio backend feature-seam test in lib.rs (backlog#1823)"
+)]
 pub const fn backend_name() -> &'static str {
     #[cfg(feature = "rio-v2")]
     {
@@ -53,17 +64,6 @@ pub fn compression_metadata_value(algorithm: CompressionAlgorithm) -> String {
     }
 }
 
-pub fn compression_scheme_to_algorithm(scheme: &str) -> std::io::Result<CompressionAlgorithm> {
-    #[cfg(feature = "rio-v2")]
-    if scheme.eq_ignore_ascii_case(MINIO_S2_COMPRESSION_SCHEME) {
-        // rio_v2 currently routes all compressed-object handling through the S2
-        // reader implementation, so the enum is only a placeholder token here.
-        return Ok(CompressionAlgorithm::default());
-    }
-
-    CompressionAlgorithm::from_str(scheme)
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ReadCompressionBackend {
     Legacy,
@@ -82,6 +82,11 @@ pub fn compression_scheme_to_read_plan(scheme: &str) -> std::io::Result<(Compres
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 pub enum ReadEncryptionBackend {
     Legacy,
+    // Never constructed today — every read still selects Legacy — but the
+    // decrypt paths below carry live match arms for it. This is the rio-v2
+    // read seam (backlog#1638 / #1835), not dead code: deleting the variant
+    // would delete those arms with it.
+    #[allow(dead_code, reason = "rio-v2 read seam; match arms below are live (backlog#1823)")]
     V2,
 }
 

crates/ecstore/src/layout/endpoints.rs

@@ -209,15 +209,12 @@ impl AsMut<Vec<Endpoints>> for PoolEndpointList {
 }
 
 impl PoolEndpointList {
-    /// creates a list of endpoints per pool, resolves their relevant
-    /// hostnames and discovers those are local or remote.
-    async fn create_pool_endpoints(server_addr: &str, disks_layout: &DisksLayout) -> Result<Self> {
-        Self::create_pool_endpoints_with(server_addr, disks_layout, None, None).await
-    }
-
-    /// Same as [`create_pool_endpoints`] but lets tests inject an explicit
-    /// startup topology convergence policy and local endpoint host instead of
-    /// resolving them from the environment.
+    /// Creates a list of endpoints per pool, resolves their relevant hostnames
+    /// and discovers whether those are local or remote.
+    ///
+    /// The policy and host overrides let tests inject an explicit startup
+    /// topology convergence policy and local endpoint host instead of
+    /// resolving them from the environment; production passes `None` for both.
     async fn create_pool_endpoints_with(
         server_addr: &str,
         disks_layout: &DisksLayout,
@@ -594,6 +591,10 @@ impl PoolEndpointList {
 }
 
 const DNS_RETRY_BASE_DELAY: Duration = Duration::from_millis(500);
+#[allow(
+    dead_code,
+    reason = "retry-cap bound asserted by this file's dns_retry_delay tests (backlog#1823)"
+)]
 const DNS_RETRY_MAX_DELAY: Duration = Duration::from_secs(8);
 const DNS_RETRY_JITTER_PERCENT: u64 = 20;
 /// Minimum spacing between "still retrying" warnings so a long orchestrated

crates/ecstore/src/layout/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: set-layout contracts are staged while ECStore ownership boundaries shrink.
-#![allow(dead_code)]
 
 //! Static ECStore layout boundaries.
 //!

crates/ecstore/src/layout/set_layout.rs

@@ -4,6 +4,7 @@ use std::io::{Error, Result};
 use uuid::Uuid;
 
 #[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 pub(crate) struct StaticSetLayoutSnapshot {
     pub(crate) deployment_id: Uuid,
     pub(crate) set_count: usize,
@@ -12,6 +13,7 @@ pub(crate) struct StaticSetLayoutSnapshot {
     pub(crate) distribution_algo: DistributionAlgoVersion,
 }
 
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 impl StaticSetLayoutSnapshot {
     pub(crate) fn from_format(format: &FormatV3) -> Self {
         let disk_ids = format.erasure.sets.clone();
@@ -39,17 +41,20 @@ impl StaticSetLayoutSnapshot {
 }
 
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 pub(crate) struct SetDiskPosition {
     pub(crate) set_index: usize,
     pub(crate) disk_index: usize,
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 pub(crate) struct RuntimeSetLayoutPlan {
     pub(crate) sets: Vec<Vec<RuntimeSetDrivePlan>>,
     lock_hosts_by_set: Vec<Vec<String>>,
 }
 
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 impl RuntimeSetLayoutPlan {
     pub(crate) fn from_endpoint_hosts<S>(set_count: usize, drives_per_set: usize, endpoint_hosts: &[S]) -> Result<Self>
     where
@@ -108,6 +113,7 @@ impl RuntimeSetLayoutPlan {
 }
 
 #[derive(Debug, Clone, PartialEq, Eq)]
+#[allow(dead_code, reason = "ESET-001 layout model; exercised by this file's tests (backlog#1823)")]
 pub(crate) struct RuntimeSetDrivePlan {
     pub(crate) set_index: usize,
     pub(crate) disk_index: usize,

crates/ecstore/src/object_api/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: object API readers keep staged compatibility paths during facade migration.
-#![allow(dead_code)]
 
 use crate::bucket::metadata_sys::get_versioning_config;
 use crate::bucket::replication::{

crates/ecstore/src/object_api/readers.rs

@@ -15,6 +15,7 @@
 use super::*;
 
 use crate::io_support::rio::Index;
+use std::mem::MaybeUninit;
 
 #[cfg(feature = "rio-v2")]
 const DARE_PAYLOAD_SIZE: i64 = 64 * 1024;
@@ -448,10 +449,16 @@ impl GetObjectReader {
 }
 
 enum ReadTransform {
-    Plain {
-        visible_offset: usize,
-        visible_length: i64,
-    },
+    // Written but never read by production code: the enclosing struct already
+    // carries the same pair as `storage_offset`/`storage_length`. They survive
+    // as the read plan's test-visible record — four tests assert them by
+    // literal pattern (`Plain { visible_offset: 6, visible_length: 4 }`), which
+    // rustc does not count as a read.
+    #[allow(
+        dead_code,
+        reason = "asserted by literal pattern in this file's read-plan tests (backlog#1823)"
+    )]
+    Plain { visible_offset: usize, visible_length: i64 },
     Compressed {
         algorithm: CompressionAlgorithm,
         backend: crate::io_support::rio::ReadCompressionBackend,
@@ -922,7 +929,7 @@ struct SkipReader<R> {
     inner: R,
     bytes_to_skip: usize,
     bytes_skipped: usize,
-    scratch: Vec<u8>,
+    scratch: Box<[MaybeUninit<u8>]>,
 }
 
 impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
@@ -931,7 +938,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
             inner,
             bytes_to_skip,
             bytes_skipped: 0,
-            scratch: vec![0u8; 8192],
+            scratch: Box::<[u8]>::new_uninit_slice(8192),
         }
     }
 }
@@ -943,7 +950,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> AsyncRead for SkipReader<R> {
         while this.bytes_skipped < this.bytes_to_skip {
             let remaining = this.bytes_to_skip - this.bytes_skipped;
             let scratch_len = remaining.min(this.scratch.len());
-            let mut scratch_buf = ReadBuf::new(&mut this.scratch[..scratch_len]);
+            let mut scratch_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]);
             match Pin::new(&mut this.inner).poll_read(cx, &mut scratch_buf) {
                 Poll::Pending => return Poll::Pending,
                 Poll::Ready(Err(err)) => return Poll::Ready(Err(err)),
@@ -974,7 +981,7 @@ pub struct RangedDecompressReader<R: AsyncRead + Unpin + Send + Sync + 'static>
     target_length: usize,
     current_offset: usize,
     bytes_returned: usize,
-    scratch: Vec<u8>,
+    scratch: Box<[MaybeUninit<u8>]>,
     drain_on_done: bool,
     drain_task: Option<tokio::task::JoinHandle<()>>,
 }
@@ -1012,7 +1019,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
             target_length: actual_length,
             current_offset: 0,
             bytes_returned: 0,
-            scratch: vec![0u8; 8192],
+            scratch: Box::<[u8]>::new_uninit_slice(8192),
             drain_on_done,
             drain_task: None,
         })
@@ -1062,7 +1069,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
             }
 
             let scratch_len = std::cmp::min(this.scratch.len(), std::cmp::max(buf_capacity, 1));
-            let mut temp_read_buf = ReadBuf::new(&mut this.scratch[..scratch_len]);
+            let mut temp_read_buf = ReadBuf::uninit(&mut this.scratch[..scratch_len]);
 
             let Some(inner) = this.inner.as_mut() else {
                 return Poll::Ready(Ok(()));
@@ -1114,7 +1121,8 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
                             );
 
                             if bytes_to_return > 0 {
-                                let data_slice = &this.scratch[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
+                                let data_slice =
+                                    &temp_read_buf.filled()[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
                                 buf.put_slice(data_slice);
                                 this.bytes_returned += bytes_to_return;
 
@@ -1133,7 +1141,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
                             std::cmp::min(n, std::cmp::min(buf.remaining(), this.target_length - this.bytes_returned));
 
                         if bytes_to_return > 0 {
-                            buf.put_slice(&this.scratch[..bytes_to_return]);
+                            buf.put_slice(&temp_read_buf.filled()[..bytes_to_return]);
                             this.bytes_returned += bytes_to_return;
 
                             tracing::trace!("Returned {} bytes at offset {}", bytes_to_return, old_offset);
@@ -1203,20 +1211,7 @@ impl<R: AsyncRead + Unpin + Send + 'static> AsyncRead for StreamConsumer<R> {
 
 impl<R: AsyncRead + Unpin + Send + 'static> Drop for StreamConsumer<R> {
     fn drop(&mut self) {
-        if self.consumer_task.is_none() && self.inner.is_some() {
-            let mut inner = self.inner.take().unwrap();
-            let task = tokio::spawn(async move {
-                let mut buf = [0u8; 8192];
-                loop {
-                    match inner.read(&mut buf).await {
-                        Ok(0) => break,    // EOF
-                        Ok(_) => continue, // Keep consuming
-                        Err(_) => break,   // Error, stop consuming
-                    }
-                }
-            });
-            self.consumer_task = Some(task);
-        }
+        self.ensure_consumer_started();
     }
 }
 
@@ -1263,6 +1258,43 @@ mod tests {
     use temp_env::async_with_vars;
     use tokio::io::AsyncReadExt;
 
+    #[derive(Debug)]
+    struct PendingPartialReader {
+        data: &'static [u8],
+        position: usize,
+        pending: bool,
+    }
+
+    impl PendingPartialReader {
+        fn new(data: &'static [u8]) -> Self {
+            Self {
+                data,
+                position: 0,
+                pending: true,
+            }
+        }
+    }
+
+    impl AsyncRead for PendingPartialReader {
+        fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if self.pending {
+                self.pending = false;
+                cx.waker().wake_by_ref();
+                return Poll::Pending;
+            }
+            if self.position == self.data.len() {
+                return Poll::Ready(Ok(()));
+            }
+
+            let length = buf.remaining().min(3).min(self.data.len() - self.position);
+            let end = self.position + length;
+            buf.put_slice(&self.data[self.position..end]);
+            self.position = end;
+            self.pending = true;
+            Poll::Ready(Ok(()))
+        }
+    }
+
     const TEST_DIRECT_KEY_HEADER: &str = "x-rustfs-test-direct-key";
     const TEST_OBJECT_KEY_HEADER: &str = "x-rustfs-test-object-key";
     const TEST_NONCE_HEADER: &str = "x-rustfs-test-nonce";
@@ -1400,6 +1432,36 @@ mod tests {
         assert_eq!(result, b"World");
     }
 
+    #[tokio::test]
+    async fn uninitialized_scratch_preserves_partial_pending_and_eof_reads() {
+        let mut skipped = SkipReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5);
+        let mut skipped_output = Vec::new();
+        skipped
+            .read_to_end(&mut skipped_output)
+            .await
+            .expect("skip reader should survive partial pending reads through EOF");
+        assert_eq!(skipped_output, b"56789abcdef");
+
+        let mut ranged = RangedDecompressReader::new(PendingPartialReader::new(b"0123456789abcdef"), 5, 7, 16)
+            .expect("valid range should construct");
+        let mut ranged_output = Vec::new();
+        ranged
+            .read_to_end(&mut ranged_output)
+            .await
+            .expect("range reader should survive partial pending reads through EOF");
+        assert_eq!(ranged_output, b"56789ab");
+    }
+
+    #[tokio::test]
+    async fn uninitialized_skip_scratch_reports_early_eof() {
+        let mut reader = SkipReader::new(PendingPartialReader::new(b"short"), 6);
+        let error = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("EOF before the skip boundary must remain visible");
+        assert_eq!(error.kind(), std::io::ErrorKind::UnexpectedEof);
+    }
+
     #[tokio::test]
     async fn test_ranged_decompress_reader_from_start() {
         let original_data = b"Hello, World! This is a test.";

crates/ecstore/src/object_api/types.rs

@@ -172,6 +172,7 @@ impl ObjectLockConfigSnapshot {
         }
     }
 
+    #[allow(dead_code, reason = "snapshot-scope predicate asserted by this file's tests (backlog#1823)")]
     pub(crate) fn is_for_store_bucket(
         &self,
         store_id: Uuid,

crates/ecstore/src/runtime/global.rs

@@ -31,7 +31,6 @@ use std::{
 use tokio::sync::{OnceCell, RwLock};
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
-use uuid::Uuid;
 
 pub const DISK_ASSUME_UNKNOWN_SIZE: u64 = 1 << 30;
 pub const DISK_MIN_INODES: u64 = 1000;
@@ -109,18 +108,6 @@ pub fn set_global_rustfs_port(value: u16) {
     }
 }
 
-/// Set the global deployment id
-///
-/// # Arguments
-/// * `id` - The Uuid to set as the global deployment id
-///
-/// # Returns
-/// * None
-///
-pub fn set_global_deployment_id(id: Uuid) {
-    current_ctx().set_deployment_id(id);
-}
-
 /// Get the global deployment id
 ///
 /// # Returns
@@ -288,19 +275,6 @@ pub fn get_global_region() -> Option<s3s::region::Region> {
     current_ctx().region()
 }
 
-/// Initialize the global background services cancellation token
-///
-/// # Arguments
-/// * `cancel_token` - The CancellationToken instance to set globally
-///
-/// # Returns
-/// * `Ok(())` if successful
-/// * `Err(CancellationToken)` if setting fails
-///
-pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> Result<(), CancellationToken> {
-    current_ctx().init_background_cancel_token(cancel_token)
-}
-
 /// Get the global background services cancellation token
 ///
 /// # Returns
@@ -310,18 +284,6 @@ pub fn get_background_services_cancel_token() -> Option<CancellationToken> {
     current_ctx().background_cancel_token()
 }
 
-/// Create and initialize the global background services cancellation token
-///
-/// # Returns
-/// * `CancellationToken` - The newly created global cancellation token
-///
-pub fn create_background_services_cancel_token() -> CancellationToken {
-    let cancel_token = CancellationToken::new();
-    init_background_services_cancel_token(cancel_token.clone())
-        .expect("background services cancel token should be initialized once during startup");
-    cancel_token
-}
-
 /// Shutdown all background services gracefully
 ///
 /// # Returns

crates/ecstore/src/runtime/instance.rs

@@ -402,6 +402,10 @@ impl InstanceContext {
     }
 
     #[cfg(test)]
+    #[allow(
+        dead_code,
+        reason = "driven by the tier-delete-journal recovery test behind `--features test-util` (backlog#1823)"
+    )]
     pub(crate) fn wake_tier_delete_journal_recovery(&self) {
         self.tier_delete_journal_recovery_wakeup.notify_one();
     }

crates/ecstore/src/runtime/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: runtime source migration keeps fallback handles until all owners inject state.
-#![allow(dead_code)]
 
 pub(crate) mod global;
 pub(crate) mod instance;

crates/ecstore/src/runtime/sources.rs

@@ -38,7 +38,6 @@ use crate::{
         set_object_layer, update_erasure_type,
     },
     services::batch_processor::{GlobalBatchProcessors, get_global_processors},
-    services::event_notification::EventNotifier,
     services::notification_sys::{NotificationSys, get_global_notification_sys},
     services::tier::tier::TierConfigMgr,
     store::ECStore,
@@ -143,6 +142,10 @@ pub async fn setup_is_erasure_sd() -> bool {
     is_erasure_sd().await
 }
 
+#[allow(
+    dead_code,
+    reason = "setup-type override used only by tests across this crate (backlog#1823)"
+)]
 pub(crate) async fn current_setup_type() -> SetupType {
     if setup_is_dist_erasure().await {
         SetupType::DistErasure
@@ -155,6 +158,10 @@ pub(crate) async fn current_setup_type() -> SetupType {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "setup-type override used only by tests across this crate (backlog#1823)"
+)]
 pub(crate) async fn set_setup_type(setup_type: SetupType) {
     update_erasure_type(setup_type).await;
 }
@@ -232,14 +239,6 @@ pub(crate) fn ensure_test_rpc_secret() {
     let _ = rustfs_credentials::set_global_rpc_secret(TEST_RPC_SECRET.to_owned());
 }
 
-pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option<usize> {
-    get_global_storage_class_snapshot().get_parity_for_sc(storage_class.unwrap_or_default())
-}
-
-pub(crate) fn storage_class_should_inline(shard_size: i64, versioned: bool) -> bool {
-    get_global_storage_class_snapshot().should_inline(shard_size, versioned)
-}
-
 pub(crate) fn deployment_upload_id(upload_id: &str) -> String {
     base64_simd::URL_SAFE_NO_PAD
         .encode_to_string(format!("{}.{}", get_global_deployment_id().unwrap_or_default(), upload_id).as_bytes())
@@ -332,21 +331,6 @@ pub(crate) fn storage_class_config_snapshot() -> Arc<storageclass::Config> {
     get_global_storage_class_snapshot()
 }
 
-/// Scalar STANDARD / RRS parity for backend-info reporting.
-///
-/// Retained for the rebalance/backend-info path. `get_parity_for_sc` returns
-/// `None` when the runtime config is uninitialized or (post per-pool support)
-/// when pools disagree, so STANDARD falls back to the caller's default and RRS
-/// stays `None` — matching the pre-per-pool scalar reporting.
-pub(crate) fn backend_storage_class_parities(default_standard_parity: usize) -> (Option<usize>, Option<usize>) {
-    let sc = get_global_storage_class_snapshot();
-    let standard = sc
-        .get_parity_for_sc(storageclass::CLASS_STANDARD)
-        .or(Some(default_standard_parity));
-    let reduced_redundancy = sc.get_parity_for_sc(storageclass::RRS);
-    (standard, reduced_redundancy)
-}
-
 pub(crate) fn set_storage_class_config(config: storageclass::Config) {
     set_global_storage_class(config);
 }
@@ -414,10 +398,6 @@ pub fn transition_state_handle() -> Arc<TransitionState> {
     crate::runtime::global::current_ctx().transition_state()
 }
 
-pub(crate) fn event_notifier_handle() -> Arc<RwLock<EventNotifier>> {
-    crate::runtime::global::current_ctx().event_notifier()
-}
-
 pub(crate) async fn local_disk_by_path(path: &str) -> Option<DiskStore> {
     local_disk_map_handle().read().await.get(path).cloned().flatten()
 }
@@ -511,30 +491,6 @@ pub(crate) async fn local_disk_set_drive(
     instance_ctx.local_disk_set_drives().read().await[pool_idx][set_idx][disk_idx].clone()
 }
 
-pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option<DiskStore> {
-    let set_drives = local_disk_set_drives_handle();
-    let global_set_drives = set_drives.read().await;
-    if global_set_drives.is_empty() {
-        return local_disk_map_handle()
-            .read()
-            .await
-            .get(&endpoint.to_string())
-            .cloned()
-            .unwrap_or(None);
-    }
-
-    let pool_idx = usize::try_from(endpoint.pool_idx).ok()?;
-    let set_idx = usize::try_from(endpoint.set_idx).ok()?;
-    let disk_idx = usize::try_from(endpoint.disk_idx).ok()?;
-
-    global_set_drives
-        .get(pool_idx)
-        .and_then(|sets| sets.get(set_idx))
-        .and_then(|disks| disks.get(disk_idx))
-        .cloned()
-        .unwrap_or(None)
-}
-
 pub(crate) async fn local_disk_paths() -> Vec<String> {
     local_disk_map_handle().read().await.keys().cloned().collect()
 }

crates/ecstore/src/services/notification_sys.rs

@@ -206,6 +206,38 @@ pub(crate) fn remote_version_state_fleet_proof_matches(proof: &RemoteVersionStat
     })
 }
 
+#[cfg(test)]
+pub(crate) struct RemoteVersionStateFleetProofGuard;
+
+#[cfg(test)]
+impl Drop for RemoteVersionStateFleetProofGuard {
+    fn drop(&mut self) {
+        replace_remote_version_state_fleet_proof(None);
+    }
+}
+
+#[cfg(test)]
+pub(crate) fn install_remote_version_state_fleet_proof_for_test(topology_fingerprint: &str) -> RemoteVersionStateFleetProofGuard {
+    match REMOTE_VERSION_STATE_PROBE_TOPOLOGY.set(topology_fingerprint.to_string()) {
+        Ok(()) => {}
+        Err(_)
+            if REMOTE_VERSION_STATE_PROBE_TOPOLOGY
+                .get()
+                .is_some_and(|current| current == topology_fingerprint) => {}
+        Err(_) => panic!("remote version state test topology is already bound to another fingerprint"),
+    }
+    let peer_epochs = BTreeMap::new();
+    if let Some(err) = publish_remote_version_state_probe_result(
+        remote_version_state_fleet_proof_slot(),
+        topology_fingerprint,
+        Ok(peer_epochs),
+        Instant::now(),
+    ) {
+        panic!("test proof installation must not fail: {err}");
+    }
+    RemoteVersionStateFleetProofGuard
+}
+
 fn remote_version_state_fleet_proof_valid_at(
     proof: Option<&RemoteVersionStateFleetProof>,
     expected_topology: &str,

crates/ecstore/src/set_disk/mod.rs

@@ -880,12 +880,44 @@ mod prepared_get_object_metadata_tests {
     use super::*;
     use crate::ecstore_validation_blackbox::make_local_set_disks;
     use crate::object_api::{BLOCK_SIZE_V2, PutObjReader};
-    use crate::set_disk::core::io_primitives::disk_call_counters;
+    use crate::set_disk::core::io_primitives::{bounded_metadata_fanout_order, disk_call_counters, rename_fanout_barrier};
     use crate::storage_api_contracts::bucket::{BucketOperations as _, MakeBucketOptions};
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+    use crate::test_metrics::CapturingRecorder;
     use http::HeaderMap;
     use tokio::io::AsyncReadExt;
 
+    const READ_VERSION_BARRIER_GUARD: std::time::Duration = std::time::Duration::from_secs(10);
+
+    fn object_with_initial_data_shards(bucket: &str, prefix: &str) -> String {
+        (0..1000)
+            .map(|index| format!("{prefix}-{index}.bin"))
+            .find(|name| {
+                let order = bounded_metadata_fanout_order(bucket, name, 4, 2);
+                let distribution = FileInfo::new(&[bucket, name].join("/"), 2, 2).erasure.distribution;
+                let mut seen = [false; 2];
+                for disk_index in order.into_iter().take(3) {
+                    if let Some(block_index @ 1..=2) = distribution.get(disk_index).copied() {
+                        seen[block_index - 1] = true;
+                    }
+                }
+                seen.into_iter().all(|seen| seen)
+            })
+            .expect("test should find an object whose initial fanout covers both data shards")
+    }
+
+    fn bounded_spare_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(3)
+            .expect("4-disk test geometry should leave one bounded spare disk")
+    }
+
+    fn bounded_slow_initial_disk_index(bucket: &str, object: &str) -> usize {
+        *bounded_metadata_fanout_order(bucket, object, 4, 2)
+            .get(2)
+            .expect("4-disk test geometry should include a third initial metadata disk")
+    }
+
     #[tokio::test]
     async fn prepared_metadata_is_consumed_exactly_once() {
         let snapshot = GetObjectFileInfo::owned(FileInfo::default(), Vec::new(), Vec::new());
@@ -1002,6 +1034,307 @@ mod prepared_get_object_metadata_tests {
         );
     }
 
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn inline_data_read_early_stop_reader_returns_exact_body() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "inline-data-read-early-stop-reader";
+        let object = object_with_initial_data_shards(bucket, "inline-data-read-early-stop-reader-object");
+        let payload = b"inline early-stop reader payload".repeat(256);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, object_size, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("inline object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let slow_initial_disk = bounded_slow_initial_disk_index(bucket, &object);
+                        let barrier =
+                            rename_fanout_barrier::arm(&object, slow_initial_disk, rename_fanout_barrier::PHASE_READ_VERSION);
+                        let calls = disk_call_counters::observe(&object);
+                        let set_disks_for_read = Arc::clone(&set_disks);
+                        let opts_for_read = opts.clone();
+                        let object_for_read = object.clone();
+                        let mut open_reader = tokio::spawn(async move {
+                            set_disks_for_read
+                                .get_object_reader(bucket, &object_for_read, None, HeaderMap::new(), &opts_for_read)
+                                .await
+                        });
+
+                        tokio::time::timeout(READ_VERSION_BARRIER_GUARD, barrier.wait_until_paused())
+                            .await
+                            .expect("bounded inline GET should pause a slow initial metadata read");
+                        let mut reader = tokio::time::timeout(READ_VERSION_BARRIER_GUARD, &mut open_reader)
+                            .await
+                            .expect("production inline GET should return before the paused metadata response")
+                            .expect("inline GET reader task should not panic")
+                            .expect("inline GET reader should open");
+                        let object_size = reader.object_info.size;
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("inline GET body should stream");
+
+                        (restored, object_size, calls.total(disk_call_counters::KIND_READ_VERSION))
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(object_size, payload.len() as i64);
+        assert_eq!(restored, payload);
+        assert_eq!(calls_total, 4, "bounded production GET should schedule the initial quorum plus one spare");
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "bounded production GET should record all scheduled metadata tasks"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![3.0],
+            "bounded production GET should record only observed metadata responses as completed"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![1.0],
+            "bounded production GET should record the aborted slow metadata task"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn prepared_metadata_uses_full_fanout_even_when_data_read_early_stop_is_enabled() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "prepared-metadata-early-stop-enabled";
+        let object = object_with_initial_data_shards(bucket, "prepared-metadata-early-stop-enabled-object");
+        let payload = b"prepared metadata early-stop enabled payload".repeat(16);
+        let recorder = CapturingRecorder::default();
+        let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+
+        let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+            runtime.block_on(async {
+                let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                let opts = ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                };
+
+                set_disks
+                    .make_bucket(bucket, &MakeBucketOptions::default())
+                    .await
+                    .expect("bucket should be created");
+                let mut put_reader = PutObjReader::from_vec(payload.clone());
+                set_disks
+                    .put_object(bucket, &object, &mut put_reader, &opts)
+                    .await
+                    .expect("object should be written");
+
+                temp_env::async_with_vars(
+                    [
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                        ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                    ],
+                    async {
+                        let calls = disk_call_counters::observe(&object);
+                        let metadata = set_disks
+                            .prepare_get_object_metadata(bucket, &object, &opts)
+                            .await
+                            .expect("prepared metadata should resolve");
+                        let calls_total = calls.total(disk_call_counters::KIND_READ_VERSION);
+
+                        let mut reader = set_disks
+                            .get_object_reader_with_prepared_metadata(bucket, &object, None, HeaderMap::new(), &opts, metadata)
+                            .await
+                            .expect("prepared body reader should open");
+                        let mut restored = Vec::new();
+                        reader
+                            .stream
+                            .read_to_end(&mut restored)
+                            .await
+                            .expect("prepared body should stream");
+                        (restored, calls_total)
+                    },
+                )
+                .await
+            })
+        });
+        rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+        assert_eq!(restored, payload);
+        assert_eq!(
+            calls_total, 4,
+            "prepared metadata must opt out of data-read early-stop until the read shape is known"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_scheduled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata should schedule the full metadata fanout"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_completed",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![4.0],
+            "prepared metadata must wait for every scheduled metadata response"
+        );
+        assert_eq!(
+            recorder.histogram_values(
+                "rustfs_io_get_object_metadata_fanout_cancelled",
+                &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+            ),
+            vec![0.0],
+            "prepared metadata must not cancel metadata responses"
+        );
+    }
+
+    #[test]
+    #[serial_test::serial(body_cache_hook)]
+    fn data_read_early_stop_request_shapes_full_wait_in_production_reader() {
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("current-thread runtime should build");
+        let bucket = "data-read-early-stop-shape-reader";
+        let payload = b"shape-gated inline reader payload".repeat(256);
+
+        for (object_prefix, range, configure_opts, expected_body) in [
+            (
+                "data-read-early-stop-range-reader-object",
+                Some(HTTPRangeSpec {
+                    start: 0,
+                    end: 3,
+                    is_suffix_length: false,
+                }),
+                None,
+                payload[..4].to_vec(),
+            ),
+            ("data-read-early-stop-part-reader-object", None, Some(1), payload.clone()),
+        ] {
+            let recorder = CapturingRecorder::default();
+            let previous_gate = rustfs_io_metrics::get_stage_metrics_enabled();
+            rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+            let (restored, calls_total) = metrics::with_local_recorder(&recorder, || {
+                runtime.block_on(async {
+                    let (_dirs, set_disks) = make_local_set_disks(4, 2).await;
+                    let object = object_with_initial_data_shards(bucket, object_prefix);
+                    let mut opts = ObjectOptions {
+                        no_lock: true,
+                        ..Default::default()
+                    };
+                    opts.part_number = configure_opts;
+
+                    set_disks
+                        .make_bucket(bucket, &MakeBucketOptions::default())
+                        .await
+                        .expect("bucket should be created");
+                    let mut put_reader = PutObjReader::from_vec(payload.clone());
+                    set_disks
+                        .put_object(bucket, &object, &mut put_reader, &opts)
+                        .await
+                        .expect("inline object should be written");
+
+                    temp_env::async_with_vars(
+                        [
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_DATA_READ_EARLY_STOP_ENABLE", Some("true")),
+                            ("RUSTFS_GET_METADATA_EARLY_STOP_BOUNDED_FANOUT", Some("true")),
+                        ],
+                        async {
+                            let calls = disk_call_counters::observe(&object);
+                            let mut reader = set_disks
+                                .get_object_reader(bucket, &object, range, HeaderMap::new(), &opts)
+                                .await
+                                .expect("shape-gated GET reader should open");
+                            let mut restored = Vec::new();
+                            reader
+                                .stream
+                                .read_to_end(&mut restored)
+                                .await
+                                .expect("shape-gated GET body should stream");
+                            (restored, calls.total(disk_call_counters::KIND_READ_VERSION))
+                        },
+                    )
+                    .await
+                })
+            });
+            rustfs_io_metrics::set_get_stage_metrics_enabled(previous_gate);
+
+            assert_eq!(restored, expected_body);
+            assert_eq!(calls_total, 4, "shape-gated production GET should keep full metadata fanout");
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_scheduled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET should schedule the full metadata fanout"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_completed",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![4.0],
+                "shape-gated production GET must wait for every scheduled metadata response"
+            );
+            assert_eq!(
+                recorder.histogram_values(
+                    "rustfs_io_get_object_metadata_fanout_cancelled",
+                    &[("path", GET_OBJECT_PATH_LEGACY_DUPLEX)]
+                ),
+                vec![0.0],
+                "shape-gated production GET must not cancel metadata responses"
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial(body_cache_hook)]
     async fn prepared_reader_rebuilds_object_info_when_precomputed_value_is_absent() {
@@ -1105,7 +1438,7 @@ impl SetDisks {
         object: &str,
         opts: &ObjectOptions,
     ) -> Result<PreparedGetObjectMetadata> {
-        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, true).await?;
+        let snapshot = self.get_object_fileinfo(bucket, object, opts, true, false).await?;
         let object_info = build_get_object_info(snapshot.fi(), bucket, object, opts.versioned || opts.version_suspended);
         Ok(PreparedGetObjectMetadata {
             snapshot,
@@ -2408,12 +2741,12 @@ mod write_layout_tests {
 
         let held_layout = resolve_write_layout(&held, 0, 4, 2, None, false).expect("held snapshot should remain valid");
         assert_eq!(held_layout.parity_drives, 2);
-        assert!(held.should_inline(512, false));
+        assert!(held.should_inline(512, held_layout.data_drives, false));
 
         let current = published.load_full();
         let current_layout = resolve_write_layout(&current, 0, 4, 2, None, false).expect("new snapshot should resolve");
         assert_eq!(current_layout.parity_drives, 1);
-        assert!(!current.should_inline(512, false));
+        assert!(!current.should_inline(512, current_layout.data_drives, false));
     }
 }
 
@@ -2459,6 +2792,9 @@ pub struct SetDisks {
     get_object_metadata_cache: moka::future::Cache<GetObjectMetadataCacheKey, Arc<GetObjectMetadataCacheEntry>>,
     get_object_metadata_cache_hash_builder: std::collections::hash_map::RandomState,
     get_object_metadata_cache_generations: Arc<[AtomicU64]>,
+    /// GET codecs keyed by every persisted layout dimension that affects
+    /// decoding. Clones of a set share the memoized shells.
+    erasure_cache: Arc<ErasureCache>,
     pub lockers: Vec<Arc<dyn LockClient>>,
     shared_lockers: Arc<[Arc<dyn LockClient>]>,
     local_lock_manager: Arc<rustfs_lock::GlobalLockManager>,
@@ -2481,6 +2817,137 @@ pub struct SetDisks {
     storage_class_config_override: Arc<std::sync::RwLock<Option<Arc<storageclass::Config>>>>,
 }
 
+const ERASURE_CACHE_MAX_ENTRIES: usize = 32;
+
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
+struct ErasureCacheKey {
+    data_shards: usize,
+    parity_shards: usize,
+    block_size: usize,
+    uses_legacy: bool,
+}
+
+struct ErasureCache {
+    entries: parking_lot::RwLock<HashMap<ErasureCacheKey, Arc<coding::Erasure>>>,
+}
+
+impl std::fmt::Debug for ErasureCache {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        formatter
+            .debug_struct("ErasureCache")
+            .field("entries", &self.entries.read().len())
+            .finish()
+    }
+}
+
+impl ErasureCache {
+    fn new() -> Self {
+        Self {
+            entries: parking_lot::RwLock::new(HashMap::new()),
+        }
+    }
+
+    fn get_or_try_insert(
+        &self,
+        key: ErasureCacheKey,
+    ) -> std::result::Result<Arc<coding::Erasure>, coding::ErasureConstructionError> {
+        if let Some(erasure) = self.entries.read().get(&key) {
+            return Ok(Arc::clone(erasure));
+        }
+
+        // Serialize first construction for a key so concurrent cold GETs still
+        // create exactly one shell. Codec construction never awaits.
+        let mut entries = self.entries.write();
+        if let Some(erasure) = entries.get(&key) {
+            return Ok(Arc::clone(erasure));
+        }
+        let erasure = Arc::new(coding::Erasure::try_new_with_options(
+            key.data_shards,
+            key.parity_shards,
+            key.block_size,
+            key.uses_legacy,
+        )?);
+        if entries.len() < ERASURE_CACHE_MAX_ENTRIES {
+            entries.insert(key, Arc::clone(&erasure));
+        }
+        Ok(erasure)
+    }
+
+    fn get_for_file_info(&self, fi: &FileInfo) -> Result<Arc<coding::Erasure>> {
+        self.get_or_try_insert(ErasureCacheKey {
+            data_shards: fi.erasure.data_blocks,
+            parity_shards: fi.erasure.parity_blocks,
+            block_size: fi.erasure.block_size,
+            uses_legacy: fi.uses_legacy_checksum,
+        })
+        .map_err(Error::from)
+    }
+}
+
+#[cfg(test)]
+mod erasure_cache_tests {
+    use super::*;
+
+    #[test]
+    fn reuses_shells_and_keeps_every_layout_dimension_in_the_key() {
+        let cache = ErasureCache::new();
+        let base = ErasureCacheKey {
+            data_shards: 4,
+            parity_shards: 2,
+            block_size: 1_048_576,
+            uses_legacy: false,
+        };
+        let first = cache.get_or_try_insert(base).expect("modern shell should construct");
+        let reused = cache.get_or_try_insert(base).expect("same modern shell should be cached");
+        assert!(Arc::ptr_eq(&first, &reused));
+
+        for distinct in [
+            ErasureCacheKey { data_shards: 3, ..base },
+            ErasureCacheKey {
+                parity_shards: 1,
+                ..base
+            },
+            ErasureCacheKey {
+                block_size: 524_288,
+                ..base
+            },
+            ErasureCacheKey {
+                uses_legacy: true,
+                ..base
+            },
+        ] {
+            let shell = cache.get_or_try_insert(distinct).expect("distinct shell should construct");
+            assert!(!Arc::ptr_eq(&first, &shell));
+        }
+        assert_eq!(cache.entries.read().len(), 5);
+    }
+
+    #[test]
+    fn does_not_cache_invalid_layouts_or_grow_past_the_bound() {
+        let cache = ErasureCache::new();
+        let invalid = ErasureCacheKey {
+            data_shards: 4,
+            parity_shards: 2,
+            block_size: 0,
+            uses_legacy: false,
+        };
+        assert!(cache.get_or_try_insert(invalid).is_err());
+        assert!(cache.entries.read().is_empty());
+
+        for block_size in 1..=(ERASURE_CACHE_MAX_ENTRIES + 1) {
+            cache
+                .get_or_try_insert(ErasureCacheKey {
+                    data_shards: 4,
+                    parity_shards: 2,
+                    block_size,
+                    uses_legacy: false,
+                })
+                .expect("bounded cache fixture should construct");
+        }
+        assert_eq!(cache.entries.read().len(), ERASURE_CACHE_MAX_ENTRIES);
+    }
+}
+
 #[derive(Clone, Debug, Eq, PartialEq)]
 struct GetObjectMetadataCacheKey {
     bucket: Arc<str>,
@@ -2879,6 +3346,7 @@ impl SetDisks {
                     .map(|_| AtomicU64::new(0))
                     .collect::<Vec<_>>(),
             ),
+            erasure_cache: Arc::new(ErasureCache::new()),
             lockers,
             shared_lockers,
             // Sourced from the instance context so each instance owns its lock
@@ -3411,9 +3879,15 @@ fn collect_inline_data_shard_fileinfos_by_index<'a>(
         if block_index == 0 || block_index > data_shards {
             continue;
         }
+        if file_info.erasure.index != block_index {
+            continue;
+        }
         if !file_info.has_valid_erasure_geometry() {
             continue;
         }
+        if !core::io_primitives::metadata_early_stop_candidate_matches(file_info, fi) {
+            continue;
+        }
         if file_info.data.as_ref().is_none_or(|data| data.is_empty()) {
             continue;
         }
@@ -9221,6 +9695,9 @@ mod tests {
             HashAlgorithm::HighwayHash256S
         };
         let shards = erasure.encode_data(payload).expect("payload should encode");
+        let version_id = Some(Uuid::new_v4());
+        let data_dir = Some(Uuid::new_v4());
+        let mod_time = Some(OffsetDateTime::now_utc());
         let mut files = Vec::with_capacity(shards.len());
 
         for shard in shards {
@@ -9233,6 +9710,16 @@ mod tests {
             writer.shutdown().await.expect("inline writer should shutdown");
             let data = writer.into_inline_data().expect("inline data should be retained");
             let mut file = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
+            file.volume = "bucket".to_string();
+            file.name = "object".to_string();
+            file.size = i64::try_from(payload.len()).expect("test payload should fit i64");
+            file.is_latest = true;
+            file.version_id = version_id;
+            file.data_dir = data_dir;
+            file.mod_time = mod_time;
+            file.metadata.insert("etag".to_string(), "etag-inline".to_string());
+            file.add_object_part(1, "part-etag-inline".to_string(), payload.len(), file.mod_time, file.size, None, None);
+            file.set_inline_data();
             file.erasure.index = files.len() + 1;
             file.data = Some(Bytes::from(data));
             files.push(file);
@@ -9245,16 +9732,40 @@ mod tests {
         inline_bitrot_files_for_payload_with_mode(payload, false).await
     }
 
+    fn disk_ordered_fileinfos(files: &[FileInfo]) -> Vec<FileInfo> {
+        let distribution = &files
+            .first()
+            .expect("inline data shard fixture should include metadata")
+            .erasure
+            .distribution;
+        distribution
+            .iter()
+            .map(|block_index| {
+                files
+                    .get(block_index.checked_sub(1).expect("erasure block indexes are one-based"))
+                    .expect("inline data shard fixture should include every distributed shard")
+                    .clone()
+            })
+            .collect()
+    }
+
     fn inline_data_shard_fileinfo(
-        name: &str,
         data_blocks: usize,
         parity_blocks: usize,
         erasure_index: usize,
         distribution: &[usize],
         data: Option<&'static [u8]>,
     ) -> FileInfo {
-        let mut fi = FileInfo::new(name, data_blocks, parity_blocks);
-        fi.name = name.to_string();
+        let mut fi = FileInfo::new("object", data_blocks, parity_blocks);
+        fi.name = "object".to_string();
+        fi.volume = "bucket".to_string();
+        fi.size = 4;
+        fi.is_latest = true;
+        fi.data_dir = Some(Uuid::nil());
+        fi.mod_time = Some(OffsetDateTime::UNIX_EPOCH);
+        fi.metadata.insert("etag".to_string(), "etag-inline".to_string());
+        fi.add_object_part(1, "part-etag-inline".to_string(), 4, fi.mod_time, 4, None, None);
+        fi.set_inline_data();
         fi.erasure.index = erasure_index;
         fi.erasure.distribution = distribution.to_vec();
         fi.data = data.map(Bytes::from_static);
@@ -9264,36 +9775,41 @@ mod tests {
     #[test]
     fn collect_inline_data_shards_by_index_uses_distribution_order() {
         let distribution = vec![3, 1, 5, 2, 4, 6];
-        let mut fi = FileInfo::new("object", 4, 2);
+        let mut fi = inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-3", 4, 2, 3, &distribution, Some(b"c")),
-            inline_data_shard_fileinfo("block-1", 4, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("parity-5", 4, 2, 5, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("block-2", 4, 2, 2, &distribution, Some(b"b")),
-            inline_data_shard_fileinfo("block-4", 4, 2, 4, &distribution, Some(b"d")),
-            inline_data_shard_fileinfo("parity-6", 4, 2, 6, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(4, 2, 3, &distribution, Some(b"c")),
+            inline_data_shard_fileinfo(4, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(4, 2, 5, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(4, 2, 2, &distribution, Some(b"b")),
+            inline_data_shard_fileinfo(4, 2, 4, &distribution, Some(b"d")),
+            inline_data_shard_fileinfo(4, 2, 6, &distribution, Some(b"q")),
         ];
 
         let data_files =
             collect_inline_data_shard_fileinfos_by_index(&files, &fi, 4, |_| true).expect("all data shards should be collected");
 
         assert_eq!(
-            data_files.iter().map(|file| file.name.as_str()).collect::<Vec<_>>(),
-            ["block-1", "block-2", "block-3", "block-4"]
+            data_files
+                .iter()
+                .map(|file| file.data.as_deref().expect("fixture carries inline bytes"))
+                .collect::<Vec<_>>(),
+            [b"a".as_slice(), b"b".as_slice(), b"c".as_slice(), b"d".as_slice()]
         );
     }
 
     #[test]
     fn collect_inline_data_shards_by_index_rejects_missing_data_shard() {
         let distribution = vec![1, 2, 3, 4];
-        let mut fi = FileInfo::new("object", 2, 2);
+        let mut fi = inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"x"));
+        fi.erasure.index = 1;
         fi.erasure.distribution = distribution.clone();
         let files = vec![
-            inline_data_shard_fileinfo("block-1", 2, 2, 1, &distribution, Some(b"a")),
-            inline_data_shard_fileinfo("block-2", 2, 2, 2, &distribution, None),
-            inline_data_shard_fileinfo("parity-3", 2, 2, 3, &distribution, Some(b"p")),
-            inline_data_shard_fileinfo("parity-4", 2, 2, 4, &distribution, Some(b"q")),
+            inline_data_shard_fileinfo(2, 2, 1, &distribution, Some(b"a")),
+            inline_data_shard_fileinfo(2, 2, 2, &distribution, None),
+            inline_data_shard_fileinfo(2, 2, 3, &distribution, Some(b"p")),
+            inline_data_shard_fileinfo(2, 2, 4, &distribution, Some(b"q")),
         ];
 
         assert!(collect_inline_data_shard_fileinfos_by_index(&files, &fi, 2, |_| true).is_none());
@@ -9426,10 +9942,8 @@ mod tests {
 
         let payload = vec![b'i'; 192 * 1024];
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload.len() as i64;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload.len() as i64, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let disks = vec![Some(disk); erasure.total_shard_count()];
         let metrics_size_bucket = rustfs_io_metrics::get_object_size_bucket(fi.size);
@@ -9437,8 +9951,9 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
-            &files,
+            &disk_files,
             &disks,
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9469,10 +9984,8 @@ mod tests {
         let payload = vec![b'v'; 64 * 1024];
         let payload_size = i64::try_from(payload.len()).expect("test payload size should fit i64");
         let (erasure, files, _read_length, _checksum_algo) = inline_bitrot_files_for_payload(&payload).await;
-        let mut fi = FileInfo::new("bucket/object", erasure.data_shards, erasure.parity_shards);
-        fi.size = payload_size;
-        fi.data = files[0].data.clone();
-        fi.add_object_part(1, String::new(), payload.len(), None, payload_size, None, None);
+        let fi = files[0].clone();
+        let disk_files = disk_ordered_fileinfos(&files);
 
         let mut object_info = ObjectInfo {
             size: payload_size,
@@ -9502,8 +10015,9 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
-            &files,
+            &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
             true,
             GET_CODEC_STREAMING_OBJECT_CLASS_PLAIN_SINGLE_PART,
@@ -9582,6 +10096,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             &fi,
             &files,
             &disks,
@@ -9667,6 +10182,7 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
+                Arc::new(ErasureCache::new()),
                 range_offset,
                 range_length as i64,
                 &mut writer,
@@ -9778,6 +10294,7 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
+                Arc::new(ErasureCache::new()),
                 0,
                 total_size as i64,
                 &mut writer,

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

@@ -1425,6 +1425,33 @@ impl SetDisks {
     /// post-heal tail — reclaim identically. Never fails the heal: delete errors
     /// are logged and swallowed. Callers must gate this on `!opts.dry_run`.
     async fn reclaim_orphan_data_dirs_best_effort(&self, bucket: &str, object: &str) {
+        match self.reconcile_old_data_cleanup_receipts(bucket, object).await {
+            Ok(removed) if removed > 0 => {
+                debug!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket,
+                    object,
+                    removed,
+                    state = "old_data_cleanup_receipt_reconciled",
+                    "Set disk old-data cleanup receipts reconciled"
+                );
+            }
+            Ok(_) => {}
+            Err(e) => {
+                warn!(
+                    event = EVENT_SET_DISK_HEAL,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_SET_DISK,
+                    bucket,
+                    object,
+                    error = %e,
+                    state = "old_data_cleanup_receipt_reconcile_failed",
+                    "Set disk old-data cleanup receipt reconcile failed"
+                );
+            }
+        }
         match self.reclaim_orphan_data_dirs(bucket, object).await {
             Ok(removed) if removed > 0 => {
                 debug!(

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

@@ -22,6 +22,11 @@
 
 use super::super::*;
 use super::bitrot_self_verify::{BitrotSelfVerifyTarget, drop_failed_writer_disks, verify_written_bitrot_shards};
+use super::object::{
+    assign_object_transaction_epoch, object_transaction_fencing_fleet_proof, object_transaction_fencing_fleet_proof_matches,
+    object_transaction_fencing_requested, old_data_cleanup_receipt_path, read_object_transaction_epoch_fence,
+    verify_object_transaction_epoch_fence,
+};
 use crate::crash_inject::{self, CrashPoint};
 use crate::multipart_listing::paginate_multipart_listing;
 use futures::{StreamExt, stream};
@@ -63,6 +68,9 @@ pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
+    BeforeTransactionEpochVerify,
+    BeforeObjectPublication,
+    AfterObjectPublication,
     AfterRename,
 }
 
@@ -153,13 +161,24 @@ impl Drop for MultipartCommitBarrier {
 
 #[cfg(test)]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
-    let barrier = MULTIPART_COMMIT_BARRIER
-        .get_or_init(|| std::sync::Mutex::new(None))
-        .lock()
-        .expect("multipart commit barrier mutex should not poison")
-        .as_ref()
-        .filter(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
-        .cloned();
+    let barrier = {
+        let mut slot = MULTIPART_COMMIT_BARRIER
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("multipart commit barrier mutex should not poison");
+        if slot
+            .as_ref()
+            .is_some_and(|barrier| barrier.bucket == bucket && barrier.object == object && barrier.pause == pause)
+        {
+            if pause == MultipartCommitPause::BeforeTransactionEpochVerify {
+                slot.take()
+            } else {
+                slot.clone()
+            }
+        } else {
+            None
+        }
+    };
     if let Some(barrier) = barrier
         && let Ok(previous) = barrier.arrivals.fetch_update(Ordering::AcqRel, Ordering::Acquire, |current| {
             (current < barrier.expected_arrivals).then_some(current + 1)
@@ -2296,140 +2315,196 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
-        let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
-
-        // Crash-consistency injection: hard power loss after the upload is fully
-        // staged and locked but before the authoritative rename_data commit. No
-        // disk has moved the staged data, so a crash here must leave any prior
-        // committed version byte-for-byte intact (rustfs/backlog#864) and the
-        // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, object) {
-            return Err(StorageError::Unexpected);
+        let transaction_fencing_proof = object_transaction_fencing_fleet_proof();
+        if object_transaction_fencing_requested() && transaction_fencing_proof.is_none() {
+            return Err(Error::other("object transaction fencing requires a live fleet capability proof"));
         }
+        let transaction_epoch_fence = if transaction_fencing_proof.is_some() {
+            Some(read_object_transaction_epoch_fence(self.as_ref(), bucket, object).await?)
+        } else {
+            None
+        };
+        let transaction_epoch =
+            transaction_epoch_fence.map(|_| assign_object_transaction_epoch(&shuffle_disks, &mut parts_metadatas));
 
-        // The trailing `_` drops the rename_data old-size backfill
-        // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
-        // `get_object_info` lookup, so the backfill has no consumer here yet.
-        let (online_disks, convergence, op_old_dir, cleanup_disks, _) = Self::rename_data(
-            &shuffle_disks,
-            RUSTFS_META_MULTIPART_BUCKET,
-            &upload_id_path,
-            &parts_metadatas,
-            bucket,
-            object,
-            write_quorum,
-        )
-        .await?;
-
-        // Detach admission before any post-commit await: client cancellation
-        // must not couple durable convergence repair to cleanup work.
-        if convergence.needs_heal() {
-            let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
-                bucket.to_string(),
-                Some(object.to_string()),
-                false,
-                Some(HealChannelPriority::Normal),
-                Some(self.pool_index),
-                Some(self.set_index),
-            );
-            request.object_version_id = fi
-                .version_id
-                .or_else(|| opts.version_suspended.then(Uuid::nil))
-                .map(|version_id| version_id.to_string());
-            tokio::spawn(async move {
-                let _ = rustfs_common::heal_channel::send_heal_request(request).await;
-            });
-        }
-
-        // Crash-consistency injection: hard power loss after the authoritative
-        // rename_data commit succeeded but before the stale part.N.meta cleanup.
-        // The new version is durably committed and visible, so a crash here must
-        // leave the object readable as the new version; the un-reclaimed staging
-        // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
-        // Compiles to a no-op outside `#[cfg(test)]`.
-        if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object) {
-            return Err(StorageError::Unexpected);
-        }
-
-        // backlog#946: reclaim the stale per-part metadata (and any superfluous
-        // part.N data files no longer in the completed set) only *after* the
-        // authoritative rename_data commit above has succeeded. If rename_data
-        // fails write quorum and returns via `?`, the upload directory must keep
-        // its part.N.meta so a retried CompleteMultipartUpload can still read the
-        // parts; deleting them before the commit would strand the upload
-        // permanently. This mirrors the "clean up only after commit" pattern
-        // already used for the old data-dir GC and the upload-dir delete_all below.
-        self.cleanup_multipart_path(&parts).await;
-
-        if let Some(old_dir) = op_old_dir {
-            let committed_dir = fi.data_dir.unwrap_or_default().to_string();
-            // backlog#898: best-effort reclaim of the dereferenced old data dir.
-            // Returns a receipt (never `Err`); a failed GC must not turn an
-            // already-committed multipart completion into a 503.
-            let cleanup = self
-                .commit_rename_data_dir(&cleanup_disks, bucket, object, &old_dir.to_string(), &committed_dir, write_quorum)
-                .await;
-            self.report_old_data_dir_cleanup(bucket, object, &old_dir.to_string(), &cleanup)
-                .await;
-        }
-
-        if let Some(stage_start) = complete_tail_stage_start {
-            rustfs_io_metrics::record_put_object_stage_duration(
-                "multipart_complete_tail",
-                stage_start.elapsed().as_secs_f64() * 1000.0,
-            );
-        }
-
-        #[cfg(test)]
-        pause_multipart_commit(bucket, object, MultipartCommitPause::AfterRename).await;
-
-        let cleanup_store = self.clone();
-        let cleanup_upload_id_path = upload_id_path.clone();
-        let cleanup_bucket = bucket.to_owned();
-        let cleanup_object = object.to_owned();
-        let cleanup_upload_id = upload_id.to_owned();
-        let cleanup_handle = tokio::spawn(async move {
+        let commit_set = self.clone();
+        let commit_bucket = bucket.to_owned();
+        let commit_object = object.to_owned();
+        let commit_upload_id = upload_id.to_owned();
+        let commit_upload_id_path = upload_id_path.clone();
+        let commit_version_suspended = opts.version_suspended;
+        let commit_is_versioned = opts.versioned || opts.version_suspended;
+        let commit_capacity_scope_token = opts.capacity_scope_token;
+        let commit_object_lock_guard = object_lock_guard.take();
+        let detach_commit_owner = commit_object_lock_guard.is_some() || upload_guard.is_some();
+        let commit = async move {
+            let _object_lock_guard = commit_object_lock_guard;
             let _upload_guard = upload_guard;
-            if let Err(err) = cleanup_store
-                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &cleanup_upload_id_path, write_quorum)
+            let complete_tail_stage_start = rustfs_io_metrics::put_stage_metrics_enabled().then(Instant::now);
+
+            // Crash-consistency injection: hard power loss after the upload is fully
+            // staged and locked but before the authoritative rename_data commit. No
+            // disk has moved the staged data, so a crash here must leave any prior
+            // committed version byte-for-byte intact (rustfs/backlog#864) and the
+            // upload fully retryable. Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartBeforeCommitRename, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            // The trailing `_` drops the rename_data old-size backfill
+            // (rustfs/backlog#1009): CompleteMultipartUpload keeps its pre-commit
+            // `get_object_info` lookup, so the backfill has no consumer here yet.
+            if let Some(proof) = transaction_fencing_proof.as_ref()
+                && !object_transaction_fencing_fleet_proof_matches(proof)
+            {
+                return Err(Error::other(
+                    "object transaction fencing fleet capability changed during complete_multipart_upload",
+                ));
+            }
+            if let Some(expected) = transaction_epoch_fence {
+                #[cfg(test)]
+                pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeTransactionEpochVerify).await;
+                verify_object_transaction_epoch_fence(&commit_set, &commit_bucket, &commit_object, expected).await?;
+            }
+            let (online_disks, convergence, op_old_dir, cleanup_disks, _) = SetDisks::rename_data(
+                &shuffle_disks,
+                RUSTFS_META_MULTIPART_BUCKET,
+                &commit_upload_id_path,
+                &parts_metadatas,
+                &commit_bucket,
+                &commit_object,
+                write_quorum,
+            )
+            .await?;
+
+            // Detach admission before any post-commit await: client cancellation
+            // must not couple durable convergence repair to cleanup work.
+            if convergence.needs_heal() {
+                let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
+                    commit_bucket.clone(),
+                    Some(commit_object.clone()),
+                    false,
+                    Some(HealChannelPriority::Normal),
+                    Some(commit_set.pool_index),
+                    Some(commit_set.set_index),
+                );
+                request.object_version_id = fi
+                    .version_id
+                    .or_else(|| commit_version_suspended.then(Uuid::nil))
+                    .map(|version_id| version_id.to_string());
+                tokio::spawn(async move {
+                    let _ = rustfs_common::heal_channel::send_heal_request(request).await;
+                });
+            }
+
+            if let Some(old_dir) = op_old_dir {
+                commit_set
+                    .persist_old_data_cleanup_receipts(
+                        &cleanup_disks,
+                        &commit_bucket,
+                        &commit_object,
+                        old_dir,
+                        fi.data_dir,
+                        transaction_epoch,
+                    )
+                    .await;
+            }
+
+            // Crash-consistency injection: hard power loss after the authoritative
+            // rename_data commit succeeded but before the stale part.N.meta cleanup.
+            // The new version is durably committed and visible, so a crash here must
+            // leave the object readable as the new version; the un-reclaimed staging
+            // parts are swept by a retried completion or upload GC (rustfs/backlog#946).
+            // Compiles to a no-op outside `#[cfg(test)]`.
+            if crash_inject::should_crash_at(CrashPoint::MultipartAfterCommitBeforePartsCleanup, &commit_object) {
+                return Err(StorageError::Unexpected);
+            }
+
+            if let Some(committed_slot) = online_disks.iter().position(Option::is_some) {
+                fi = parts_metadatas[committed_slot].clone();
+            }
+            let committed_dir = fi.data_dir.unwrap_or_default().to_string();
+
+            commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
+
+            fi.is_latest = true;
+
+            #[cfg(test)]
+            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::BeforeObjectPublication).await;
+
+            commit_set
+                .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
+                .await;
+
+            drop(_object_lock_guard); // release the object lock before multipart cleanup tail IO.
+
+            #[cfg(test)]
+            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterObjectPublication).await;
+
+            // backlog#946: reclaim the stale per-part metadata (and any superfluous
+            // part.N data files no longer in the completed set) only *after* the
+            // authoritative rename_data commit above has succeeded. If rename_data
+            // fails write quorum and returns via `?`, the upload directory must keep
+            // its part.N.meta so a retried CompleteMultipartUpload can still read the
+            // parts; deleting them before the commit would strand the upload
+            // permanently. This mirrors the "clean up only after commit" pattern
+            // already used for the old data-dir GC and the upload-dir delete_all below.
+            commit_set.cleanup_multipart_path(&parts).await;
+
+            if let Some(old_dir) = op_old_dir {
+                // backlog#898: best-effort reclaim of the dereferenced old data dir.
+                // Returns a receipt (never `Err`); a failed GC must not turn an
+                // already-committed multipart completion into a 503.
+                let cleanup = commit_set
+                    .commit_rename_data_dir(
+                        &cleanup_disks,
+                        &commit_bucket,
+                        &commit_object,
+                        &old_dir.to_string(),
+                        &committed_dir,
+                        write_quorum,
+                    )
+                    .await;
+                commit_set
+                    .report_old_data_dir_cleanup(&commit_bucket, &commit_object, &old_dir.to_string(), &cleanup)
+                    .await;
+            }
+
+            if let Some(stage_start) = complete_tail_stage_start {
+                rustfs_io_metrics::record_put_object_stage_duration(
+                    "multipart_complete_tail",
+                    stage_start.elapsed().as_secs_f64() * 1000.0,
+                );
+            }
+
+            #[cfg(test)]
+            pause_multipart_commit(&commit_bucket, &commit_object, MultipartCommitPause::AfterRename).await;
+
+            if let Err(err) = commit_set
+                .delete_all_with_quorum(RUSTFS_META_MULTIPART_BUCKET, &commit_upload_id_path, write_quorum)
                 .await
             {
                 warn!(
-                    bucket = %cleanup_bucket,
-                    object = %cleanup_object,
-                    upload_id = %cleanup_upload_id,
+                    bucket = %commit_bucket,
+                    object = %commit_object,
+                    upload_id = %commit_upload_id,
                     error = ?err,
                     "completed multipart upload staging cleanup did not reach write quorum"
                 );
             }
-        });
-        if let Err(err) = cleanup_handle.await {
-            warn!(
-                bucket = %bucket,
-                object = %object,
-                upload_id = %upload_id,
-                error = ?err,
-                "completed multipart upload staging cleanup task failed"
-            );
+
+            drop(_upload_guard);
+
+            Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
+        };
+
+        if detach_commit_owner {
+            tokio::spawn(commit)
+                .await
+                .map_err(|err| Error::other(format!("complete_multipart_upload commit task failed: {err}")))?
+        } else {
+            commit.await
         }
-        drop(object_lock_guard); // drop object lock guard to release the lock
-
-        for (i, op_disk) in online_disks.iter().enumerate() {
-            if let Some(disk) = op_disk
-                && disk.is_online().await
-            {
-                fi = parts_metadatas[i].clone();
-                break;
-            }
-        }
-
-        self.record_capacity_scope_if_needed(opts.capacity_scope_token, &online_disks);
-
-        fi.is_latest = true;
-
-        self.invalidate_get_object_metadata_cache(bucket, object).await;
-
-        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
     }
 }
 
@@ -2452,9 +2527,10 @@ fn resolve_complete_etag(opts: &ObjectOptions, uploaded_parts: &[CompletePart])
 mod tests {
     use super::*;
     use crate::config::storageclass::lookup_config_for_pools_without_env;
-    use crate::disk::DiskAPI as _;
+    use crate::disk::{DiskAPI as _, ReadOptions};
     use crate::disk::{endpoint::Endpoint, format::FormatV3};
     use crate::layout::endpoints::SetupType;
+    use crate::services::notification_sys::install_remote_version_state_fleet_proof_for_test;
     // No-locker helpers resolve to the isolated-context variants (see
     // `hermetic_set_disks_isolated`); the guard-based tests build through
     // `hermetic_set_disks_with_lockers`, which stays on the bootstrap context
@@ -2463,6 +2539,7 @@ mod tests {
         hermetic_set_disks_for_pool_with_default_parity_isolated as hermetic_set_disks_for_pool_with_default_parity,
         hermetic_set_disks_isolated as hermetic_set_disks, hermetic_set_disks_with_lockers,
     };
+    use crate::set_disk::ops::object::{PutObjectCommitBarrier, PutObjectCommitPause};
     use crate::storage_api_contracts::namespace::NamespaceLocking as _;
     use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
     use rustfs_config::server_config::KVS;
@@ -2865,6 +2942,208 @@ mod tests {
         }
     }
 
+    async fn object_transaction_epochs(disks: &[DiskStore], bucket: &str, object: &str) -> Vec<Option<Uuid>> {
+        let mut epochs = Vec::with_capacity(disks.len());
+        for (disk_index, disk) in disks.iter().enumerate() {
+            let file_info = disk
+                .read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .unwrap_or_else(|err| panic!("disk {disk_index} should persist object metadata: {err}"));
+            epochs.push(
+                file_info
+                    .object_transaction_epoch()
+                    .unwrap_or_else(|err| panic!("disk {disk_index} transaction epoch should decode: {err}")),
+            );
+        }
+        epochs
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_requires_live_fleet_proof_before_multipart_commit() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-transaction-fencing-no-proof";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) = stage_upload_with_create_opts(
+            &set_disks,
+            bucket,
+            object,
+            b"must-not-complete-without-proof",
+            &ObjectOptions::default(),
+        )
+        .await;
+
+        let err = temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                set_disks
+                    .clone()
+                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
+                    .await
+            },
+        )
+        .await
+        .expect_err("multipart completion must fail closed without a live fleet proof");
+
+        assert!(
+            err.to_string()
+                .contains("object transaction fencing requires a live fleet capability proof"),
+            "unexpected error: {err:?}"
+        );
+        set_disks
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect_err("failed fenced completion must not publish object metadata");
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_persists_epoch_on_multipart_commit() {
+        let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-object-transaction-epoch";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+        let (upload_id, parts) =
+            stage_upload_with_create_opts(&set_disks, bucket, object, b"multipart fenced epoch", &ObjectOptions::default()).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                set_disks
+                    .clone()
+                    .complete_multipart_upload(bucket, object, &upload_id, parts.clone(), &ObjectOptions::default())
+                    .await
+                    .expect("fenced multipart completion should commit with a live proof");
+            },
+        )
+        .await;
+
+        let epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
+        let first = epochs[0].expect("fenced multipart completion should persist an epoch");
+        assert!(!first.is_nil());
+        assert!(epochs.into_iter().all(|epoch| epoch == Some(first)));
+    }
+
+    #[tokio::test]
+    #[serial(storage_class_env)]
+    async fn object_transaction_fencing_rejects_stale_multipart_epoch() {
+        let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "multipart-object-transaction-stale-epoch";
+        let object = "object";
+        make_bucket_on_all(&disk_stores, bucket).await;
+
+        temp_env::async_with_vars(
+            [
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+            ],
+            async {
+                let mut initial_reader = PutObjReader::from_vec(b"initial fenced object".to_vec());
+                set_disks
+                    .put_object(
+                        bucket,
+                        object,
+                        &mut initial_reader,
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("initial fenced PUT should commit");
+                let initial_epoch = object_transaction_epochs(&disk_stores, bucket, object)
+                    .await
+                    .into_iter()
+                    .next()
+                    .flatten()
+                    .expect("initial fenced PUT should persist an epoch");
+
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, b"stale multipart body", &ObjectOptions::default())
+                        .await;
+                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeTransactionEpochVerify);
+                let stale_set = Arc::clone(&set_disks);
+                let stale = tokio::spawn(async move {
+                    stale_set
+                        .clone()
+                        .complete_multipart_upload(
+                            bucket,
+                            object,
+                            &upload_id,
+                            parts,
+                            &ObjectOptions {
+                                no_lock: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                barrier.wait_until_paused().await;
+
+                let mut winner_reader = PutObjReader::from_vec(b"winning put body".to_vec());
+                set_disks
+                    .put_object(
+                        bucket,
+                        object,
+                        &mut winner_reader,
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("concurrent fenced PUT should advance the epoch");
+                let winning_epoch = object_transaction_epochs(&disk_stores, bucket, object)
+                    .await
+                    .into_iter()
+                    .next()
+                    .flatten()
+                    .expect("winning fenced PUT should persist an epoch");
+                assert_ne!(winning_epoch, initial_epoch);
+
+                barrier.release();
+                let err = stale
+                    .await
+                    .expect("stale multipart task should not panic")
+                    .expect_err("stale epoch multipart completion must be rejected");
+                assert_eq!(err, StorageError::PreconditionFailed);
+
+                let final_epochs = object_transaction_epochs(&disk_stores, bucket, object).await;
+                assert!(final_epochs.into_iter().all(|epoch| epoch == Some(winning_epoch)));
+                let mut reader = set_disks
+                    .get_object_reader(
+                        bucket,
+                        object,
+                        None,
+                        HeaderMap::new(),
+                        &ObjectOptions {
+                            no_lock: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("winning object should remain readable");
+                let mut restored = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut restored)
+                    .await
+                    .expect("winning body should stream");
+                assert_eq!(restored, b"winning put body");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     async fn complete_multipart_quota_rejection_preserves_destination_and_upload() {
         let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -4883,6 +5162,250 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn cancelled_complete_keeps_upload_lock_through_tail_cleanup() {
+        temp_env::async_with_vars(
+            [
+                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
+            ],
+            async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+                let bucket = "multipart-cancelled-tail-lock-bucket";
+                let object = "object";
+                make_bucket_on_all(&disk_stores, bucket).await;
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, &[0x53; 4096], &ObjectOptions::default()).await;
+                let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+                signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+                signaling.clear_observed();
+                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+                let barrier = MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterRename);
+
+                let complete_store = set_disks.clone();
+                let complete_upload_id = upload_id.clone();
+                let complete = tokio::spawn(async move {
+                    complete_store
+                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                        .await
+                });
+                barrier.wait_until_paused().await;
+
+                let abort_store = set_disks.clone();
+                let abort_upload_id = upload_id.clone();
+                let abort = tokio::spawn(async move {
+                    abort_store
+                        .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                        .await
+                });
+                signaling.wait_for_attempts(2).await;
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "abort must wait while completion tail owns the upload lock");
+
+                complete.abort();
+                assert!(
+                    complete
+                        .await
+                        .expect_err("the completion request should be cancellable while the tail is paused")
+                        .is_cancelled()
+                );
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "cancelling the completion waiter must not release the upload lock");
+
+                barrier.release();
+                let abort_err = abort
+                    .await
+                    .expect("abort task should not panic")
+                    .expect_err("the committed upload should no longer exist when abort acquires the lock");
+                assert!(
+                    matches!(abort_err, StorageError::InvalidUploadID(..)),
+                    "abort should return InvalidUploadID after the detached completion tail, got {abort_err:?}"
+                );
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn complete_releases_object_lock_before_cleanup_and_keeps_upload_lock() {
+        temp_env::async_with_vars(
+            [
+                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
+            ],
+            async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+                let bucket = "multipart-object-lock-short-tail-bucket";
+                let object = "object";
+                make_bucket_on_all(&disk_stores, bucket).await;
+                let completed_body = vec![0x63; 4096];
+                let replacement_body = vec![0x64; 4096];
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, &completed_body, &ObjectOptions::default()).await;
+                let upload_id_path = SetDisks::get_upload_id_dir(bucket, object, &upload_id);
+                signaling.set_target(rustfs_lock::ObjectKey::new(RUSTFS_META_MULTIPART_BUCKET, upload_id_path));
+                signaling.clear_observed();
+                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+                let completion_barrier =
+                    MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::AfterObjectPublication);
+
+                let complete_store = set_disks.clone();
+                let complete_upload_id = upload_id.clone();
+                let complete = tokio::spawn(async move {
+                    complete_store
+                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                        .await
+                });
+                completion_barrier.wait_until_paused().await;
+
+                let mut reader = tokio::time::timeout(
+                    Duration::from_secs(10),
+                    set_disks.get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()),
+                )
+                .await
+                .expect("GET should not wait for multipart cleanup after object publication")
+                .expect("completed object should be readable while the upload tail is paused");
+                let mut observed_body = Vec::new();
+                tokio::time::timeout(Duration::from_secs(10), reader.stream.read_to_end(&mut observed_body))
+                    .await
+                    .expect("completed object body should stream while the upload tail is paused")
+                    .expect("completed object body should read successfully");
+                assert_eq!(observed_body, completed_body);
+
+                let put_barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterNamespace);
+                let put_store = set_disks.clone();
+                let put_payload = replacement_body.clone();
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(put_payload);
+                    put_store
+                        .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                        .await
+                });
+                put_barrier.wait_until_paused().await;
+
+                let abort_store = set_disks.clone();
+                let abort_upload_id = upload_id.clone();
+                let abort = tokio::spawn(async move {
+                    abort_store
+                        .abort_multipart_upload(bucket, object, &abort_upload_id, &ObjectOptions::default())
+                        .await
+                });
+                signaling.wait_for_attempts(2).await;
+                tokio::task::yield_now().await;
+                assert!(
+                    !abort.is_finished(),
+                    "abort must still wait while the completion tail owns the upload lock"
+                );
+
+                complete.abort();
+                assert!(
+                    complete
+                        .await
+                        .expect_err("the completion waiter should remain cancellable after object publication")
+                        .is_cancelled()
+                );
+                tokio::task::yield_now().await;
+                assert!(!abort.is_finished(), "cancelling the waiter must not release the upload lock");
+
+                completion_barrier.release();
+                let abort_err = abort
+                    .await
+                    .expect("abort task should not panic")
+                    .expect_err("the committed upload should no longer exist after the detached tail drains");
+                assert!(
+                    matches!(abort_err, StorageError::InvalidUploadID(..)),
+                    "abort should return InvalidUploadID after the completion tail, got {abort_err:?}"
+                );
+
+                put_barrier.release();
+                put.await
+                    .expect("same-key PUT task should not panic")
+                    .expect("same-key PUT should commit after the object lock is released early");
+
+                let mut reader = set_disks
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("final object should be readable");
+                let mut final_body = Vec::new();
+                reader
+                    .stream
+                    .read_to_end(&mut final_body)
+                    .await
+                    .expect("final object should stream fully");
+                assert_eq!(final_body, replacement_body);
+            },
+        )
+        .await;
+    }
+
+    #[tokio::test(flavor = "multi_thread")]
+    #[serial]
+    async fn complete_keeps_object_lock_until_publication_fence() {
+        temp_env::async_with_vars(
+            [
+                (crate::object_api::ENV_RUSTFS_ENCRYPTED_RANGE_SEEK, Some("true")),
+                (rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("60")),
+            ],
+            async {
+                let manager = Arc::new(rustfs_lock::GlobalLockManager::new());
+                let signaling = Arc::new(SignalingLockClient::new(Arc::new(LocalClient::with_manager(manager))));
+                let lockers: Vec<Arc<dyn LockClient>> = vec![signaling.clone()];
+                let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks_with_lockers(4, 0, 2, lockers).await;
+                let bucket = "multipart-publication-fence-lock-bucket";
+                let object = "object";
+                make_bucket_on_all(&disk_stores, bucket).await;
+                let (upload_id, parts) =
+                    stage_upload_with_create_opts(&set_disks, bucket, object, &[0x65; 4096], &ObjectOptions::default()).await;
+                let _setup_type_guard = SetupTypeGuard::switch_to(SetupType::DistErasure).await;
+                let completion_barrier =
+                    MultipartCommitBarrier::install(bucket, object, MultipartCommitPause::BeforeObjectPublication);
+
+                let complete_store = set_disks.clone();
+                let complete_upload_id = upload_id.clone();
+                let complete = tokio::spawn(async move {
+                    complete_store
+                        .complete_multipart_upload(bucket, object, &complete_upload_id, parts, &ObjectOptions::default())
+                        .await
+                });
+                completion_barrier.wait_until_paused().await;
+
+                let before_namespace_barrier =
+                    PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
+                let after_namespace_barrier =
+                    PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::AfterNamespace);
+                let put_store = set_disks.clone();
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![0x66; 4096]);
+                    put_store
+                        .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                        .await
+                });
+                before_namespace_barrier.wait_until_paused().await;
+                before_namespace_barrier.release_and_wait_until_namespace_pending().await;
+
+                completion_barrier.release();
+                after_namespace_barrier.wait_until_paused().await;
+                after_namespace_barrier.release();
+                complete
+                    .await
+                    .expect("completion task should not panic")
+                    .expect("completion should commit after publication fence");
+                put.await
+                    .expect("same-key PUT task should not panic")
+                    .expect("same-key PUT should commit after completion publishes and releases the object lock");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test(flavor = "multi_thread")]
     #[serial]
     async fn complete_validates_parts_after_an_inflight_upload_part_commit() {
@@ -6108,6 +6631,24 @@ mod tests {
             (body, etag)
         }
 
+        async fn current_data_dir(disk: &DiskStore, bucket: &str, object: &str) -> Uuid {
+            disk.read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("current object metadata should read")
+                .data_dir
+                .expect("test object should be stored out-of-line")
+        }
+
+        async fn data_dir_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool {
+            disk.read_all(bucket, &format!("{object}/{data_dir}/part.1")).await.is_ok()
+        }
+
+        async fn cleanup_receipt_exists(disk: &DiskStore, bucket: &str, object: &str, data_dir: Uuid) -> bool {
+            disk.read_all(bucket, &old_data_cleanup_receipt_path(object, data_dir))
+                .await
+                .is_ok()
+        }
+
         async fn upload_is_listed(set_disks: &Arc<SetDisks>, bucket: &str, object: &str, upload_id: &str) -> bool {
             let page = set_disks
                 .list_multipart_uploads_for_incarnation(bucket, object, None, None, None, 1000, None)
@@ -6228,6 +6769,106 @@ mod tests {
             let (body_after, _) = read_object(&set_disks, bucket, object).await;
             assert_eq!(body_after, new, "reclaiming the leftover upload must not disturb the committed object");
         }
+
+        #[tokio::test]
+        #[serial(storage_class_env)]
+        async fn post_commit_crash_receipt_reclaims_old_data_after_restart() {
+            let _proof = install_remote_version_state_fleet_proof_for_test("object-transaction-fencing-test");
+            let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+            let bucket = "multipart-crash-old-data-receipt";
+            let object = "crash-old-data-object";
+            make_bucket_on_all(&disk_stores, bucket).await;
+
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_WRITE, Some("true")),
+                    (rustfs_config::ENV_OBJECT_TRANSACTION_FENCING_FLEET_CONFIRMED, Some("true")),
+                ],
+                async {
+                    let old = payload(0x51);
+                    let (u_old, parts_old) = stage_upload(&set_disks, bucket, object, &old).await;
+                    complete(&set_disks, bucket, object, &u_old, parts_old)
+                        .await
+                        .expect("the old version should commit");
+                    let old_dir = current_data_dir(&disk_stores[0], bucket, object).await;
+
+                    let new = payload(0x52);
+                    let (u_new, parts_new) = stage_upload(&set_disks, bucket, object, &new).await;
+                    crash_inject::arm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
+                    let crashed = complete(&set_disks, bucket, object, &u_new, parts_new).await;
+                    assert!(
+                        matches!(crashed, Err(StorageError::Unexpected)),
+                        "the post-commit crash point must surface as unexpected, got {crashed:?}"
+                    );
+                    crash_inject::disarm(CrashPoint::MultipartAfterCommitBeforePartsCleanup, object);
+
+                    let (body, _) = read_object(&set_disks, bucket, object).await;
+                    assert_eq!(body, new, "the committed replacement must remain readable after the crash");
+                    for disk in &disk_stores {
+                        assert!(
+                            cleanup_receipt_exists(disk, bucket, object, old_dir).await,
+                            "post-commit crash must leave a durable old-data cleanup receipt"
+                        );
+                        assert!(
+                            data_dir_exists(disk, bucket, object, old_dir).await,
+                            "post-commit crash must leave old data for restart reconciliation"
+                        );
+                    }
+
+                    let restarted_endpoints = temp_dirs
+                        .iter()
+                        .enumerate()
+                        .map(|(disk_idx, dir)| {
+                            let mut endpoint = Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8"))
+                                .expect("endpoint should parse");
+                            endpoint.set_pool_index(0);
+                            endpoint.set_set_index(0);
+                            endpoint.set_disk_index(disk_idx);
+                            endpoint
+                        })
+                        .collect::<Vec<_>>();
+                    let mut reloaded = Vec::with_capacity(restarted_endpoints.len());
+                    for endpoint in &restarted_endpoints {
+                        reloaded.push(
+                            new_disk(
+                                endpoint,
+                                &DiskOption {
+                                    cleanup: false,
+                                    health_check: false,
+                                },
+                            )
+                            .await
+                            .expect("disk should restart"),
+                        );
+                    }
+                    let restarted_set = SetDisks::new_with_instance_ctx(
+                        "restart-cleanup-receipt-test-owner".to_string(),
+                        Arc::new(RwLock::new(reloaded.iter().cloned().map(Some).collect())),
+                        4,
+                        2,
+                        0,
+                        0,
+                        restarted_endpoints,
+                        set_disks.format.clone(),
+                        Vec::new(),
+                        Arc::new(crate::runtime::instance::InstanceContext::new()),
+                    )
+                    .await;
+                    let removed = restarted_set
+                        .reconcile_old_data_cleanup_receipts(bucket, object)
+                        .await
+                        .expect("restart receipt reconciliation should succeed");
+                    assert_eq!(removed, 4, "restart reconciler should delete all receipt targets");
+                    for disk in &reloaded {
+                        assert!(
+                            !data_dir_exists(disk, bucket, object, old_dir).await,
+                            "restart reconciler must reclaim the old data dir"
+                        );
+                    }
+                },
+            )
+            .await;
+        }
     }
 
     #[test]

crates/ecstore/src/set_disk/read.rs

@@ -482,6 +482,7 @@ impl SetDisks {
     pub(super) async fn try_get_object_direct_data_shards_with_fileinfo(
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         fi: &FileInfo,
         files: &[FileInfo],
         disks: &[Option<DiskStore>],
@@ -502,13 +503,7 @@ impl SetDisks {
             return Ok(None);
         }
 
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
+        let erasure = erasure_cache.get_for_file_info(fi)?;
 
         let checksum_info = fi.erasure.get_checksum_info(part.number);
         let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -636,6 +631,7 @@ impl SetDisks {
         // &self,
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         offset: usize,
         length: i64,
         writer: &mut W,
@@ -730,13 +726,7 @@ impl SetDisks {
             object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
         );
 
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
+        let erasure = erasure_cache.get_for_file_info(&fi)?;
 
         let part_indices: Vec<usize> = (part_index..=last_part_index).collect();
         debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
@@ -1170,6 +1160,7 @@ impl SetDisks {
     pub(super) async fn get_object_decode_reader_with_fileinfo(
         bucket: &str,
         object: &str,
+        erasure_cache: Arc<ErasureCache>,
         fi: &FileInfo,
         files: &[FileInfo],
         disks: &[Option<DiskStore>],
@@ -1180,14 +1171,7 @@ impl SetDisks {
         metrics_size_bucket: &'static str,
         prefer_data_blocks_first_reader_setup: bool,
     ) -> Result<GetCodecStreamingReaderBuildOutcome> {
-        let erasure = coding::Erasure::try_new_with_options(
-            fi.erasure.data_blocks,
-            fi.erasure.parity_blocks,
-            fi.erasure.block_size,
-            fi.uses_legacy_checksum,
-        )
-        .map_err(Error::from)?;
-
+        let erasure = erasure_cache.get_for_file_info(fi)?;
         let (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
 
         if fi.parts.len() == 1 {
@@ -1574,7 +1558,7 @@ struct LazyCodecPartContext {
     fi: FileInfo,
     files: Vec<FileInfo>,
     disks: Vec<Option<DiskStore>>,
-    erasure: coding::Erasure,
+    erasure: Arc<coding::Erasure>,
     skip_verify_bitrot: bool,
     metrics_object_class: &'static str,
     metrics_size_bucket: &'static str,
@@ -2058,6 +2042,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2088,6 +2073,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             2,
             1,
             &mut output,
@@ -2111,6 +2097,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             usize::MAX,
             1,
             &mut output,
@@ -2132,6 +2119,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             1,
             1,
             &mut output,
@@ -2155,6 +2143,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2192,6 +2181,7 @@ mod metadata_cache_tests {
         SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             0,
             &mut output,
@@ -2224,6 +2214,7 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
+            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -4128,6 +4119,7 @@ mod tests {
         let result = SetDisks::get_object_decode_reader_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             &fi,
             &[],
             &[],
@@ -4150,6 +4142,7 @@ mod tests {
         let invalid_size = SetDisks::get_object_decode_reader_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             &single_part,
             &[],
             &[],
@@ -4170,6 +4163,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4194,6 +4188,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4222,6 +4217,7 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
+                    Arc::new(ErasureCache::new()),
                     &multipart,
                     &[],
                     &[],
@@ -4275,6 +4271,7 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
+                    Arc::new(ErasureCache::new()),
                     &fi,
                     &files,
                     &disks,
@@ -4328,6 +4325,7 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
+                Arc::new(ErasureCache::new()),
                 &fi,
                 &files,
                 &disks,
@@ -4372,6 +4370,7 @@ mod tests {
         SetDisks::get_object_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
+            Arc::new(ErasureCache::new()),
             0,
             part_data.len() as i64,
             &mut output,

crates/ecstore/src/set_disk/replication.rs

@@ -48,6 +48,23 @@ impl RestoreCleanupIdentity {
     }
 }
 
+fn ensure_restore_metadata_lock_held(bucket: &str, object: &str, opts: &ObjectOptions, mode: &'static str) -> Result<()> {
+    if opts
+        .namespace_lock_fence
+        .as_ref()
+        .is_some_and(NamespaceLockFence::is_lock_lost)
+    {
+        return Err(StorageError::NamespaceLockQuorumUnavailable {
+            mode,
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            required: 1,
+            achieved: 0,
+        });
+    }
+    Ok(())
+}
+
 impl SetDisks {
     pub(super) async fn finalize_restore_metadata(
         &self,
@@ -88,6 +105,7 @@ impl SetDisks {
         if !expected.matches_file_info(&fi, &expected_etag) {
             return Err(Error::other("restored object changed before restore metadata finalization"));
         }
+        ensure_restore_metadata_lock_held(bucket, object, opts, "restore_finalize_metadata")?;
         let restore_expiry =
             lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1));
         fi.metadata.insert(
@@ -159,6 +177,7 @@ impl SetDisks {
         if !expected.matches_file_info(&fi, &expected_etag) {
             return Ok(());
         }
+        ensure_restore_metadata_lock_held(bucket, object, opts, "restore_cleanup_metadata")?;
         fi.metadata.remove(X_AMZ_RESTORE.as_str());
         fi.metadata.remove(AMZ_RESTORE_EXPIRY_DAYS);
         fi.metadata.remove(AMZ_RESTORE_REQUEST_DATE);

crates/ecstore/src/store/init.rs

@@ -13,7 +13,7 @@
 // limitations under the License.
 
 use super::*;
-use crate::core::pools::local_decommission_queue_prefix;
+use crate::core::pools::{local_decommission_queue_prefix, pool_meta_has_active_decommission};
 use crate::error::is_err_decommission_running;
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -109,14 +109,6 @@ fn should_auto_start_rebalance_after_init(decommission_running: bool, rebalance_
     rebalance_meta_loaded && !decommission_running
 }
 
-fn pool_meta_has_active_decommission(meta: &PoolMeta) -> bool {
-    meta.pools.iter().any(|pool| {
-        pool.decommission
-            .as_ref()
-            .is_some_and(|info| info.has_decommission_state() && !info.complete && !info.failed && !info.canceled)
-    })
-}
-
 async fn wait_for_local_decommission_resume_delay(rx: &CancellationToken, delay: Duration) -> bool {
     tokio::select! {
         _ = rx.cancelled() => false,

crates/filemeta/src/fileinfo.rs

@@ -18,8 +18,9 @@ use rmp_serde::Serializer;
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::http::{
     AMZ_OBJECT_TAGGING, SUFFIX_COMPRESSION, SUFFIX_DATA_MOVED, SUFFIX_DATA_MOVED_TAGS, SUFFIX_FREE_VERSION, SUFFIX_HEALING,
-    SUFFIX_INLINE_DATA, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID, contains_key_str, get_str,
-    has_internal_suffix, insert_str, is_encryption_metadata_key, starts_with_ignore_ascii_case,
+    SUFFIX_INLINE_DATA, SUFFIX_OBJECT_TRANSACTION_EPOCH, SUFFIX_TIER_FV_ID, SUFFIX_TIER_FV_MARKER, SUFFIX_TIER_SKIP_FV_ID,
+    contains_key_str, get_consistent_str, get_str, has_internal_suffix, insert_str, is_encryption_metadata_key,
+    starts_with_ignore_ascii_case,
 };
 use s3s::dto::{RestoreStatus, Timestamp};
 use s3s::header::X_AMZ_RESTORE;
@@ -1172,6 +1173,22 @@ impl FileInfo {
         insert_str(&mut self.metadata, SUFFIX_DATA_MOVED, String::new());
     }
 
+    pub fn set_object_transaction_epoch(&mut self, epoch: Uuid) {
+        insert_str(&mut self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH, epoch.to_string());
+    }
+
+    pub fn object_transaction_epoch(&self) -> Result<Option<Uuid>> {
+        if !contains_key_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH) {
+            return Ok(None);
+        }
+        let value = get_consistent_str(&self.metadata, SUFFIX_OBJECT_TRANSACTION_EPOCH).ok_or(Error::FileCorrupt)?;
+        let epoch = Uuid::parse_str(value).map_err(|_| Error::FileCorrupt)?;
+        if epoch.is_nil() {
+            return Err(Error::FileCorrupt);
+        }
+        Ok(Some(epoch))
+    }
+
     pub fn inline_data(&self) -> bool {
         contains_key_str(&self.metadata, SUFFIX_INLINE_DATA) && !self.is_remote()
     }
@@ -1484,6 +1501,46 @@ mod tests {
         assert_eq!(ei.get_checksum_info(99).algorithm, HashAlgorithm::HighwayHash256S);
     }
 
+    #[test]
+    fn object_transaction_epoch_uses_consistent_dual_internal_metadata() {
+        let mut fi = validation_test_fileinfo();
+        assert_eq!(fi.object_transaction_epoch().expect("absent epoch should decode"), None);
+
+        let epoch = Uuid::new_v4();
+        let epoch_text = epoch.to_string();
+        fi.set_object_transaction_epoch(epoch);
+        assert_eq!(fi.object_transaction_epoch().expect("written epoch should decode"), Some(epoch));
+        assert_eq!(fi.metadata.get("x-rustfs-internal-object-transaction-epoch"), Some(&epoch_text));
+        assert_eq!(fi.metadata.get("x-minio-internal-object-transaction-epoch"), Some(&epoch_text));
+
+        let mut rustfs_only = validation_test_fileinfo();
+        rustfs_only
+            .metadata
+            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), epoch_text);
+        assert_eq!(
+            rustfs_only
+                .object_transaction_epoch()
+                .expect("single compatibility key should decode"),
+            Some(epoch)
+        );
+
+        let mut conflicting = fi.clone();
+        conflicting
+            .metadata
+            .insert("x-minio-internal-object-transaction-epoch".to_string(), Uuid::new_v4().to_string());
+        assert_eq!(conflicting.object_transaction_epoch(), Err(Error::FileCorrupt));
+
+        let mut malformed = validation_test_fileinfo();
+        malformed
+            .metadata
+            .insert("x-rustfs-internal-object-transaction-epoch".to_string(), "not-a-uuid".to_string());
+        assert_eq!(malformed.object_transaction_epoch(), Err(Error::FileCorrupt));
+
+        let mut nil = validation_test_fileinfo();
+        nil.set_object_transaction_epoch(Uuid::nil());
+        assert_eq!(nil.object_transaction_epoch(), Err(Error::FileCorrupt));
+    }
+
     // backlog#949: distribution range/permutation validation.
     #[test]
     fn is_valid_distribution_accepts_permutation() {

crates/heal/src/error.rs

@@ -34,36 +34,21 @@ pub enum Error {
     #[error("Configuration error: {0}")]
     Config(String),
 
-    #[error("Heal configuration error: {message}")]
-    ConfigurationError { message: String },
-
     #[error("Other error: {0}")]
     Other(String),
 
     #[error("Serialization error: {0}")]
     Serialization(String),
 
-    #[error("IO error: {0}")]
-    IO(String),
-
-    #[error("Not found: {0}")]
-    NotFound(String),
-
     #[error("Invalid checkpoint: {0}")]
     InvalidCheckpoint(String),
 
     #[error("Heal task not found: {task_id}")]
     TaskNotFound { task_id: String },
 
-    #[error("Heal task already exists: {task_id}")]
-    TaskAlreadyExists { task_id: String },
-
     #[error("Invalid heal client token")]
     InvalidClientToken,
 
-    #[error("Heal manager is not running")]
-    ManagerNotRunning,
-
     #[error("Heal task execution failed: {message}")]
     TaskExecutionFailed { message: String },
 
@@ -78,12 +63,6 @@ pub enum Error {
 
     #[error("Heal task timeout")]
     TaskTimeout,
-
-    #[error("Heal event processing failed: {message}")]
-    EventProcessingFailed { message: String },
-
-    #[error("Heal progress tracking failed: {message}")]
-    ProgressTrackingFailed { message: String },
 }
 
 /// A specialized Result type for heal operations
@@ -129,9 +108,7 @@ impl Error {
                         | DiskError::FaultyDisk
                 ) || is_recoverable_heal_error_message(&err.to_string())
             }
-            Error::TaskExecutionFailed { message } | Error::IO(message) | Error::Other(message) => {
-                is_recoverable_heal_error_message(message)
-            }
+            Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message),
             Error::Io(err) => is_recoverable_heal_error_message(&err.to_string()),
             _ => false,
         }

crates/heal/src/heal/task.rs

@@ -597,7 +597,7 @@ impl HealTask {
                 | EcstoreError::ObjectNotFound(_, _)
                 | EcstoreError::VersionNotFound(_, _, _),
             ) => true,
-            Error::Other(message) | Error::IO(message) => {
+            Error::Other(message) => {
                 message.contains("File not found")
                     || message.contains("file not found")
                     || message.contains("File version not found")

crates/heal/tests/endpoint_index_test.rs

@@ -14,6 +14,8 @@
 
 //! test endpoint index settings
 
+#![recursion_limit = "256"]
+
 use std::net::SocketAddr;
 use tempfile::TempDir;
 use tokio_util::sync::CancellationToken;

crates/heal/tests/heal_b5_versioned_regression_test.rs

@@ -22,6 +22,8 @@
 //! bucket-metadata-sys OnceCell) — under `cargo nextest` each test runs
 //! in its own process so the OnceCell never collides.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/heal/tests/heal_b920_subquorum_union_test.rs

@@ -21,6 +21,8 @@
 //! These drive the REAL `ECStoreHealStorage` + `ECStore` against real disks.
 //! Every test is `#[serial]`; under `cargo nextest` each runs in its own process.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::storage::{

crates/heal/tests/heal_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use http::HeaderMap;
 use rustfs_common::heal_channel::{HealOpts, HealScanMode};
 use rustfs_heal::heal::{

crates/io-metrics/src/lib.rs

@@ -812,6 +812,17 @@ pub fn record_get_object_metadata_fanout_shape(path: &'static str, total: usize,
         .record(metadata_fanout_count_to_f64(non_valid));
 }
 
+/// Record task lifecycle shape for one GetObject metadata fanout.
+#[inline(always)]
+pub fn record_get_object_metadata_fanout_lifecycle(path: &'static str, scheduled: usize, completed: usize, cancelled: usize) {
+    if !get_stage_metrics_enabled() {
+        return;
+    }
+    histogram!("rustfs_io_get_object_metadata_fanout_scheduled", "path" => path).record(metadata_fanout_count_to_f64(scheduled));
+    histogram!("rustfs_io_get_object_metadata_fanout_completed", "path" => path).record(metadata_fanout_count_to_f64(completed));
+    histogram!("rustfs_io_get_object_metadata_fanout_cancelled", "path" => path).record(metadata_fanout_count_to_f64(cancelled));
+}
+
 /// Record a guarded metadata early-stop hit for GetObject.
 #[inline(always)]
 pub fn record_get_object_metadata_early_stop_hit(path: &'static str, reason: &'static str) {
@@ -2692,12 +2703,17 @@ mod tests {
         record_get_object_reader_prefetch_wait("codec_streaming", 0.0002);
         record_get_object_response_handoff("standard", "selected", 8192, 1024, 0.0001);
         record_get_object_metadata_fanout_duration("legacy_duplex", 0.001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_resolve", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_path_check", 0.0001);
+        record_get_object_stage_duration("legacy_duplex", "read_version_xlmeta_read", 0.0005);
+        record_get_object_stage_duration("legacy_duplex", "read_version_decode", 0.0002);
         record_get_object_first_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_first_valid_metadata_response_latency("legacy_duplex", 0.001);
         record_get_object_slowest_metadata_response_latency("legacy_duplex", 0.003);
         record_get_object_quorum_reached_latency("legacy_duplex", 0.002);
         record_get_object_metadata_response("legacy_duplex", "valid");
         record_get_object_metadata_fanout_shape("legacy_duplex", 4, 3, 1, 1);
+        record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
         record_get_object_metadata_early_stop_hit("legacy_duplex", "valid_quorum");
         record_get_object_metadata_early_stop_miss("legacy_duplex", "insufficient_quorum");
         record_get_object_metadata_early_stop_saved_responses("legacy_duplex", 1);
@@ -2768,6 +2784,38 @@ mod tests {
         assert!(remote_scheduled >= remote_avoid_potential);
     }
 
+    #[test]
+    fn metadata_fanout_lifecycle_records_named_histograms() {
+        let _guard = METRICS_FLAG_LOCK.lock().unwrap_or_else(|e| e.into_inner());
+        let recorder = DebuggingRecorder::new();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, || {
+            set_get_stage_metrics_enabled(true);
+            record_get_object_metadata_fanout_lifecycle("legacy_duplex", 4, 3, 1);
+            set_get_stage_metrics_enabled(false);
+        });
+
+        let metrics = snapshotter.snapshot().into_vec();
+        for (name, expected) in [
+            ("rustfs_io_get_object_metadata_fanout_scheduled", 4.0),
+            ("rustfs_io_get_object_metadata_fanout_completed", 3.0),
+            ("rustfs_io_get_object_metadata_fanout_cancelled", 1.0),
+        ] {
+            let value = metrics.iter().find_map(|(composite, _, _, value)| {
+                let has_path = composite
+                    .key()
+                    .labels()
+                    .any(|label| label.key() == "path" && label.value() == "legacy_duplex");
+                (composite.kind() == MetricKind::Histogram && composite.key().name() == name && has_path).then_some(value)
+            });
+            assert!(
+                matches!(value, Some(DebugValue::Histogram(values)) if values.len() == 1 && values[0].0 == expected),
+                "{name} must record the exact fanout lifecycle sample"
+            );
+        }
+    }
+
     #[test]
     fn test_record_get_object_fill_metrics() {
         record_get_object_fill_queued("codec_streaming", "single_inflight", 1);

crates/kms/src/config.rs

@@ -41,6 +41,7 @@ pub const ENV_KMS_AWS_ENDPOINT_URL: &str = "RUSTFS_KMS_AWS_ENDPOINT_URL";
 /// unset leaves rotation readiness unreported. Read once when the manager is
 /// built, by [`crate::manager::KmsManager`].
 pub const ENV_KMS_ROTATION_MAX_AGE_SECS: &str = "RUSTFS_KMS_ROTATION_MAX_AGE_SECS";
+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";

crates/kms/src/manager.rs

@@ -17,7 +17,7 @@
 use crate::audit::{KmsAuditOperation, KmsAuditRecord, KmsAuditSink};
 use crate::backends::KmsBackend;
 use crate::cache::{KmsCache, KmsCacheStats};
-use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, KmsConfig};
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, ENV_KMS_ROTATION_MAX_WRAPS, KmsConfig};
 use crate::deletion_worker::DeletionReferenceChecker;
 use crate::error::{KmsError, Result};
 use crate::types::{
@@ -42,6 +42,13 @@ use tracing::warn;
 /// after it was rotated, which trains operators to ignore the signal.
 const MIN_ROTATION_MAX_AGE: Duration = Duration::from_secs(3600);
 
+/// Smallest wrap budget that can be configured.
+///
+/// Wraps are accounted in reserved blocks, so any threshold below one block
+/// would be crossed by a single reservation and report a key that has barely
+/// wrapped anything as overdue.
+const MIN_ROTATION_MAX_WRAPS: u64 = 1_000_000;
+
 /// Rotation age from the environment, or `None` when the signal is off.
 ///
 /// Unset leaves it off rather than guessing a policy: how often a deployment
@@ -68,6 +75,33 @@ fn parse_rotation_max_age(value: Option<&str>) -> Option<Duration> {
     Some(Duration::from_secs(seconds).max(MIN_ROTATION_MAX_AGE))
 }
 
+/// Wrap budget from the environment, or `None` when the signal is off.
+///
+/// Same discipline as the age threshold: unset means unreported rather than a
+/// guessed policy, and an unparsable value is refused loudly instead of
+/// falling back to a number the operator did not write. Clamped to
+/// [`MIN_ROTATION_MAX_WRAPS`] because the backend accounts for wraps in
+/// reserved blocks, so a threshold below one block would trip on the first
+/// reservation regardless of how many wraps actually happened.
+fn configured_rotation_max_wraps() -> Option<u64> {
+    parse_rotation_max_wraps(std::env::var(ENV_KMS_ROTATION_MAX_WRAPS).ok().as_deref())
+}
+
+fn parse_rotation_max_wraps(value: Option<&str>) -> Option<u64> {
+    let value = value?;
+    let Ok(wraps) = value.trim().parse::<u64>() else {
+        warn!(
+            variable = ENV_KMS_ROTATION_MAX_WRAPS,
+            "ignoring unparsable KMS rotation wrap budget; rotation readiness stays unreported"
+        );
+        return None;
+    };
+    if wraps == 0 {
+        return None;
+    }
+    Some(wraps.max(MIN_ROTATION_MAX_WRAPS))
+}
+
 #[derive(Clone)]
 pub struct KmsManager {
     backend: Arc<dyn KmsBackend>,
@@ -82,6 +116,7 @@ pub struct KmsManager {
     /// the verdict unreported. Read once at construction so a listing cannot
     /// change its answer halfway through.
     rotation_max_age: Option<Duration>,
+    rotation_max_wraps: Option<u64>,
 }
 
 impl KmsManager {
@@ -103,6 +138,7 @@ impl KmsManager {
             allow_immediate_deletion: config.allow_immediate_deletion,
             reference_checker: None,
             rotation_max_age: configured_rotation_max_age(),
+            rotation_max_wraps: configured_rotation_max_wraps(),
         }
     }
 
@@ -314,9 +350,22 @@ impl KmsManager {
             key.rotation_due_reason = Some(RotationDueReason::Unsupported);
             return;
         }
+        key.rotation_due = false;
+        key.rotation_due_reason = None;
+
+        // The wrap budget is checked first: it is the cryptographic bound (the
+        // AES-GCM random-nonce ceiling), whereas the age threshold is a policy
+        // choice, so when both are crossed the reason an operator most needs to
+        // see is the one they cannot negotiate.
+        if let (Some(max_wraps), Some(wraps)) = (self.rotation_max_wraps, key.wrap_budget_reserved)
+            && wraps >= max_wraps
+        {
+            key.rotation_due = true;
+            key.rotation_due_reason = Some(RotationDueReason::Wraps);
+            return;
+        }
+
         let Some(max_age) = self.rotation_max_age else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
             return;
         };
 
@@ -333,9 +382,6 @@ impl KmsManager {
         if age >= max_age {
             key.rotation_due = true;
             key.rotation_due_reason = Some(reason);
-        } else {
-            key.rotation_due = false;
-            key.rotation_due_reason = None;
         }
     }
 
@@ -1685,10 +1731,15 @@ mod tests {
     }
 
     fn readiness_manager(rotation_max_age: Option<Duration>) -> KmsManager {
+        readiness_manager_with(rotation_max_age, None)
+    }
+
+    fn readiness_manager_with(rotation_max_age: Option<Duration>, rotation_max_wraps: Option<u64>) -> KmsManager {
         let temp_dir = tempfile::tempdir().expect("temp dir");
         let config = KmsConfig::local(temp_dir.path().to_path_buf()).with_insecure_development_defaults();
         let mut manager = KmsManager::new(Arc::new(ScriptedBackend::succeeding()), config);
         manager.rotation_max_age = rotation_max_age;
+        manager.rotation_max_wraps = rotation_max_wraps;
         manager
     }
 
@@ -1765,6 +1816,85 @@ mod tests {
         assert!(!key.rotation_due, "clock skew must not manufacture an overdue key");
     }
 
+    /// The wrap-budget half of the verdict: the cryptographic bound, checked
+    /// independently of the age policy and reported under its own reason.
+    #[test]
+    fn rotation_readiness_reports_an_exhausted_wrap_budget() {
+        let now = Zoned::now();
+        let recently = &now - jiff::Span::new().hours(1);
+        let long_ago = &now - jiff::Span::new().days(400);
+        let day = Duration::from_secs(86_400);
+        let budget = 2_000_000;
+
+        let with_wraps = |manager: &KmsManager, wraps: Option<u64>, rotated_at: Option<Zoned>| {
+            let mut key = aged_key(rotated_at, recently.clone());
+            key.wrap_budget_reserved = wraps;
+            manager.apply_rotation_readiness(&mut key, true, &now);
+            (key.rotation_due, key.rotation_due_reason)
+        };
+
+        // Budget configured and exceeded on a freshly rotated key: due, and the
+        // reason names the wrap budget rather than an age nobody crossed.
+        let manager = readiness_manager_with(Some(day), Some(budget));
+        assert_eq!(
+            with_wraps(&manager, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // At the threshold exactly, not only past it: the bound is a ceiling.
+        assert_eq!(
+            with_wraps(&manager, Some(budget + 1), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+        // Under the threshold: no verdict from the wrap half.
+        assert_eq!(with_wraps(&manager, Some(budget - 1), Some(recently.clone())), (false, None));
+
+        // The cryptographic bound outranks the policy one when both are crossed.
+        let mut key = aged_key(Some(long_ago.clone()), long_ago);
+        key.wrap_budget_reserved = Some(budget);
+        manager.apply_rotation_readiness(&mut key, true, &now);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Wraps));
+
+        // No wrap threshold configured: an enormous count reports nothing, the
+        // same way an unset age threshold does.
+        let age_only = readiness_manager_with(Some(day), None);
+        assert_eq!(with_wraps(&age_only, Some(u64::MAX), Some(recently.clone())), (false, None));
+
+        // Backend reports no count (Transit, AWS, or a pre-accounting record):
+        // the wrap half stays silent instead of guessing, and the age half
+        // still decides.
+        let wraps_only = readiness_manager_with(None, Some(budget));
+        assert_eq!(with_wraps(&wraps_only, None, Some(recently.clone())), (false, None));
+        assert_eq!(
+            with_wraps(&wraps_only, Some(budget), Some(recently.clone())),
+            (true, Some(RotationDueReason::Wraps))
+        );
+
+        // A backend that cannot rotate is never told to, whatever it wrapped.
+        let mut key = aged_key(None, recently);
+        key.wrap_budget_reserved = Some(u64::MAX);
+        wraps_only.apply_rotation_readiness(&mut key, false, &now);
+        assert!(!key.rotation_due);
+        assert_eq!(key.rotation_due_reason, Some(RotationDueReason::Unsupported));
+    }
+
+    /// Threshold parsing matches the age threshold's discipline: unset and
+    /// unparsable both disable the signal rather than inventing a policy.
+    #[test]
+    fn rotation_wrap_threshold_parsing_refuses_to_guess() {
+        assert_eq!(parse_rotation_max_wraps(None), None);
+        assert_eq!(parse_rotation_max_wraps(Some("not-a-number")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("-1")), None);
+        assert_eq!(parse_rotation_max_wraps(Some("0")), None);
+        // Clamped: below one reservation block the first reservation would trip it.
+        assert_eq!(parse_rotation_max_wraps(Some("1")), Some(MIN_ROTATION_MAX_WRAPS));
+        assert_eq!(
+            parse_rotation_max_wraps(Some(" 5000000 ")),
+            Some(5_000_000),
+            "a configured budget above the floor is honored verbatim"
+        );
+    }
+
     /// The two fields are additive on the wire: a payload written before they
     /// existed still deserializes, and a key with no verdict serializes exactly
     /// as it did before.

crates/kms/src/types.rs

@@ -217,6 +217,12 @@ pub enum RotationDueReason {
     /// The key has never been rotated and has existed longer than the
     /// configured maximum age.
     NeverRotated,
+    /// The key has wrapped more data keys than the configured maximum.
+    ///
+    /// Counted per key-material version, so a rotation restarts the budget.
+    /// The count is an over-estimate by construction (see the backend's
+    /// reservation accounting), so this verdict errs toward rotating early.
+    Wraps,
     /// The backend cannot rotate keys at all, so no age makes one due.
     Unsupported,
 }

crates/obs/src/metrics/collectors/audit.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Audit metrics collector.
 //!
 //! Collects audit log metrics including failed messages, queue length,

crates/obs/src/metrics/collectors/cluster_config.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Cluster config metrics collector.
 //!
 //! Collects cluster configuration metrics including storage class

crates/obs/src/metrics/collectors/cluster_erasure_set.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Cluster erasure set metrics collector.
 //!
 //! Collects erasure coding set metrics including parity, quorum,

crates/obs/src/metrics/collectors/cluster_health.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Cluster health metrics collector.
 //!
 //! Collects cluster-wide health metrics including drive counts

crates/obs/src/metrics/collectors/cluster_iam.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Cluster IAM metrics collector.
 //!
 //! Collects IAM (Identity and Access Management) metrics including

crates/obs/src/metrics/collectors/cluster_usage.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Cluster usage metrics collector.
 //!
 //! Collects cluster-wide and per-bucket usage metrics including

crates/obs/src/metrics/collectors/ilm.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! ILM (Information Lifecycle Management) metrics collector.
 //!
 //! Collects ILM metrics including pending tasks, active tasks,

crates/obs/src/metrics/collectors/notification.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Notification metrics collector.
 //!
 //! Collects notification system metrics including events sent,

crates/obs/src/metrics/collectors/notification_target.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::metrics::report::PrometheusMetric;
 use crate::metrics::schema::notification_target::{
     NOTIFICATION_TARGET_FAILED_MESSAGES_BY_SERVER_MD, NOTIFICATION_TARGET_FAILED_MESSAGES_MD,

crates/obs/src/metrics/collectors/replication.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Replication metrics collector.
 //!
 //! Collects cluster-wide replication metrics including queue stats,

crates/obs/src/metrics/collectors/request.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! API request metrics collector.
 //!
 //! Collects API request metrics including request counts, errors,

crates/obs/src/metrics/collectors/scanner.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Scanner metrics collector.
 //!
 //! Collects background scanner metrics including bucket-drive scans,

crates/obs/src/metrics/collectors/system_cpu.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System CPU metrics collector.
 //!
 //! Collects CPU metrics including load average, CPU time distribution,

crates/obs/src/metrics/collectors/system_drive.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System drive metrics collector.
 //!
 //! Collects detailed drive/disk metrics including capacity, I/O statistics,

crates/obs/src/metrics/collectors/system_gpu.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System GPU metrics collector.
 //!
 //! Collects GPU memory usage metrics using NVML library.

crates/obs/src/metrics/collectors/system_memory.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System memory metrics collector.
 //!
 //! Collects memory-related metrics including total, used, free,

crates/obs/src/metrics/collectors/system_network.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System network metrics collector.
 //!
 //! Collects internode network metrics including errors, dial times,

crates/obs/src/metrics/collectors/system_process.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! System process metrics collector.
 //!
 //! Collects process-level metrics including file descriptors, memory,

crates/obs/src/metrics/schema/audit.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/bucket.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/bucket_replication.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_config.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_erasure_set.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_health.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_iam.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_notification.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/cluster_usage.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/ilm.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/node_bucket.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, MetricSubsystem, new_gauge_md};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/node_disk.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, MetricSubsystem, new_gauge_md};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/notification_target.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/process_resource.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, MetricSubsystem, new_gauge_md};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/replication.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/request.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, MetricSubsystem, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 /// name label
@@ -32,6 +30,13 @@ const API_SERVER_NAME_TYPE_LE_LABELS: [&str; 4] = [SERVER_LABEL, NAME_LABEL, TYP
 const API_TYPE_LABELS: [&str; 1] = [TYPE_LABEL];
 const API_SERVER_TYPE_LABELS: [&str; 2] = [SERVER_LABEL, TYPE_LABEL];
 
+// Declared for MinIO metric parity but never emitted: no collector passes these
+// descriptors to `PrometheusMetric::from_descriptor`, so the wire names
+// (`rejected_auth_total`, `rejected_header_total`, `rejected_timestamp_total`,
+// `rejected_invalid_total`, `waiting_total`, `incoming_total`) never appear in a
+// scrape. Kept so the gap stays greppable rather than silently disappearing with
+// their `MetricName` variants; wiring an emitter is what retires these allows.
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REJECTED_AUTH_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ApiRejectedAuthTotal,
@@ -41,6 +46,7 @@ pub static API_REJECTED_AUTH_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::ne
     )
 });
 
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REJECTED_HEADER_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ApiRejectedHeaderTotal,
@@ -50,6 +56,7 @@ pub static API_REJECTED_HEADER_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::
     )
 });
 
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REJECTED_TIMESTAMP_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ApiRejectedTimestampTotal,
@@ -59,6 +66,7 @@ pub static API_REJECTED_TIMESTAMP_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLoc
     )
 });
 
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REJECTED_INVALID_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_counter_md(
         MetricName::ApiRejectedInvalidTotal,
@@ -68,6 +76,7 @@ pub static API_REJECTED_INVALID_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock:
     )
 });
 
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REQUESTS_WAITING_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::ApiRequestsWaitingTotal,
@@ -77,6 +86,7 @@ pub static API_REQUESTS_WAITING_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock:
     )
 });
 
+#[allow(dead_code, reason = "declared metric with no emitter; see note above (backlog#1823)")]
 pub static API_REQUESTS_INCOMING_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
     new_gauge_md(
         MetricName::ApiRequestsIncomingTotal,

crates/obs/src/metrics/schema/scanner.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/system_cpu.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 /// CPU system-related metric descriptors

crates/obs/src/metrics/schema/system_drive.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;
 

crates/obs/src/metrics/schema/system_gpu.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! GPU-related metric descriptors.
 //!
 //! This module defines metric descriptors for GPU monitoring,

crates/obs/src/metrics/schema/system_memory.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_network.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_network_host.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, new_counter_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_process.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 use crate::node_identity::SERVER_LABEL;
 use crate::{MetricDescriptor, MetricName, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/stats_collector.rs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
-#![allow(dead_code)]
-
 //! Statistics collection functions for metrics.
 //!
 //! This module contains functions that collect statistics from various

crates/protocols/tests/swift_metadata_persistence.rs

@@ -23,6 +23,7 @@
 //! two are tested together because a reload is the only way to tell a real
 //! merge from one that happened to look right in the cache.
 
+#![recursion_limit = "256"]
 #![cfg(feature = "swift")]
 
 use std::collections::HashMap;

crates/scanner/src/lib.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
 #![cfg_attr(docsrs, feature(doc_auto_cfg))]
 #![warn(
     // missing_docs,

crates/scanner/tests/lifecycle_integration_test.rs

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#![recursion_limit = "256"]
+
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
 use rustfs_scanner::scanner_folder::ScannerItem;