diff --git a/crates/notify/src/bucket_config_manager.rs b/crates/notify/src/bucket_config_manager.rs index 1de84b4ea..811a34783 100644 --- a/crates/notify/src/bucket_config_manager.rs +++ b/crates/notify/src/bucket_config_manager.rs @@ -53,7 +53,9 @@ impl NotifyBucketConfigManager { self.rule_engine.has_subscriber(bucket, event).await } - pub async fn load_bucket_notification_config( + /// Validates that the runtime can accept `cfg` for `bucket` without + /// mutating any state. + pub async fn validate_bucket_notification_config( &self, bucket: &str, cfg: &BucketNotificationConfig, @@ -98,6 +100,11 @@ impl NotifyBucketConfigManager { ); } + Ok(()) + } + + /// Publishes an already validated `cfg` to the runtime rule state. + pub async fn apply_bucket_notification_config(&self, bucket: &str, cfg: &BucketNotificationConfig) { self.subscriber_view.apply_bucket_config(bucket, cfg); self.rule_engine.set_bucket_rules(bucket, cfg.get_rules_map().clone()).await; info!( @@ -109,6 +116,15 @@ impl NotifyBucketConfigManager { rule_count = cfg.get_rules_map().inner().len(), "notify bucket config state" ); + } + + pub async fn load_bucket_notification_config( + &self, + bucket: &str, + cfg: &BucketNotificationConfig, + ) -> Result<(), NotificationError> { + self.validate_bucket_notification_config(bucket, cfg).await?; + self.apply_bucket_notification_config(bucket, cfg).await; Ok(()) } @@ -122,7 +138,7 @@ impl NotifyBucketConfigManager { mod tests { use super::NotifyBucketConfigManager; use crate::{ - BucketNotificationConfig, integration::NotificationMetrics, + BucketNotificationConfig, NotificationError, integration::NotificationMetrics, notification_system_subscriber::NotificationSystemSubscriberView, notifier::EventNotifier, rule_engine::NotifyRuleEngine, }; use rustfs_s3_types::EventName; @@ -156,4 +172,37 @@ mod tests { manager.remove_bucket_notification_config("bucket").await; assert!(!manager.subscriber_view.has_subscriber("bucket", &EventName::ObjectCreatedPut)); } + + #[tokio::test] + async fn validate_bucket_notification_config_rejects_missing_targets_without_applying_rules() { + let manager = build_manager(); + let target_id = TargetID::new("primary".to_string(), "webhook".to_string()); + let mut cfg = BucketNotificationConfig::new("us-east-1"); + cfg.add_rule(&[EventName::ObjectCreatedPut], "*".to_string(), target_id); + + let err = manager + .validate_bucket_notification_config("bucket", &cfg) + .await + .expect_err("validation must fail when the runtime has no notify targets"); + assert!(matches!(err, NotificationError::Configuration(_))); + assert!(!manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await); + assert!(!manager.subscriber_view.has_subscriber("bucket", &EventName::ObjectCreatedPut)); + } + + #[tokio::test] + async fn failed_validation_keeps_previous_bucket_rules() { + let manager = build_manager(); + let target_id = TargetID::new("primary".to_string(), "webhook".to_string()); + let mut cfg = BucketNotificationConfig::new("us-east-1"); + cfg.add_rule(&[EventName::ObjectCreatedPut], "*".to_string(), target_id); + manager.apply_bucket_notification_config("bucket", &cfg).await; + assert!(manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await); + + let err = manager + .validate_bucket_notification_config("bucket", &cfg) + .await + .expect_err("validation must fail when the runtime has no notify targets"); + assert!(matches!(err, NotificationError::Configuration(_))); + assert!(manager.has_subscriber("bucket", &EventName::ObjectCreatedPut).await); + } } diff --git a/crates/notify/src/global.rs b/crates/notify/src/global.rs index c33e47f34..8249c7fca 100644 --- a/crates/notify/src/global.rs +++ b/crates/notify/src/global.rs @@ -239,6 +239,42 @@ pub mod notifier_global { .await } + fn bucket_notification_config_for_rules( + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> BucketNotificationConfig { + let mut bucket_config = BucketNotificationConfig::new(region); + + for (event_names, prefix, suffix, target_ids) in event_rules { + // Use `new_pattern` to construct a matching pattern + let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str())); + + for target_id in target_ids { + bucket_config.add_rule(event_names, pattern.clone(), target_id.clone()); + } + } + + bucket_config + } + + /// Checks that the runtime can accept `event_rules` for `region` without + /// publishing them. Callers use this to fail before persisting a bucket + /// notification configuration the notify subsystem would reject. + pub async fn validate_event_specific_rules( + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), NotificationError> { + let bucket_config = bucket_notification_config_for_rules(region, event_rules); + + // Get global NotificationSystem instance + let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?; + + notification_sys + .validate_bucket_notification_config(bucket_name, &bucket_config) + .await + } + /// Dynamically add notification rules according to different event types. /// /// # Parameter @@ -256,16 +292,7 @@ pub mod notifier_global { region: &str, event_rules: &[(Vec, String, String, Vec)], ) -> Result<(), NotificationError> { - let mut bucket_config = BucketNotificationConfig::new(region); - - for (event_names, prefix, suffix, target_ids) in event_rules { - // Use `new_pattern` to construct a matching pattern - let pattern = crate::rules::pattern::new_pattern(Some(prefix.as_str()), Some(suffix.as_str())); - - for target_id in target_ids { - bucket_config.add_rule(event_names, pattern.clone(), target_id.clone()); - } - } + let bucket_config = bucket_notification_config_for_rules(region, event_rules); // Get global NotificationSystem instance let notification_sys = notification_system().ok_or(NotificationError::Lifecycle(LifecycleError::NotInitialized))?; diff --git a/crates/notify/src/integration.rs b/crates/notify/src/integration.rs index 65d1b5f66..a113395f9 100644 --- a/crates/notify/src/integration.rs +++ b/crates/notify/src/integration.rs @@ -372,6 +372,19 @@ impl NotificationSystem { self.services.config_manager.lifecycle().is_converged() } + /// Validates a bucket notification configuration against the runtime targets + /// without publishing it, so callers can reject a request before persisting. + pub async fn validate_bucket_notification_config( + &self, + bucket: &str, + cfg: &BucketNotificationConfig, + ) -> Result<(), NotificationError> { + self.services + .bucket_config_manager + .validate_bucket_notification_config(bucket, cfg) + .await + } + /// Loads the bucket notification configuration pub async fn load_bucket_notification_config( &self, diff --git a/rustfs/src/app/bucket_usecase.rs b/rustfs/src/app/bucket_usecase.rs index 3c138bccd..dcfbba4c4 100644 --- a/rustfs/src/app/bucket_usecase.rs +++ b/rustfs/src/app/bucket_usecase.rs @@ -92,7 +92,7 @@ use rustfs_policy::policy::{ use rustfs_s3_ops::S3Operation; use rustfs_targets::{ EventName, - arn::{ARN, TargetIDError}, + arn::{ARN, TargetID, TargetIDError}, }; use rustfs_trusted_proxies::ClientInfo; use rustfs_utils::http::{SUFFIX_FORCE_DELETE, get_header}; @@ -508,6 +508,46 @@ fn validate_notification_configuration_filters(notification_configuration: &Noti Ok(()) } +fn parse_notification_target_id(arn_str: &str) -> Result { + ARN::parse(arn_str) + .map(|arn| arn.target_id) + .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) +} + +type NotificationEventRule = (Vec, String, String, Vec); + +/// Builds the notify runtime rules for a bucket notification configuration +/// without touching the store or the runtime rule state. +fn build_notification_event_rules( + notification_configuration: &NotificationConfiguration, +) -> S3Result> { + let mut event_rules = Vec::new(); + let invalid_arn = |e: TargetIDError| { + S3Error::with_message(S3ErrorCode::InvalidArgument, format!("Invalid ARN in notification configuration: {e}")) + }; + + process_queue_configurations( + &mut event_rules, + notification_configuration.queue_configurations.clone(), + parse_notification_target_id, + ) + .map_err(invalid_arn)?; + process_topic_configurations( + &mut event_rules, + notification_configuration.topic_configurations.clone(), + parse_notification_target_id, + ) + .map_err(invalid_arn)?; + process_lambda_configurations( + &mut event_rules, + notification_configuration.lambda_function_configurations.clone(), + parse_notification_target_id, + ) + .map_err(invalid_arn)?; + + Ok(event_rules) +} + fn sr_bucket_meta_item(bucket: String, item_type: &str) -> SRBucketMeta { SRBucketMeta { bucket, @@ -2440,6 +2480,17 @@ impl DefaultBucketUsecase { .await .map_err(ApiError::from)?; + let region = resolve_notification_region(self.global_region(), request_region); + let notify = current_notify_interface_for_context(self.context.as_deref()); + let event_rules = build_notification_event_rules(¬ification_configuration)?; + + // Reject the request before the store write so a failure cannot leave a + // persisted configuration that the notify runtime refused to activate. + notify + .validate_event_specific_rules(&bucket, region.as_str(), &event_rules) + .await + .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}"))?; + let data = serialize_config(¬ification_configuration)?; update_bucket_config_for_incarnation(&bucket, BUCKET_NOTIFICATION_CONFIG, data, expected_incarnation_id) .await @@ -2447,40 +2498,10 @@ impl DefaultBucketUsecase { notify_bucket_metadata_reload(bucket.clone(), "put bucket notification", request_context, false).await; - 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); - let parse_rules = async { - let mut event_rules = Vec::new(); - - process_queue_configurations(&mut event_rules, notification_configuration.queue_configurations.clone(), |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - })?; - process_topic_configurations(&mut event_rules, notification_configuration.topic_configurations.clone(), |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - })?; - process_lambda_configurations( - &mut event_rules, - notification_configuration.lambda_function_configurations.clone(), - |arn_str| { - ARN::parse(arn_str) - .map(|arn| arn.target_id) - .map_err(|e| TargetIDError::InvalidFormat(e.to_string())) - }, - )?; - - Ok::<_, TargetIDError>(event_rules) - }; - - let (clear_result, event_rules_result) = tokio::join!(clear_rules, parse_rules); - - clear_result.map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?; - let event_rules = - event_rules_result.map_err(|e| s3_error!(InvalidArgument, "Invalid ARN in notification configuration: {e}"))?; + notify + .clear_bucket_notification_rules(&bucket) + .await + .map_err(|e| s3_error!(InternalError, "Failed to clear rules: {e}"))?; warn!("notify event rules: {:?}", &event_rules); notify .add_event_specific_rules(&bucket, region.as_str(), &event_rules) diff --git a/rustfs/src/app/context/handles.rs b/rustfs/src/app/context/handles.rs index 8a2bdc29c..fdc9f99cc 100644 --- a/rustfs/src/app/context/handles.rs +++ b/rustfs/src/app/context/handles.rs @@ -167,6 +167,15 @@ impl NotifyInterface for NotifyHandle { runtime_sources::notify(args).await; } + async fn validate_event_specific_rules( + &self, + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), NotificationError> { + runtime_sources::validate_event_specific_rules(bucket_name, region, event_rules).await + } + async fn add_event_specific_rules( &self, bucket_name: &str, diff --git a/rustfs/src/app/context/interfaces.rs b/rustfs/src/app/context/interfaces.rs index b36a560dd..eeea0da4e 100644 --- a/rustfs/src/app/context/interfaces.rs +++ b/rustfs/src/app/context/interfaces.rs @@ -73,6 +73,13 @@ pub trait OutboundTlsRuntimeInterface: Send + Sync { pub trait NotifyInterface: Send + Sync { async fn notify(&self, args: EventArgs); + async fn validate_event_specific_rules( + &self, + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), NotificationError>; + async fn add_event_specific_rules( &self, bucket_name: &str, diff --git a/rustfs/src/app/context/runtime_sources.rs b/rustfs/src/app/context/runtime_sources.rs index f93b83a69..4831cd509 100644 --- a/rustfs/src/app/context/runtime_sources.rs +++ b/rustfs/src/app/context/runtime_sources.rs @@ -71,6 +71,14 @@ pub async fn notify(args: EventArgs) { notifier_global::notify(args).await; } +pub async fn validate_event_specific_rules( + bucket_name: &str, + region: &str, + event_rules: &[(Vec, String, String, Vec)], +) -> Result<(), NotificationError> { + notifier_global::validate_event_specific_rules(bucket_name, region, event_rules).await +} + pub async fn add_event_specific_rules( bucket_name: &str, region: &str, diff --git a/rustfs/src/app/object/extract.rs b/rustfs/src/app/object/extract.rs index 204a0c1f3..26f19c432 100644 --- a/rustfs/src/app/object/extract.rs +++ b/rustfs/src/app/object/extract.rs @@ -2747,6 +2747,15 @@ mod tests { let _ = self.events.send(args.version_id); } + async fn validate_event_specific_rules( + &self, + _bucket_name: &str, + _region: &str, + _event_rules: &[(Vec, String, String, Vec)], + ) -> Result<(), rustfs_notify::NotificationError> { + Ok(()) + } + async fn add_event_specific_rules( &self, _bucket_name: &str, diff --git a/scripts/test_security_workflow.py b/scripts/test_security_workflow.py index 544603851..c5704ba25 100644 --- a/scripts/test_security_workflow.py +++ b/scripts/test_security_workflow.py @@ -303,11 +303,12 @@ class SecurityWorkflowTests(WorkflowSteps, unittest.TestCase): source = (ROOT / f".github/workflows/rustfs-{suite}-test.yml").read_text().splitlines() # Workflow-level concurrency covers every job, including cleanup, # regardless of trigger or the runner hosting the job. + expected_group = "rustfs-performance-suite" if suite == "performance" else "rustfs-shared-functional-tests-v2" self.assertEqual([ line.strip() for line in yaml_block(source, "concurrency", 0) if line.strip() and not line.lstrip().startswith("#") ], [ - "group: rustfs-shared-functional-tests-v2", "cancel-in-progress: false", + f"group: {expected_group}", "cancel-in-progress: false", ]) self.assertIsNotNone(yaml_block(source, "workflow_dispatch", 2)) self.assertIsNotNone(yaml_block(source, "repository_dispatch", 2))