diff --git a/crates/ecstore/src/notification_sys.rs b/crates/ecstore/src/notification_sys.rs index 4f2488663..630bc89e1 100644 --- a/crates/ecstore/src/notification_sys.rs +++ b/crates/ecstore/src/notification_sys.rs @@ -485,31 +485,29 @@ impl NotificationSys { Ok(()) } - pub async fn load_bucket_metadata(&self, bucket: &str) -> Vec { + pub async fn load_bucket_metadata(&self, bucket: &str) -> Result<()> { + let operation = format!("load_bucket_metadata({bucket})"); + let mut failures = Vec::new(); let mut futures = Vec::with_capacity(self.peer_clients.len()); - for client in self.peer_clients.iter() { - let b = bucket.to_string(); - futures.push(async move { - if let Some(client) = client { - match client.load_bucket_metadata(&b).await { - Ok(_) => NotificationPeerErr { - host: client.host.to_string(), - err: None, - }, - Err(e) => NotificationPeerErr { - host: client.host.to_string(), - err: Some(e), - }, - } - } else { - NotificationPeerErr { - host: "".to_string(), - err: Some(Error::other("peer is not reachable")), - } - } - }); + for (idx, client) in self.peer_clients.iter().enumerate() { + if let Some(client) = client { + let host = client.host.to_string(); + let b = bucket.to_string(); + futures.push(async move { client.load_bucket_metadata(&b).await.map_err(|err| (host, err)) }); + } else { + failures.push(format!("peer[{idx}] {operation} failed: peer is not reachable")); + } } - join_all(futures).await + + for result in join_all(futures).await { + if let Err((host, err)) = result { + let failure = format!("peer {host} {operation} failed: {err}"); + error!("notification {operation} err {failure}"); + failures.push(failure); + } + } + + aggregate_notification_failures(&operation, failures) } pub async fn delete_bucket_metadata(&self, bucket: &str) -> Vec { @@ -904,4 +902,22 @@ mod tests { assert!(msg.contains("peer-1 failed")); assert!(msg.contains("local save failed")); } + + #[tokio::test] + async fn load_bucket_metadata_reports_unreachable_peers() { + let sys = NotificationSys { + peer_clients: vec![None], + all_peer_clients: Vec::new(), + }; + + let err = sys + .load_bucket_metadata("bucket-a") + .await + .expect_err("unreachable peers should fail bucket metadata reload"); + + let msg = err.to_string(); + assert!(msg.contains("load_bucket_metadata(bucket-a)")); + assert!(msg.contains("1 failure(s)")); + assert!(msg.contains("peer[0]")); + } } diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 68f05fdcb..ab5294d42 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -54,6 +54,7 @@ use rustfs_ecstore::bucket::{ use rustfs_ecstore::client::object_api_utils::to_s3s_etag; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::new_object_layer_fn; +use rustfs_ecstore::notification_sys::get_global_notification_sys; use rustfs_ecstore::store_api::{ BucketOperations, BucketOptions, DeleteBucketOptions, ListObjectVersionsInfo, ListObjectsV2Info, ListOperations, MakeBucketOptions, ObjectInfo, @@ -194,6 +195,20 @@ fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta { } } +fn notify_bucket_metadata_reload( + bucket: String, + operation: &'static str, + request_context: Option, +) { + spawn_background_with_context(request_context, async move { + if let Some(notification_sys) = get_global_notification_sys() + && let Err(err) = notification_sys.load_bucket_metadata(&bucket).await + { + warn!(bucket = %bucket, error = %err, "failed to notify peers after {operation}"); + } + }); +} + fn replication_target_arns(config: &ReplicationConfiguration) -> HashSet { let mut arns = HashSet::new(); @@ -977,6 +992,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let DeleteBucketLifecycleInput { bucket, .. } = req.input; let Some(store) = new_object_layer_fn() else { @@ -992,6 +1008,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "delete bucket lifecycle", request_context); + let item = sr_bucket_meta_item(bucket.clone(), "lc-config"); if let Err(err) = site_replication_bucket_meta_hook(item).await { warn!(bucket = %bucket, error = ?err, "site replication bucket lifecycle delete hook failed"); @@ -1508,6 +1526,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketLifecycleConfigurationInput { bucket, lifecycle_configuration, @@ -1543,6 +1562,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket lifecycle", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "lc-config"); item.expiry_lc_config = Some(serialize_config(&input_cfg).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?);