chore(ecstore): drop the services dead_code blanket

Removing the blanket exposes twenty-five items across tier, notification and rebalance. Only eight are deleted — the lowest ratio of this burn-down so far, and the reason is that these subsystems carry heavy test coverage, so the blanket was mostly hiding test-only seams rather than dead weight.

Deleted:

- crates/ecstore/src/services/tier/warm_backend_s3sdk.rs entirely (200 lines). Its WarmBackendS3 is never constructed; the type of the same name in warm_backend_s3.rs is the live one, wrapped by the Azure backend. Two implementations of one S3 warm backend, one of them never wired.
- TierConfigMgr::begin_publish_transition and publish_candidate_inner, thin wrappers whose _with_allowed_mutation_blocks siblings carry every real caller, plus retire_driver.
- The GCS backend's MAX_MULTIPART_PUT_OBJECT_SIZE, MAX_PARTS_COUNT and MIN_PART_SIZE, and its write-only storage_class field.
- mark_started_rebalance_pools_stopped and the RStats alias.

Two deletions were withdrawn after a per-name grep, both because of an inference rather than a check:

AsyncBatchProcessor::new was deleted on the strength of grepping only BATCH_PROCESSOR_OPERATION_CUSTOM, whose two hits are its definition and its use inside new. That looked like a self-contained dead pair; new in fact has seven test callers. The warning listed both items, and only one of them was actually checked.

Deleting the two dead publish wrappers then revealed a second layer — publish_candidate_owned, remove_and_save_with, clear_and_save_with, save_tiering_config_if_current. These are not dead: publish_candidate, their caller, is #[cfg(test)], so a callee that lives in the main body has no caller in the lib build and a live one in the test build. rustc reports the roots of a dead subgraph, and the next layer down can have a different character, so each layer needs its own grep.

Kept with allows: the tier mutation-intent record helpers (asserted by store::init tests), affected_targets, tier_object_blocks_target_rebind, the rebalance snapshot and retry-wait helpers, notification_sys's tier_config_reload_worker_active and call_peer_with_timeout, and active_operation_lease_count, whose only caller sits behind #[cfg(feature = "test-util")].

Also kept, with a module note rather than removal: the ecstore-side EventNotifier. All four of its methods are unreachable and init_bucket_targets logs that it is a no-op in this build; the working stack is rustfs-notify, whose own EventNotifier drives bucket configuration. Removing it means also retiring the InstanceContext slot that holds it (backlog#939 Phase 5), which belongs in its own PR.

Worth a separate issue: MAX_MULTIPART_PUT_OBJECT_SIZE, MAX_PARTS_COUNT and MIN_PART_SIZE are declared independently in eight warm-backend files plus client/constants.rs. Only the GCS copies were dead; the other seven backends each use their own.

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).

crates/ecstore/src/event/targetlist.rs

@@ -20,6 +20,10 @@ use std::sync::atomic::AtomicI64;
 /// 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)]
+#[allow(
+    dead_code,
+    reason = "held only by the dead ecstore EventNotifier; see services/event_notification.rs (backlog#1823)"
+)]
 pub struct TargetList {
     pub total_events: AtomicI64,
 }

crates/ecstore/src/services/batch_processor.rs

@@ -23,6 +23,10 @@ use std::sync::{Arc, Mutex};
 use std::time::{Duration, Instant};
 use tokio::task::JoinSet;
 
+#[allow(
+    dead_code,
+    reason = "default operation label for the test-only AsyncBatchProcessor::new (backlog#1823)"
+)]
 const BATCH_PROCESSOR_OPERATION_CUSTOM: &str = "custom";
 const BATCH_PROCESSOR_OPERATION_READ: &str = "read";
 const BATCH_PROCESSOR_OPERATION_WRITE: &str = "write";
@@ -211,6 +215,7 @@ pub struct AsyncBatchProcessor {
 }
 
 impl AsyncBatchProcessor {
+    #[allow(dead_code, reason = "constructor used only by this file's tests (backlog#1823)")]
     pub fn new(max_concurrent: usize) -> Self {
         Self::new_with_operation(max_concurrent, BATCH_PROCESSOR_OPERATION_CUSTOM)
     }

crates/ecstore/src/services/event_notification.rs

@@ -26,11 +26,26 @@ use std::sync::atomic::Ordering;
 use tokio::sync::RwLock;
 use tracing::warn;
 
+/// Dead ecstore-side notification skeleton.
+///
+/// The working notification stack is `rustfs-notify`, whose own `EventNotifier`
+/// is the one bucket configuration actually drives. Nothing calls the methods
+/// below; `init_bucket_targets` even logs that it is a no-op in this build.
+/// Removing it means also retiring the `InstanceContext` slot that holds it
+/// (backlog#939 Phase 5), so it is left explicit here rather than half-removed.
+#[allow(
+    dead_code,
+    reason = "ecstore-side notification skeleton superseded by rustfs-notify; see module note (backlog#1823)"
+)]
 pub struct EventNotifier {
     target_list: TargetList,
     //bucket_rules_map: HashMap<String , HashMap<EventName, Rules>>,
 }
 
+#[allow(
+    dead_code,
+    reason = "ecstore-side notification skeleton superseded by rustfs-notify; see module note (backlog#1823)"
+)]
 impl EventNotifier {
     pub fn new() -> Arc<RwLock<Self>> {
         Arc::new(RwLock::new(Self {

crates/ecstore/src/services/mod.rs

@@ -13,7 +13,6 @@
 // limitations under the License.
 
 // #730: background service owners still contain staged notification/rebalance/tier paths.
-#![allow(dead_code)]
 
 pub(crate) mod batch_processor;
 pub(crate) mod event_notification;

crates/ecstore/src/services/notification_sys.rs

@@ -1539,6 +1539,7 @@ impl NotificationSys {
         workers.peers.remove(host);
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn tier_config_reload_worker_active(&self, host: &str) -> bool {
         self.tier_config_reload_workers
             .lock()
@@ -1712,6 +1713,7 @@ where
         .map_err(|_| Error::other(format!("scanner activity peer {host} timed out after {timeout_duration:?}")))?
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 async fn call_peer_with_timeout<F, Fut>(
     timeout_dur: Duration,
     host_label: &str,

crates/ecstore/src/services/rebalance/meta.rs

@@ -864,6 +864,10 @@ pub(super) fn merge_rebalance_meta(remote: &mut RebalanceMeta, local: &Rebalance
     RebalanceMetaMergeOutcome::Merged
 }
 
+#[allow(
+    dead_code,
+    reason = "stop-transition helper retained beside stop_rebalance_meta_snapshot; no caller yet (backlog#1823)"
+)]
 pub(super) fn mark_started_rebalance_pools_stopped(meta: &mut RebalanceMeta, stop_time: OffsetDateTime) {
     for pool_stat in meta.pool_stats.iter_mut() {
         if pool_stat.info.status == RebalStatus::Started {
@@ -964,6 +968,7 @@ pub(super) fn rollback_rebalance_start_meta_snapshot_for_id(
     })
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(super) fn stop_rebalance_meta_snapshot(meta: Option<&mut RebalanceMeta>, now: OffsetDateTime) -> Option<RebalanceMeta> {
     let meta = meta?;
     stop_rebalance_state(meta, now);

crates/ecstore/src/services/rebalance/migration.rs

@@ -171,6 +171,7 @@ where
 }
 
 #[allow(clippy::too_many_arguments)]
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 pub(super) async fn migrate_entry_version_with_retry_wait<Backend, F, Fut, D, DFut, W, WFut>(
     set: &Backend,
     bucket: String,

crates/ecstore/src/services/rebalance/types.rs

@@ -1,5 +1,4 @@
 use serde::{Deserialize, Serialize};
-use std::sync::Arc;
 use time::OffsetDateTime;
 use tokio_util::sync::CancellationToken;
 
@@ -32,8 +31,6 @@ pub struct RebalanceStats {
     pub cleanup_warnings: RebalanceCleanupWarnings,
 }
 
-pub type RStats = Vec<Arc<RebalanceStats>>;
-
 #[derive(Debug, Default)]
 pub(super) struct RebalanceBucketConfigs {
     pub(super) bucket_incarnation_id: Option<uuid::Uuid>,

crates/ecstore/src/services/tier/mod.rs

@@ -30,6 +30,5 @@ pub mod warm_backend_minio;
 pub mod warm_backend_r2;
 pub mod warm_backend_rustfs;
 pub mod warm_backend_s3;
-pub mod warm_backend_s3sdk;
 pub mod warm_backend_tencent;
 pub mod warm_backend_wasabi;

crates/ecstore/src/services/tier/tier.rs

@@ -488,6 +488,7 @@ impl TierCandidateMutation {
         targets
     }
 
+    #[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
     fn affected_targets(
         &self,
         manager: &TierConfigMgr,
@@ -802,6 +803,7 @@ fn tier_persisted_reference_blocks_any_target(
         .any(|target| tier_persisted_reference_blocks_target(tier_name, backend_identity, target))
 }
 
+#[allow(dead_code, reason = "asserted by this file's tests (backlog#1823)")]
 fn tier_object_blocks_target_rebind(object: &ObjectInfo, target: &TierMutationIntentTarget) -> io::Result<bool> {
     tier_object_blocks_any_target_rebind(object, std::slice::from_ref(target))
 }
@@ -2726,14 +2728,6 @@ impl TierConfigMgr {
         Self::publish_candidate_owned(handle, candidate, driver_tier.map(str::to_string), update).await
     }
 
-    fn begin_publish_transition(
-        handle: &Arc<RwLock<Self>>,
-        manager: &mut Self,
-        candidate: &Self,
-    ) -> std::result::Result<TierPublishTransition, AdminError> {
-        Self::begin_publish_transition_with_allowed_mutation_blocks(handle, manager, candidate, None)
-    }
-
     fn begin_publish_transition_with_allowed_mutation_blocks(
         handle: &Arc<RwLock<Self>>,
         manager: &mut Self,
@@ -2819,14 +2813,6 @@ impl TierConfigMgr {
         })
     }
 
-    async fn publish_candidate_inner(
-        handle: &Arc<RwLock<Self>>,
-        candidate: Self,
-        driver_tier: Option<&str>,
-    ) -> std::result::Result<(), AdminError> {
-        Self::publish_candidate_inner_with_allowed_mutation_blocks(handle, candidate, driver_tier, None).await
-    }
-
     async fn publish_candidate_inner_with_allowed_mutation_blocks(
         handle: &Arc<RwLock<Self>>,
         candidate: Self,
@@ -2939,6 +2925,7 @@ impl TierConfigMgr {
         admin_err
     }
 
+    #[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
     async fn publish_candidate_owned(
         handle: &Arc<RwLock<Self>>,
         candidate: Self,
@@ -3541,6 +3528,7 @@ impl TierConfigMgr {
         Self::update_candidate_with_config_lock(handle, api, TierCandidateMutation::Remove(tier_name.to_string(), force)).await
     }
 
+    #[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
     async fn remove_and_save_with<S>(
         handle: &Arc<RwLock<Self>>,
         api: Arc<S>,
@@ -3574,6 +3562,7 @@ impl TierConfigMgr {
         Self::update_candidate_with_config_lock(handle, api, TierCandidateMutation::Clear(force)).await
     }
 
+    #[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
     async fn clear_and_save_with<S>(
         handle: &Arc<RwLock<Self>>,
         api: Arc<S>,
@@ -3612,6 +3601,10 @@ impl TierConfigMgr {
     }
 
     #[cfg(test)]
+    #[allow(
+        dead_code,
+        reason = "lease accounting asserted by a bucket_lifecycle_ops test behind `--features test-util` (backlog#1823)"
+    )]
     pub(crate) async fn active_operation_lease_count(handle: &Arc<RwLock<Self>>, tier_name: &str) -> usize {
         let manager = handle.read().await;
         let Some(runtime) = registered_tier_driver_runtime(&manager) else {
@@ -3717,10 +3710,6 @@ impl TierConfigMgr {
         Ok(())
     }
 
-    fn retire_driver(&mut self, tier_name: &str) {
-        self.revoke_driver(tier_name);
-    }
-
     fn revoke_all_drivers(&mut self) {
         if let Some(runtime) = registered_tier_driver_runtime(self) {
             let mut runtime = lock_unpoisoned(&runtime);
@@ -3884,6 +3873,7 @@ impl TierConfigMgr {
         self.save_config(api, &config_file, data).await
     }
 
+    #[allow(dead_code, reason = "reached only through #[cfg(test)] helpers in this file (backlog#1823)")]
     async fn save_tiering_config_if_current<S>(
         &self,
         api: Arc<S>,

crates/ecstore/src/services/tier/tier_mutation_intent.rs

@@ -305,6 +305,10 @@ impl TierMutationIntent {
     }
 }
 
+#[allow(
+    dead_code,
+    reason = "intent-record persistence asserted by store::init tests (backlog#1823)"
+)]
 pub(crate) fn tier_mutation_intent_record_object_name(mutation_id: Uuid) -> Result<String> {
     tier_mutation_intent_record_object_name_with_prefix(TIER_MUTATION_INTENT_RECORD_PREFIX, mutation_id)
 }
@@ -317,6 +321,10 @@ fn tier_mutation_intent_record_object_name_with_prefix(prefix: &str, mutation_id
     Ok(format!("{}/{}/{}/{}.json", prefix, &mutation_key[..2], &mutation_key[2..4], mutation_key))
 }
 
+#[allow(
+    dead_code,
+    reason = "intent-record persistence asserted by store::init tests (backlog#1823)"
+)]
 pub(crate) fn tier_mutation_intent_id_from_record_object_name(object: &str) -> Result<Uuid> {
     tier_mutation_intent_id_from_record_object_name_with_prefix(TIER_MUTATION_INTENT_RECORD_PREFIX, object)
 }
@@ -355,6 +363,10 @@ fn tier_mutation_intent_id_from_record_object_name_with_prefix(prefix: &str, obj
     Uuid::parse_str(mutation_key).map_err(|_| TierMutationIntentError::Corrupt("intent record path has invalid uuid"))
 }
 
+#[allow(
+    dead_code,
+    reason = "intent-record persistence asserted by store::init tests (backlog#1823)"
+)]
 pub(crate) async fn save_tier_mutation_intent_record<S>(api: Arc<S>, intent: &TierMutationIntent) -> EcstoreResult<()>
 where
     S: EcstoreObjectIO,
@@ -446,6 +458,10 @@ where
     Ok((intent, etag))
 }
 
+#[allow(
+    dead_code,
+    reason = "intent-record persistence asserted by store::init tests (backlog#1823)"
+)]
 pub(crate) async fn save_tier_mutation_intent_record_if_current<S>(
     api: Arc<S>,
     intent: &TierMutationIntent,

crates/ecstore/src/services/tier/warm_backend_gcs.rs

@@ -41,10 +41,7 @@ use crate::services::tier::{
 };
 use tracing::warn;
 
-const MAX_MULTIPART_PUT_OBJECT_SIZE: i64 = 1024 * 1024 * 1024 * 1024 * 5;
-const MAX_PARTS_COUNT: i64 = 10000;
 const _MAX_PART_SIZE: i64 = 1024 * 1024 * 1024 * 5;
-const MIN_PART_SIZE: i64 = 1024 * 1024 * 128;
 
 fn parse_generation(remote_version: &str) -> Result<Option<i64>, Error> {
     if remote_version.is_empty() {
@@ -64,7 +61,6 @@ pub struct WarmBackendGCS {
     pub control: Arc<StorageControl>,
     pub bucket: String,
     pub prefix: String,
-    pub storage_class: String,
 }
 
 impl WarmBackendGCS {
@@ -104,7 +100,6 @@ impl WarmBackendGCS {
             control,
             bucket: conf.bucket.clone(),
             prefix: conf.prefix.strip_suffix("/").unwrap_or(&conf.prefix).to_owned(),
-            storage_class: "".to_string(),
         })
     }
 

crates/ecstore/src/services/tier/warm_backend_s3sdk.rs

@@ -1,200 +0,0 @@
-// Copyright 2024 RustFS Team
-//
-// Licensed under the Apache License, Version 2.0 (the "License");
-// you may not use this file except in compliance with the License.
-// You may obtain a copy of the License at
-//
-//     http://www.apache.org/licenses/LICENSE-2.0
-//
-// Unless required by applicable law or agreed to in writing, software
-// distributed under the License is distributed on an "AS IS" BASIS,
-// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
-// See the License for the specific language governing permissions and
-// limitations under the License.
-#![allow(unused_imports)]
-#![allow(unused_variables)]
-#![allow(unused_mut)]
-#![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
-
-use std::collections::HashMap;
-use std::sync::Arc;
-use url::Url;
-
-use aws_config::meta::region::RegionProviderChain;
-use aws_sdk_s3::Client;
-use aws_sdk_s3::config::{Credentials, Region};
-use aws_sdk_s3::primitives::ByteStream;
-
-use crate::client::{
-    api_get_options::GetObjectOptions,
-    api_put_object::PutObjectOptions,
-    api_remove::RemoveObjectOptions,
-    transition_api::{ReadCloser, ReaderImpl},
-};
-use crate::error::ErrorResponse;
-use crate::error::error_resp_to_object_err;
-use crate::services::tier::{
-    tier_config::TierS3,
-    warm_backend::{WarmBackend, WarmBackendGetOpts},
-};
-
-pub struct WarmBackendS3 {
-    pub client: Arc<Client>,
-    pub bucket: String,
-    pub prefix: String,
-    pub storage_class: String,
-}
-
-impl WarmBackendS3 {
-    pub async fn new(conf: &TierS3, tier: &str) -> Result<Self, std::io::Error> {
-        let u = match Url::parse(&conf.endpoint) {
-            Ok(u) => u,
-            Err(err) => {
-                return Err(std::io::Error::other(err.to_string()));
-            }
-        };
-
-        if conf.aws_role_web_identity_token_file == "" && conf.aws_role_arn != ""
-            || conf.aws_role_web_identity_token_file != "" && conf.aws_role_arn == ""
-        {
-            return Err(std::io::Error::other("both the token file and the role ARN are required"));
-        } else if conf.access_key == "" && conf.secret_key != "" || conf.access_key != "" && conf.secret_key == "" {
-            return Err(std::io::Error::other("both the access and secret keys are required"));
-        } else if conf.aws_role
-            && (conf.aws_role_web_identity_token_file != ""
-                || conf.aws_role_arn != ""
-                || conf.access_key != ""
-                || conf.secret_key != "")
-        {
-            return Err(std::io::Error::other(
-                "AWS Role cannot be activated with static credentials or the web identity token file",
-            ));
-        } else if conf.bucket == "" {
-            return Err(std::io::Error::other("no bucket name was provided"));
-        }
-
-        let creds;
-        if conf.access_key != "" && conf.secret_key != "" {
-            creds = Credentials::new(
-                conf.access_key.clone(), // access_key_id
-                conf.secret_key.clone(), // secret_access_key
-                None,                    // session_token (optional)
-                None,
-                "Static",
-            );
-        } else {
-            return Err(std::io::Error::other("insufficient parameters for S3 backend authentication"));
-        }
-        let region_provider = RegionProviderChain::default_provider().or_else(Region::new(conf.region.clone()));
-        #[allow(deprecated)]
-        let config = aws_config::from_env()
-            .endpoint_url(conf.endpoint.clone())
-            .region(region_provider)
-            .credentials_provider(creds)
-            .load()
-            .await;
-        let client = Client::new(&config);
-        let client = Arc::new(client);
-        Ok(Self {
-            client,
-            bucket: conf.bucket.clone(),
-            prefix: conf.prefix.clone().trim_matches('/').to_string(),
-            storage_class: conf.storage_class.clone(),
-        })
-    }
-
-    pub fn get_dest(&self, object: &str) -> String {
-        let mut dest_obj = object.to_string();
-        if self.prefix != "" {
-            dest_obj = format!("{}/{}", &self.prefix, object);
-        }
-        return dest_obj;
-    }
-}
-
-#[async_trait::async_trait]
-impl WarmBackend for WarmBackendS3 {
-    async fn put_with_meta(
-        &self,
-        object: &str,
-        r: ReaderImpl,
-        length: i64,
-        meta: HashMap<String, String>,
-    ) -> Result<String, std::io::Error> {
-        let client = self.client.clone();
-        let Ok(res) = client
-            .put_object()
-            .bucket(&self.bucket)
-            .key(&self.get_dest(object))
-            .body(match r {
-                ReaderImpl::Body(content_body) => ByteStream::from(content_body.to_vec()),
-                ReaderImpl::ObjectBody(mut content_body) => ByteStream::from(content_body.read_all().await?),
-            })
-            .send()
-            .await
-        else {
-            return Err(std::io::Error::other("put_object error"));
-        };
-
-        Ok(res.version_id().unwrap_or("").to_string())
-    }
-
-    async fn put(&self, object: &str, r: ReaderImpl, length: i64) -> Result<String, std::io::Error> {
-        self.put_with_meta(object, r, length, HashMap::new()).await
-    }
-
-    async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-        let client = self.client.clone();
-        let mut req = client.get_object().bucket(&self.bucket).key(&self.get_dest(object));
-
-        if !rv.is_empty() {
-            req = req.version_id(rv);
-        }
-
-        if opts.start_offset >= 0 && opts.length > 0 {
-            let end = opts
-                .start_offset
-                .checked_add(opts.length)
-                .and_then(|v| v.checked_sub(1))
-                .ok_or_else(|| std::io::Error::other("invalid range: overflow"))?;
-            req = req.range(format!("bytes={}-{}", opts.start_offset, end));
-        }
-
-        let res = req.send().await.map_err(|e| std::io::Error::other(e.to_string()))?;
-
-        Ok(ReadCloser::new(std::io::Cursor::new(
-            res.body.collect().await.map(|data| data.into_bytes().to_vec())?,
-        )))
-    }
-
-    async fn remove(&self, object: &str, rv: &str) -> Result<(), std::io::Error> {
-        let client = self.client.clone();
-        let mut req = client.delete_object().bucket(&self.bucket).key(&self.get_dest(object));
-
-        if !rv.is_empty() {
-            req = req.version_id(rv);
-        }
-
-        req.send().await.map_err(|e| std::io::Error::other(e.to_string()))?;
-
-        Ok(())
-    }
-
-    async fn in_use(&self) -> Result<bool, std::io::Error> {
-        let client = self.client.clone();
-        let Ok(res) = client
-            .list_objects_v2()
-            .bucket(&self.bucket)
-            //.max_keys(10)
-            //.into_paginator()
-            .send()
-            .await
-        else {
-            return Err(std::io::Error::other("list_objects_v2 error"));
-        };
-
-        Ok(res.common_prefixes.unwrap_or_default().len() > 0 || res.contents.unwrap_or_default().len() > 0)
-    }
-}