fix(kms): stop an oversized Vault credential window from panicking the request path

refresh_safety_window_secs is operator-supplied and unbounded, so a window like
u64::MAX passed validation and then reached `Instant::now() + safety_window` in
VaultCredentialProvider::current. After a lease-bearing login the first request
panicked with "overflow when adding duration to instant" — reachable from the
admin configure API for every login-based auth method.

Use checked arithmetic and collapse an unrepresentable window to "refuse": such
a window means every token is always inside it, so that is both the fail-closed
answer and the one the arithmetic was reaching for. The comparison moves into
one helper because current() and record_credential_gauges() must apply the same
gate, which their doc comments already require.

The other operand had the same defect: expires_at and renew_at add a TTL built
from the lease_duration the Vault server sent, an unvalidated u64 off the wire.
An unrepresentable TTL now collapses to None, which is indistinguishable from
the no-expiry case Vault already produces for zero-lease tokens; the token stays
in use and Vault still validates it on every call. Leaving that side unchecked
would have kept the same panic reachable through the lease instead of the
window.

.github/workflows/cache-warm.yml

@@ -182,12 +182,7 @@ jobs:
             echo '```'
           } >> "$GITHUB_STEP_SUMMARY"
 
-  # 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.
+  # Readers: test-and-lint-rio-v2, build-rustfs-debug-binary-rio-v2.
   warm-ci-feat-rio:
     name: Warm ci-feat-rio
     runs-on: sm-standard-4

.github/workflows/ci.yml

@@ -533,12 +533,7 @@ jobs:
 
   build-rustfs-debug-binary-rio-v2:
     name: Build RustFS Debug Binary (rio-v2)
-    # 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'
+    if: github.event_name != 'pull_request' || github.event.action != 'closed'
     needs: [ quick-checks ]
     runs-on: sm-standard-4
     timeout-minutes: 30
@@ -829,9 +824,6 @@ 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,10 +101,7 @@ 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/`. `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)).
+module split is tracked under `docs/architecture/`.
 
 ## 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", # MinIO on-disk format compatibility I/O layer (feature-gated, ships in no default build)
+    "crates/rio-v2", # Next-generation Rust I/O compatibility layer
     "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,22 +137,6 @@ 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
@@ -689,13 +673,4 @@ 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,6 +13,7 @@
 // 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,7 +256,6 @@ 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,10 +43,6 @@ 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";
@@ -138,10 +134,6 @@ 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,
@@ -158,10 +150,6 @@ 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,
@@ -233,17 +221,11 @@ 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> {
@@ -252,12 +234,7 @@ 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;
 }
 
@@ -275,15 +252,6 @@ 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();
@@ -293,16 +261,6 @@ 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());
@@ -712,10 +670,6 @@ 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,6 +854,7 @@ 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
@@ -885,6 +886,7 @@ 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()),
@@ -903,6 +905,10 @@ 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);
@@ -1202,10 +1208,6 @@ 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
@@ -1402,10 +1404,6 @@ 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
 }
@@ -1587,7 +1585,15 @@ 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,17 +57,15 @@ 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, ReadBuf},
+    io::{self, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt},
     net::TcpStream,
     time::timeout,
 };
@@ -216,231 +214,6 @@ 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>>,
@@ -2696,7 +2469,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        let request = ReadStreamRequest {
+        self.open_read_with_retry(ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2704,9 +2477,8 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        };
-        let reader = self.open_read_with_retry(request.clone()).await?;
-        Ok(Box::new(RetryingRemoteReader::new(reader, Arc::clone(&self.data_transport), request)))
+        })
+        .await
     }
 
     async fn read_file_stream_chunks(
@@ -2721,7 +2493,7 @@ impl DiskAPI for RemoteDisk {
         }
         let disk = self.disk_ref().await;
         let stall_timeout = get_object_disk_read_timeout();
-        let request = ReadStreamRequest {
+        self.open_read_chunks_with_retry(ReadStreamRequest {
             endpoint: self.endpoint.grid_host(),
             disk,
             volume: volume.to_string(),
@@ -2729,12 +2501,8 @@ impl DiskAPI for RemoteDisk {
             offset,
             length,
             stall_timeout: (!stall_timeout.is_zero()).then_some(stall_timeout),
-        };
-        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
-        }))
+        })
+        .await
     }
 
     /// Buffered read for remote disks.
@@ -4355,278 +4123,6 @@ 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,6 +48,10 @@ 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,6 +46,7 @@ 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;
@@ -199,6 +200,8 @@ 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,6 +13,7 @@
 // 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,7 +101,6 @@ 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![
@@ -151,8 +150,6 @@ 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)]
@@ -189,10 +186,6 @@ 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;
@@ -240,19 +233,17 @@ impl Config {
             .map(|(pool_index, pool)| (pool_index, pool.drives_per_set))
     }
 
-    pub fn should_inline(&self, shard_size: i64, data_shards: usize, versioned: bool) -> bool {
-        if shard_size < 0 || data_shards == 0 {
+    pub fn should_inline(&self, shard_size: i64, versioned: bool) -> bool {
+        if shard_size < 0 {
             return false;
         }
 
         let shard_size = shard_size as usize;
-        // 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)
-        };
+
+        let mut inline_block = DEFAULT_INLINE_BLOCK;
+        if self.initialized {
+            inline_block = self.inline_block;
+        }
 
         if versioned {
             shard_size <= inline_block / 8
@@ -401,7 +392,6 @@ 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>()
@@ -434,7 +424,6 @@ fn lookup_config_for_pools_with_env(
         optimize,
         inline_block,
         initialized: true,
-        inline_block_explicit,
         standard_parities,
         rrs_parities,
     })
@@ -552,26 +541,22 @@ mod tests {
     }
 
     #[test]
-    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");
+    fn should_inline_preserves_exact_default_shard_boundaries() {
+        let config = Config::default();
 
-        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),
+        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),
         ] {
             assert_eq!(
-                config.should_inline(shard_size, data_shards, versioned),
+                config.should_inline(shard_size, versioned),
                 expected,
-                "{case}: shard_size={shard_size}, data_shards={data_shards}, versioned={versioned}"
+                "{case}: shard_size={shard_size}, versioned={versioned}"
             );
         }
     }
@@ -592,28 +577,13 @@ 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, erasure.data_shards, versioned),
+                config.should_inline(shard_size, 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]);
@@ -807,7 +777,6 @@ 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));
@@ -817,25 +786,6 @@ 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,6 +13,7 @@
 // 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,7 +226,6 @@ 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)?;
@@ -1059,19 +1058,11 @@ 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,6 +13,7 @@
 // limitations under the License.
 
 // #730: data-movement migration keeps staged cleanup helpers until copy paths converge.
+#![allow(dead_code)]
 
 pub(crate) mod backpressure;
 
@@ -1018,10 +1019,6 @@ 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>,
 }
@@ -1031,10 +1028,6 @@ 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 {
@@ -1173,7 +1166,6 @@ 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,7 +653,6 @@ 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,49 +46,21 @@ 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";
@@ -112,28 +84,12 @@ 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";
@@ -199,20 +155,8 @@ 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)]
@@ -498,10 +442,6 @@ 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,6 +13,8 @@
 // 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

@@ -0,0 +1,30 @@
+// 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,11 +15,6 @@
 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};
@@ -9845,12 +9840,6 @@ 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) {
@@ -9860,46 +9849,37 @@ 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 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 (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 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?;
+        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()?;
         if fi.is_canonical_delete_marker() {
             return Ok(fi);
         }
@@ -10582,108 +10562,6 @@ 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,7 +26,6 @@ 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;
 }
 
@@ -34,14 +33,11 @@ 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,7 +24,6 @@ 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,7 +213,6 @@ 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 {
@@ -409,10 +408,6 @@ 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,
@@ -425,7 +420,6 @@ 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,
@@ -444,7 +438,6 @@ 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,
@@ -521,7 +514,6 @@ 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,
@@ -1338,6 +1330,10 @@ 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]
@@ -1543,7 +1539,6 @@ 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,
@@ -1614,9 +1609,9 @@ impl Erasure {
             *ret_err = Some(err.into());
         }
 
-        // 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.
+        // 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.
         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,10 +138,6 @@ 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)
     }
@@ -683,10 +679,6 @@ 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)
     }
@@ -813,7 +805,6 @@ 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,7 +166,6 @@ 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,16 +71,10 @@ 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)]
@@ -147,61 +141,43 @@ pub fn calc_shard_size_legacy(block_size: usize, data_shards: usize) -> usize {
 struct LegacyReedSolomonEncoder {
     data_shards: usize,
     parity_shards: usize,
-    cache_workspaces: bool,
-    encoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonEncoder>>,
-    decoder_cache: RwLock<Option<reed_solomon_simd::ReedSolomonDecoder>>,
+    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),
+        }
+    }
 }
 
 impl LegacyReedSolomonEncoder {
-    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> {
+    fn new(_data_shards: usize, _parity_shards: usize) -> io::Result<Self> {
         Ok(Self {
-            data_shards,
-            parity_shards,
-            cache_workspaces,
-            encoder_cache: RwLock::new(None),
-            decoder_cache: RwLock::new(None),
+            data_shards: _data_shards,
+            parity_shards: _parity_shards,
+            encoder_cache: std::sync::RwLock::new(None),
+            decoder_cache: std::sync::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 = {
-            match cached_encoder {
+            let mut cache_guard = self
+                .encoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to acquire encoder cache lock"))?;
+            match cache_guard.take() {
                 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)
@@ -228,15 +204,10 @@ impl LegacyReedSolomonEncoder {
             }
         }
         drop(result);
-        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);
-            }
-        }
+        *self
+            .encoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to return encoder to cache"))? = Some(encoder);
         Ok(())
     }
 
@@ -250,13 +221,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 = {
-            match cached_decoder {
+            let mut cache_guard = self
+                .decoder_cache
+                .write()
+                .map_err(|_| io::Error::other("Failed to acquire decoder cache lock"))?;
+
+            match cache_guard.take() {
                 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);
@@ -303,15 +274,10 @@ impl LegacyReedSolomonEncoder {
 
         drop(result);
 
-        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);
-            }
-        }
+        *self
+            .decoder_cache
+            .write()
+            .map_err(|_| io::Error::other("Failed to return decoder to cache"))? = Some(decoder);
 
         Ok(())
     }
@@ -469,39 +435,6 @@ 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<()>,
@@ -618,7 +551,7 @@ pub struct Erasure {
     pub data_shards: usize,
     pub parity_shards: usize,
     encoder: Option<ReedSolomonEncoder>,
-    legacy_encoder: Option<Arc<LegacyReedSolomonEncoder>>,
+    legacy_encoder: Option<LegacyReedSolomonEncoder>,
     pub block_size: usize,
     uses_legacy: bool,
     _id: Uuid,
@@ -754,7 +687,7 @@ impl Erasure {
 
         let legacy_encoder = if uses_legacy && parity_shards > 0 {
             Some(
-                cached_legacy_reed_solomon(data_shards, parity_shards)
+                LegacyReedSolomonEncoder::new(data_shards, parity_shards)
                     .map_err(|source| ErasureConstructionError::LegacyEncoder { source })?,
             )
         } else {
@@ -1110,10 +1043,6 @@ 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,
@@ -1476,7 +1405,7 @@ mod tests {
         assert_eq!(cloned.block_size, legacy.block_size);
         assert!(cloned.uses_legacy);
 
-        let data = b"legacy clone should preserve SIMD codec behavior";
+        let data = b"legacy clone should keep independent SIMD caches";
         let encoded = cloned.encode_data(data).expect("legacy clone should encode");
         let mut shards = optional_shards(&encoded);
         shards[0] = None;
@@ -1484,93 +1413,6 @@ 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,6 +13,7 @@
 // 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,12 +13,13 @@
 // 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::S3ErrorCode;
+use s3s::{S3Error, S3ErrorCode};
 
 pub type Error = StorageError;
 pub type Result<T> = core::result::Result<T, Error>;
@@ -901,7 +902,6 @@ 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,11 +910,14 @@ 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`.
@@ -1075,9 +1078,21 @@ 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),
 }
@@ -1160,6 +1175,30 @@ 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,6 +13,8 @@
 // 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

@@ -0,0 +1,25 @@
+#![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,16 +12,18 @@
 // 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 {
@@ -29,3 +31,14 @@ 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,6 +13,7 @@
 // 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,20 +25,9 @@ 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")]
     {
@@ -64,6 +53,17 @@ 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,11 +82,6 @@ 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,12 +209,15 @@ impl AsMut<Vec<Endpoints>> for PoolEndpointList {
 }
 
 impl PoolEndpointList {
-    /// 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.
+    /// 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.
     async fn create_pool_endpoints_with(
         server_addr: &str,
         disks_layout: &DisksLayout,
@@ -591,10 +594,6 @@ 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,6 +13,7 @@
 // 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,7 +4,6 @@ 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,
@@ -13,7 +12,6 @@ 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();
@@ -41,20 +39,17 @@ 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
@@ -113,7 +108,6 @@ 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,6 +13,7 @@
 // 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,7 +15,6 @@
 use super::*;
 
 use crate::io_support::rio::Index;
-use std::mem::MaybeUninit;
 
 #[cfg(feature = "rio-v2")]
 const DARE_PAYLOAD_SIZE: i64 = 64 * 1024;
@@ -449,16 +448,10 @@ impl GetObjectReader {
 }
 
 enum ReadTransform {
-    // 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 },
+    Plain {
+        visible_offset: usize,
+        visible_length: i64,
+    },
     Compressed {
         algorithm: CompressionAlgorithm,
         backend: crate::io_support::rio::ReadCompressionBackend,
@@ -929,7 +922,7 @@ struct SkipReader<R> {
     inner: R,
     bytes_to_skip: usize,
     bytes_skipped: usize,
-    scratch: Box<[MaybeUninit<u8>]>,
+    scratch: Vec<u8>,
 }
 
 impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
@@ -938,7 +931,7 @@ impl<R: AsyncRead + Unpin + Send + Sync> SkipReader<R> {
             inner,
             bytes_to_skip,
             bytes_skipped: 0,
-            scratch: Box::<[u8]>::new_uninit_slice(8192),
+            scratch: vec![0u8; 8192],
         }
     }
 }
@@ -950,7 +943,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::uninit(&mut this.scratch[..scratch_len]);
+            let mut scratch_buf = ReadBuf::new(&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)),
@@ -981,7 +974,7 @@ pub struct RangedDecompressReader<R: AsyncRead + Unpin + Send + Sync + 'static>
     target_length: usize,
     current_offset: usize,
     bytes_returned: usize,
-    scratch: Box<[MaybeUninit<u8>]>,
+    scratch: Vec<u8>,
     drain_on_done: bool,
     drain_task: Option<tokio::task::JoinHandle<()>>,
 }
@@ -1019,7 +1012,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> RangedDecompressReader<R> {
             target_length: actual_length,
             current_offset: 0,
             bytes_returned: 0,
-            scratch: Box::<[u8]>::new_uninit_slice(8192),
+            scratch: vec![0u8; 8192],
             drain_on_done,
             drain_task: None,
         })
@@ -1069,7 +1062,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::uninit(&mut this.scratch[..scratch_len]);
+            let mut temp_read_buf = ReadBuf::new(&mut this.scratch[..scratch_len]);
 
             let Some(inner) = this.inner.as_mut() else {
                 return Poll::Ready(Ok(()));
@@ -1121,8 +1114,7 @@ impl<R: AsyncRead + Unpin + Send + Sync + 'static> AsyncRead for RangedDecompres
                             );
 
                             if bytes_to_return > 0 {
-                                let data_slice =
-                                    &temp_read_buf.filled()[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
+                                let data_slice = &this.scratch[data_start_in_buffer..data_start_in_buffer + bytes_to_return];
                                 buf.put_slice(data_slice);
                                 this.bytes_returned += bytes_to_return;
 
@@ -1141,7 +1133,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(&temp_read_buf.filled()[..bytes_to_return]);
+                            buf.put_slice(&this.scratch[..bytes_to_return]);
                             this.bytes_returned += bytes_to_return;
 
                             tracing::trace!("Returned {} bytes at offset {}", bytes_to_return, old_offset);
@@ -1211,7 +1203,20 @@ impl<R: AsyncRead + Unpin + Send + 'static> AsyncRead for StreamConsumer<R> {
 
 impl<R: AsyncRead + Unpin + Send + 'static> Drop for StreamConsumer<R> {
     fn drop(&mut self) {
-        self.ensure_consumer_started();
+        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);
+        }
     }
 }
 
@@ -1258,43 +1263,6 @@ 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";
@@ -1432,36 +1400,6 @@ 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,7 +172,6 @@ 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,6 +31,7 @@ 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;
@@ -108,6 +109,18 @@ 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
@@ -275,6 +288,19 @@ 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
@@ -284,6 +310,18 @@ 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,10 +402,6 @@ 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,6 +13,7 @@
 // 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,6 +38,7 @@ 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,
@@ -142,10 +143,6 @@ 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
@@ -158,10 +155,6 @@ 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;
 }
@@ -239,6 +232,14 @@ 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())
@@ -331,6 +332,21 @@ 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);
 }
@@ -398,6 +414,10 @@ 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()
 }
@@ -491,6 +511,30 @@ 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,38 +206,6 @@ 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, held_layout.data_drives, false));
+        assert!(held.should_inline(512, 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, current_layout.data_drives, false));
+        assert!(!current.should_inline(512, false));
     }
 }
 
@@ -2792,9 +2792,6 @@ 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>,
@@ -2817,137 +2814,6 @@ 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>,
@@ -3346,7 +3212,6 @@ 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
@@ -9951,7 +9816,6 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             "bucket",
             "object",
-            Arc::new(ErasureCache::new()),
             &fi,
             &disk_files,
             &disks,
@@ -10015,7 +9879,6 @@ 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()],
@@ -10096,7 +9959,6 @@ mod tests {
         let body = SetDisks::try_get_object_direct_data_shards_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             &fi,
             &files,
             &disks,
@@ -10182,7 +10044,6 @@ mod tests {
             SetDisks::get_object_with_fileinfo(
                 bucket,
                 object,
-                Arc::new(ErasureCache::new()),
                 range_offset,
                 range_length as i64,
                 &mut writer,
@@ -10294,7 +10155,6 @@ 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,33 +1425,6 @@ 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,11 +22,6 @@
 
 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};
@@ -68,9 +63,6 @@ pub(crate) enum MultipartCommitPause {
     PutPartBeforeLockLost,
     PutPartAfterRename,
     BeforeLockLost,
-    BeforeTransactionEpochVerify,
-    BeforeObjectPublication,
-    AfterObjectPublication,
     AfterRename,
 }
 
@@ -161,24 +153,13 @@ impl Drop for MultipartCommitBarrier {
 
 #[cfg(test)]
 async fn pause_multipart_commit(bucket: &str, object: &str, pause: MultipartCommitPause) {
-    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
-        }
-    };
+    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();
     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)
@@ -2315,18 +2296,6 @@ 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();
@@ -2354,18 +2323,6 @@ 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,
@@ -2397,19 +2354,6 @@ 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
@@ -2420,27 +2364,6 @@ 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
@@ -2452,6 +2375,7 @@ 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.
@@ -2493,6 +2417,24 @@ 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))
@@ -2527,10 +2469,9 @@ 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 _, ReadOptions};
+    use crate::disk::DiskAPI as _;
     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
@@ -2539,7 +2480,6 @@ 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;
@@ -2942,208 +2882,6 @@ 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;
@@ -5230,182 +4968,6 @@ 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() {
@@ -6631,24 +6193,6 @@ 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)
@@ -6769,106 +6313,6 @@ 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,7 +482,6 @@ 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>],
@@ -503,7 +502,13 @@ impl SetDisks {
             return Ok(None);
         }
 
-        let erasure = erasure_cache.get_for_file_info(fi)?;
+        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 checksum_info = fi.erasure.get_checksum_info(part.number);
         let checksum_algo = if fi.uses_legacy_checksum && checksum_info.algorithm == HashAlgorithm::HighwayHash256S {
@@ -631,7 +636,6 @@ impl SetDisks {
         // &self,
         bucket: &str,
         object: &str,
-        erasure_cache: Arc<ErasureCache>,
         offset: usize,
         length: i64,
         writer: &mut W,
@@ -726,7 +730,13 @@ impl SetDisks {
             object, offset, length, end_offset, part_index, last_part_index, last_part_relative_offset, "Multipart read bounds"
         );
 
-        let erasure = erasure_cache.get_for_file_info(&fi)?;
+        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 part_indices: Vec<usize> = (part_index..=last_part_index).collect();
         debug!(bucket, object, ?part_indices, "Multipart part indices to stream");
@@ -1160,7 +1170,6 @@ 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>],
@@ -1171,7 +1180,14 @@ impl SetDisks {
         metrics_size_bucket: &'static str,
         prefer_data_blocks_first_reader_setup: bool,
     ) -> Result<GetCodecStreamingReaderBuildOutcome> {
-        let erasure = erasure_cache.get_for_file_info(fi)?;
+        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 (disks, files) = Self::shuffle_disks_and_parts_metadata_by_index(disks, files, fi);
 
         if fi.parts.len() == 1 {
@@ -1558,7 +1574,7 @@ struct LazyCodecPartContext {
     fi: FileInfo,
     files: Vec<FileInfo>,
     disks: Vec<Option<DiskStore>>,
-    erasure: Arc<coding::Erasure>,
+    erasure: coding::Erasure,
     skip_verify_bitrot: bool,
     metrics_object_class: &'static str,
     metrics_size_bucket: &'static str,
@@ -2042,7 +2058,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             "bucket",
             "object",
-            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2073,7 +2088,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             2,
             1,
             &mut output,
@@ -2097,7 +2111,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             usize::MAX,
             1,
             &mut output,
@@ -2119,7 +2132,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             1,
             1,
             &mut output,
@@ -2143,7 +2155,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -2181,7 +2192,6 @@ mod metadata_cache_tests {
         SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             0,
             0,
             &mut output,
@@ -2214,7 +2224,6 @@ mod metadata_cache_tests {
         let err = SetDisks::get_object_with_fileinfo(
             bucket,
             object,
-            Arc::new(ErasureCache::new()),
             0,
             1,
             &mut output,
@@ -4119,7 +4128,6 @@ mod tests {
         let result = SetDisks::get_object_decode_reader_with_fileinfo(
             CODEC_STREAMING_TEST_BUCKET,
             CODEC_STREAMING_TEST_OBJECT,
-            Arc::new(ErasureCache::new()),
             &fi,
             &[],
             &[],
@@ -4142,7 +4150,6 @@ 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,
             &[],
             &[],
@@ -4163,7 +4170,6 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
-                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4188,7 +4194,6 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
-                Arc::new(ErasureCache::new()),
                 &multipart,
                 &[],
                 &[],
@@ -4217,7 +4222,6 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
-                    Arc::new(ErasureCache::new()),
                     &multipart,
                     &[],
                     &[],
@@ -4271,7 +4275,6 @@ mod tests {
                 SetDisks::get_object_decode_reader_with_fileinfo(
                     CODEC_STREAMING_TEST_BUCKET,
                     CODEC_STREAMING_TEST_OBJECT,
-                    Arc::new(ErasureCache::new()),
                     &fi,
                     &files,
                     &disks,
@@ -4325,7 +4328,6 @@ mod tests {
             SetDisks::get_object_decode_reader_with_fileinfo(
                 CODEC_STREAMING_TEST_BUCKET,
                 CODEC_STREAMING_TEST_OBJECT,
-                Arc::new(ErasureCache::new()),
                 &fi,
                 &files,
                 &disks,
@@ -4370,7 +4372,6 @@ 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,23 +48,6 @@ 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,
@@ -105,7 +88,6 @@ 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(
@@ -177,7 +159,6 @@ 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, pool_meta_has_active_decommission};
+use crate::core::pools::local_decommission_queue_prefix;
 use crate::error::is_err_decommission_running;
 use crate::runtime::instance::InstanceContext;
 use crate::runtime::sources as runtime_sources;
@@ -109,6 +109,14 @@ 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,9 +18,8 @@ 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_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,
+    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,
 };
 use s3s::dto::{RestoreStatus, Timestamp};
 use s3s::header::X_AMZ_RESTORE;
@@ -1173,22 +1172,6 @@ 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()
     }
@@ -1501,46 +1484,6 @@ 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,21 +34,36 @@ 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 },
 
@@ -63,6 +78,12 @@ 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
@@ -108,7 +129,9 @@ impl Error {
                         | DiskError::FaultyDisk
                 ) || is_recoverable_heal_error_message(&err.to_string())
             }
-            Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message),
+            Error::TaskExecutionFailed { message } | Error::IO(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::Other(message) | Error::IO(message) => {
                 message.contains("File not found")
                     || message.contains("file not found")
                     || message.contains("File version not found")

crates/io-metrics/src/lib.rs

@@ -2703,10 +2703,6 @@ 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/api_types.rs

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

crates/kms/src/backends/vault.rs

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

crates/kms/src/backends/vault_credentials.rs

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

crates/kms/src/backends/vault_transit.rs

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

crates/kms/src/backup/vault_restore.rs

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

crates/kms/src/config.rs

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

crates/kms/src/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, ENV_KMS_ROTATION_MAX_WRAPS, KmsConfig};
+use crate::config::{ENV_KMS_ALLOW_IMMEDIATE_DELETION, ENV_KMS_ROTATION_MAX_AGE_SECS, KmsConfig};
 use crate::deletion_worker::DeletionReferenceChecker;
 use crate::error::{KmsError, Result};
 use crate::types::{
@@ -42,13 +42,6 @@ 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
@@ -75,33 +68,6 @@ 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>,
@@ -116,7 +82,6 @@ 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 {
@@ -138,7 +103,6 @@ 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(),
         }
     }
 
@@ -350,22 +314,9 @@ 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;
         };
 
@@ -382,6 +333,9 @@ 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;
         }
     }
 
@@ -1731,15 +1685,10 @@ 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
     }
 
@@ -1816,85 +1765,6 @@ 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,12 +217,6 @@ 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"]),
+            anchors: strings(["Heal task execution failed", "Heal manager is not running"]),
             ..base(
                 "heal-task-failure",
                 P2Degraded,

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

@@ -12,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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,6 +12,8 @@
 // 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;