fix(admin): decouple target inventory from runtime convergence (#8921)

* fix(admin): list destinations without runtime reconciliation Read configured notification destinations directly from persisted config and merge existing runtime health without starting target reconciliation. Cover blocked reconciliation, absent and live-only runtimes, corrupt config, and authorization precedence through the real admin handler. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * refactor(admin): unify passive target inventory reads Read one durable configuration alongside existing runtime observations for event, audit, plugin, and extension inventories. Share the configuration across plugin domains and retain strict mutation readiness separately. Propagate module-state read failures and cover fresh configuration, secret redaction, concurrent reads, cancellation, and runtime independence. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(admin): route inventory errors through the S3 facade Keep the shared target reader behind the existing admin protocol boundary and reuse its error factory without changing error codes or messages. Tighten the error-macro ratchet to the actual reduced footprint. 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>
2026-10-26 22:07:53 +00:00 · 2026-10-25 15:11:27 +08:00
parent 124cbfc00a
commit 2ccc8d8647
8 changed files with 384 additions and 73 deletions
@@ -14,13 +14,14 @@

 use crate::admin::{
    auth::authorize_admin_request,
-    handlers::audit_runtime_config::{load_server_config_from_store, update_audit_config_and_reload},
+    handlers::audit_runtime_config::update_audit_config_and_reload,
    handlers::target_descriptor::{
        AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin,
-        build_enabled_target_kvs, build_json_response, collect_runtime_statuses, extract_supported_target_params,
+        build_enabled_target_kvs, build_json_response, extract_supported_target_params,
        merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason,
        target_mutation_block_reason as shared_target_mutation_block_reason,
    },
+    handlers::target_read::{collect_audit_runtime_statuses, load_server_config_from_store, load_target_read_snapshot},
    router::{AdminOperation, Operation, S3Router},
 };
 use crate::server::{
@@ -29,7 +30,6 @@ use crate::server::{
 use http::StatusCode;
 use hyper::Method;
 use matchit::Params;
-use rustfs_audit::audit_system;
 use rustfs_config::audit::AUDIT_ROUTE_PREFIX;
 use rustfs_config::server_config::Config;
 use rustfs_config::{AUDIT_DEFAULT_DIR, MAX_ADMIN_REQUEST_BODY_SIZE};
@@ -344,12 +344,7 @@ impl Operation for ListAuditTargets {
    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
        authorize_audit_admin_request(&req, AdminAction::GetBucketTargetAction).await?;

-        let mut runtime_statuses = HashMap::new();
-        if let Some(system) = audit_system() {
-            runtime_statuses = collect_runtime_statuses(system.get_target_values().await).await;
-        }
-
-        let config = load_server_config_from_store().await?;
+        let (config, runtime_statuses) = load_target_read_snapshot(collect_audit_runtime_statuses()).await?;
        let audit_endpoints = merge_audit_endpoints(&config, runtime_statuses)?;
        let data = serde_json::to_vec(&AuditEndpointsResponse { audit_endpoints }).map_err(|e| {
            log_audit_target_request_failed!("list_audit_targets", "serialize_audit_targets_failed", None, None, e);
@@ -17,8 +17,8 @@ use crate::admin::handlers::target_descriptor::AdminTargetSpec;
 use crate::admin::runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context};
 use crate::admin::service::config::with_runtime_config_reload_lock;
 use crate::admin::storage_api::config::{
-    read_admin_config_without_migrate, read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock,
-    save_admin_server_config_snapshot, with_admin_server_config_read_lock,
+    read_admin_server_config_snapshot, read_existing_admin_server_config_no_lock, save_admin_server_config_snapshot,
+    with_admin_server_config_read_lock,
 };
 use rustfs_audit::{audit_system, start_audit_system as start_global_audit_system, system::AuditSystemState};
 use rustfs_config::DEFAULT_DELIMITER;
@@ -26,21 +26,6 @@ use rustfs_config::server_config::Config;
 use s3s::{S3Error, S3Result, s3_error};
 use tracing::warn;

-pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result<Config> {
-    let Some(store) = current_object_store_handle_for_context(context) else {
-        return Ok(Config::new());
-    };
-
-    read_admin_config_without_migrate(store)
-        .await
-        .map_err(|e| s3_error!(InternalError, "failed to read server config: {}", e))
-}
-
-pub(crate) async fn load_server_config_from_store() -> S3Result<Config> {
-    let context = current_app_context();
-    load_server_config_from_store_for_context(context.as_deref()).await
-}
-
 fn has_any_audit_targets(specs: &[AdminTargetSpec], config: &Config) -> bool {
    specs.iter().any(|spec| {
        config
@@ -188,7 +173,7 @@ mod tests {
    use super::*;
    use crate::admin::handlers::target_descriptor::admin_target_spec_from_builtin;
    use crate::admin::runtime_sources::{IamInterface, KmsInterface};
-    use crate::admin::storage_api::config::save_admin_server_config;
+    use crate::admin::storage_api::config::{read_admin_config_without_migrate, save_admin_server_config};
    use crate::admin::storage_api::error::StorageError;
    use rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS;
    use rustfs_config::server_config::KVS;
@@ -14,7 +14,7 @@

 use crate::admin::{
    auth::authorize_admin_request,
-    handlers::notify_runtime_access::{get_notification_system, load_notification_config_snapshot},
+    handlers::notify_runtime_access::{get_notification_system, load_notification_mutation_snapshot},
    handlers::supervise_admin_mutation,
    handlers::target_descriptor::{
        AdminTargetSpec, EndpointKey, RuntimeHealthStatus, TargetEndpointSource, admin_target_spec_from_builtin,
@@ -22,6 +22,7 @@ use crate::admin::{
        merge_target_endpoints as shared_merge_target_endpoints, target_module_disabled_reason,
        target_mutation_block_reason as shared_target_mutation_block_reason,
    },
+    handlers::target_read::{collect_notification_runtime_statuses, load_target_read_snapshot},
    router::{AdminOperation, Operation, S3Router},
    runtime_sources::object_store_from_extensions,
    runtime_sources::{AppContext, app_context_from_req},
@@ -464,7 +465,7 @@ impl Operation for NotificationTarget {
            log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
-        let (ns, config_snapshot) = load_notification_config_snapshot().await?;
+        let (ns, config_snapshot) = load_notification_mutation_snapshot().await?;
        if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? {
            log_notification_target_operation_blocked!("set_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
@@ -537,8 +538,7 @@ impl Operation for ListNotificationTargets {
            s3_error!(InternalError, "failed to refresh notification module state")
        })?;
        let notify_enabled = refresh_notify_module_enabled();
-        let (ns, config) = load_notification_config_snapshot().await?;
-        let runtime_statuses = collect_runtime_statuses(ns.get_target_values().await).await;
+        let (config, runtime_statuses) = load_target_read_snapshot(collect_notification_runtime_statuses()).await?;
        let notification_endpoints = merge_notification_endpoints(&config, runtime_statuses)?;

        let data = serde_json::to_vec(&NotificationEndpointsResponse {
@@ -610,7 +610,7 @@ impl Operation for RemoveNotificationTarget {
            log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
-        let (ns, config_snapshot) = load_notification_config_snapshot().await?;
+        let (ns, config_snapshot) = load_notification_mutation_snapshot().await?;
        if let Some(reason) = target_mutation_block_reason(&config_snapshot, target_type, target_name)? {
            log_notification_target_operation_blocked!("remove_target_config", Some(target_type), Some(target_name), &reason);
            return Err(s3_error!(InvalidRequest, "{reason}"));
@@ -676,6 +676,268 @@ mod tests {
        }
    }

+    #[tokio::test]
+    #[serial]
+    async fn target_list_reads_durable_config_without_runtime_reconciliation() {
+        use crate::admin::handlers::{
+            audit::ListAuditTargets,
+            extensions::ListExtensionInstancesHandler,
+            plugins_instances::{GetPluginInstanceHandler, ListPluginInstancesHandler},
+            target_read::load_target_read_snapshot,
+        };
+        use crate::admin::storage_api::config::{
+            read_admin_config, save_admin_config, save_admin_server_config, with_admin_server_config_write_lock,
+        };
+        use crate::server::{mark_event_notifier_unreconciled, with_notify_runtime_reconcile_lock};
+        use http_body_util::BodyExt as _;
+        use std::sync::Arc;
+        use std::time::Duration;
+
+        async fn response_json(
+            call: impl std::future::Future<Output = S3Result<S3Response<(StatusCode, Body)>>>,
+        ) -> S3Result<Value> {
+            let response = tokio::time::timeout(Duration::from_secs(10), call)
+                .await
+                .expect("destination inventory must not wait for runtime reconciliation")?;
+            assert_eq!(response.output.0, StatusCode::OK);
+            let body = response
+                .output
+                .1
+                .collect()
+                .await
+                .expect("collect inventory response")
+                .to_bytes();
+            Ok(serde_json::from_slice(&body).expect("decode inventory response"))
+        }
+
+        let _ = rustfs_credentials::init_global_action_credentials(
+            Some("notification-list-root".to_string()),
+            Some("notification-list-secret".to_string()),
+        );
+        let temp_dir = tempfile::tempdir().expect("create notification list test directory");
+        let env = rustfs_test_utils::TestECStoreEnv::builder()
+            .base_dir(temp_dir.path())
+            .disk_count(1)
+            .init_bucket_metadata(false)
+            .build()
+            .await;
+        crate::runtime_sources::install_test_app_context(Arc::clone(&env.ecstore)).await;
+
+        let credentials = crate::runtime_sources::current_action_credentials().expect("test root credentials");
+        let request = |uri: &str| S3Request {
+            input: Body::empty(),
+            method: Method::GET,
+            uri: uri.parse().expect("target inventory URI"),
+            headers: http::HeaderMap::new(),
+            extensions: http::Extensions::new(),
+            credentials: Some(s3s::auth::Credentials {
+                access_key: credentials.access_key.clone(),
+                secret_key: credentials.secret_key.clone().into(),
+            }),
+            region: None,
+            service: None,
+            trailing_headers: None,
+        };
+
+        temp_env::async_with_vars([(rustfs_config::ENV_NOTIFY_ENABLE, Some("true"))], async {
+            let mut config = Config::new();
+            let mut kvs = enabled_kvs("on");
+            kvs.insert(rustfs_config::WEBHOOK_ENDPOINT.to_string(), "http://127.0.0.1:9/hook".to_string());
+            kvs.insert(rustfs_config::WEBHOOK_AUTH_TOKEN.to_string(), "inventory-secret-token".to_string());
+            config
+                .0
+                .entry(rustfs_config::audit::AUDIT_WEBHOOK_SUB_SYS.to_string())
+                .or_default()
+                .insert("audit-unavailable".to_string(), kvs.clone());
+            config
+                .0
+                .entry(NOTIFY_WEBHOOK_SUB_SYS.to_string())
+                .or_default()
+                .insert("unavailable".to_string(), kvs);
+            save_admin_server_config(env.ecstore.clone(), &config)
+                .await
+                .expect("persist target without loading it into the runtime");
+
+            let (locked_tx, locked_rx) = tokio::sync::oneshot::channel();
+            let (release_tx, release_rx) = tokio::sync::oneshot::channel();
+            let store = env.ecstore.clone();
+            let writer = tokio::spawn(async move {
+                with_admin_server_config_write_lock(store, move |_guard| async move {
+                    locked_tx.send(()).expect("signal config write lock acquisition");
+                    release_rx.await.expect("release config writer");
+                })
+                .await
+                .expect("hold config write lock");
+            });
+            locked_rx.await.expect("config write lock should be held");
+            let (observing_tx, observing_rx) = tokio::sync::oneshot::channel();
+            let snapshot = tokio::spawn(load_target_read_snapshot(async move {
+                observing_tx.send(()).expect("signal runtime observation");
+            }));
+            tokio::time::timeout(Duration::from_secs(10), observing_rx)
+                .await
+                .expect("runtime observation must start while the durable read is blocked")
+                .expect("runtime observer should start");
+            assert!(!snapshot.is_finished(), "the snapshot must still await its blocked durable read");
+            release_tx.send(()).expect("release config write lock");
+            writer.await.expect("config writer task");
+            let (observed_config, ()) = snapshot.await.expect("snapshot task").expect("read durable snapshot");
+            assert!(observed_config.0[NOTIFY_WEBHOOK_SUB_SYS].contains_key("unavailable"));
+            mark_event_notifier_unreconciled();
+
+            with_notify_runtime_reconcile_lock(async {
+                for initialize_runtime in [false, true] {
+                    if initialize_runtime {
+                        rustfs_notify::ensure_live_events();
+                    }
+                    let value =
+                        response_json(ListNotificationTargets {}.call(request("/rustfs/admin/v3/target/list"), Params::new()))
+                            .await
+                            .expect("read configured destinations while the runtime is unavailable");
+                    assert_eq!(value["notify_enabled"], true);
+                    let endpoints = value["notification_endpoints"].as_array().expect("destination inventory");
+                    let endpoint = endpoints
+                        .iter()
+                        .find(|endpoint| endpoint["account_id"] == "unavailable" && endpoint["service"] == "webhook")
+                        .expect("persisted destination remains visible");
+                    assert_eq!(endpoint["status"], "offline");
+                    assert_eq!(endpoint["health_state"], "offline");
+                    assert_eq!(endpoint["health_reason"], "not_loaded_in_runtime");
+                    assert_eq!(endpoint["source"], "config");
+
+                    let audit =
+                        response_json(ListAuditTargets {}.call(request("/rustfs/admin/v3/audit/target/list"), Params::new()))
+                            .await
+                            .expect("read audit destinations without runtime initialization");
+                    assert!(
+                        audit["audit_endpoints"]
+                            .as_array()
+                            .expect("audit inventory")
+                            .iter()
+                            .any(|entry| {
+                                entry["account_id"] == "audit-unavailable" && entry["health_reason"] == "not_loaded_in_runtime"
+                            })
+                    );
+
+                    for value in [
+                        response_json(
+                            ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()),
+                        )
+                        .await
+                        .expect("read plugin inventory without runtime reconciliation"),
+                        response_json(
+                            ListExtensionInstancesHandler {}
+                                .call(request("/rustfs/admin/v4/extensions/instances"), Params::new()),
+                        )
+                        .await
+                        .expect("read extension inventory without runtime reconciliation"),
+                    ] {
+                        let instances = value["instances"].as_array().expect("instance inventory");
+                        for (domain, account_id) in [("notify", "unavailable"), ("audit", "audit-unavailable")] {
+                            let entry = instances
+                                .iter()
+                                .find(|entry| entry["domain"] == domain && entry["account_id"] == account_id)
+                                .expect("both target domains remain visible");
+                            assert_eq!(entry["status"], "offline");
+                            assert_eq!(entry["config"][rustfs_config::WEBHOOK_AUTH_TOKEN], "***redacted***");
+                        }
+                        assert!(!value.to_string().contains("inventory-secret-token"));
+                    }
+
+                    let mut router = Router::new();
+                    router.insert("/{id}", ()).expect("plugin detail route");
+                    let matched = router
+                        .at("/builtin:webhook:notify:unavailable")
+                        .expect("plugin detail params");
+                    let detail = response_json(GetPluginInstanceHandler {}.call(
+                        request("/rustfs/admin/v4/plugins/instances/builtin:webhook:notify:unavailable"),
+                        matched.params,
+                    ))
+                    .await
+                    .expect("read plugin details without runtime reconciliation");
+                    assert_eq!(detail["config"][rustfs_config::WEBHOOK_AUTH_TOKEN], "***redacted***");
+                    assert!(!detail.to_string().contains("inventory-secret-token"));
+                    assert!(!crate::server::is_event_notifier_reconciled());
+                    if let Some(system) = rustfs_notify::notification_system() {
+                        assert_eq!(system.runtime_lifecycle_state(), rustfs_notify::NotificationRuntimeState::LiveOnly);
+                    }
+                }
+
+                let target = config.0[NOTIFY_WEBHOOK_SUB_SYS]["unavailable"].clone();
+                let targets = config.0.get_mut(NOTIFY_WEBHOOK_SUB_SYS).expect("notification target config");
+                targets.remove("unavailable");
+                targets.insert("latest".to_string(), target);
+                save_admin_server_config(env.ecstore.clone(), &config)
+                    .await
+                    .expect("persist a newer target configuration while runtime stays unreconciled");
+                let latest = response_json(
+                    ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()),
+                )
+                .await
+                .expect("inventory must read the latest durable config");
+                let instances = latest["instances"].as_array().expect("latest inventory");
+                assert!(instances.iter().any(|entry| entry["account_id"] == "latest"));
+                assert!(!instances.iter().any(|entry| entry["account_id"] == "unavailable"));
+
+                let valid_config = read_admin_config(env.ecstore.clone(), "config/config.json")
+                    .await
+                    .expect("read valid config bytes for restoration");
+                save_admin_config(env.ecstore.clone(), "config/config.json", b"invalid-server-config".to_vec())
+                    .await
+                    .expect("inject corrupt persisted configuration");
+                let error = ListNotificationTargets {}
+                    .call(request("/rustfs/admin/v3/target/list"), Params::new())
+                    .await
+                    .expect_err("unreadable persisted config must not become an empty inventory");
+                assert_eq!(error.code(), &S3ErrorCode::InternalError);
+
+                let error = response_json(
+                    ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()),
+                )
+                .await
+                .expect_err("plugin inventory must not hide corrupt durable config");
+                assert_eq!(error.code(), &S3ErrorCode::InternalError);
+                let error =
+                    tokio::time::timeout(Duration::from_secs(10), load_target_read_snapshot(std::future::pending::<()>()))
+                        .await
+                        .expect("a config read failure must cancel stalled runtime observations")
+                        .expect_err("config read failure must remain visible");
+                assert_eq!(error.code(), &S3ErrorCode::InternalError);
+
+                let mut unsigned = request("/rustfs/admin/v3/target/list");
+                unsigned.credentials = None;
+                let error = ListNotificationTargets {}
+                    .call(unsigned, Params::new())
+                    .await
+                    .expect_err("authorization must precede persisted configuration access");
+                assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+                assert_eq!(error.message(), Some("credentials not found"));
+
+                save_admin_config(env.ecstore.clone(), "config/config.json", valid_config)
+                    .await
+                    .expect("restore valid target configuration");
+                save_admin_config(env.ecstore.clone(), "config/module_switches.json", b"invalid-module-switches".to_vec())
+                    .await
+                    .expect("inject corrupt persisted module switches");
+                let error = response_json(
+                    ListPluginInstancesHandler {}.call(request("/rustfs/admin/v4/plugins/instances"), Params::new()),
+                )
+                .await
+                .expect_err("plugin diagnostics must not silently reuse stale module switches");
+                assert_eq!(error.code(), &S3ErrorCode::InternalError);
+                let mut unsigned = request("/rustfs/admin/v4/plugins/instances");
+                unsigned.credentials = None;
+                let error = response_json(ListPluginInstancesHandler {}.call(unsigned, Params::new()))
+                    .await
+                    .expect_err("plugin authorization must precede module and config access");
+                assert_eq!(error.code(), &S3ErrorCode::InvalidRequest);
+                assert_eq!(error.message(), Some("authentication required"));
+            })
+            .await;
+        })
+        .await;
+    }
+
    fn subscription_router() -> Router<()> {
        let mut router = Router::new();
        router
@@ -1319,7 +1581,7 @@ mod tests {
            .find("let notify_enabled = refresh_notify_module_enabled();")
            .expect("target list should resolve the effective env and persisted module state");
        let load_index = list_block
-            .find("load_notification_config_snapshot().await?")
+            .find("load_target_read_snapshot(collect_notification_runtime_statuses()).await?")
            .expect("target list should load notification config");
        assert!(
            authorize_index < refresh_index && refresh_index < effective_state_index && effective_state_index < load_index,
@@ -69,6 +69,7 @@ pub mod sts;
 pub mod system;
 pub mod table_catalog;
 mod target_descriptor;
+mod target_read;
 pub mod tier;
 pub mod tls_certificate;
 pub mod tls_debug;
@@ -36,7 +36,7 @@ pub(crate) async fn get_notification_system() -> S3Result<Arc<rustfs_notify::Not
    Ok(system)
 }

-pub(crate) async fn load_notification_config_snapshot() -> S3Result<(Arc<rustfs_notify::NotificationSystem>, Config)> {
+pub(crate) async fn load_notification_mutation_snapshot() -> S3Result<(Arc<rustfs_notify::NotificationSystem>, Config)> {
    let system = get_notification_system().await?;
    let config = system.config_snapshot().await;
    Ok((system, config))
@@ -14,20 +14,25 @@

 use crate::admin::{
    auth::authorize_admin_request,
-    handlers::audit_runtime_config::{load_server_config_from_store, remove_audit_target_config, set_audit_target_config},
+    handlers::audit_runtime_config::{remove_audit_target_config, set_audit_target_config},
    handlers::notify_runtime_access::{
-        load_notification_config_snapshot, remove_notification_target_config, set_notification_target_config,
+        load_notification_mutation_snapshot, remove_notification_target_config, set_notification_target_config,
    },
    handlers::target_descriptor::{
        AdminTargetSpec, RuntimeHealthStatus, TargetEndpointSource, TargetInstanceReadModel, admin_target_spec_from_builtin,
-        build_enabled_target_kvs, build_json_response, collect_runtime_statuses, collect_target_instances, find_target_instance,
+        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,
    },
+    handlers::target_read::{
+        collect_audit_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},
+    storage_api::s3::{self, S3ErrorCode},
 };
 use crate::server::{
    ADMIN_PREFIX, is_audit_module_enabled, is_notify_module_enabled, refresh_audit_module_enabled, refresh_notify_module_enabled,
@@ -35,7 +40,6 @@ use crate::server::{
 };
 use hyper::{Method, StatusCode};
 use matchit::Params;
-use rustfs_audit::audit_system;
 use rustfs_config::audit::AUDIT_ROUTE_PREFIX;
 use rustfs_config::notify::NOTIFY_ROUTE_PREFIX;
 use rustfs_config::server_config::{Config, KVS};
@@ -597,13 +601,15 @@ fn plugin_instance_mutation_block_reason(
 /// Reloads persisted module switches from the config store. Call once per
 /// admin request before evaluating [`module_disabled_block_reason`] so that
 /// list/detail paths do not hit the store once per domain or per instance.
-async fn refresh_persisted_module_switches() {
+async fn refresh_persisted_module_switches() -> S3Result<()> {
    if let Err(err) = refresh_persisted_module_switches_from_store().await {
        warn!(
            error = %err,
            "failed to reload persisted module switches before checking plugin instance operation gating"
        );
+        return Err(s3::error(S3ErrorCode::InternalError, "failed to refresh target module state"));
    }
+    Ok(())
 }

 fn module_disabled_block_reason(domain: PluginContractDomain, action: &str) -> Option<String> {
@@ -619,53 +625,53 @@ fn module_disabled_block_reason(domain: PluginContractDomain, action: &str) -> O
    }
 }

-async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> Option<String> {
-    refresh_persisted_module_switches().await;
-    module_disabled_block_reason(context.domain, action)
+async fn plugin_instance_operation_block_reason(context: PluginInstanceDomainContext, action: &str) -> S3Result<Option<String>> {
+    refresh_persisted_module_switches().await?;
+    Ok(module_disabled_block_reason(context.domain, action))
 }

 async fn plugin_instance_runtime_statuses(
    context: PluginInstanceDomainContext,
-) -> S3Result<HashMap<(String, String), RuntimeHealthStatus>> {
+) -> HashMap<(String, String), RuntimeHealthStatus> {
    match context.domain {
-        PluginContractDomain::Notify => {
-            let (ns, _) = load_notification_config_snapshot().await?;
-            Ok(collect_runtime_statuses(ns.get_target_values().await).await)
-        }
-        PluginContractDomain::Audit => {
-            let mut runtime_statuses = HashMap::new();
-            if let Some(system) = audit_system() {
-                runtime_statuses = collect_runtime_statuses(system.get_target_values().await).await;
-            }
-            Ok(runtime_statuses)
-        }
+        PluginContractDomain::Notify => collect_notification_runtime_statuses().await,
+        PluginContractDomain::Audit => collect_audit_runtime_statuses().await,
    }
 }

-async fn plugin_instance_config_snapshot(context: PluginInstanceDomainContext) -> S3Result<Config> {
+async fn plugin_instance_mutation_config_snapshot(context: PluginInstanceDomainContext) -> S3Result<Config> {
    match context.domain {
-        PluginContractDomain::Notify => load_notification_config_snapshot().await.map(|(_, config)| config),
+        PluginContractDomain::Notify => load_notification_mutation_snapshot().await.map(|(_, config)| config),
        PluginContractDomain::Audit => load_server_config_from_store().await,
    }
 }

-async fn collect_domain_instances(context: PluginInstanceDomainContext) -> S3Result<Vec<PluginInstanceEntry>> {
-    let runtime_statuses = plugin_instance_runtime_statuses(context).await?;
-    let config = plugin_instance_config_snapshot(context).await?;
+fn collect_domain_instances(
+    context: PluginInstanceDomainContext,
+    config: &Config,
+    runtime_statuses: HashMap<(String, String), RuntimeHealthStatus>,
+) -> S3Result<Vec<PluginInstanceEntry>> {
    let module_disabled_reason = module_disabled_block_reason(context.domain, "listing plugin instances");
    let mut entries = Vec::new();
-    for instance in collect_target_instances(context.specs, context.route_prefix, &config, runtime_statuses)? {
+    for instance in collect_target_instances(context.specs, context.route_prefix, config, runtime_statuses)? {
        entries.push(plugin_instance_list_entry(instance, module_disabled_reason.clone()));
    }
    Ok(entries)
 }

 pub(super) async fn collect_all_instances() -> S3Result<Vec<PluginInstanceEntry>> {
-    refresh_persisted_module_switches().await;
-    let (mut notify_instances, audit_instances) = tokio::try_join!(
-        collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Notify)),
-        collect_domain_instances(plugin_instance_domain_context(PluginContractDomain::Audit))
-    )?;
+    refresh_persisted_module_switches().await?;
+    let notify_context = plugin_instance_domain_context(PluginContractDomain::Notify);
+    let audit_context = plugin_instance_domain_context(PluginContractDomain::Audit);
+    let (config, (notify_statuses, audit_statuses)) = load_target_read_snapshot(async {
+        tokio::join!(
+            plugin_instance_runtime_statuses(notify_context),
+            plugin_instance_runtime_statuses(audit_context)
+        )
+    })
+    .await?;
+    let mut notify_instances = collect_domain_instances(notify_context, &config, notify_statuses)?;
+    let audit_instances = collect_domain_instances(audit_context, &config, audit_statuses)?;
    notify_instances.extend(audit_instances);
    notify_instances.sort_by(|a, b| a.service.cmp(&b.service).then_with(|| a.account_id.cmp(&b.account_id)));
    Ok(notify_instances)
@@ -673,8 +679,7 @@ pub(super) async fn collect_all_instances() -> S3Result<Vec<PluginInstanceEntry>

 async fn find_plugin_instance(instance_id: &str) -> S3Result<Option<TargetInstanceReadModel>> {
    let context = plugin_instance_domain_context(parse_plugin_instance_id(instance_id)?.1);
-    let runtime_statuses = plugin_instance_runtime_statuses(context).await?;
-    let config = plugin_instance_config_snapshot(context).await?;
+    let (config, runtime_statuses) = load_target_read_snapshot(plugin_instance_runtime_statuses(context)).await?;
    find_target_instance(context.specs, context.route_prefix, &config, runtime_statuses, instance_id)
 }

@@ -738,7 +743,7 @@ impl Operation for GetPluginInstanceHandler {
            .get("id")
            .ok_or_else(|| s3_error!(InvalidArgument, "missing required parameter: 'id'"))?;

-        refresh_persisted_module_switches().await;
+        refresh_persisted_module_switches().await?;
        let instance = find_plugin_instance(instance_id)
            .await?
            .ok_or_else(|| s3_error!(NoSuchKey, "plugin instance not found"))?;
@@ -761,7 +766,8 @@ impl Operation for PutPluginInstanceHandler {
        let resolved = resolve_plugin_instance_target(instance_id)?;
        let context = resolved.context;

-        if let Some(reason) = plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await
+        if let Some(reason) =
+            plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await?
        {
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }
@@ -774,7 +780,7 @@ 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_config_snapshot(context).await?;
+        let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?;
        if let Some(reason) = plugin_instance_mutation_block_reason(
            context,
            &config_snapshot,
@@ -821,12 +827,13 @@ impl Operation for DeletePluginInstanceHandler {
        let resolved = resolve_plugin_instance_target(instance_id)?;
        let context = resolved.context;

-        if let Some(reason) = plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await
+        if let Some(reason) =
+            plugin_instance_operation_block_reason(context, "managing plugin instances from the console").await?
        {
            return Err(s3_error!(InvalidRequest, "{reason}"));
        }

-        let config_snapshot = plugin_instance_config_snapshot(context).await?;
+        let config_snapshot = plugin_instance_mutation_config_snapshot(context).await?;
        if let Some(reason) = plugin_instance_mutation_block_reason(
            context,
            &config_snapshot,
@@ -0,0 +1,61 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+use crate::admin::{
+    handlers::target_descriptor::{EndpointKey, RuntimeHealthStatus, collect_runtime_statuses},
+    runtime_sources::{AppContext, current_app_context, current_object_store_handle_for_context},
+    storage_api::{
+        config::read_admin_config_without_migrate,
+        s3::{self, S3Error, S3ErrorCode, S3Result},
+    },
+};
+use rustfs_config::server_config::Config;
+use std::{collections::HashMap, future::Future};
+
+pub(crate) async fn load_server_config_from_store_for_context(context: Option<&AppContext>) -> S3Result<Config> {
+    let Some(store) = current_object_store_handle_for_context(context) else {
+        return Ok(Config::new());
+    };
+
+    read_admin_config_without_migrate(store)
+        .await
+        .map_err(|e| s3::error(S3ErrorCode::InternalError, format!("failed to read server config: {e}")))
+}
+
+pub(crate) async fn load_server_config_from_store() -> S3Result<Config> {
+    let context = current_app_context();
+    load_server_config_from_store_for_context(context.as_deref()).await
+}
+
+/// Reads one durable configuration alongside passive runtime observations.
+/// Neither source waits for target initialization or lifecycle convergence.
+pub(crate) async fn load_target_read_snapshot<T>(runtime: impl Future<Output = T>) -> S3Result<(Config, T)> {
+    tokio::try_join!(load_server_config_from_store(), async { Ok::<_, S3Error>(runtime.await) })
+}
+
+pub(crate) async fn collect_notification_runtime_statuses() -> HashMap<EndpointKey, RuntimeHealthStatus> {
+    if let Some(system) = rustfs_notify::notification_system() {
+        collect_runtime_statuses(system.get_target_values().await).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
+    } else {
+        HashMap::new()
+    }
+}
@@ -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=1572
+S3_ERROR_LINES_BASELINE=1571
 # 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;