Merge branch 'main' into cxymds/fix-5469-config-atomicity

Co-Authored-By: heihutu <heihutu@gmail.com>

.config/nextest.toml

@@ -300,6 +300,9 @@ path = "junit.xml"
 # negative-path siblings of each family stay in as regression guards.
 #   * rustfs#4843 — over-limit archive entry paths hard-reject the whole
 #     archive even under ignore-errors semantics.
+#   * rustfs#4846 — distributed-lock quorum tests misclassify as timeout
+#     under parallel load (multi-node in-process clusters; natural home is
+#     ci-7's nightly cluster lane).
 [profile.e2e-full]
 default-filter = """
     package(e2e_test)
@@ -308,6 +311,7 @@ default-filter = """
     & !test(/^replication_extension_test::/)
     & !test(/^multipart_auth_test::test_signed_put_object_extract_skips_invalid_entry_when_ignore_errors_enabled$/)
     & !test(/^snowball_auto_extract_test::tests::snowball_auto_extract_(ignores_invalid_entries_when_requested|supports_standard_headers_with_combined_extract_options)$/)
+    & !test(/^reliant::lock::test_distributed_lock_(2_nodes_grpc_read_survives_failed_node|4_nodes_grpc_read_write_quorum_split_with_two_failed_nodes)$/)
 """
 fail-fast = false
 

crates/e2e_test/src/reliant/lock.rs

@@ -15,7 +15,9 @@
 
 use super::{grpc_lock_client::GrpcLockClient, grpc_lock_server::spawn_lock_server};
 use rustfs_lock::client::{LockClient, local::LocalClient};
-use rustfs_lock::{GlobalLockManager, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey};
+use rustfs_lock::{
+    GlobalLockManager, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockType, NamespaceLock, ObjectKey,
+};
 use std::sync::Arc;
 use std::time::Duration;
 
@@ -33,11 +35,7 @@ struct FailingClient;
 #[async_trait::async_trait]
 impl rustfs_lock::LockClient for FailingClient {
     async fn acquire_lock(&self, _request: &rustfs_lock::LockRequest) -> rustfs_lock::Result<LockResponse> {
-        // Match RemoteClient's transport-failure response so the coordinator can count this node toward quorum loss.
-        Ok(LockResponse::failure(
-            "Remote lock RPC failed: simulated gRPC node failure",
-            Duration::ZERO,
-        ))
+        Err(LockError::internal("simulated gRPC node failure"))
     }
 
     async fn release(&self, _lock_id: &rustfs_lock::LockId) -> rustfs_lock::Result<bool> {

crates/ecstore/src/api/mod.rs

@@ -267,12 +267,13 @@ pub mod config {
     pub mod com {
         pub use crate::config::com::{
             COMMA_SEPARATED_LISTS, CONFIG_PREFIX, ENV_CONFIG_RECOVER_ON_CORRUPTION, STORAGE_CLASS_SUB_SYS,
-            ServerConfigCorruptError, ServerConfigSnapshot, delete_config, is_server_config_corrupt_error, lookup_configs,
-            read_config, read_config_no_lock, read_config_with_metadata, read_config_without_migrate,
-            read_config_without_migrate_no_lock, read_existing_server_config_no_lock, read_server_config_snapshot, save_config,
-            save_config_no_lock, save_config_with_opts, save_server_config, save_server_config_no_lock,
-            save_server_config_snapshot, server_config_path, try_migrate_server_config, with_config_object_read_lock,
-            with_config_object_write_lock, with_server_config_read_lock, with_server_config_write_lock,
+            ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config,
+            is_server_config_corrupt_error, lookup_configs, read_config, read_config_no_lock, read_config_with_metadata,
+            read_config_without_migrate, read_config_without_migrate_no_lock, read_existing_server_config_no_lock,
+            read_server_config_snapshot, save_config, save_config_no_lock, save_config_with_opts, save_server_config,
+            save_server_config_no_lock, save_server_config_snapshot, save_server_config_snapshot_with_generation,
+            server_config_path, try_migrate_server_config, with_config_object_read_lock, with_config_object_write_lock,
+            with_server_config_read_lock, with_server_config_write_lock,
         };
     }
 

crates/ecstore/src/config/com.rs

@@ -53,12 +53,14 @@ use std::sync::LazyLock;
 use std::sync::{Arc, RwLock};
 use tokio::sync::{OwnedRwLockWriteGuard, RwLock as AsyncRwLock};
 use tracing::{debug, error, info, instrument, warn};
+use uuid::Uuid;
 
 pub const CONFIG_PREFIX: &str = "config";
 const SERVER_CONFIG_OBJECT: &str = "config/config.json";
+const CONFIG_TRANSACTION_LOCK_SUFFIX: &str = ".transaction.lock";
 
-// Server-config lock order: SERVER_CONFIG_LOCK -> distributed namespace lock
-// for SERVER_CONFIG_OBJECT. Readers and writers must never reverse this order.
+// Server-config lock order: SERVER_CONFIG_LOCK -> transaction lock ->
+// SERVER_CONFIG_OBJECT. Readers and writers must never reverse this order.
 static SERVER_CONFIG_LOCK: LazyLock<Arc<AsyncRwLock<()>>> = LazyLock::new(|| Arc::new(AsyncRwLock::new(())));
 
 fn config_task_join_error(operation: &'static str, error: tokio::task::JoinError) -> Error {
@@ -76,8 +78,11 @@ where
     T: Send + 'static,
 {
     tokio::spawn(async move {
-        // Lock order: SERVER_CONFIG_LOCK -> namespace write lock.
+        // Lock order: SERVER_CONFIG_LOCK -> transaction lock -> object lock.
         let _local_guard = SERVER_CONFIG_LOCK.write().await;
+        let transaction_lock = server_config_transaction_lock_path();
+        let transaction_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?;
+        let _transaction_guard = transaction_lock.get_write_lock(get_lock_acquire_timeout()).await?;
         let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, SERVER_CONFIG_OBJECT).await?;
         let _write_guard = namespace_lock.get_write_lock(get_lock_acquire_timeout()).await?;
         Ok(operation().await)
@@ -96,8 +101,11 @@ where
     T: Send + 'static,
 {
     tokio::spawn(async move {
-        // Lock order: SERVER_CONFIG_LOCK -> namespace read lock.
+        // Lock order: SERVER_CONFIG_LOCK -> transaction lock -> object lock.
         let _local_guard = SERVER_CONFIG_LOCK.read().await;
+        let transaction_lock = server_config_transaction_lock_path();
+        let transaction_lock = store.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?;
+        let _transaction_guard = transaction_lock.get_read_lock(get_lock_acquire_timeout()).await?;
         let namespace_lock = store.new_ns_lock(RUSTFS_META_BUCKET, SERVER_CONFIG_OBJECT).await?;
         let _read_guard = namespace_lock.get_read_lock(get_lock_acquire_timeout()).await?;
         Ok(operation().await)
@@ -567,6 +575,21 @@ where
 }
 
 pub async fn save_config_with_opts<S>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<()>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        >,
+{
+    save_config_with_opts_and_metadata(api, file, data, opts).await.map(|_| ())
+}
+
+async fn save_config_with_opts_and_metadata<S>(api: Arc<S>, file: &str, data: Vec<u8>, opts: &ObjectOptions) -> Result<ObjectInfo>
 where
     S: ObjectIO<
             Error = Error,
@@ -579,11 +602,13 @@ where
         >,
 {
     let mut put_data = PutObjReader::from_vec(data);
-    if let Err(err) = api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await {
-        error!("save_config_with_opts: err: {:?}, file: {}", err, file);
-        return Err(err);
+    match api.put_object(RUSTFS_META_BUCKET, file, &mut put_data, opts).await {
+        Ok(object_info) => Ok(object_info),
+        Err(err) => {
+            error!("save_config_with_opts: err: {:?}, file: {}", err, file);
+            Err(err)
+        }
     }
-    Ok(())
 }
 
 fn new_server_config() -> Config {
@@ -594,8 +619,12 @@ async fn new_and_save_server_config<S>(api: Arc<S>) -> Result<Config>
 where
     S: EcstoreObjectIO + StorageAdminApi + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
+    let snapshot = read_server_config_snapshot(api.clone()).await?;
+    if snapshot.object_exists() {
+        return Ok(snapshot.config.clone());
+    }
     let cfg = new_server_config();
-    save_server_config(api, &cfg).await?;
+    save_server_config_snapshot(api, &cfg, &snapshot).await?;
 
     Ok(cfg)
 }
@@ -617,6 +646,10 @@ pub fn server_config_path() -> String {
     SERVER_CONFIG_OBJECT.to_string()
 }
 
+fn server_config_transaction_lock_path() -> String {
+    format!("{}{CONFIG_TRANSACTION_LOCK_SUFFIX}", server_config_path())
+}
+
 fn storage_class_kvs_mut(cfg: &mut Config) -> &mut KVS {
     let sub_cfg = cfg.0.entry(STORAGE_CLASS_SUB_SYS.to_string()).or_insert_with(|| {
         let mut section = HashMap::new();
@@ -819,6 +852,9 @@ fn apply_external_scalar_config_map(
     let Some(config_value) = root.get(descriptor.subsystem_key) else {
         return Ok(false);
     };
+    if descriptor.subsystem_key == HEAL_SUB_SYS && config_value.is_null() {
+        return Ok(false);
+    }
     let overrides = decode_scalar_config_value(config_value, descriptor)?;
 
     if overrides.is_empty() {
@@ -1463,21 +1499,142 @@ fn build_audit_object(cfg: &Config) -> Map<String, Value> {
     build_target_object(cfg, &audit_target_descriptors())
 }
 
+fn sync_rendered_target_instance(existing: Value, rendered: Option<&Value>, valid_keys: &[&str]) -> Option<Value> {
+    match existing {
+        Value::Object(mut instance) => {
+            for key in valid_keys {
+                instance.remove(*key);
+            }
+            if let Some(Value::Object(rendered)) = rendered {
+                instance.extend(rendered.clone());
+            }
+            (!instance.is_empty()).then_some(Value::Object(instance))
+        }
+        Value::Array(entries) => {
+            let mut pending = rendered
+                .and_then(Value::as_object)
+                .map(|rendered| {
+                    rendered
+                        .iter()
+                        .filter_map(|(key, value)| parse_target_scalar_value(key, value).map(|value| (key.clone(), value)))
+                        .collect::<HashMap<_, _>>()
+                })
+                .unwrap_or_default();
+            let mut updated = Vec::with_capacity(entries.len().saturating_add(pending.len()));
+            for entry in entries {
+                let Some(entry_obj) = entry.as_object() else {
+                    updated.push(entry);
+                    continue;
+                };
+                let Some(key) = entry_obj.get("key").and_then(Value::as_str) else {
+                    updated.push(entry);
+                    continue;
+                };
+                if !valid_keys.contains(&key) {
+                    updated.push(entry);
+                    continue;
+                }
+                let Some(value) = pending.remove(key) else {
+                    continue;
+                };
+                let mut entry_obj = entry_obj.clone();
+                entry_obj.insert("value".to_string(), Value::String(value));
+                updated.push(Value::Object(entry_obj));
+            }
+            updated.extend(rendered_scalar_config_kvs_entries(&pending));
+            (!updated.is_empty()).then_some(Value::Array(updated))
+        }
+        value if rendered.is_none() => Some(value),
+        _ => rendered.cloned(),
+    }
+}
+
 fn sync_rendered_target_object(
     target_obj: &mut Map<String, Value>,
     rendered_target: &Map<String, Value>,
     descriptors: &[TargetConfigDescriptor],
 ) {
     for descriptor in descriptors {
-        match rendered_target.get(descriptor.external_key) {
-            Some(Value::Object(v)) => {
-                target_obj.insert(descriptor.external_key.to_string(), Value::Object(v.clone()));
-                target_obj.remove(descriptor.subsystem_key);
+        let existing = target_obj.remove(descriptor.external_key);
+        let alias = target_obj.remove(descriptor.subsystem_key);
+        let mut section = existing
+            .or(alias)
+            .and_then(|value| value.as_object().cloned())
+            .unwrap_or_default();
+        let rendered = rendered_target.get(descriptor.external_key).and_then(Value::as_object);
+
+        if is_target_instance_shorthand(&section, descriptor.valid_keys) {
+            let has_named_instances = rendered.is_some_and(|instances| instances.keys().any(|name| name != "default"));
+            if !has_named_instances {
+                if let Some(section) = sync_rendered_target_instance(
+                    Value::Object(section),
+                    rendered.and_then(|instances| instances.get("default")),
+                    descriptor.valid_keys,
+                ) {
+                    target_obj.insert(descriptor.external_key.to_string(), section);
+                }
+                continue;
             }
-            _ => {
-                target_obj.remove(descriptor.external_key);
-                target_obj.remove(descriptor.subsystem_key);
+
+            let mut nested = Map::new();
+            if let Some(default) = sync_rendered_target_instance(
+                Value::Object(section),
+                rendered.and_then(|instances| instances.get("default")),
+                descriptor.valid_keys,
+            ) {
+                nested.insert("default".to_string(), default);
             }
+            if let Some(rendered) = rendered {
+                for (instance_name, instance) in rendered {
+                    if instance_name != "default" {
+                        nested.insert(instance_name.clone(), instance.clone());
+                    }
+                }
+            }
+            if !nested.is_empty() {
+                target_obj.insert(descriptor.external_key.to_string(), Value::Object(nested));
+            }
+            continue;
+        }
+
+        if let Some(default_alias) = section.remove(DEFAULT_DELIMITER) {
+            if let Some(default) = section.get_mut("default") {
+                if let Some(alias) = sync_rendered_target_instance(default_alias, None, descriptor.valid_keys) {
+                    match (default, alias) {
+                        (Value::Object(default), Value::Object(alias)) => {
+                            for (key, value) in alias {
+                                default.entry(key).or_insert(value);
+                            }
+                        }
+                        (Value::Array(default), Value::Array(alias)) => default.extend(alias),
+                        _ => {}
+                    }
+                }
+            } else {
+                section.insert("default".to_string(), default_alias);
+            }
+        }
+
+        let mut merged = Map::new();
+        for (instance_name, instance) in section {
+            if let Some(instance) = sync_rendered_target_instance(
+                instance,
+                rendered.and_then(|instances| instances.get(&instance_name)),
+                descriptor.valid_keys,
+            ) {
+                merged.insert(instance_name, instance);
+            }
+        }
+        if let Some(rendered) = rendered {
+            for (instance_name, instance) in rendered {
+                if !merged.contains_key(instance_name) {
+                    merged.insert(instance_name.clone(), instance.clone());
+                }
+            }
+        }
+
+        if !merged.is_empty() {
+            target_obj.insert(descriptor.external_key.to_string(), Value::Object(merged));
         }
     }
 }
@@ -1496,6 +1653,14 @@ fn encode_server_config_blob(cfg: &Config, seed: Option<&[u8]>) -> Result<Vec<u8
         Some(Value::Object(v)) => v,
         _ => Map::new(),
     };
+    for key in [
+        storageclass::CLASS_STANDARD,
+        storageclass::CLASS_RRS,
+        storageclass::OPTIMIZE,
+        storageclass::INLINE_BLOCK,
+    ] {
+        sc_obj.remove(key);
+    }
     for (k, v) in build_storageclass_object(cfg) {
         sc_obj.insert(k, v);
     }
@@ -1503,7 +1668,10 @@ fn encode_server_config_blob(cfg: &Config, seed: Option<&[u8]>) -> Result<Vec<u8
     root.remove("storage_class");
 
     for descriptor in [scanner_config_descriptor(), heal_config_descriptor()] {
-        let existing = root.remove(descriptor.subsystem_key);
+        let mut existing = root.remove(descriptor.subsystem_key);
+        if descriptor.subsystem_key == HEAL_SUB_SYS && existing.as_ref().is_some_and(Value::is_null) {
+            existing = None;
+        }
         let rendered = build_scalar_config_object(cfg, descriptor);
         if let Some(config_value) = sync_rendered_scalar_config_value(existing, &rendered, descriptor)? {
             root.insert(descriptor.subsystem_key.to_string(), config_value);
@@ -1560,6 +1728,7 @@ fn is_standard_object_server_config(data: &[u8]) -> bool {
     matches!(root.get("version"), Some(Value::String(v)) if !v.trim().is_empty())
         && matches!(root.get("storageclass"), Some(Value::Object(_)))
         && !root.contains_key("storage_class")
+        && !matches!(root.get(HEAL_SUB_SYS), Some(Value::Null))
 }
 
 fn configs_semantically_equal(lhs: &Config, rhs: &Config) -> bool {
@@ -1593,7 +1762,7 @@ where
             FileInfo = FileInfo,
             ObjectToDelete = ObjectToDelete,
             DeletedObject = DeletedObject,
-        >,
+        > + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
 {
     if let Some(decrypt) = &decrypt_fn {
         register_server_config_decrypt_fn(decrypt.clone());
@@ -1601,14 +1770,7 @@ where
 
     let config_file = server_config_path();
     match api
-        .get_object_info(
-            RUSTFS_META_BUCKET,
-            &config_file,
-            &ObjectOptions {
-                no_lock: true,
-                ..Default::default()
-            },
-        )
+        .get_object_info(RUSTFS_META_BUCKET, &config_file, &ObjectOptions::default())
         .await
     {
         Ok(_) => {
@@ -1624,7 +1786,6 @@ where
 
     let opts = ObjectOptions {
         max_parity: true,
-        no_lock: true,
         ..Default::default()
     };
 
@@ -1677,7 +1838,33 @@ where
         }
     };
 
-    match save_config(api, &config_file, normalized).await {
+    let snapshot = match read_server_config_snapshot(api.clone()).await {
+        Ok(snapshot) => snapshot,
+        Err(err) => {
+            warn!("recheck target server config failed, skip migration: {:?}", err);
+            return;
+        }
+    };
+    if snapshot.object_exists() {
+        debug!("server config was created while legacy migration was preparing, skip migration");
+        return;
+    }
+
+    match save_config_with_opts(
+        api,
+        &config_file,
+        normalized,
+        &ObjectOptions {
+            max_parity: true,
+            http_preconditions: Some(HTTPPreconditions {
+                if_none_match: Some("*".to_string()),
+                ..Default::default()
+            }),
+            ..Default::default()
+        },
+    )
+    .await
+    {
         Ok(()) => {
             info!("Migrated compatible server config from legacy metadata bucket");
         }
@@ -1769,8 +1956,13 @@ where
 {
     let config_file = server_config_path();
 
-    // Try to read the configuration file
-    match read_config_no_lock(api.clone(), &config_file).await {
+    // Try to read the configuration file.
+    let data = if namespace_lock_held {
+        read_config_no_lock(api.clone(), &config_file).await
+    } else {
+        read_config(api.clone(), &config_file).await
+    };
+    match data {
         Ok(data) => read_server_config(api, &data, namespace_lock_held).await,
         Err(Error::ConfigNotFound) => handle_missing_config(api, "Read the main configuration", namespace_lock_held).await,
         Err(err) => handle_config_read_error(err, &config_file),
@@ -1787,7 +1979,12 @@ where
         warn!("Received empty configuration data, try to reread from '{}'", config_file);
 
         // Try to read the configuration again
-        match read_config_no_lock(api.clone(), &config_file).await {
+        let data = if namespace_lock_held {
+            read_config_no_lock(api.clone(), &config_file).await
+        } else {
+            read_config(api.clone(), &config_file).await
+        };
+        match data {
             Ok(cfg_data) => {
                 let cfg = decode_persisted_server_config(&cfg_data)?;
                 return Ok(cfg.merge());
@@ -2036,11 +2233,16 @@ pub struct ServerConfigSnapshot {
     raw: Option<Vec<u8>>,
     seed: Option<Vec<u8>>,
     etag: Option<String>,
+    generation: Option<Uuid>,
     _local_guard: OwnedRwLockWriteGuard<()>,
     _guard: rustfs_lock::NamespaceLockGuard,
 }
 
 impl ServerConfigSnapshot {
+    pub fn object_exists(&self) -> bool {
+        self.raw.is_some()
+    }
+
     pub fn ensure_lock_held(&self) -> Result<()> {
         if self._guard.is_lock_lost() {
             return Err(Error::other("server config transaction lock was lost"));
@@ -2051,12 +2253,34 @@ impl ServerConfigSnapshot {
     pub fn is_lock_lost(&self) -> bool {
         self._guard.is_lock_lost()
     }
+
+    pub fn generation(&self) -> Option<Uuid> {
+        self.generation
+    }
 }
 
-/// Read a server config transaction snapshot while holding the same local and
-/// distributed write locks used by every other server-config writer. Internal
-/// reads and the later conditional write use no-lock object I/O; the guards
-/// remain live until the snapshot is dropped.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct ServerConfigSaveResult {
+    persisted: bool,
+    generation: Option<Uuid>,
+}
+
+impl ServerConfigSaveResult {
+    pub fn persisted(&self) -> bool {
+        self.persisted
+    }
+
+    pub fn generation(&self) -> Option<Uuid> {
+        self.generation
+    }
+}
+
+/// Read a server config transaction snapshot while holding a dedicated
+/// transaction lock. The config object's normal namespace lock remains
+/// available to fence reads and the conditional write at commit time.
+/// The transaction guard remains live until the snapshot is dropped,
+/// serializing persistence and history ordering across admin nodes. Runtime
+/// state is reloaded from the durable object after this guard is released.
 pub async fn read_server_config_snapshot<S>(api: Arc<S>) -> Result<ServerConfigSnapshot>
 where
     S: ObjectIO<
@@ -2071,12 +2295,10 @@ where
 {
     let config_file = server_config_path();
     let local_guard = SERVER_CONFIG_LOCK.clone().write_owned().await;
-    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &config_file).await?;
+    let transaction_lock = server_config_transaction_lock_path();
+    let lock = api.new_ns_lock(RUSTFS_META_BUCKET, &transaction_lock).await?;
     let guard = lock.get_write_lock(get_lock_acquire_timeout()).await?;
-    let read_options = ObjectOptions {
-        no_lock: true,
-        ..Default::default()
-    };
+    let read_options = ObjectOptions::default();
     match read_config_with_metadata_inner(api, &config_file, &read_options, true).await {
         Ok((raw, object_info)) => {
             let (config, seed) = decode_persisted_server_config_with_seed(&raw)?;
@@ -2085,6 +2307,7 @@ where
                 raw: Some(raw),
                 seed: Some(seed),
                 etag: object_info.etag,
+                generation: object_info.data_dir.filter(|generation| !generation.is_nil()),
                 _local_guard: local_guard,
                 _guard: guard,
             })
@@ -2094,6 +2317,7 @@ where
             raw: None,
             seed: None,
             etag: None,
+            generation: None,
             _local_guard: local_guard,
             _guard: guard,
         }),
@@ -2108,6 +2332,27 @@ where
 /// lock, so a concurrent update or transaction lease loss cannot commit an
 /// unfenced overwrite.
 pub async fn save_server_config_snapshot<S>(api: Arc<S>, cfg: &Config, snapshot: &ServerConfigSnapshot) -> Result<bool>
+where
+    S: ObjectIO<
+            Error = Error,
+            RangeSpec = HTTPRangeSpec,
+            HeaderMap = HeaderMap,
+            ObjectOptions = ObjectOptions,
+            ObjectInfo = ObjectInfo,
+            GetObjectReader = GetObjectReader,
+            PutObjectReader = PutObjReader,
+        > + NamespaceLocking<Error = Error, NamespaceLock = rustfs_lock::NamespaceLockWrapper>,
+{
+    save_server_config_snapshot_with_generation(api, cfg, snapshot)
+        .await
+        .map(|result| result.persisted())
+}
+
+pub async fn save_server_config_snapshot_with_generation<S>(
+    api: Arc<S>,
+    cfg: &Config,
+    snapshot: &ServerConfigSnapshot,
+) -> Result<ServerConfigSaveResult>
 where
     S: ObjectIO<
             Error = Error,
@@ -2126,13 +2371,19 @@ where
         && configs_semantically_equal(&snapshot.config, cfg)
     {
         debug!("server config unchanged and already in standard object shape, skip write");
-        return Ok(false);
+        return Ok(ServerConfigSaveResult {
+            persisted: false,
+            generation: snapshot.generation(),
+        });
     }
 
     let data = encode_server_config_blob(cfg, snapshot.seed.as_deref())?;
     if snapshot.raw.as_deref().is_some_and(|current| current == data.as_slice()) {
         debug!("server config bytes unchanged after encode, skip write");
-        return Ok(false);
+        return Ok(ServerConfigSaveResult {
+            persisted: false,
+            generation: snapshot.generation(),
+        });
     }
 
     let http_preconditions = if snapshot.raw.is_some() {
@@ -2152,19 +2403,22 @@ where
         }
     };
 
-    save_config_with_opts(
+    snapshot.ensure_lock_held()?;
+    let object_info = save_config_with_opts_and_metadata(
         api,
         &config_file,
         data,
         &ObjectOptions {
             max_parity: true,
-            no_lock: true,
             http_preconditions: Some(http_preconditions),
             ..Default::default()
         },
     )
     .await?;
-    Ok(true)
+    Ok(ServerConfigSaveResult {
+        persisted: true,
+        generation: object_info.data_dir.filter(|generation| !generation.is_nil()),
+    })
 }
 
 /// Saves the server config while an upper layer holds the namespace write
@@ -2301,8 +2555,9 @@ mod tests {
     use super::{
         SERVER_CONFIG_LOCK, ServerConfigSnapshot, apply_dynamic_config_for_sub_sys_with, config_task_join_error,
         configs_semantically_equal, decode_server_config_blob, encode_server_config_blob, is_standard_object_server_config,
-        lookup_configs, read_config, read_config_preserve_empty, read_config_with_metadata, read_config_without_migrate,
-        read_server_config_snapshot, save_server_config, save_server_config_snapshot, server_config_path, storage_class_kvs_mut,
+        lookup_configs, new_and_save_server_config, read_config, read_config_preserve_empty, read_config_with_metadata,
+        read_config_without_migrate, read_server_config_snapshot, save_server_config, save_server_config_snapshot,
+        save_server_config_snapshot_with_generation, server_config_transaction_lock_path, storage_class_kvs_mut,
     };
     use crate::config::{audit, heal, notify, oidc, scanner};
     use crate::disk::endpoint::Endpoint;
@@ -2311,7 +2566,9 @@ mod tests {
     use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
     use crate::runtime::sources as runtime_sources;
     use crate::set_disk::SetDisks;
-    use crate::storage_api_contracts::{admin::StorageAdminApi, namespace::NamespaceLocking as _, range::HTTPRangeSpec};
+    use crate::storage_api_contracts::{
+        admin::StorageAdminApi, namespace::NamespaceLocking as _, object::HTTPPreconditions, range::HTTPRangeSpec,
+    };
     use http::HeaderMap;
     use rustfs_config::audit::{AUDIT_AMQP_SUB_SYS, AUDIT_KAFKA_SUB_SYS, AUDIT_MQTT_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS};
     use rustfs_config::notify::{
@@ -3104,6 +3361,85 @@ mod tests {
         );
     }
 
+    #[test]
+    fn root_heal_null_decodes_as_no_override_and_is_canonicalized_on_save() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{"standard":"","rrs":""},
+          "heal":null,
+          "future_root":{"mode":"keep"},
+          "openid":{"default":{
+            "config_url":"https://issuer.example/.well-known/openid-configuration",
+            "client_id":"console",
+            "client_secret":"oidc-secret",
+            "future_provider_control":"keep"
+          }},
+          "notify":{"webhook":{"primary":{
+            "enable":true,
+            "endpoint":"https://notify.example/hook",
+            "auth_token":"notify-secret",
+            "future_notify_control":"keep"
+          }}},
+          "logger":{"webhook":{"primary":{
+            "enable":true,
+            "endpoint":"https://audit.example/hook",
+            "auth_token":"audit-secret",
+            "future_audit_control":"keep"
+          }}}
+        }"#;
+
+        let cfg = decode_server_config_blob(seed).expect("root heal null should mean no persisted override");
+        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none());
+        assert!(!is_standard_object_server_config(seed));
+
+        let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("legacy seed should canonicalize on an authorized save");
+        let value: Value = serde_json::from_slice(&encoded).expect("canonical config should be valid JSON");
+        assert!(value.get(HEAL_SUB_SYS).is_none());
+        assert_eq!(value["future_root"]["mode"].as_str(), Some("keep"));
+        assert_eq!(value["openid"]["default"]["client_secret"].as_str(), Some("oidc-secret"));
+        assert_eq!(value["openid"]["default"]["future_provider_control"].as_str(), Some("keep"));
+        assert_eq!(value["notify"]["webhook"]["primary"]["auth_token"].as_str(), Some("notify-secret"));
+        assert_eq!(value["notify"]["webhook"]["primary"]["future_notify_control"].as_str(), Some("keep"));
+        assert_eq!(value["logger"]["webhook"]["primary"]["auth_token"].as_str(), Some("audit-secret"));
+        assert_eq!(value["logger"]["webhook"]["primary"]["future_audit_control"].as_str(), Some("keep"));
+        assert!(is_standard_object_server_config(&encoded));
+    }
+
+    #[test]
+    fn invalid_scalar_and_nested_null_config_shapes_remain_rejected() {
+        let invalid_sections = [
+            r#""scanner":null"#,
+            r#""heal":"""#,
+            r#""heal":false"#,
+            r#""heal":0"#,
+            r#""heal":{"default":null}"#,
+            r#""heal":{"_":null}"#,
+            r#""heal":{"bitrot_cycle":null}"#,
+            r#""heal":[{"key":"bitrot_cycle","value":null}]"#,
+        ];
+
+        for section in invalid_sections {
+            let input = format!(r#"{{"version":"33","storageclass":{{"standard":"","rrs":""}},{section}}}"#);
+            let err = decode_server_config_blob(input.as_bytes()).expect_err("invalid scalar shape must remain rejected");
+            assert!(
+                err.to_string().contains("expected"),
+                "invalid section {section} returned an unrelated error: {err}"
+            );
+        }
+    }
+
+    #[test]
+    fn valid_heal_object_and_kvs_array_shapes_remain_accepted() {
+        let empty_object = br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":{}}"#;
+        let cfg = decode_server_config_blob(empty_object).expect("empty heal object should decode as no override");
+        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_none());
+
+        let kvs_array =
+            br#"{"version":"33","storageclass":{"standard":"","rrs":""},"heal":[{"key":"bitrot_cycle","value":"off"}]}"#;
+        let cfg = decode_server_config_blob(kvs_array).expect("heal KVS array should decode");
+        assert!(cfg.get_value(HEAL_SUB_SYS, DEFAULT_DELIMITER).is_some());
+    }
+
     #[test]
     fn scanner_update_preserves_unknown_root_and_oidc_provider_fields() {
         let seed = br#"{
@@ -3131,6 +3467,171 @@ mod tests {
         assert_eq!(value["openid"]["default"]["client_id"].as_str(), Some("console"));
     }
 
+    #[test]
+    fn storageclass_reset_removes_stale_inline_block_from_seed() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{
+            "standard":"EC:2",
+            "rrs":"EC:1",
+            "optimize":"availability",
+            "inline_block":"64KiB",
+            "future_storage_control":"keep"
+          }
+        }"#;
+
+        let encoded = encode_server_config_blob(&Config::new(), Some(seed)).expect("storageclass reset should encode");
+        let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json");
+        let storageclass = value["storageclass"].as_object().expect("storageclass object");
+
+        assert!(storageclass.get(crate::config::storageclass::INLINE_BLOCK).is_none());
+        assert_eq!(storageclass["future_storage_control"].as_str(), Some("keep"));
+    }
+
+    #[test]
+    fn target_update_preserves_unknown_fields_without_restoring_removed_instances() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{"standard":"","rrs":""},
+          "notify":{"webhook":{
+            "primary":{
+              "enable":true,
+              "endpoint":"https://notify.example/old",
+              "auth_token":"notify-secret",
+              "future_control":"keep"
+            },
+            "removed":{"enable":true,"endpoint":"https://notify.example/removed"},
+            "retained":{"enable":true,"endpoint":"https://notify.example/retained","future_control":"keep"},
+            "enable":{"enable":true,"endpoint":"https://notify.example/named-enable","future_control":"keep"}
+          }}
+        }"#;
+        let mut cfg = decode_server_config_blob(seed).expect("target seed should decode");
+        let webhook = cfg
+            .0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .expect("notify webhook subsystem should exist");
+        webhook
+            .get_mut("primary")
+            .expect("primary target should exist")
+            .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string());
+        webhook.remove("removed");
+        webhook.remove("retained");
+
+        let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("target update should encode");
+        let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json");
+        let webhook = value["notify"]["webhook"].as_object().expect("webhook section");
+
+        assert_eq!(webhook["primary"]["endpoint"].as_str(), Some("https://notify.example/new"));
+        assert_eq!(webhook["primary"]["auth_token"].as_str(), Some("notify-secret"));
+        assert_eq!(webhook["primary"]["future_control"].as_str(), Some("keep"));
+        assert!(webhook.get("removed").is_none());
+        assert_eq!(webhook["retained"]["future_control"].as_str(), Some("keep"));
+        assert!(webhook["retained"].get("enable").is_none());
+        assert!(webhook["retained"].get("endpoint").is_none());
+        assert_eq!(webhook["enable"]["endpoint"].as_str(), Some("https://notify.example/named-enable"));
+        assert_eq!(webhook["enable"]["future_control"].as_str(), Some("keep"));
+    }
+
+    #[test]
+    fn shorthand_target_update_preserves_shape_and_unknown_nested_fields() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{"standard":"","rrs":""},
+          "notify":{"webhook":{
+            "enable":true,
+            "endpoint":"https://notify.example/old",
+            "future_control":{"endpoint":"leave-untouched","mode":"keep"}
+          }}
+        }"#;
+        let mut cfg = decode_server_config_blob(seed).expect("shorthand target should decode");
+        cfg.0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .and_then(|targets| targets.get_mut(DEFAULT_DELIMITER))
+            .expect("default webhook target should exist")
+            .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string());
+
+        let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("shorthand target update should encode");
+        let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json");
+        let webhook = value["notify"]["webhook"].as_object().expect("webhook shorthand object");
+
+        assert_eq!(webhook["endpoint"].as_str(), Some("https://notify.example/new"));
+        assert!(webhook.get("default").is_none());
+        assert_eq!(webhook["future_control"]["endpoint"].as_str(), Some("leave-untouched"));
+        assert_eq!(webhook["future_control"]["mode"].as_str(), Some("keep"));
+        let decoded = decode_server_config_blob(&encoded).expect("updated shorthand target should remain decodable");
+        assert_eq!(
+            decoded
+                .get_value(NOTIFY_WEBHOOK_SUB_SYS, DEFAULT_DELIMITER)
+                .expect("updated default webhook target")
+                .get(rustfs_config::WEBHOOK_ENDPOINT),
+            "https://notify.example/new"
+        );
+    }
+
+    #[test]
+    fn target_kvs_update_preserves_unknown_entries_and_attributes() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{"standard":"","rrs":""},
+          "notify":{"webhook":{"primary":[
+            {"key":"enable","value":"on","hidden_if_empty":false},
+            {"key":"endpoint","value":"https://notify.example/old","future_attribute":"keep-endpoint"},
+            {"key":"future_control","value":"keep","future_attribute":"keep-control"}
+          ]}}
+        }"#;
+        let mut cfg = decode_server_config_blob(seed).expect("target KVS seed should decode");
+        cfg.0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .and_then(|targets| targets.get_mut("primary"))
+            .expect("primary webhook target should exist")
+            .insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "https://notify.example/new".to_string());
+
+        let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("target KVS update should encode");
+        let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json");
+        let entries = value["notify"]["webhook"]["primary"]
+            .as_array()
+            .expect("target KVS shape should be preserved");
+        let endpoint = entries
+            .iter()
+            .find(|entry| entry["key"].as_str() == Some(rustfs_config::WEBHOOK_ENDPOINT))
+            .expect("endpoint entry should remain");
+        let future = entries
+            .iter()
+            .find(|entry| entry["key"].as_str() == Some("future_control"))
+            .expect("unknown target entry should remain");
+
+        assert_eq!(endpoint["value"].as_str(), Some("https://notify.example/new"));
+        assert_eq!(endpoint["future_attribute"].as_str(), Some("keep-endpoint"));
+        assert_eq!(future["value"].as_str(), Some("keep"));
+        assert_eq!(future["future_attribute"].as_str(), Some("keep-control"));
+    }
+
+    #[test]
+    fn target_default_alias_is_canonicalized_without_losing_unknown_fields() {
+        let seed = br#"{
+          "version":"33",
+          "storageclass":{"standard":"","rrs":""},
+          "notify":{"webhook":{
+            "_":{"enable":false,"endpoint":"https://notify.example/alias","future_alias":"keep"},
+            "default":{"enable":true,"endpoint":"https://notify.example/default","future_default":"keep"}
+          }}
+        }"#;
+        let cfg = decode_server_config_blob(seed).expect("dual default aliases should decode");
+        let expected_endpoint = cfg
+            .get_value(NOTIFY_WEBHOOK_SUB_SYS, DEFAULT_DELIMITER)
+            .expect("default webhook target should exist")
+            .get(rustfs_config::WEBHOOK_ENDPOINT);
+
+        let encoded = encode_server_config_blob(&cfg, Some(seed)).expect("default alias should canonicalize");
+        let value: Value = serde_json::from_slice(&encoded).expect("encoded config should be valid json");
+        let webhook = value["notify"]["webhook"].as_object().expect("webhook section");
+
+        assert!(webhook.get(DEFAULT_DELIMITER).is_none());
+        assert_eq!(webhook["default"]["endpoint"].as_str(), Some(expected_endpoint.as_str()));
+        assert_eq!(webhook["default"]["future_alias"].as_str(), Some("keep"));
+        assert_eq!(webhook["default"]["future_default"].as_str(), Some("keep"));
+    }
+
     #[test]
     fn test_scanner_config_changes_are_semantically_significant() {
         let baseline = Config::new();
@@ -4124,6 +4625,7 @@ mod tests {
 
     /// What reads of the config object currently return.
     enum RecoveryReadState {
+        Missing,
         Blob(Vec<u8>),
         QuorumError,
     }
@@ -4136,6 +4638,7 @@ mod tests {
         heal_calls: AtomicUsize,
         write_calls: AtomicUsize,
         last_put_no_lock: AtomicBool,
+        last_put_preconditions: Mutex<Option<HTTPPreconditions>>,
         revision: AtomicUsize,
         drive_counts: Vec<usize>,
         lock_manager: Arc<rustfs_lock::GlobalLockManager>,
@@ -4150,6 +4653,7 @@ mod tests {
                 heal_calls: AtomicUsize::new(0),
                 write_calls: AtomicUsize::new(0),
                 last_put_no_lock: AtomicBool::new(false),
+                last_put_preconditions: Mutex::new(None),
                 revision: AtomicUsize::new(1),
                 drive_counts: vec![2],
                 lock_manager: Arc::new(rustfs_lock::GlobalLockManager::new()),
@@ -4206,6 +4710,7 @@ mod tests {
             _opts: &ObjectOptions,
         ) -> Result<GetObjectReader> {
             let data = match &*self.state.lock().expect("state lock poisoned") {
+                RecoveryReadState::Missing => return Err(Error::ConfigNotFound),
                 RecoveryReadState::Blob(data) => data.clone(),
                 RecoveryReadState::QuorumError => return Err(Error::ErasureReadQuorum),
             };
@@ -4213,6 +4718,9 @@ mod tests {
                 size: data.len() as i64,
                 actual_size: data.len() as i64,
                 etag: Some(format!("config-{}", self.revision.load(Ordering::SeqCst))),
+                data_dir: Some(uuid::Uuid::from_u128(
+                    u128::try_from(self.revision.load(Ordering::SeqCst)).expect("test revision should fit in u128"),
+                )),
                 ..Default::default()
             };
             Ok(GetObjectReader {
@@ -4231,15 +4739,19 @@ mod tests {
             opts: &ObjectOptions,
         ) -> Result<ObjectInfo> {
             let current_etag = format!("config-{}", self.revision.load(Ordering::SeqCst));
+            let object_exists = matches!(&*self.state.lock().expect("state lock poisoned"), RecoveryReadState::Blob(_));
             if let Some(preconditions) = &opts.http_preconditions
-                && (preconditions.if_match_value().is_some_and(|etag| etag != current_etag)
-                    || preconditions.if_none_match_value() == Some("*"))
+                && (preconditions
+                    .if_match_value()
+                    .is_some_and(|etag| !object_exists || etag != current_etag)
+                    || (object_exists && preconditions.if_none_match_value() == Some("*")))
             {
                 return Err(Error::PreconditionFailed);
             }
             let mut body = Vec::new();
             data.stream.read_to_end(&mut body).await?;
             self.last_put_no_lock.store(opts.no_lock, Ordering::SeqCst);
+            *self.last_put_preconditions.lock().expect("preconditions lock poisoned") = opts.http_preconditions.clone();
             self.write_calls.fetch_add(1, Ordering::SeqCst);
             *self.state.lock().expect("state lock poisoned") = RecoveryReadState::Blob(body.clone());
             let revision = self.revision.fetch_add(1, Ordering::SeqCst) + 1;
@@ -4247,6 +4759,7 @@ mod tests {
                 size: i64::try_from(body.len()).expect("test config should fit in i64"),
                 actual_size: i64::try_from(body.len()).expect("test config should fit in i64"),
                 etag: Some(format!("config-{revision}")),
+                data_dir: Some(uuid::Uuid::from_u128(u128::try_from(revision).expect("test revision should fit in u128"))),
                 ..Default::default()
             })
         }
@@ -4284,10 +4797,18 @@ mod tests {
             .expect("scanner-only config change should be persisted");
 
         assert_eq!(store.write_calls.load(Ordering::SeqCst), 1);
-        assert!(store.last_put_no_lock.load(Ordering::SeqCst));
+        assert!(!store.last_put_no_lock.load(Ordering::SeqCst));
+        let preconditions = store
+            .last_put_preconditions
+            .lock()
+            .expect("preconditions lock poisoned")
+            .clone()
+            .expect("existing config update must be conditional");
+        assert_eq!(preconditions.if_match_value(), Some("config-1"));
+        assert_eq!(preconditions.if_none_match_value(), None);
         assert_eq!(
             store.lock_resources.lock().expect("lock resources mutex poisoned").as_slice(),
-            &[server_config_path()]
+            &[server_config_transaction_lock_path()]
         );
         let decoded = read_config_without_migrate(store)
             .await
@@ -4301,6 +4822,73 @@ mod tests {
         );
     }
 
+    #[tokio::test]
+    async fn server_config_snapshot_save_returns_committed_generation() {
+        let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode");
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None));
+        let snapshot = read_server_config_snapshot(store.clone())
+            .await
+            .expect("server config snapshot");
+
+        let result = save_server_config_snapshot_with_generation(store, &config_with_scanner_cycle("61"), &snapshot)
+            .await
+            .expect("conditional config save");
+
+        assert!(result.persisted());
+        assert_eq!(result.generation(), Some(uuid::Uuid::from_u128(2)));
+    }
+
+    #[tokio::test]
+    async fn missing_server_config_is_created_with_if_none_match() {
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Missing, None));
+        let cfg = config_with_scanner_cycle("61");
+
+        save_server_config(store.clone(), &cfg)
+            .await
+            .expect("missing config should be created conditionally");
+
+        assert_eq!(store.write_calls.load(Ordering::SeqCst), 1);
+        assert!(!store.last_put_no_lock.load(Ordering::SeqCst));
+        let preconditions = store
+            .last_put_preconditions
+            .lock()
+            .expect("preconditions lock poisoned")
+            .clone()
+            .expect("missing config create must be conditional");
+        assert_eq!(preconditions.if_match_value(), None);
+        assert_eq!(preconditions.if_none_match_value(), Some("*"));
+        let persisted = read_config_without_migrate(store)
+            .await
+            .expect("created config should reload");
+        assert_eq!(
+            persisted
+                .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER)
+                .expect("persisted scanner config")
+                .get(SCANNER_CYCLE),
+            "61"
+        );
+    }
+
+    #[tokio::test]
+    async fn missing_config_initialization_recheck_preserves_concurrent_config() {
+        let existing = config_with_scanner_cycle("73");
+        let baseline = encode_server_config_blob(&existing, None).expect("existing config should encode");
+        let store = Arc::new(RecoveryMockStore::new(RecoveryReadState::Blob(baseline), None));
+
+        let observed = new_and_save_server_config(store.clone())
+            .await
+            .expect("initialization recheck should return the config created by another writer");
+
+        assert_eq!(store.write_calls.load(Ordering::SeqCst), 0);
+        assert_eq!(
+            observed
+                .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER)
+                .expect("concurrent scanner config")
+                .get(SCANNER_CYCLE),
+            "73"
+        );
+    }
+
     #[tokio::test]
     async fn stale_server_config_snapshot_cannot_overwrite_newer_update() {
         let baseline = encode_server_config_blob(&Config::new(), None).expect("baseline config should encode");
@@ -4357,7 +4945,7 @@ mod tests {
         let lock = rustfs_lock::NamespaceLock::new("server-config-lease-loss".to_string(), client.clone());
         let guard = lock
             .lock_guard(
-                rustfs_lock::ObjectKey::new(crate::disk::RUSTFS_META_BUCKET, server_config_path()),
+                rustfs_lock::ObjectKey::new(crate::disk::RUSTFS_META_BUCKET, server_config_transaction_lock_path()),
                 "server-config-lease-loss",
                 std::time::Duration::from_secs(1),
                 std::time::Duration::from_millis(120),
@@ -4372,6 +4960,7 @@ mod tests {
             raw: Some(baseline.clone()),
             seed: None,
             etag: Some("config-0".to_string()),
+            generation: Some(uuid::Uuid::from_u128(1)),
             _local_guard: local_guard,
             _guard: guard,
         };

crates/ecstore/src/set_disk/mod.rs

@@ -4714,6 +4714,7 @@ mod tests {
     use crate::layout::endpoints::SetupType;
     use crate::object_api::BLOCK_SIZE_V2;
     use crate::object_api::ObjectInfo;
+    use crate::set_disk::core::io_primitives::rename_fanout_barrier;
     use crate::storage_api_contracts::{
         heal::HealOperations as _, lifecycle::TransitionedObject, list::ListOperations as _, multipart::CompletePart,
         namespace::NamespaceLocking as _, object::ObjectIO as _, object::ObjectOperations as _,
@@ -9565,6 +9566,15 @@ mod tests {
         make_local_bucket_test_set_disks_with_drive_count(2).await
     }
 
+    fn assert_exclusive_object_lock_held(set_disks: &SetDisks, bucket: &str, object: &str) {
+        let lock = set_disks
+            .local_lock_manager_for_test()
+            .get_lock_info(&ObjectKey::new(bucket, object))
+            .expect("object lock should be visible while rename is paused");
+        assert!(matches!(lock.mode, rustfs_lock::LockMode::Exclusive));
+        assert_eq!(lock.owner.as_ref(), set_disks.locker_owner.as_str());
+    }
+
     async fn make_local_bucket_test_set_disks_with_drive_count(drive_count: usize) -> Arc<SetDisks> {
         let format = FormatV3::new(1, drive_count);
         let mut endpoints = Vec::new();
@@ -9599,7 +9609,9 @@ mod tests {
             disks.push(Some(disk));
         }
 
-        let set_disks = SetDisks::new(
+        let instance_ctx = Arc::new(InstanceContext::new());
+        instance_ctx.update_erasure_type(SetupType::Erasure).await;
+        let set_disks = SetDisks::new_with_instance_ctx(
             "test-owner".to_string(),
             Arc::new(RwLock::new(disks)),
             drive_count,
@@ -9609,6 +9621,7 @@ mod tests {
             endpoints,
             format,
             Vec::new(),
+            instance_ctx,
         )
         .await;
         set_disks.set_test_storage_class_config(
@@ -10262,6 +10275,216 @@ mod tests {
         ));
     }
 
+    #[tokio::test]
+    async fn conditional_replace_holds_object_lock_through_rename() {
+        let set_disks = make_local_bucket_test_set_disks().await;
+        let bucket = "bucket-conditional-replace-fence";
+        let object = "config/conditional-replace.json";
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+
+        let mut initial_reader = PutObjReader::from_vec(b"initial config".to_vec());
+        let initial = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut initial_reader,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("initial config should be written");
+        let initial_etag = initial.etag.expect("initial config should have an ETag");
+
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let writer_store = set_disks.clone();
+        let expected_etag = initial_etag.clone();
+        let writer = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(b"replacement config".to_vec());
+            writer_store
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        preserve_etag: Some("replacement-etag".to_string()),
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: Some(expected_etag),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        tokio::time::timeout(std::time::Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("conditional replace should reach the rename barrier");
+        assert_exclusive_object_lock_held(&set_disks, bucket, object);
+        barrier.release();
+        writer
+            .await
+            .expect("conditional writer task should finish")
+            .expect("matching conditional replace should commit");
+        assert!(
+            set_disks
+                .local_lock_manager_for_test()
+                .get_lock_info(&ObjectKey::new(bucket, object))
+                .is_none(),
+            "conditional replace should release the object lock after commit"
+        );
+        let contender = set_disks
+            .new_ns_lock(bucket, object)
+            .await
+            .expect("contender namespace lock should be created");
+        let contender_guard = contender
+            .get_write_lock(std::time::Duration::from_secs(30))
+            .await
+            .expect("contender should acquire after conditional replace commits");
+        drop(contender_guard);
+
+        let mut stale_reader = PutObjReader::from_vec(b"stale config".to_vec());
+        let err = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut stale_reader,
+                &ObjectOptions {
+                    http_preconditions: Some(HTTPPreconditions {
+                        if_match: Some(initial_etag),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("the old ETag must fail after the fenced replacement commits");
+        assert_eq!(err, StorageError::PreconditionFailed);
+    }
+
+    #[tokio::test]
+    async fn repeated_body_write_keeps_etag_but_changes_data_dir_generation() {
+        let set_disks = make_local_bucket_test_set_disks().await;
+        let bucket = "bucket-write-generation";
+        let object = "config/write-generation.json";
+        let body = b"identical config body".to_vec();
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+
+        let mut first_reader = PutObjReader::from_vec(body.clone());
+        let first = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut first_reader,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("first config body should be written");
+        let mut second_reader = PutObjReader::from_vec(body);
+        let second = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut second_reader,
+                &ObjectOptions {
+                    no_lock: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("identical config body should be rewritten");
+
+        assert_eq!(first.etag, second.etag, "content ETag should expose the ABA collision");
+        assert_ne!(first.data_dir, second.data_dir, "each committed body write needs a unique generation");
+        assert!(first.data_dir.is_some() && second.data_dir.is_some());
+    }
+
+    #[tokio::test]
+    async fn conditional_create_holds_object_lock_through_rename() {
+        let set_disks = make_local_bucket_test_set_disks().await;
+        let bucket = "bucket-conditional-create-fence";
+        let object = "config/conditional-create.json";
+        set_disks
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let writer_store = set_disks.clone();
+        let writer = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(b"created config".to_vec());
+            writer_store
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_none_match: Some("*".to_string()),
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+
+        tokio::time::timeout(std::time::Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("conditional create should reach the rename barrier");
+        assert_exclusive_object_lock_held(&set_disks, bucket, object);
+        barrier.release();
+        writer
+            .await
+            .expect("conditional writer task should finish")
+            .expect("first conditional create should commit");
+        assert!(
+            set_disks
+                .local_lock_manager_for_test()
+                .get_lock_info(&ObjectKey::new(bucket, object))
+                .is_none(),
+            "conditional create should release the object lock after commit"
+        );
+        let contender = set_disks
+            .new_ns_lock(bucket, object)
+            .await
+            .expect("contender namespace lock should be created");
+        let contender_guard = contender
+            .get_write_lock(std::time::Duration::from_secs(30))
+            .await
+            .expect("contender should acquire after conditional create commits");
+        drop(contender_guard);
+
+        let mut duplicate_reader = PutObjReader::from_vec(b"duplicate config".to_vec());
+        let err = set_disks
+            .put_object(
+                bucket,
+                object,
+                &mut duplicate_reader,
+                &ObjectOptions {
+                    http_preconditions: Some(HTTPPreconditions {
+                        if_none_match: Some("*".to_string()),
+                        ..Default::default()
+                    }),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("a second create-only write must not replace the committed config");
+        assert_eq!(err, StorageError::PreconditionFailed);
+    }
+
     #[tokio::test]
     async fn set_level_if_none_match_fails_closed_without_read_quorum() {
         let set_disks = make_local_bucket_test_set_disks_with_drive_count(4).await;

crates/notify/src/config_manager.rs

@@ -19,7 +19,7 @@ use crate::{
     resolve_notify_object_store_handle,
     rule_engine::NotifyRuleEngine,
     runtime_facade::NotifyRuntimeFacade,
-    with_notify_server_config_read_lock, with_notify_server_config_write_lock,
+    with_notify_server_config_read_lock,
 };
 use rustfs_config::notify::{
     NOTIFY_AMQP_SUB_SYS, NOTIFY_KAFKA_SUB_SYS, NOTIFY_MQTT_SUB_SYS, NOTIFY_MYSQL_SUB_SYS, NOTIFY_NATS_SUB_SYS,
@@ -41,12 +41,32 @@ enum NotifyConfigStoreError {
     StorageNotAvailable,
     Read(String),
     Save(String),
+    Converge { persisted: bool, error: String },
+}
+
+fn notification_convergence_error(persisted: bool, error: impl std::fmt::Display) -> NotificationError {
+    let durable_state = if persisted { "persisted" } else { "unchanged" };
+    NotificationError::Configuration(format!(
+        "configuration was {durable_state} but notification runtime convergence failed: {error}"
+    ))
+}
+
+async fn supervise_config_update<T>(
+    mutation: impl std::future::Future<Output = Result<T, NotifyConfigStoreError>> + Send + 'static,
+) -> Result<T, NotifyConfigStoreError>
+where
+    T: Send + 'static,
+{
+    tokio::spawn(mutation).await.map_err(|error| {
+        let outcome = if error.is_cancelled() { "cancelled" } else { "panicked" };
+        NotifyConfigStoreError::Lock(format!("notify config update task {outcome}"))
+    })?
 }
 
 async fn update_server_config<F>(
-    modifier: F,
+    mut modifier: F,
     lifecycle: NotifyLifecycleCoordinator,
-) -> Result<Option<crate::lifecycle::NotificationLifecycleTransition>, NotifyConfigStoreError>
+) -> Result<Option<(crate::lifecycle::NotificationLifecycleTransition, bool)>, NotifyConfigStoreError>
 where
     F: FnMut(&mut Config) -> bool + Send + 'static,
 {
@@ -54,51 +74,31 @@ where
         return Err(NotifyConfigStoreError::StorageNotAvailable);
     };
 
-    let store_for_read = store.clone();
-    let store_for_save = store.clone();
-    with_notify_server_config_write_lock(store, move || {
-        read_modify_write(
-            modifier,
-            move || async move {
-                crate::read_notify_server_config_without_migrate_no_lock(store_for_read)
-                    .await
-                    .map_err(NotifyConfigStoreError::Read)
-            },
-            move |config| async move {
-                crate::save_notify_server_config_no_lock(store_for_save, &config)
-                    .await
-                    .map_err(NotifyConfigStoreError::Save)
-            },
-            move |config| lifecycle.update_config(config),
-        )
+    supervise_config_update(async move {
+        let snapshot = crate::read_notify_server_config_snapshot(store.clone())
+            .await
+            .map_err(NotifyConfigStoreError::Read)?;
+        let mut config = snapshot.config.clone();
+        if !modifier(&mut config) {
+            return Ok(None);
+        }
+
+        let persisted = crate::save_notify_server_config_snapshot(store.clone(), &config, &snapshot)
+            .await
+            .map_err(NotifyConfigStoreError::Save)?;
+        drop(snapshot);
+
+        let read_store = store.clone();
+        with_notify_server_config_read_lock(store, move || async move {
+            let latest = crate::read_existing_notify_server_config_no_lock(read_store)
+                .await
+                .map_err(|error| NotifyConfigStoreError::Converge { persisted, error })?;
+            Ok::<_, NotifyConfigStoreError>(Some((lifecycle.update_config(latest), persisted)))
+        })
+        .await
+        .map_err(|error| NotifyConfigStoreError::Converge { persisted, error })?
     })
     .await
-    .map_err(NotifyConfigStoreError::Lock)?
-}
-
-async fn read_modify_write<F, R, RFut, S, SFut, P, T>(
-    mut modifier: F,
-    read: R,
-    save: S,
-    publish: P,
-) -> Result<Option<T>, NotifyConfigStoreError>
-where
-    F: FnMut(&mut Config) -> bool,
-    R: FnOnce() -> RFut,
-    RFut: std::future::Future<Output = Result<Config, NotifyConfigStoreError>>,
-    S: FnOnce(Config) -> SFut,
-    SFut: std::future::Future<Output = Result<(), NotifyConfigStoreError>>,
-    P: FnOnce(Config) -> T,
-{
-    let mut new_config = read().await?;
-
-    if !modifier(&mut new_config) {
-        return Ok(None);
-    }
-
-    save(new_config.clone()).await?;
-
-    Ok(Some(publish(new_config)))
 }
 
 pub(crate) fn notify_configuration_hint() -> String {
@@ -345,16 +345,18 @@ impl NotifyConfigManager {
         if self.lifecycle.state() == NotificationRuntimeState::Terminated {
             return Err(NotificationError::Initialization("Notification runtime has terminated".to_string()));
         }
-        let Some(transition) = update_server_config(modifier, self.lifecycle.clone())
-            .await
-            .map_err(|err| match err {
-                NotifyConfigStoreError::Lock(err) => NotificationError::StorageNotAvailable(err),
-                NotifyConfigStoreError::StorageNotAvailable => NotificationError::StorageNotAvailable(
-                    "Failed to save target configuration: server storage not initialized".to_string(),
-                ),
-                NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err),
-                NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err),
-            })?
+        let Some((transition, persisted)) =
+            update_server_config(modifier, self.lifecycle.clone())
+                .await
+                .map_err(|err| match err {
+                    NotifyConfigStoreError::Lock(err) => NotificationError::StorageNotAvailable(err),
+                    NotifyConfigStoreError::StorageNotAvailable => NotificationError::StorageNotAvailable(
+                        "Failed to save target configuration: server storage not initialized".to_string(),
+                    ),
+                    NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err),
+                    NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err),
+                    NotifyConfigStoreError::Converge { persisted, error } => notification_convergence_error(persisted, error),
+                })?
         else {
             debug!(
                 event = EVENT_NOTIFY_CONFIG_UPDATE,
@@ -367,23 +369,53 @@ impl NotifyConfigManager {
             return Ok(());
         };
 
+        let result = if persisted { "updated" } else { "unchanged" };
         info!(
             event = EVENT_NOTIFY_CONFIG_UPDATE,
             component = LOG_COMPONENT_NOTIFY,
             subsystem = LOG_SUBSYSTEM_CONFIG,
             action = "reload_if_changed",
-            result = "updated",
+            result,
             "notify config update"
         );
-        transition.wait().await
+        transition
+            .wait()
+            .await
+            .map_err(|err| notification_convergence_error(persisted, err))
     }
 }
 
 #[cfg(test)]
 mod tests {
-    use super::{NotifyConfigManager, NotifyConfigStoreError, read_modify_write, runtime_target_id_for_subsystem};
+    use super::{
+        NotifyConfigManager, NotifyConfigStoreError, notification_convergence_error, runtime_target_id_for_subsystem,
+        supervise_config_update,
+    };
     use crate::rules::RulesMap;
     use crate::{NotificationError, NotificationRuntimeState};
+
+    #[tokio::test]
+    async fn supervised_config_update_survives_waiter_cancellation() {
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
+
+        let waiter = tokio::spawn(async move {
+            supervise_config_update(async move {
+                let _ = started_tx.send(());
+                let _ = release_rx.await;
+                let _ = completed_tx.send(());
+                Ok::<_, NotifyConfigStoreError>(())
+            })
+            .await
+        });
+
+        started_rx.await.expect("config update should start");
+        waiter.abort();
+        release_tx.send(()).expect("config update should be released");
+        let completed = tokio::time::timeout(std::time::Duration::from_secs(30), completed_rx).await;
+        assert!(matches!(completed, Ok(Ok(()))), "detached config update should complete");
+    }
     use crate::{
         integration::NotificationMetrics, notifier::EventNotifier, registry::TargetRegistry, rule_engine::NotifyRuleEngine,
         runtime_facade::NotifyRuntimeFacade,
@@ -392,14 +424,11 @@ 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, KVS};
+    use rustfs_config::server_config::Config;
     use rustfs_s3_types::EventName;
     use rustfs_targets::ReplayWorkerManager;
     use rustfs_targets::arn::TargetID;
-    use std::sync::{
-        Arc,
-        atomic::{AtomicBool, Ordering},
-    };
+    use std::sync::Arc;
     use tokio::sync::{RwLock, Semaphore};
 
     fn build_manager() -> NotifyConfigManager {
@@ -463,38 +492,6 @@ mod tests {
         assert!(matches!(manager.lifecycle().state(), NotificationRuntimeState::TargetsEnabled { .. }));
     }
 
-    #[tokio::test]
-    async fn read_modify_write_publishes_only_after_save() {
-        let saved = Arc::new(AtomicBool::new(false));
-        let saved_by_writer = saved.clone();
-        let observed_by_publisher = saved.clone();
-
-        read_modify_write(
-            |config| {
-                config
-                    .0
-                    .entry(NOTIFY_WEBHOOK_SUB_SYS.to_string())
-                    .or_default()
-                    .insert("primary".to_string(), KVS::default());
-                true
-            },
-            || async { Ok::<_, NotifyConfigStoreError>(Config::default()) },
-            move |_config| async move {
-                saved_by_writer.store(true, Ordering::Release);
-                Ok::<_, NotifyConfigStoreError>(())
-            },
-            move |_config| {
-                assert!(
-                    observed_by_publisher.load(Ordering::Acquire),
-                    "publication must observe the completed save"
-                );
-            },
-        )
-        .await
-        .expect("read-modify-write should succeed")
-        .expect("changed config should publish");
-    }
-
     #[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");
@@ -550,4 +547,16 @@ mod tests {
         assert_eq!(target_id.id, "audittrail");
         assert_eq!(target_id.name, "postgres");
     }
+
+    #[test]
+    fn convergence_error_reports_durable_write_state() {
+        for (persisted, expected) in [(true, "configuration was persisted"), (false, "configuration was unchanged")] {
+            let error = notification_convergence_error(persisted, "injected failure");
+            let NotificationError::Configuration(message) = error else {
+                panic!("convergence error should be a configuration error");
+            };
+            assert!(message.contains(expected), "unexpected convergence error: {message}");
+            assert!(message.contains("injected failure"));
+        }
+    }
 }

crates/notify/src/lib.rs

@@ -57,7 +57,6 @@ pub use services::NotifyServices;
 pub use status_view::NotifyStatusView;
 pub use storage_api::NotifyStore;
 pub(crate) use storage_api::crate_boundary::{
-    read_existing_notify_server_config_no_lock, read_notify_server_config_without_migrate_no_lock,
-    resolve_notify_object_store_handle, save_notify_server_config_no_lock, with_notify_server_config_read_lock,
-    with_notify_server_config_write_lock,
+    read_existing_notify_server_config_no_lock, read_notify_server_config_snapshot, resolve_notify_object_store_handle,
+    save_notify_server_config_snapshot, with_notify_server_config_read_lock,
 };

crates/notify/src/storage_api.rs

@@ -15,11 +15,10 @@
 use std::sync::Arc;
 
 use rustfs_ecstore::api::config::com::{
-    read_config_without_migrate_no_lock as read_notify_config_without_migrate_from_backend_no_lock,
     read_existing_server_config_no_lock as read_existing_notify_config_from_backend_no_lock,
-    save_server_config_no_lock as save_notify_server_config_to_backend_no_lock,
+    read_server_config_snapshot as read_notify_server_config_snapshot_from_backend,
+    save_server_config_snapshot as save_notify_server_config_snapshot_to_backend,
     with_server_config_read_lock as with_notify_server_config_read_lock_from_backend,
-    with_server_config_write_lock as with_notify_server_config_write_lock_from_backend,
 };
 use rustfs_ecstore::api::runtime::object_store_handle as resolve_notify_object_store_handle_from_backend;
 pub use rustfs_ecstore::api::storage::ECStore as NotifyStore;
@@ -28,10 +27,10 @@ pub(crate) fn resolve_notify_object_store_handle() -> Option<Arc<NotifyStore>> {
     resolve_notify_object_store_handle_from_backend()
 }
 
-pub(crate) async fn read_notify_server_config_without_migrate_no_lock(
-    store: Arc<NotifyStore>,
-) -> Result<rustfs_config::server_config::Config, String> {
-    read_notify_config_without_migrate_from_backend_no_lock(store)
+pub(crate) type NotifyServerConfigSnapshot = rustfs_ecstore::api::config::com::ServerConfigSnapshot;
+
+pub(crate) async fn read_notify_server_config_snapshot(store: Arc<NotifyStore>) -> Result<NotifyServerConfigSnapshot, String> {
+    read_notify_server_config_snapshot_from_backend(store)
         .await
         .map_err(|err| err.to_string())
 }
@@ -44,22 +43,12 @@ pub(crate) async fn read_existing_notify_server_config_no_lock(
         .map_err(|err| err.to_string())
 }
 
-pub(crate) async fn save_notify_server_config_no_lock(
+pub(crate) async fn save_notify_server_config_snapshot(
     store: Arc<NotifyStore>,
     config: &rustfs_config::server_config::Config,
-) -> Result<(), String> {
-    save_notify_server_config_to_backend_no_lock(store, config)
-        .await
-        .map_err(|err| err.to_string())
-}
-
-pub(crate) async fn with_notify_server_config_write_lock<F, Fut, T>(store: Arc<NotifyStore>, operation: F) -> Result<T, String>
-where
-    F: FnOnce() -> Fut + Send + 'static,
-    Fut: std::future::Future<Output = T> + Send + 'static,
-    T: Send + 'static,
-{
-    with_notify_server_config_write_lock_from_backend(store, operation)
+    snapshot: &NotifyServerConfigSnapshot,
+) -> Result<bool, String> {
+    save_notify_server_config_snapshot_to_backend(store, config, snapshot)
         .await
         .map_err(|err| err.to_string())
 }
@@ -77,8 +66,7 @@ where
 
 pub(crate) mod crate_boundary {
     pub(crate) use super::{
-        read_existing_notify_server_config_no_lock, read_notify_server_config_without_migrate_no_lock,
-        resolve_notify_object_store_handle, save_notify_server_config_no_lock, with_notify_server_config_read_lock,
-        with_notify_server_config_write_lock,
+        read_existing_notify_server_config_no_lock, read_notify_server_config_snapshot, resolve_notify_object_store_handle,
+        save_notify_server_config_snapshot, with_notify_server_config_read_lock,
     };
 }

rustfs/src/admin/handlers/audit_runtime_config.rs

@@ -15,13 +15,15 @@
 use crate::admin::handlers::supervise_admin_mutation;
 use crate::admin::handlers::target_descriptor::AdminTargetSpec;
 use crate::admin::runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context};
+use crate::admin::service::config::with_runtime_config_reload_lock;
 use crate::admin::storage_api::config::{
-    read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_server_config_snapshot,
+    read_admin_config_without_migrate, read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock,
+    save_admin_server_config_snapshot, with_admin_server_config_read_lock,
 };
 use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system, system::AuditSystemState};
 use rustfs_config::DEFAULT_DELIMITER;
 use rustfs_config::server_config::Config;
-use s3s::{S3Result, s3_error};
+use s3s::{S3Error, S3Result, s3_error};
 use tracing::warn;
 
 pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result<Config> {
@@ -48,6 +50,14 @@ fn has_any_audit_targets(specs: &[AdminTargetSpec], config: &Config) -> bool {
     })
 }
 
+fn audit_config_convergence_error(persisted: bool, error: impl std::fmt::Display) -> S3Error {
+    if persisted {
+        s3_error!(InternalError, "audit config persisted but runtime convergence failed: {}", error)
+    } else {
+        s3_error!(InternalError, "audit config unchanged but runtime convergence failed: {}", error)
+    }
+}
+
 pub(crate) async fn apply_audit_runtime_config(specs: &[AdminTargetSpec], config: Config) -> S3Result<()> {
     let has_targets = has_any_audit_targets(specs, &config);
 
@@ -107,15 +117,23 @@ where
             return Ok(());
         }
 
-        save_admin_server_config_snapshot(store, &config, &snapshot)
+        let persisted = save_admin_server_config_snapshot(store.clone(), &config, &snapshot)
             .await
-            .map(|_| ())
-            .map_err(|e| s3_error!(InternalError, "failed to save audit config: {}", e))?;
+            .map_err(|e| s3_error!(InternalError, "failed to save audit config: {}", e))?
+            .persisted();
+        drop(snapshot);
 
-        // Keep persistence and runtime publication in one detached, serialized
-        // mutation. Otherwise a cancelled caller or two concurrent updates can
-        // leave the persisted config and active audit generation disagreeing.
-        apply_audit_runtime_config(&specs, config).await
+        let read_store = store.clone();
+        with_runtime_config_reload_lock(async move {
+            let latest = with_admin_server_config_read_lock(store, move || read_existing_admin_server_config_no_lock(read_store))
+                .await
+                .map_err(|e| s3_error!(InternalError, "failed to lock server config for audit reload: {}", e))?
+                .map_err(|e| s3_error!(InternalError, "failed to read latest server config for audit reload: {}", e))?;
+
+            apply_audit_runtime_config(&specs, latest).await
+        })
+        .await
+        .map_err(|e| audit_config_convergence_error(persisted, e))
     })
     .await
 }
@@ -164,3 +182,154 @@ pub(crate) async fn remove_audit_target_config(specs: &[AdminTargetSpec], subsys
     })
     .await
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::admin::handlers::target_descriptor::admin_target_spec_from_builtin;
+    use crate::admin::runtime_sources::{IamInterface, KmsInterface};
+    use crate::admin::storage_api::config::save_admin_server_config;
+    use rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS;
+    use rustfs_config::server_config::KVS;
+    use rustfs_config::{ENABLE_KEY, EnableState, SCANNER_CYCLE, SCANNER_SUB_SYS, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR};
+    use rustfs_iam::{store::object::ObjectStore, sys::IamSys};
+    use rustfs_kms::KmsServiceManager;
+    use rustfs_targets::catalog::builtin::builtin_audit_target_admin_descriptors;
+    use std::sync::Arc;
+    use std::time::Duration;
+    use tempfile::TempDir;
+
+    struct TestIam;
+
+    impl IamInterface for TestIam {
+        fn handle(&self) -> Arc<IamSys<ObjectStore>> {
+            unreachable!("audit config tests do not use IAM")
+        }
+
+        fn is_ready(&self) -> bool {
+            false
+        }
+    }
+
+    struct TestKms;
+
+    impl KmsInterface for TestKms {
+        fn handle(&self) -> Arc<KmsServiceManager> {
+            Arc::new(KmsServiceManager::new())
+        }
+    }
+
+    fn audit_specs() -> Vec<AdminTargetSpec> {
+        builtin_audit_target_admin_descriptors()
+            .into_iter()
+            .map(|descriptor| admin_target_spec_from_builtin(&descriptor))
+            .collect()
+    }
+
+    async fn wait_for_persisted_target(store: Arc<crate::admin::storage_api::runtime::ECStore>, subsystem: &str, target: &str) {
+        tokio::time::timeout(Duration::from_secs(30), async {
+            let mut poll = tokio::time::interval(Duration::from_millis(10));
+            loop {
+                poll.tick().await;
+                let config = read_admin_config_without_migrate(store.clone())
+                    .await
+                    .expect("read persisted server config");
+                if config.0.get(subsystem).is_some_and(|targets| targets.contains_key(target)) {
+                    return;
+                }
+            }
+        })
+        .await
+        .expect("config mutation should become durable");
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial_test::serial]
+    async fn audit_reload_reads_latest_durable_config_after_releasing_write_snapshot() {
+        let temp_dir = TempDir::new().expect("audit config temp dir");
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .base_dir(temp_dir.path())
+            .disk_count(1)
+            .init_bucket_metadata(false)
+            .build()
+            .await;
+        save_admin_server_config(env.ecstore.clone(), &Config::new())
+            .await
+            .expect("persist baseline server config");
+        let context = Arc::new(AppContext::new(env.ecstore.clone(), Arc::new(TestIam), Arc::new(TestKms)));
+
+        let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let blocker = tokio::spawn(async move {
+            with_runtime_config_reload_lock(async move {
+                locked_tx.send(()).expect("signal runtime reload lock acquisition");
+                release_rx.await.expect("release runtime reload lock");
+                Ok(())
+            })
+            .await
+            .expect("runtime reload lock blocker");
+        });
+        locked_rx.await.expect("runtime reload lock should be held");
+
+        let older_context = context.clone();
+        let older = tokio::spawn(async move {
+            let specs = audit_specs();
+            update_audit_config_and_reload_for_context(Some(older_context.as_ref()), &specs, |config| {
+                config
+                    .0
+                    .entry(SCANNER_SUB_SYS.to_string())
+                    .or_default()
+                    .entry(DEFAULT_DELIMITER.to_string())
+                    .or_insert_with(KVS::new)
+                    .insert(SCANNER_CYCLE.to_string(), "15s".to_string());
+                true
+            })
+            .await
+        });
+        wait_for_persisted_target(env.ecstore.clone(), SCANNER_SUB_SYS, DEFAULT_DELIMITER).await;
+        assert!(!older.is_finished(), "audit runtime publication must wait for the shared reload lock");
+
+        let snapshot = read_admin_server_config_snapshot(env.ecstore.clone())
+            .await
+            .expect("read newer server config snapshot");
+        let mut latest = snapshot.config.clone();
+        let mut latest_target = KVS::new();
+        latest_target.insert(ENABLE_KEY.to_string(), EnableState::On.to_string());
+        latest_target.insert(WEBHOOK_ENDPOINT.to_string(), "https://audit.invalid/hook".to_string());
+        latest_target.insert(
+            WEBHOOK_QUEUE_DIR.to_string(),
+            temp_dir.path().join("audit-queue").to_string_lossy().into_owned(),
+        );
+        latest
+            .0
+            .entry(AUDIT_WEBHOOK_SUB_SYS.to_string())
+            .or_default()
+            .insert("latest".to_string(), latest_target);
+        save_admin_server_config_snapshot(env.ecstore.clone(), &latest, &snapshot)
+            .await
+            .expect("persist newer audit config");
+        drop(snapshot);
+
+        release_tx.send(()).expect("release runtime reload blocker");
+        blocker.await.expect("runtime reload blocker task");
+        tokio::time::timeout(Duration::from_secs(30), older)
+            .await
+            .expect("older audit update should complete")
+            .expect("older audit update task should not panic")
+            .expect("older audit update should converge from the latest durable config");
+
+        let system = audit_system().expect("latest durable audit target should start the audit system");
+        let targets = system.list_targets().await;
+        assert!(targets.iter().any(|target| target.contains("latest")), "active targets: {targets:?}");
+        system.close().await.expect("audit system should stop after the test");
+    }
+
+    #[test]
+    fn audit_convergence_error_reports_durable_write_state() {
+        for (persisted, expected) in [(true, "audit config persisted"), (false, "audit config unchanged")] {
+            let error = audit_config_convergence_error(persisted, "injected failure");
+            assert!(error.to_string().contains(expected), "unexpected convergence error: {error}");
+            assert!(error.to_string().contains("injected failure"));
+        }
+    }
+}

rustfs/src/admin/handlers/config_admin.rs

@@ -17,18 +17,17 @@ use crate::admin::handlers::supervise_admin_mutation;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{
     current_action_credentials, current_app_context, current_object_store_handle_for_context, current_server_config_for_context,
-    publish_server_config,
 };
 use crate::admin::service::config::{
-    CONFIG_WORKER_RELOAD_FAILURE_STATE, EVENT_CONFIG_WORKER_RELOAD_FAILED, FULL_CONFIG_WORKER_SUBSYSTEMS, LOG_COMPONENT_ADMIN,
-    LOG_SUBSYSTEM_CONFIG, PreparedRuntimeConfig, is_dynamic_config_subsystem, preflight_dynamic_config_reload,
-    prepare_server_config, reload_dynamic_config_runtime_state, reload_runtime_config_snapshot, signal_config_snapshot_reload,
-    signal_config_snapshot_reload_checked, signal_dynamic_config_reload_checked,
+    FULL_CONFIG_WORKER_SUBSYSTEMS, is_dynamic_config_subsystem, preflight_dynamic_config_reload, prepare_server_config,
+    publish_latest_runtime_config_snapshot, reload_dynamic_config_runtime_state, reload_runtime_config_snapshot,
+    signal_config_snapshot_reload, signal_config_snapshot_reload_checked, signal_dynamic_config_reload_checked,
 };
 use crate::admin::storage_api::config::storageclass::{INLINE_BLOCK_ENV, OPTIMIZE_ENV, RRS_ENV, STANDARD_ENV};
 use crate::admin::storage_api::config::{
-    AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, read_admin_config,
-    read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config, save_admin_server_config_snapshot,
+    AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config,
+    read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config,
+    save_admin_server_config_snapshot,
 };
 use crate::admin::storage_api::contract::list::ListOperations as _;
 use crate::admin::utils::{encode_compatible_admin_payload, is_compat_admin_request, read_compatible_admin_body};
@@ -766,7 +765,10 @@ async fn load_server_config_snapshot_from_store() -> S3Result<AdminServerConfigS
         .map_err(Into::into)
 }
 
-async fn save_server_config_to_store(config: &ServerConfig, snapshot: &AdminServerConfigSnapshot) -> S3Result<bool> {
+async fn save_server_config_to_store(
+    config: &ServerConfig,
+    snapshot: &AdminServerConfigSnapshot,
+) -> S3Result<AdminServerConfigSaveResult> {
     let store = object_store()?;
     save_admin_server_config_snapshot(store, config, snapshot)
         .await
@@ -987,8 +989,8 @@ fn decode_config_history_snapshot(data: &[u8]) -> S3Result<ServerConfig> {
     Ok(config)
 }
 
-fn validate_restore_rollback_generation(current: &ServerConfig, restored: &ServerConfig) -> S3Result<()> {
-    if current == restored {
+fn validate_restore_rollback_generation(current: Option<Uuid>, committed: Option<Uuid>) -> S3Result<()> {
+    if current.is_some() && current == committed {
         Ok(())
     } else {
         Err(s3_error!(
@@ -1004,12 +1006,12 @@ async fn save_server_config_history_snapshot(config: &ServerConfig) -> S3Result<
     save_server_config_history(&sealed).await
 }
 
-async fn cleanup_failed_config_history_snapshot(restore_id: &str) {
+async fn cleanup_config_history_snapshot(restore_id: &str) {
     if let Err(err) = delete_server_config_history(restore_id).await {
         warn!(
             restore_id,
             error = %err,
-            "Failed to remove config history snapshot for an uncommitted mutation"
+            "Failed to remove config history snapshot"
         );
     }
 }
@@ -1843,23 +1845,6 @@ async fn reject_lost_config_transaction<T>(snapshot: AdminServerConfigSnapshot,
     ))
 }
 
-fn publish_prepared_config_snapshots(config: ServerConfig, prepared: PreparedRuntimeConfig) -> S3Result<()> {
-    prepared.publish_storage_class()?;
-    publish_server_config(config);
-    Ok(())
-}
-
-/// Re-apply local mutable worker families after a full-config replacement.
-/// Peers receive one full-snapshot signal after this returns; signaling each
-/// family here as well would recreate audit/scanner targets twice per peer.
-fn publish_notify_config_intent(
-    config: &ServerConfig,
-    sub_system: Option<&str>,
-) -> Option<rustfs_notify::NotificationLifecycleTransition> {
-    (sub_system.is_none() || sub_system.is_some_and(|sub_system| NOTIFY_SUB_SYSTEMS.contains(&sub_system)))
-        .then(|| rustfs_notify::ensure_live_events().publish_config(config.clone()))
-}
-
 fn config_preflight_subsystems(sub_system: Option<&str>) -> Vec<&str> {
     if let Some(sub_system) = sub_system {
         return is_dynamic_config_subsystem(sub_system)
@@ -1884,78 +1869,26 @@ async fn preflight_config_intent(sub_system: Option<&str>) -> S3Result<()> {
     Ok(())
 }
 
-async fn wait_notify_config_intent(transition: Option<rustfs_notify::NotificationLifecycleTransition>) -> S3Result<bool> {
-    let Some(transition) = transition else {
-        return Ok(false);
-    };
-    transition.wait().await.map_err(|err| {
-        warn!(error = %err, "Failed to apply local notification config");
-        s3_error!(InternalError, "failed to apply notification config")
-    })?;
-    Ok(true)
-}
-
-async fn reload_non_notify_dynamic_subsystems() -> Vec<String> {
-    let mut failures = Vec::new();
-    for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS {
-        if NOTIFY_SUB_SYSTEMS.contains(&sub_system) {
-            continue;
-        }
-        if reload_dynamic_config_runtime_state(sub_system).await.is_err() {
-            failures.push(format!("local {sub_system}"));
-            warn!(
-                event = EVENT_CONFIG_WORKER_RELOAD_FAILED,
-                component = LOG_COMPONENT_ADMIN,
-                subsystem = LOG_SUBSYSTEM_CONFIG,
-                config_subsystem = sub_system,
-                state = CONFIG_WORKER_RELOAD_FAILURE_STATE,
-                reason = "apply_failed",
-                "Published server config but failed to reload a local worker subsystem"
-            );
-        }
-    }
-    failures
-}
-
-fn finish_config_reconciliation(errors: Vec<String>) -> S3Result<()> {
+fn finish_config_reconciliation(errors: Vec<String>, persisted: bool) -> S3Result<()> {
     if errors.is_empty() {
         Ok(())
-    } else {
+    } else if persisted {
         Err(s3_error!(
             InternalError,
             "server config persisted but runtime convergence failed: {}",
             errors.join("; ")
         ))
+    } else {
+        Err(s3_error!(
+            InternalError,
+            "server config was unchanged but runtime convergence failed: {}",
+            errors.join("; ")
+        ))
     }
 }
 
-async fn reconcile_targeted_config(
-    sub_system: Option<String>,
-    storage_class_applied: bool,
-    notify_transition: Option<rustfs_notify::NotificationLifecycleTransition>,
-) -> S3Result<bool> {
-    let mut errors = Vec::new();
-    let notify_applied = notify_transition.is_some();
-    if let Err(err) = wait_notify_config_intent(notify_transition).await {
-        warn!(error = %err, "Local notification config failed to converge");
-        errors.push("local notify".to_string());
-    }
-
-    let config_applied = if notify_applied {
-        if let Some(sub_system) = sub_system.as_deref()
-            && let Err(err) = signal_dynamic_config_reload_checked(sub_system).await
-        {
-            warn!(config_subsystem = sub_system, error = %err, "Peer config reload failed");
-            errors.push(format!("peer {sub_system}"));
-        }
-        true
-    } else if storage_class_applied {
-        if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await {
-            warn!(error = %err, "Peer storage-class reload failed");
-            errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}"));
-        }
-        true
-    } else if let Some(sub_system) = sub_system.as_deref()
+async fn reconcile_targeted_config(sub_system: Option<String>, persisted: bool, mut errors: Vec<String>) -> S3Result<bool> {
+    let config_applied = if let Some(sub_system) = sub_system.as_deref()
         && is_dynamic_config_subsystem(sub_system)
     {
         let config_applied = match reload_dynamic_config_runtime_state(sub_system).await {
@@ -1979,21 +1912,19 @@ async fn reconcile_targeted_config(
         false
     };
 
-    finish_config_reconciliation(errors)?;
+    finish_config_reconciliation(errors, persisted)?;
     Ok(config_applied)
 }
 
-async fn reconcile_full_config(notify_transition: Option<rustfs_notify::NotificationLifecycleTransition>) -> S3Result<()> {
-    let mut errors = Vec::new();
-    if let Err(err) = wait_notify_config_intent(notify_transition).await {
-        warn!(error = %err, "Local notification config failed to converge");
+async fn reconcile_full_config(persisted: bool, mut errors: Vec<String>) -> S3Result<()> {
+    if let Err(err) = reload_dynamic_config_runtime_state(NOTIFY_WEBHOOK_SUB_SYS).await {
+        warn!(error = %err, "Local notification config failed to converge from durable config");
         errors.push("local notify".to_string());
     }
     if let Err(err) = signal_dynamic_config_reload_checked(STORAGE_CLASS_SUB_SYS).await {
         warn!(error = %err, "Peer storage-class reload failed");
         errors.push(format!("peer {STORAGE_CLASS_SUB_SYS}"));
     }
-    errors.extend(reload_non_notify_dynamic_subsystems().await);
     if let Err(err) = signal_dynamic_config_reload_checked(NOTIFY_WEBHOOK_SUB_SYS).await {
         warn!(error = %err, "Peer notification config reload failed");
         errors.push("peer notify".to_string());
@@ -2002,91 +1933,82 @@ async fn reconcile_full_config(notify_transition: Option<rustfs_notify::Notifica
         warn!(error = %err, "Peer config snapshot reload failed");
         errors.push("peer config snapshot".to_string());
     }
-    finish_config_reconciliation(errors)
+    finish_config_reconciliation(errors, persisted)
 }
 
 struct PersistedConfigTransaction {
     previous_config: ServerConfig,
     history_restore_id: Option<String>,
-    storage_class_applied: bool,
-    notify_transition: Option<rustfs_notify::NotificationLifecycleTransition>,
+    persisted: bool,
+    committed_generation: Option<Uuid>,
 }
 
 async fn persist_server_config_transaction(
     config: ServerConfig,
-    prepared: PreparedRuntimeConfig,
     snapshot: AdminServerConfigSnapshot,
-    sub_system: Option<&str>,
 ) -> S3Result<PersistedConfigTransaction> {
     snapshot.ensure_lock_held().map_err(ApiError::from).map_err(S3Error::from)?;
     let previous_config = snapshot.config.clone();
     let history_restore_id = save_server_config_history_snapshot(&previous_config).await?;
     if snapshot.is_lock_lost() {
-        cleanup_failed_config_history_snapshot(&history_restore_id).await;
+        cleanup_config_history_snapshot(&history_restore_id).await;
         return reject_lost_config_transaction(snapshot, "before persistence").await;
     }
 
-    let persisted = match save_server_config_to_store(&config, &snapshot).await {
-        Ok(persisted) => persisted,
+    let save_result = match save_server_config_to_store(&config, &snapshot).await {
+        Ok(result) => result,
         Err(err) => {
-            cleanup_failed_config_history_snapshot(&history_restore_id).await;
+            cleanup_config_history_snapshot(&history_restore_id).await;
             return Err(err);
         }
     };
+    let persisted = save_result.persisted();
+    let committed_generation = save_result.generation();
     let history_restore_id = if persisted {
         Some(history_restore_id)
     } else {
-        cleanup_failed_config_history_snapshot(&history_restore_id).await;
+        cleanup_config_history_snapshot(&history_restore_id).await;
         None
     };
-    if snapshot.is_lock_lost() {
-        return reject_lost_config_transaction(snapshot, "after persistence").await;
-    }
-
-    let storage_class_applied = sub_system.is_none_or(|value| value == STORAGE_CLASS_SUB_SYS);
-    if storage_class_applied {
-        if let Err(err) = publish_prepared_config_snapshots(config.clone(), prepared) {
-            return Err(match history_restore_id.as_deref() {
-                Some(restore_id) => s3_error!(
-                    InternalError,
-                    "config persisted but runtime publish failed; recovery snapshot restoreId={}: {}",
-                    restore_id,
-                    err
-                ),
-                None => err,
-            });
-        }
-    } else {
-        publish_server_config(config.clone());
-    }
-    let notify_transition = publish_notify_config_intent(&config, sub_system);
-    if snapshot.is_lock_lost() {
-        return reject_lost_config_transaction(snapshot, "while publishing runtime snapshots").await;
-    }
-
     drop(snapshot);
     Ok(PersistedConfigTransaction {
         previous_config,
         history_restore_id,
-        storage_class_applied,
-        notify_transition,
+        persisted,
+        committed_generation,
     })
 }
 
+async fn reconcile_committed_config(sub_system: Option<String>, persisted: bool) -> S3Result<bool> {
+    let mut errors = Vec::new();
+    let publication_result = match sub_system.as_deref() {
+        Some(sub_system) => publish_latest_runtime_config_snapshot(sub_system).await,
+        None => reload_runtime_config_snapshot().await,
+    };
+    if let Err(err) = publication_result {
+        warn!(error = %err, "Failed to publish the latest durable server config");
+        errors.push("local config snapshot".to_string());
+    }
+
+    if sub_system.is_none() {
+        reconcile_full_config(persisted, errors).await?;
+        return Ok(false);
+    }
+
+    reconcile_targeted_config(sub_system, persisted, errors).await
+}
+
 async fn commit_server_config_transaction(
     config: ServerConfig,
-    prepared: PreparedRuntimeConfig,
     snapshot: AdminServerConfigSnapshot,
     sub_system: Option<String>,
 ) -> S3Result<bool> {
-    let transaction = persist_server_config_transaction(config, prepared, snapshot, sub_system.as_deref()).await?;
-
-    if sub_system.is_none() {
-        reconcile_full_config(transaction.notify_transition).await?;
-        return Ok(false);
+    let transaction = persist_server_config_transaction(config, snapshot).await?;
+    let result = reconcile_committed_config(sub_system, transaction.persisted).await;
+    match (result, transaction.history_restore_id.as_deref()) {
+        (Err(err), Some(restore_id)) => Err(s3_error!(InternalError, "{}; recovery snapshot restoreId={}", err, restore_id)),
+        (result, _) => result,
     }
-
-    reconcile_targeted_config(sub_system, transaction.storage_class_applied, transaction.notify_transition).await
 }
 
 pub struct GetConfigKVHandler {}
@@ -2127,8 +2049,8 @@ impl Operation for SetConfigKVHandler {
             let snapshot = load_server_config_snapshot_from_store().await?;
             let mut config = snapshot.config.clone();
             apply_set_directives(&mut config, &directives)?;
-            let prepared = prepare_server_config(&config, sub_system.as_deref()).await?;
-            commit_server_config_transaction(config, prepared, snapshot, sub_system).await
+            prepare_server_config(&config, sub_system.as_deref()).await?;
+            commit_server_config_transaction(config, snapshot, sub_system).await
         })
         .await?;
 
@@ -2155,8 +2077,8 @@ impl Operation for DelConfigKVHandler {
             let snapshot = load_server_config_snapshot_from_store().await?;
             let mut config = snapshot.config.clone();
             apply_delete_directives(&mut config, &directives);
-            let prepared = prepare_server_config(&config, sub_system.as_deref()).await?;
-            commit_server_config_transaction(config, prepared, snapshot, sub_system).await
+            prepare_server_config(&config, sub_system.as_deref()).await?;
+            commit_server_config_transaction(config, snapshot, sub_system).await
         })
         .await?;
 
@@ -2239,15 +2161,19 @@ impl Operation for RestoreConfigHistoryKVHandler {
         supervise_admin_mutation("config mutation", async move {
             preflight_config_intent(None).await?;
             let snapshot = load_server_config_snapshot_from_store().await?;
-            let prepared = prepare_server_config(&config, None).await?;
-            let restored_config = config.clone();
-            let transaction = persist_server_config_transaction(config, prepared, snapshot, None).await?;
-            let Err(restore_error) = reconcile_full_config(transaction.notify_transition).await else {
+            prepare_server_config(&config, None).await?;
+            let transaction = persist_server_config_transaction(config, snapshot).await?;
+            let Err(restore_error) = reconcile_committed_config(None, transaction.persisted).await else {
                 return Ok(());
             };
 
+            if !transaction.persisted {
+                return Err(restore_error);
+            }
+
             let previous_config = transaction.previous_config;
             let recovery_restore_id = transaction.history_restore_id;
+            let committed_generation = transaction.committed_generation;
             let recovery_reference = recovery_restore_id.as_deref().unwrap_or("not-created");
             let rollback_snapshot = match load_server_config_snapshot_from_store().await {
                 Ok(snapshot) => snapshot,
@@ -2261,7 +2187,9 @@ impl Operation for RestoreConfigHistoryKVHandler {
                     ));
                 }
             };
-            if let Err(rollback_error) = validate_restore_rollback_generation(&rollback_snapshot.config, &restored_config) {
+            if let Err(rollback_error) =
+                validate_restore_rollback_generation(rollback_snapshot.generation(), committed_generation)
+            {
                 return Err(s3_error!(
                     InternalError,
                     "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}",
@@ -2271,12 +2199,20 @@ impl Operation for RestoreConfigHistoryKVHandler {
                 ));
             }
 
-            let rollback_prepared = prepare_server_config(&previous_config, None).await?;
-            rollback_snapshot
+            if let Err(rollback_error) = prepare_server_config(&previous_config, None).await {
+                return Err(s3_error!(
+                    InternalError,
+                    "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}",
+                    restore_error,
+                    rollback_error,
+                    recovery_reference
+                ));
+            }
+            if let Err(rollback_error) = rollback_snapshot
                 .ensure_lock_held()
                 .map_err(ApiError::from)
-                .map_err(S3Error::from)?;
-            if let Err(rollback_error) = save_server_config_to_store(&previous_config, &rollback_snapshot).await {
+                .map_err(S3Error::from)
+            {
                 return Err(s3_error!(
                     InternalError,
                     "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}",
@@ -2285,10 +2221,9 @@ impl Operation for RestoreConfigHistoryKVHandler {
                     recovery_reference
                 ));
             }
-            if rollback_snapshot.is_lock_lost() {
-                if let Err(rollback_error) =
-                    reject_lost_config_transaction::<()>(rollback_snapshot, "after restore rollback persistence").await
-                {
+            let rollback_persisted = match save_server_config_to_store(&previous_config, &rollback_snapshot).await {
+                Ok(result) => result.persisted(),
+                Err(rollback_error) => {
                     return Err(s3_error!(
                         InternalError,
                         "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}",
@@ -2297,46 +2232,20 @@ impl Operation for RestoreConfigHistoryKVHandler {
                         recovery_reference
                     ));
                 }
-                return Err(s3_error!(InternalError, "restore rollback lock-loss handling returned unexpectedly"));
-            }
-
-            if let Err(rollback_error) = publish_prepared_config_snapshots(previous_config.clone(), rollback_prepared) {
-                return Err(s3_error!(
-                    InternalError,
-                    "config restore failed: {}; automatic rollback publish failed: {}; recovery snapshot restoreId={}",
-                    restore_error,
-                    rollback_error,
-                    recovery_reference
-                ));
-            }
-            let rollback_transition = publish_notify_config_intent(&previous_config, None);
-            if rollback_snapshot.is_lock_lost() {
-                if let Err(rollback_error) =
-                    reject_lost_config_transaction::<()>(rollback_snapshot, "while publishing restore rollback").await
-                {
-                    return Err(s3_error!(
-                        InternalError,
-                        "config restore failed: {}; automatic rollback failed: {}; recovery snapshot restoreId={}",
-                        restore_error,
-                        rollback_error,
-                        recovery_reference
-                    ));
-                }
-                return Err(s3_error!(InternalError, "restore rollback lock-loss handling returned unexpectedly"));
-            }
+            };
             drop(rollback_snapshot);
 
-            if let Err(rollback_error) = reconcile_full_config(rollback_transition).await {
+            if let Err(rollback_error) = reconcile_committed_config(None, rollback_persisted).await {
                 return Err(s3_error!(
                     InternalError,
-                    "config restore failed: {}; persisted rollback did not converge: {}; recovery snapshot restoreId={}",
+                    "config restore failed: {}; automatic rollback convergence failed: {}; recovery snapshot restoreId={}",
                     restore_error,
                     rollback_error,
                     recovery_reference
                 ));
             }
-            if let Some(recovery_restore_id) = recovery_restore_id.as_deref() {
-                cleanup_failed_config_history_snapshot(recovery_restore_id).await;
+            if rollback_persisted && let Some(recovery_restore_id) = recovery_restore_id.as_deref() {
+                cleanup_config_history_snapshot(recovery_restore_id).await;
             }
             Err(s3_error!(InternalError, "config restore failed and was rolled back: {}", restore_error))
         })
@@ -2377,8 +2286,8 @@ impl Operation for SetConfigHandler {
             let snapshot = load_server_config_snapshot_from_store().await?;
             let mut config = ServerConfig::new();
             apply_set_directives(&mut config, &directives)?;
-            let prepared = prepare_server_config(&config, None).await?;
-            commit_server_config_transaction(config, prepared, snapshot, None).await?;
+            prepare_server_config(&config, None).await?;
+            commit_server_config_transaction(config, snapshot, None).await?;
             Ok(())
         })
         .await?;
@@ -2408,6 +2317,23 @@ mod tests {
         );
     }
 
+    #[test]
+    fn reconciliation_error_reports_whether_config_was_persisted() {
+        let persisted = finish_config_reconciliation(vec!["local config snapshot".to_string()], true)
+            .expect_err("committed config convergence failure must be reported");
+        let unchanged = finish_config_reconciliation(vec!["local config snapshot".to_string()], false)
+            .expect_err("unchanged config convergence failure must be reported");
+
+        assert_eq!(
+            persisted.message(),
+            Some("server config persisted but runtime convergence failed: local config snapshot")
+        );
+        assert_eq!(
+            unchanged.message(),
+            Some("server config was unchanged but runtime convergence failed: local config snapshot")
+        );
+    }
+
     #[test]
     fn tokenize_config_line_handles_quotes_and_escapes() {
         let tokens = tokenize_config_line(r#"identity_openid client_id="console app" client_secret="s3cr\"et" enable=on"#)
@@ -3191,23 +3117,32 @@ notify_webhook:secondary endpoint="https://secondary.example" auth_token="second
     }
 
     #[test]
-    fn restore_rollback_generation_rejects_concurrent_config_change() {
-        crate::admin::storage_api::config::init_admin_config_defaults();
-        let restored = ServerConfig::new();
-        let mut concurrent = restored.clone();
-        apply_set_directives(
-            &mut concurrent,
-            &parse_config_directives(r#"identity_openid client_id="concurrent-client""#, false).expect("parse concurrent"),
-        )
-        .expect("apply concurrent");
+    fn restore_rollback_generation_accepts_committed_write_identity() {
+        let committed = Uuid::from_u128(1);
+        validate_restore_rollback_generation(Some(committed), Some(committed)).expect("matching committed generation");
+    }
 
-        validate_restore_rollback_generation(&restored, &restored).expect("unchanged restore generation");
-        let error = validate_restore_rollback_generation(&concurrent, &restored)
-            .expect_err("concurrent change must fence automatic rollback");
+    #[test]
+    fn restore_rollback_generation_rejects_aba_with_matching_config_content() {
+        let error = validate_restore_rollback_generation(Some(Uuid::from_u128(2)), Some(Uuid::from_u128(1)))
+            .expect_err("a later generation must fence automatic rollback even when config content matches");
         assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
         assert!(error.to_string().contains("concurrent configuration change"));
     }
 
+    #[test]
+    fn restore_rollback_generation_rejects_missing_write_identity() {
+        for (current, committed) in [
+            (None, Some(Uuid::from_u128(1))),
+            (Some(Uuid::from_u128(1)), None),
+            (None, None),
+        ] {
+            let error = validate_restore_rollback_generation(current, committed)
+                .expect_err("missing generation metadata must fence automatic rollback");
+            assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+        }
+    }
+
     #[test]
     fn legacy_directive_history_is_rejected_without_being_applied_as_a_snapshot() {
         let error = decode_config_history_snapshot(b"identity_openid client_id=\"legacy\"")

rustfs/src/admin/service/config.rs

@@ -16,7 +16,9 @@ use crate::admin::runtime_sources::{
     AppContext, current_app_context, current_notification_system_for_context, current_object_store_handle_for_context,
     publish_server_config, publish_storage_class_config,
 };
-use crate::admin::storage_api::config::{STORAGE_CLASS_SUB_SYS, read_admin_config_without_migrate, storageclass};
+use crate::admin::storage_api::config::{
+    STORAGE_CLASS_SUB_SYS, read_existing_admin_server_config_no_lock, storageclass, with_admin_server_config_read_lock,
+};
 use crate::admin::storage_api::contract::admin::StorageAdminApi;
 use crate::admin::storage_api::runtime::ECStore;
 use crate::server::{
@@ -43,6 +45,25 @@ use url::Url;
 
 static RUNTIME_CONFIG_RELOAD_MUTEX: AsyncMutex<()> = AsyncMutex::const_new(());
 
+// Runtime publication lock order: reload mutex -> server-config local lock ->
+// transaction lock -> config object lock.
+pub(crate) async fn with_runtime_config_reload_lock<Fut, T>(operation: Fut) -> S3Result<T>
+where
+    Fut: Future<Output = S3Result<T>> + Send + 'static,
+    T: Send + 'static,
+{
+    let reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await;
+    tokio::spawn(async move {
+        let _reload_guard = reload_guard;
+        operation.await
+    })
+    .await
+    .map_err(|err| {
+        let outcome = if err.is_cancelled() { "cancelled" } else { "panicked" };
+        internal_error(format!("runtime config reload task {outcome}"))
+    })?
+}
+
 pub fn is_dynamic_config_subsystem(sub_system: &str) -> bool {
     NOTIFY_SUB_SYSTEMS.contains(&sub_system)
         || matches!(
@@ -121,11 +142,6 @@ impl PreparedRuntimeConfig {
     fn publish_storage_class_for_context(self, context: Option<&AppContext>) -> S3Result<()> {
         self.publish_storage_class_for_context_with(context, publish_storage_class_config)
     }
-
-    pub(crate) fn publish_storage_class(self) -> S3Result<()> {
-        let context = current_app_context();
-        self.publish_storage_class_for_context(context.as_deref())
-    }
 }
 
 fn publish_server_config_for_context(context: Option<&AppContext>, config: ServerConfig) {
@@ -396,7 +412,18 @@ pub async fn apply_dynamic_config_for_subsystem(config: &ServerConfig, sub_syste
 }
 
 pub async fn reload_dynamic_config_runtime_state_for_context(context: Option<&AppContext>, sub_system: &str) -> S3Result<()> {
-    let _reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await;
+    let context = context.cloned();
+    let sub_system = sub_system.to_owned();
+    with_runtime_config_reload_lock(async move {
+        reload_dynamic_config_runtime_state_under_reload_lock_for_context(context.as_ref(), &sub_system).await
+    })
+    .await
+}
+
+async fn reload_dynamic_config_runtime_state_under_reload_lock_for_context(
+    context: Option<&AppContext>,
+    sub_system: &str,
+) -> S3Result<()> {
     if sub_system == MODULE_SWITCHES_SIGNAL_SUBSYSTEM {
         let store = resolve_runtime_config_store_for_context(context)?;
         let notify_result = reconcile_event_notifier_from_store(store).await;
@@ -414,18 +441,54 @@ pub async fn reload_dynamic_config_runtime_state_for_context(context: Option<&Ap
     }
 
     let store = resolve_runtime_config_store_for_context(context)?;
-    let config = read_admin_config_without_migrate(store).await.map_err(|err| {
-        warn!("peer reload_dynamic_config: failed to load server config for {sub_system}: {err}");
-        internal_error(format!("failed to load server config: {err}"))
-    })?;
+    let read_store = store.clone();
+    let publication_context = context.cloned();
+    let publication_sub_system = sub_system.to_owned();
+    let notify_config = with_admin_server_config_read_lock(store, move || async move {
+        let config = read_existing_admin_server_config_no_lock(read_store).await.map_err(|err| {
+            warn!("peer reload_dynamic_config: failed to load server config for {publication_sub_system}: {err}");
+            internal_error(format!("failed to load server config: {err}"))
+        })?;
+        let prepared = prepare_server_config_for_context(publication_context.as_ref(), &config, Some(&publication_sub_system))
+            .await
+            .map_err(|err| {
+                if publication_sub_system == STORAGE_CLASS_SUB_SYS {
+                    internal_error(format!("failed to apply storage class config: {err}"))
+                } else {
+                    err
+                }
+            })?;
 
-    if matches!(sub_system, SCANNER_SUB_SYS | HEAL_SUB_SYS) {
-        validate_server_config_for_context(context, &config, Some(sub_system)).await?;
-        // Scanner cycles refresh from the process-wide server config before
-        // each pass. Publish the same validated snapshot first so that refresh
-        // cannot overwrite this peer's dynamic scanner update with stale data.
-        publish_server_config_for_context(context, config.clone());
-    }
+        if publication_sub_system == STORAGE_CLASS_SUB_SYS {
+            prepared.publish_storage_class_for_context(publication_context.as_ref())?;
+            return Ok::<Option<ServerConfig>, S3Error>(None);
+        }
+        if NOTIFY_SUB_SYSTEMS.contains(&publication_sub_system.as_str()) {
+            return Ok(Some(config));
+        }
+        if matches!(publication_sub_system.as_str(), SCANNER_SUB_SYS | HEAL_SUB_SYS) {
+            publish_server_config_for_context(publication_context.as_ref(), config.clone());
+        }
+        apply_dynamic_config_for_subsystem_for_context(publication_context.as_ref(), &config, &publication_sub_system)
+            .await
+            .inspect_err(|_| {
+                warn!(
+                    config_subsystem = publication_sub_system,
+                    reason = "apply_failed",
+                    "Peer dynamic config apply failed"
+                );
+            })?;
+        Ok(None)
+    })
+    .await
+    .map_err(|err| {
+        warn!("peer reload_dynamic_config: failed to acquire server config publication fence for {sub_system}: {err}");
+        internal_error(format!("failed to lock server config: {err}"))
+    })??;
+
+    let Some(config) = notify_config else {
+        return Ok(());
+    };
     apply_dynamic_config_for_subsystem_for_context(context, &config, sub_system)
         .await
         .inspect_err(|_| {
@@ -439,23 +502,16 @@ pub async fn reload_dynamic_config_runtime_state(sub_system: &str) -> S3Result<(
     reload_dynamic_config_runtime_state_for_context(context.as_deref(), sub_system).await
 }
 
-async fn reload_runtime_config_snapshot_with<ReadFuture, Prepare, PrepareFuture, Publish, ApplyWorkers, ApplyWorkersFuture>(
-    read: ReadFuture,
-    prepare: Prepare,
-    publish: Publish,
+async fn reload_runtime_config_snapshot_with<PublishFuture, ApplyWorkers, ApplyWorkersFuture>(
+    publish_snapshot: PublishFuture,
     apply_workers: ApplyWorkers,
 ) -> S3Result<()>
 where
-    ReadFuture: Future<Output = S3Result<ServerConfig>>,
-    Prepare: FnOnce(ServerConfig) -> PrepareFuture,
-    PrepareFuture: Future<Output = S3Result<(ServerConfig, PreparedRuntimeConfig)>>,
-    Publish: FnOnce(&ServerConfig, PreparedRuntimeConfig) -> S3Result<()>,
+    PublishFuture: Future<Output = S3Result<ServerConfig>>,
     ApplyWorkers: FnOnce(ServerConfig) -> ApplyWorkersFuture,
     ApplyWorkersFuture: Future<Output = S3Result<()>>,
 {
-    let config = read.await?;
-    let (config, prepared) = prepare(config).await?;
-    publish(&config, prepared)?;
+    let config = publish_snapshot.await?;
 
     // Worker reloads mutate live state and have no rollback contract, so the
     // validated snapshots stay published. The RPC still reports convergence
@@ -474,55 +530,105 @@ where
     Ok(())
 }
 
-pub async fn reload_runtime_config_snapshot_for_context(context: Option<&AppContext>) -> S3Result<()> {
-    let _reload_guard = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await;
+async fn publish_latest_runtime_config_snapshot_under_reload_lock_for_context<ApplyWorkers, ApplyWorkersFuture>(
+    context: Option<&AppContext>,
+    sub_system: Option<&str>,
+    apply_workers: ApplyWorkers,
+) -> S3Result<()>
+where
+    ApplyWorkers: FnOnce(ServerConfig) -> ApplyWorkersFuture + Send + 'static,
+    ApplyWorkersFuture: Future<Output = S3Result<()>> + Send + 'static,
+{
     let store = resolve_runtime_config_store_for_context(context)?;
+    let read_store = store.clone();
+    let publication_context = context.cloned();
+    let publication_sub_system = sub_system.map(str::to_owned);
 
-    reload_runtime_config_snapshot_with(
-        async move {
-            read_admin_config_without_migrate(store).await.map_err(|err| {
-                warn!("peer reload_runtime_config_snapshot: failed to load server config: {err}");
-                internal_error(format!("failed to load server config: {err}"))
-            })
-        },
-        |config| async move {
-            let prepared = prepare_server_config_for_context(context, &config, None).await.map_err(|_| {
-                warn!("peer reload_runtime_config_snapshot: failed to prepare server config");
-                internal_error("failed to prepare server config")
-            })?;
-            Ok((config, prepared))
-        },
-        |config, prepared| {
-            prepared.publish_storage_class_for_context(context)?;
-            publish_server_config_for_context(context, config.clone());
-            Ok(())
-        },
-        |config| async move {
-            let mut failures = Vec::new();
-            for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS {
-                if apply_dynamic_config_for_subsystem_for_context(context, &config, sub_system)
-                    .await
-                    .is_err()
+    with_admin_server_config_read_lock(store, move || async move {
+        let config = read_existing_admin_server_config_no_lock(read_store).await.map_err(|err| {
+            warn!("runtime config publication failed to load server config: {err}");
+            internal_error(format!("failed to load server config: {err}"))
+        })?;
+        let prepared =
+            prepare_server_config_for_context(publication_context.as_ref(), &config, publication_sub_system.as_deref())
+                .await
+                .map_err(|err| match publication_sub_system.as_deref() {
+                    None => {
+                        warn!("peer reload_runtime_config_snapshot: failed to prepare server config");
+                        internal_error("failed to prepare server config")
+                    }
+                    Some(STORAGE_CLASS_SUB_SYS) => internal_error(format!("failed to apply storage class config: {err}")),
+                    Some(_) => err,
+                })?;
+        reload_runtime_config_snapshot_with(
+            async move {
+                if publication_sub_system
+                    .as_deref()
+                    .is_none_or(|sub_system| sub_system == STORAGE_CLASS_SUB_SYS)
                 {
-                    failures.push(sub_system);
-                    warn!(
-                        event = EVENT_CONFIG_WORKER_RELOAD_FAILED,
-                        component = LOG_COMPONENT_ADMIN,
-                        subsystem = LOG_SUBSYSTEM_CONFIG,
-                        config_subsystem = sub_system,
-                        state = CONFIG_WORKER_RELOAD_FAILURE_STATE,
-                        reason = "apply_failed",
-                        "Peer runtime config snapshot was published but a subsystem worker reload failed"
-                    );
+                    prepared.publish_storage_class_for_context(publication_context.as_ref())?;
                 }
+                publish_server_config_for_context(publication_context.as_ref(), config.clone());
+                Ok(config)
+            },
+            apply_workers,
+        )
+        .await
+    })
+    .await
+    .map_err(|err| {
+        warn!("runtime config publication failed to acquire server config publication fence: {err}");
+        internal_error(format!("failed to lock server config: {err}"))
+    })?
+}
+
+pub(crate) async fn publish_latest_runtime_config_snapshot(sub_system: &str) -> S3Result<()> {
+    let context = current_app_context();
+    let sub_system = sub_system.to_owned();
+    with_runtime_config_reload_lock(async move {
+        publish_latest_runtime_config_snapshot_under_reload_lock_for_context(context.as_deref(), Some(&sub_system), |_| async {
+            Ok(())
+        })
+        .await
+    })
+    .await
+}
+
+pub async fn reload_runtime_config_snapshot_for_context(context: Option<&AppContext>) -> S3Result<()> {
+    let context = context.cloned();
+    with_runtime_config_reload_lock(async move {
+        reload_runtime_config_snapshot_under_reload_lock_for_context(context.as_ref()).await
+    })
+    .await
+}
+
+async fn reload_runtime_config_snapshot_under_reload_lock_for_context(context: Option<&AppContext>) -> S3Result<()> {
+    let worker_context = context.cloned();
+    publish_latest_runtime_config_snapshot_under_reload_lock_for_context(context, None, move |config| async move {
+        let mut failures = Vec::new();
+        for sub_system in FULL_CONFIG_WORKER_SUBSYSTEMS {
+            if apply_dynamic_config_for_subsystem_for_context(worker_context.as_ref(), &config, sub_system)
+                .await
+                .is_err()
+            {
+                failures.push(sub_system);
+                warn!(
+                    event = EVENT_CONFIG_WORKER_RELOAD_FAILED,
+                    component = LOG_COMPONENT_ADMIN,
+                    subsystem = LOG_SUBSYSTEM_CONFIG,
+                    config_subsystem = sub_system,
+                    state = CONFIG_WORKER_RELOAD_FAILURE_STATE,
+                    reason = "apply_failed",
+                    "Peer runtime config snapshot was published but a subsystem worker reload failed"
+                );
             }
-            if failures.is_empty() {
-                Ok(())
-            } else {
-                Err(internal_error(format!("runtime worker reload failed: {}", failures.join("; "))))
-            }
-        },
-    )
+        }
+        if failures.is_empty() {
+            Ok(())
+        } else {
+            Err(internal_error(format!("runtime worker reload failed: {}", failures.join("; "))))
+        }
+    })
     .await
 }
 
@@ -699,6 +805,77 @@ mod tests {
         }
     }
 
+    #[tokio::test]
+    async fn runtime_reload_waiter_cancellation_does_not_leave_queued_work() {
+        let _blocker = RUNTIME_CONFIG_RELOAD_MUTEX.lock().await;
+        let (operation_started_tx, mut operation_started_rx) = tokio::sync::oneshot::channel();
+        let mut reload = Box::pin(with_runtime_config_reload_lock(async move {
+            let _ = operation_started_tx.send(());
+            Ok(())
+        }));
+
+        poll_fn(|cx| match reload.as_mut().poll(cx) {
+            Poll::Pending => Poll::Ready(()),
+            Poll::Ready(_) => panic!("reload must wait while the mutex is held"),
+        })
+        .await;
+        drop(reload);
+
+        assert!(
+            matches!(operation_started_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Closed)),
+            "cancelling a queued reload must drop its work instead of detaching it"
+        );
+    }
+
+    #[tokio::test]
+    async fn runtime_reload_lock_survives_waiter_cancellation() {
+        let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+        let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+        let (completed_tx, completed_rx) = tokio::sync::oneshot::channel();
+
+        let waiter = tokio::spawn(async move {
+            with_runtime_config_reload_lock(async move {
+                started_tx.send(()).expect("signal runtime reload start");
+                release_rx.await.expect("release supervised runtime reload");
+                completed_tx.send(()).expect("signal runtime reload completion");
+                Ok(())
+            })
+            .await
+        });
+
+        started_rx.await.expect("supervised runtime reload started");
+        waiter.abort();
+        assert!(waiter.await.expect_err("reload waiter should be cancelled").is_cancelled());
+
+        let (contender_polled_tx, contender_polled_rx) = tokio::sync::oneshot::channel();
+        let (contender_entered_tx, mut contender_entered_rx) = tokio::sync::oneshot::channel();
+        let contender = tokio::spawn(async move {
+            let mut lock = Box::pin(RUNTIME_CONFIG_RELOAD_MUTEX.lock());
+            let mut contender_polled_tx = Some(contender_polled_tx);
+            let _guard = poll_fn(|cx| {
+                if let Some(tx) = contender_polled_tx.take() {
+                    let _ = tx.send(());
+                }
+                lock.as_mut().poll(cx)
+            })
+            .await;
+            let _ = contender_entered_tx.send(());
+        });
+
+        contender_polled_rx.await.expect("contending reload should be polled");
+        assert!(
+            matches!(contender_entered_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Empty)),
+            "a cancelled waiter must not release the reload mutex while its detached reload is still running"
+        );
+
+        release_tx.send(()).expect("release detached runtime reload");
+        completed_rx.await.expect("detached runtime reload should complete");
+        contender_entered_rx
+            .await
+            .expect("contending reload should enter after completion");
+        contender.await.expect("contending reload task should not panic");
+    }
+
     #[tokio::test]
     async fn checked_scanner_reload_reports_unreachable_peer() {
         let temp_dir = TempDir::new().expect("scanner reload temp dir");
@@ -1221,6 +1398,123 @@ mod tests {
         .await;
     }
 
+    #[tokio::test(flavor = "multi_thread", worker_threads = 4)]
+    #[serial_test::serial(storage_class_env)]
+    async fn full_reload_keeps_worker_apply_inside_durable_read_fence() {
+        temp_env::async_with_vars(
+            [
+                (storageclass::STANDARD_ENV, None::<&str>),
+                (storageclass::RRS_ENV, None::<&str>),
+                (storageclass::OPTIMIZE_ENV, None::<&str>),
+                (storageclass::INLINE_BLOCK_ENV, None::<&str>),
+            ],
+            async {
+                let fixture = runtime_config_reload_fixture().await;
+                let older = scanner_server_config("41");
+                save_admin_server_config(fixture.context.object_store(), &older)
+                    .await
+                    .expect("persist older scanner config");
+
+                let (worker_entered_tx, worker_entered_rx) = tokio::sync::oneshot::channel();
+                let (release_worker_tx, release_worker_rx) = tokio::sync::oneshot::channel();
+                let reload_context = fixture.context.clone();
+                let reload = tokio::spawn(async move {
+                    publish_latest_runtime_config_snapshot_under_reload_lock_for_context(
+                        Some(&reload_context),
+                        None,
+                        move |config| async move {
+                            worker_entered_tx.send(config).expect("signal runtime config worker entry");
+                            release_worker_rx.await.expect("release runtime config worker");
+                            Ok(())
+                        },
+                    )
+                    .await
+                });
+
+                let worker_config = tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, worker_entered_rx)
+                    .await
+                    .expect("worker apply should enter")
+                    .expect("worker apply entry signal should be delivered");
+                assert_eq!(
+                    worker_config
+                        .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER)
+                        .expect("older scanner config should be loaded")
+                        .get(SCANNER_CYCLE),
+                    "41"
+                );
+
+                let latest = scanner_server_config("71");
+                let writer_store = fixture.context.object_store();
+                let (writer_polled_tx, writer_polled_rx) = tokio::sync::oneshot::channel();
+                let (writer_entered_tx, mut writer_entered_rx) = tokio::sync::oneshot::channel();
+                let writer = tokio::spawn(async move {
+                    let transaction_store = writer_store.clone();
+                    let transaction = with_admin_server_config_write_lock(writer_store, move || async move {
+                        writer_entered_tx.send(()).expect("signal writer entry");
+                        save_admin_server_config_no_lock(transaction_store, &latest).await
+                    });
+                    tokio::pin!(transaction);
+                    let mut writer_polled_tx = Some(writer_polled_tx);
+                    poll_fn(|cx| match transaction.as_mut().poll(cx) {
+                        Poll::Pending => {
+                            if let Some(tx) = writer_polled_tx.take() {
+                                let _ = tx.send(());
+                            }
+                            Poll::Ready(())
+                        }
+                        Poll::Ready(_) => panic!("writer entered before the worker apply released its read fence"),
+                    })
+                    .await;
+                    transaction
+                        .await
+                        .expect("writer should acquire the server-config locks")
+                        .expect("writer should persist the latest config");
+                });
+
+                tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, writer_polled_rx)
+                    .await
+                    .expect("writer should be polled")
+                    .expect("writer poll signal should be delivered");
+                assert!(
+                    matches!(writer_entered_rx.try_recv(), Err(tokio::sync::oneshot::error::TryRecvError::Empty)),
+                    "a server-config writer must wait until the old worker apply completes"
+                );
+
+                release_worker_tx.send(()).expect("release worker apply");
+                tokio::time::timeout(REAL_STORE_TEST_TIMEOUT, async {
+                    reload
+                        .await
+                        .expect("full reload task should not panic")
+                        .expect("full reload should finish");
+                    writer.await.expect("writer task should not panic");
+                })
+                .await
+                .expect("reload and writer should finish");
+
+                reload_runtime_config_snapshot_for_context(Some(&fixture.context))
+                    .await
+                    .expect("a later full reload should publish the latest durable config");
+                let snapshot = fixture
+                    .server_snapshot
+                    .lock()
+                    .expect("server config result lock")
+                    .clone()
+                    .expect("latest server config should be published");
+                assert_eq!(
+                    snapshot
+                        .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER)
+                        .expect("latest scanner config should be present")
+                        .get(SCANNER_CYCLE),
+                    "71"
+                );
+                assert_eq!(rustfs_scanner::scanner_runtime_config_status().cycle_interval_seconds.value, 71);
+
+                rustfs_scanner::apply_scanner_runtime_config(&ServerConfig::new()).expect("restore scanner runtime defaults");
+            },
+        )
+        .await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
     async fn peer_full_reload_rejects_later_pool_without_publishing() {
@@ -1251,11 +1545,9 @@ mod tests {
         let worker_events = events.clone();
 
         let err = reload_runtime_config_snapshot_with(
-            async { Ok(ServerConfig::new()) },
-            |config| async { Ok((config, PreparedRuntimeConfig::default())) },
-            move |_config, _prepared| {
+            async move {
                 publish_events.lock().expect("reload event lock").push("publish");
-                Ok(())
+                Ok(ServerConfig::new())
             },
             move |_config| async move {
                 let mut events = worker_events.lock().expect("reload event lock");

rustfs/src/admin/storage_api.rs

@@ -644,12 +644,17 @@ pub(crate) async fn read_admin_config_without_migrate(api: Arc<ECStore>) -> Resu
     ecstore_config::com::read_config_without_migrate(api).await
 }
 
+pub(crate) async fn read_existing_admin_server_config_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
+    ecstore_config::com::read_existing_server_config_no_lock(api).await
+}
+
 #[cfg(test)]
 pub(crate) async fn read_admin_config_without_migrate_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
     ecstore_config::com::read_config_without_migrate_no_lock(api).await
 }
 
 pub(crate) type AdminServerConfigSnapshot = ecstore_config::com::ServerConfigSnapshot;
+pub(crate) type AdminServerConfigSaveResult = ecstore_config::com::ServerConfigSaveResult;
 
 pub(crate) async fn save_admin_config(api: Arc<ECStore>, file: &str, data: Vec<u8>) -> Result<()> {
     ecstore_config::com::save_config(api, file, data).await
@@ -690,8 +695,17 @@ pub(crate) async fn save_admin_server_config_snapshot(
     api: Arc<ECStore>,
     cfg: &rustfs_config::server_config::Config,
     snapshot: &AdminServerConfigSnapshot,
-) -> Result<bool> {
-    ecstore_config::com::save_server_config_snapshot(api, cfg, snapshot).await
+) -> Result<AdminServerConfigSaveResult> {
+    ecstore_config::com::save_server_config_snapshot_with_generation(api, cfg, snapshot).await
+}
+
+pub(crate) async fn with_admin_server_config_read_lock<F, Fut, T>(api: Arc<ECStore>, operation: F) -> Result<T>
+where
+    F: FnOnce() -> Fut + Send + 'static,
+    Fut: std::future::Future<Output = T> + Send + 'static,
+    T: Send + 'static,
+{
+    ecstore_config::com::with_server_config_read_lock(api, operation).await
 }
 
 pub(crate) fn init_admin_config_defaults() {
@@ -793,9 +807,10 @@ pub(crate) mod cluster {
 pub(crate) mod config {
     pub(crate) use super::storageclass;
     pub(crate) use super::{
-        AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config, init_admin_config_defaults,
-        read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot, save_admin_config,
-        save_admin_server_config_snapshot,
+        AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config,
+        init_admin_config_defaults, read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot,
+        read_existing_admin_server_config_no_lock, save_admin_config, save_admin_server_config_snapshot,
+        with_admin_server_config_read_lock,
     };
     #[cfg(test)]
     pub(crate) use super::{

scripts/e2e-run.sh

@@ -40,7 +40,19 @@ export RUST_BACKTRACE=full
 "$BIN" --address "127.0.0.1:${RUSTFS_TEST_PORT}" "$VOLUME" > "${RUSTFS_TEST_LOG}" 2>&1 &
 RUSTFS_PID=$!
 
-sleep 10
+for _ in {1..60}; do
+    if curl --noproxy '*' --silent --fail "http://127.0.0.1:${RUSTFS_TEST_PORT}/health/ready" >/dev/null; then
+        break
+    fi
+
+    if ! kill -0 "${RUSTFS_PID}" 2>/dev/null; then
+        wait "${RUSTFS_PID}"
+    fi
+
+    sleep 1
+done
+
+curl --noproxy '*' --silent --show-error --fail "http://127.0.0.1:${RUSTFS_TEST_PORT}/health/ready" >/dev/null
 
 export AWS_ACCESS_KEY_ID="${RUSTFS_ACCESS_KEY:-rustfsadmin}"
 export AWS_SECRET_ACCESS_KEY="${RUSTFS_SECRET_KEY:-rustfsadmin}"