fix: reload bucket metadata after lifecycle updates (#2822)

Co-authored-by: houseme <housemecn@gmail.com>

crates/ecstore/src/notification_sys.rs

@@ -485,31 +485,29 @@ impl NotificationSys {
         Ok(())
     }
 
-    pub async fn load_bucket_metadata(&self, bucket: &str) -> Vec<NotificationPeerErr> {
+    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<NotificationPeerErr> {
@@ -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]"));
+    }
 }

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<request_context::RequestContext>,
+) {
+    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<String> {
     let mut arns = HashSet::new();
 
@@ -977,6 +992,7 @@ impl DefaultBucketUsecase {
         &self,
         req: S3Request<DeleteBucketLifecycleInput>,
     ) -> S3Result<S3Response<DeleteBucketLifecycleOutput>> {
+        let request_context = req.extensions.get::<request_context::RequestContext>().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<PutBucketLifecycleConfigurationInput>,
     ) -> S3Result<S3Response<PutBucketLifecycleConfigurationOutput>> {
+        let request_context = req.extensions.get::<request_context::RequestContext>().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))?);