From 8c9249054f4291653a16b6c40e4db6d9237fb02b Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Fri, 14 Aug 2026 08:12:47 +0800 Subject: [PATCH] chore(ecstore): drop the runtime and error dead_code blankets (#6085) --- crates/ecstore/src/error/mod.rs | 45 +-------------- crates/ecstore/src/runtime/global.rs | 38 ------------- crates/ecstore/src/runtime/instance.rs | 4 ++ crates/ecstore/src/runtime/mod.rs | 1 - crates/ecstore/src/runtime/sources.rs | 56 +++---------------- .../background-services-inventory.md | 2 +- 6 files changed, 16 insertions(+), 130 deletions(-) diff --git a/crates/ecstore/src/error/mod.rs b/crates/ecstore/src/error/mod.rs index 2208f359e..4e8ff791f 100644 --- a/crates/ecstore/src/error/mod.rs +++ b/crates/ecstore/src/error/mod.rs @@ -13,13 +13,12 @@ // limitations under the License. // #730: error taxonomy still exposes compatibility variants while callers move to contracts. -#![allow(dead_code)] use crate::bucket::error::BucketMetadataError; use crate::disk::error::DiskError; use crate::storage_api_contracts::{error::StorageErrorCode, range::HTTPRangeError}; use rustfs_utils::path::decode_dir_object; -use s3s::{S3Error, S3ErrorCode}; +use s3s::S3ErrorCode; pub type Error = StorageError; pub type Result = core::result::Result; @@ -902,6 +901,7 @@ pub fn is_err_decommission_running(err: &Error) -> bool { matches!(err, &StorageError::DecommissionAlreadyRunning) } +#[allow(dead_code, reason = "predicate asserted by this file's tests (backlog#1823)")] pub fn is_err_rebalance_running(err: &Error) -> bool { matches!(err, &StorageError::RebalanceAlreadyRunning) } @@ -910,14 +910,11 @@ pub fn is_err_operation_canceled(err: &Error) -> bool { matches!(err, &StorageError::OperationCanceled) } +#[allow(dead_code, reason = "predicate asserted by this file's tests (backlog#1823)")] pub fn is_err_not_initialized(err: &Error) -> bool { err.to_string().contains("errServerNotInitialized") || err.to_string().contains("ServerNotInitialized") } -pub fn is_err_io(err: &Error) -> bool { - matches!(err, &StorageError::Io(_)) -} - /// Strict "not found" predicate that only matches genuine object/version/volume /// absence errors: `FileNotFound`/`VolumeNotFound`/`FileVersionNotFound`/ /// `ObjectNotFound`/`VersionNotFound`. @@ -1078,21 +1075,9 @@ pub struct GenericError { #[derive(Debug, thiserror::Error, PartialEq, Eq)] pub enum ObjectApiError { - #[error("Operation timed out")] - OperationTimedOut, - - #[error("etag of the object has changed")] - InvalidETag, - #[error("BackendDown")] BackendDown(String), - #[error("Unsupported headers in Metadata")] - UnsupportedMetadata, - - #[error("Method not allowed: {}/{}", .0.bucket, .0.object)] - MethodNotAllowed(GenericError), - #[error("The operation is not valid for the current state of the object {}/{}({})", .0.bucket, .0.object, .0.version_id)] InvalidObjectState(GenericError), } @@ -1175,30 +1160,6 @@ pub fn error_resp_to_object_err(err: ErrorResponse, params: Vec<&str>) -> std::i err } -pub fn storage_to_object_err(err: Error, params: Vec<&str>) -> S3Error { - let storage_err = &err; - let mut bucket: String = "".to_string(); - let mut object: String = "".to_string(); - if !params.is_empty() { - bucket = params[0].to_string(); - } - if params.len() >= 2 { - object = decode_dir_object(params[1]); - } - match storage_err { - StorageError::MethodNotAllowed => S3Error::with_message( - S3ErrorCode::MethodNotAllowed, - ObjectApiError::MethodNotAllowed(GenericError { - bucket, - object, - ..Default::default() - }) - .to_string(), - ), - _ => s3s::S3Error::with_message(S3ErrorCode::Custom("err".into()), err.to_string()), - } -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/ecstore/src/runtime/global.rs b/crates/ecstore/src/runtime/global.rs index f8c0c0100..fcc4411f0 100644 --- a/crates/ecstore/src/runtime/global.rs +++ b/crates/ecstore/src/runtime/global.rs @@ -31,7 +31,6 @@ use std::{ use tokio::sync::{OnceCell, RwLock}; use tokio_util::sync::CancellationToken; use tracing::warn; -use uuid::Uuid; pub const DISK_ASSUME_UNKNOWN_SIZE: u64 = 1 << 30; pub const DISK_MIN_INODES: u64 = 1000; @@ -109,18 +108,6 @@ pub fn set_global_rustfs_port(value: u16) { } } -/// Set the global deployment id -/// -/// # Arguments -/// * `id` - The Uuid to set as the global deployment id -/// -/// # Returns -/// * None -/// -pub fn set_global_deployment_id(id: Uuid) { - current_ctx().set_deployment_id(id); -} - /// Get the global deployment id /// /// # Returns @@ -288,19 +275,6 @@ pub fn get_global_region() -> Option { current_ctx().region() } -/// Initialize the global background services cancellation token -/// -/// # Arguments -/// * `cancel_token` - The CancellationToken instance to set globally -/// -/// # Returns -/// * `Ok(())` if successful -/// * `Err(CancellationToken)` if setting fails -/// -pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> Result<(), CancellationToken> { - current_ctx().init_background_cancel_token(cancel_token) -} - /// Get the global background services cancellation token /// /// # Returns @@ -310,18 +284,6 @@ pub fn get_background_services_cancel_token() -> Option { 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 diff --git a/crates/ecstore/src/runtime/instance.rs b/crates/ecstore/src/runtime/instance.rs index 95ff16ffe..71a1898cf 100644 --- a/crates/ecstore/src/runtime/instance.rs +++ b/crates/ecstore/src/runtime/instance.rs @@ -402,6 +402,10 @@ impl InstanceContext { } #[cfg(test)] + #[allow( + dead_code, + reason = "driven by the tier-delete-journal recovery test behind `--features test-util` (backlog#1823)" + )] pub(crate) fn wake_tier_delete_journal_recovery(&self) { self.tier_delete_journal_recovery_wakeup.notify_one(); } diff --git a/crates/ecstore/src/runtime/mod.rs b/crates/ecstore/src/runtime/mod.rs index 81812cac3..9dd84d401 100644 --- a/crates/ecstore/src/runtime/mod.rs +++ b/crates/ecstore/src/runtime/mod.rs @@ -13,7 +13,6 @@ // limitations under the License. // #730: runtime source migration keeps fallback handles until all owners inject state. -#![allow(dead_code)] pub(crate) mod global; pub(crate) mod instance; diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 6c0b96c3b..26a3e296c 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -38,7 +38,6 @@ use crate::{ set_object_layer, update_erasure_type, }, services::batch_processor::{GlobalBatchProcessors, get_global_processors}, - services::event_notification::EventNotifier, services::notification_sys::{NotificationSys, get_global_notification_sys}, services::tier::tier::TierConfigMgr, store::ECStore, @@ -143,6 +142,10 @@ pub async fn setup_is_erasure_sd() -> bool { is_erasure_sd().await } +#[allow( + dead_code, + reason = "setup-type override used only by tests across this crate (backlog#1823)" +)] pub(crate) async fn current_setup_type() -> SetupType { if setup_is_dist_erasure().await { SetupType::DistErasure @@ -155,6 +158,10 @@ pub(crate) async fn current_setup_type() -> SetupType { } } +#[allow( + dead_code, + reason = "setup-type override used only by tests across this crate (backlog#1823)" +)] pub(crate) async fn set_setup_type(setup_type: SetupType) { update_erasure_type(setup_type).await; } @@ -232,10 +239,6 @@ pub(crate) fn ensure_test_rpc_secret() { let _ = rustfs_credentials::set_global_rpc_secret(TEST_RPC_SECRET.to_owned()); } -pub(crate) fn storage_class_parity(storage_class: Option<&str>) -> Option { - 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()) @@ -328,21 +331,6 @@ pub(crate) fn storage_class_config_snapshot() -> Arc { 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, Option) { - 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); } @@ -410,10 +398,6 @@ pub fn transition_state_handle() -> Arc { crate::runtime::global::current_ctx().transition_state() } -pub(crate) fn event_notifier_handle() -> Arc> { - crate::runtime::global::current_ctx().event_notifier() -} - pub(crate) async fn local_disk_by_path(path: &str) -> Option { local_disk_map_handle().read().await.get(path).cloned().flatten() } @@ -507,30 +491,6 @@ pub(crate) async fn local_disk_set_drive( instance_ctx.local_disk_set_drives().read().await[pool_idx][set_idx][disk_idx].clone() } -pub(crate) async fn local_disk_for_endpoint(endpoint: &Endpoint) -> Option { - 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 { local_disk_map_handle().read().await.keys().cloned().collect() } diff --git a/docs/architecture/background-services-inventory.md b/docs/architecture/background-services-inventory.md index 587934670..d484a1004 100644 --- a/docs/architecture/background-services-inventory.md +++ b/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. | Inventory search found no current startup caller for `create_background_services_cancel_token()`. | +| 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. | | 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. |