From 09157485aace25761c5bbbc1eeef2d4e7c8e1127 Mon Sep 17 00:00:00 2001 From: houseme Date: Wed, 29 Jul 2026 23:00:13 +0800 Subject: [PATCH] fix(notify): reconcile persisted bucket rules (#5437) Restore persisted bucket notification rules after the notification target runtime converges so restarted nodes rebuild their local rule engine without requiring an unchanged PUT bucket notification request. Fixes #5428 Co-authored-by: heihutu --- rustfs/src/init.rs | 270 ++++++++++++++++++++++--------------- rustfs/src/server/event.rs | 60 ++++++++- 2 files changed, 222 insertions(+), 108 deletions(-) diff --git a/rustfs/src/init.rs b/rustfs/src/init.rs index f94523c00..8ac27e2e6 100644 --- a/rustfs/src/init.rs +++ b/rustfs/src/init.rs @@ -15,6 +15,7 @@ use crate::runtime_sources::current_region; use crate::server::ShutdownHandle; use crate::server::runtime_sources::current_notify_interface; +use crate::storage_api::startup::bucket_metadata::contract::bucket::{BucketOperations, BucketOptions}; use crate::storage_api::startup::init::{ get_bucket_notification_config, process_lambda_configurations, process_queue_configurations, process_topic_configurations, }; @@ -23,11 +24,14 @@ use rustfs_config::{ DEFAULT_BUFFER_MAX_SIZE, DEFAULT_BUFFER_MIN_SIZE, DEFAULT_BUFFER_PROFILE, DEFAULT_BUFFER_UNKNOWN_SIZE, DEFAULT_UPDATE_CHECK, ENV_RUSTFS_BUFFER_DEFAULT_SIZE, ENV_RUSTFS_BUFFER_MAX_SIZE, ENV_RUSTFS_BUFFER_MIN_SIZE, ENV_UPDATE_CHECK, RUSTFS_REGION, }; -use rustfs_targets::arn::{ARN, TargetIDError}; +use rustfs_notify::NotificationError; +use rustfs_s3_types::EventName; +use rustfs_targets::arn::{ARN, TargetID, TargetIDError}; use rustfs_utils::get_env_usize; use s3s::s3_error; use std::env; use std::io::Error; +use std::sync::Arc; use tracing::{debug, error, info, instrument, warn}; const LOG_COMPONENT_INIT: &str = "init"; @@ -41,6 +45,8 @@ const LOG_SUBSYSTEM_PROTOCOL: &str = "protocol"; const EVENT_PROTOCOL_RUNTIME_STATE: &str = "protocol_runtime_state"; const EVENT_PROTOCOL_SERVER_STATE: &str = "protocol_server_state"; +type NotificationEventRule = (Vec, String, String, Vec); + #[instrument] pub fn print_server_info() { let current_year = jiff::Zoned::now().year(); @@ -151,6 +157,57 @@ fn arn_to_target_id(arn_str: &str) -> Result Result, TargetIDError> { + let mut event_rules = Vec::new(); + process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), arn_to_target_id)?; + process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), arn_to_target_id)?; + process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), arn_to_target_id)?; + Ok(event_rules) +} + +async fn apply_bucket_notification_configuration(bucket: &str, region: &str) -> Result { + let has_notification_config = get_bucket_notification_config(bucket) + .await + .map_err(|err| NotificationError::StorageNotAvailable(format!("load bucket notification config for {bucket}: {err}")))?; + + match has_notification_config { + Some(cfg) => { + info!( + target: "rustfs::init", + event = "notification_config_loaded", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + queue_configuration_count = cfg.queue_configurations.as_ref().map_or(0, Vec::len), + topic_configuration_count = cfg.topic_configurations.as_ref().map_or(0, Vec::len), + lambda_configuration_count = cfg.lambda_function_configurations.as_ref().map_or(0, Vec::len), + "Loaded bucket notification configuration" + ); + + let event_rules = + notification_config_to_event_rules(&cfg).map_err(|err| NotificationError::BucketNotification(err.to_string()))?; + current_notify_interface() + .add_event_specific_rules(bucket, region, &event_rules) + .await?; + Ok(true) + } + None => { + info!( + target: "rustfs::init", + event = "notification_config_missing", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + "Bucket notification configuration not found" + ); + current_notify_interface().clear_bucket_notification_rules(bucket).await?; + Ok(false) + } + } +} + /// Add existing bucket notification configurations to the global notifier system. /// This function retrieves notification configurations for each bucket /// and registers the corresponding event rules with the notifier system. @@ -159,11 +216,52 @@ fn arn_to_target_id(arn_str: &str) -> Result) { + let region = notification_region(); + for bucket in buckets.iter() { + if let Err(err) = apply_bucket_notification_configuration(bucket, region.as_str()).await { + let err = s3_error!(InternalError, "Failed to add rules: {err}"); + error!( + target: "rustfs::init", + event = "notification_rules_registration_failed", + component = LOG_COMPONENT_INIT, + subsystem = LOG_SUBSYSTEM_NOTIFICATION, + bucket = %bucket, + region, + error = ?err, + "Failed to register bucket notification rules" + ); + } + } +} + +pub(crate) async fn reconcile_persisted_bucket_notification_configurations( + store: Arc, +) -> Result { + let bucket_infos = store + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .map_err(|err| NotificationError::StorageNotAvailable(format!("list buckets for notification reconciliation: {err}")))?; + let region = notification_region(); + let mut configured_bucket_count = 0; + + for bucket in bucket_infos { + if apply_bucket_notification_configuration(&bucket.name, region.as_str()).await? { + configured_bucket_count += 1; + } + } + + Ok(configured_bucket_count) +} + +fn notification_region() -> String { let global_region = current_region(); - let region = global_region + global_region .as_ref() .filter(|r| !r.as_str().is_empty()) - .map(|r| r.as_str()) + .map(|r| r.to_string()) .unwrap_or_else(|| { warn!( target: "rustfs::init", @@ -173,107 +271,8 @@ pub async fn add_bucket_notification_configuration(buckets: Vec) { fallback_region = RUSTFS_REGION, "Notification configuration falling back to default region" ); - RUSTFS_REGION - }); - for bucket in buckets.iter() { - let has_notification_config = get_bucket_notification_config(bucket).await.unwrap_or_else(|err| { - warn!( - target: "rustfs::init", - event = "notification_config_load_failed", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - error = ?err, - "Failed to load bucket notification configuration" - ); - None - }); - - match has_notification_config { - Some(cfg) => { - info!( - target: "rustfs::init", - event = "notification_config_loaded", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - queue_configuration_count = cfg.queue_configurations.as_ref().map_or(0, Vec::len), - topic_configuration_count = cfg.topic_configurations.as_ref().map_or(0, Vec::len), - lambda_configuration_count = cfg.lambda_function_configurations.as_ref().map_or(0, Vec::len), - "Loaded bucket notification configuration" - ); - - let mut event_rules = Vec::new(); - if let Err(e) = process_queue_configurations(&mut event_rules, cfg.queue_configurations.clone(), arn_to_target_id) - { - error!( - target: "rustfs::init", - event = "notification_config_parse_failed", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - config_type = "queue", - error = ?e, - "Failed to parse bucket notification configuration" - ); - } - if let Err(e) = process_topic_configurations(&mut event_rules, cfg.topic_configurations.clone(), arn_to_target_id) - { - error!( - target: "rustfs::init", - event = "notification_config_parse_failed", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - config_type = "topic", - error = ?e, - "Failed to parse bucket notification configuration" - ); - } - if let Err(e) = - process_lambda_configurations(&mut event_rules, cfg.lambda_function_configurations.clone(), arn_to_target_id) - { - error!( - target: "rustfs::init", - event = "notification_config_parse_failed", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - config_type = "lambda", - error = ?e, - "Failed to parse bucket notification configuration" - ); - } - - if let Err(e) = current_notify_interface() - .add_event_specific_rules(bucket, region, &event_rules) - .await - .map_err(|e| s3_error!(InternalError, "Failed to add rules: {e}")) - { - error!( - target: "rustfs::init", - event = "notification_rules_registration_failed", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - region, - error = ?e, - "Failed to register bucket notification rules" - ); - } - } - None => { - info!( - target: "rustfs::init", - event = "notification_config_missing", - component = LOG_COMPONENT_INIT, - subsystem = LOG_SUBSYSTEM_NOTIFICATION, - bucket = %bucket, - "Bucket notification configuration not found" - ); - } - } - } + RUSTFS_REGION.to_string() + }) } /// Build KMS configuration for local backend @@ -1354,9 +1353,13 @@ pub async fn init_sftp_system() -> Result, Box = OnceLock::new(); const EVENT_NOTIFIER_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); @@ -54,6 +56,19 @@ pub(crate) fn mark_event_notifier_reconciled() { pub(crate) fn mark_event_notifier_unreconciled() { NOTIFY_RUNTIME_RECONCILED.store(false, Ordering::Release); + NOTIFY_BUCKET_RULES_RECONCILED.store(false, Ordering::Release); +} + +fn are_bucket_notification_rules_reconciled() -> bool { + NOTIFY_BUCKET_RULES_RECONCILED.load(Ordering::Acquire) +} + +fn mark_bucket_notification_rules_reconciled() { + NOTIFY_BUCKET_RULES_RECONCILED.store(true, Ordering::Release); +} + +fn should_reconcile_bucket_notification_rules(runtime_changed: bool, notify_enabled: bool) -> bool { + notify_enabled && (runtime_changed || !are_bucket_notification_rules_reconciled()) } pub fn refresh_notify_module_enabled() -> bool { @@ -180,7 +195,7 @@ pub(crate) async fn reconcile_event_notifier_from_store( let system = ensure_live_events_initialized(); let transition_system = system.clone(); let transition_store = store.clone(); - let transition = with_refreshed_notify_module_state_from(store, move |resolution| async move { + let transition = with_refreshed_notify_module_state_from(store.clone(), move |resolution| async move { NOTIFY_MODULE_ENABLED.store(resolution.enabled, Ordering::Relaxed); let read_store = transition_store.clone(); let config_system = transition_system.clone(); @@ -208,11 +223,25 @@ pub(crate) async fn reconcile_event_notifier_from_store( .await .map_err(|err| NotificationError::Initialization(format!("failed to refresh notify module switch: {err}")))??; + let runtime_changed = transition.is_some(); if let Some(transition) = transition { transition.wait().await?; } - ensure_event_notifier_converged(&system) + ensure_event_notifier_converged(&system)?; + if should_reconcile_bucket_notification_rules(runtime_changed, is_notify_module_enabled()) { + let configured_bucket_count = reconcile_persisted_bucket_notification_configurations(store).await?; + mark_bucket_notification_rules_reconciled(); + info!( + event = EVENT_NOTIFY_RUNTIME_RECONCILE, + component = "notify", + subsystem = "bucket_rules", + configured_bucket_count, + "Persisted bucket notification rules reconciled" + ); + } + + Ok(()) } .await; @@ -405,7 +434,8 @@ mod tests { set_persisted_module_switches, }; use super::{ - convert_ecstore_object_info, init_event_notifier_with_store, parse_host_and_port, run_persisted_event_notifier_reconciler, + convert_ecstore_object_info, init_event_notifier_with_store, mark_bucket_notification_rules_reconciled, + parse_host_and_port, run_persisted_event_notifier_reconciler, should_reconcile_bucket_notification_rules, }; use crate::server::is_event_notifier_reconciled; use crate::storage_api::server::event::StorageObjectInfo; @@ -533,6 +563,30 @@ mod tests { assert_eq!(converted.transitioned_tier.as_deref(), Some("DEEP_ARCHIVE")); } + #[test] + fn bucket_rule_reconcile_runs_once_and_after_runtime_change() { + super::mark_event_notifier_unreconciled(); + + assert!( + should_reconcile_bucket_notification_rules(false, true), + "enabled notify must restore bucket rules until the first successful replay" + ); + + mark_bucket_notification_rules_reconciled(); + assert!( + !should_reconcile_bucket_notification_rules(false, true), + "steady-state reconcile must not rescan buckets every tick" + ); + assert!( + should_reconcile_bucket_notification_rules(true, true), + "target runtime changes must replay persisted bucket rules" + ); + assert!( + !should_reconcile_bucket_notification_rules(true, false), + "disabled notify must not load external target rules" + ); + } + #[tokio::test(start_paused = true)] async fn persisted_reconciler_converges_after_one_injected_tick() { let persisted_generation = Arc::new(AtomicUsize::new(1));