From 2490d4ee22923ffd64e943d2411c1f0fb1378256 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Wed, 8 Jul 2026 15:02:10 +0800 Subject: [PATCH] fix(notify): serialize persisted config read-modify-write to prevent lost updates (#4425) fix(notify): serialize persisted config read-modify-write to prevent lost updates (backlog#968) The notify config write path performed a read -> modify -> write over the persisted server config with no serialization: two concurrent updates could both read the same base config, apply disjoint changes, and race their full-config writes. The later write silently overwrote the earlier one, losing updates (e.g. adding two targets concurrently could drop one) with no error surfaced. Guard the whole read -> modify -> write with a process-global tokio Mutex (NOTIFY_CONFIG_RMW_LOCK). The persisted config is a single process-global resource, so serializing the RMW makes concurrent updates apply in sequence and every change is preserved. The lock is only acquired inside update_server_config and is released before the in-memory reload, so it never nests with the per-manager config RwLock and introduces no lock-ordering risk. The RMW sequence is extracted into serialized_read_modify_write with injected read/save operations so the exact production serialization path is exercised by a regression test that concurrently adds 32 distinct targets and asserts all of them survive. Refs: https://github.com/rustfs/backlog/issues/968 --- crates/notify/src/config_manager.rs | 140 +++++++++++++++++++++++++--- 1 file changed, 129 insertions(+), 11 deletions(-) diff --git a/crates/notify/src/config_manager.rs b/crates/notify/src/config_manager.rs index 6ea0103dd..7cd83ffe3 100644 --- a/crates/notify/src/config_manager.rs +++ b/crates/notify/src/config_manager.rs @@ -22,10 +22,23 @@ use rustfs_config::notify::{ }; use rustfs_config::server_config::{Config, KVS}; use rustfs_targets::{Target, arn::TargetID}; -use std::sync::Arc; -use tokio::sync::RwLock; +use std::sync::{Arc, LazyLock}; +use tokio::sync::{Mutex, RwLock}; use tracing::{debug, info}; +/// Serializes the read-modify-write sequence over the persisted notify server +/// config. The persisted config is a single process-global resource (there is +/// only one backing object store), so without this guard two concurrent updates +/// can both read the same base config, apply disjoint changes, and race their +/// full-config writes — the later write silently overwrites the earlier one, +/// losing updates. Holding this mutex across the whole read→modify→write makes +/// concurrent updates apply serially so every change is preserved. +/// +/// The lock is only ever acquired inside `update_server_config`; it never nests +/// with the per-manager `config` RwLock (the in-memory reload runs after this +/// guard is released), so it introduces no lock-ordering risk. +static NOTIFY_CONFIG_RMW_LOCK: LazyLock> = LazyLock::new(|| Mutex::new(())); + const LOG_COMPONENT_NOTIFY: &str = "notify"; const LOG_SUBSYSTEM_CONFIG: &str = "config"; const EVENT_NOTIFY_RUNTIME_LIFECYCLE: &str = "notify_runtime_lifecycle"; @@ -38,7 +51,7 @@ enum NotifyConfigStoreError { Save(String), } -async fn update_server_config(mut modifier: F) -> Result, NotifyConfigStoreError> +async fn update_server_config(modifier: F) -> Result, NotifyConfigStoreError> where F: FnMut(&mut Config) -> bool, { @@ -46,17 +59,50 @@ where return Err(NotifyConfigStoreError::StorageNotAvailable); }; - let mut new_config = crate::read_notify_server_config_without_migrate(store.clone()) - .await - .map_err(NotifyConfigStoreError::Read)?; + let store_for_save = store.clone(); + serialized_read_modify_write( + modifier, + move || async move { + crate::read_notify_server_config_without_migrate(store) + .await + .map_err(NotifyConfigStoreError::Read) + }, + move |config| async move { + crate::save_notify_server_config(store_for_save, &config) + .await + .map_err(NotifyConfigStoreError::Save) + }, + ) + .await +} + +/// Runs a `read → modify → write` over the persisted notify config while holding +/// [`NOTIFY_CONFIG_RMW_LOCK`], so concurrent updates serialize and cannot clobber +/// each other's changes (backlog#968). `read`/`save` are injected so the exact +/// production serialization path can be exercised in tests without a live store. +async fn serialized_read_modify_write( + mut modifier: F, + read: R, + save: S, +) -> Result, NotifyConfigStoreError> +where + F: FnMut(&mut Config) -> bool, + R: FnOnce() -> RFut, + RFut: std::future::Future>, + S: FnOnce(Config) -> SFut, + SFut: std::future::Future>, +{ + // Hold the RMW lock across the entire read→modify→write so concurrent + // updates serialize and cannot clobber each other's changes (backlog#968). + let _rmw_guard = NOTIFY_CONFIG_RMW_LOCK.lock().await; + + let mut new_config = read().await?; if !modifier(&mut new_config) { return Ok(None); } - crate::save_notify_server_config(store, &new_config) - .await - .map_err(NotifyConfigStoreError::Save)?; + save(new_config.clone()).await?; Ok(Some(new_config)) } @@ -380,7 +426,7 @@ impl NotifyConfigManager { #[cfg(test)] mod tests { - use super::{NotifyConfigManager, runtime_target_id_for_subsystem}; + use super::{NotifyConfigManager, NotifyConfigStoreError, runtime_target_id_for_subsystem, serialized_read_modify_write}; use crate::{ integration::NotificationMetrics, notifier::EventNotifier, registry::TargetRegistry, rule_engine::NotifyRuleEngine, runtime_facade::NotifyRuntimeFacade, @@ -389,7 +435,7 @@ mod tests { NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_NATS_SUB_SYS, NOTIFY_POSTGRES_SUB_SYS, NOTIFY_PULSAR_SUB_SYS, NOTIFY_REDIS_SUB_SYS, NOTIFY_WEBHOOK_SUB_SYS, }; - use rustfs_config::server_config::Config; + use rustfs_config::server_config::{Config, KVS}; use rustfs_targets::ReplayWorkerManager; use std::sync::Arc; use tokio::sync::{RwLock, Semaphore}; @@ -426,6 +472,78 @@ mod tests { .expect("reload_config should succeed for empty targets"); } + // Regression test for backlog#968: the read-modify-write over the persisted + // notify config must be serialized. Many tasks concurrently add a distinct + // target through the same production RMW path (`serialized_read_modify_write`, + // which holds the global RMW lock across read→modify→write) against a shared + // in-memory backend. Every update must survive — no lost updates. Without the + // lock, concurrent tasks would read the same base config and clobber each + // other's writes, leaving only a subset of targets. + #[tokio::test(flavor = "multi_thread", worker_threads = 4)] + async fn concurrent_config_updates_preserve_all_targets() { + // Shared in-memory stand-in for the persisted config backend. + let backend = Arc::new(RwLock::new(Config::default())); + const TASKS: usize = 32; + + let mut handles = Vec::with_capacity(TASKS); + for idx in 0..TASKS { + let backend = backend.clone(); + handles.push(tokio::spawn(async move { + let ttype = NOTIFY_WEBHOOK_SUB_SYS.to_lowercase(); + let tname = format!("target-{idx}"); + + let read_backend = backend.clone(); + let save_backend = backend.clone(); + + let result = serialized_read_modify_write( + |config: &mut Config| { + config + .0 + .entry(ttype.clone()) + .or_default() + .insert(tname.clone(), KVS::default()); + true + }, + move || async move { + let snapshot = read_backend.read().await.clone(); + // Yield inside the critical section to widen the race window: + // if the RMW were not serialized, other tasks would read this + // same base snapshot and their writes would clobber ours. + tokio::task::yield_now().await; + Ok::<_, NotifyConfigStoreError>(snapshot) + }, + move |config: Config| async move { + *save_backend.write().await = config; + Ok::<_, NotifyConfigStoreError>(()) + }, + ) + .await + .expect("serialized RMW should succeed"); + assert!(result.is_some(), "modifier reported a change, expected Some(config)"); + })); + } + + for handle in handles { + handle.await.expect("update task should not panic"); + } + + let final_config = backend.read().await; + let webhook_targets = final_config + .0 + .get(&NOTIFY_WEBHOOK_SUB_SYS.to_lowercase()) + .expect("webhook subsystem should exist after updates"); + + assert_eq!( + webhook_targets.len(), + TASKS, + "all concurrent target additions must be preserved (no lost updates)" + ); + for idx in 0..TASKS { + let tname = format!("target-{idx}"); + assert!(webhook_targets.contains_key(&tname), "missing target {tname}: concurrent update was lost"); + } + } + #[test] fn runtime_target_id_for_subsystem_maps_notify_webhook_to_runtime_type() { let target_id = runtime_target_id_for_subsystem(NOTIFY_WEBHOOK_SUB_SYS, "Primary");