diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 07ca12f86..cb4ac1dd4 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -1010,6 +1010,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let DeleteBucketEncryptionInput { bucket, .. } = req.input; let Some(store) = self.object_store() else { @@ -1025,6 +1026,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "delete bucket encryption", request_context); + let item = sr_bucket_meta_item(bucket.clone(), "sse-config"); if let Err(err) = site_replication_bucket_meta_hook(item).await { warn!(bucket = %bucket, error = ?err, "site replication bucket encryption delete hook failed"); @@ -1038,6 +1041,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let DeleteBucketCorsInput { bucket, .. } = req.input; let Some(store) = self.object_store() else { @@ -1053,6 +1057,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "delete bucket cors", request_context); + let item = sr_bucket_meta_item(bucket.clone(), "cors-config"); if let Err(err) = site_replication_bucket_meta_hook(item).await { warn!(bucket = %bucket, error = ?err, "site replication bucket cors delete hook failed"); @@ -1127,6 +1133,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let DeleteBucketReplicationInput { bucket, .. } = req.input; let Some(store) = self.object_store() else { @@ -1160,6 +1167,8 @@ impl DefaultBucketUsecase { return Err(err); } + notify_bucket_metadata_reload(bucket.clone(), "delete bucket replication", request_context); + let item = sr_bucket_meta_item(bucket.clone(), "replication-config"); if let Err(err) = site_replication_bucket_meta_hook(item).await { warn!(bucket = %bucket, error = ?err, "site replication bucket replication-config delete hook failed"); @@ -1176,12 +1185,15 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let DeleteBucketTaggingInput { bucket, .. } = req.input; metadata_sys::delete(&bucket, BUCKET_TAGGING_CONFIG) .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "delete bucket tagging", request_context); + let item = sr_bucket_meta_item(bucket.clone(), "tags"); if let Err(err) = site_replication_bucket_meta_hook(item).await { warn!(bucket = %bucket, error = ?err, "site replication bucket tagging delete hook failed"); @@ -1615,6 +1627,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketEncryptionInput { bucket, mut server_side_encryption_configuration, @@ -1656,6 +1669,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket encryption", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "sse-config"); item.sse_config = Some( serialize_config(&server_side_encryption_configuration) @@ -1769,6 +1784,7 @@ impl DefaultBucketUsecase { req: S3Request, ) -> S3Result> { let request_region = req.region.clone(); + let request_context = req.extensions.get::().cloned(); let PutBucketNotificationConfigurationInput { bucket, @@ -1792,6 +1808,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context); + let region = resolve_notification_region(self.global_region(), request_region); let notify = current_notify_interface_for_context(self.context.as_deref()); let clear_rules = notify.clear_bucket_notification_rules(&bucket); @@ -1906,6 +1924,7 @@ impl DefaultBucketUsecase { #[instrument(level = "debug", skip(self))] pub async fn execute_put_bucket_cors(&self, req: S3Request) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketCorsInput { bucket, cors_configuration, @@ -1926,6 +1945,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket cors", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "cors-config"); item.cors = Some(serialize_config(&cors_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?); @@ -1940,6 +1961,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketReplicationInput { bucket, replication_configuration, @@ -1962,6 +1984,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket replication", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "replication-config"); item.replication_config = Some( serialize_config(&replication_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?, @@ -2010,6 +2034,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketTaggingInput { bucket, tagging, .. } = req.input; let Some(store) = self.object_store() else { @@ -2027,6 +2052,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket tagging", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "tags"); item.tags = Some(serialize_config(&tagging).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?); if let Err(err) = site_replication_bucket_meta_hook(item).await { @@ -2042,6 +2069,7 @@ impl DefaultBucketUsecase { &self, req: S3Request, ) -> S3Result> { + let request_context = req.extensions.get::().cloned(); let PutBucketVersioningInput { bucket, versioning_configuration, @@ -2056,6 +2084,8 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + notify_bucket_metadata_reload(bucket.clone(), "put bucket versioning", request_context); + let mut item = sr_bucket_meta_item(bucket.clone(), "version-config"); item.versioning = Some( serialize_config(&versioning_configuration).and_then(|bytes| String::from_utf8(bytes).map_err(to_internal_error))?, @@ -2298,13 +2328,25 @@ mod tests { } #[test] - fn bucket_policy_and_public_access_block_changes_notify_peer_metadata_reload() { + fn bucket_metadata_config_changes_notify_peer_metadata_reload() { let source = include_str!("bucket_usecase.rs"); for (method, operation) in [ ("execute_delete_bucket_policy", "delete bucket policy"), ("execute_put_bucket_policy", "put bucket policy"), ("execute_delete_public_access_block", "delete public access block"), ("execute_put_public_access_block", "put public access block"), + ("execute_delete_bucket_lifecycle", "delete bucket lifecycle"), + ("execute_put_bucket_lifecycle_configuration", "put bucket lifecycle"), + ("execute_put_bucket_versioning", "put bucket versioning"), + ("execute_delete_bucket_tagging", "delete bucket tagging"), + ("execute_put_bucket_tagging", "put bucket tagging"), + ("execute_delete_bucket_replication", "delete bucket replication"), + ("execute_put_bucket_replication", "put bucket replication"), + ("execute_delete_bucket_cors", "delete bucket cors"), + ("execute_put_bucket_cors", "put bucket cors"), + ("execute_delete_bucket_encryption", "delete bucket encryption"), + ("execute_put_bucket_encryption", "put bucket encryption"), + ("execute_put_bucket_notification_configuration", "put bucket notification"), ] { let body = usecase_method_source(source, method); assert!(