From 5a0367969aa8102acbc52e3818bd16fe123e779b Mon Sep 17 00:00:00 2001 From: GatewayJ <835269233@qq.com> Date: Tue, 25 Aug 2026 21:21:30 +0800 Subject: [PATCH] fix(replication): retry startup resync lock failures (#6570) --- .../bucket/replication/replication_pool.rs | 49 ++++- rustfs/src/startup_bucket_metadata.rs | 6 +- rustfs/src/startup_services.rs | 2 +- rustfs/src/storage/storage_api.rs | 181 +++++++++++++++++- 4 files changed, 224 insertions(+), 14 deletions(-) diff --git a/crates/ecstore/src/bucket/replication/replication_pool.rs b/crates/ecstore/src/bucket/replication/replication_pool.rs index 9bc1951b5..8807168f8 100644 --- a/crates/ecstore/src/bucket/replication/replication_pool.rs +++ b/crates/ecstore/src/bucket/replication/replication_pool.rs @@ -1862,6 +1862,12 @@ impl ReplicationPool { Ok(status) } + /// Read `resync.bin` directly without consulting or replacing this node's + /// progress cache when the persisted cross-node intent is authoritative. + pub async fn read_durable_bucket_resync_status(&self, bucket: &str) -> Result { + load_bucket_resync_metadata(bucket, self.storage.clone()).await + } + pub async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError> { self.resyncer.cancel(&opts).await; self.resyncer @@ -2803,6 +2809,7 @@ pub trait ReplicationPoolTrait: std::fmt::Debug { async fn persist_mrf_entry(&self, entry: MrfReplicateEntry) -> ReplicationQueueAdmission; async fn resize(&self, priority: ReplicationPriority, max_workers: usize, max_l_workers: usize); async fn get_bucket_resync_status(&self, bucket: &str) -> Result; + async fn read_durable_bucket_resync_status(&self, bucket: &str) -> Result; async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>; async fn cancel_bucket_resync_for_removed_target( self: Arc, @@ -2858,6 +2865,10 @@ impl ReplicationPoolTrait for ReplicationPool { self.get_bucket_resync_status(bucket).await } + async fn read_durable_bucket_resync_status(&self, bucket: &str) -> Result { + self.read_durable_bucket_resync_status(bucket).await + } + async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError> { self.cancel_bucket_resync(opts).await } @@ -3858,12 +3869,16 @@ mod tests { } fn load_resync_test_metadata() -> Vec { + resync_test_metadata_with_status("load-resync-lock", ResyncStatusType::ResyncCompleted) + } + + fn resync_test_metadata_with_status(bucket: &str, resync_status: ResyncStatusType) -> Vec { let mut status = BucketReplicationResyncStatus::new(); status.targets_map.insert( "arn:test".to_string(), TargetReplicationResyncStatus { - bucket: "load-resync-lock".to_string(), - resync_status: ResyncStatusType::ResyncCompleted, + bucket: bucket.to_string(), + resync_status, ..Default::default() }, ); @@ -3897,6 +3912,36 @@ mod tests { }) } + #[tokio::test] + async fn durable_resync_read_observes_cancellation_after_cached_pending_intent() { + let bucket = "startup-cancel-race"; + let shared = empty_resync_shared_state(); + *shared.data.lock().expect("test data lock should not be poisoned") = + resync_test_metadata_with_status(bucket, ResyncStatusType::ResyncPending); + let pool = new_test_replication_pool(Arc::new(LoadResyncNodeStore::new("node-a", shared.clone()))).await; + + let cached = pool + .get_bucket_resync_status(bucket) + .await + .expect("the pending intent should populate the node cache"); + assert_eq!(cached.targets_map["arn:test"].resync_status, ResyncStatusType::ResyncPending); + + *shared.data.lock().expect("test data lock should not be poisoned") = + resync_test_metadata_with_status(bucket, ResyncStatusType::ResyncCanceled); + let stale = pool + .get_bucket_resync_status(bucket) + .await + .expect("the ordinary status read should expose the stale-cache precondition"); + assert_eq!(stale.targets_map["arn:test"].resync_status, ResyncStatusType::ResyncPending); + + let durable = pool + .read_durable_bucket_resync_status(bucket) + .await + .expect("the lock-protected status read should bypass the stale cache"); + assert_eq!(durable.targets_map["arn:test"].resync_status, ResyncStatusType::ResyncCanceled); + assert_eq!(shared.read_count.load(Ordering::SeqCst), 2); + } + async fn hold_resync_runtime_lock( shared: &Arc, bucket: &str, diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index 7c7a39f61..ad98a7f68 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/rustfs/src/startup_bucket_metadata.rs @@ -23,7 +23,7 @@ use std::{ }; use tokio_util::sync::CancellationToken; -pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc) -> Result> { +pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc, ctx: &CancellationToken) -> Result> { let buckets_list = store .list_bucket(&BucketOptions { no_metadata: true, @@ -37,7 +37,7 @@ pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc) - try_migrate_bucket_metadata(store.clone()).await; init_bucket_metadata_sys(store.clone(), buckets.clone()).await; try_migrate_iam_config(store).await; - reconcile_bucket_resync_target_intents(&buckets).await?; + reconcile_bucket_resync_target_intents(&buckets, ctx).await?; Ok(buckets) } @@ -57,7 +57,7 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance try_migrate_iam_config(store.clone()).await; init_bucket_metadata_sys(store, buckets.clone()).await; - reconcile_bucket_resync_target_intents(&buckets).await?; + reconcile_bucket_resync_target_intents(&buckets, &ctx).await?; if let Some(pool) = get_global_replication_pool() { pool.init_resync(ctx, buckets.clone()).await?; diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 8a0292615..d66acee71 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -59,7 +59,7 @@ pub(crate) async fn init_embedded_startup_runtime_services( server_ctx: Arc, ) -> Result { init_embedded_optional_service_runtime(config).await; - let buckets = init_embedded_bucket_metadata_runtime(store.clone()).await?; + let buckets = init_embedded_bucket_metadata_runtime(store.clone(), &ctx).await?; let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness, server_ctx) .await .map_err(|err| std::io::Error::other(format!("IAM bootstrap setup: {err}")))?; diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index cd0e45460..80dd5bed9 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -18,11 +18,19 @@ use std::collections::hash_map::DefaultHasher; use std::hash::{Hash, Hasher}; use std::sync::{Arc, LazyLock}; +use rand::RngExt as _; use rustfs_storage_api as storage_contracts; use tokio::sync::{Mutex, OwnedMutexGuard}; use tokio_util::sync::CancellationToken; const BUCKET_TARGETS_METADATA_LOCK_SHARDS: usize = 256; +const BUCKET_RESYNC_LOCK_RETRY_BASE_MS: u64 = 100; +const BUCKET_RESYNC_LOCK_RETRY_MAX_MS: u64 = 2_000; +const EVENT_REPLICATION_RESYNC_INTENT_ORPHANED: &str = "replication_resync_intent_orphaned"; +const EVENT_REPLICATION_RESYNC_STARTUP_LOCK_RECOVERED: &str = "replication_resync_startup_lock_recovered"; +const EVENT_REPLICATION_RESYNC_STARTUP_LOCK_RETRY: &str = "replication_resync_startup_lock_retry"; +const LOG_COMPONENT_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_REPLICATION: &str = "replication"; static BUCKET_TARGETS_METADATA_LOCKS: LazyLock>>> = LazyLock::new(|| { (0..BUCKET_TARGETS_METADATA_LOCK_SHARDS) .map(|_| Arc::new(Mutex::new(()))) @@ -905,9 +913,9 @@ fn apply_active_resync_intents( } let Some(target) = targets.targets.iter_mut().find(|target| target.arn == *arn) else { tracing::warn!( - event = "replication_resync_intent_orphaned", - component = "storage", - subsystem = "replication", + event = EVENT_REPLICATION_RESYNC_INTENT_ORPHANED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REPLICATION, result = "skipped", bucket, arn = %arn, @@ -925,14 +933,107 @@ fn apply_active_resync_intents( changed } -pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String]) -> Result<()> { +fn bucket_resync_transaction_lock_retry_reason(error: &Error) -> Option<&'static str> { + match error { + Error::Lock(rustfs_lock::LockError::Timeout { .. }) => Some("timeout"), + Error::Lock(rustfs_lock::LockError::Network { .. }) => Some("network"), + Error::Lock(rustfs_lock::LockError::InsufficientNodes { .. }) => Some("insufficient_nodes"), + Error::Lock(rustfs_lock::LockError::QuorumNotReached { .. }) => Some("quorum_not_reached"), + _ => None, + } +} + +fn bucket_resync_transaction_lock_retry_ceiling_ms(attempt: u32) -> u64 { + let shift = attempt.saturating_sub(1).min(5); + BUCKET_RESYNC_LOCK_RETRY_BASE_MS + .saturating_mul(1_u64 << shift) + .min(BUCKET_RESYNC_LOCK_RETRY_MAX_MS) +} + +fn bucket_resync_transaction_lock_retry_delay(attempt: u32) -> std::time::Duration { + let ceiling_ms = bucket_resync_transaction_lock_retry_ceiling_ms(attempt); + std::time::Duration::from_millis(rand::rng().random_range(0..=ceiling_ms)) +} + +async fn retry_bucket_resync_transaction_lock(bucket: &str, shutdown: &CancellationToken, mut acquire: F) -> Result +where + F: FnMut() -> Fut, + Fut: std::future::Future>, +{ + let mut attempt = 0_u32; + loop { + let result = tokio::select! { + biased; + _ = shutdown.cancelled() => return Err(Error::OperationCanceled), + result = acquire() => result, + }; + match result { + Ok(guard) => { + if attempt > 0 { + metrics::counter!("rustfs_replication_resync_startup_lock_recovered_total").increment(1); + tracing::info!( + event = EVENT_REPLICATION_RESYNC_STARTUP_LOCK_RECOVERED, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + result = "acquired", + bucket, + attempts = attempt, + "startup resync reconcile acquired the bucket metadata transaction lock after retry" + ); + } + return Ok(guard); + } + Err(error) => { + let Some(reason) = bucket_resync_transaction_lock_retry_reason(&error) else { + return Err(error); + }; + attempt = attempt.saturating_add(1); + let retry_delay = bucket_resync_transaction_lock_retry_delay(attempt); + metrics::counter!("rustfs_replication_resync_startup_lock_retry_total", "reason" => reason).increment(1); + tracing::warn!( + event = EVENT_REPLICATION_RESYNC_STARTUP_LOCK_RETRY, + component = LOG_COMPONENT_STORAGE, + subsystem = LOG_SUBSYSTEM_REPLICATION, + state = "retrying", + bucket, + attempt, + reason, + retry_delay_ms = retry_delay.as_millis() as u64, + error = %error, + "startup resync reconcile is retrying a transient bucket metadata transaction lock failure" + ); + tokio::select! { + biased; + _ = shutdown.cancelled() => return Err(Error::OperationCanceled), + _ = tokio::time::sleep(retry_delay) => {} + } + } + } + } +} + +async fn acquire_bucket_resync_transaction_lock( + bucket: &str, + shutdown: &CancellationToken, +) -> Result { + retry_bucket_resync_transaction_lock(bucket, shutdown, || { + ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock(bucket) + }) + .await +} + +pub(crate) async fn reconcile_bucket_resync_target_intents(buckets: &[String], shutdown: &CancellationToken) -> Result<()> { let Some(pool) = ecstore_bucket::replication::get_global_replication_pool() else { return Err(Error::other("replication pool is not initialized")); }; for bucket in buckets { - let transaction_guard = ecstore_bucket::metadata_sys::acquire_bucket_metadata_transaction_lock(bucket).await?; - let status = pool.get_bucket_resync_status(bucket).await?; + let status = pool.read_durable_bucket_resync_status(bucket).await?; + if status.targets_map.is_empty() { + continue; + } + let transaction_guard = acquire_bucket_resync_transaction_lock(bucket, shutdown).await?; + let status = pool.read_durable_bucket_resync_status(bucket).await?; if status.targets_map.is_empty() { continue; } @@ -1917,8 +2018,10 @@ pub(crate) async fn init_compression_total_memory_from_backend(store: Arc>).await { + Ok(_) => panic!("cancelled startup must not acquire a transaction lock"), + Err(error) => error, + }; + assert!(matches!(error, super::Error::OperationCanceled)); + } + #[test] fn fresh_instance_context_installs_object_encryption_resolver() { assert!(new_instance_ctx().object_encryption_resolver().is_some());