fix(replication): retry startup resync lock failures (#6570)

crates/ecstore/src/bucket/replication/replication_pool.rs

@@ -1862,6 +1862,12 @@ impl<S: ReplicationStorage> ReplicationPool<S> {
         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<BucketReplicationResyncStatus, EcstoreError> {
+        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<BucketReplicationResyncStatus, EcstoreError>;
+    async fn read_durable_bucket_resync_status(&self, bucket: &str) -> Result<BucketReplicationResyncStatus, EcstoreError>;
     async fn cancel_bucket_resync(&self, opts: ResyncOpts) -> Result<(), EcstoreError>;
     async fn cancel_bucket_resync_for_removed_target(
         self: Arc<Self>,
@@ -2858,6 +2865,10 @@ impl<S: ReplicationStorage> ReplicationPoolTrait for ReplicationPool<S> {
         self.get_bucket_resync_status(bucket).await
     }
 
+    async fn read_durable_bucket_resync_status(&self, bucket: &str) -> Result<BucketReplicationResyncStatus, EcstoreError> {
+        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<u8> {
+        resync_test_metadata_with_status("load-resync-lock", ResyncStatusType::ResyncCompleted)
+    }
+
+    fn resync_test_metadata_with_status(bucket: &str, resync_status: ResyncStatusType) -> Vec<u8> {
         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<LoadResyncSharedState>,
         bucket: &str,

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<ECStore>) -> Result<Vec<String>> {
+pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, ctx: &CancellationToken) -> Result<Vec<String>> {
     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<ECStore>) -
     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<ECStore>, 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?;

rustfs/src/startup_services.rs

@@ -59,7 +59,7 @@ pub(crate) async fn init_embedded_startup_runtime_services(
     server_ctx: Arc<ServerContextSlot>,
 ) -> Result<EmbeddedStartupServiceRuntime> {
     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}")))?;

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<Vec<Arc<Mutex<()>>>> = 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<T, F, Fut>(bucket: &str, shutdown: &CancellationToken, mut acquire: F) -> Result<T>
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = Result<T>>,
+{
+    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<ecstore_bucket::metadata_sys::BucketMetadataMutationGuard> {
+    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<ECStor
 #[cfg(test)]
 mod tests {
     use super::{
-        apply_active_resync_intents, bucket_targets_metadata_lock_shard, ecstore_bucket, lock_bucket_targets_metadata,
-        new_instance_ctx, scanner_maintenance_config_file,
+        BUCKET_RESYNC_LOCK_RETRY_MAX_MS, apply_active_resync_intents, bucket_resync_transaction_lock_retry_ceiling_ms,
+        bucket_resync_transaction_lock_retry_delay, bucket_resync_transaction_lock_retry_reason,
+        bucket_targets_metadata_lock_shard, ecstore_bucket, lock_bucket_targets_metadata, new_instance_ctx,
+        retry_bucket_resync_transaction_lock, scanner_maintenance_config_file,
     };
     use std::time::Duration;
 
@@ -1949,6 +2052,68 @@ mod tests {
         );
     }
 
+    #[test]
+    fn startup_resync_retries_only_transient_transaction_lock_errors() {
+        let timeout = super::Error::Lock(rustfs_lock::LockError::timeout("bucket", Duration::from_secs(5)));
+        assert_eq!(bucket_resync_transaction_lock_retry_reason(&timeout), Some("timeout"));
+
+        let network = super::Error::Lock(rustfs_lock::LockError::network(
+            "unreachable",
+            std::io::Error::new(std::io::ErrorKind::ConnectionRefused, "unreachable"),
+        ));
+        assert_eq!(bucket_resync_transaction_lock_retry_reason(&network), Some("network"));
+        assert_eq!(
+            bucket_resync_transaction_lock_retry_reason(&super::Error::Lock(rustfs_lock::LockError::QuorumNotReached {
+                required: 3,
+                achieved: 2,
+            },)),
+            Some("quorum_not_reached")
+        );
+        assert_eq!(bucket_resync_transaction_lock_retry_reason(&super::Error::DiskNotFound), None);
+    }
+
+    #[test]
+    fn startup_resync_transaction_lock_backoff_is_capped() {
+        assert_eq!(bucket_resync_transaction_lock_retry_ceiling_ms(1), 100);
+        assert_eq!(bucket_resync_transaction_lock_retry_ceiling_ms(2), 200);
+        assert_eq!(bucket_resync_transaction_lock_retry_ceiling_ms(3), 400);
+        assert_eq!(bucket_resync_transaction_lock_retry_ceiling_ms(8), BUCKET_RESYNC_LOCK_RETRY_MAX_MS);
+        assert_eq!(bucket_resync_transaction_lock_retry_ceiling_ms(u32::MAX), BUCKET_RESYNC_LOCK_RETRY_MAX_MS);
+        assert!(bucket_resync_transaction_lock_retry_delay(8) <= Duration::from_millis(BUCKET_RESYNC_LOCK_RETRY_MAX_MS));
+    }
+
+    #[tokio::test]
+    async fn startup_resync_retries_a_transaction_lock_timeout() {
+        let shutdown = tokio_util::sync::CancellationToken::new();
+        let mut attempts = 0;
+        let guard = retry_bucket_resync_transaction_lock("bucket", &shutdown, || {
+            attempts += 1;
+            std::future::ready(if attempts == 1 {
+                Err(super::Error::Lock(rustfs_lock::LockError::timeout("bucket", Duration::from_secs(5))))
+            } else {
+                Ok("guard")
+            })
+        })
+        .await
+        .expect("startup reconcile must retry a transaction lock timeout");
+
+        assert_eq!(guard, "guard");
+        assert_eq!(attempts, 2);
+    }
+
+    #[tokio::test]
+    async fn startup_resync_lock_wait_honors_shutdown() {
+        let shutdown = tokio_util::sync::CancellationToken::new();
+        shutdown.cancel();
+
+        let error =
+            match retry_bucket_resync_transaction_lock("bucket", &shutdown, std::future::pending::<super::Result<()>>).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());