fix(admin): protect notification target edits (#9048)

* fix(admin): protect notification target edits

Use a durable configuration revision and conditional mutation to preserve concurrent changes and inherited target settings.

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

* fix(admin): use the gateway error boundary

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

* refactor(admin): limit target detail health probes

Reuse normalized source information and probe only runtime targets matching the requested instance.

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

* refactor(admin): streamline target config validation

Preserve duplicate-key order when comparing snapshots, avoid copying secret values, and reuse the validated target configuration.

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

---------

Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>

crates/ecstore/src/api/mod.rs

@@ -385,10 +385,10 @@ pub mod config {
             ServerConfigCorruptError, ServerConfigSaveResult, ServerConfigSnapshot, delete_config, delete_config_no_lock,
             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_config_with_opts_quiet,
-            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,
+            read_server_config_snapshot, read_server_config_with_generation, save_config, save_config_no_lock,
+            save_config_with_opts, save_config_with_opts_quiet, 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

@@ -2617,6 +2617,15 @@ where
     save_server_config_snapshot(api, cfg, &snapshot).await.map(|_| ())
 }
 
+/// Read config and generation from the same object without taking the transaction write lock.
+pub async fn read_server_config_with_generation(api: Arc<ECStore>) -> Result<(Config, Option<Uuid>)> {
+    match read_config_with_metadata_inner(api.clone(), &server_config_path(), &ObjectOptions::default(), false, None).await {
+        Ok((raw, info)) => Ok((decode_persisted_server_config(&raw)?.merge(), info.data_dir.filter(|id| !id.is_nil()))),
+        Err(Error::ConfigNotFound) => read_config_without_migrate(api).await.map(|config| (config, None)),
+        Err(error) => Err(error),
+    }
+}
+
 /// One serialized read of the persisted server config used as the seed and
 /// compare-and-swap fence for an admin update.
 #[derive(Debug)]

crates/notify/src/config_manager.rs

@@ -42,13 +42,14 @@ enum NotifyConfigStoreError {
     Read(String),
     Save(String),
     Converge { persisted: bool, error: String },
+    Conflict,
 }
 
 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}"
-    ))
+    NotificationError::ConfigurationConvergence {
+        persisted,
+        detail: error.to_string(),
+    }
 }
 
 async fn supervise_config_update<T>(
@@ -63,12 +64,62 @@ where
     })?
 }
 
+fn configs_equal(first: Option<&KVS>, second: Option<&KVS>) -> bool {
+    match (first, second) {
+        (None, None) => true,
+        (Some(first), Some(second)) => {
+            if first == second {
+                return true;
+            }
+            let mut first: Vec<_> = first.0.iter().collect();
+            let mut second: Vec<_> = second.0.iter().collect();
+            // Lookup uses the first occurrence of a key, so preserve duplicate-key order.
+            first.sort_by(|a, b| a.key.cmp(&b.key));
+            second.sort_by(|a, b| a.key.cmp(&b.key));
+            first == second
+        }
+        _ => false,
+    }
+}
+
+fn replace_target_if_unchanged(
+    config: &mut Config,
+    target_type: &str,
+    target_name: &str,
+    expected: Option<&KVS>,
+    expected_default: Option<&KVS>,
+    kvs: &KVS,
+) -> Result<bool, NotifyConfigStoreError> {
+    let targets = config.0.get(target_type);
+    let mut matching = targets
+        .into_iter()
+        .flat_map(|targets| targets.iter())
+        .filter(|(name, _)| name.eq_ignore_ascii_case(target_name));
+    let current = matching.next();
+    if matching.next().is_some() {
+        return Err(NotifyConfigStoreError::Conflict);
+    }
+    let stored_name = current
+        .map(|(name, _)| name.clone())
+        .unwrap_or_else(|| target_name.to_owned());
+    let current_default = targets.and_then(|targets| targets.get(rustfs_config::DEFAULT_DELIMITER));
+    if !configs_equal(current.map(|(_, kvs)| kvs), expected) || !configs_equal(current_default, expected_default) {
+        return Err(NotifyConfigStoreError::Conflict);
+    }
+    config
+        .0
+        .entry(target_type.to_owned())
+        .or_default()
+        .insert(stored_name, kvs.clone());
+    Ok(true)
+}
+
 async fn update_server_config<F>(
     mut modifier: F,
     lifecycle: NotifyLifecycleCoordinator,
 ) -> Result<Option<(crate::lifecycle::NotificationLifecycleTransition, bool)>, NotifyConfigStoreError>
 where
-    F: FnMut(&mut Config) -> bool + Send + 'static,
+    F: FnMut(&mut Config, Option<String>) -> Result<bool, NotifyConfigStoreError> + Send + 'static,
 {
     let Some(store) = resolve_notify_object_store_handle() else {
         return Err(NotifyConfigStoreError::StorageNotAvailable);
@@ -79,7 +130,8 @@ where
             .await
             .map_err(NotifyConfigStoreError::Read)?;
         let mut config = snapshot.config.clone();
-        if !modifier(&mut config) {
+        let generation = snapshot.generation().map(|value| value.to_string());
+        if !modifier(&mut config, generation)? {
             return Ok(None);
         }
 
@@ -245,6 +297,30 @@ impl NotifyConfigManager {
         .await
     }
 
+    /// Replace a target only when its persisted configuration is unchanged since validation.
+    pub async fn set_target_config_if_unchanged(
+        &self,
+        target_type: &str,
+        target_name: &str,
+        expected: Option<KVS>,
+        expected_default: Option<KVS>,
+        expected_generation: Option<String>,
+        kvs: KVS,
+    ) -> Result<(), NotificationError> {
+        let ttype = target_type.to_lowercase();
+        let tname = target_name.to_lowercase();
+        self.update_config_and_reload_checked(move |config, generation| {
+            if expected_generation
+                .as_ref()
+                .is_some_and(|expected| generation.as_ref() != Some(expected))
+            {
+                return Err(NotifyConfigStoreError::Conflict);
+            }
+            replace_target_if_unchanged(config, &ttype, &tname, expected.as_ref(), expected_default.as_ref(), &kvs)
+        })
+        .await
+    }
+
     pub async fn remove_target_config(&self, target_type: &str, target_name: &str) -> Result<(), NotificationError> {
         debug!(
             event = EVENT_NOTIFY_CONFIG_UPDATE,
@@ -338,9 +414,17 @@ impl NotifyConfigManager {
         Ok(())
     }
 
-    async fn update_config_and_reload<F>(&self, modifier: F) -> Result<(), NotificationError>
+    async fn update_config_and_reload<F>(&self, mut modifier: F) -> Result<(), NotificationError>
     where
         F: FnMut(&mut Config) -> bool + Send + 'static,
+    {
+        self.update_config_and_reload_checked(move |config, _generation| Ok(modifier(config)))
+            .await
+    }
+
+    async fn update_config_and_reload_checked<F>(&self, modifier: F) -> Result<(), NotificationError>
+    where
+        F: FnMut(&mut Config, Option<String>) -> Result<bool, NotifyConfigStoreError> + Send + 'static,
     {
         if self.lifecycle.state() == NotificationRuntimeState::Terminated {
             return Err(NotificationError::Initialization("Notification runtime has terminated".to_string()));
@@ -356,6 +440,7 @@ impl NotifyConfigManager {
                     NotifyConfigStoreError::Read(err) => NotificationError::ReadConfig(err),
                     NotifyConfigStoreError::Save(err) => NotificationError::SaveConfig(err),
                     NotifyConfigStoreError::Converge { persisted, error } => notification_convergence_error(persisted, error),
+                    NotifyConfigStoreError::Conflict => NotificationError::ConfigurationConflict,
                 })?
         else {
             debug!(
@@ -394,6 +479,126 @@ mod tests {
     use crate::rules::RulesMap;
     use crate::{NotificationError, NotificationRuntimeState};
 
+    #[test]
+    fn config_comparison_preserves_duplicates_and_hidden_flags() {
+        use rustfs_config::server_config::{KV, KVS};
+        let field = |key: &str, value: &str, hidden_if_empty: bool| KV {
+            key: key.into(),
+            value: value.into(),
+            hidden_if_empty,
+        };
+        let original = KVS(vec![
+            field("password", "secret", false),
+            field("token", "", true),
+            field("password", "secret", false),
+        ]);
+        let reordered = KVS(vec![
+            field("password", "secret", false),
+            field("password", "secret", false),
+            field("token", "", true),
+        ]);
+        let missing_duplicate = KVS(vec![field("password", "secret", false), field("token", "", true)]);
+        let changed_flag = KVS(vec![
+            field("password", "secret", false),
+            field("password", "secret", false),
+            field("token", "", false),
+        ]);
+        assert!(super::configs_equal(Some(&original), Some(&reordered)));
+        assert!(!super::configs_equal(Some(&original), Some(&missing_duplicate)));
+        assert!(!super::configs_equal(Some(&original), Some(&changed_flag)));
+        assert!(!super::configs_equal(Some(&original), None));
+        let first_value = KVS(vec![
+            field("password", "old", false),
+            field("token", "", true),
+            field("password", "new", false),
+        ]);
+        let reordered_keys = KVS(vec![
+            field("token", "", true),
+            field("password", "old", false),
+            field("password", "new", false),
+        ]);
+        let reordered_duplicates = KVS(vec![
+            field("token", "", true),
+            field("password", "new", false),
+            field("password", "old", false),
+        ]);
+        assert!(super::configs_equal(Some(&first_value), Some(&reordered_keys)));
+        assert!(!super::configs_equal(Some(&first_value), Some(&reordered_duplicates)));
+    }
+
+    #[test]
+    fn conditional_target_update_rejects_rotation_and_deletion() {
+        use rustfs_config::server_config::{Config, KVS};
+        let mut old = KVS::new();
+        old.insert("password".into(), "old".into());
+        let mut rotated = KVS::new();
+        rotated.insert("password".into(), "new".into());
+        let mut config = Config::new();
+        config
+            .0
+            .entry("notify_nats".into())
+            .or_default()
+            .insert("primary".into(), rotated.clone());
+        let result = super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &old);
+        assert!(matches!(result, Err(super::NotifyConfigStoreError::Conflict)));
+        assert_eq!(config.0["notify_nats"]["primary"], rotated);
+        config.0.get_mut("notify_nats").expect("subsystem exists").remove("primary");
+        assert!(matches!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &old),
+            Err(super::NotifyConfigStoreError::Conflict)
+        ));
+        assert!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", None, None, &old)
+                .expect("create unchanged target")
+        );
+        assert!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), None, &rotated)
+                .expect("replace unchanged target")
+        );
+        assert_eq!(config.0["notify_nats"]["primary"], rotated);
+
+        let mut first = KVS::new();
+        first.insert("url".into(), "nats://example.com".into());
+        first.insert("token".into(), "secret".into());
+        let mut reordered = KVS::new();
+        reordered.insert("token".into(), "secret".into());
+        reordered.insert("url".into(), "nats://example.com".into());
+        config
+            .0
+            .get_mut("notify_nats")
+            .expect("subsystem")
+            .insert("primary".into(), first);
+        assert!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&reordered), None, &rotated)
+                .expect("key order does not signal a change")
+        );
+        let targets = config.0.get_mut("notify_nats").expect("subsystem");
+        targets.remove("primary");
+        targets.insert("Primary".into(), rotated.clone());
+        assert!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&rotated), None, &old)
+                .expect("case-insensitive match")
+        );
+        let targets = &config.0["notify_nats"];
+        assert!(targets.contains_key("Primary"));
+        assert!(!targets.contains_key("primary"));
+        let mut old_default = KVS::new();
+        old_default.insert("password".into(), "old-default".into());
+        let mut new_default = KVS::new();
+        new_default.insert("password".into(), "new-default".into());
+        config
+            .0
+            .get_mut("notify_nats")
+            .expect("subsystem")
+            .insert("_".into(), new_default.clone());
+        assert!(matches!(
+            super::replace_target_if_unchanged(&mut config, "notify_nats", "primary", Some(&old), Some(&old_default), &rotated),
+            Err(super::NotifyConfigStoreError::Conflict)
+        ));
+        assert_eq!(config.0["notify_nats"]["Primary"], old);
+        assert_eq!(config.0["notify_nats"]["_"], new_default);
+    }
+
     #[tokio::test]
     async fn supervised_config_update_survives_waiter_cancellation() {
         let (started_tx, started_rx) = tokio::sync::oneshot::channel();
@@ -550,13 +755,9 @@ mod tests {
 
     #[test]
     fn convergence_error_reports_durable_write_state() {
-        for (persisted, expected) in [(true, "configuration was persisted"), (false, "configuration was unchanged")] {
+        for persisted in [true, false] {
             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"));
+            assert!(matches!(error, NotificationError::ConfigurationConvergence { persisted: value, .. } if value == persisted));
         }
     }
 }

crates/notify/src/error.rs

@@ -38,6 +38,12 @@ pub enum NotificationError {
     #[error("Configuration error: {0}")]
     Configuration(String),
 
+    #[error("Target configuration changed during validation")]
+    ConfigurationConflict,
+
+    #[error("Notification configuration convergence failed (persisted: {persisted}): {detail}")]
+    ConfigurationConvergence { persisted: bool, detail: String },
+
     #[error("ARN not found: {0}")]
     ARNNotFound(String),
 

crates/notify/src/integration.rs

@@ -298,6 +298,22 @@ impl NotificationSystem {
             .await
     }
 
+    /// Replace a target only if its persisted configuration still matches the validated snapshot.
+    pub async fn set_target_config_if_unchanged(
+        &self,
+        target_type: &str,
+        target_name: &str,
+        expected: Option<KVS>,
+        expected_default: Option<KVS>,
+        expected_generation: Option<String>,
+        kvs: KVS,
+    ) -> Result<(), NotificationError> {
+        self.services
+            .config_manager
+            .set_target_config_if_unchanged(target_type, target_name, expected, expected_default, expected_generation, kvs)
+            .await
+    }
+
     /// Removes all notification configurations for a bucket.
     /// If the configuration is successfully removed, the entire notification system will be automatically reloaded.
     ///

rustfs/Cargo.toml

@@ -392,6 +392,7 @@ object = { workspace = true, optional = true, features = ["read", "std", "elf"]
 symbolic-demangle = { version = "=13.9.0", default-features = false, features = ["rust"], optional = true }
 
 [dev-dependencies]
+rustfs-targets = { workspace = true, features = ["test-support"] }
 uuid = { workspace = true, features = ["v4", "v5", "fast-rng", "macro-diagnostics"] }
 rustfs-iam = { workspace = true, features = ["test-util"] }
 serial_test = { workspace = true }

rustfs/src/admin/handlers/event.rs

@@ -50,7 +50,7 @@ use tracing::{error, info, warn};
 
 const LOG_COMPONENT_ADMIN_API: &str = "admin_api";
 
-async fn converge_target_mutation_on_cluster(
+pub(super) async fn converge_target_mutation_on_cluster(
     context: Option<&AppContext>,
     target_type: &str,
     local_result: Result<(), rustfs_notify::NotificationError>,
@@ -60,18 +60,26 @@ async fn converge_target_mutation_on_cluster(
 
     match (local_result, peer_result) {
         (Ok(()), Ok(())) => Ok(()),
-        (Err(local), Ok(())) => {
-            warn!(target_type, error = %local, "Local notification runtime failed to converge");
-            Err(s3_error!(InternalError, "local notification runtime failed to converge"))
-        }
-        (Ok(()), Err(peer)) => Err(peer),
-        (Err(local), Err(peer)) => {
-            warn!(target_type, error = %local, "Local notification runtime failed while peer convergence also failed");
-            Err(s3_error!(
-                InternalError,
-                "local notification runtime and peer convergence failed: {}",
-                peer
-            ))
+        (Ok(()), Err(_)) => Err(s3_error!(
+            InternalError,
+            "notification target update completed locally but peer convergence failed"
+        )),
+        (Err(local), peer) => {
+            let peer_failed = peer.is_err();
+            let message = match local {
+                rustfs_notify::NotificationError::ConfigurationConvergence { persisted: true, .. } => {
+                    "notification target configuration was persisted but local runtime convergence failed"
+                }
+                rustfs_notify::NotificationError::ConfigurationConvergence { persisted: false, .. } => {
+                    "notification target configuration was unchanged but local runtime convergence failed"
+                }
+                _ => "notification target update failed; persistence status is unknown",
+            };
+            if peer_failed {
+                Err(s3_error!(InternalError, "{}; peer convergence also failed", message))
+            } else {
+                Err(s3_error!(InternalError, "{}", message))
+            }
         }
     }
 }
@@ -84,7 +92,7 @@ fn notification_target_subsystem(target_type: &str) -> S3Result<&'static str> {
         .ok_or_else(|| s3_error!(InvalidArgument, "unsupported notification target type: {}", target_type))
 }
 
-async fn preflight_target_mutation_on_cluster(context: Option<&AppContext>, target_type: &str) -> S3Result<()> {
+pub(super) async fn preflight_target_mutation_on_cluster(context: Option<&AppContext>, target_type: &str) -> S3Result<()> {
     preflight_dynamic_config_reload_for_context(context, notification_target_subsystem(target_type)?).await
 }
 
@@ -682,7 +690,7 @@ mod tests {
         use crate::admin::handlers::{
             audit::ListAuditTargets,
             extensions::ListExtensionInstancesHandler,
-            plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler},
+            plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler, PutPluginInstanceHandler},
             target_read::load_target_read_snapshot,
         };
         use crate::admin::storage_api::config::{
@@ -758,6 +766,154 @@ mod tests {
                 .await
                 .expect("persist target without loading it into the runtime");
 
+            // A concurrent credential rotation must be observed under the commit transaction.
+            let initial = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
+                .await
+                .expect("read initial notify config");
+            let generation_before = {
+                let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
+                    .await
+                    .expect("read initial generation");
+                snapshot.generation().expect("test config has an object generation")
+            };
+            let expected = initial.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
+            let expected_default = initial.0[NOTIFY_WEBHOOK_SUB_SYS]
+                .get(rustfs_config::DEFAULT_DELIMITER)
+                .cloned();
+            let mut rotated = initial.clone();
+            rotated
+                .0
+                .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+                .expect("subsystem")
+                .get_mut("unavailable")
+                .expect("target")
+                .insert(rustfs_config::WEBHOOK_AUTH_TOKEN.to_owned(), "rotated-secret".to_owned());
+            save_admin_server_config(env.ecstore.clone(), &rotated)
+                .await
+                .expect("persist concurrent rotation");
+            let generation_after = {
+                let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
+                    .await
+                    .expect("read rotated generation");
+                snapshot.generation().expect("rotated config has an object generation")
+            };
+            assert_ne!(generation_before, generation_after, "content change must publish a new generation");
+            let system = rustfs_notify::NotificationSystem::new(Config::default());
+            let result = system
+                .set_target_config_if_unchanged(
+                    NOTIFY_WEBHOOK_SUB_SYS,
+                    "unavailable",
+                    Some(expected.clone()),
+                    expected_default,
+                    None,
+                    expected,
+                )
+                .await;
+            assert!(matches!(result, Err(rustfs_notify::NotificationError::ConfigurationConflict)));
+            let actual = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
+                .await
+                .expect("read rotated config");
+            assert_eq!(
+                actual.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"]
+                    .lookup(rustfs_config::WEBHOOK_AUTH_TOKEN)
+                    .as_deref(),
+                Some("rotated-secret")
+            );
+            save_admin_server_config(env.ecstore.clone(), &initial)
+                .await
+                .expect("restore inventory config");
+            let generation = {
+                let snapshot = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
+                    .await
+                    .expect("read revision baseline");
+                snapshot.generation().expect("baseline generation").to_string()
+            };
+            let mut unrelated = initial.clone();
+            let mut other = KVS::new();
+            other.insert("enable".into(), "off".into());
+            other.insert("endpoint".into(), "http://127.0.0.1:9/other".into());
+            unrelated
+                .0
+                .entry(NOTIFY_WEBHOOK_SUB_SYS.to_owned())
+                .or_default()
+                .insert("other".to_owned(), other);
+            save_admin_server_config(env.ecstore.clone(), &unrelated)
+                .await
+                .expect("persist unrelated update");
+            let target = initial.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
+            let default = initial.0[NOTIFY_WEBHOOK_SUB_SYS]
+                .get(rustfs_config::DEFAULT_DELIMITER)
+                .cloned();
+            let result = system
+                .set_target_config_if_unchanged(
+                    NOTIFY_WEBHOOK_SUB_SYS,
+                    "unavailable",
+                    Some(target.clone()),
+                    default,
+                    Some(generation),
+                    target,
+                )
+                .await;
+            assert!(
+                matches!(result, Err(rustfs_notify::NotificationError::ConfigurationConflict)),
+                "revision changes must reject even when target remains unchanged"
+            );
+            save_admin_server_config(env.ecstore.clone(), &initial)
+                .await
+                .expect("restore after unrelated update");
+
+            let mut with_default = initial.clone();
+            let mut default = KVS::new();
+            default.insert("queue_limit".into(), "15".into());
+            with_default
+                .0
+                .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+                .expect("subsystem")
+                .insert("_".into(), default.clone());
+            save_admin_server_config(env.ecstore.clone(), &with_default)
+                .await
+                .expect("persist inherited default");
+            let saved = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
+                .await
+                .expect("read inherited config");
+            let original = saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
+            let old_default = saved.0[NOTIFY_WEBHOOK_SUB_SYS].get("_").cloned();
+            let expected = std::collections::HashMap::from([
+                ("endpoint".to_owned(), "http://127.0.0.1:9/hook".to_owned()),
+                ("queue_limit".to_owned(), "15".to_owned()),
+            ]);
+            let mut validated = original.clone();
+            validated.insert("endpoint".into(), "http://127.0.0.1:9/changed".into());
+            validated.insert("queue_limit".into(), "15".into());
+            let update = crate::admin::handlers::plugins_instances::merge_conditional_target_update(
+                &original,
+                &expected,
+                &["endpoint".into()],
+                &validated,
+            )
+            .expect("merge only edited field");
+            let result = system
+                .set_target_config_if_unchanged(NOTIFY_WEBHOOK_SUB_SYS, "unavailable", Some(original), old_default, None, update)
+                .await;
+            assert!(
+                result.is_ok()
+                    || matches!(
+                        result,
+                        Err(rustfs_notify::NotificationError::ConfigurationConvergence { persisted: true, .. })
+                    )
+            );
+            let saved = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
+                .await
+                .expect("read conditional update");
+            assert_eq!(
+                saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].lookup("endpoint").as_deref(),
+                Some("http://127.0.0.1:9/changed")
+            );
+            assert_eq!(saved.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].lookup("queue_limit"), None);
+            assert_eq!(saved.0[NOTIFY_WEBHOOK_SUB_SYS]["_"].lookup("queue_limit").as_deref(), Some("15"));
+            save_admin_server_config(env.ecstore.clone(), &initial)
+                .await
+                .expect("restore inventory config");
             let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
             let (release_tx, release_rx) = tokio::sync::oneshot::channel();
             let store = env.ecstore.clone();
@@ -785,6 +941,53 @@ mod tests {
             assert!(observed_config.0[NOTIFY_WEBHOOK_SUB_SYS].contains_key("unavailable"));
             mark_event_notifier_unreconciled();
 
+            // Conditional edits reject stale revisions before target validation or persistence.
+            let mut router = Router::new();
+            router.insert("/{id}", ()).expect("plugin detail route");
+            let matched = router.at("/builtin:webhook:notify:unavailable").expect("plugin params");
+            let transaction_guard = crate::admin::storage_api::config::read_admin_server_config_snapshot(env.ecstore.clone())
+                .await
+                .expect("hold config transaction lock");
+            let generation = transaction_guard.generation().expect("config generation").to_string();
+            let detail = response_json(GetPluginInstanceHandler {}.call(
+                request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable"),
+                matched.params,
+            ))
+            .await
+            .expect("read conditional edit baseline");
+            let revision = detail["config_revision"].as_str().expect("GET includes opaque revision");
+            assert_eq!(
+                revision,
+                crate::admin::handlers::plugins_instances::config_revision("builtin:webhook:notify:unavailable", &generation)
+            );
+            assert_eq!(detail["config"]["auth_token"], "***redacted***");
+            drop(transaction_guard);
+            let mut put = request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable");
+            put.method = Method::PUT;
+            put.input = Body::from(
+                serde_json::to_vec(&serde_json::json!({
+                    "expected_config": detail["config"],
+                    "expected_revision": "stale",
+                    "key_values": [{"key":"auth_token", "value":"replacement-secret"}]
+                }))
+                .expect("encode conditional edit"),
+            );
+            let matched = router.at("/builtin:webhook:notify:unavailable").expect("plugin params");
+            let err = PutPluginInstanceHandler {}
+                .call(put, matched.params)
+                .await
+                .expect_err("stale revision must be rejected");
+            assert_eq!(err.code(), &s3s::S3ErrorCode::InvalidRequest);
+            let after = crate::admin::storage_api::config::read_admin_config_without_migrate(env.ecstore.clone())
+                .await
+                .expect("read unchanged config");
+            assert_eq!(
+                after.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"]
+                    .lookup(rustfs_config::WEBHOOK_AUTH_TOKEN)
+                    .as_deref(),
+                Some("inventory-secret-token")
+            );
+
             with_notify_runtime_reconcile_lock(async {
                 for initialize_runtime in [false, true] {
                     if initialize_runtime {

rustfs/src/admin/handlers/plugins_instances.rs

@@ -15,23 +15,29 @@
 use crate::admin::{
     auth::authorize_admin_request,
     handlers::audit_runtime_config::{remove_audit_target_config, set_audit_target_config},
+    handlers::event::{converge_target_mutation_on_cluster, preflight_target_mutation_on_cluster},
     handlers::notify_runtime_access::{
-        load_notification_mutation_snapshot, remove_notification_target_config, set_notification_target_config,
+        get_notification_system, load_notification_mutation_snapshot, remove_notification_target_config,
+        set_notification_target_config,
     },
+    handlers::supervise_admin_mutation,
     handlers::target_descriptor::{
         AdminTargetSpec, RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin,
-        build_enabled_target_kvs, build_json_response, collect_target_instances, find_target_instance,
-        target_module_disabled_reason, target_mutation_block_reason as shared_target_mutation_block_reason,
+        allowed_target_keys, build_enabled_target_kvs, build_json_response, build_validated_target_kvs, collect_target_instances,
+        collect_validated_key_values, find_target_instance, target_module_disabled_reason,
+        target_mutation_block_reason as shared_target_mutation_block_reason,
     },
     handlers::target_read::{
-        collect_audit_runtime_statuses, collect_notification_runtime_statuses, load_server_config_from_store,
-        load_target_read_snapshot,
+        collect_audit_runtime_statuses, collect_notification_instance_runtime_statuses, collect_notification_runtime_statuses,
+        load_server_config_from_store, load_target_read_snapshot,
     },
     plugin_contract::{
         PluginContractDomain, PluginInstanceDetail, PluginInstanceDiagnostic, PluginInstanceDiagnosticCode,
         PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, PluginInstancesResponse,
     },
     router::{AdminOperation, Operation, S3Router},
+    runtime_sources::{app_context_from_req, current_object_store_handle_for_context},
+    storage_api::config::{read_admin_server_config_snapshot, read_admin_server_config_with_generation},
     storage_api::s3::{self, S3ErrorCode},
 };
 use crate::server::{
@@ -48,6 +54,7 @@ use rustfs_policy::policy::action::{Action, AdminAction};
 use rustfs_targets::catalog::builtin::{builtin_audit_target_admin_descriptors, builtin_notify_target_admin_descriptors};
 use rustfs_targets::{builtin_target_plugin_operational_state, runtime_state_from_status_label};
 use s3s::{Body, S3Request, S3Response, S3Result, s3_error};
+use sha2::{Digest, Sha256};
 use std::collections::{BTreeMap, HashMap};
 use std::sync::LazyLock;
 use tracing::warn;
@@ -281,6 +288,7 @@ fn plugin_instance_detail(instance: TargetInstanceReadModel) -> PluginInstanceDe
     mapped.diagnostic_codes = diagnostics.iter().map(|item| item.code.clone()).collect();
 
     PluginInstanceDetail {
+        config_revision: None,
         instance: mapped,
         diagnostics,
     }
@@ -320,6 +328,10 @@ struct KeyValue {
 #[serde(deny_unknown_fields)]
 struct PluginInstanceBody {
     key_values: Vec<KeyValue>,
+    #[serde(default)]
+    expected_config: Option<HashMap<String, String>>,
+    #[serde(default)]
+    expected_revision: Option<String>,
 }
 
 #[derive(Debug, Default, Clone, PartialEq, Eq)]
@@ -683,6 +695,106 @@ async fn find_plugin_instance(instance_id: &str) -> S3Result<Option<TargetInstan
     find_target_instance(context.specs, context.route_prefix, &config, runtime_statuses, instance_id)
 }
 
+fn effective_validation_fields(
+    context: PluginInstanceDomainContext,
+    subsystem: &str,
+    effective: &KVS,
+    requested: &[KeyValue],
+) -> S3Result<HashMap<String, String>> {
+    let allowed = allowed_target_keys(context.specs, subsystem);
+    let requested = collect_validated_key_values(
+        requested.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
+        &allowed,
+        subsystem,
+        "plugin instance",
+    )?;
+    let mut candidate: HashMap<String, String> = effective
+        .0
+        .iter()
+        .filter(|kv| allowed.contains(kv.key.as_str()))
+        .map(|kv| (kv.key.clone(), kv.value.clone()))
+        .collect();
+    candidate.extend(requested);
+    Ok(candidate)
+}
+
+pub(super) fn merge_conditional_target_update(
+    original: &KVS,
+    expected: &HashMap<String, String>,
+    changed_keys: &[String],
+    validated: &KVS,
+) -> S3Result<KVS> {
+    let mut result = original.clone();
+    for key in changed_keys {
+        if key == rustfs_config::ENABLE_KEY {
+            return Err(s3::error(
+                S3ErrorCode::InvalidRequest,
+                "target enable state cannot be changed using expected_config",
+            ));
+        }
+        let value = validated
+            .lookup(key)
+            .ok_or_else(|| s3::error(S3ErrorCode::InvalidArgument, "missing validated target field"))?;
+        if expected.get(key) != Some(&value) {
+            result.insert(key.clone(), value);
+        }
+    }
+    Ok(result)
+}
+
+pub(super) fn config_revision(instance_id: &str, generation: &str) -> String {
+    let mut hasher = Sha256::new();
+    hasher.update(b"rustfs-plugin-instance-config-revision-v1\0");
+    hasher.update(generation.as_bytes());
+    hasher.update(b"\0");
+    hasher.update(instance_id.as_bytes());
+    faster_hex::hex_string(&hasher.finalize())
+}
+
+fn validate_expected_config(
+    context: PluginInstanceDomainContext,
+    config: &Config,
+    instance_id: &str,
+    expected: &HashMap<String, String>,
+) -> S3Result<TargetInstanceReadModel> {
+    if context.domain != PluginContractDomain::Notify {
+        return Err(s3::error(
+            S3ErrorCode::InvalidArgument,
+            "expected_config is only supported for notification targets",
+        ));
+    }
+    let resolved = resolve_plugin_instance_target(instance_id)?;
+    let instance = find_target_instance(
+        std::slice::from_ref(&resolved.target_spec),
+        context.route_prefix,
+        config,
+        HashMap::new(),
+        instance_id,
+    )?
+    .filter(|instance| instance.source == TargetEndpointSource::Config)
+    .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "expected_config requires an existing config-managed target"))?;
+    if instance.has_env_default {
+        return Err(s3::error(
+            S3ErrorCode::InvalidRequest,
+            "notification target inherits environment configuration and cannot be edited",
+        ));
+    }
+    if !instance.enabled {
+        return Err(s3::error(
+            S3ErrorCode::InvalidRequest,
+            "disabled targets cannot be edited with expected_config",
+        ));
+    }
+    if &map_instance_config(instance.config.clone(), secret_fields_for_instance(context.domain, &instance.subsystem)) != expected
+    {
+        return Err(s3::error(
+            S3ErrorCode::InvalidRequest,
+            "target configuration changed since it was read; refresh and retry",
+        ));
+    }
+    Ok(instance)
+}
+
 async fn set_plugin_instance_config(
     context: PluginInstanceDomainContext,
     resolved: &ResolvedPluginInstanceTarget,
@@ -744,12 +856,31 @@ impl Operation for GetPluginInstanceHandler {
             .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?;
 
         refresh_persisted_module_switches().await?;
-        let instance = find_plugin_instance(instance_id)
-            .await?
-            .ok_or_else(|| s3_error!(NoSuchKey, "plugin instance not found"))?;
-
-        let data = serde_json::to_vec(&plugin_instance_detail(instance))
-            .map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?;
+        let domain = parse_plugin_instance_id(instance_id)?.1;
+        let (instance, generation) = if domain == PluginContractDomain::Notify {
+            let app_context = app_context_from_req(&req);
+            let context = plugin_instance_domain_context(domain);
+            let read_config = async {
+                match current_object_store_handle_for_context(app_context.as_deref()) {
+                    Some(store) => read_admin_server_config_with_generation(store)
+                        .await
+                        .map_err(|_| s3::error(S3ErrorCode::InternalError, "failed to read server config")),
+                    None => Ok((Config::new(), None)),
+                }
+            };
+            let ((config, generation), runtime) = tokio::try_join!(read_config, async {
+                Ok::<_, s3s::S3Error>(collect_notification_instance_runtime_statuses(context.specs, instance_id).await)
+            })?;
+            let instance = find_target_instance(context.specs, context.route_prefix, &config, runtime, instance_id)?;
+            (instance, generation.map(|value| value.to_string()))
+        } else {
+            (find_plugin_instance(instance_id).await?, None)
+        };
+        let instance = instance.ok_or_else(|| s3::error(S3ErrorCode::NoSuchKey, "plugin instance not found"))?;
+        let mut detail = plugin_instance_detail(instance);
+        detail.config_revision = generation.map(|value| config_revision(instance_id, &value));
+        let data = serde_json::to_vec(&detail)
+            .map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to serialize response: {}", e)))?;
         Ok(build_json_response(StatusCode::OK, Body::from(data), req.headers.get("x-request-id")))
     }
 }
@@ -760,6 +891,7 @@ pub struct PutPluginInstanceHandler {}
 impl Operation for PutPluginInstanceHandler {
     async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
         authorize_plugin_instance_write_request(&req).await?;
+        let app_context = app_context_from_req(&req);
         let instance_id = params
             .get("id")
             .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?;
@@ -780,7 +912,16 @@ impl Operation for PutPluginInstanceHandler {
         let mut body: PluginInstanceBody = serde_json::from_slice(&body_bytes)
             .map_err(|e| s3_error!(InvalidArgument, "invalid json body for plugin instance config: {}", e))?;
 
-        let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?;
+        let (config_snapshot, generation) = if context.domain == PluginContractDomain::Notify {
+            let store = current_object_store_handle_for_context(app_context.as_deref())
+                .ok_or_else(|| s3::error(S3ErrorCode::InternalError, "server storage not initialized"))?;
+            let snapshot = read_admin_server_config_snapshot(store)
+                .await
+                .map_err(|_| s3::error(S3ErrorCode::InternalError, "failed to read server config"))?;
+            (snapshot.config.clone(), snapshot.generation().map(|value| value.to_string()))
+        } else {
+            (plugin_instance_mutation_config_snapshot(context).await?, None)
+        };
         if let Some(reason) = plugin_instance_mutation_block_reason(
             context,
             &config_snapshot,
@@ -791,25 +932,104 @@ impl Operation for PutPluginInstanceHandler {
             return Err(s3_error!(InvalidRequest, "{reason}"));
         }
 
+        if body.expected_revision.is_some() && body.expected_config.is_none() {
+            return Err(s3::error(S3ErrorCode::InvalidArgument, "expected_revision requires expected_config"));
+        }
+        let effective_instance = if let Some(expected) = body.expected_config.as_ref() {
+            let revision = generation
+                .as_ref()
+                .ok_or_else(|| s3::error(S3ErrorCode::InvalidRequest, "target configuration revision unavailable"))?;
+            if body.expected_revision.as_deref() != Some(config_revision(instance_id, revision).as_str()) {
+                return Err(s3::error(
+                    S3ErrorCode::InvalidRequest,
+                    "target configuration changed since it was read; refresh and retry",
+                ));
+            }
+            Some(validate_expected_config(context, &config_snapshot, instance_id, expected)?.config)
+        } else {
+            None
+        };
+
+        let changed_keys: Vec<String> = body
+            .expected_config
+            .as_ref()
+            .map(|expected| {
+                body.key_values
+                    .iter()
+                    .filter(|kv| expected.get(&kv.key) != Some(&kv.value))
+                    .map(|kv| kv.key.clone())
+                    .collect()
+            })
+            .unwrap_or_default();
+
         // GET responses redact secret values; a read-modify-write round trip
         // sends the placeholder back, which must restore the stored secret
         // instead of persisting the placeholder literal.
-        restore_redacted_secret_values(
-            &mut body.key_values,
-            resolved.target_spec.secret_fields,
-            stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name),
-        )?;
+        let stored_secrets = if body.expected_config.is_some() && context.domain == PluginContractDomain::Notify {
+            effective_instance.as_ref()
+        } else {
+            stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name)
+        };
+        restore_redacted_secret_values(&mut body.key_values, resolved.target_spec.secret_fields, stored_secrets)?;
 
-        let kvs = build_enabled_target_kvs(
-            context.specs,
-            body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
-            resolved.target_spec.subsystem,
-            context.default_queue_dir,
-            "plugin instance",
-        )
-        .await?;
+        let validated_kvs = if let Some(effective) = effective_instance.as_ref() {
+            let candidate = effective_validation_fields(context, resolved.target_spec.subsystem, effective, &body.key_values)?;
+            build_validated_target_kvs(&resolved.target_spec, candidate, context.default_queue_dir).await?
+        } else {
+            build_enabled_target_kvs(
+                context.specs,
+                body.key_values.iter().map(|kv| (kv.key.as_str(), kv.value.as_str())),
+                resolved.target_spec.subsystem,
+                context.default_queue_dir,
+                "plugin instance",
+            )
+            .await?
+        };
 
-        set_plugin_instance_config(context, &resolved, kvs).await?;
+        let kvs = if let Some(expected_config) = body.expected_config.as_ref() {
+            let original = stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name)
+                .ok_or_else(|| {
+                    s3::error(S3ErrorCode::InvalidRequest, "expected_config requires an existing config-managed target")
+                })?;
+            merge_conditional_target_update(original, expected_config, &changed_keys, &validated_kvs)?
+        } else {
+            validated_kvs
+        };
+
+        if context.domain == PluginContractDomain::Notify {
+            let expected =
+                stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, &resolved.target_name).cloned();
+            let expected_default =
+                stored_instance_config(&config_snapshot, resolved.target_spec.subsystem, rustfs_config::DEFAULT_DELIMITER)
+                    .cloned();
+            let expected_generation = if body.expected_config.is_some() { generation } else { None };
+            let system = get_notification_system().await?;
+            let subsystem = resolved.target_spec.subsystem.to_owned();
+            let target_name = resolved.target_name.clone();
+            supervise_admin_mutation("plugin notification target mutation", async move {
+                preflight_target_mutation_on_cluster(app_context.as_deref(), &subsystem).await?;
+                let local = system
+                    .set_target_config_if_unchanged(
+                        &subsystem,
+                        &target_name,
+                        expected,
+                        expected_default,
+                        expected_generation,
+                        kvs,
+                    )
+                    .await;
+                if matches!(local, Err(rustfs_notify::NotificationError::ConfigurationConflict)) {
+                    return Err(s3::error(
+                        S3ErrorCode::InvalidRequest,
+                        "target configuration changed during validation; retry the request",
+                    ));
+                }
+                converge_target_mutation_on_cluster(app_context.as_deref(), &subsystem, local).await
+            })
+            .await?;
+        } else {
+            set_plugin_instance_config(context, &resolved, kvs).await?;
+        }
 
         Ok(build_json_response(StatusCode::OK, Body::empty(), req.headers.get("x-request-id")))
     }
@@ -1102,6 +1322,7 @@ mod tests {
             health_state: "offline".to_string(),
             health_reason: "not_loaded_in_runtime".to_string(),
             runtime_present: false,
+            has_env_default: false,
             source: TargetEndpointSource::Config,
             enabled: true,
             config: enabled_kvs("on"),
@@ -1126,6 +1347,7 @@ mod tests {
             health_state: "online".to_string(),
             health_reason: "reachable".to_string(),
             runtime_present: true,
+            has_env_default: false,
             source: TargetEndpointSource::Config,
             enabled: true,
             config: KVS(vec![
@@ -1218,6 +1440,202 @@ mod tests {
         assert_eq!(found.lookup(WEBHOOK_AUTH_TOKEN).as_deref(), Some("stored-secret-token"));
     }
 
+    #[test]
+    fn expected_config_round_trips_redacted_read_and_rejects_stale_values() {
+        let mut target = KVS::new();
+        target.insert(ENABLE_KEY.into(), "on".into());
+        target.insert(WEBHOOK_ENDPOINT.into(), "https://example.com/a".into());
+        target.insert(WEBHOOK_AUTH_TOKEN.into(), "secret-one".into());
+        let mut config = Config(HashMap::from([(
+            NOTIFY_WEBHOOK_SUB_SYS.into(),
+            HashMap::from([("primary".into(), target)]),
+        )]));
+        let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
+        let id = canonical_target_instance_id("builtin:webhook", TargetDomain::Notify, "primary");
+        let instance = super::find_target_instance(context.specs, context.route_prefix, &config, HashMap::new(), &id)
+            .expect("find instance")
+            .expect("configured target");
+        let expected = super::map_instance(instance).config;
+        assert_eq!(expected.get(WEBHOOK_AUTH_TOKEN).map(String::as_str), Some(super::REDACTED_SECRET_VALUE));
+        super::validate_expected_config(context, &config, &id, &expected).expect("GET config should be accepted");
+        config
+            .0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .expect("subsystem")
+            .get_mut("primary")
+            .expect("target")
+            .insert(WEBHOOK_AUTH_TOKEN.into(), "secret-two".into());
+        super::validate_expected_config(context, &config, &id, &expected).expect("rotation remains redacted");
+        config
+            .0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .expect("subsystem")
+            .get_mut("primary")
+            .expect("target")
+            .insert(WEBHOOK_ENDPOINT.into(), "https://example.com/b".into());
+        assert!(super::validate_expected_config(context, &config, &id, &expected).is_err());
+        config
+            .0
+            .get_mut(NOTIFY_WEBHOOK_SUB_SYS)
+            .expect("subsystem")
+            .get_mut("primary")
+            .expect("target")
+            .insert(ENABLE_KEY.into(), "off".into());
+        let disabled = super::find_target_instance(context.specs, context.route_prefix, &config, HashMap::new(), &id)
+            .expect("find disabled instance")
+            .expect("disabled target exists");
+        let disabled_expected = super::map_instance(disabled).config;
+        let error = super::validate_expected_config(context, &config, &id, &disabled_expected)
+            .expect_err("disabled target must not be enabled by editing");
+        assert!(error.to_string().contains("disabled targets"));
+        config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("subsystem").remove("primary");
+        assert!(super::validate_expected_config(context, &config, &id, &expected).is_err());
+        let legacy: PluginInstanceBody = serde_json::from_str(r#"{"key_values":[]}"#).expect("legacy input");
+        assert!(legacy.expected_config.is_none());
+    }
+
+    #[test]
+    #[serial_test::serial]
+    fn conditional_edit_rejects_inherited_environment_defaults() {
+        temp_env::with_vars(
+            [(rustfs_config::notify::ENV_NOTIFY_WEBHOOK_ENDPOINT, Some("https://example.com/hook"))],
+            || {
+                let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
+                let mut config = Config::new();
+                let mut target = KVS::new();
+                target.insert(ENABLE_KEY.into(), "on".into());
+                config
+                    .0
+                    .entry(NOTIFY_WEBHOOK_SUB_SYS.into())
+                    .or_default()
+                    .insert("Primary".into(), target);
+                let id = canonical_target_instance_id("builtin:webhook", TargetDomain::Notify, "primary");
+                let error = super::validate_expected_config(context, &config, &id, &HashMap::new())
+                    .expect_err("inherited environment defaults must prevent edits");
+                assert!(error.to_string().contains("inherits environment configuration"));
+            },
+        );
+    }
+
+    #[test]
+    fn conditional_edit_keeps_inherited_fields_in_default_layer() {
+        let mut original = KVS::new();
+        original.insert("endpoint".into(), "https://example.com/old".into());
+        let expected = HashMap::from([
+            ("endpoint".into(), "https://example.com/old".into()),
+            ("auth_token".into(), super::REDACTED_SECRET_VALUE.into()),
+            ("queue_limit".into(), "10".into()),
+        ]);
+        let mut validated = KVS::new();
+        validated.insert("endpoint".into(), "https://example.com/new".into());
+        validated.insert("auth_token".into(), "default-secret".into());
+        validated.insert("queue_limit".into(), "10".into());
+        let result = super::merge_conditional_target_update(&original, &expected, &["endpoint".into()], &validated)
+            .expect("merge edited endpoint");
+        assert_eq!(result.lookup("endpoint").as_deref(), Some("https://example.com/new"));
+        assert_eq!(result.lookup("auth_token"), None);
+        assert_eq!(result.lookup("queue_limit"), None);
+        for value in ["replacement-secret", ""] {
+            validated.insert("auth_token".into(), value.into());
+            let changed = super::merge_conditional_target_update(&original, &expected, &["auth_token".into()], &validated)
+                .expect("merge explicit secret change");
+            assert_eq!(changed.lookup("auth_token").as_deref(), Some(value));
+            assert_eq!(changed.lookup("queue_limit"), None);
+        }
+    }
+
+    #[tokio::test]
+    async fn partial_nats_update_validates_merged_authentication() {
+        use rustfs_config::{NATS_ADDRESS, NATS_PASSWORD, NATS_SUBJECT, NATS_TOKEN, NATS_USERNAME};
+        let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
+        let subsystem = rustfs_config::notify::NOTIFY_NATS_SUB_SYS;
+        let mut effective = KVS::new();
+        effective.insert(NATS_ADDRESS.into(), "nats://127.0.0.1:4222".into());
+        effective.insert(NATS_SUBJECT.into(), "events".into());
+        effective.insert(NATS_USERNAME.into(), "user".into());
+        effective.insert(NATS_PASSWORD.into(), "password".into());
+        for mut requested in [
+            vec![super::KeyValue {
+                key: NATS_TOKEN.into(),
+                value: "new-token".into(),
+            }],
+            vec![
+                super::KeyValue {
+                    key: NATS_ADDRESS.into(),
+                    value: "nats://127.0.0.1:4222".into(),
+                },
+                super::KeyValue {
+                    key: NATS_SUBJECT.into(),
+                    value: "events".into(),
+                },
+                super::KeyValue {
+                    key: NATS_USERNAME.into(),
+                    value: "user".into(),
+                },
+                super::KeyValue {
+                    key: NATS_PASSWORD.into(),
+                    value: super::REDACTED_SECRET_VALUE.into(),
+                },
+                super::KeyValue {
+                    key: NATS_TOKEN.into(),
+                    value: "new-token".into(),
+                },
+            ],
+        ] {
+            super::restore_redacted_secret_values(&mut requested, &[NATS_PASSWORD, NATS_TOKEN], Some(&effective))
+                .expect("restore placeholders before validation");
+            let candidate =
+                super::effective_validation_fields(context, subsystem, &effective, &requested).expect("merge request");
+            assert_eq!(candidate.get(NATS_PASSWORD).map(String::as_str), Some("password"));
+            let spec = context
+                .specs
+                .iter()
+                .find(|spec| spec.subsystem == subsystem)
+                .expect("NATS descriptor");
+            let error = super::build_validated_target_kvs(spec, candidate, context.default_queue_dir)
+                .await
+                .expect_err("mixed NATS authentication must fail before connection");
+            assert!(error.to_string().contains("only one auth method"));
+        }
+    }
+
+    #[test]
+    fn conditional_update_rejects_unknown_and_duplicate_keys() {
+        let context = super::plugin_instance_domain_context(PluginContractDomain::Notify);
+        let subsystem = rustfs_config::notify::NOTIFY_NATS_SUB_SYS;
+        let unknown = [super::KeyValue {
+            key: "unknown".into(),
+            value: "value".into(),
+        }];
+        let duplicate = [
+            super::KeyValue {
+                key: rustfs_config::NATS_SUBJECT.into(),
+                value: "one".into(),
+            },
+            super::KeyValue {
+                key: rustfs_config::NATS_SUBJECT.into(),
+                value: "two".into(),
+            },
+        ];
+        for request in [&unknown[..], &duplicate[..]] {
+            assert!(super::effective_validation_fields(context, subsystem, &KVS::new(), request).is_err());
+        }
+    }
+
+    #[test]
+    fn config_revision_is_scoped_to_instance_and_generation() {
+        let revision = super::config_revision("builtin:webhook:notify:one", "00000000-0000-4000-8000-000000000001");
+        assert_eq!(revision.len(), 64);
+        assert_ne!(
+            revision,
+            super::config_revision("builtin:webhook:notify:two", "00000000-0000-4000-8000-000000000001")
+        );
+        assert_ne!(
+            revision,
+            super::config_revision("builtin:webhook:notify:one", "00000000-0000-4000-8000-000000000002")
+        );
+    }
+
     #[test]
     fn plugin_instance_body_rejects_unknown_fields() {
         let err = serde_json::from_str::<PluginInstanceBody>(r#"{"key_values":[],"unexpected_field":true}"#)
@@ -1559,6 +1977,7 @@ mod tests {
             health_state: "offline".to_string(),
             health_reason: "not_loaded_in_runtime".to_string(),
             runtime_present: false,
+            has_env_default: false,
             source: TargetEndpointSource::Config,
             enabled: true,
             config: enabled_kvs("on"),
@@ -1585,6 +2004,7 @@ mod tests {
             health_state: "offline".to_string(),
             health_reason: "unreachable".to_string(),
             runtime_present: true,
+            has_env_default: false,
             source: TargetEndpointSource::Runtime,
             enabled: true,
             config: KVS::new(),
@@ -1611,6 +2031,7 @@ mod tests {
             health_state: "disabled".to_string(),
             health_reason: "disabled".to_string(),
             runtime_present: false,
+            has_env_default: false,
             source: TargetEndpointSource::Mixed,
             enabled: false,
             config: enabled_kvs("off"),
@@ -1654,6 +2075,7 @@ mod tests {
             health_state: "offline".to_string(),
             health_reason: "not_loaded_in_runtime".to_string(),
             runtime_present: false,
+            has_env_default: false,
             source: TargetEndpointSource::Config,
             enabled: true,
             config: enabled_kvs("on"),

rustfs/src/admin/handlers/target_descriptor.rs

@@ -105,6 +105,7 @@ pub(crate) struct TargetInstanceReadModel {
     pub health_state: String,
     pub health_reason: String,
     pub runtime_present: bool,
+    pub has_env_default: bool,
     pub source: TargetEndpointSource,
     pub enabled: bool,
     pub config: KVS,
@@ -391,6 +392,31 @@ pub(crate) fn merge_target_endpoints(
     Ok(endpoints)
 }
 
+fn runtime_target_plugin_id(specs: &[AdminTargetSpec], service: &str) -> &'static str {
+    target_spec_by_service(specs, service)
+        .map(|spec| builtin_target_manifest(spec.service).plugin_id)
+        .unwrap_or("custom:target")
+}
+
+pub(crate) async fn collect_instance_runtime_statuses<E>(
+    targets: Vec<SharedTarget<E>>,
+    specs: &[AdminTargetSpec],
+    domain: TargetDomain,
+    instance_id: &str,
+) -> HashMap<EndpointKey, RuntimeHealthStatus>
+where
+    E: rustfs_targets::PluginEvent,
+{
+    let targets = targets
+        .into_iter()
+        .filter(|target| {
+            let id = target.id();
+            canonical_target_instance_id(runtime_target_plugin_id(specs, &id.name), domain, &id.id) == instance_id
+        })
+        .collect();
+    collect_runtime_statuses(targets).await
+}
+
 pub(crate) fn canonical_target_instance_id(plugin_id: &str, domain: TargetDomain, instance_id: &str) -> String {
     format!("{plugin_id}:{}:{}", canonical_domain_label(domain), instance_id.to_lowercase())
 }
@@ -444,6 +470,7 @@ pub(crate) fn collect_target_instances(
             health_state: health.state,
             health_reason: health.reason,
             runtime_present,
+            has_env_default: instance.source_hints.has_env_default,
             source,
             enabled: instance.enabled,
             config: instance.effective_config,
@@ -455,9 +482,10 @@ pub(crate) fn collect_target_instances(
             continue;
         }
 
-        let (plugin_id, subsystem): (String, String) = target_spec_by_service(specs, &service)
-            .map(|spec| (builtin_target_manifest(spec.service).plugin_id.to_string(), spec.subsystem.to_string()))
-            .unwrap_or_else(|| ("custom:target".to_string(), format!("{}_{}", canonical_domain_label(domain), service)));
+        let plugin_id = runtime_target_plugin_id(specs, &service).to_owned();
+        let subsystem = target_spec_by_service(specs, &service)
+            .map(|spec| spec.subsystem.to_string())
+            .unwrap_or_else(|| format!("{}_{}", canonical_domain_label(domain), service));
         instances.push(TargetInstanceReadModel {
             canonical_id: canonical_target_instance_id(&plugin_id, domain, &account_id),
             plugin_id,
@@ -469,6 +497,7 @@ pub(crate) fn collect_target_instances(
             health_state: status.state,
             health_reason: status.reason,
             runtime_present: true,
+            has_env_default: false,
             source: TargetEndpointSource::Runtime,
             enabled: true,
             config: KVS::new(),
@@ -573,6 +602,15 @@ where
     let kv_map = collect_validated_key_values(key_values, &allowed_keys, target_type, target_label)?;
     let spec = target_spec(specs, target_type)
         .ok_or_else(|| s3_error!(InvalidArgument, "unsupported target type: '{}'", target_type))?;
+    build_validated_target_kvs(spec, kv_map, default_queue_dir).await
+}
+
+/// Validate and build from keys already checked against the target descriptor.
+pub(crate) async fn build_validated_target_kvs(
+    spec: &AdminTargetSpec,
+    kv_map: HashMap<String, String>,
+    default_queue_dir: &str,
+) -> S3Result<KVS> {
     timeout(Duration::from_secs(10), validate_target_request(spec, &kv_map, default_queue_dir))
         .await
         .map_err(|_| s3_error!(InvalidArgument, "target validation timed out after 10s"))??;
@@ -982,3 +1020,57 @@ mod webhook_request_tests {
             .expect("a public HTTPS webhook should pass outbound preflight");
     }
 }
+
+#[cfg(test)]
+mod instance_runtime_status_tests {
+    use super::{admin_target_spec_from_builtin, collect_instance_runtime_statuses};
+    use rustfs_targets::{
+        SharedTarget, TargetDomain, catalog::builtin::builtin_notify_target_admin_descriptors, testkit::MockTarget,
+    };
+    use std::sync::Arc;
+    use std::time::Duration;
+    use tokio::sync::Notify;
+
+    #[tokio::test]
+    async fn instance_health_only_probes_matching_canonical_targets() {
+        let specs: Vec<_> = builtin_notify_target_admin_descriptors()
+            .iter()
+            .map(admin_target_spec_from_builtin)
+            .collect();
+        let wanted = MockTarget::new("Primary", "webhook");
+        let unrelated = MockTarget::new("Primary", "nats").with_health_gate(Arc::new(Notify::new()));
+        let unrelated_started = unrelated.health_started();
+        let targets: Vec<SharedTarget<rustfs_notify::Event>> = vec![Arc::new(wanted), Arc::new(unrelated)];
+        let statuses = tokio::time::timeout(
+            Duration::from_secs(1),
+            collect_instance_runtime_statuses(targets, &specs, TargetDomain::Notify, "builtin:webhook:notify:primary"),
+        )
+        .await
+        .expect("unrelated blocked health must not delay instance detail");
+        assert_eq!(statuses.len(), 1);
+        assert!(statuses.contains_key(&("Primary".into(), "webhook".into())));
+        assert!(
+            tokio::time::timeout(Duration::from_millis(20), unrelated_started.notified())
+                .await
+                .is_err()
+        );
+    }
+
+    #[tokio::test]
+    async fn instance_health_preserves_unknown_runtime_identity_collisions() {
+        let specs: Vec<_> = builtin_notify_target_admin_descriptors()
+            .iter()
+            .map(admin_target_spec_from_builtin)
+            .collect();
+        let targets: Vec<SharedTarget<rustfs_notify::Event>> = vec![
+            Arc::new(MockTarget::new("Primary", "custom-one")),
+            Arc::new(MockTarget::new("PRIMARY", "custom-two")),
+            Arc::new(MockTarget::new("Primary", "webhook")),
+        ];
+        let statuses =
+            collect_instance_runtime_statuses(targets, &specs, TargetDomain::Notify, "custom:target:notify:primary").await;
+        assert_eq!(statuses.len(), 2);
+        assert!(statuses.contains_key(&("Primary".into(), "custom-one".into())));
+        assert!(statuses.contains_key(&("PRIMARY".into(), "custom-two".into())));
+    }
+}

rustfs/src/admin/handlers/target_read.rs

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 use crate::admin::{
-    handlers::target_descriptor::{EndpointKey, RuntimeHealthStatus, collect_runtime_statuses},
+    handlers::target_descriptor::{
+        AdminTargetSpec, EndpointKey, RuntimeHealthStatus, collect_instance_runtime_statuses, collect_runtime_statuses,
+    },
     runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context},
     storage_api::{
         config::read_admin_config_without_migrate,
@@ -52,6 +54,23 @@ pub(crate) async fn collect_notification_runtime_statuses() -> HashMap<EndpointK
     }
 }
 
+pub(crate) async fn collect_notification_instance_runtime_statuses(
+    specs: &[AdminTargetSpec],
+    instance_id: &str,
+) -> HashMap<EndpointKey, RuntimeHealthStatus> {
+    if let Some(system) = rustfs_notify::notification_system() {
+        collect_instance_runtime_statuses(
+            system.get_target_values().await,
+            specs,
+            rustfs_targets::TargetDomain::Notify,
+            instance_id,
+        )
+        .await
+    } else {
+        HashMap::new()
+    }
+}
+
 pub(crate) async fn collect_audit_runtime_statuses() -> HashMap<EndpointKey, RuntimeHealthStatus> {
     if let Some(system) = rustfs_audit::audit_system() {
         collect_runtime_statuses(system.get_target_values().await).await

rustfs/src/admin/plugin_contract.rs

@@ -361,6 +361,8 @@ pub(crate) struct PluginInstanceDiagnosticCount {
 
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub(crate) struct PluginInstanceDetail {
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub config_revision: Option<String>,
     #[serde(flatten)]
     pub instance: PluginInstanceEntry,
     #[serde(skip_serializing_if = "Vec::is_empty")]
@@ -539,6 +541,7 @@ mod tests {
     #[test]
     fn plugin_instance_detail_contract_serializes_diagnostics_when_present() {
         let detail = PluginInstanceDetail {
+            config_revision: None,
             instance: PluginInstanceEntry {
                 id: "builtin:webhook:notify:primary".to_string(),
                 plugin_id: "builtin:webhook".to_string(),

rustfs/src/admin/storage_api.rs

@@ -826,6 +826,12 @@ pub(crate) async fn read_admin_config_without_migrate(api: Arc<ECStore>) -> Resu
     ecstore_config::com::read_config_without_migrate(api).await
 }
 
+pub(crate) async fn read_admin_server_config_with_generation(
+    api: Arc<ECStore>,
+) -> Result<(rustfs_config::server_config::Config, Option<uuid::Uuid>)> {
+    crate::storage::storage_api::read_server_config_with_generation(api).await
+}
+
 pub(crate) async fn read_existing_admin_server_config_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
     ecstore_config::com::read_existing_server_config_no_lock(api).await
 }
@@ -1001,8 +1007,8 @@ pub(crate) mod config {
     pub(crate) use super::{
         AdminServerConfigSaveResult, AdminServerConfigSnapshot, RUSTFS_META_BUCKET, STORAGE_CLASS_SUB_SYS, delete_admin_config,
         init_admin_config_defaults, read_admin_config, read_admin_config_without_migrate, read_admin_server_config_snapshot,
-        read_existing_admin_server_config_no_lock, save_admin_config, save_admin_server_config_snapshot,
-        with_admin_server_config_read_lock,
+        read_admin_server_config_with_generation, read_existing_admin_server_config_no_lock, save_admin_config,
+        save_admin_server_config_snapshot, with_admin_server_config_read_lock,
     };
     #[cfg(test)]
     pub(crate) use super::{

rustfs/src/storage/storage_api.rs

@@ -1349,6 +1349,12 @@ pub(crate) async fn read_config_no_lock(api: Arc<ECStore>, file: &str) -> Result
     ecstore_config::com::read_config_no_lock(api, file).await
 }
 
+pub(crate) async fn read_server_config_with_generation(
+    api: Arc<ECStore>,
+) -> Result<(rustfs_config::server_config::Config, Option<uuid::Uuid>)> {
+    ecstore_config::com::read_server_config_with_generation(api).await
+}
+
 pub(crate) async fn read_existing_server_config_no_lock(api: Arc<ECStore>) -> Result<rustfs_config::server_config::Config> {
     ecstore_config::com::read_existing_server_config_no_lock(api).await
 }

scripts/check_s3s_footprint.sh

@@ -65,7 +65,7 @@ cd "$(dirname "$0")/.."
 # other reductions. The merged tree measures 209 import files and 1586
 # invocation lines, tightening main's reviewed 1592-line baseline by six.
 S3S_IMPORT_FILES_BASELINE=209
-S3_ERROR_LINES_BASELINE=1571
+S3_ERROR_LINES_BASELINE=1570
 # ecstore-scoped ratchet (rustfs/backlog#1842): the storage engine must not
 # know S3 wire/DTO types (ARCHITECTURE.md invariant 4). The S3-*consuming*
 # client was extracted to crates/s3-client, where s3s usage is legitimate;