From 81005b705b51537c19d95d932828f1fcccaa120e Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 10 Sep 2026 03:12:00 +0800 Subject: [PATCH] fix(scanner): bound native backlog fixture resources Keep native scanner backlog restart fixtures within low file descriptor limits by reducing the native-only disk layout, making fixture shutdown drain background work, and avoiding long-lived ECStore retention from background loops. Convert ECStore-backed background refresh/recovery/monitor tasks to upgrade weak owners only while doing work so completed test stores release their disk graph before the next native fixture starts. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 13 +- crates/ecstore/src/bucket/metadata_sys.rs | 104 ++++++++------- crates/ecstore/src/core/sets.rs | 32 +++-- crates/ecstore/src/services/tier/tier.rs | 41 +++++- crates/ecstore/src/store/init.rs | 65 ++++++++-- crates/scanner/src/scanner/backlog.rs | 119 ++++++++++++++---- crates/scanner/src/scanner/tests.rs | 41 +++++- 7 files changed, 316 insertions(+), 99 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 93627d2ab..dfed86927 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -1220,7 +1220,7 @@ impl ExpiryState { while state.tasks_tx.len() < n { let (tx, rx) = mpsc::channel(EXPIRY_WORKER_QUEUE_CAPACITY); - let api = api.clone(); + let api = Arc::downgrade(&api); let rx = Arc::new(tokio::sync::Mutex::new(rx)); let stats = Arc::clone(&state.stats); let recovery_notify = Arc::clone(&state.recovery_notify); @@ -1248,14 +1248,18 @@ impl ExpiryState { async fn worker( rx: &mut Receiver>, - api: Arc, + api: Weak, stats: Arc, recovery_notify: Arc, ) { - let cancel_token = api.ctx.background_cancel_token().unwrap_or_else(|| { + let Some(initial_api) = api.upgrade() else { + return; + }; + let cancel_token = initial_api.ctx.background_cancel_token().unwrap_or_else(|| { static FALLBACK: std::sync::OnceLock = std::sync::OnceLock::new(); FALLBACK.get_or_init(tokio_util::sync::CancellationToken::new).clone() }); + drop(initial_api); loop { select! { @@ -1284,6 +1288,9 @@ impl ExpiryState { let v = v.expect("received None after None check"); stats.decrement_pending_tasks(); let _active_task = ExpiryActiveTask::begin(Arc::clone(&stats)); + let Some(api) = api.upgrade() else { + return; + }; if v.as_any().is::() { let v = v.as_any().downcast_ref::().expect("ExpiryTask downcast failed"); //debug!("lifecycle expiry worker received task: {:?}", v.obj_info); diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 1cdea7e5c..fe376f343 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -248,10 +248,9 @@ fn validate_authoritative_object_lock_config(config: &ObjectLockConfiguration) - } pub async fn init_bucket_metadata_sys(api: Arc, buckets: Vec) { - // The metadata system is inherently per-store (it holds the store handle - // and that store's bucket cache), so it lives on the store's own instance - // context (backlog#1052 S3) — a second instance initializes its own cell - // instead of panicking on the process-global one. + // The metadata system is inherently per-store, so it lives on the store's + // own instance context (backlog#1052 S3). It resolves the store through a + // weak handle so the context cache cannot keep the store and disks alive. let instance_ctx = api.ctx.clone(); let is_dist_erasure = instance_ctx.is_dist_erasure().await; @@ -317,18 +316,22 @@ fn start_refresh_buckets_metadata_loop(sys: Arc>) { warn!("bucket metadata refresh loop skipped because background cancellation token is not initialized"); return; }; + let sys = Arc::downgrade(&sys); tokio::spawn(async move { refresh_buckets_metadata_loop(sys, cancel_token).await; }); } -async fn refresh_buckets_metadata_loop(sys: Arc>, cancel_token: CancellationToken) { +async fn refresh_buckets_metadata_loop(sys: Weak>, cancel_token: CancellationToken) { loop { if !wait_refresh_interval_or_cancel(&cancel_token, BUCKET_METADATA_REFRESH_INTERVAL).await { break; } - refresh_buckets_metadata_once(sys.clone()).await; + let Some(sys) = sys.upgrade() else { + break; + }; + refresh_buckets_metadata_once(sys).await; } } @@ -455,7 +458,7 @@ pub(crate) async fn object_store_in(ctx: &crate::runtime::instance::InstanceCont pub(crate) async fn object_store_if_initialized_in(ctx: &crate::runtime::instance::InstanceContext) -> Option> { let sys = ctx.bucket_metadata_sys().or_else(get_global_bucket_metadata_sys)?; - Some(sys.read().await.api.clone()) + sys.read().await.object_store_if_live() } pub(crate) async fn get_in(ctx: &crate::runtime::instance::InstanceContext, bucket: &str) -> Result> { @@ -475,7 +478,7 @@ pub(crate) async fn get_config_from_disk_with_presence_in( bucket: &str, ) -> Result<(BucketMetadata, bool)> { let sys = bucket_metadata_sys_of(ctx)?; - let api = sys.read().await.api.clone(); + let api = sys.read().await.object_store(); load_bucket_metadata_parse_with_presence(api, bucket, true).await } @@ -738,7 +741,7 @@ pub async fn acquire_scanner_bucket_incarnation_fence( ) -> Result { super::utils::check_valid_bucket_name(bucket)?; let sys = get_bucket_metadata_sys()?; - if expected_owner_id.is_nil() || sys.read().await.api.id != expected_owner_id || expected_incarnation_id.is_nil() { + if expected_owner_id.is_nil() || sys.read().await.object_store().id != expected_owner_id || expected_incarnation_id.is_nil() { return Err(Error::other("scanner bucket incarnation owner does not match")); } acquire_config_write_guard_with_migration(sys, bucket, Some(expected_incarnation_id), false).await @@ -751,7 +754,7 @@ async fn acquire_config_write_guard_with_migration( migrate: bool, ) -> Result { let metadata_sys = sys.read().await.clone(); - let lifecycle_guard = metadata_sys.api.acquire_bucket_lifecycle_read_lock(bucket).await?; + let lifecycle_guard = metadata_sys.object_store().acquire_bucket_lifecycle_read_lock(bucket).await?; // Legacy buckets are migrated while the lifecycle fence prevents a // same-name replacement. The second read under the write transaction is @@ -782,7 +785,7 @@ async fn acquire_config_write_guard_with_migration( "bucket config existence transaction validation", async { match metadata_sys - .api + .object_store() .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default()) .await { @@ -802,7 +805,7 @@ async fn acquire_config_write_guard_with_migration( Some(&transaction_guard), bucket, "bucket config incarnation transaction validation", - load_bucket_incarnation(metadata_sys.api.clone(), bucket), + load_bucket_incarnation(metadata_sys.object_store(), bucket), ), ) .await? @@ -1461,7 +1464,7 @@ pub struct BucketMetadataSys { /// Physically missing names are TTL-bounded to limit memory under bogus /// name floods while avoiding repeated namespace and erasure reads. missing_buckets: moka::future::Cache, - api: Arc, + api: Weak, } impl BucketMetadataSys { @@ -1489,12 +1492,17 @@ impl BucketMetadataSys { .max_capacity(MISSING_BUCKET_MAX_ENTRIES) .time_to_live(MISSING_BUCKET_TTL) .build(), - api, + api: Arc::downgrade(&api), } } pub(crate) fn object_store(&self) -> Arc { - self.api.clone() + self.object_store_if_live() + .expect("bucket metadata object store should still be live") + } + + fn object_store_if_live(&self) -> Option> { + self.api.upgrade() } fn metadata_publish_lock(&self, bucket: &str) -> Arc> { @@ -1549,7 +1557,7 @@ impl BucketMetadataSys { ) -> Result { await_bucket_namespace_operation(Some(namespace_guard), bucket, operation, async { match self - .api + .object_store() .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default()) .await { @@ -1566,7 +1574,7 @@ impl BucketMetadataSys { } async fn init_internal(&self, buckets: Vec) -> Result<()> { let count = self - .api + .object_store() .pools .iter() .map(|pool| pool.disk_set.len()) @@ -1598,7 +1606,7 @@ impl BucketMetadataSys { let mut futures = Vec::new(); for bucket in buckets.iter() { - let api = self.api.clone(); + let api = self.object_store(); let bucket = bucket.clone(); futures.push(async move { sleep(Duration::from_millis(30)).await; @@ -1644,7 +1652,9 @@ impl BucketMetadataSys { let bucket = bucket.clone(); futures.push(async move { sleep(Duration::from_millis(30)).await; - let api = sys.read().await.api.clone(); + let Some(api) = sys.read().await.object_store_if_live() else { + return Ok(()); + }; let namespace_lock = api.new_ns_lock(&bucket, &bucket).await?; let namespace_guard = namespace_lock .get_read_lock(crate::set_disk::get_lock_acquire_timeout()) @@ -1685,7 +1695,7 @@ impl BucketMetadataSys { Some(namespace_guard), bucket, "bucket metadata heal existence check", - self.api.bucket_exists_for_heal(bucket), + self.object_store().bucket_exists_for_heal(bucket), ) .await? { @@ -1709,7 +1719,7 @@ impl BucketMetadataSys { Some(namespace_guard), bucket, "bucket metadata heal", - self.api.heal_bucket( + self.object_store().heal_bucket( bucket, &HealOpts { recreate: true, @@ -1723,7 +1733,7 @@ impl BucketMetadataSys { Some(namespace_guard), bucket, "bucket metadata load", - load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true), + load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true), ) .await?; match mode { @@ -1903,7 +1913,7 @@ impl BucketMetadataSys { // (backlog#1052 S7). Reading from the ambient handle instead made the // read and the write of a single read-modify-write able to target // different instances. - let mut bm = Box::pin(Self::load_bucket_metadata_for_update(self.api.clone(), bucket, parse)).await?; + let mut bm = Box::pin(Self::load_bucket_metadata_for_update(self.object_store(), bucket, parse)).await?; if !bm.bucket_incarnation_sidecar || bm.bucket_incarnation_id != expected_incarnation_id { return Err(Error::BucketNotFound(bucket.to_string())); } @@ -1942,7 +1952,7 @@ impl BucketMetadataSys { where F: FnOnce(&BucketMetadata) -> Result> + Send, { - let mut bm = Box::pin(Self::load_bucket_metadata_for_update(self.api.clone(), bucket, true)).await?; + let mut bm = Box::pin(Self::load_bucket_metadata_for_update(self.object_store(), bucket, true)).await?; if !bm.bucket_incarnation_sidecar || bm.bucket_incarnation_id != expected_incarnation_id { return Err(Error::BucketNotFound(bucket.to_string())); } @@ -1993,7 +2003,7 @@ impl BucketMetadataSys { /// server's metadata never leaks into the ambient (first) instance. pub(crate) async fn persist_and_set(&self, bm: BucketMetadata) -> Result<()> { let mut bm = bm; - bm.save_with_store(self.api.clone()).await?; + bm.save_with_store(self.object_store()).await?; self.set(bm.name.clone(), Arc::new(bm)).await; @@ -2001,8 +2011,8 @@ impl BucketMetadataSys { } async fn persist_new_and_set(&self, mut bm: BucketMetadata) -> Result<()> { - bm.save_with_store(self.api.clone()).await?; - save_bucket_incarnation(self.api.clone(), &bm.name, bm.bucket_incarnation_id).await?; + bm.save_with_store(self.object_store()).await?; + save_bucket_incarnation(self.object_store(), &bm.name, bm.bucket_incarnation_id).await?; bm.bucket_incarnation_sidecar = true; self.set(bm.name.clone(), Arc::new(bm)).await; Ok(()) @@ -2019,7 +2029,7 @@ impl BucketMetadataSys { return Err(Error::other("errInvalidArgument")); } - load_bucket_metadata(self.api.clone(), bucket).await + load_bucket_metadata(self.object_store(), bucket).await } /// Reload persisted metadata under the bucket namespace generation fence. @@ -2033,7 +2043,7 @@ impl BucketMetadataSys { return Err(Error::other("errInvalidArgument")); } - let namespace_lock = self.api.new_ns_lock(bucket, bucket).await?; + let namespace_lock = self.object_store().new_ns_lock(bucket, bucket).await?; let namespace_guard = namespace_lock .get_read_lock(crate::set_disk::get_lock_acquire_timeout()) .await?; @@ -2056,7 +2066,7 @@ impl BucketMetadataSys { Some(namespace_guard), bucket, "peer bucket metadata load", - load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true), + load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true), ) .await?; if !persisted { @@ -2101,11 +2111,11 @@ impl BucketMetadataSys { #[cfg(test)] self.lazy_disk_loads.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let lock = self.api.new_ns_lock(bucket, bucket).await?; + let lock = self.object_store().new_ns_lock(bucket, bucket).await?; let guard = lock.get_read_lock(crate::set_disk::get_lock_acquire_timeout()).await?; #[cfg(test)] if self.lazy_load_lock_probe.load(std::sync::atomic::Ordering::Relaxed) { - let competing = self.api.new_ns_lock(bucket, bucket).await?; + let competing = self.object_store().new_ns_lock(bucket, bucket).await?; assert!( competing.get_write_lock(Duration::from_millis(20)).await.is_err(), "lazy metadata IO must start while the bucket namespace read lock is held" @@ -2115,7 +2125,7 @@ impl BucketMetadataSys { Some(&guard), bucket, "lazy bucket metadata load", - Box::pin(load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true)), + Box::pin(load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true)), ) .await?; @@ -2127,7 +2137,7 @@ impl BucketMetadataSys { bucket, "lazy bucket metadata existence check", Box::pin(async { - self.api + self.object_store() .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default()) .await .map(|_| ()) @@ -2334,13 +2344,13 @@ impl BucketMetadataSys { async fn get_bucket_incarnation_id_from_disk(&self, bucket: &str) -> Result { let transaction_lock = self - .api + .object_store() .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket)) .await?; let _transaction_guard = transaction_lock .get_read_lock(crate::set_disk::get_lock_acquire_timeout()) .await?; - let incarnation_id = load_bucket_incarnation(self.api.clone(), bucket).await?; + let incarnation_id = load_bucket_incarnation(self.object_store(), bucket).await?; if _transaction_guard.is_lock_lost() { return Err(Error::other(format!("bucket incarnation metadata transaction lock was lost: {bucket}"))); } @@ -2376,7 +2386,7 @@ impl BucketMetadataSys { async fn migrate_legacy_metadata(&self, bucket: &str) -> Result { let transaction_lock = self - .api + .object_store() .new_ns_lock(RUSTFS_META_BUCKET, &bucket_metadata_transaction_lock_key(bucket)) .await?; let _transaction_guard = transaction_lock @@ -2401,7 +2411,7 @@ impl BucketMetadataSys { return Err(Error::other(format!("injected Object Lock metadata disk read failure: {bucket}"))); } - let namespace_lock = self.api.new_ns_lock(bucket, bucket).await?; + let namespace_lock = self.object_store().new_ns_lock(bucket, bucket).await?; let namespace_guard = namespace_lock .get_read_lock(crate::set_disk::get_lock_acquire_timeout()) .await?; @@ -2411,7 +2421,7 @@ impl BucketMetadataSys { bucket, "legacy bucket metadata existence check", async { - self.api + self.object_store() .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default()) .await }, @@ -2435,7 +2445,7 @@ impl BucketMetadataSys { Some(&namespace_guard), bucket, "legacy bucket metadata confirmation", - load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true), + load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true), ) .await?; if persisted && !metadata.bucket_incarnation_sidecar && !metadata.bucket_incarnation_id.is_nil() { @@ -2457,20 +2467,20 @@ impl BucketMetadataSys { } #[cfg(test)] if self.legacy_migration_lock_probe.load(std::sync::atomic::Ordering::Relaxed) { - let competing = self.api.new_ns_lock(bucket, bucket).await?; + let competing = self.object_store().new_ns_lock(bucket, bucket).await?; assert!( competing.get_write_lock(Duration::from_millis(20)).await.is_err(), "bucket delete/recreate must not cross the legacy metadata migration fence" ); } - save_bucket_incarnation(self.api.clone(), bucket, metadata.bucket_incarnation_id).await?; + save_bucket_incarnation(self.object_store(), bucket, metadata.bucket_incarnation_id).await?; metadata.bucket_incarnation_sidecar = true; if !persisted { await_bucket_namespace_operation( Some(&namespace_guard), bucket, "legacy bucket metadata migration", - metadata.save_with_store(self.api.clone()), + metadata.save_with_store(self.object_store()), ) .await?; } @@ -2506,7 +2516,7 @@ impl BucketMetadataSys { return Err(Error::other(format!("injected Object Lock metadata disk read failure: {bucket}"))); } - let namespace_lock = self.api.new_ns_lock(bucket, bucket).await?; + let namespace_lock = self.object_store().new_ns_lock(bucket, bucket).await?; let namespace_guard = namespace_lock .get_read_lock(crate::set_disk::get_lock_acquire_timeout()) .await?; @@ -2515,7 +2525,7 @@ impl BucketMetadataSys { bucket, "bucket metadata snapshot existence check", async { - self.api + self.object_store() .get_bucket_info_from_sets(bucket, &crate::storage_api_contracts::bucket::BucketOptions::default()) .await }, @@ -2531,7 +2541,7 @@ impl BucketMetadataSys { Some(&namespace_guard), bucket, "bucket metadata authoritative snapshot", - load_bucket_metadata_parse_with_presence(self.api.clone(), bucket, true), + load_bucket_metadata_parse_with_presence(self.object_store(), bucket, true), ) .await?; if persisted { @@ -3695,7 +3705,7 @@ mod tests { let mut stale = BucketMetadata::new("recreated-bucket"); stale.policy_config_json = b"old-generation".to_vec(); let namespace_lock = sys - .api + .object_store() .new_ns_lock("recreated-bucket", "recreated-bucket") .await .expect("namespace lock should be created"); diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index ab51167e8..7040776c8 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -55,7 +55,7 @@ use rustfs_madmin::heal_commands::HealResultItem; use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash}; use std::{ collections::{HashMap, HashSet}, - sync::Arc, + sync::{Arc, Weak}, }; use tokio::sync::RwLock; use tokio::sync::broadcast::{Receiver, Sender}; @@ -308,10 +308,9 @@ impl Sets { ctx: instance_ctx, }); - let asets = sets.clone(); - let rx1 = rx.resubscribe(); - tokio::spawn(async move { asets.monitor_and_connect_endpoints(rx1).await }); + let weak_sets = Arc::downgrade(&sets); + tokio::spawn(async move { Self::monitor_and_connect_endpoints_task(weak_sets, rx1).await }); Ok(sets) } @@ -326,12 +325,26 @@ impl Sets { &self.ctx } - pub async fn monitor_and_connect_endpoints(&self, mut rx: Receiver<()>) { - tokio::time::sleep(Duration::from_secs(5)).await; + async fn monitor_and_connect_endpoints_task(sets: Weak, mut rx: Receiver<()>) { + let startup_delay = tokio::time::sleep(Duration::from_secs(5)); + tokio::pin!(startup_delay); + + tokio::select! { + _ = &mut startup_delay => {} + _ = rx.recv() => { + warn!("monitor_and_connect_endpoints ctx cancelled"); + return; + } + } info!("start monitor_and_connect_endpoints"); - self.connect_disks().await; + let Some(current) = sets.upgrade() else { + warn!("monitor_and_connect_endpoints exit"); + return; + }; + current.connect_disks().await; + drop(current); // TODO(backlog): make monitor_and_connect interval configurable instead of hardcoded 15s let mut interval = tokio::time::interval(Duration::from_secs(15)); @@ -339,7 +352,10 @@ impl Sets { tokio::select! { _= interval.tick()=>{ // debug!("tick..."); - self.connect_disks().await; + let Some(current) = sets.upgrade() else { + break; + }; + current.connect_disks().await; interval.reset(); }, diff --git a/crates/ecstore/src/services/tier/tier.rs b/crates/ecstore/src/services/tier/tier.rs index 1d5adf39b..f48f58fc0 100644 --- a/crates/ecstore/src/services/tier/tier.rs +++ b/crates/ecstore/src/services/tier/tier.rs @@ -6433,9 +6433,48 @@ impl TierConfigMgr { } pub(crate) async fn refresh_tier_config_handle(handle: Arc>, api: Arc) { - Self::refresh_tier_config_handle_with(handle, api).await; + Self::refresh_tier_config_handle_with_weak(handle, Arc::downgrade(&api)).await; } + async fn refresh_tier_config_handle_with_weak(handle: Arc>, api: Weak) { + // The periodic refresh remains the recovery fallback; committed mutations + // notify this worker so a successful peer commit converges immediately. + let mutation_refresh = Self::mutation_refresh_notifier(&handle).await; + let r = rand::rng().random_range(0.0..1.0); + let rand_interval = || Duration::from_secs((r * 60_f64).round() as u64); + + let refresh_interval = TIER_CFG_REFRESH + rand_interval(); + let mut t = delayed_tier_refresh_interval(refresh_interval); + loop { + select! { + _ = t.tick() => { + let Some(api) = Weak::upgrade(&api) else { + return; + }; + if let Err(err) = Self::reload_handle_with(&handle, api).await { + warn!( + event = EVENT_TIER_CONFIG_REFRESH, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_TIER, + trigger = "periodic", + result = "failed", + error = ?err, + "tier configuration refresh" + ); + } + } + _ = mutation_refresh.notified() => { + let Some(api) = Weak::upgrade(&api) else { + return; + }; + Self::reload_after_committed_mutation(&handle, api).await; + } + } + t.reset(); + } + } + + #[allow(dead_code, reason = "used by focused tier refresh tests and non-ECStore generic harnesses")] pub(crate) async fn refresh_tier_config_handle_with(handle: Arc>, api: Arc) where S: EcstoreObjectIO diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index b431bc4a8..238c43649 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -22,7 +22,9 @@ use crate::runtime::sources as runtime_sources; use crate::storage_api_contracts::object::EcstoreObjectIO; use rustfs_config::server_config::KVS; use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token}; +#[cfg(test)] use std::future::Future; +use std::sync::Weak; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_ECSTORE: &str = "ecstore"; @@ -240,6 +242,7 @@ where Ok(committed) } +#[cfg(test)] async fn run_local_decommission_watchdog(rx: CancellationToken, mut reconcile: F) where F: FnMut() -> Fut, @@ -294,16 +297,45 @@ async fn reconcile_local_decommission_after_init(store: &Arc, rx: Cance store.spawn_missing_local_decommission_routines_with_token(rx).await } -async fn supervise_local_decommission_after_init(store: Arc, rx: CancellationToken) { - run_local_decommission_watchdog(rx.clone(), || { - let store = store.clone(); - let worker_rx = rx.clone(); - async move { reconcile_local_decommission_after_init(&store, worker_rx).await } - }) - .await; +async fn supervise_local_decommission_after_init(store: Weak, rx: CancellationToken) { + let mut consecutive_failures = 0u32; + loop { + if rx.is_cancelled() { + return; + } + + let Some(store) = store.upgrade() else { + return; + }; + let delay = match reconcile_local_decommission_after_init(&store, rx.clone()).await { + Ok(()) => { + consecutive_failures = 0; + LOCAL_DECOMMISSION_WATCHDOG_INTERVAL + } + Err(err) => { + consecutive_failures = consecutive_failures.saturating_add(1); + let retry_delay = local_decommission_watchdog_retry_delay(consecutive_failures); + warn!( + event = EVENT_DECOMMISSION_RESUME_RETRY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_STORE_INIT, + consecutive_failures, + retry_delay_secs = retry_delay.as_secs(), + error = %err, + "Retrying decommission worker recovery" + ); + retry_delay + } + }; + drop(store); + + if !wait_for_local_decommission_resume_delay(&rx, delay).await { + return; + } + } } -async fn resume_rebalance_after_init(store: Arc, rx: CancellationToken) { +async fn resume_rebalance_after_init(store: Weak, rx: CancellationToken) { if !wait_for_rebalance_resume_delay(&rx, REBALANCE_INITIAL_RESUME_DELAY).await { return; } @@ -313,6 +345,9 @@ async fn resume_rebalance_after_init(store: Arc, rx: CancellationToken) return; } + let Some(store) = store.upgrade() else { + return; + }; let resume_required = store .rebalance_meta .read() @@ -327,6 +362,7 @@ async fn resume_rebalance_after_init(store: Arc, rx: CancellationToken) store.ctx.is_dist_erasure().await, crate::services::notification_sys::acquire_cross_pool_fence_fleet_proof().is_some(), ) { + drop(store); if !wait_for_rebalance_resume_retry(&rx).await { return; } @@ -791,10 +827,10 @@ impl ECStore { if has_local_decommission_leadership { // The watchdog checks recovery safety and retries transient failures. // Resume persisted work without an unconditional cold-start delay. - tokio::spawn(supervise_local_decommission_after_init(self.clone(), rx.clone())); + tokio::spawn(supervise_local_decommission_after_init(Arc::downgrade(self), rx.clone())); } - let recovery_store = self.clone(); + let recovery_store = Arc::downgrade(self); let recovery_rx = rx.clone(); tokio::spawn(async move { let mut delay = std::time::Duration::from_secs(5); @@ -803,9 +839,12 @@ impl ECStore { _ = recovery_rx.cancelled() => return, _ = tokio::time::sleep(delay) => {} } + let Some(store) = recovery_store.upgrade() else { + return; + }; let result = tokio::select! { _ = recovery_rx.cancelled() => return, - result = tokio::time::timeout(std::time::Duration::from_secs(30), recovery_store.recover_pool_meta_transaction()) => result, + result = tokio::time::timeout(std::time::Duration::from_secs(30), store.recover_pool_meta_transaction()) => result, }; delay = match result { Ok(Ok(_)) => std::time::Duration::from_secs(5), @@ -814,7 +853,7 @@ impl ECStore { Ok(Err(error)) => error, _ => Error::Timeout, }; - recovery_store.record_pool_meta_recovery_failure(error); + store.record_pool_meta_recovery_failure(error); (delay * 2).min(std::time::Duration::from_secs(60)) } }; @@ -835,7 +874,7 @@ impl ECStore { } if rebalance_auto_start_deferred { - let store = self.clone(); + let store = Arc::downgrade(self); tokio::spawn(resume_rebalance_after_init(store, rx)); } diff --git a/crates/scanner/src/scanner/backlog.rs b/crates/scanner/src/scanner/backlog.rs index 3f2e10d85..3248a37f3 100644 --- a/crates/scanner/src/scanner/backlog.rs +++ b/crates/scanner/src/scanner/backlog.rs @@ -1798,6 +1798,8 @@ pub(super) fn scanner_pause_backlog_now() -> u64 { mod tests { use super::*; + const NATIVE_RETIREMENT_DRIVES_PER_SET: usize = 2; + fn run_native_retirement_test(case: C) where C: FnOnce() -> F + Send + 'static, @@ -1825,10 +1827,29 @@ mod tests { async fn native_retirement_store() -> (tempfile::TempDir, Arc) { register_scanner_pause_backlog_retirement(); let root = tempfile::tempdir().expect("native retirement fixture directory"); - let store = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, 3, 2).await; + let store = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + 3, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; (root, store) } + async fn shutdown_native_retirement_store(store: Arc) { + if let Some(token) = store.background_cancel_token() { + token.cancel(); + } + drop(store); + for _ in 0..8 { + tokio::task::yield_now().await; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + async fn native_replica_bytes(set: &SetDisks) -> (Vec, String) { let mut reader = set .get_object_reader( @@ -1871,7 +1892,15 @@ mod tests { register_scanner_pause_backlog_retirement(); let root = tempfile::tempdir().unwrap(); - let old = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, old_pool_count, 2).await; + let old = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + old_pool_count, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; let fault = unstable_source .then(|| NativeScannerPauseBacklogWriteFault::fail_before_write(Arc::clone(&old.pools[0].disk_set[0]), "publish", 2)); let now = unix_now(); @@ -1889,10 +1918,14 @@ mod tests { } else { controller.observe(observation(now + 1, true, 4)).await; controller.observe(observation(now + 2, false, 4)).await; - assert!(matches!( - controller.begin_attempt(now + 2).await, - ScannerPauseBacklogAttemptDecision::Tracked(_) - )); + let decision = controller.begin_attempt(now + 2).await; + assert!( + matches!(decision, ScannerPauseBacklogAttemptDecision::Tracked(_)), + "expected tracked native expansion setup attempt, got {decision:?}; ledger={:?}; persistence_disabled={}; runtime_error={:?}", + controller.loaded.ledger, + controller.persistence_disabled, + runtime_error() + ); assert!(controller.loaded.ledger.has_unfinished_attempt()); } let original = controller.loaded.ledger.clone(); @@ -1902,9 +1935,16 @@ mod tests { old.pool_meta_write_status() .await .expect("healthy pool metadata keeps the background recovery loop read-only"); - old.background_cancel_token().expect("old store shutdown token").cancel(); - drop(old); - let expanded = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, 3, 2).await; + shutdown_native_retirement_store(old).await; + let expanded = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + 3, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; for pool in expanded.pools.iter().skip(old_pool_count) { for set in &pool.disk_set { let replica = read_scanner_pause_backlog_replica(Arc::clone(set)).await; @@ -1999,12 +2039,16 @@ mod tests { } store.pool_meta_write_status().await.expect("healthy metadata before restart"); - store - .background_cancel_token() - .expect("expanded store shutdown token") - .cancel(); - drop(store); - let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, 3, 2).await; + shutdown_native_retirement_store(store).await; + let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + 3, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; let reloaded = load_scanner_pause_backlog(Arc::clone(&restarted)) .await .expect("a new store must recover from disk without the failed controller"); @@ -2021,6 +2065,8 @@ mod tests { assert_eq!(retry_ledger.current_attempt_serial, original.current_attempt_serial); assert_eq!(retry_ledger.last_finished_attempt_serial, original.current_attempt_serial); assert_eq!(retry_ledger.consecutive_failures, original.consecutive_failures + 1); + drop(retried); + shutdown_native_retirement_store(restarted).await; } async fn assert_current_native_writer_ledger(store: &Arc, expected: &ScannerPauseBacklogLedger) { @@ -2124,6 +2170,8 @@ mod tests { .await .expect("retry must seed, commit and stabilize the stale member"); assert_current_native_writer_ledger(&store, &expected).await; + drop(target); + shutdown_native_retirement_store(store).await; }); } @@ -2175,6 +2223,8 @@ mod tests { .await .expect("a fresh caller may finish the seeded membership transition"); assert_current_native_writer_ledger(&store, &expected).await; + drop(pool_meta_lock); + shutdown_native_retirement_store(store).await; }); } @@ -2235,6 +2285,7 @@ mod tests { .await .expect("retry must seed the newly visible members before committing"); assert_current_native_writer_ledger(&store, &expected).await; + shutdown_native_retirement_store(store).await; }); } @@ -2292,6 +2343,7 @@ mod tests { assert_current_native_ledger(&store, &original).await; } assert!(original.has_unfinished_attempt()); + shutdown_native_retirement_store(store).await; } }); } @@ -2337,14 +2389,22 @@ mod tests { .pool_meta_write_status() .await .expect("healthy pool metadata keeps the background recovery loop read-only"); - store.background_cancel_token().expect("old store shutdown token").cancel(); - drop(store); - let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, 3, 2).await; + shutdown_native_retirement_store(store).await; + let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + 3, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; for set_index in 0..2 { restarted.retire_scanner_pause_backlog_for_test(0, set_index).await.unwrap(); assert_native_source_missing(&restarted, set_index).await; } assert_current_native_ledger(&restarted, &original).await; + shutdown_native_retirement_store(restarted).await; } }); } @@ -2384,9 +2444,16 @@ mod tests { .pool_meta_write_status() .await .expect("healthy pool metadata keeps the background recovery loop read-only"); - store.background_cancel_token().expect("old store shutdown token").cancel(); - drop(store); - let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_sets(root.path(), false, 3, 2).await; + shutdown_native_retirement_store(store).await; + let restarted = super::super::tests::setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root.path(), + false, + 3, + 2, + NATIVE_RETIREMENT_DRIVES_PER_SET, + false, + ) + .await; for set_index in 0..2 { restarted.retire_scanner_pause_backlog_for_test(0, set_index).await.unwrap(); assert_native_source_missing(&restarted, set_index).await; @@ -2396,6 +2463,7 @@ mod tests { assert_eq!(bootstrapped.ledger.generation, 1); assert!(bootstrapped.ledger.last_updated_at_unix_secs < future_now); assert_current_native_ledger(&restarted, &bootstrapped.ledger).await; + shutdown_native_retirement_store(restarted).await; }); } @@ -2441,6 +2509,8 @@ mod tests { assert_current_native_ledger(&store, &original).await; store.retire_scanner_pause_backlog_for_test(0, 1).await.unwrap(); assert_native_source_missing(&store, 1).await; + drop(pool_meta_lock); + shutdown_native_retirement_store(store).await; }); } @@ -2536,6 +2606,9 @@ mod tests { "retirement never copied a stale source record" ); } + drop(pool_meta_lock); + drop(barrier); + shutdown_native_retirement_store(store).await; }); } @@ -2669,11 +2742,12 @@ mod tests { .retire_scanner_pause_backlog_for_test(0, 1) .await .expect("the remaining source set follows the same native proof"); - let after = load_scanner_pause_backlog(store) + let after = load_scanner_pause_backlog(Arc::clone(&store)) .await .expect("native restart selection after handoff"); assert_eq!(after.ledger, new_ledger); assert!(after.durable && after.stable_matches_ledger); + shutdown_native_retirement_store(store).await; }); } @@ -2745,6 +2819,7 @@ mod tests { .expect("retained target intent snapshot") }; assert_eq!(after, before, "native cleanup never clears or estimates a target mutation intent"); + shutdown_native_retirement_store(store).await; }); } diff --git a/crates/scanner/src/scanner/tests.rs b/crates/scanner/src/scanner/tests.rs index 35805d5d7..5c9dae5be 100644 --- a/crates/scanner/src/scanner/tests.rs +++ b/crates/scanner/src/scanner/tests.rs @@ -69,13 +69,42 @@ pub(super) async fn setup_scanner_cycle_store_at_path_with_sets( seed_usage_baseline: bool, pool_count: usize, sets_per_pool: usize, +) -> Arc { + setup_scanner_cycle_store_at_path_with_layout(root, seed_usage_baseline, pool_count, sets_per_pool, 4).await +} + +pub(super) async fn setup_scanner_cycle_store_at_path_with_layout( + root: &Path, + seed_usage_baseline: bool, + pool_count: usize, + sets_per_pool: usize, + drives_per_set: usize, +) -> Arc { + setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root, + seed_usage_baseline, + pool_count, + sets_per_pool, + drives_per_set, + true, + ) + .await +} + +pub(super) async fn setup_scanner_cycle_store_at_path_with_layout_and_disk_preinit( + root: &Path, + seed_usage_baseline: bool, + pool_count: usize, + sets_per_pool: usize, + drives_per_set: usize, + preinitialize_disks: bool, ) -> Arc { init_ecstore_config_for_scanner_tests(); let mut pools = Vec::with_capacity(pool_count); for pool_index in 0..pool_count { let mut endpoints = Vec::new(); for set_index in 0..sets_per_pool { - for disk_index in 0..4 { + for disk_index in 0..drives_per_set { let disk_path = if sets_per_pool == 1 { root.join(format!("pool{pool_index}/disk{disk_index}")) } else { @@ -95,7 +124,7 @@ pub(super) async fn setup_scanner_cycle_store_at_path_with_sets( pools.push(PoolEndpoints { legacy: false, set_count: sets_per_pool, - drives_per_set: 4, + drives_per_set, endpoints: Endpoints::from(endpoints), cmd_line: if pool_count == 1 && sets_per_pool == 1 { "scanner-cycle-metrics".to_string() @@ -108,9 +137,11 @@ pub(super) async fn setup_scanner_cycle_store_at_path_with_sets( let endpoint_pools = EndpointServerPools::from(pools); let instance_ctx = Arc::new(InstanceContext::new()); instance_ctx.set_endpoints(endpoint_pools.clone()); - init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) - .await - .expect("scanner cycle test disks should initialize"); + if preinitialize_disks { + init_local_disks_with_instance_ctx(&instance_ctx, endpoint_pools.clone()) + .await + .expect("scanner cycle test disks should initialize"); + } let store = ECStore::new_with_instance_ctx( "127.0.0.1:0".parse().expect("test address should parse"), endpoint_pools,