Merge branch 'main' into codex/fix-4846-lock-quorum-timeout

crates/e2e_test/src/copy_object_metadata_test.rs

@@ -117,9 +117,14 @@ mod tests {
             .key("assets/explicit-copy.js")
             .copy_source(format!("{bucket}/{key}"))
             .metadata_directive(MetadataDirective::Copy)
+            .customize()
+            .mutate_request(|request| {
+                request.headers_mut().insert("content-type", "application/octet-stream");
+                request.headers_mut().insert("x-amz-meta-request-only", "ignored");
+            })
             .send()
             .await
-            .expect("explicit COPY directive failed");
+            .expect("explicit COPY directive with request metadata failed");
         let explicit_copy_head = client
             .head_object()
             .bucket(bucket)
@@ -128,6 +133,18 @@ mod tests {
             .await
             .expect("HEAD failed after explicit COPY");
         assert_eq!(explicit_copy_head.cache_control(), Some("max-age=60"));
+        assert_eq!(explicit_copy_head.content_type(), Some("text/javascript; charset=utf-8"));
+        assert_eq!(
+            explicit_copy_head.metadata().and_then(|metadata| metadata.get("mtime")),
+            Some(&"1777992333".to_string())
+        );
+        assert_eq!(
+            explicit_copy_head
+                .metadata()
+                .and_then(|metadata| metadata.get("request-only")),
+            None,
+            "COPY must ignore request metadata"
+        );
         assert_eq!(
             explicit_copy_head.website_redirect_location(),
             None,
@@ -571,20 +588,6 @@ mod tests {
             Some("InvalidArgument")
         );
 
-        let ignored_replacement = client
-            .copy_object()
-            .bucket(bucket)
-            .key(key)
-            .copy_source(format!("{bucket}/{key}"))
-            .content_type("application/ignored")
-            .send()
-            .await
-            .expect_err("Replacement fields without REPLACE should be rejected");
-        assert_eq!(
-            ignored_replacement.as_service_error().and_then(|error| error.code()),
-            Some("InvalidRequest")
-        );
-
         let unchanged = client
             .get_object()
             .bucket(bucket)

crates/e2e_test/src/delete_marker_migration_semantics_test.rs

@@ -56,6 +56,21 @@ mod tests {
         );
     }
 
+    async fn assert_current_list_hides_delete_marker(client: &Client, bucket: &str, key: &str) {
+        let listed = client
+            .list_objects_v2()
+            .bucket(bucket)
+            .prefix(key)
+            .send()
+            .await
+            .expect("list current objects after delete marker");
+
+        assert!(
+            listed.contents().iter().all(|object| object.key() != Some(key)),
+            "ListObjectsV2 must hide an object whose latest version is a delete marker"
+        );
+    }
+
     #[tokio::test]
     #[serial]
     async fn test_versioning_only_delete_marker_has_minio_compatible_visibility_for_migration_proof() {
@@ -94,6 +109,7 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
+        assert_current_list_hides_delete_marker(&client, bucket, key).await;
     }
 
     #[tokio::test]
@@ -118,6 +134,17 @@ mod tests {
             .await
             .expect("put historical version");
         let data_version_id = put.version_id().expect("put should return data version id");
+        let listed_before_delete = client
+            .list_objects_v2()
+            .bucket(bucket)
+            .prefix(key)
+            .send()
+            .await
+            .expect("list current object before creating delete marker");
+        assert!(
+            listed_before_delete.contents().iter().any(|object| object.key() == Some(key)),
+            "ListObjectsV2 must include the current object before it is deleted"
+        );
 
         let delete_marker = client
             .delete_object()
@@ -145,6 +172,7 @@ mod tests {
         assert_eq!(markers[0].version_id(), Some(delete_marker_version_id));
         assert_eq!(markers[0].is_latest(), Some(true));
         assert_current_get_is_delete_marker_not_found(&client, bucket, key).await;
+        assert_current_list_hides_delete_marker(&client, bucket, key).await;
 
         let historical = client
             .get_object()

crates/e2e_test/src/object_lock/object_lock_test.rs

@@ -26,6 +26,7 @@
 
 use super::common::*;
 use aws_sdk_s3::Client;
+use aws_sdk_s3::error::ProvideErrorMetadata;
 use aws_sdk_s3::primitives::{ByteStream, DateTimeFormat};
 use aws_sdk_s3::types::{
     CompletedMultipartUpload, CompletedPart, Delete, MetadataDirective, ObjectIdentifier, ObjectLockLegalHoldStatus,
@@ -2120,6 +2121,127 @@ async fn test_multipart_default_retention_fixed_at_create() {
 // Versioning Auto-Enable Tests
 // ============================================================================
 
+#[tokio::test]
+#[serial]
+async fn test_unretained_object_lock_object_delete_and_bucket_cleanup() {
+    init_logging();
+    info!("🧪 Test: Unretained Object Lock object delete and bucket cleanup (Issue #5339)");
+
+    let mut env = ObjectLockTestEnvironment::new()
+        .await
+        .expect("failed to create Object Lock test environment");
+    env.start_rustfs().await.expect("failed to start RustFS");
+
+    let bucket = "test-object-lock-delete-cleanup";
+    let key = "unretained-object";
+
+    env.create_object_lock_bucket(bucket)
+        .await
+        .expect("failed to create Object Lock bucket");
+    let client = env.s3_client();
+
+    let put_response = client
+        .put_object()
+        .bucket(bucket)
+        .key(key)
+        .body(ByteStream::from_static(b"unretained data"))
+        .send()
+        .await
+        .expect("failed to upload unretained object");
+    let object_version_id = put_response
+        .version_id()
+        .expect("Object Lock buckets must create versioned objects")
+        .to_string();
+
+    let delete_response = client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect("failed to create delete marker");
+    assert_eq!(delete_response.delete_marker(), Some(true));
+    let delete_marker_version_id = delete_response
+        .version_id()
+        .expect("Deleting without a version ID must create a delete marker")
+        .to_string();
+
+    let get_error = client
+        .get_object()
+        .bucket(bucket)
+        .key(key)
+        .send()
+        .await
+        .expect_err("GET must not return an object hidden by a delete marker");
+    assert_eq!(get_error.raw_response().map(|response| response.status().as_u16()), Some(404));
+    assert_eq!(get_error.as_service_error().and_then(|error| error.code()), Some("NoSuchKey"));
+
+    let listed_objects = client
+        .list_objects_v2()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list current objects");
+    assert!(
+        listed_objects.contents().iter().all(|object| object.key() != Some(key)),
+        "ListObjectsV2 must hide objects whose latest version is a delete marker"
+    );
+
+    let listed_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list object versions");
+    assert!(
+        listed_versions
+            .versions()
+            .iter()
+            .any(|version| version.key() == Some(key) && version.version_id() == Some(object_version_id.as_str())),
+        "The data version must remain until it is explicitly deleted"
+    );
+    assert!(
+        listed_versions
+            .delete_markers()
+            .iter()
+            .any(|marker| marker.key() == Some(key) && marker.version_id() == Some(delete_marker_version_id.as_str())),
+        "ListObjectVersions must expose the delete marker"
+    );
+
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(object_version_id)
+        .send()
+        .await
+        .expect("failed to delete the data version");
+    client
+        .delete_object()
+        .bucket(bucket)
+        .key(key)
+        .version_id(delete_marker_version_id)
+        .send()
+        .await
+        .expect("failed to delete the delete marker");
+
+    let remaining_versions = client
+        .list_object_versions()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("failed to list versions after cleanup");
+    assert!(remaining_versions.versions().is_empty());
+    assert!(remaining_versions.delete_markers().is_empty());
+
+    client
+        .delete_bucket()
+        .bucket(bucket)
+        .send()
+        .await
+        .expect("Deleting every version must remove xl.meta so the bucket can be deleted normally");
+}
+
 #[tokio::test]
 #[serial]
 async fn test_versioning_auto_enabled_with_object_lock() {

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/kms/src/backends/local.rs

@@ -46,7 +46,10 @@ use zeroize::Zeroizing;
 /// `key_dir` is accepted, so identifiers already in use by existing deployments keep
 /// resolving. Only separators, traversal and the degenerate cases are refused, which is
 /// what stops `key_dir.join(...)` from escaping.
-fn validate_key_id(key_id: &str) -> Result<()> {
+///
+/// pub(crate) because the backup restore path applies the same containment
+/// rule to key identifiers recovered from bundle artifacts.
+pub(crate) fn validate_key_id(key_id: &str) -> Result<()> {
     if key_id.is_empty() {
         return Err(KmsError::invalid_key("key identifier must not be empty"));
     }
@@ -68,7 +71,15 @@ fn validate_key_id(key_id: &str) -> Result<()> {
     }
 }
 
-const LOCAL_KMS_MASTER_KEY_SALT_FILE: &str = ".master-key.salt";
+// The salt and restore-marker file names are pub(crate) so the backup/restore
+// modules (`crate::backup`) address the exact on-disk names instead of copies
+// that could drift.
+pub(crate) const LOCAL_KMS_MASTER_KEY_SALT_FILE: &str = ".master-key.salt";
+/// Commit marker of an in-progress Local restore cutover (see
+/// `crate::backup::local_restore`). Its presence means the key directory is
+/// mid-cutover: startup must fail closed until the restore is rolled forward
+/// or explicitly aborted.
+pub(crate) const LOCAL_RESTORE_COMMIT_MARKER_FILE: &str = ".restore-commit.json";
 // The KDF parameters are pub(crate) so the backup manifest contract
 // (`crate::backup`) records the exact compiled-in derivation instead of a
 // copy that could drift.
@@ -87,7 +98,11 @@ pub(crate) const LOCAL_KMS_ARGON2_P_COST: u32 = 1;
 /// `foo.tmp-<uuid>` is stored as `foo.tmp-<uuid>.key` — so the `.key` guard
 /// plus the exact hyphenated-UUID check makes it impossible to match an
 /// authoritative file.
-fn is_orphan_commit_temp_name(file_name: &str) -> bool {
+///
+/// pub(crate) because the backup restore path applies the same classification
+/// when it re-enters an interrupted run: a leftover commit temp is never
+/// authoritative state, so it does not make a target non-empty.
+pub(crate) fn is_orphan_commit_temp_name(file_name: &str) -> bool {
     if file_name.ends_with(".key") {
         return false;
     }
@@ -110,12 +125,15 @@ fn is_orphan_commit_temp_name(file_name: &str) -> bool {
 ///
 /// This intentionally mirrors ecstore's fsync helpers without depending on the
 /// ecstore crate: the KMS backend stays decoupled from storage internals.
-mod durable_file {
+///
+/// pub(crate) because the backup restore path (`crate::backup::local_restore`)
+/// commits staged files and its cutover marker through the same protocol.
+pub(crate) mod durable_file {
     use std::io::{self, Write};
     use std::path::{Path, PathBuf};
 
     /// How the fully written temp file becomes visible under its final name.
-    pub(super) enum Publish {
+    pub(crate) enum Publish {
         /// Atomically replace whatever is at the destination via `rename`.
         Replace,
         /// Publish via `hard_link`, failing with [`CommitError::AlreadyExists`]
@@ -124,7 +142,7 @@ mod durable_file {
     }
 
     #[derive(Debug)]
-    pub(super) enum CommitError {
+    pub(crate) enum CommitError {
         AlreadyExists,
         Io(io::Error),
         /// Test-only simulated crash: the protocol stops after the given step
@@ -158,14 +176,14 @@ mod durable_file {
     /// to prove that every interrupted prefix recovers to either the complete
     /// old state or the complete new state.
     #[derive(Debug, Clone, Copy, PartialEq, Eq)]
-    pub(super) enum CommitStep {
+    pub(crate) enum CommitStep {
         TempWritten,
         FileSynced,
         Published,
         DirSynced,
     }
 
-    pub(super) async fn commit(
+    pub(crate) async fn commit(
         temp_path: PathBuf,
         final_path: PathBuf,
         content: Vec<u8>,
@@ -179,7 +197,7 @@ mod durable_file {
 
     /// Remove a published file durably: without the parent directory fsync a
     /// deleted key could resurface after power loss.
-    pub(super) async fn remove_durably(path: PathBuf) -> io::Result<()> {
+    pub(crate) async fn remove_durably(path: PathBuf) -> io::Result<()> {
         tokio::task::spawn_blocking(move || {
             std::fs::remove_file(&path)?;
             let parent = path
@@ -191,6 +209,41 @@ mod durable_file {
         .map_err(io::Error::other)?
     }
 
+    /// Publish an already-durable file under a second name via `hard_link`,
+    /// then fsync the destination's parent directory.
+    ///
+    /// This is the restore cutover primitive: the source (a staged file that
+    /// went through [`commit`]) is already durable, so linking plus a parent
+    /// fsync is a complete publish. `AlreadyExists` is idempotent success only
+    /// when the destination content is byte-identical to the source — that is
+    /// exactly the re-entry case of a cutover interrupted after this link —
+    /// and a hard failure otherwise, so the primitive can never clobber or
+    /// silently accept foreign state.
+    pub(crate) async fn link_durably(source: PathBuf, dest: PathBuf) -> io::Result<()> {
+        tokio::task::spawn_blocking(move || {
+            match std::fs::hard_link(&source, &dest) {
+                Ok(()) => {}
+                Err(error) if error.kind() == io::ErrorKind::AlreadyExists => {
+                    let existing = std::fs::read(&dest)?;
+                    let staged = std::fs::read(&source)?;
+                    if existing != staged {
+                        return Err(io::Error::new(
+                            io::ErrorKind::AlreadyExists,
+                            format!("destination {} already exists with different content", dest.display()),
+                        ));
+                    }
+                }
+                Err(error) => return Err(error),
+            }
+            let parent = dest
+                .parent()
+                .ok_or_else(|| io::Error::other("destination has no parent directory"))?;
+            fsync_dir(parent)
+        })
+        .await
+        .map_err(io::Error::other)?
+    }
+
     fn commit_blocking(
         temp_path: &Path,
         final_path: &Path,
@@ -355,7 +408,7 @@ mod durable_file {
     /// Test-only failpoints simulating a crash after a given commit step.
     /// Armed per directory so parallel tests never affect each other.
     #[cfg(test)]
-    pub(super) mod failpoint {
+    pub(crate) mod failpoint {
         use super::CommitStep;
         use std::path::{Path, PathBuf};
         use std::sync::Mutex;
@@ -460,6 +513,11 @@ impl LocalKmsClient {
             debug!(path = ?config.key_dir, "KMS key directory created");
         }
 
+        // The restore-marker guard must run before anything else touches the
+        // directory (in particular before salt load/creation): a directory
+        // mid-cutover holds an arbitrary mix of old and new state.
+        Self::ensure_no_restore_marker(&config).await?;
+
         // Initialize master cipher if master key is provided
         let (master_cipher, legacy_master_cipher) = if let Some(ref master_key) = config.master_key {
             let salt = Self::load_or_create_master_key_salt(&config).await?;
@@ -491,6 +549,7 @@ impl LocalKmsClient {
         if !fs::try_exists(&config.key_dir).await? {
             return Err(KmsError::configuration_error("Local KMS key directory does not exist"));
         }
+        Self::ensure_no_restore_marker(&config).await?;
 
         let (master_cipher, legacy_master_cipher) = if let Some(ref master_key) = config.master_key {
             let legacy_key = Self::derive_legacy_master_key(master_key)?;
@@ -567,8 +626,36 @@ impl LocalKmsClient {
         Self::master_key_salt_path(&self.config)
     }
 
+    /// Operator-configured master key string, exposed for the backup export
+    /// module so it can record a one-way verifier in the bundle manifest.
+    /// Never log or persist this value.
+    pub(crate) fn configured_master_key(&self) -> Option<&str> {
+        self.config.master_key.as_deref()
+    }
+
+    /// Fail closed while a restore cutover marker is present: the directory
+    /// then holds an arbitrary mix of pre-restore and restored state, and the
+    /// only valid next steps are re-running the restore with the same bundle
+    /// (roll forward) or explicitly aborting it. This mirrors the missing-salt
+    /// guard: startup must never paper over a half-applied restore.
+    async fn ensure_no_restore_marker(config: &LocalConfig) -> Result<()> {
+        let marker = config.key_dir.join(LOCAL_RESTORE_COMMIT_MARKER_FILE);
+        if fs::try_exists(&marker).await? {
+            return Err(KmsError::configuration_error(format!(
+                "Local KMS key directory has an unfinished restore (marker {} present); \
+                 re-run the restore with the same bundle to roll it forward or abort it explicitly",
+                marker.display()
+            )));
+        }
+        Ok(())
+    }
+
     /// Derive a 256-bit key from the master key string using a persistent Argon2id salt.
-    fn derive_master_key(master_key: &str, salt: &[u8]) -> Result<Key<Aes256Gcm>> {
+    ///
+    /// pub(crate) because the backup restore path derives the same key from
+    /// the operator-supplied master key and the bundled salt for its verifier
+    /// check and staged decryption probe.
+    pub(crate) fn derive_master_key(master_key: &str, salt: &[u8]) -> Result<Key<Aes256Gcm>> {
         let params = Params::new(
             LOCAL_KMS_ARGON2_M_COST_KIB,
             LOCAL_KMS_ARGON2_T_COST,
@@ -585,7 +672,7 @@ impl LocalKmsClient {
         Ok(key)
     }
 
-    fn derive_legacy_master_key(master_key: &str) -> Result<Key<Aes256Gcm>> {
+    pub(crate) fn derive_legacy_master_key(master_key: &str) -> Result<Key<Aes256Gcm>> {
         let mut hasher = Sha256::new();
         hasher.update(master_key.as_bytes());
         hasher.update(b"rustfs-kms-local");
@@ -2453,6 +2540,34 @@ mod tests {
         assert!(!is_orphan_commit_temp_name("mykey.tmp-"));
     }
 
+    #[tokio::test]
+    async fn link_durably_publishes_no_clobber_and_is_content_idempotent() {
+        let temp_dir = TempDir::new().expect("temp dir");
+        let source = temp_dir.path().join("staged");
+        let dest = temp_dir.path().join("published");
+        fs::write(&source, b"staged-content").await.expect("write source");
+
+        durable_file::link_durably(source.clone(), dest.clone())
+            .await
+            .expect("first link must succeed");
+        assert_eq!(fs::read(&dest).await.expect("read dest"), b"staged-content");
+
+        // Re-entry with identical content is idempotent success — exactly the
+        // resumed-cutover case.
+        durable_file::link_durably(source.clone(), dest.clone())
+            .await
+            .expect("re-linking identical content must be idempotent");
+
+        // Existing content that differs is a hard failure, never a clobber.
+        let foreign = temp_dir.path().join("foreign");
+        fs::write(&foreign, b"different-content").await.expect("write foreign");
+        let error = durable_file::link_durably(foreign, dest.clone())
+            .await
+            .expect_err("differing content must not be clobbered");
+        assert_eq!(error.kind(), std::io::ErrorKind::AlreadyExists);
+        assert_eq!(fs::read(&dest).await.expect("dest unchanged"), b"staged-content");
+    }
+
     #[tokio::test]
     async fn durable_commit_fsyncs_every_write_path() {
         use durable_file::fsync_recorder;

crates/kms/src/backends/scripted_vault.rs

@@ -61,7 +61,7 @@ impl ScriptedResponse {
 pub(crate) struct ScriptedVault {
     /// Base address (`http://127.0.0.1:port`) to point a Vault client at.
     pub(crate) address: String,
-    requests: Arc<Mutex<Vec<String>>>,
+    requests: Arc<Mutex<Vec<(String, String)>>>,
 }
 
 impl ScriptedVault {
@@ -81,13 +81,10 @@ impl ScriptedVault {
                 let Ok((mut stream, _)) = listener.accept().await else {
                     return;
                 };
-                let Some(request_line) = read_request(&mut stream).await else {
+                let Some(request) = read_request(&mut stream).await else {
                     continue;
                 };
-                recorded
-                    .lock()
-                    .expect("scripted vault request log poisoned")
-                    .push(request_line);
+                recorded.lock().expect("scripted vault request log poisoned").push(request);
                 let response = responses
                     .next()
                     .unwrap_or_else(|| ScriptedResponse::error(599, "scripted vault: script exhausted"));
@@ -107,14 +104,32 @@ impl ScriptedVault {
 
     /// The `METHOD /path` lines of every request served so far, in order.
     pub(crate) fn requests(&self) -> Vec<String> {
-        self.requests.lock().expect("scripted vault request log poisoned").clone()
+        self.requests
+            .lock()
+            .expect("scripted vault request log poisoned")
+            .iter()
+            .map(|(line, _)| line.clone())
+            .collect()
+    }
+
+    /// The request bodies, in the same order as [`Self::requests`]; empty for
+    /// bodyless requests. Lets tests assert what a write actually persisted
+    /// (record contents, check-and-set options), not just that a write happened.
+    pub(crate) fn request_bodies(&self) -> Vec<String> {
+        self.requests
+            .lock()
+            .expect("scripted vault request log poisoned")
+            .iter()
+            .map(|(_, body)| body.clone())
+            .collect()
     }
 }
 
 /// Read one HTTP/1.1 request (head plus content-length body) and return its
-/// `METHOD /path` line. Draining the body before responding keeps the client
-/// from seeing a connection reset while it is still writing.
-async fn read_request(stream: &mut TcpStream) -> Option<String> {
+/// `METHOD /path` line together with the body. Draining the body before
+/// responding keeps the client from seeing a connection reset while it is
+/// still writing.
+async fn read_request(stream: &mut TcpStream) -> Option<(String, String)> {
     let mut buffer = Vec::new();
     let mut chunk = [0u8; 4096];
     let head_end = loop {
@@ -146,14 +161,17 @@ async fn read_request(stream: &mut TcpStream) -> Option<String> {
         })
         .next()
         .unwrap_or(0);
-    let mut remaining = content_length.saturating_sub(buffer.len() - head_end);
+    let mut body = buffer[head_end..].to_vec();
+    let mut remaining = content_length.saturating_sub(body.len());
     while remaining > 0 {
         let read = stream.read(&mut chunk).await.ok()?;
         if read == 0 {
             break;
         }
+        body.extend_from_slice(&chunk[..read]);
         remaining = remaining.saturating_sub(read);
     }
+    body.truncate(content_length);
 
-    Some(format!("{method} {path}"))
+    Some((format!("{method} {path}"), String::from_utf8_lossy(&body).into_owned()))
 }

crates/kms/src/backup/local_export.rs

@@ -14,9 +14,9 @@
 
 //! Local backend backup export: sealed, KEK-protected bundle production.
 //!
-//! This is the producer side only; restore lives in a follow-up change. The
-//! admin API is not wired here either — callers construct the request and
-//! supply the backup KEK explicitly.
+//! This is the producer side; the consumer side lives in
+//! [`crate::backup::local_restore`]. The admin API is not wired here —
+//! callers construct the request and supply the backup KEK explicitly.
 //!
 //! # Bundle layout
 //!
@@ -63,6 +63,7 @@ use aes_gcm::{
 use jiff::Zoned;
 use rand::RngExt;
 use serde::Deserialize;
+use sha2::{Digest, Sha256};
 use std::path::{Path, PathBuf};
 use tokio::fs;
 use tokio::io::AsyncWriteExt;
@@ -76,6 +77,12 @@ const SALT_ARTIFACT_PATH: &str = "artifacts/master-key.salt.enc";
 const AEAD_NONCE_LEN: usize = 12;
 /// Domain-separation context for the artifact AAD binding.
 const BUNDLE_AAD_CONTEXT: &str = "rustfs-kms-local-backup:v1";
+/// Domain-separation context for the master-key verifier.
+const MASTER_KEY_VERIFIER_CONTEXT: &str = "rustfs-kms-local-master-key-verifier:v1";
+/// Verifier scheme prefix for the Argon2id (salted) derivation.
+pub(crate) const MASTER_KEY_VERIFIER_ARGON2ID_PREFIX: &str = "argon2id-v1:";
+/// Verifier scheme prefix for the legacy pre-beta.9 SHA-256 derivation.
+pub(crate) const MASTER_KEY_VERIFIER_LEGACY_PREFIX: &str = "legacy-sha256-v1:";
 
 /// Caller-supplied backup KEK: a trust root deliberately separate from the
 /// business KMS hierarchy (it must not be a key that is itself part of the
@@ -208,7 +215,15 @@ pub async fn export_local_backup(
         ));
     }
 
-    let manifest = build_and_write_bundle(kek, request, &snapshot).await?;
+    // The verifier lets a restore detect a wrong operator-supplied master key
+    // before touching any target state; dev-mode directories have no master
+    // key to verify.
+    let master_key_verifier = match client.configured_master_key() {
+        Some(master_key) => Some(compute_master_key_verifier(master_key, snapshot.salt.as_deref(), &request.backup_id)?),
+        None => None,
+    };
+
+    let manifest = build_and_write_bundle(kek, request, &snapshot, master_key_verifier).await?;
     Ok(manifest)
 }
 
@@ -262,6 +277,21 @@ pub async fn decrypt_bundle_artifact(
         Err(error) => return Err(error.into()),
     };
 
+    decrypt_artifact_payload(&manifest.backup_id, manifest.snapshot_generation, descriptor, kek, &payload)
+}
+
+/// Verify and decrypt one artifact payload already read into memory.
+///
+/// Fail-closed order: declared length, encrypted digest, nonce framing, then
+/// AEAD authentication. Shared by [`decrypt_bundle_artifact`] and the export
+/// pre-seal probe so producer and consumer can never drift on the framing.
+fn decrypt_artifact_payload(
+    backup_id: &str,
+    snapshot_generation: u64,
+    descriptor: &ArtifactDescriptor,
+    kek: &BackupKek,
+    payload: &[u8],
+) -> Result<Zeroizing<Vec<u8>>> {
     if (payload.len() as u64) < descriptor.len {
         return Err(BackupError::truncated(format!(
             "artifact '{}' is {} bytes, manifest declares {}",
@@ -280,7 +310,7 @@ pub async fn decrypt_bundle_artifact(
         ))
         .into());
     }
-    if ContentDigest::sha256_of(&payload) != descriptor.encrypted_digest {
+    if ContentDigest::sha256_of(payload) != descriptor.encrypted_digest {
         return Err(BackupError::corrupted(format!("artifact '{}' does not match its manifest digest", descriptor.path)).into());
     }
     if payload.len() < AEAD_NONCE_LEN {
@@ -290,7 +320,7 @@ pub async fn decrypt_bundle_artifact(
     let (nonce_bytes, ciphertext) = payload.split_at(AEAD_NONCE_LEN);
     let mut nonce = [0u8; AEAD_NONCE_LEN];
     nonce.copy_from_slice(nonce_bytes);
-    let aad = artifact_aad(&manifest.backup_id, manifest.snapshot_generation, &descriptor.path);
+    let aad = artifact_aad(backup_id, snapshot_generation, &descriptor.path);
     let plaintext = kek
         .cipher()
         .decrypt(
@@ -360,6 +390,7 @@ async fn build_and_write_bundle(
     kek: &BackupKek,
     request: &LocalBackupExportRequest,
     snapshot: &CollectedSnapshot,
+    master_key_verifier: Option<String>,
 ) -> Result<BackupManifest> {
     let mut artifacts = Vec::with_capacity(snapshot.records.len() + 1);
     for record in &snapshot.records {
@@ -392,7 +423,7 @@ async fn build_and_write_bundle(
         snapshot_generation: request.snapshot_generation,
         backup_kek: kek.descriptor(),
         artifacts,
-        local_kdf: Some(local_kdf_descriptor(snapshot)),
+        local_kdf: Some(local_kdf_descriptor(snapshot, master_key_verifier)),
         key_versions: None,
         capability_discovery: None,
         completeness: CompletenessState::InProgress,
@@ -448,13 +479,25 @@ async fn encrypt_and_write_artifact(
         )));
     }
 
-    Ok(ArtifactDescriptor {
+    let descriptor = ArtifactDescriptor {
         kind,
         path: artifact_path.to_string(),
         len: payload.len() as u64,
         aead_algorithm: AeadAlgorithm::Aes256Gcm,
         encrypted_digest: digest,
-    })
+    };
+
+    // Pre-seal decryption probe: digest equality only proves the ciphertext
+    // landed intact; this proves the stored artifact actually opens under the
+    // backup KEK and AAD binding before the manifest may reference it.
+    let reopened = decrypt_artifact_payload(&request.backup_id, request.snapshot_generation, &descriptor, kek, &written)?;
+    if reopened.as_slice() != plaintext {
+        return Err(KmsError::internal_error(format!(
+            "bundle artifact '{artifact_path}' failed the pre-seal decryption probe"
+        )));
+    }
+
+    Ok(descriptor)
 }
 
 /// AAD binding an artifact to its bundle identity and path. A JSON tuple
@@ -464,6 +507,29 @@ fn artifact_aad(backup_id: &str, snapshot_generation: u64, artifact_path: &str)
         .expect("AAD tuple of strings and integers always serializes")
 }
 
+/// Compute the opaque one-way master-key verifier recorded in the manifest:
+/// `<scheme-prefix>` + `hex(SHA-256(json(context, backup_id) || derived_key))`.
+///
+/// The derivation follows the directory's KDF state: Argon2id over the
+/// persistent salt when one exists, the legacy SHA-256 derivation otherwise.
+/// The verifier is not an offline-guessing oracle: computing a candidate
+/// requires the salt, which exists only inside the KEK-sealed bundle, and a
+/// party holding the KEK already has the strictly stronger oracle of the
+/// artifact AEAD tags — while every guess still pays the full Argon2id cost.
+/// Binding `backup_id` prevents cross-bundle fingerprint correlation.
+pub(crate) fn compute_master_key_verifier(master_key: &str, salt: Option<&[u8]>, backup_id: &str) -> Result<String> {
+    let framing = serde_json::to_vec(&(MASTER_KEY_VERIFIER_CONTEXT, backup_id))
+        .expect("verifier framing tuple of strings always serializes");
+    let (prefix, derived) = match salt {
+        Some(salt) => (MASTER_KEY_VERIFIER_ARGON2ID_PREFIX, LocalKmsClient::derive_master_key(master_key, salt)?),
+        None => (MASTER_KEY_VERIFIER_LEGACY_PREFIX, LocalKmsClient::derive_legacy_master_key(master_key)?),
+    };
+    let mut hasher = Sha256::new();
+    hasher.update(&framing);
+    hasher.update(derived.as_slice());
+    Ok(format!("{prefix}{}", hex::encode(hasher.finalize())))
+}
+
 /// The bundle-level protection label is the weakest state observed across
 /// records: any plaintext-dev-only record marks the whole bundle, then any
 /// legacy-unspecified marker (unknown until read), and only a uniformly
@@ -484,7 +550,7 @@ fn weakest_observed_protection(records: &[CollectedRecord]) -> AtRestProtection
     }
 }
 
-fn local_kdf_descriptor(snapshot: &CollectedSnapshot) -> LocalKdfDescriptor {
+fn local_kdf_descriptor(snapshot: &CollectedSnapshot, master_key_verifier: Option<String>) -> LocalKdfDescriptor {
     let mut modes = Vec::new();
     for (marker, mode) in [
         (StoredKeyProtection::EncryptedMasterKey, AtRestProtection::EncryptedMasterKey),
@@ -508,9 +574,7 @@ fn local_kdf_descriptor(snapshot: &CollectedSnapshot) -> LocalKdfDescriptor {
     LocalKdfDescriptor {
         derivation,
         protection_modes: modes,
-        // The verifier shape is left to the restore change; the schema keeps
-        // it optional so bundles without one stay valid.
-        master_key_verifier: None,
+        master_key_verifier,
     }
 }
 
@@ -543,8 +607,9 @@ async fn write_new_file(path: &Path, bytes: &[u8]) -> Result<()> {
 
 /// Fsync a directory so freshly created bundle entries survive power loss.
 /// No-op on non-Unix platforms where directories cannot be opened for
-/// syncing (mirrors the local backend's durable commit helper).
-async fn fsync_dir(path: &Path) -> Result<()> {
+/// syncing (mirrors the local backend's durable commit helper). Shared with
+/// the restore module for the same durability points on the target side.
+pub(crate) async fn fsync_dir(path: &Path) -> Result<()> {
     #[cfg(unix)]
     {
         let path = path.to_path_buf();
@@ -560,7 +625,6 @@ async fn fsync_dir(path: &Path) -> Result<()> {
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::backends::KmsClient;
     use crate::config::LocalConfig;
     use std::sync::Arc;
     use tempfile::TempDir;
@@ -726,6 +790,37 @@ mod tests {
         assert_eq!(decrypted.as_slice(), source_record.as_slice());
     }
 
+    #[tokio::test]
+    async fn export_records_a_master_key_verifier_matching_recomputation() {
+        let (client, _key_dir) = encrypted_client().await;
+        client.create_key("verified", "AES_256", None).await.expect("create key");
+
+        let bundle = TempDir::new().expect("bundle dir");
+        let manifest = export_local_backup(&client, &test_kek(), &export_request(bundle.path().join("bundle")))
+            .await
+            .expect("export should succeed");
+
+        let verifier = manifest
+            .local_kdf
+            .as_ref()
+            .expect("local kdf descriptor")
+            .master_key_verifier
+            .as_deref()
+            .expect("encrypted bundles must record a verifier");
+        assert!(verifier.starts_with(MASTER_KEY_VERIFIER_ARGON2ID_PREFIX), "got {verifier:?}");
+
+        // The verifier is a pure function of (master key, salt, backup id):
+        // the restore side recomputes it from the operator-supplied key and
+        // the bundled salt.
+        let salt = fs::read(client.master_key_salt_file()).await.expect("salt");
+        let recomputed = compute_master_key_verifier("test-master-key", Some(&salt), "backup-0001").expect("recompute");
+        assert_eq!(recomputed, verifier);
+        let wrong_key = compute_master_key_verifier("wrong-master-key", Some(&salt), "backup-0001").expect("recompute");
+        assert_ne!(wrong_key, verifier, "a different master key must change the verifier");
+        let other_bundle = compute_master_key_verifier("test-master-key", Some(&salt), "backup-0002").expect("recompute");
+        assert_ne!(other_bundle, verifier, "verifiers must be bundle-bound");
+    }
+
     #[tokio::test]
     async fn export_fence_blocks_writers_until_released() {
         let (client, _key_dir) = encrypted_client().await;

crates/kms/src/backup/mod.rs

@@ -16,9 +16,10 @@
 //!
 //! The contract side defines the versioned backup manifest, the per-backend
 //! responsibility matrix, typed failure modes, and the restore dry-run
-//! report. [`local_export`] implements the producer side for the Local
-//! backend as a crate-internal API; restore orchestration and the admin API
-//! build on these pieces in follow-up changes.
+//! report. [`local_export`] implements the producer side and
+//! [`local_restore`] the consumer side for the Local backend as
+//! crate-internal APIs; the admin API builds on these pieces in follow-up
+//! changes.
 //!
 //! # Bundle model
 //!
@@ -51,6 +52,7 @@ mod capability;
 mod dry_run;
 mod error;
 pub mod local_export;
+pub mod local_restore;
 mod manifest;
 
 pub use capability::{AtRestProtection, BackupBackendKind, BackupResponsibility};
@@ -62,6 +64,10 @@ pub use local_export::{
     BackupKek, LOCAL_BUNDLE_MANIFEST_FILE, LocalBackupExportRequest, decrypt_bundle_artifact, export_local_backup,
     read_local_bundle_manifest,
 };
+pub use local_restore::{
+    LocalRestoreReport, LocalRestoreRequest, RestoreConflictPolicy, abort_local_restore, dry_run_local_restore,
+    restore_local_backup,
+};
 pub use manifest::{
     AeadAlgorithm, ArtifactDescriptor, ArtifactKind, BackupKekDescriptor, BackupManifest, CompletenessState, ContentDigest,
     DigestAlgorithm, LocalKdfDescriptor, LocalKeyDerivation, ReservedSlot,

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,
     };
 }

crates/policy/src/policy.rs

@@ -72,4 +72,10 @@ pub enum Error {
 
     #[error("invalid resource, type: '{0}', pattern: '{1}'")]
     InvalidResource(String, String),
+
+    #[error("KMS resources require a statement whose actions are all KMS actions")]
+    KmsResourceWithNonKmsAction,
+
+    #[error("bucket policies do not support KMS actions or resources")]
+    KmsUnsupportedInBucketPolicy,
 }

crates/policy/src/policy/action.rs

@@ -728,6 +728,8 @@ pub enum KmsAction {
     ListKeysAction,
     #[strum(serialize = "kms:DescribeKey")]
     DescribeKeyAction,
+    #[strum(serialize = "kms:Decrypt")]
+    DecryptAction,
 }
 
 #[cfg(test)]
@@ -764,6 +766,7 @@ mod tests {
             ("kms:RotateKey", KmsAction::RotateKeyAction),
             ("kms:ListKeys", KmsAction::ListKeysAction),
             ("kms:DescribeKey", KmsAction::DescribeKeyAction),
+            ("kms:Decrypt", KmsAction::DecryptAction),
         ] {
             let action = Action::try_from(raw).expect("Should parse KMS action");
             assert_eq!(action, Action::KmsAction(expected));

crates/policy/src/policy/policy.rs

@@ -1760,6 +1760,391 @@ mod test {
         );
     }
 
+    fn kms_args<'a>(
+        action: Action,
+        key_id: &'a str,
+        conditions: &'a HashMap<String, Vec<String>>,
+        claims: &'a HashMap<String, Value>,
+    ) -> Args<'a> {
+        Args {
+            account: "testuser",
+            groups: &None,
+            action,
+            bucket: "",
+            conditions,
+            is_owner: false,
+            object: key_id,
+            claims,
+            deny_only: false,
+        }
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_key_resource_scopes_by_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DisableKeyAction);
+
+        assert!(
+            policy.is_allowed(&kms_args(action, "key-a", &conditions, &claims)).await,
+            "the granted key must be allowed"
+        );
+        assert!(
+            !policy.is_allowed(&kms_args(action, "key-b", &conditions, &claims)).await,
+            "a key outside the granted resource must be denied"
+        );
+        assert!(
+            !policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::EnableKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "an action outside the grant must stay denied even for the granted key"
+        );
+        assert!(
+            policy.is_allowed(&kms_args(action, "", &conditions, &claims)).await,
+            "call sites that do not pass a key resource keep the legacy match-every-key behaviour"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_wildcard_key_resource() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:GenerateDataKey"],
+      "Resource": ["arn:aws:kms:::key/app-*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::GenerateDataKeyAction);
+
+        assert!(
+            policy
+                .is_allowed(&kms_args(action, "app-primary", &conditions, &claims))
+                .await
+        );
+        assert!(
+            !policy
+                .is_allowed(&kms_args(action, "backup-primary", &conditions, &claims))
+                .await
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_without_resource_matches_every_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+
+        for key_id in ["", "key-a", "any-other-key"] {
+            assert!(
+                policy
+                    .is_allowed(&kms_args(Action::KmsAction(KmsAction::DisableKeyAction), key_id, &conditions, &claims))
+                    .await,
+                "resource-less KMS statement must keep matching every key (key_id: {key_id:?})"
+            );
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_deny_with_wildcard_resource_overrides_allow() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:*"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    },
+    {
+      "Effect": "Deny",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:kms:::key/*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+
+        assert!(
+            !policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::DisableKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "a wildcard Deny must override the narrower Allow"
+        );
+        assert!(
+            policy
+                .is_allowed(&kms_args(Action::KmsAction(KmsAction::RotateKeyAction), "key-a", &conditions, &claims))
+                .await,
+            "actions outside the Deny keep the Allow"
+        );
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_not_resource_excludes_keys() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DescribeKey"],
+      "NotResource": ["arn:aws:kms:::key/prod-*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DescribeKeyAction);
+
+        assert!(policy.is_allowed(&kms_args(action, "dev-key", &conditions, &claims)).await);
+        assert!(!policy.is_allowed(&kms_args(action, "prod-key", &conditions, &claims)).await);
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_statement_with_s3_resource_is_treated_as_unscoped() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        // Statements combining KMS actions with S3 resources predate KMS resource
+        // support and were always evaluated as if unscoped. Pin that they still
+        // match every key (a warning is logged during evaluation).
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DisableKey"],
+      "Resource": ["arn:aws:s3:::somebucket/*"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DisableKeyAction);
+
+        for key_id in ["", "key-a"] {
+            assert!(
+                policy.is_allowed(&kms_args(action, key_id, &conditions, &claims)).await,
+                "malformed KMS statement with S3 resources must keep matching every key (key_id: {key_id:?})"
+            );
+        }
+
+        Ok(())
+    }
+
+    #[tokio::test]
+    async fn test_kms_alias_resource_parses_but_matches_no_key() -> Result<()> {
+        use crate::policy::action::{Action, KmsAction};
+
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:DescribeKey"],
+      "Resource": ["arn:aws:kms:::alias/app-alias"]
+    }
+  ]
+}"#,
+        )?;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let action = Action::KmsAction(KmsAction::DescribeKeyAction);
+
+        assert!(
+            !policy.is_allowed(&kms_args(action, "app-alias", &conditions, &claims)).await,
+            "alias patterns are parse-only until alias resolution lands and must not match key requests"
+        );
+        assert!(
+            policy.is_allowed(&kms_args(action, "", &conditions, &claims)).await,
+            "call sites without a key resource keep the legacy match-every-key behaviour"
+        );
+
+        Ok(())
+    }
+
+    #[test]
+    fn test_kms_resource_policy_round_trips() {
+        let policy = Policy::parse_config(
+            br#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Action": ["kms:GenerateDataKey", "kms:Decrypt"],
+      "Resource": ["arn:aws:kms:::key/app-*", "arn:aws:kms:::alias/app-alias"]
+    }
+  ]
+}"#,
+        )
+        .expect("KMS resource policy should parse");
+
+        let json = serde_json::to_string(&policy).expect("policy should serialize");
+        let round_trip = Policy::parse_config(json.as_bytes()).expect("serialized KMS resource policy should re-parse");
+        assert_eq!(round_trip.statements[0].resources, policy.statements[0].resources);
+        assert_eq!(round_trip.statements[0].actions, policy.statements[0].actions);
+
+        let value: serde_json::Value = serde_json::from_str(&json).expect("JSON valid");
+        let resources: Vec<_> = value["Statement"][0]["Resource"]
+            .as_array()
+            .expect("Resource should serialize as an array")
+            .iter()
+            .map(|resource| resource.as_str().expect("Resource entries should be strings"))
+            .collect();
+        assert_eq!(resources, vec!["arn:aws:kms:::key/app-*", "arn:aws:kms:::alias/app-alias"]);
+    }
+
+    #[test]
+    fn test_kms_resource_with_non_kms_action_is_invalid() {
+        for action in ["s3:GetObject", "admin:ServerInfo", "sts:AssumeRole"] {
+            let data = format!(
+                r#"{{
+  "Version": "2012-10-17",
+  "Statement": [
+    {{
+      "Effect": "Allow",
+      "Action": ["{action}"],
+      "Resource": ["arn:aws:kms:::key/key-a"]
+    }}
+  ]
+}}"#
+            );
+
+            let result = Policy::parse_config(data.as_bytes());
+            assert!(
+                matches!(result.as_ref().unwrap_err(), Error::PolicyError(IamError::KmsResourceWithNonKmsAction)),
+                "{action} with a KMS resource should fail with KmsResourceWithNonKmsAction, got: {result:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn test_bucket_policy_with_kms_statement_is_invalid() {
+        for statement in [
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"Action":["kms:GenerateDataKey"],"Resource":["arn:aws:s3:::bucket/*"]}"#,
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"Action":["s3:GetObject"],"Resource":["arn:aws:kms:::key/key-a"]}"#,
+            r#"{"Effect":"Allow","Principal":{"AWS":"*"},"NotAction":["kms:*"],"Resource":["arn:aws:s3:::bucket/*"]}"#,
+        ] {
+            let data = format!(r#"{{"Version":"2012-10-17","Statement":[{statement}]}}"#);
+            let policy: BucketPolicy =
+                serde_json::from_str(&data).expect("bucket policy with KMS content should still deserialize");
+            let result = policy.is_valid();
+            assert!(
+                matches!(result.as_ref().unwrap_err(), Error::PolicyError(IamError::KmsUnsupportedInBucketPolicy)),
+                "bucket policy statement {statement} should fail with KmsUnsupportedInBucketPolicy, got: {result:?}"
+            );
+        }
+    }
+
+    #[tokio::test]
+    async fn test_stored_bucket_policy_with_kms_statement_loads_and_is_ignored() -> Result<()> {
+        // Policies stored before KMS statements were rejected at validation must keep
+        // deserializing, and their KMS statements must not affect bucket traffic.
+        let bucket_policy: BucketPolicy = serde_json::from_str(
+            r#"{
+  "Version": "2012-10-17",
+  "Statement": [
+    {
+      "Effect": "Allow",
+      "Principal": {"AWS": "*"},
+      "Action": ["kms:GenerateDataKey"],
+      "Resource": ["arn:aws:s3:::bucket/*"]
+    },
+    {
+      "Effect": "Allow",
+      "Principal": {"AWS": "*"},
+      "Action": ["s3:GetObject"],
+      "Resource": ["arn:aws:s3:::bucket/*"]
+    }
+  ]
+}"#,
+        )?;
+
+        let conditions = HashMap::new();
+        let args = BucketPolicyArgs {
+            account: "testuser",
+            groups: &None,
+            action: Action::S3Action(crate::policy::action::S3Action::GetObjectAction),
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "a.txt",
+        };
+        assert!(
+            bucket_policy.is_allowed(&args).await,
+            "the S3 statement must keep working alongside an ignored KMS statement"
+        );
+
+        let put_args = BucketPolicyArgs {
+            account: "testuser",
+            groups: &None,
+            action: Action::S3Action(crate::policy::action::S3Action::PutObjectAction),
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "a.txt",
+        };
+        assert!(
+            !bucket_policy.is_allowed(&put_args).await,
+            "the ignored KMS statement must not grant anything"
+        );
+
+        Ok(())
+    }
+
     #[test]
     fn test_mixed_action_families_are_invalid_even_with_resource() {
         let data = r#"

crates/policy/src/policy/resource.rs

@@ -40,7 +40,7 @@ impl Serialize for ResourceSet {
         for resource in &self.0 {
             let resource_str = match resource {
                 Resource::S3(value) => format!("{}{}", Resource::S3_PREFIX, value),
-                Resource::Kms(value) => value.clone(),
+                Resource::Kms(value) => format!("{}{}", Resource::KMS_PREFIX, value),
             };
             seq.serialize_element(&resource_str)?;
         }
@@ -171,6 +171,20 @@ pub enum Resource {
 
 impl Resource {
     pub const S3_PREFIX: &'static str = "arn:aws:s3:::";
+    /// KMS ARNs use the same empty-account form as [`Self::S3_PREFIX`]; the suffix is
+    /// `key/<key_id>` (wildcards allowed in the id), `alias/<name>`, or a bare `*`.
+    pub const KMS_PREFIX: &'static str = "arn:aws:kms:::";
+    /// Resource-type segment for key ids; request-side KMS resource strings are
+    /// `key/<key_id>` so they line up with these patterns.
+    pub const KMS_KEY_SEGMENT: &'static str = "key/";
+    /// Resource-type segment reserved for key aliases. Alias patterns parse and
+    /// validate, but requests are always evaluated against `key/<key_id>` strings,
+    /// so an alias pattern matches nothing until alias resolution lands.
+    pub const KMS_ALIAS_SEGMENT: &'static str = "alias/";
+
+    pub fn is_kms(&self) -> bool {
+        matches!(self, Resource::Kms(_))
+    }
 
     pub async fn is_match(&self, resource: &str, conditions: &HashMap<String, Vec<String>>) -> bool {
         self.is_match_with_resolver(resource, conditions, None).await
@@ -228,10 +242,13 @@ impl Resource {
 impl TryFrom<&str> for Resource {
     type Error = Error;
     fn try_from(value: &str) -> std::result::Result<Self, Self::Error> {
-        let Some(value) = value.strip_prefix(Self::S3_PREFIX) else {
+        let resource = if let Some(suffix) = value.strip_prefix(Self::S3_PREFIX) {
+            Resource::S3(suffix.into())
+        } else if let Some(suffix) = value.strip_prefix(Self::KMS_PREFIX) {
+            Resource::Kms(suffix.into())
+        } else {
             return Err(IamError::InvalidResource("unknown".into(), value.into()).into());
         };
-        let resource = Resource::S3(value.into());
 
         resource.is_valid()?;
         Ok(resource)
@@ -248,13 +265,17 @@ impl Validator for Resource {
                 }
             }
             Self::Kms(pattern) => {
-                if pattern.is_empty()
+                // A bare `*` matches every key resource; anything else must carry a
+                // resource-type segment. Key ids never contain separators (the KMS
+                // backends reject them), while alias names may nest ("alias/aws/s3").
+                let well_formed = pattern == "*"
                     || pattern
-                        .char_indices()
-                        .find(|&(_, c)| c == '/' || c == '\\' || c == '.')
-                        .map(|(i, _)| i)
-                        .is_some()
-                {
+                        .strip_prefix(Self::KMS_KEY_SEGMENT)
+                        .is_some_and(|id| !id.is_empty() && !id.contains('/') && !id.contains('\\'))
+                    || pattern
+                        .strip_prefix(Self::KMS_ALIAS_SEGMENT)
+                        .is_some_and(|name| !name.is_empty() && !name.contains('\\'));
+                if !well_formed {
                     return Err(IamError::InvalidResource("kms".into(), pattern.into()).into());
                 }
             }
@@ -270,7 +291,7 @@ impl Serialize for Resource {
     {
         match self {
             Resource::S3(s) => serializer.serialize_str(&format!("{}{}", Self::S3_PREFIX, s)),
-            Resource::Kms(s) => serializer.serialize_str(s),
+            Resource::Kms(s) => serializer.serialize_str(&format!("{}{}", Self::KMS_PREFIX, s)),
         }
     }
 }
@@ -308,9 +329,71 @@ mod tests {
     #[test_case("arn:aws:s3:::mybucket","mybucket/myobject" => false; "15")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/../victim-bucket/evil.txt" => false; "16")]
     #[test_case("arn:aws:s3:::attacker-bucket/*","attacker-bucket/safe/../../victim-bucket/evil.txt" => false; "17")]
+    #[test_case("arn:aws:kms:::key/mykey","key/mykey" => true; "kms exact key")]
+    #[test_case("arn:aws:kms:::key/mykey","key/otherkey" => false; "kms wrong key")]
+    #[test_case("arn:aws:kms:::key/*","key/mykey" => true; "kms key wildcard")]
+    #[test_case("arn:aws:kms:::key/app-*","key/app-primary" => true; "kms key prefix wildcard")]
+    #[test_case("arn:aws:kms:::key/app-*","key/backup-primary" => false; "kms key prefix mismatch")]
+    #[test_case("arn:aws:kms:::key/mykey?","key/mykey1" => true; "kms key question mark")]
+    #[test_case("arn:aws:kms:::*","key/mykey" => true; "kms bare star")]
+    #[test_case("arn:aws:kms:::key/mykey","key/mykey/../otherkey" => false; "kms traversal cleaned")]
+    #[test_case("arn:aws:kms:::alias/myalias","key/myalias" => false; "kms alias never matches key")]
+    #[test_case("arn:aws:kms:::alias/*","key/mykey" => false; "kms alias wildcard never matches key")]
+    #[test_case("arn:aws:kms:::key/mykey","mykey" => false; "kms bare id lacks key segment")]
     fn test_resource_is_match(resource: &str, object: &str) -> bool {
         let resource: Resource = resource.try_into().unwrap();
 
         pollster::block_on(resource.is_match(object, &HashMap::new()))
     }
+
+    #[test_case("arn:aws:kms:::key/mykey" => true; "key id parses")]
+    #[test_case("arn:aws:kms:::key/*" => true; "key wildcard parses")]
+    #[test_case("arn:aws:kms:::key/app-key.v2" => true; "key id with dot parses")]
+    #[test_case("arn:aws:kms:::alias/myalias" => true; "alias parses")]
+    #[test_case("arn:aws:kms:::alias/aws/s3" => true; "nested alias parses")]
+    #[test_case("arn:aws:kms:::*" => true; "bare star parses")]
+    #[test_case("arn:aws:kms:::" => false; "empty suffix rejected")]
+    #[test_case("arn:aws:kms:::key/" => false; "empty key id rejected")]
+    #[test_case("arn:aws:kms:::alias/" => false; "empty alias name rejected")]
+    #[test_case("arn:aws:kms:::mykey" => false; "missing resource type segment rejected")]
+    #[test_case("arn:aws:kms:::key/a/b" => false; "separator in key id rejected")]
+    #[test_case("arn:aws:kms:::key/a\\b" => false; "backslash in key id rejected")]
+    #[test_case("arn:aws:kms:us-east-1:123456789012:key/mykey" => false; "region and account form rejected")]
+    fn test_kms_resource_parse(resource: &str) -> bool {
+        Resource::try_from(resource).is_ok()
+    }
+
+    #[test]
+    fn test_kms_resource_serialization_round_trip() {
+        for raw in [
+            "arn:aws:kms:::key/mykey",
+            "arn:aws:kms:::key/*",
+            "arn:aws:kms:::alias/myalias",
+            "arn:aws:kms:::*",
+        ] {
+            let resource = Resource::try_from(raw).expect("KMS resource should parse");
+            assert!(resource.is_kms());
+
+            let json = serde_json::to_string(&resource).expect("KMS resource should serialize");
+            assert_eq!(json, format!("\"{raw}\""), "serialization must write back the full ARN");
+
+            let round_trip: Resource = serde_json::from_str(&json).expect("serialized KMS resource should deserialize");
+            assert_eq!(round_trip, resource);
+        }
+    }
+
+    #[test]
+    fn test_kms_resource_set_serialization_round_trip() {
+        use crate::policy::resource::ResourceSet;
+
+        let set: ResourceSet =
+            serde_json::from_str(r#"["arn:aws:kms:::key/app-*","arn:aws:kms:::alias/myalias"]"#).expect("set should parse");
+        assert_eq!(set.len(), 2);
+
+        let json = serde_json::to_string(&set).expect("set should serialize");
+        assert_eq!(json, r#"["arn:aws:kms:::key/app-*","arn:aws:kms:::alias/myalias"]"#);
+
+        let round_trip: ResourceSet = serde_json::from_str(&json).expect("serialized set should deserialize");
+        assert_eq!(round_trip, set);
+    }
 }

crates/policy/src/policy/statement.rs

@@ -16,6 +16,7 @@ use super::{
     ActionSet, Args, BucketPolicyArgs, Effect, Error as IamError, Functions, ID, Principal, ResourceSet, Validator,
     action::{Action, S3Action},
     function::key_name::{KeyName, S3KeyName},
+    resource::Resource,
     variables::{VariableContext, VariableResolver},
 };
 use crate::error::{Error, Result};
@@ -173,13 +174,79 @@ impl Statement {
         Some(ActionFamily::Mixed)
     }
 
+    /// Resource scope check for KMS statements, which match `arn:aws:kms:::key/<key_id>`
+    /// patterns instead of the bucket/object path grammar used by S3 statements.
+    ///
+    /// Call-site contract (wired up by the admin/SSE authorization paths): the requested
+    /// key identifier (pre-alias-resolution) travels in `args.object` with `args.bucket`
+    /// left empty. An empty `args.object` means the caller did not scope the request to a
+    /// key, which preserves the legacy match-every-key behaviour.
+    async fn kms_key_scope_matches(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
+        let kms_resources: Vec<&Resource> = self.resources.iter().filter(|resource| resource.is_kms()).collect();
+        let kms_not_resources: Vec<&Resource> = self.not_resources.iter().filter(|resource| resource.is_kms()).collect();
+
+        if kms_resources.len() != self.resources.len() || kms_not_resources.len() != self.not_resources.len() {
+            // Statements combining KMS actions with S3 resources predate KMS resource
+            // support and were always evaluated as if unscoped; keep that behaviour
+            // but surface it, since the S3 patterns never constrain key access.
+            tracing::warn!(
+                sid = %self.sid.0,
+                "KMS statement carries non-KMS resources; they are ignored and the statement matches every key"
+            );
+        }
+
+        if kms_resources.is_empty() && kms_not_resources.is_empty() {
+            // No KMS resources: the statement scopes by action only (legacy form).
+            return true;
+        }
+
+        if args.object.is_empty() {
+            // Call sites that do not pass a key resource keep the pre-resource-scoping
+            // behaviour where any key matches.
+            return true;
+        }
+
+        let requested = format!("{}{}", Resource::KMS_KEY_SEGMENT, args.object);
+
+        if !kms_resources.is_empty() {
+            let mut matched = false;
+            for resource in kms_resources {
+                if resource
+                    .is_match_with_resolver(&requested, args.conditions, Some(resolver))
+                    .await
+                {
+                    matched = true;
+                    break;
+                }
+            }
+            if !matched {
+                return false;
+            }
+        }
+
+        for resource in kms_not_resources {
+            if resource
+                .is_match_with_resolver(&requested, args.conditions, Some(resolver))
+                .await
+            {
+                return false;
+            }
+        }
+
+        true
+    }
+
     /// Returns true when this statement would reach `conditions.evaluate_with_resolver` in
-    /// [`Statement::is_allowed`] (including the KMS shortcut path). Does not evaluate conditions.
+    /// [`Statement::is_allowed`] (including the KMS resource path). Does not evaluate conditions.
     pub(crate) async fn request_reaches_condition_eval(&self, args: &Args<'_>, resolver: &VariableResolver) -> bool {
         if (!self.actions.is_match(&args.action) && !self.actions.is_empty()) || self.not_actions.is_match(&args.action) {
             return false;
         }
 
+        if self.is_kms() {
+            return self.kms_key_scope_matches(args, resolver).await;
+        }
+
         let resource = build_resource(
             &args.action,
             args.bucket,
@@ -187,10 +254,6 @@ impl Statement {
             self.conditions.references_key_name(&KeyName::S3(S3KeyName::S3Prefix)),
         );
 
-        if self.is_kms() && (resource == "/" || self.resources.is_empty()) {
-            return true;
-        }
-
         if self.resources.is_empty() && self.not_resources.is_empty() && !self.is_admin() && !self.is_sts() {
             return false;
         }
@@ -273,6 +336,19 @@ impl Validator for Statement {
             return Err(IamError::BothResourceAndNotResource.into());
         }
 
+        // KMS resources only make sense on pure-KMS statements. The reverse
+        // combination (KMS actions with S3 resources) predates KMS resources,
+        // may already be stored, and stays loadable; evaluation treats it as
+        // unscoped and warns.
+        let has_kms_resource = self
+            .resources
+            .iter()
+            .chain(self.not_resources.iter())
+            .any(|resource| resource.is_kms());
+        if has_kms_resource && !matches!(action_family, Some(ActionFamily::Kms)) {
+            return Err(IamError::KmsResourceWithNonKmsAction.into());
+        }
+
         self.actions.is_valid()?;
         self.not_actions.is_valid()?;
         self.resources.is_valid()?;
@@ -323,6 +399,18 @@ pub struct BPStatement {
 impl BPStatement {
     /// Returns true when this statement would reach `conditions.evaluate` in [`BPStatement::is_allowed`].
     pub(crate) async fn request_reaches_condition_eval(&self, args: &BucketPolicyArgs<'_>) -> bool {
+        if !self.actions.is_empty() && self.actions.iter().all(|action| matches!(action, Action::KmsAction(_))) {
+            // Bucket policies cannot grant or deny KMS access; such statements are
+            // rejected at validation but may exist in policies stored before that
+            // check. Skip them so they never influence bucket traffic. Statements
+            // mixing KMS with S3 actions keep evaluating their S3 actions as before.
+            tracing::warn!(
+                sid = %self.sid.0,
+                "ignoring bucket policy statement with KMS actions during evaluation"
+            );
+            return false;
+        }
+
         if !self.principal.is_match(args.account) {
             return false;
         }
@@ -379,6 +467,24 @@ impl Validator for BPStatement {
             return Err(IamError::BothActionAndNotAction.into());
         }
 
+        // Bucket policies govern S3 access; KMS grants belong in identity policies.
+        // Rejected here (PutBucketPolicy) only: deserialization stays permissive so
+        // stored policies from before this check keep loading, and evaluation skips
+        // pure-KMS statements with a warning.
+        let has_kms_action = self
+            .actions
+            .iter()
+            .chain(self.not_actions.iter())
+            .any(|action| matches!(action, Action::KmsAction(_)));
+        let has_kms_resource = self
+            .resources
+            .iter()
+            .chain(self.not_resources.iter())
+            .any(|resource| resource.is_kms());
+        if has_kms_action || has_kms_resource {
+            return Err(IamError::KmsUnsupportedInBucketPolicy.into());
+        }
+
         if self.resources.is_empty() && self.not_resources.is_empty() {
             return Err(IamError::NonResource.into());
         }

crates/policy/tests/policy_eval_proptest.rs

@@ -26,6 +26,12 @@
 //!       policy is also allowed by the widened policy;
 //!   (c) an empty policy denies every non-owner request (default deny).
 //!
+//! Properties (d)-(g) extend the same invariants to KMS key resources
+//! (`arn:aws:kms:::key/<key_id>`, backlog#1582): Deny-first over key scopes,
+//! wildcard/resource-less supersets implying concrete key grants, exact key
+//! scoping without cross-key leaks, and the legacy resource-less
+//! match-every-key compatibility pin.
+//!
 //! Pure evaluation: no IO, no global state, parallel-safe. Statements are built
 //! from JSON exactly like production policies arriving via PutPolicy. Generated
 //! bucket/key/action pools avoid wildcard metacharacters so resource patterns
@@ -47,10 +53,23 @@ const OBJECT_ACTIONS: &[&str] = &[
     "s3:PutObjectTagging",
 ];
 
+/// Key-scoped KMS actions safe to pair with an `arn:aws:kms:::key/<id>` resource.
+const KMS_KEY_ACTIONS: &[&str] = &[
+    "kms:GenerateDataKey",
+    "kms:Decrypt",
+    "kms:DisableKey",
+    "kms:RotateKey",
+    "kms:DescribeKey",
+];
+
 fn statement_json(effect: &str, action: &str, resource: &str) -> String {
     format!(r#"{{"Effect":"{effect}","Action":["{action}"],"Resource":["{resource}"]}}"#)
 }
 
+fn resourceless_statement_json(effect: &str, action: &str) -> String {
+    format!(r#"{{"Effect":"{effect}","Action":["{action}"]}}"#)
+}
+
 fn policy_from_statements(statements: &[String]) -> Policy {
     let json = format!(r#"{{"Version":"2012-10-17","Statement":[{}]}}"#, statements.join(","));
     serde_json::from_str(&json).expect("generated policy JSON should parse")
@@ -88,6 +107,22 @@ fn action_strategy() -> impl Strategy<Value = &'static str> {
     proptest::sample::select(OBJECT_ACTIONS)
 }
 
+/// Strategy: a KMS key id without wildcard metacharacters or separators.
+fn key_id_strategy() -> impl Strategy<Value = String> {
+    "[a-z][a-z0-9-]{2,11}"
+}
+
+/// Strategy: one action name from the key-scoped KMS action pool.
+fn kms_action_strategy() -> impl Strategy<Value = &'static str> {
+    proptest::sample::select(KMS_KEY_ACTIONS)
+}
+
+/// KMS evaluation contract: the requested key id travels in `args.object` with an
+/// empty bucket (see `Statement::kms_key_scope_matches`).
+fn is_allowed_for_key(policy: &Policy, action: &str, key_id: &str) -> bool {
+    is_allowed(policy, action, "", key_id)
+}
+
 proptest! {
     /// (a) Deny anywhere wins: a Deny statement matching the request denies it,
     /// no matter how many broad Allow statements surround it or at which index
@@ -205,4 +240,122 @@ proptest! {
             "Policy::default() must deny {action} on {bucket}/{key}"
         );
     }
+
+    /// (d) KMS Deny anywhere wins: a Deny scoped to the exact key (or `key/*`)
+    /// denies the request no matter how many broad KMS Allow statements
+    /// (resource-less or `key/*`-scoped) surround it.
+    #[test]
+    fn kms_explicit_deny_anywhere_denies(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+        allow_count in 0usize..4,
+        deny_pos_seed in 0usize..16,
+        broad in proptest::bool::ANY,
+        wildcard_deny in proptest::bool::ANY,
+    ) {
+        let mut statements: Vec<String> = (0..allow_count)
+            .map(|_| {
+                if broad {
+                    resourceless_statement_json("Allow", "kms:*")
+                } else {
+                    statement_json("Allow", "kms:*", "arn:aws:kms:::key/*")
+                }
+            })
+            .collect();
+
+        let deny_resource = if wildcard_deny {
+            "arn:aws:kms:::key/*".to_string()
+        } else {
+            format!("arn:aws:kms:::key/{key_id}")
+        };
+        let deny = statement_json("Deny", action, &deny_resource);
+        let deny_pos = deny_pos_seed % (statements.len() + 1);
+        statements.insert(deny_pos, deny);
+
+        let policy = policy_from_statements(&statements);
+
+        if allow_count > 0 {
+            let mut allows_only = statements.clone();
+            allows_only.remove(deny_pos);
+            let allow_policy = policy_from_statements(&allows_only);
+            prop_assert!(
+                is_allowed_for_key(&allow_policy, action, &key_id),
+                "sanity: the KMS Allow statements alone should permit {action} on key {key_id}"
+            );
+        }
+
+        prop_assert!(
+            !is_allowed_for_key(&policy, action, &key_id),
+            "explicit KMS Deny at index {deny_pos} of {} statements must deny {action} on key {key_id}",
+            statements.len()
+        );
+    }
+
+    /// (e) KMS wildcard superset implies the concrete key grant: whatever an
+    /// exact `key/<id>` Allow permits is also permitted by `key/*`, by a bare
+    /// `arn:aws:kms:::*`, and by the legacy resource-less statement form.
+    #[test]
+    fn kms_wildcard_superset_implies_concrete_match(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        let narrow = policy_from_statements(&[statement_json(
+            "Allow",
+            action,
+            &format!("arn:aws:kms:::key/{key_id}"),
+        )]);
+        let widened = policy_from_statements(&[statement_json("Allow", "kms:*", "arn:aws:kms:::key/*")]);
+        let star = policy_from_statements(&[statement_json("Allow", "kms:*", "arn:aws:kms:::*")]);
+        let resourceless = policy_from_statements(&[resourceless_statement_json("Allow", "kms:*")]);
+
+        prop_assert!(is_allowed_for_key(&narrow, action, &key_id), "narrow KMS policy must allow its own grant");
+        prop_assert!(is_allowed_for_key(&widened, action, &key_id), "kms:* on key/* must imply the concrete grant");
+        prop_assert!(is_allowed_for_key(&star, action, &key_id), "kms:* on arn:aws:kms:::* must imply the concrete grant");
+        prop_assert!(
+            is_allowed_for_key(&resourceless, action, &key_id),
+            "the legacy resource-less KMS statement must imply the concrete grant"
+        );
+    }
+
+    /// (f) Key scoping is exact: an Allow on `key/<a>` never leaks to a
+    /// different key id, while the compatibility contract keeps unscoped
+    /// requests (no key id passed) matching.
+    #[test]
+    fn kms_key_scope_does_not_leak_across_keys(
+        key_a in key_id_strategy(),
+        key_b in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        prop_assume!(key_a != key_b);
+
+        let policy = policy_from_statements(&[statement_json(
+            "Allow",
+            action,
+            &format!("arn:aws:kms:::key/{key_a}"),
+        )]);
+
+        prop_assert!(is_allowed_for_key(&policy, action, &key_a), "the granted key must be allowed");
+        prop_assert!(
+            !is_allowed_for_key(&policy, action, &key_b),
+            "an Allow scoped to key {key_a} must not leak to key {key_b}"
+        );
+        prop_assert!(
+            is_allowed_for_key(&policy, action, ""),
+            "call sites that pass no key resource keep the legacy match-every-key behaviour"
+        );
+    }
+
+    /// (g) Legacy compatibility pin: resource-less KMS statements match every
+    /// generated key id, exactly as before KMS resources existed.
+    #[test]
+    fn kms_resourceless_statement_matches_every_key(
+        key_id in key_id_strategy(),
+        action in kms_action_strategy(),
+    ) {
+        let policy = policy_from_statements(&[resourceless_statement_json("Allow", action)]);
+        prop_assert!(
+            is_allowed_for_key(&policy, action, &key_id),
+            "resource-less KMS statement must keep matching {action} on key {key_id}"
+        );
+    }
 }