gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-23 11:03:26 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: serialize notify runtime module switch reconciliation (#8020)

Browse Source
This commit is contained in:
cxymds
2026-09-19 13:41:39 +08:00
committed by GitHub
parent c3269ad4c2
commit 0e03bcda49
3 changed files with 107 additions and 43 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/handlers/module_switch.rs
+46 -41
View File
@@ -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<ModuleSwitchSnapshot> {
async fn apply_module_switch_update(context: Arc<AppContext>, 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");
rustfs/src/server/event.rs
+60 -1
View File
@@ -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<Fut, T>(operation: Fut) -> T
where
Fut: Future<Output = T>,
{
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<rustfs_notify::NotifyStore>,
) -> 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<rustfs_notify::NotifyStore>,
) -> 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<CurrentStore>(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());
rustfs/src/server/mod.rs
+1 -1
View File
@@ -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::{