diff --git a/crates/ecstore/src/event/targetlist.rs b/crates/ecstore/src/event/targetlist.rs index b63b786ef..96f4ea91f 100644 --- a/crates/ecstore/src/event/targetlist.rs +++ b/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, } diff --git a/crates/ecstore/src/services/batch_processor.rs b/crates/ecstore/src/services/batch_processor.rs index 61212ec00..006fd1f78 100644 --- a/crates/ecstore/src/services/batch_processor.rs +++ b/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) } diff --git a/crates/ecstore/src/services/event_notification.rs b/crates/ecstore/src/services/event_notification.rs index a5a5c7acd..d9f5267be 100644 --- a/crates/ecstore/src/services/event_notification.rs +++ b/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>, } +#[allow( + dead_code, + reason = "ecstore-side notification skeleton superseded by rustfs-notify; see module note (backlog#1823)" +)] impl EventNotifier { pub fn new() -> Arc> { Arc::new(RwLock::new(Self { diff --git a/crates/ecstore/src/services/mod.rs b/crates/ecstore/src/services/mod.rs index 303ceb94b..6335c441f 100644 --- a/crates/ecstore/src/services/mod.rs +++ b/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; diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 41fc15970..f3dd0a416 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -1623,6 +1623,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() @@ -1796,6 +1797,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( timeout_dur: Duration, host_label: &str, diff --git a/crates/ecstore/src/services/rebalance/meta.rs b/crates/ecstore/src/services/rebalance/meta.rs index 39f2a2da1..2a1ea259b 100644 --- a/crates/ecstore/src/services/rebalance/meta.rs +++ b/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 { let meta = meta?; stop_rebalance_state(meta, now); diff --git a/crates/ecstore/src/services/rebalance/migration.rs b/crates/ecstore/src/services/rebalance/migration.rs index 7e23c9ed1..c3e1e5e69 100644 --- a/crates/ecstore/src/services/rebalance/migration.rs +++ b/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( set: &Backend, bucket: String, diff --git a/crates/ecstore/src/services/rebalance/types.rs b/crates/ecstore/src/services/rebalance/types.rs index c8c8378ea..b43e075b8 100644 --- a/crates/ecstore/src/services/rebalance/types.rs +++ b/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>; - #[derive(Debug, Default)] pub(super) struct RebalanceBucketConfigs { pub(super) bucket_incarnation_id: Option, diff --git a/crates/ecstore/src/services/tier/mod.rs b/crates/ecstore/src/services/tier/mod.rs index acc0bd93b..801c1e3c9 100644 --- a/crates/ecstore/src/services/tier/mod.rs +++ b/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; diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index f6c9114f2..581f1a5a4 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/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 { 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>, - manager: &mut Self, - candidate: &Self, - ) -> std::result::Result { - Self::begin_publish_transition_with_allowed_mutation_blocks(handle, manager, candidate, None) - } - fn begin_publish_transition_with_allowed_mutation_blocks( handle: &Arc>, manager: &mut Self, @@ -2819,14 +2813,6 @@ impl TierConfigMgr { }) } - async fn publish_candidate_inner( - handle: &Arc>, - 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>, 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>, 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( handle: &Arc>, api: Arc, @@ -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( handle: &Arc>, api: Arc, @@ -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>, 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( &self, api: Arc, diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index 28a303e32..94c940653 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/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 { 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 { 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(api: Arc, 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( api: Arc, intent: &TierMutationIntent, diff --git a/crates/ecstore/src/services/tier/warm_backend_gcs.rs b/crates/ecstore/src/services/tier/warm_backend_gcs.rs index ef537002f..35aecfb6d 100644 --- a/crates/ecstore/src/services/tier/warm_backend_gcs.rs +++ b/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, Error> { if remote_version.is_empty() { @@ -64,7 +61,6 @@ pub struct WarmBackendGCS { pub control: Arc, 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(), }) } diff --git a/crates/ecstore/src/services/tier/warm_backend_s3sdk.rs b/crates/ecstore/src/services/tier/warm_backend_s3sdk.rs deleted file mode 100644 index fc1790e0a..000000000 --- a/crates/ecstore/src/services/tier/warm_backend_s3sdk.rs +++ /dev/null @@ -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, - pub bucket: String, - pub prefix: String, - pub storage_class: String, -} - -impl WarmBackendS3 { - pub async fn new(conf: &TierS3, tier: &str) -> Result { - 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, - ) -> Result { - 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 { - self.put_with_meta(object, r, length, HashMap::new()).await - } - - async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result { - 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 { - 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) - } -}