refactor: reuse embedded startup phase boundaries (#3640)

2026-08-12 08:06:54 +00:00 · 2026-06-20 06:58:48 +08:00
parent f8117eb46b
commit 131e9dc804
5 changed files with 188 additions and 126 deletions
@@ -49,28 +49,18 @@
 use crate::config::Config;
 use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system};
 use crate::server::{ShutdownHandle, shutdown_event_notifier, start_http_server, stop_audit_system};
-use crate::startup_fs_guard::enforce_unsupported_fs_policy;
 use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
-use crate::storage_compat::init_lock_clients;
-use crate::storage_compat::{
-    ECStore, EndpointServerPools, init as init_ecstore_config, init_background_replication, init_bucket_metadata_sys,
-    init_global_config_sys, init_local_disks, set_global_endpoints, set_global_rustfs_port, try_migrate_bucket_metadata,
-    try_migrate_iam_config, try_migrate_server_config, update_erasure_type,
-};
-use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
-use rustfs_credentials::init_global_action_credentials;
+use crate::startup_server::init_embedded_startup_listen_context;
+use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime};
+use crate::storage_compat::{init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config};
 use rustfs_obs::{init_obs, set_global_guard};
 use rustfs_storage_api::{BucketOperations, BucketOptions};
-use rustfs_utils::net::parse_and_resolve_address;
 use rustls::crypto::aws_lc_rs::default_provider;
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::path::{Path, PathBuf};
-use std::sync::{
-    Arc,
-    atomic::{AtomicBool, Ordering},
-};
+use std::sync::atomic::{AtomicBool, Ordering};
 use tokio_util::sync::CancellationToken;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, info, warn};

 const LOG_COMPONENT_EMBEDDED: &str = "embedded";
 const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded";
@@ -302,110 +292,41 @@ impl RustFSServerBuilder {
        // Trusted proxies.
        rustfs_trusted_proxies::init();

-        // Resolve listen address before endpoint/global initialization.
-        let server_addr =
-            parse_and_resolve_address(config.address.as_str()).map_err(|e| ServerError::Init(format!("address: {e}")))?;
-
-        if server_addr.port() == 0 {
-            return Err(ServerError::Init(
-                "port 0 is not supported in embedded mode because startup requires \
-                 a stable listen address and port before endpoint/global initialization. \
-                 Use `find_available_port()` to obtain a free port."
-                    .to_string(),
-            ));
-        }
-
-        // Credentials.
-        init_global_action_credentials(Some(config.access_key.clone()), Some(config.secret_key.clone()))
-            .map_err(|e| ServerError::Init(format!("credentials: {e:?}")))?;
-
-        // Region.
-        if let Some(region_str) = &config.region {
-            let region = region_str
-                .parse()
-                .map_err(|e| ServerError::Init(format!("invalid region '{region_str}': {e}")))?;
-            crate::storage_compat::set_global_region(region);
-        }
-
-        let server_port = server_addr.port();
-
-        set_global_rustfs_port(server_port);
-        set_global_addr(&config.address).await;
+        let listen_context = init_embedded_startup_listen_context(&config)
+            .await
+            .map_err(|e| ServerError::Init(e.to_string()))?;

        set_global_init_guard()?;

-        // Endpoints / erasure setup.
-        let server_addr_str = server_addr.to_string();
-        let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_addr_str.as_str(), config.volumes.clone())
+        let endpoint_pools = init_embedded_startup_storage_foundation(&listen_context.server_address, &config.volumes)
            .await
-            .map_err(|e| ServerError::Init(format!("endpoints: {e}")))?;
-        enforce_unsupported_fs_policy(&endpoint_pools).map_err(|e| ServerError::Init(format!("unsupported fs guard: {e}")))?;
-
-        set_global_endpoints(endpoint_pools.as_ref().clone());
-        update_erasure_type(setup_type).await;
-
-        // Local disks.
-        init_local_disks(endpoint_pools.clone())
-            .await
-            .map_err(|e| ServerError::Init(format!("local disks: {e}")))?;
-        init_lock_clients(endpoint_pools.clone());
-
-        // Service state.
-        let readiness = Arc::new(GlobalReadiness::new());
+            .map_err(|e| ServerError::Init(e.to_string()))?;

        // Start HTTP server.
        let mut s3_config = config.clone();
        s3_config.console_enable = false;
-        let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness.clone()).await?;
+        let (shutdown_handle, bound_addr) = start_http_server(&s3_config, listen_context.readiness.clone()).await?;
        let ctx = CancellationToken::new();
        let shutdown_embedded_server = || {
            shutdown_handle.signal();
            ctx.cancel();
        };

-        // Storage engine.
-        let store = match ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()).await {
-            Ok(store) => store,
+        let storage_runtime = match init_embedded_startup_storage_runtime(
+            listen_context.server_addr,
+            &endpoint_pools,
+            listen_context.readiness.clone(),
+            ctx.clone(),
+        )
+        .await
+        {
+            Ok(runtime) => runtime,
            Err(e) => {
-                error!(
-                    component = LOG_COMPONENT_EMBEDDED,
-                    subsystem = LOG_SUBSYSTEM_EMBEDDED,
-                    event = "embedded_storage_init_failed",
-                    stage = "ecstore_new",
-                    error = ?e,
-                    "Embedded storage initialization failed"
-                );
                shutdown_embedded_server();
-                return Err(ServerError::Init(format!("ECStore: {e}")));
+                return Err(ServerError::Init(e.to_string()));
            }
        };
-
-        init_ecstore_config();
-        try_migrate_server_config(store.clone()).await;
-
-        // Global config system (with retry).
-        let mut retry = 0;
-        while let Err(e) = init_global_config_sys(store.clone()).await {
-            retry += 1;
-            if retry > 15 {
-                shutdown_embedded_server();
-                return Err(ServerError::Init(format!("init_global_config_sys failed after 15 retries: {e}")));
-            }
-            debug!(
-                component = LOG_COMPONENT_EMBEDDED,
-                subsystem = LOG_SUBSYSTEM_EMBEDDED,
-                event = "embedded_storage_init_retry",
-                stage = "global_config_sys",
-                retry,
-                error = %e,
-                "Embedded storage initialization retry scheduled"
-            );
-            tokio::time::sleep(std::time::Duration::from_secs(1)).await;
-        }
-        readiness.mark_stage(SystemStage::StorageReady);
-
-        // Replication.
-        init_background_replication(store.clone()).await;
+        let store = storage_runtime.store;

        // KMS (optional, non-fatal for embedded).
        if let Err(e) = init_kms_system(&config).await {
@@ -455,12 +376,13 @@ impl RustFSServerBuilder {
        // IAM.
        let kms_interface =
            rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager);
-        let iam_bootstrap = bootstrap_or_defer_iam_init(store.clone(), kms_interface, readiness.clone(), None, Some(ctx.clone()))
-            .await
-            .map_err(|e| {
-                shutdown_embedded_server();
-                ServerError::Init(format!("IAM bootstrap setup: {e}"))
-            })?;
+        let iam_bootstrap =
+            bootstrap_or_defer_iam_init(store.clone(), kms_interface, listen_context.readiness.clone(), None, Some(ctx.clone()))
+                .await
+                .map_err(|e| {
+                    shutdown_embedded_server();
+                    ServerError::Init(format!("IAM bootstrap setup: {e}"))
+                })?;

        // Bucket notifications.
        add_bucket_notification_configuration(buckets.clone()).await;
@@ -476,7 +398,7 @@ impl RustFSServerBuilder {
            );
        }

-        publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), None)
+        publish_ready_for_iam_bootstrap(iam_bootstrap, listen_context.readiness.as_ref(), None)
            .await
            .map_err(|e| {
                shutdown_embedded_server();
@@ -44,6 +44,12 @@ pub struct StartupListenContext {
    pub server_address: String,
 }

+pub struct EmbeddedStartupListenContext {
+    pub readiness: Arc<GlobalReadiness>,
+    pub server_addr: SocketAddr,
+    pub server_address: String,
+}
+
 pub struct StartupHttpServers {
    pub state_manager: Arc<ServiceStateManager>,
    pub s3_shutdown_tx: Option<ShutdownHandle>,
@@ -99,6 +105,39 @@ pub async fn init_startup_listen_context(config: &Config) -> Result<StartupListe
    })
 }

+pub async fn init_embedded_startup_listen_context(config: &Config) -> Result<EmbeddedStartupListenContext> {
+    let readiness = Arc::new(GlobalReadiness::new());
+
+    let server_addr =
+        parse_and_resolve_address(config.address.as_str()).map_err(|err| Error::other(format!("address: {err}")))?;
+    if server_addr.port() == 0 {
+        return Err(Error::other(
+            "port 0 is not supported in embedded mode because startup requires \
+             a stable listen address and port before endpoint/global initialization. \
+             Use `find_available_port()` to obtain a free port.",
+        ));
+    }
+
+    init_global_action_credentials(Some(config.access_key.clone()), Some(config.secret_key.clone()))
+        .map_err(|err| Error::other(format!("credentials: {err:?}")))?;
+
+    if let Some(region_str) = &config.region {
+        region_str
+            .parse::<s3s::region::Region>()
+            .map(crate::storage_compat::set_global_region)
+            .map_err(|err| Error::other(format!("invalid region '{region_str}': {err}")))?;
+    }
+
+    set_global_rustfs_port(server_addr.port());
+    set_global_addr(&config.address).await;
+
+    Ok(EmbeddedStartupListenContext {
+        readiness,
+        server_addr,
+        server_address: server_addr.to_string(),
+    })
+}
+
 pub async fn init_startup_http_servers(config: &Config, readiness: Arc<GlobalReadiness>) -> Result<StartupHttpServers> {
    init_capacity_management().await;
    let state_manager = Arc::new(ServiceStateManager::new());
@@ -28,12 +28,16 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, error, info, warn};

 const LOG_COMPONENT_MAIN: &str = "main";
+const LOG_COMPONENT_EMBEDDED: &str = "embedded";
 const LOG_SUBSYSTEM_STORAGE: &str = "storage";
+const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded";
 const GLOBAL_CONFIG_INIT_MAX_RETRIES: usize = 15;
 const EVENT_ENDPOINT_PARSING_STARTED: &str = "endpoint_parsing_started";
 const EVENT_STARTUP_STORAGE_STAGE: &str = "startup_storage_stage";
 const EVENT_STORAGE_POOL_FORMATTING: &str = "storage_pool_formatting";
 const EVENT_STORAGE_POOL_HOST_RISK: &str = "storage_pool_host_risk";
+const EVENT_EMBEDDED_STORAGE_INIT_FAILED: &str = "embedded_storage_init_failed";
+const EVENT_EMBEDDED_STORAGE_INIT_RETRY: &str = "embedded_storage_init_retry";

 pub struct StartupStorageRuntime {
    pub store: Arc<ECStore>,
@@ -102,6 +106,23 @@ pub async fn init_startup_storage_foundation(server_address: &str, volumes: &[St
    Ok(endpoint_pools)
 }

+pub async fn init_embedded_startup_storage_foundation(server_address: &str, volumes: &[String]) -> Result<EndpointServerPools> {
+    let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address, volumes.to_vec())
+        .await
+        .map_err(|err| Error::other(format!("endpoints: {err}")))?;
+    enforce_unsupported_fs_policy(&endpoint_pools).map_err(|err| Error::other(format!("unsupported fs guard: {err}")))?;
+
+    set_global_endpoints(endpoint_pools.as_ref().clone());
+    update_erasure_type(setup_type).await;
+
+    init_local_disks(endpoint_pools.clone())
+        .await
+        .map_err(|err| Error::other(format!("local disks: {err}")))?;
+    init_lock_clients(endpoint_pools.clone());
+
+    Ok(endpoint_pools)
+}
+
 pub async fn init_startup_storage_runtime(
    server_addr: SocketAddr,
    endpoint_pools: &EndpointServerPools,
@@ -143,6 +164,34 @@ pub async fn init_startup_storage_runtime(
    })
 }

+pub async fn init_embedded_startup_storage_runtime(
+    server_addr: SocketAddr,
+    endpoint_pools: &EndpointServerPools,
+    readiness: Arc<GlobalReadiness>,
+    shutdown_token: CancellationToken,
+) -> Result<StartupStorageRuntime> {
+    let store = match ECStore::new(server_addr, endpoint_pools.clone(), shutdown_token.clone()).await {
+        Ok(store) => store,
+        Err(err) => {
+            error!(
+                component = LOG_COMPONENT_EMBEDDED,
+                subsystem = LOG_SUBSYSTEM_EMBEDDED,
+                event = EVENT_EMBEDDED_STORAGE_INIT_FAILED,
+                stage = "ecstore_new",
+                error = ?err,
+                "Embedded storage initialization failed"
+            );
+            return Err(Error::other(format!("ECStore: {err}")));
+        }
+    };
+
+    init_embedded_startup_storage_global_config(store.clone()).await?;
+    readiness.mark_stage(SystemStage::StorageReady);
+    init_background_replication(store.clone()).await;
+
+    Ok(StartupStorageRuntime { store, shutdown_token })
+}
+
 async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
    init_ecstore_config();
    try_migrate_server_config(store.clone()).await;
@@ -172,6 +221,33 @@ async fn init_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
    Ok(())
 }

+async fn init_embedded_startup_storage_global_config(store: Arc<ECStore>) -> Result<()> {
+    init_ecstore_config();
+    try_migrate_server_config(store.clone()).await;
+
+    let mut retry = 0;
+    while let Err(err) = init_global_config_sys(store.clone()).await {
+        retry += 1;
+        if retry > GLOBAL_CONFIG_INIT_MAX_RETRIES {
+            return Err(Error::other(format!(
+                "init_global_config_sys failed after {GLOBAL_CONFIG_INIT_MAX_RETRIES} retries: {err}"
+            )));
+        }
+        debug!(
+            component = LOG_COMPONENT_EMBEDDED,
+            subsystem = LOG_SUBSYSTEM_EMBEDDED,
+            event = EVENT_EMBEDDED_STORAGE_INIT_RETRY,
+            stage = "global_config_sys",
+            retry,
+            error = %err,
+            "Embedded storage initialization retry scheduled"
+        );
+        tokio::time::sleep(tokio::time::Duration::from_secs(1)).await;
+    }
+
+    Ok(())
+}
+
 fn log_storage_pool_layout(endpoint_pools: &EndpointServerPools) {
    for (i, eps) in endpoint_pools.as_ref().iter().enumerate() {
        info!(