refactor: extract embedded startup control helpers (#3661)

rustfs/src/embedded.rs

@@ -47,22 +47,18 @@
 //! start a second server will return an error.
 
 use crate::server::ShutdownHandle;
-use crate::startup_lifecycle::{log_embedded_server_ready, publish_embedded_startup_ready};
+use crate::startup_lifecycle::{EmbeddedStartupGuard, log_embedded_server_ready, publish_embedded_startup_ready};
 use crate::startup_runtime_hooks::init_embedded_runtime_hooks;
 use crate::startup_server::{
     EmbeddedStartupConfig, init_embedded_startup_listen_context, prepare_embedded_startup_config, start_embedded_http_server,
 };
 use crate::startup_services::init_embedded_startup_runtime_services;
-use crate::startup_shutdown::run_embedded_server_shutdown;
+use crate::startup_shutdown::{run_embedded_server_shutdown, signal_embedded_startup_shutdown};
 use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime};
 use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
 use std::path::PathBuf;
-use std::sync::atomic::{AtomicBool, Ordering};
 use tokio_util::sync::CancellationToken;
 
-/// Tracks whether a server has been started in this process.
-static SERVER_STARTED: AtomicBool = AtomicBool::new(false);
-
 /// Error type for embedded server operations.
 #[derive(Debug)]
 pub enum ServerError {
@@ -223,20 +219,7 @@ impl RustFSServerBuilder {
         // Build is allowed to fail before irreversible global initialization
         // (for example on temporary I/O or directory setup errors), and in that
         // case callers can retry.
-        let mut global_init_started = false;
-        let mut set_global_init_guard = || -> Result<(), ServerError> {
-            if global_init_started {
-                return Ok(());
-            }
-            if SERVER_STARTED
-                .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
-                .is_err()
-            {
-                return Err(ServerError::AlreadyStarted);
-            }
-            global_init_started = true;
-            Ok(())
-        };
+        let mut startup_guard = EmbeddedStartupGuard::new();
 
         let EmbeddedStartupConfig { config, temp_dir_guard } = prepare_embedded_startup_config(
             self.address.clone(),
@@ -258,7 +241,9 @@ impl RustFSServerBuilder {
             .await
             .map_err(|e| ServerError::Init(e.to_string()))?;
 
-        set_global_init_guard()?;
+        startup_guard
+            .mark_global_init_started()
+            .map_err(|_| ServerError::AlreadyStarted)?;
 
         let endpoint_pools = init_embedded_startup_storage_foundation(&listen_context.server_address, &config.volumes)
             .await
@@ -268,11 +253,6 @@ impl RustFSServerBuilder {
         let shutdown_handle = http_server.shutdown_handle;
         let bound_addr = http_server.bound_addr;
         let ctx = CancellationToken::new();
-        let shutdown_embedded_server = || {
-            shutdown_handle.signal();
-            ctx.cancel();
-        };
-
         let storage_runtime = match init_embedded_startup_storage_runtime(
             listen_context.server_addr,
             &endpoint_pools,
@@ -283,7 +263,7 @@ impl RustFSServerBuilder {
         {
             Ok(runtime) => runtime,
             Err(e) => {
-                shutdown_embedded_server();
+                signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
                 return Err(ServerError::Init(e.to_string()));
             }
         };
@@ -293,14 +273,14 @@ impl RustFSServerBuilder {
             init_embedded_startup_runtime_services(&config, endpoint_pools, store, ctx.clone(), listen_context.readiness.clone())
                 .await
                 .map_err(|e| {
-                    shutdown_embedded_server();
+                    signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
                     ServerError::Init(e.to_string())
                 })?;
 
         publish_embedded_startup_ready(service_runtime.iam_bootstrap, listen_context.readiness.as_ref())
             .await
             .map_err(|e| {
-                shutdown_embedded_server();
+                signal_embedded_startup_shutdown(&shutdown_handle, &ctx);
                 ServerError::Init(format!("runtime readiness: {e}"))
             })?;
 

rustfs/src/startup_lifecycle.rs

@@ -21,7 +21,14 @@ use crate::{
 };
 use rustfs_common::GlobalReadiness;
 use rustfs_scanner::init_data_scanner;
-use std::{io::Result, net::SocketAddr, sync::Arc};
+use std::{
+    io::Result,
+    net::SocketAddr,
+    sync::{
+        Arc,
+        atomic::{AtomicBool, Ordering},
+    },
+};
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 
@@ -33,6 +40,48 @@ const EVENT_SERVER_READY: &str = "server_ready";
 const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state";
 const EVENT_EMBEDDED_SERVER_STATE: &str = "embedded_server_state";
 
+static EMBEDDED_SERVER_STARTED: AtomicBool = AtomicBool::new(false);
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) struct EmbeddedStartupAlreadyStarted;
+
+pub(crate) struct EmbeddedStartupGuard {
+    global_init_started: bool,
+}
+
+impl EmbeddedStartupGuard {
+    pub(crate) fn new() -> Self {
+        Self {
+            global_init_started: false,
+        }
+    }
+
+    pub(crate) fn mark_global_init_started(&mut self) -> std::result::Result<(), EmbeddedStartupAlreadyStarted> {
+        mark_embedded_global_init_started(&EMBEDDED_SERVER_STARTED, &mut self.global_init_started)
+    }
+}
+
+impl Default for EmbeddedStartupGuard {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+fn mark_embedded_global_init_started(
+    server_started: &AtomicBool,
+    global_init_started: &mut bool,
+) -> std::result::Result<(), EmbeddedStartupAlreadyStarted> {
+    if *global_init_started {
+        return Ok(());
+    }
+
+    server_started
+        .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst)
+        .map_err(|_| EmbeddedStartupAlreadyStarted)?;
+    *global_init_started = true;
+    Ok(())
+}
+
 pub struct StartupRuntimeLifecycle {
     pub server_address: String,
     pub state_manager: Arc<ServiceStateManager>,
@@ -119,3 +168,34 @@ pub fn log_embedded_server_ready(endpoint_address: SocketAddr) {
         endpoint_address
     );
 }
+
+#[cfg(test)]
+mod tests {
+    use super::mark_embedded_global_init_started;
+    use std::sync::atomic::{AtomicBool, Ordering};
+
+    #[test]
+    fn embedded_global_init_guard_allows_local_retry_before_mark() {
+        let server_started = AtomicBool::new(false);
+        let mut global_init_started = false;
+
+        mark_embedded_global_init_started(&server_started, &mut global_init_started)
+            .expect("first irreversible startup should mark global init");
+        mark_embedded_global_init_started(&server_started, &mut global_init_started)
+            .expect("repeated mark in same startup should be idempotent");
+
+        assert!(global_init_started);
+        assert!(server_started.load(Ordering::SeqCst));
+    }
+
+    #[test]
+    fn embedded_global_init_guard_rejects_second_startup_after_mark() {
+        let server_started = AtomicBool::new(true);
+        let mut global_init_started = false;
+
+        let result = mark_embedded_global_init_started(&server_started, &mut global_init_started);
+
+        assert!(result.is_err());
+        assert!(!global_init_started);
+    }
+}

rustfs/src/startup_shutdown.rs

@@ -216,6 +216,11 @@ pub async fn run_embedded_shutdown_cleanup() {
     }
 }
 
+pub(crate) fn signal_embedded_startup_shutdown(shutdown_handle: &ShutdownHandle, ctx: &CancellationToken) {
+    shutdown_handle.signal();
+    ctx.cancel();
+}
+
 pub async fn run_embedded_server_shutdown(
     ctx: &CancellationToken,
     shutdown_handle: &mut Option<ShutdownHandle>,
@@ -264,7 +269,11 @@ pub async fn run_embedded_server_shutdown(
 
 #[cfg(test)]
 mod tests {
-    use super::{BackgroundShutdownStep, background_shutdown_steps};
+    use super::{BackgroundShutdownStep, background_shutdown_steps, signal_embedded_startup_shutdown};
+    use crate::server::ShutdownHandle;
+    use std::time::Duration;
+    use tokio::sync::broadcast;
+    use tokio_util::sync::CancellationToken;
 
     #[test]
     fn background_shutdown_plan_keeps_scanner_before_ahm() {
@@ -279,4 +288,21 @@ mod tests {
         assert_eq!(background_shutdown_steps(false, true), vec![BackgroundShutdownStep::Ahm]);
         assert!(background_shutdown_steps(false, false).is_empty());
     }
+
+    #[tokio::test]
+    async fn signal_embedded_startup_shutdown_signals_handle_and_cancels_token() {
+        let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
+        let shutdown_task = tokio::spawn(async move {
+            let _ = shutdown_rx.recv().await;
+        });
+        let shutdown_handle = ShutdownHandle::new(shutdown_tx, shutdown_task);
+        let cancel_token = CancellationToken::new();
+
+        signal_embedded_startup_shutdown(&shutdown_handle, &cancel_token);
+
+        tokio::time::timeout(Duration::from_secs(1), shutdown_handle.wait())
+            .await
+            .expect("shutdown task should observe startup signal");
+        assert!(cancel_token.is_cancelled());
+    }
 }