refactor: extract embedded handle boundaries (#3664)

rustfs/src/embedded.rs

@@ -48,9 +48,9 @@
 
 use crate::server::ShutdownHandle;
 use crate::startup_embedded::{EmbeddedStartupArgs, EmbeddedStartupError, run_embedded_startup};
-use crate::startup_lifecycle::log_embedded_server_ready;
-use crate::startup_shutdown::run_embedded_server_shutdown;
-use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr};
+use crate::startup_lifecycle::{embedded_endpoint_address, log_embedded_server_ready};
+use crate::startup_shutdown::{run_embedded_server_drop_cleanup, run_embedded_server_shutdown};
+use std::net::SocketAddr;
 use std::path::PathBuf;
 use tokio_util::sync::CancellationToken;
 
@@ -263,14 +263,7 @@ pub struct RustFSServer {
 
 impl RustFSServer {
     fn endpoint_address(&self) -> SocketAddr {
-        let ip = match self.address.ip() {
-            ip @ IpAddr::V4(v4) if !v4.is_unspecified() => ip,
-            IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
-            ip @ IpAddr::V6(v6) if !v6.is_unspecified() => ip,
-            IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
-        };
-
-        SocketAddr::new(ip, self.address.port())
+        embedded_endpoint_address(self.address)
     }
 
     /// The HTTP endpoint URL (e.g. `"http://127.0.0.1:54321"`).
@@ -312,14 +305,7 @@ impl RustFSServer {
 
 impl Drop for RustFSServer {
     fn drop(&mut self) {
-        // Best-effort synchronous cleanup.
-        self.cancel_token.cancel();
-        if let Some(shutdown_handle) = self.shutdown_handle.take() {
-            shutdown_handle.signal();
-        }
-        if let Some(ref dir) = self.temp_dir {
-            let _ = std::fs::remove_dir_all(dir);
-        }
+        run_embedded_server_drop_cleanup(&self.cancel_token, &mut self.shutdown_handle, self.temp_dir.as_deref());
     }
 }
 

rustfs/src/startup_lifecycle.rs

@@ -23,7 +23,7 @@ use rustfs_common::GlobalReadiness;
 use rustfs_scanner::init_data_scanner;
 use std::{
     io::Result,
-    net::SocketAddr,
+    net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
     sync::{
         Arc,
         atomic::{AtomicBool, Ordering},
@@ -157,6 +157,17 @@ pub async fn publish_embedded_startup_ready(iam_bootstrap: IamBootstrapDispositi
     Ok(())
 }
 
+pub(crate) fn embedded_endpoint_address(address: SocketAddr) -> SocketAddr {
+    let ip = match address.ip() {
+        ip @ IpAddr::V4(v4) if !v4.is_unspecified() => ip,
+        IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST),
+        ip @ IpAddr::V6(v6) if !v6.is_unspecified() => ip,
+        IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST),
+    };
+
+    SocketAddr::new(ip, address.port())
+}
+
 pub fn log_embedded_server_ready(endpoint_address: SocketAddr) {
     info!(
         target: "rustfs::embedded",
@@ -171,8 +182,11 @@ pub fn log_embedded_server_ready(endpoint_address: SocketAddr) {
 
 #[cfg(test)]
 mod tests {
-    use super::mark_embedded_global_init_started;
-    use std::sync::atomic::{AtomicBool, Ordering};
+    use super::{embedded_endpoint_address, mark_embedded_global_init_started};
+    use std::{
+        net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr},
+        sync::atomic::{AtomicBool, Ordering},
+    };
 
     #[test]
     fn embedded_global_init_guard_allows_local_retry_before_mark() {
@@ -198,4 +212,28 @@ mod tests {
         assert!(result.is_err());
         assert!(!global_init_started);
     }
+
+    #[test]
+    fn embedded_endpoint_address_rewrites_unspecified_hosts() {
+        assert_eq!(
+            embedded_endpoint_address(SocketAddr::new(IpAddr::V4(Ipv4Addr::UNSPECIFIED), 9000)),
+            SocketAddr::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9000)
+        );
+        assert_eq!(
+            embedded_endpoint_address(SocketAddr::new(IpAddr::V6(Ipv6Addr::UNSPECIFIED), 9001)),
+            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 9001)
+        );
+    }
+
+    #[test]
+    fn embedded_endpoint_address_preserves_bound_hosts() {
+        assert_eq!(
+            embedded_endpoint_address(SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 10)), 9000)),
+            SocketAddr::new(IpAddr::V4(Ipv4Addr::new(192, 0, 2, 10)), 9000)
+        );
+        assert_eq!(
+            embedded_endpoint_address(SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 9001)),
+            SocketAddr::new(IpAddr::V6(Ipv6Addr::LOCALHOST), 9001)
+        );
+    }
 }

rustfs/src/startup_shutdown.rs

@@ -221,6 +221,20 @@ pub(crate) fn signal_embedded_startup_shutdown(shutdown_handle: &ShutdownHandle,
     ctx.cancel();
 }
 
+pub(crate) fn run_embedded_server_drop_cleanup(
+    ctx: &CancellationToken,
+    shutdown_handle: &mut Option<ShutdownHandle>,
+    temp_dir: Option<&Path>,
+) {
+    ctx.cancel();
+    if let Some(shutdown_handle) = shutdown_handle.take() {
+        shutdown_handle.signal();
+    }
+    if let Some(dir) = temp_dir {
+        let _ = std::fs::remove_dir_all(dir);
+    }
+}
+
 pub async fn run_embedded_server_shutdown(
     ctx: &CancellationToken,
     shutdown_handle: &mut Option<ShutdownHandle>,
@@ -269,7 +283,9 @@ pub async fn run_embedded_server_shutdown(
 
 #[cfg(test)]
 mod tests {
-    use super::{BackgroundShutdownStep, background_shutdown_steps, signal_embedded_startup_shutdown};
+    use super::{
+        BackgroundShutdownStep, background_shutdown_steps, run_embedded_server_drop_cleanup, signal_embedded_startup_shutdown,
+    };
     use crate::server::ShutdownHandle;
     use std::time::Duration;
     use tokio::sync::broadcast;
@@ -305,4 +321,29 @@ mod tests {
             .expect("shutdown task should observe startup signal");
         assert!(cancel_token.is_cancelled());
     }
+
+    #[tokio::test]
+    async fn embedded_drop_cleanup_signals_handle_cancels_token_and_removes_temp_dir() {
+        let (shutdown_tx, mut shutdown_rx) = broadcast::channel(1);
+        let (observed_tx, observed_rx) = tokio::sync::oneshot::channel();
+        let shutdown_task = tokio::spawn(async move {
+            let _ = shutdown_rx.recv().await;
+            let _ = observed_tx.send(());
+        });
+        let shutdown_handle = ShutdownHandle::new(shutdown_tx, shutdown_task);
+        let mut shutdown_handle = Some(shutdown_handle);
+        let cancel_token = CancellationToken::new();
+        let temp_dir = tempfile::tempdir().expect("temp dir should create");
+        let temp_path = temp_dir.path().to_path_buf();
+
+        run_embedded_server_drop_cleanup(&cancel_token, &mut shutdown_handle, Some(temp_dir.path()));
+
+        tokio::time::timeout(Duration::from_secs(1), observed_rx)
+            .await
+            .expect("drop cleanup should signal shutdown")
+            .expect("shutdown signal should be delivered");
+        assert!(cancel_token.is_cancelled());
+        assert!(shutdown_handle.is_none());
+        assert!(!temp_path.exists());
+    }
 }