chore(ecstore): drop the cluster and erasure dead_code blankets

Removing both blankets exposes 23 items, of which only four are deleted. The ratio is the point: close to the core data path the blankets were hiding test assertions and migration seams, not dead code.

A cfg-split function is the reason two symbols in the internode transport look dead when neither is. build_internode_data_transport_from_env has two bodies, one under #[cfg(test)] that calls build_internode_data_transport directly and one under #[cfg(not(test))] that goes through the INTERNODE_DATA_TRANSPORT static so tests do not share process-global transport state. Each half's helper is live in exactly one build, and because cargo check --tests compiles both the lib target and the test harness, both symbols appear in one warning list. Deleting either one breaks the other lane. Both are kept with allows naming their half.

Three deletion candidates were withdrawn after a per-name grep: ParallelReader::new, ErasureDecodeReader::new and SyncErasureDecodeReader::new all have test callers. The last two are exactly the shape of the dead wrapper deleted in #6084 — a thin forward to a new_with_metrics_path sibling — except that sibling is live in production (set_disk/read.rs) and the wrappers are used by tests.

Deleted:

- RemotePeerS3Client::get_addr and RemoteLocker::from_url, neither with a consumer in any lane.
- RemotePeerS3Client's node field, which new writes after using it to derive addr and nothing ever reads. Its only other writer was a test helper that built a whole Node solely to fill the field; that block goes too.
- ParallelReader::can_decode, superseded by an inlined copy. The copy's comment named the method it replaced, so deleting the method alone would have left a dangling reference; the comment now describes the check instead of pointing at a method that no longer exists.

Kept with allows: the erasure items are decode/encode invariants asserted by their own files' tests (shard_read_launch_order, decode_with_read_costs, emit_data_shards, queued_block_bytes, the engine trait facets, the ParallelReader and decode-reader constructors, encode_stream_callback_async). On the cluster side, peer_replay_state, heal_bucket_local and clone_drives are test-only, InternodeDataTransportCapabilities and tcp_http are constructed only by transport test doubles, and the InternodeDataTransport trait's name/capabilities pair is an unused capability-negotiation facet kept for the transport split (backlog#1350) — six impls provide them and no caller negotiates on them yet.

Verification, four lanes warning-free: default, --tests, --features rio-v2 --tests, --features test-util --tests. cargo nextest run -p rustfs-ecstore 4041 passed; clippy --lib --tests -D warnings clean; make pre-commit exit 0.

Ref rustfs/backlog#1823 (step 2).

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

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

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

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

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

@@ -43,6 +43,10 @@ use tokio::io::{AsyncReadExt, AsyncWrite};
 use tokio::sync::OnceCell;
 use uuid::Uuid;
 
+#[allow(
+    dead_code,
+    reason = "live in the cfg(not(test)) half of build_internode_data_transport_from_env (backlog#1823)"
+)]
 static INTERNODE_DATA_TRANSPORT: OnceLock<std::result::Result<Arc<dyn InternodeDataTransport>, String>> = OnceLock::new();
 
 const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream";
@@ -134,6 +138,10 @@ fn put_file_capability_status_is_legacy(status: u16) -> bool {
 }
 
 #[derive(Debug, Clone, Copy, Eq, PartialEq)]
+#[allow(
+    dead_code,
+    reason = "capability-negotiation seam; constructed only by transport test doubles (backlog#1823)"
+)]
 pub struct InternodeDataTransportCapabilities {
     /// Backend can open a streaming remote disk reader.
     pub streaming_read: bool,
@@ -150,6 +158,10 @@ pub struct InternodeDataTransportCapabilities {
 }
 
 impl InternodeDataTransportCapabilities {
+    #[allow(
+        dead_code,
+        reason = "capability-negotiation seam; used by transport test doubles (backlog#1823)"
+    )]
     pub const fn tcp_http() -> Self {
         Self {
             streaming_read: true,
@@ -234,7 +246,12 @@ pub trait InternodeDataTransport: Send + Sync + std::fmt::Debug {
     async fn probe_ns_scanner(&self, _request: NsScannerCapabilityRequest) -> Result<Uuid> {
         Err(Error::MethodNotAllowed)
     }
+    // Interface facet nobody calls yet: every transport implements both, but no
+    // caller negotiates on them. Kept for the internode transport split
+    // (backlog#1350); deleting them would delete the seam and six impls.
+    #[allow(dead_code, reason = "unused capability-negotiation facet (backlog#1823)")]
     fn name(&self) -> &'static str;
+    #[allow(dead_code, reason = "unused capability-negotiation facet (backlog#1823)")]
     fn capabilities(&self) -> InternodeDataTransportCapabilities;
 }
 
@@ -670,6 +687,10 @@ fn build_internode_data_transport_result(
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "live in the cfg(test) half of build_internode_data_transport_from_env, which bypasses the process static (backlog#1823)"
+)]
 pub fn build_internode_data_transport(configured_transport: Option<&str>) -> Result<Arc<dyn InternodeDataTransport>> {
     build_internode_data_transport_result(configured_transport).map_err(Error::other)
 }

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

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

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

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

crates/ecstore/src/config/com.rs

@@ -46,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

@@ -189,10 +189,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;

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";
@@ -117,23 +89,11 @@ pub(crate) const GET_STAGE_READ_VERSION_PATH_CHECK: &str = "read_version_path_ch
 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 +159,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)]

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

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

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

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

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

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

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

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

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

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

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

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

crates/ecstore/src/erasure/mod.rs

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

crates/ecstore/src/error/mod.rs

@@ -13,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

@@ -449,16 +449,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,
@@ -1211,7 +1205,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);
+        }
     }
 }
 

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,10 @@ 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 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 +328,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 +410,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 +507,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/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/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/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;
 

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

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

@@ -12,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/system_cpu.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, new_gauge_md, subsystems};
 /// CPU system-related metric descriptors

crates/obs/src/metrics/schema/system_drive.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/system_gpu.rs

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

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

@@ -12,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, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_network.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, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_network_host.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, new_counter_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/schema/system_process.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, new_counter_md, new_gauge_md, subsystems};
 use std::sync::LazyLock;

crates/obs/src/metrics/stats_collector.rs

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

docs/architecture/background-services-inventory.md

@@ -22,7 +22,7 @@ not define a new scheduler, controller framework, or shutdown contract.
 | Scanner | `rustfs/src/main.rs::run` calls `init_data_scanner(ctx.clone(), store.clone())` after successful startup log and global init time. | Main shutdown calls `ctx.cancel()`; if scanner was enabled it also calls `shutdown_background_services()`. | Scanner loop receives the main runtime token. |
 | Heal/AHM | Main creates `create_ahm_services_cancel_token()` before scanner/heal feature checks and calls `init_heal_manager(...)` when heal or scanner is enabled. | Main shutdown calls `shutdown_ahm_services()` when heal or scanner was enabled. | Global AHM token plus channel/worker-local state. |
 | Replication pool | Main calls `init_background_replication(store.clone())` after global config init, then `pool.init_resync(ctx.clone(), buckets.clone())` after bucket listing. | No direct main shutdown call for the replication pool; resync receives the main runtime token. | Resync routine uses the main runtime token; per-bucket resync uses registered cancel tokens. |
-| Lifecycle expiry/transition | `ECStore::init` calls `init_background_expiry(self.clone())` and `init_background_stale_multipart_upload_cleanup(self.clone())`. | Expiry workers read `get_background_services_cancel_token()` and fall back to a private token if none exists. Stale multipart cleanup exits when the weak ECStore reference cannot upgrade. | `ECStore::init` binds the main runtime token into the instance context with `bind_background_cancel_token(ctx)` before expiry starts, so the private-token fallback is a defensive path rather than the normal one. |
+| Lifecycle expiry/transition | `ECStore::init` calls `init_background_expiry(self.clone())` and `init_background_stale_multipart_upload_cleanup(self.clone())`. | Expiry workers read `get_background_services_cancel_token()` and fall back to a private token if none exists. Stale multipart cleanup exits when the weak ECStore reference cannot upgrade. | Inventory search found no current startup caller for `create_background_services_cancel_token()`. |
 | Notification runtime | Main calls `init_event_notifier()` after buffer profile init. | Main shutdown calls `shutdown_event_notifier().await`. | Notification runtime owns target/replay shutdown internally. |
 | Audit runtime | Main calls `start_audit_system().await`. | Main shutdown calls `stop_audit_system().await`. | Audit runtime owns target/replay shutdown internally. |
 | Metrics and memory loops | Main calls `init_metrics_runtime(ctx.clone())`, `init_memory_observability(ctx.clone())`, and `init_auto_tuner(ctx.clone())` when observability metrics are enabled. | Main shutdown only cancels the shared runtime token. | Shared runtime token. |

docs/architecture/compat-cleanup-register.md

@@ -34,7 +34,6 @@ for later deletion.
 - `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
 - `rustfs-5063` pre-beta.9 Local KMS recovery: persisted Local KMS configs from beta.8 and earlier predate the explicit insecure-development flag, and encrypted key files use the legacy SHA-256 KDF. Remove the config fallback after supported upgrades have rewritten or explicitly resaved all pre-beta.9 configs with the development-default field, and remove the legacy KDF after supported upgrades have rewritten all pre-beta.9 Local KMS key files with explicit at-rest protection.
 - `sse-local-dek-json-v1` legacy local SSE DEK decoding: releases before the JSON envelope wrote wrapped DEKs as `base64(nonce):base64(ciphertext)`, so readers retain that decoder while all new writes use the versioned JSON envelope. Remove the colon decoder after the minimum supported direct-upgrade release writes JSON envelopes and migration tooling has rewritten every retained legacy object.
-- `rio-v2-dormant-variant` dormant `rio-v2` build variant: `crates/rio-v2` and the `rio-v2` feature ship in no default or release build and exist only as the candidate MinIO stream-format implementation for the rustfs/backlog#1638 SSE-interop adjudication. Per-PR CI keeps only `test-and-lint-rio-v2` to guard the `#[cfg(feature = "rio-v2")]` seam; the full-suite lanes (`build-rustfs-debug-binary-rio-v2`, `e2e-tests-rio-v2` in `.github/workflows/ci.yml`) run on schedule/workflow_dispatch only — see the "`rio-v2` variant lifecycle" section in [minio-file-format-compat.md](minio-file-format-compat.md). While both implementations exist, DARE/S2 stream fixes must land in both `crates/rio` and `crates/rio-v2`. No `RUSTFS_COMPAT_TODO` source marker applies: the temporary surface is CI workflow YAML plus an entire feature-gated candidate crate, not a compatibility code path inside shipping code, and workflow files are outside the marker convention's Rust scope. Remove after the #1638 adjudication lands and converges on one implementation: delete the losing implementation, its feature seam, and the gating CI jobs.
 
 ## Review Checklist
 

docs/architecture/minio-file-format-compat.md

@@ -271,32 +271,6 @@ Seam 2 surfaces its own error, but only for objects that got past seam 1.
 
 The interop harness reflects this. The reader tests are `#[ignore]` (`rustfs/src/storage/minio_generated_read_test.rs:244`, `:250`), the workflow that would run them is disabled at the GitHub Actions level and states in its own header that end-to-end MinIO-to-RustFS SSE interop is not implemented (`.github/workflows/minio-interop.yml:24-29`, `:34-39`), and the fixture suite's scope note says the tests "do not yet validate full plaintext reconstruction from MinIO-written encrypted data" (`crates/rio-v2/tests/README.md:55`).
 
-### `rio-v2` variant lifecycle
-
-The variant is deliberately **dormant** until rustfs/backlog#1638 is
-adjudicated. Dormant means:
-
-- **Per-PR CI keeps one guard job.** Only `test-and-lint-rio-v2` in
-  `.github/workflows/ci.yml` runs per PR; its job is to keep the
-  `#[cfg(feature = "rio-v2")]` seam compiling and its unit tests green so the
-  variant does not bit-rot. The full-suite lanes —
-  `build-rustfs-debug-binary-rio-v2` and `e2e-tests-rio-v2` — run only on the
-  weekly `schedule` and on `workflow_dispatch`, not per PR, per main push, or
-  in the merge queue.
-- **Post-1.0 the variant is promoted or deleted.** The #1638 adjudication
-  converges on one implementation: either `rio-v2` becomes a shipped
-  configuration, or the losing side is removed together with its feature seam
-  and its gating CI jobs. Tracked as `rio-v2-dormant-variant` in
-  [compat-cleanup-register.md](compat-cleanup-register.md).
-- **DARE/S2 fixes land in both crates.** While both implementations exist,
-  any fix to the DARE V2 stream format or the S2 compression framing/index
-  must be applied to `crates/rio` **and** `crates/rio-v2` (each has its own
-  `encrypt_reader.rs` and `compress_reader.rs`). Both implement the same
-  stream primitives; a single-sided fix forks on-disk behavior between
-  default and `rio-v2` builds and invalidates the dormant variant as an
-  interop baseline — and with full-suite CI now weekly-only, the divergence
-  could go unnoticed for up to a week.
-
 ### Reverse direction
 
 Migrating back is also unsupported. Under `rio-v2` RustFS writes its own DEK envelope into MinIO's sealed-key metadata slots and labels it with MinIO's seal algorithm (`rustfs/src/storage/sse.rs:1830-1852`), so the metadata is MinIO-shaped while the key bytes are not MinIO-openable. Default builds do not populate those slots at all (`rustfs/src/storage/sse.rs:1796-1798`). Treat RustFS-written SSE objects as readable only by RustFS.

rustfs/src/admin/handlers/tier.rs

@@ -672,6 +672,7 @@ impl Operation for RemoveTier {
     }
 }
 
+#[allow(dead_code)]
 pub struct VerifyTier {}
 #[async_trait::async_trait]
 impl Operation for VerifyTier {
@@ -775,6 +776,7 @@ fn filter_tier_stats(daily_stats: DailyAllTierStats, tier_name: Option<&str>) ->
         .collect()
 }
 
+#[allow(dead_code)]
 fn map_tier_verify_error(err: std::io::Error) -> S3Error {
     if let Some(admin_err) = err.get_ref().and_then(|inner| inner.downcast_ref::<AdminError>()) {
         return match admin_err.code.as_str() {