Merge branch 'cxymds/fix-1854-copyobject-error-cause' into cxymds/fix-1855-stack

* test(table-catalog): move the store-side stateful object backend into test_support

First half of the issue's PR2: the store tests' TestCatalogObjectBackend cluster (state/record/locks/pause types, the seed/fail/pause instrumented inherent impl, the TableCatalogObjectBackend trait impl, and the BlockingObjectPublication/UnserializedTestPublication commit-publication fakes — 544 lines) moves verbatim from table_catalog/tests.rs into test_support.rs, with pub(crate) visibility on the items and fields the tests reach directly. Pure move, no behavior change; the admin handler tests' TestTableCatalogObjectBackend union (its put barrier / fail-path / lock-attempt instrumentation folding into this fake) is the second half.

Verification: cargo test -p rustfs --lib table_catalog 481 passed; clippy --lib --tests -D warnings clean; make pre-commit green.

Ref rustfs/backlog#1837 (PR2, part 1).

* test(table-catalog): fold the admin object backend into the shared fake

Second half of PR2: the admin handler tests' TestTableCatalogObjectBackend (struct, inherent impl, trait impl, lock alias — 201 lines) is deleted and its instrumentation folded into the shared TestCatalogObjectBackend, which the admin tests now take through a type alias so no call site is renamed.

Two behavioral differences between the two fakes were found by the test suites rather than assumed away, and both are preserved:

- Lock observability: the admin fake implemented only acquire_write_lock, so the trait's default acquire_read_lock -> acquire_write_lock delegation made read acquisitions visible in lock_attempts. The shared fake implements both independently, so five fence/lock tests timed out until the read path also records attempts.

- Etag generation: the admin fake used content-addressed sha256 etags (its tests observe an etag and expect rewriting identical bytes to reproduce it) while the store fake uses an incrementing counter. Instead of silently picking one, the union carries a content_addressed_etags flag; the 80 admin construction sites go through TestCatalogObjectBackend::content_addressed() and the store tests keep counter semantics.

The six one-shot path-keyed injection knobs (fail/corrupt put, missing/fail read, put barrier) run before the store fake's attempt-indexed injection maps, matching each fake's original ordering.

Verification: cargo test -p rustfs --lib table_catalog 481 passed; --lib admin::handlers::table_catalog 167 passed; clippy --lib --tests -D warnings clean; make pre-commit green.

Ref rustfs/backlog#1837 (PR2, part 2).

.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/cluster/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: cluster/RPC migration leaves transport capabilities staged for upcoming owners.
-#![allow(dead_code)]
 
 mod control_plane;
 pub(crate) mod rpc;

crates/ecstore/src/cluster/rpc/client.rs

@@ -256,6 +256,7 @@ impl<S> ReplayScopeChannel<S> {
     }
 }
 
+#[allow(dead_code, reason = "replay-state probe asserted by this file's tests (backlog#1823)")]
 fn peer_replay_state(audience: &str) -> PeerReplayState {
     PEER_REPLAY_STATES
         .lock()

crates/ecstore/src/cluster/rpc/internode_data_transport.rs

@@ -43,6 +43,10 @@ use tokio::io::{AsyncReadExt, AsyncWrite};
 use tokio::sync::OnceCell;
 use uuid::Uuid;
 
+#[allow(
+    dead_code,
+    reason = "live in the cfg(not(test)) half of build_internode_data_transport_from_env (backlog#1823)"
+)]
 static INTERNODE_DATA_TRANSPORT: OnceLock<std::result::Result<Arc<dyn InternodeDataTransport>, String>> = OnceLock::new();
 
 const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream";
@@ -134,6 +138,10 @@ fn put_file_capability_status_is_legacy(status: u16) -> bool {
 }
 
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
+#[allow(
+    dead_code,
+    reason = "capability-negotiation seam; constructed only by transport test doubles (backlog#1823)"
+)]
 pub struct InternodeDataTransportCapabilities {
     /// Backend can open a streaming remote disk reader.
     pub streaming_read: bool,
@@ -150,6 +158,10 @@ pub struct InternodeDataTransportCapabilities {
 }
 
 impl InternodeDataTransportCapabilities {
+    #[allow(
+        dead_code,
+        reason = "capability-negotiation seam; used by transport test doubles (backlog#1823)"
+    )]
     pub const fn tcp_http() -> Self {
         Self {
             streaming_read: true,
@@ -221,11 +233,17 @@ pub struct NsScannerCapabilityRequest {
 #[async_trait]
 pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
     async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader>;
+    async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
+        self.open_read(request).await
+    }
     /// Opens an owned-chunk stream when this transport can retain receive-buffer
     /// ownership. `None` preserves the established `open_read` fallback.
     async fn open_read_chunks(&self, _request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
         Ok(None)
     }
+    async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
+        self.open_read_chunks(request).await
+    }
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter>;
     async fn open_walk_dir(&self, request: WalkDirStreamRequest) -> Result<FileReader>;
     async fn open_ns_scanner(&self, _request: NsScannerStreamRequest) -> Result<FileReader> {
@@ -234,7 +252,12 @@ pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
     async fn probe_ns_scanner(&self, _request: NsScannerCapabilityRequest) -> Result<Uuid> {
         Err(Error::MethodNotAllowed)
     }
+    // Interface facet nobody calls yet: every transport implements both, but no
+    // caller negotiates on them. Kept for the internode transport split
+    // (backlog#1350); deleting them would delete the seam and six impls.
+    #[allow(dead_code, reason = "unused capability-negotiation facet (backlog#1823)")]
     fn name(&self) -> &'static str;
+    #[allow(dead_code, reason = "unused capability-negotiation facet (backlog#1823)")]
     fn capabilities(&self) -> InternodeDataTransportCapabilities;
 }
 
@@ -252,6 +275,15 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
         ))
     }
 
+    async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
+        let url = build_read_file_stream_url(&request);
+        let mut headers = json_headers();
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Box::new(
+            HttpReader::new_fresh_connection_with_stall_timeout(url, Method::GET, headers, None, request.stall_timeout).await?,
+        ))
+    }
+
     async fn open_read_chunks(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
         let url = build_read_file_stream_url(&request);
         let mut headers = json_headers();
@@ -261,6 +293,16 @@ impl InternodeDataTransport for TcpHttpInternodeDataTransport {
         )))
     }
 
+    async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<ChunkReaderBox>> {
+        let url = build_read_file_stream_url(&request);
+        let mut headers = json_headers();
+        build_auth_headers(&url, &Method::GET, &mut headers)?;
+        Ok(Some(Box::new(
+            HttpChunkReader::new_fresh_connection_with_stall_timeout(url, Method::GET, headers, None, request.stall_timeout)
+                .await?,
+        )))
+    }
+
     async fn open_write(&self, request: WriteStreamRequest) -> Result<FileWriter> {
         let server_epoch = self.put_file_auth_capability(&request.endpoint).await?;
         let nonce = server_epoch.map(|_| Uuid::new_v4());
@@ -670,6 +712,10 @@ fn build_internode_data_transport_result(
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "live in the cfg(test) half of build_internode_data_transport_from_env, which bypasses the process static (backlog#1823)"
+)]
 pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result<Arc<dyn InternodeDataTransport>> {
     build_internode_data_transport_result(configured_transport).map_err(Error::other)
 }

crates/ecstore/src/cluster/rpc/peer_s3_client.rs

@@ -854,7 +854,6 @@ impl PeerS3Client for LocalPeerS3Client {
 
 #[derive(Debug)]
 pub struct RemotePeerS3Client {
-    pub node: Option<Node>,
     pub pools: Option<Vec<usize>>,
     addr: String,
     /// Health tracker for connection monitoring
@@ -886,7 +885,6 @@ impl RemotePeerS3Client {
     pub fn new(node: Option<Node>, pools: Option<Vec<usize>>) -> Self {
         let addr = node.as_ref().map(|v| v.url.to_string()).unwrap_or_default();
         let client = Self {
-            node,
             pools,
             addr,
             health: Arc::new(DiskHealthTracker::new()),
@@ -905,10 +903,6 @@ impl RemotePeerS3Client {
             .map_err(|err| Error::other(format!("can not get client, err: {err}")))
     }
 
-    pub fn get_addr(&self) -> String {
-        self.addr.clone()
-    }
-
     /// Start health monitoring for the remote peer
     fn start_health_monitoring(&self) {
         let health = Arc::clone(&self.health);
@@ -1208,6 +1202,10 @@ impl PeerS3Client for RemotePeerS3Client {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "local bucket-heal path reached only by this file's tests (backlog#1823)"
+)]
 pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result<HealResultItem> {
     let disks = clone_drives().await;
     heal_bucket_local_on_disks(bucket, opts, disks).await
@@ -1404,6 +1402,10 @@ pub(crate) async fn heal_bucket_local_on_disks(
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "reached only through heal_bucket_local, which only tests call (backlog#1823)"
+)]
 async fn clone_drives() -> Vec<Option<DiskStore>> {
     runtime_sources::local_disk_entries().await
 }
@@ -1585,15 +1587,7 @@ mod tests {
     }
 
     fn test_remote_peer(addr: &str) -> RemotePeerS3Client {
-        let node = Node {
-            url: url::Url::parse(addr).expect("test peer URL should parse"),
-            pools: vec![0],
-            is_local: false,
-            grid_host: addr.to_string(),
-        };
-
         RemotePeerS3Client {
-            node: Some(node),
             pools: Some(vec![0]),
             addr: addr.to_string(),
             health: Arc::new(DiskHealthTracker::new()),

crates/ecstore/src/cluster/rpc/remote_disk.rs

@@ -57,15 +57,17 @@ use serde::{Serialize, de::DeserializeOwned};
 use std::{
     io::Cursor,
     path::PathBuf,
+    pin::Pin,
     sync::{
         Arc,
         atomic::{AtomicU32, Ordering},
     },
+    task::{Context, Poll},
     time::Duration,
 };
 use tokio::time;
 use tokio::{
-    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
+    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, ReadBuf},
     net::TcpStream,
     time::timeout,
 };
@@ -214,6 +216,231 @@ where
     }
 }
 
+fn is_retryable_remote_body_error(error: &io::Error) -> bool {
+    if error
+        .get_ref()
+        .and_then(|source| source.downcast_ref::<rustfs_rio::BodyStalled>())
+        .is_some()
+    {
+        return true;
+    }
+
+    matches!(
+        error.kind(),
+        io::ErrorKind::ConnectionReset
+            | io::ErrorKind::BrokenPipe
+            | io::ErrorKind::ConnectionAborted
+            | io::ErrorKind::UnexpectedEof
+    )
+}
+
+fn resumed_read_request(request: &ReadStreamRequest, emitted: usize) -> io::Result<ReadStreamRequest> {
+    let offset = request
+        .offset
+        .checked_add(emitted)
+        .ok_or_else(|| io::Error::other("remote read resume offset overflow"))?;
+    let length = if request.length == 0 {
+        0
+    } else {
+        request
+            .length
+            .checked_sub(emitted)
+            .ok_or_else(|| io::Error::other("remote read resume offset exceeds requested length"))?
+    };
+    Ok(ReadStreamRequest {
+        offset,
+        length,
+        ..request.clone()
+    })
+}
+
+type ReadResumeFuture = tokio::task::JoinHandle<Result<FileReader>>;
+
+struct RetryingRemoteReader {
+    reader: Option<FileReader>,
+    transport: Arc<dyn InternodeDataTransport>,
+    request: ReadStreamRequest,
+    emitted: usize,
+    retried: bool,
+    resume: Option<ReadResumeFuture>,
+}
+
+impl RetryingRemoteReader {
+    fn new(reader: FileReader, transport: Arc<dyn InternodeDataTransport>, request: ReadStreamRequest) -> Self {
+        Self {
+            reader: Some(reader),
+            transport,
+            request,
+            emitted: 0,
+            retried: false,
+            resume: None,
+        }
+    }
+
+    fn start_resume(&mut self) -> io::Result<()> {
+        if self.request.length != 0 && self.emitted >= self.request.length {
+            self.reader = None;
+            return Ok(());
+        }
+        let request = resumed_read_request(&self.request, self.emitted)?;
+        let transport = Arc::clone(&self.transport);
+        self.resume = Some(tokio::spawn(async move { transport.open_read_fresh(request).await }));
+        Ok(())
+    }
+}
+
+impl AsyncRead for RetryingRemoteReader {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        loop {
+            if let Some(resume) = self.resume.as_mut() {
+                match Pin::new(resume).poll(cx) {
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(Ok(Ok(reader))) => {
+                        self.resume = None;
+                        self.reader = Some(reader);
+                    }
+                    Poll::Ready(Ok(Err(error))) => {
+                        self.resume = None;
+                        return Poll::Ready(Err(io::Error::other(error)));
+                    }
+                    Poll::Ready(Err(error)) => {
+                        self.resume = None;
+                        return Poll::Ready(Err(io::Error::other(error)));
+                    }
+                }
+            }
+
+            let Some(reader) = self.reader.as_mut() else {
+                return Poll::Ready(Ok(()));
+            };
+            let before = buf.filled().len();
+            match Pin::new(reader).poll_read(cx, buf) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Ok(())) => {
+                    let produced = buf.filled().len() - before;
+                    self.emitted = match self.emitted.checked_add(produced) {
+                        Some(emitted) => emitted,
+                        None => return Poll::Ready(Err(io::Error::other("remote read emitted byte count overflow"))),
+                    };
+                    return Poll::Ready(Ok(()));
+                }
+                Poll::Ready(Err(error)) if !self.retried && is_retryable_remote_body_error(&error) => {
+                    self.retried = true;
+                    if let Err(resume_error) = self.start_resume() {
+                        return Poll::Ready(Err(resume_error));
+                    }
+                    continue;
+                }
+                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+            }
+        }
+    }
+}
+
+type ChunkResumeFuture = tokio::task::JoinHandle<Result<Option<rustfs_rio::ChunkReaderBox>>>;
+
+struct RetryingRemoteChunkReader {
+    reader: Option<rustfs_rio::ChunkReaderBox>,
+    transport: Arc<dyn InternodeDataTransport>,
+    request: ReadStreamRequest,
+    emitted: usize,
+    retried: bool,
+    resume: Option<ChunkResumeFuture>,
+}
+
+impl RetryingRemoteChunkReader {
+    fn new(reader: rustfs_rio::ChunkReaderBox, transport: Arc<dyn InternodeDataTransport>, request: ReadStreamRequest) -> Self {
+        Self {
+            reader: Some(reader),
+            transport,
+            request,
+            emitted: 0,
+            retried: false,
+            resume: None,
+        }
+    }
+
+    fn start_resume(&mut self) -> io::Result<()> {
+        if self.request.length != 0 && self.emitted >= self.request.length {
+            self.reader = None;
+            return Ok(());
+        }
+        let request = resumed_read_request(&self.request, self.emitted)?;
+        let transport = Arc::clone(&self.transport);
+        self.resume = Some(tokio::spawn(async move { transport.open_read_chunks_fresh(request).await }));
+        Ok(())
+    }
+}
+
+impl AsyncRead for RetryingRemoteChunkReader {
+    fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+        if buf.remaining() == 0 {
+            return Poll::Ready(Ok(()));
+        }
+        match rustfs_rio::ChunkReader::poll_read_chunk(self.as_mut(), cx, buf.remaining()) {
+            Poll::Ready(Ok(Some(chunk))) => {
+                buf.put_slice(&chunk);
+                Poll::Ready(Ok(()))
+            }
+            Poll::Ready(Ok(None)) => Poll::Ready(Ok(())),
+            Poll::Ready(Err(error)) => Poll::Ready(Err(error)),
+            Poll::Pending => Poll::Pending,
+        }
+    }
+}
+
+impl rustfs_rio::ChunkReader for RetryingRemoteChunkReader {
+    fn poll_read_chunk(mut self: Pin<&mut Self>, cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<Bytes>>> {
+        loop {
+            if let Some(resume) = self.resume.as_mut() {
+                match Pin::new(resume).poll(cx) {
+                    Poll::Pending => return Poll::Pending,
+                    Poll::Ready(Ok(Ok(Some(reader)))) => {
+                        self.resume = None;
+                        self.reader = Some(reader);
+                    }
+                    Poll::Ready(Ok(Ok(None))) => {
+                        self.resume = None;
+                        self.reader = None;
+                        return Poll::Ready(Err(io::Error::other("remote resume transport did not provide a chunk reader")));
+                    }
+                    Poll::Ready(Ok(Err(error))) => {
+                        self.resume = None;
+                        return Poll::Ready(Err(io::Error::other(error)));
+                    }
+                    Poll::Ready(Err(error)) => {
+                        self.resume = None;
+                        return Poll::Ready(Err(io::Error::other(error)));
+                    }
+                }
+            }
+
+            let Some(reader) = self.reader.as_mut() else {
+                return Poll::Ready(Ok(None));
+            };
+            match rustfs_rio::ChunkReader::poll_read_chunk(Pin::new(reader.as_mut()), cx, max) {
+                Poll::Pending => return Poll::Pending,
+                Poll::Ready(Ok(Some(chunk))) => {
+                    self.emitted = match self.emitted.checked_add(chunk.len()) {
+                        Some(emitted) => emitted,
+                        None => return Poll::Ready(Err(io::Error::other("remote read emitted byte count overflow"))),
+                    };
+                    return Poll::Ready(Ok(Some(chunk)));
+                }
+                Poll::Ready(Ok(None)) => return Poll::Ready(Ok(None)),
+                Poll::Ready(Err(error)) if !self.retried && is_retryable_remote_body_error(&error) => {
+                    self.retried = true;
+                    if let Err(resume_error) = self.start_resume() {
+                        return Poll::Ready(Err(resume_error));
+                    }
+                    continue;
+                }
+                Poll::Ready(Err(error)) => return Poll::Ready(Err(error)),
+            }
+        }
+    }
+}
+
 #[derive(Debug)]
 pub struct RemoteDisk {
     pub id: Mutex<Option<Uuid>>,
@@ -2469,7 +2696,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        self.open_read_with_retry(ReadStreamRequest {
+        let request = ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2477,8 +2704,9 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        })
-        .await
+        };
+        let reader = self.open_read_with_retry(request.clone()).await?;
+        Ok(Box::new(RetryingRemoteReader::new(reader, Arc::clone(&self.data_transport), request)))
     }
 
     async fn read_file_stream_chunks(
@@ -2493,7 +2721,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        self.open_read_chunks_with_retry(ReadStreamRequest {
+        let request = ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2501,8 +2729,12 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        })
-        .await
+        };
+        let reader = self.open_read_chunks_with_retry(request.clone()).await?;
+        Ok(reader.map(|reader| {
+            Box::new(RetryingRemoteChunkReader::new(reader, Arc::clone(&self.data_transport), request))
+                as rustfs_rio::ChunkReaderBox
+        }))
     }
 
     /// Buffered read for remote disks.
@@ -4123,6 +4355,278 @@ mod tests {
         }
     }
 
+    #[derive(Debug, Clone)]
+    enum ResumeReadStep {
+        PartialThenReset(Vec<u8>),
+        Data(Vec<u8>),
+    }
+
+    #[derive(Debug, Default)]
+    struct ResumeTransport {
+        read_steps: Mutex<Vec<ResumeReadStep>>,
+        chunk_steps: Mutex<Vec<ResumeReadStep>>,
+        read_requests: Mutex<Vec<ReadStreamRequest>>,
+        chunk_requests: Mutex<Vec<ReadStreamRequest>>,
+        fresh_read_requests: Mutex<Vec<ReadStreamRequest>>,
+        fresh_chunk_requests: Mutex<Vec<ReadStreamRequest>>,
+    }
+
+    impl ResumeTransport {
+        fn with_read_steps(read_steps: Vec<ResumeReadStep>) -> Self {
+            Self {
+                read_steps: Mutex::new(read_steps),
+                ..Self::default()
+            }
+        }
+
+        fn with_chunk_steps(chunk_steps: Vec<ResumeReadStep>) -> Self {
+            Self {
+                chunk_steps: Mutex::new(chunk_steps),
+                ..Self::default()
+            }
+        }
+    }
+
+    #[derive(Debug)]
+    struct ChunkPartialThenErrorReader {
+        data: Option<Bytes>,
+        error: Option<io::Error>,
+    }
+
+    impl rustfs_rio::ChunkReader for ChunkPartialThenErrorReader {
+        fn poll_read_chunk(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, max: usize) -> Poll<io::Result<Option<Bytes>>> {
+            if let Some(mut data) = self.data.take() {
+                let take = data.len().min(max);
+                let chunk = data.split_to(take);
+                if !data.is_empty() {
+                    self.data = Some(data);
+                }
+                return Poll::Ready(Ok(Some(chunk)));
+            }
+            if let Some(error) = self.error.take() {
+                return Poll::Ready(Err(error));
+            }
+            Poll::Ready(Ok(None))
+        }
+    }
+
+    impl AsyncRead for ChunkPartialThenErrorReader {
+        fn poll_read(self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Err(io::Error::other("chunk reader must use chunk handoff")))
+        }
+    }
+
+    fn resume_step_reader(step: ResumeReadStep) -> FileReader {
+        match step {
+            ResumeReadStep::PartialThenReset(data) => Box::new(PartialThenErrorReader {
+                cursor: Cursor::new(data),
+                error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")),
+            }),
+            ResumeReadStep::Data(data) => Box::new(Cursor::new(data)),
+        }
+    }
+
+    fn resume_step_chunk_reader(step: ResumeReadStep) -> rustfs_rio::ChunkReaderBox {
+        match step {
+            ResumeReadStep::PartialThenReset(data) => Box::new(ChunkPartialThenErrorReader {
+                data: Some(Bytes::from(data)),
+                error: Some(io::Error::new(std_io::ErrorKind::ConnectionReset, "stream reset")),
+            }),
+            ResumeReadStep::Data(data) => Box::new(ChunkPartialThenErrorReader {
+                data: Some(Bytes::from(data)),
+                error: None,
+            }),
+        }
+    }
+
+    #[async_trait::async_trait]
+    impl InternodeDataTransport for ResumeTransport {
+        async fn open_read(&self, request: ReadStreamRequest) -> Result<FileReader> {
+            self.read_requests
+                .lock()
+                .expect("read request lock should not be poisoned")
+                .push(request);
+            let step = self
+                .read_steps
+                .lock()
+                .expect("read steps lock should not be poisoned")
+                .remove(0);
+            Ok(resume_step_reader(step))
+        }
+
+        async fn open_read_fresh(&self, request: ReadStreamRequest) -> Result<FileReader> {
+            self.fresh_read_requests
+                .lock()
+                .expect("fresh read request lock should not be poisoned")
+                .push(request.clone());
+            self.open_read(request).await
+        }
+
+        async fn open_read_chunks(&self, request: ReadStreamRequest) -> Result<Option<rustfs_rio::ChunkReaderBox>> {
+            self.chunk_requests
+                .lock()
+                .expect("chunk request lock should not be poisoned")
+                .push(request);
+            let step = self
+                .chunk_steps
+                .lock()
+                .expect("chunk steps lock should not be poisoned")
+                .remove(0);
+            Ok(Some(resume_step_chunk_reader(step)))
+        }
+
+        async fn open_read_chunks_fresh(&self, request: ReadStreamRequest) -> Result<Option<rustfs_rio::ChunkReaderBox>> {
+            self.fresh_chunk_requests
+                .lock()
+                .expect("fresh chunk request lock should not be poisoned")
+                .push(request.clone());
+            self.open_read_chunks(request).await
+        }
+
+        async fn open_write(&self, _request: WriteStreamRequest) -> Result<FileWriter> {
+            panic!("open_write should not be used in remote read resume tests");
+        }
+
+        async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result<FileReader> {
+            panic!("open_walk_dir should not be used in remote read resume tests");
+        }
+
+        fn name(&self) -> &'static str {
+            "resume-test"
+        }
+
+        fn capabilities(&self) -> InternodeDataTransportCapabilities {
+            InternodeDataTransportCapabilities::tcp_http()
+        }
+    }
+
+    fn resume_request(length: usize) -> ReadStreamRequest {
+        ReadStreamRequest {
+            endpoint: "http://remote".to_string(),
+            disk: "disk".to_string(),
+            volume: "volume".to_string(),
+            path: "path".to_string(),
+            offset: 7,
+            length,
+            stall_timeout: None,
+        }
+    }
+
+    #[tokio::test]
+    async fn remote_reader_resumes_from_emitted_bytes_without_duplicates() {
+        let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::Data(b"456789".to_vec())]));
+        let request = resume_request(10);
+        let reader = resume_step_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec()));
+        let mut reader = RetryingRemoteReader::new(reader, transport.clone(), request);
+        let mut output = Vec::new();
+        reader
+            .read_to_end(&mut output)
+            .await
+            .expect("one body reset should be resumed");
+
+        assert_eq!(output, b"0123456789");
+        let requests = transport
+            .read_requests
+            .lock()
+            .expect("read request lock should not be poisoned");
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].offset, 11);
+        assert_eq!(requests[0].length, 6);
+        assert_eq!(
+            transport
+                .fresh_read_requests
+                .lock()
+                .expect("fresh read request lock should not be poisoned")
+                .len(),
+            1
+        );
+    }
+
+    #[tokio::test]
+    async fn remote_chunk_reader_resumes_from_emitted_bytes_without_duplicates() {
+        let transport = Arc::new(ResumeTransport::with_chunk_steps(vec![ResumeReadStep::Data(b"456789".to_vec())]));
+        let request = resume_request(10);
+        let reader = resume_step_chunk_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec()));
+        let mut reader = RetryingRemoteChunkReader::new(reader, transport.clone(), request);
+        let mut output = Vec::new();
+        reader
+            .read_to_end(&mut output)
+            .await
+            .expect("chunk body reset should be resumed");
+
+        assert_eq!(output, b"0123456789");
+        let requests = transport
+            .chunk_requests
+            .lock()
+            .expect("chunk request lock should not be poisoned");
+        assert_eq!(requests.len(), 1);
+        assert_eq!(requests[0].offset, 11);
+        assert_eq!(requests[0].length, 6);
+        assert_eq!(
+            transport
+                .fresh_chunk_requests
+                .lock()
+                .expect("fresh chunk request lock should not be poisoned")
+                .len(),
+            1
+        );
+    }
+
+    #[tokio::test]
+    async fn remote_reader_retries_at_most_once_and_preserves_non_retryable_errors() {
+        let transport = Arc::new(ResumeTransport::with_read_steps(vec![ResumeReadStep::PartialThenReset(b"456".to_vec())]));
+        let mut reader = RetryingRemoteReader::new(
+            resume_step_reader(ResumeReadStep::PartialThenReset(b"0123".to_vec())),
+            transport.clone(),
+            resume_request(7),
+        );
+        let error = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("second reset must not retry");
+        assert_eq!(error.kind(), std_io::ErrorKind::ConnectionReset);
+        assert_eq!(
+            transport
+                .read_requests
+                .lock()
+                .expect("read request lock should not be poisoned")
+                .len(),
+            1
+        );
+
+        let transport = Arc::new(ResumeTransport::default());
+        let reader = PartialThenErrorReader {
+            cursor: Cursor::new(b"data".to_vec()),
+            error: Some(io::Error::new(std_io::ErrorKind::PermissionDenied, "permission denied")),
+        };
+        let mut reader = RetryingRemoteReader::new(Box::new(reader), transport.clone(), resume_request(4));
+        let error = reader
+            .read_to_end(&mut Vec::new())
+            .await
+            .expect_err("non-retryable errors must not retry");
+        assert_eq!(error.kind(), std_io::ErrorKind::PermissionDenied);
+        assert!(
+            transport
+                .read_requests
+                .lock()
+                .expect("read request lock should not be poisoned")
+                .is_empty()
+        );
+    }
+
+    #[test]
+    fn resumed_read_request_checks_large_offsets() {
+        let request = ReadStreamRequest {
+            offset: usize::MAX - 1,
+            length: 0,
+            ..resume_request(0)
+        };
+        assert!(resumed_read_request(&request, 2).is_err());
+
+        let request = resume_request(4);
+        assert!(resumed_read_request(&request, 5).is_err());
+    }
+
     fn init_tracing(filter_level: Level) {
         INIT.call_once(|| {
             let _ = tracing_subscriber::fmt()

crates/ecstore/src/cluster/rpc/remote_locker.rs

@@ -48,10 +48,6 @@ impl RemoteClient {
         Self { addr: endpoint }
     }
 
-    pub fn from_url(url: url::Url) -> Self {
-        Self { addr: url.to_string() }
-    }
-
     fn build_ping_request() -> PingRequest {
         let mut fbb = flatbuffers::FlatBufferBuilder::new();
         let payload = fbb.create_vector(b"health-check");

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/codec/bridge.rs

@@ -26,6 +26,7 @@ pub(crate) const GET_RECONSTRUCT_OUTCOME_SKIP_DATA_COMPLETE: &str = "skip_data_c
 pub(crate) const GET_RECONSTRUCT_OUTCOME_SKIP_EMPTY_PAYLOAD: &str = "skip_empty_payload";
 
 pub(crate) trait DecodeWorkspace: Send + Sync + 'static {
+    #[allow(dead_code, reason = "workspace width asserted by decode_reader tests (backlog#1823)")]
     fn shard_len(&self) -> usize;
 }
 
@@ -33,11 +34,14 @@ pub(crate) trait ErasureDecodeEngine: Send + Sync + 'static {
     type Workspace: DecodeWorkspace;
 
     fn data_shards(&self) -> usize;
+    #[allow(dead_code, reason = "engine trait facet asserted by decode_reader tests (backlog#1823)")]
     fn parity_shards(&self) -> usize;
     fn block_size(&self) -> usize;
     fn engine_name(&self) -> &'static str;
 
+    #[allow(dead_code, reason = "engine trait facet asserted by decode_reader tests (backlog#1823)")]
     fn supports_progressive_decode(&self) -> bool;
+    #[allow(dead_code, reason = "engine trait facet asserted by decode_reader tests (backlog#1823)")]
     fn supports_aligned_shards(&self) -> bool;
 
     fn prepare_workspace(&self, shard_len: usize) -> io::Result<Self::Workspace>;

crates/ecstore/src/erasure/codec/workspace.rs

@@ -24,6 +24,7 @@ impl RustfsCodecDecodeWorkspace {
     }
 
     #[inline]
+    #[allow(dead_code, reason = "workspace width asserted by decode_reader tests (backlog#1823)")]
     pub(crate) fn shard_len(&self) -> usize {
         self.shard_len
     }

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

@@ -213,6 +213,7 @@ fn shard_read_launch_rank(cost: ShardReadCost) -> u8 {
     }
 }
 
+#[allow(dead_code, reason = "launch ordering asserted by this file's tests (backlog#1823)")]
 fn shard_read_launch_order(read_costs: &[ShardReadCost], num_readers: usize, locality_preference_enabled: bool) -> Vec<usize> {
     let mut order: Vec<usize> = (0..num_readers).collect();
     if locality_preference_enabled {
@@ -408,6 +409,10 @@ where
     R: crate::erasure::coding::ShardSource,
 {
     // Readers should handle disk errors before being passed in, ensuring each reader reaches the available number of BitrotReaders
+    #[allow(
+        dead_code,
+        reason = "ParallelReader constructor used only by this file's tests (backlog#1823)"
+    )]
     pub fn new(readers: Vec<Option<BitrotReader<R>>>, e: Erasure, offset: usize, total_length: usize) -> Self {
         Self::new_with_metrics_path_read_timeout_and_reconstruction_verification(
             readers,
@@ -420,6 +425,7 @@ where
         )
     }
 
+    #[allow(dead_code, reason = "constructor used only by this file's tests (backlog#1823)")]
     pub fn new_with_metrics_path(
         readers: Vec<Option<BitrotReader<R>>>,
         e: Erasure,
@@ -438,6 +444,7 @@ where
         )
     }
 
+    #[allow(dead_code, reason = "constructor used only by this file's tests (backlog#1823)")]
     pub fn new_with_metrics_path_and_read_costs(
         readers: Vec<Option<BitrotReader<R>>>,
         e: Erasure,
@@ -514,6 +521,7 @@ where
         )
     }
 
+    #[allow(dead_code, reason = "constructor used only by this file's tests (backlog#1823)")]
     fn new_with_read_timeout(
         readers: Vec<Option<BitrotReader<R>>>,
         e: Erasure,
@@ -1330,10 +1338,6 @@ where
             }
         }
     }
-
-    pub fn can_decode(&self, shards: &[Option<Vec<u8>>]) -> bool {
-        shards.iter().filter(|s| s.is_some()).count() >= self.data_shards
-    }
 }
 
 #[async_trait::async_trait]
@@ -1539,6 +1543,7 @@ impl Erasure {
             .await
     }
 
+    #[allow(dead_code, reason = "read-cost decode path asserted by this file's tests (backlog#1823)")]
     pub(crate) async fn decode_with_read_costs<W, R>(
         &self,
         writer: &mut W,
@@ -1609,9 +1614,9 @@ impl Erasure {
             *ret_err = Some(err.into());
         }
 
-        // Equivalent to `ParallelReader::can_decode`; inlined so this helper does
-        // not need to borrow the reader, leaving the reader free for the
-        // concurrent next-stripe read under prefetch.
+        // Shard-availability check, written out here rather than called on the
+        // reader so this helper does not need to borrow it, leaving the reader
+        // free for the concurrent next-stripe read under prefetch.
         let available_shards = shards.iter().filter(|shard| shard.is_some()).count();
         if available_shards < self.data_shards {
             let reason = GetObjectFailureReason::ReadQuorum;

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

@@ -138,6 +138,10 @@ where
     S: ShardStripeSource + Send + 'static,
     E: ErasureDecodeEngine + Clone + Send + Sync + 'static,
 {
+    #[allow(
+        dead_code,
+        reason = "default-metrics-path constructor used only by this file's tests (backlog#1823)"
+    )]
     pub(crate) fn new(source: S, engine: E, total_length: usize) -> io::Result<Self> {
         Self::new_with_metrics_path(source, engine, total_length, GET_OBJECT_PATH_CODEC_STREAMING)
     }
@@ -679,6 +683,10 @@ pub(crate) struct SyncErasureDecodeReader<R> {
 }
 
 impl<R> SyncErasureDecodeReader<R> {
+    #[allow(
+        dead_code,
+        reason = "default-metrics-path constructor used only by this file's tests (backlog#1823)"
+    )]
     pub(crate) fn new(inner: R) -> Self {
         Self::new_with_metrics_path(inner, GET_OBJECT_PATH_CODEC_STREAMING)
     }
@@ -805,6 +813,7 @@ where
     Ok(true)
 }
 
+#[allow(dead_code, reason = "shard emission asserted by this file's tests (backlog#1823)")]
 fn emit_data_shards(state: &StripeReadState, data_shards: usize, block_size: usize, remaining: usize) -> io::Result<Vec<u8>> {
     let mut output = Vec::new();
     emit_data_shards_into(state, data_shards, block_size, remaining, &mut output)?;

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

@@ -166,6 +166,7 @@ where
     if total == 0 { Ok(None) } else { Ok(Some(total)) }
 }
 
+#[allow(dead_code, reason = "byte accounting asserted by this file's tests (backlog#1823)")]
 fn queued_block_bytes(block: &[Bytes]) -> usize {
     block.iter().map(Bytes::len).sum()
 }

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 {
@@ -1043,6 +1110,10 @@ impl Erasure {
     ///
     /// # Errors
     /// Returns error if reading from reader fails or if callback returns error
+    #[allow(
+        dead_code,
+        reason = "callback encode path exercised only by this file's tests (backlog#1823)"
+    )]
     pub(crate) async fn encode_stream_callback_async<F, Fut, E, R>(
         self: std::sync::Arc<Self>,
         reader: &mut R,
@@ -1405,7 +1476,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 +1484,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/erasure/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: erasure codec migration keeps staged streaming decode paths in this module.
-#![allow(dead_code)]
 
 pub(crate) mod codec;
 pub(crate) mod coding;

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

@@ -2741,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));
     }
 }
 
@@ -2792,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>,
@@ -2814,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>,
@@ -3212,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
@@ -9816,6 +9951,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
             &disk_files,
             &disks,
@@ -9879,6 +10015,7 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
+            Arc::new(ErasureCache::new()),
             &fi,
             &disk_files,
             &vec![Some(disk); erasure.total_shard_count()],
@@ -9959,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,
@@ -10044,6 +10182,7 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
+                Arc::new(ErasureCache::new()),
                 range_offset,
                 range_length as i64,
                 &mut writer,
@@ -10155,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,6 +2315,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
         }
         ensure_multipart_bucket_lifecycle_lock_held(bucket, object, opts)?;
 
+        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));
+
         let commit_set = self.clone();
         let commit_bucket = bucket.to_owned();
         let commit_object = object.to_owned();
@@ -2323,6 +2354,18 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             // 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,
@@ -2354,6 +2397,19 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 });
             }
 
+            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
@@ -2364,6 +2420,27 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 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
@@ -2375,7 +2452,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
             commit_set.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.
@@ -2417,24 +2493,6 @@ impl crate::storage_api_contracts::multipart::MultipartOperations for SetDisks {
                 );
             }
 
-            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;
-                }
-            }
-
-            commit_set.record_capacity_scope_if_needed(commit_capacity_scope_token, &online_disks);
-
-            fi.is_latest = true;
-
-            commit_set
-                .invalidate_get_object_metadata_cache(&commit_bucket, &commit_object)
-                .await;
-
-            drop(_object_lock_guard); // drop object lock guard to release the lock
             drop(_upload_guard);
 
             Ok(ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned))
@@ -2469,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
@@ -2480,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;
@@ -2882,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;
@@ -4968,6 +5230,182 @@ mod tests {
         .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() {
@@ -6193,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)
@@ -6313,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/io-metrics/src/lib.rs

@@ -2703,6 +2703,10 @@ 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);

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/log-analyzer/src/rules/seed/heal.rs

@@ -97,7 +97,7 @@ pub(super) fn rules() -> Vec<Rule> {
             )
         },
         Rule {
-            anchors: strings(["Heal task execution failed", "Heal manager is not running"]),
+            anchors: strings(["Heal task execution failed"]),
             ..base(
                 "heal-task-failure",
                 P2Degraded,

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,