diff --git a/rustfs/src/connect/offline/bundle_writer.rs b/rustfs/src/connect/offline/bundle_writer.rs index 9456a03d5..e46eb1a43 100644 --- a/rustfs/src/connect/offline/bundle_writer.rs +++ b/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() diff --git a/rustfs/src/startup_bucket_metadata.rs b/rustfs/src/startup_bucket_metadata.rs index ad98a7f68..33e90cb1b 100644 --- a/rustfs/src/startup_bucket_metadata.rs +++ b/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, ctx: &CancellationToken) -> Result> { +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, ctx: &CancellationToken) -> IoResult> { 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 = 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, ctx: CancellationToken) -> Result> { +pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: CancellationToken) -> IoResult> { let buckets_list = store .list_bucket(&BucketOptions { no_metadata: true, ..Default::default() }) .await - .map_err(Error::other)?; + .map_err(IoError::other)?; let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); @@ -57,11 +61,60 @@ 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, &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, 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, + 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" + ))); + } +} diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index 8b1ff9f23..6c12018b6 100644 --- a/rustfs/src/storage_api.rs +++ b/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, }; }