From 0e03bcda49c6e8e8890f4a4b15ae1c389f4495ea Mon Sep 17 00:00:00 2001 From: cxymds Date: Sat, 19 Sep 2026 13:41:39 +0800 Subject: [PATCH] fix: serialize notify runtime module switch reconciliation (#8020) --- rustfs/src/admin/handlers/module_switch.rs | 87 ++++++++++++---------- rustfs/src/server/event.rs | 61 ++++++++++++++- rustfs/src/server/mod.rs | 2 +- 3 files changed, 107 insertions(+), 43 deletions(-) diff --git a/rustfs/src/admin/handlers/module_switch.rs b/rustfs/src/admin/handlers/module_switch.rs index f560df991..47739bf9a 100644 --- a/rustfs/src/admin/handlers/module_switch.rs +++ b/rustfs/src/admin/handlers/module_switch.rs @@ -26,7 +26,7 @@ use crate::server::{ apply_audit_module_switch_for_context, current_module_switch_snapshot, mark_event_notifier_reconciled, mark_event_notifier_unreconciled, refresh_audit_module_enabled, refresh_notify_module_enabled, refresh_persisted_module_switches_from, refresh_persisted_module_switches_from_store, save_persisted_module_switches_to, - validate_module_switch_update, + validate_module_switch_update, with_notify_runtime_reconcile_lock, }; use http::{HeaderMap, StatusCode}; use hyper::Method; @@ -138,48 +138,53 @@ async fn refresh_module_switch_snapshot() -> S3Result { async fn apply_module_switch_update(context: Arc, switches: PersistedModuleSwitches) -> S3Result<()> { preflight_dynamic_config_reload_for_context(Some(context.as_ref()), MODULE_SWITCHES_SIGNAL_SUBSYSTEM).await?; let store = context.object_store(); - mark_event_notifier_unreconciled(); - let notification_system = rustfs_notify::ensure_live_events(); - if switches.notify_enabled { - notification_system - .reload_persisted_config_from_store(store.clone()) - .await - .map_err(|err| { - tracing::warn!(error = %err, "Failed to load notification config for module switch update"); - s3_error!(InternalError, "failed to load notification config") - })?; - } + let mut failures = with_notify_runtime_reconcile_lock(async { + mark_event_notifier_unreconciled(); + let notification_system = rustfs_notify::ensure_live_events(); + if switches.notify_enabled { + notification_system + .reload_persisted_config_from_store(store.clone()) + .await + .map_err(|err| { + tracing::warn!(error = %err, "Failed to load notification config for module switch update"); + s3_error!(InternalError, "failed to load notification config") + })?; + } - let transition_system = notification_system.clone(); - let notify_transition = save_persisted_module_switches_to(store.clone(), switches, move || { - let enabled = refresh_notify_module_enabled(); - transition_system.publish_targets_enabled(enabled, None) + let transition_system = notification_system.clone(); + let notify_transition = save_persisted_module_switches_to(store.clone(), switches, move || { + let enabled = refresh_notify_module_enabled(); + transition_system.publish_targets_enabled(enabled, None) + }) + .await + .map_err(|err| { + tracing::warn!(error = %err, "Failed to save module switches"); + s3_error!(InternalError, "failed to save module switches") + })?; + + let mut failures = Vec::new(); + let mut notify_converged = true; + if let Err(err) = notify_transition.wait().await { + tracing::warn!(error = %err, "Local notification runtime failed to apply module switch update"); + notify_converged = false; + failures.push("local notify"); + } + if !switches.notify_enabled + && let Err(err) = notification_system.reload_persisted_config_from_store(store).await + { + tracing::warn!(error = %err, "Local notification config cache failed to reload after module disable"); + notify_converged = false; + failures.push("local notify config cache"); + } + if notify_converged && notification_system.runtime_lifecycle_is_converged() { + mark_event_notifier_reconciled(); + } else if notify_converged { + failures.push("local notify convergence"); + } + + Ok::<_, s3s::S3Error>(failures) }) - .await - .map_err(|err| { - tracing::warn!(error = %err, "Failed to save module switches"); - s3_error!(InternalError, "failed to save module switches") - })?; - - let mut failures = Vec::new(); - let mut notify_converged = true; - if let Err(err) = notify_transition.wait().await { - tracing::warn!(error = %err, "Local notification runtime failed to apply module switch update"); - notify_converged = false; - failures.push("local notify"); - } - if !switches.notify_enabled - && let Err(err) = notification_system.reload_persisted_config_from_store(store).await - { - tracing::warn!(error = %err, "Local notification config cache failed to reload after module disable"); - notify_converged = false; - failures.push("local notify config cache"); - } - if notify_converged && notification_system.runtime_lifecycle_is_converged() { - mark_event_notifier_reconciled(); - } else if notify_converged { - failures.push("local notify convergence"); - } + .await?; if apply_audit_module_switch_for_context(Some(context.as_ref())).await.is_err() { tracing::warn!(reason = "apply_failed", "Local audit runtime failed to apply module switch update"); diff --git a/rustfs/src/server/event.rs b/rustfs/src/server/event.rs index 6f0769774..37b22bb71 100644 --- a/rustfs/src/server/event.rs +++ b/rustfs/src/server/event.rs @@ -29,6 +29,7 @@ use std::sync::OnceLock; use std::sync::atomic::{AtomicBool, Ordering}; use std::time::Duration; use tokio::spawn; +use tokio::sync::Mutex as AsyncMutex; use tokio::task::JoinHandle; use tokio::time::{Instant, MissedTickBehavior}; use tokio_util::sync::CancellationToken; @@ -38,6 +39,12 @@ static NOTIFY_RUNTIME_RECONCILED: AtomicBool = AtomicBool::new(false); static NOTIFY_BUCKET_RULES_RECONCILED: AtomicBool = AtomicBool::new(false); static ECSTORE_EVENT_DISPATCH_HOOK: OnceLock<()> = OnceLock::new(); +// Serialize local notification lifecycle publication with the periodic and +// peer-triggered reconciler. Lock order is this guard before the module-switch +// RMW/object locks and the server-config read lock; the module-switch handler's +// preliminary server-config read is released before it takes the RMW lock. +static NOTIFY_RUNTIME_RECONCILE_LOCK: AsyncMutex<()> = AsyncMutex::const_new(()); + const EVENT_NOTIFIER_RECONCILE_INTERVAL: Duration = Duration::from_secs(5); const EVENT_NOTIFIER_RECONCILE_ATTEMPT_TIMEOUT: Duration = Duration::from_secs(120); const EVENT_NOTIFY_RUNTIME_RECONCILE: &str = "notify_runtime_reconcile"; @@ -55,6 +62,14 @@ pub(crate) fn mark_event_notifier_unreconciled() { NOTIFY_BUCKET_RULES_RECONCILED.store(false, Ordering::Release); } +pub(crate) async fn with_notify_runtime_reconcile_lock(operation: Fut) -> T +where + Fut: Future, +{ + let _guard = NOTIFY_RUNTIME_RECONCILE_LOCK.lock().await; + operation.await +} + fn are_bucket_notification_rules_reconciled() -> bool { NOTIFY_BUCKET_RULES_RECONCILED.load(Ordering::Acquire) } @@ -161,6 +176,12 @@ fn ensure_event_notifier_converged(system: &NotificationSystem) -> Result<(), No pub(crate) async fn reconcile_event_notifier_from_store( store: std::sync::Arc, +) -> Result<(), NotificationError> { + with_notify_runtime_reconcile_lock(reconcile_event_notifier_from_store_unlocked(store)).await +} + +async fn reconcile_event_notifier_from_store_unlocked( + store: std::sync::Arc, ) -> Result<(), NotificationError> { let result = async { validate_notify_module_env().map_err(NotificationError::Initialization)?; @@ -310,7 +331,7 @@ pub async fn shutdown_event_notifier() -> Result<(), NotificationError> { #[instrument] pub async fn init_event_notifier() -> Result<(), NotificationError> { - init_event_notifier_with_store(runtime_sources::current_object_store_handle).await + with_notify_runtime_reconcile_lock(init_event_notifier_with_store(runtime_sources::current_object_store_handle)).await } async fn init_event_notifier_with_store(current_store: CurrentStore) -> Result<(), NotificationError> @@ -408,6 +429,7 @@ mod tests { use super::{ 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, + with_notify_runtime_reconcile_lock, }; use crate::server::is_event_notifier_reconciled; use crate::storage_api::server::event::StorageObjectInfo; @@ -471,6 +493,43 @@ mod tests { .await; } + #[tokio::test] + #[serial] + async fn notify_runtime_reconcile_lock_serializes_overlapping_operations() { + let entered = Arc::new(Notify::new()); + let release = Arc::new(Notify::new()); + let second_attempted = Arc::new(Notify::new()); + let second_started = Arc::new(AtomicBool::new(false)); + + let first_entered = entered.clone(); + let first_release = release.clone(); + let first = tokio::spawn(with_notify_runtime_reconcile_lock(async move { + first_entered.notify_one(); + first_release.notified().await; + })); + entered.notified().await; + + let second_attempted_signal = second_attempted.clone(); + let second_started_flag = second_started.clone(); + let second = tokio::spawn(async move { + second_attempted_signal.notify_one(); + with_notify_runtime_reconcile_lock(async move { + second_started_flag.store(true, Ordering::Release); + }) + .await; + }); + second_attempted.notified().await; + assert!( + !second_started.load(Ordering::Acquire), + "overlapping reconcile work must wait for the first operation" + ); + + release.notify_one(); + first.await.expect("first reconcile operation should finish"); + second.await.expect("second reconcile operation should finish"); + assert!(second_started.load(Ordering::Acquire)); + } + #[test] fn parse_host_and_port_with_ipv4_and_port() { let (host, port) = parse_host_and_port("127.0.0.1:9000".to_string()); diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 91b1ebc4d..7d79791ef 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -48,7 +48,7 @@ pub use service_state::wait_for_shutdown; // Items only used within the library crate (admin handlers, server/http.rs, etc.). pub(crate) use event::{ is_event_notifier_reconciled, mark_event_notifier_reconciled, mark_event_notifier_unreconciled, - reconcile_event_notifier_from_store, start_persisted_event_notifier_reconciler, + reconcile_event_notifier_from_store, start_persisted_event_notifier_reconciler, with_notify_runtime_reconcile_lock, }; #[cfg(test)] pub(crate) use health::{