fix(targets): isolate invalid target instances instead of failing the whole subsystem (#5118)

A single instance with a malformed `enable` value (e.g. the typo `enable`
instead of `enabled`/`on`) made every configured notification/audit target
of every type fail to load. `create_targets_from_config_with_store_mode`
collected instance configs through the strict `try_collect_target_configs`,
whose `.collect::<Result<Vec, _>>()` short-circuits on the first error and
drops all siblings; the surrounding `?` then aborted the loop over every
plugin type.

This contradicted the function's own documented contract ("Creation is
fault-isolated per instance") and was inconsistent: target construction
failures were already isolated into the `failures` accumulator, only config
collection failures were fatal.

Add `collect_target_config_results[_from_env]`, which returns the enabled
`(id, config)` pairs plus a per-instance failure summary, and use it in the
create path. Per-instance collection errors now flow into the same
`failures`/`creation_failures` accumulator as construction failures: healthy
siblings and other target types still load, while the malformed instance is
surfaced so the notify lifecycle stays non-converged/retryable and an Admin
write cannot report a false success. Audit likewise keeps logging through
its valid targets instead of going fully Stopped.

Follow-up to #5088.

crates/targets/src/config/loader.rs

@@ -37,6 +37,22 @@ pub fn try_collect_target_configs(
     try_collect_target_configs_from_env(config, route_prefix, target_type, valid_fields, std::env::vars())
 }
 
+/// Collects enabled target instance configs with per-instance fault isolation.
+///
+/// Unlike [`try_collect_target_configs`], an instance with an invalid
+/// configuration (for example an unparseable `enable` value) does not abort the
+/// whole target type. The enabled `(id, config)` pairs are returned alongside a
+/// summary line for every instance that could not be collected, so callers can
+/// load the healthy instances while still surfacing the failed ones.
+pub fn collect_target_config_results(
+    config: &Config,
+    route_prefix: &str,
+    target_type: &str,
+    valid_fields: &HashSet<String>,
+) -> (Vec<(String, KVS)>, Vec<String>) {
+    collect_target_config_results_from_env(config, route_prefix, target_type, valid_fields, std::env::vars())
+}
+
 fn is_sensitive_target_field(field_name: &str) -> bool {
     let field_name = field_name.to_ascii_lowercase();
     field_name.contains("password")
@@ -171,6 +187,35 @@ where
     .collect())
 }
 
+pub fn collect_target_config_results_from_env<I>(
+    config: &Config,
+    route_prefix: &str,
+    target_type: &str,
+    valid_fields: &HashSet<String>,
+    env_vars: I,
+) -> (Vec<(String, KVS)>, Vec<String>)
+where
+    I: IntoIterator<Item = (String, String)>,
+{
+    let mut configs = Vec::new();
+    let mut failures = Vec::new();
+    for result in collect_merged_target_config_results_from_env(
+        config,
+        &format!("{route_prefix}{target_type}").to_lowercase(),
+        route_prefix,
+        target_type,
+        valid_fields,
+        env_vars,
+    ) {
+        match result {
+            Ok(record) if record.enabled => configs.push((record.instance_id, record.effective_config)),
+            Ok(_) => {}
+            Err((record, err)) => failures.push(format!("{target_type}/{}: {err}", record.instance_id)),
+        }
+    }
+    (configs, failures)
+}
+
 pub(crate) fn collect_merged_target_configs_from_env<I>(
     config: &Config,
     section_name: &str,
@@ -330,8 +375,8 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        collect_env_target_instance_ids_from_env, collect_target_configs_from_env, redact_target_field_value,
-        redacted_target_config, try_collect_target_configs_from_env,
+        collect_env_target_instance_ids_from_env, collect_target_config_results_from_env, collect_target_configs_from_env,
+        redact_target_field_value, redacted_target_config, try_collect_target_configs_from_env,
     };
     use crate::TargetError;
     use rustfs_config::notify::{
@@ -516,6 +561,48 @@ mod tests {
         assert_eq!(configs[0].0, "good");
     }
 
+    #[test]
+    fn collect_target_config_results_isolates_invalid_instance_and_reports_it() {
+        let (configs, failures) = collect_target_config_results_from_env(
+            &Config(HashMap::new()),
+            NOTIFY_ROUTE_PREFIX,
+            "webhook",
+            &HashSet::from([ENABLE_KEY.to_string(), WEBHOOK_ENDPOINT.to_string()]),
+            vec![
+                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_GOOD".to_string(), "on".to_string()),
+                ("RUSTFS_NOTIFY_WEBHOOK_ENDPOINT_GOOD".to_string(), "https://example.com/good".to_string()),
+                // "enable" is a common typo: EnableState accepts "enabled"/"on", not "enable".
+                ("RUSTFS_NOTIFY_WEBHOOK_ENABLE_BAD".to_string(), "enable".to_string()),
+            ],
+        );
+
+        // The healthy instance still loads even though a sibling is malformed...
+        assert_eq!(configs.len(), 1);
+        assert_eq!(configs[0].0, "good");
+        // ...and the malformed instance is surfaced (not silently dropped, not
+        // aborting the whole target type) so callers can fail-report it.
+        assert_eq!(failures.len(), 1);
+        assert!(
+            failures[0].contains("webhook/bad") && failures[0].contains("enable"),
+            "unexpected failure summary: {}",
+            failures[0]
+        );
+    }
+
+    #[test]
+    fn collect_target_config_results_does_not_report_validly_disabled_instances() {
+        let (configs, failures) = collect_target_config_results_from_env(
+            &Config(HashMap::new()),
+            NOTIFY_ROUTE_PREFIX,
+            "webhook",
+            &HashSet::from([ENABLE_KEY.to_string(), WEBHOOK_ENDPOINT.to_string()]),
+            vec![("RUSTFS_NOTIFY_WEBHOOK_ENABLE_PRIMARY".to_string(), "off".to_string())],
+        );
+
+        assert!(configs.is_empty());
+        assert!(failures.is_empty(), "a validly disabled instance is not a failure");
+    }
+
     #[test]
     fn collect_target_configs_preserves_whitespace_padded_legacy_value() {
         let configs = try_collect_target_configs_from_env(

crates/targets/src/config/mod.rs

@@ -24,8 +24,9 @@ pub use instance::{
     try_normalize_target_plugin_instances, try_normalize_target_plugin_instances_from_env,
 };
 pub use loader::{
-    collect_env_target_instance_ids, collect_env_target_instance_ids_from_env, collect_target_configs,
-    collect_target_configs_from_env, try_collect_target_configs, try_collect_target_configs_from_env,
+    collect_env_target_instance_ids, collect_env_target_instance_ids_from_env, collect_target_config_results,
+    collect_target_config_results_from_env, collect_target_configs, collect_target_configs_from_env, try_collect_target_configs,
+    try_collect_target_configs_from_env,
 };
 pub use target_args::{
     build_amqp_args, build_kafka_args, build_mqtt_args, build_mysql_args, build_nats_args, build_postgres_args,

crates/targets/src/plugin.rs

@@ -14,7 +14,7 @@
 
 use crate::{
     PluginRuntimeAdapter, RuntimeActivation, Target, TargetError,
-    config::try_collect_target_configs,
+    config::collect_target_config_results,
     manifest::{TargetPluginManifest, builtin_target_manifest},
     target::with_deferred_queue_store_open,
 };
@@ -347,7 +347,16 @@ where
 
         for (target_type, plugin) in &self.plugins {
             info!(target_type = %target_type, "Start working on target type");
-            for (id, merged_config) in try_collect_target_configs(config, route_prefix, target_type, plugin.valid_fields_set())? {
+            // Per-instance fault isolation: an invalid instance (e.g. an
+            // unparseable `enable` value) is recorded as a failure and skipped,
+            // never aborting the remaining instances or other target types.
+            let (collected, invalid_instances) =
+                collect_target_config_results(config, route_prefix, target_type, plugin.valid_fields_set());
+            for detail in invalid_instances {
+                error!(target_type = %target_type, reason = "invalid_config", detail = %detail, "Skipping target instance with invalid configuration");
+                failures.push(detail);
+            }
+            for (id, merged_config) in collected {
                 info!(target_type = %target_type, instance_id = %id, "Target is enabled, ready to create");
                 let created = if defer_store_open {
                     with_deferred_queue_store_open(|| self.create_target(target_type, id.clone(), &merged_config))
@@ -505,4 +514,61 @@ mod tests {
         assert_eq!(activation.targets[0].id().to_string(), "primary:test");
         assert!(activation.replay_workers.is_empty());
     }
+
+    // Regression: a single instance with a malformed `enable` value must not
+    // abort the remaining instances or unrelated target types. Before this fix
+    // the collector short-circuited the whole create path, so one typo took
+    // down every notify/audit target.
+    #[tokio::test]
+    async fn create_dormant_isolates_invalid_enable_and_still_loads_other_targets() {
+        let mut registry = TargetPluginRegistry::<String>::new();
+        for target_type in ["alpha", "beta"] {
+            registry.register(TargetPluginDescriptor::new(
+                target_type,
+                &[ENABLE_KEY, "endpoint"],
+                |_config| Ok(()),
+                move |id, _config| {
+                    Ok(Box::new(TestTarget {
+                        id: crate::arn::TargetID::new(id, target_type.to_string()),
+                    }))
+                },
+            ));
+        }
+
+        let mut cfg = Config(HashMap::new());
+
+        // alpha: one healthy instance plus one with a malformed `enable` value
+        // ("enable" is a typo -- EnableState accepts "enabled"/"on", not "enable").
+        let mut alpha = HashMap::new();
+        let mut alpha_good = KVS::new();
+        alpha_good.insert(ENABLE_KEY.to_string(), "on".to_string());
+        alpha_good.insert("endpoint".to_string(), "https://example.com/alpha".to_string());
+        alpha.insert("good".to_string(), alpha_good);
+        let mut alpha_bad = KVS::new();
+        alpha_bad.insert(ENABLE_KEY.to_string(), "enable".to_string());
+        alpha.insert("bad".to_string(), alpha_bad);
+        cfg.0.insert("notify_alpha".to_string(), alpha);
+
+        // beta: a healthy instance in a different target type must survive.
+        let mut beta = HashMap::new();
+        let mut beta_primary = KVS::new();
+        beta_primary.insert(ENABLE_KEY.to_string(), "on".to_string());
+        beta_primary.insert("endpoint".to_string(), "https://example.com/beta".to_string());
+        beta.insert("primary".to_string(), beta_primary);
+        cfg.0.insert("notify_beta".to_string(), beta);
+
+        let (targets, failures) = registry
+            .create_dormant_targets_from_config(&cfg, "notify_")
+            .await
+            .expect("a malformed instance must not abort target creation");
+
+        let mut created: Vec<String> = targets.iter().map(|target| target.id().to_string()).collect();
+        created.sort();
+        assert_eq!(created, vec!["good:alpha".to_string(), "primary:beta".to_string()]);
+
+        // The malformed instance is surfaced (so an Admin write can't report a
+        // false success) rather than silently dropped or fatally aborting.
+        assert_eq!(failures.len(), 1);
+        assert!(failures[0].contains("alpha/bad"), "unexpected failure summary: {}", failures[0]);
+    }
 }