fix(startup): avoid blocking on resync reconcile (#6593)

Run replication resync target reconcile and follow-up resync recovery in a background startup task so bucket metadata transaction lock contention cannot keep a node from joining the cluster.

Co-authored-by: heihutu <heihutu@gmail.com>

rustfs/src/connect/offline/bundle_writer.rs

@@ -335,7 +335,7 @@ fn valid_payload(collector: OfflineCollector, value: &serde_json::Value) -> bool
                 && object.get("totalBytes").and_then(serde_json::Value::as_u64).is_some()
                 && object.get("underPressure").and_then(serde_json::Value::as_bool).is_some()
         }),
-        OfflineCollector::FilesystemSummary => value.as_array().is_some_and(|values| ordered_strings(values)),
+        OfflineCollector::FilesystemSummary => value.as_array().is_some_and(|values| ordered_strings(values.as_slice())),
         OfflineCollector::NetworkSummary => value.as_object().is_some_and(|object| {
             object.len() == 2
                 && object.get("bondCount").and_then(serde_json::Value::as_u64).is_some()

rustfs/src/startup_bucket_metadata.rs

@@ -14,42 +14,46 @@
 
 use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions};
 use crate::storage_api::startup::bucket_metadata::{
-    ECStore, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents,
-    try_migrate_bucket_metadata, try_migrate_iam_config,
+    ECStore, Error as StorageError, Result as StorageResult, get_global_replication_pool, init_bucket_metadata_sys,
+    reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,
 };
 use std::{
-    io::{Error, Result},
+    io::{Error as IoError, Result as IoResult},
     sync::Arc,
 };
 use tokio_util::sync::CancellationToken;
 
-pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, ctx: &CancellationToken) -> Result<Vec<String>> {
+const EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_FAILED: &str = "replication_resync_startup_background_failed";
+const LOG_COMPONENT_STARTUP_BUCKET_METADATA: &str = "startup_bucket_metadata";
+const LOG_SUBSYSTEM_REPLICATION: &str = "replication";
+
+pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc<ECStore>, ctx: &CancellationToken) -> IoResult<Vec<String>> {
     let buckets_list = store
         .list_bucket(&BucketOptions {
             no_metadata: true,
             ..Default::default()
         })
         .await
-        .map_err(|err| Error::other(format!("list_bucket: {err}")))?;
+        .map_err(|err| IoError::other(format!("list_bucket: {err}")))?;
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
     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, ctx).await?;
+    spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx.clone(), false);
 
     Ok(buckets)
 }
 
-pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: CancellationToken) -> Result<Vec<String>> {
+pub(crate) async fn init_bucket_metadata_runtime(store: Arc<ECStore>, ctx: CancellationToken) -> IoResult<Vec<String>> {
     let buckets_list = store
         .list_bucket(&BucketOptions {
             no_metadata: true,
             ..Default::default()
         })
         .await
-        .map_err(Error::other)?;
+        .map_err(IoError::other)?;
 
     let buckets: Vec<String> = buckets_list.into_iter().map(|v| v.name).collect();
 
@@ -57,11 +61,60 @@ 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, &ctx).await?;
-
-    if let Some(pool) = get_global_replication_pool() {
-        pool.init_resync(ctx, buckets.clone()).await?;
-    }
+    spawn_bucket_resync_startup_reconcile(buckets.clone(), ctx, true);
 
     Ok(buckets)
 }
+
+fn spawn_bucket_resync_startup_reconcile(buckets: Vec<String>, ctx: CancellationToken, init_resync_after_reconcile: bool) {
+    tokio::spawn(async move {
+        if let Err(error) = run_bucket_resync_startup_reconcile(buckets, ctx, init_resync_after_reconcile).await {
+            if !report_bucket_resync_startup_background_error(&error) {
+                return;
+            }
+            tracing::error!(
+                event = EVENT_REPLICATION_RESYNC_STARTUP_BACKGROUND_FAILED,
+                component = LOG_COMPONENT_STARTUP_BUCKET_METADATA,
+                subsystem = LOG_SUBSYSTEM_REPLICATION,
+                result = "failed",
+                init_resync_after_reconcile,
+                error = %error,
+                "Bucket metadata startup resync reconcile failed in background"
+            );
+        }
+    });
+}
+
+async fn run_bucket_resync_startup_reconcile(
+    buckets: Vec<String>,
+    ctx: CancellationToken,
+    init_resync_after_reconcile: bool,
+) -> StorageResult<()> {
+    reconcile_bucket_resync_target_intents(&buckets, &ctx).await?;
+
+    if init_resync_after_reconcile {
+        let Some(pool) = get_global_replication_pool() else {
+            return Err(StorageError::other("replication pool is not initialized"));
+        };
+        pool.init_resync(ctx, buckets).await?;
+    }
+
+    Ok(())
+}
+
+fn report_bucket_resync_startup_background_error(error: &StorageError) -> bool {
+    !matches!(error, StorageError::OperationCanceled)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn startup_resync_background_error_reporting_skips_shutdown() {
+        assert!(!report_bucket_resync_startup_background_error(&StorageError::OperationCanceled));
+        assert!(report_bucket_resync_startup_background_error(&StorageError::other(
+            "replication pool is not initialized"
+        )));
+    }
+}

rustfs/src/storage_api.rs

@@ -227,8 +227,8 @@ pub(crate) mod startup {
         }
 
         pub(crate) use crate::storage::storage_api::{
-            ECStore, get_global_replication_pool, init_bucket_metadata_sys, reconcile_bucket_resync_target_intents,
-            try_migrate_bucket_metadata, try_migrate_iam_config,
+            ECStore, Error, Result, get_global_replication_pool, init_bucket_metadata_sys,
+            reconcile_bucket_resync_target_intents, try_migrate_bucket_metadata, try_migrate_iam_config,
         };
     }