refactor: centralize startup protocol bootstrap (#3450)

rustfs/src/lib.rs

@@ -69,6 +69,7 @@ pub mod protocols;
 pub mod server;
 pub mod startup_fs_guard;
 pub mod startup_iam;
+pub mod startup_protocols;
 pub mod startup_services;
 pub mod storage;
 pub(crate) mod table_catalog;

rustfs/src/main.rs

@@ -17,15 +17,6 @@ use rustfs::init::{
     add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system, init_update_check, print_server_info,
 };
 
-#[cfg(feature = "ftps")]
-use rustfs::init::{init_ftp_system, init_ftps_system};
-
-#[cfg(feature = "webdav")]
-use rustfs::init::init_webdav_system;
-
-#[cfg(feature = "sftp")]
-use rustfs::init::init_sftp_system;
-
 use futures_util::future::join_all;
 use rustfs::capacity::capacity_integration::init_capacity_management;
 use rustfs::license::{init_license, license_status};
@@ -35,6 +26,7 @@ use rustfs::server::{
 };
 use rustfs::startup_fs_guard::enforce_unsupported_fs_policy;
 use rustfs::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap};
+use rustfs::startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders};
 use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr};
 use rustfs_credentials::init_global_action_credentials;
 use rustfs_ecstore::store::init_lock_clients;
@@ -606,173 +598,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
     // Initialize KMS system if enabled
     init_kms_system(&config).await?;
 
-    // Initialize FTP system if enabled
-    #[cfg(feature = "ftps")]
-    let ftp_shutdown_tx = match init_ftp_system().await {
-        Ok(Some(tx)) => {
-            debug!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftp",
-                state = "started",
-                "Protocol runtime started"
-            );
-            Some(tx)
-        }
-        Ok(None) => {
-            info!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftp",
-                state = "disabled",
-                "Protocol runtime disabled"
-            );
-            None
-        }
-        Err(e) => {
-            error!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftp",
-                state = "initialization_failed",
-                error = %e,
-                "Protocol runtime initialization failed"
-            );
-            return Err(Error::other(e));
-        }
-    };
-
-    #[cfg(not(feature = "ftps"))]
-    let ftp_shutdown_tx: Option<ShutdownHandle> = None;
-
-    // Initialize FTPS system if enabled
-    #[cfg(feature = "ftps")]
-    let ftps_shutdown_tx = match init_ftps_system().await {
-        Ok(Some(tx)) => {
-            debug!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftps",
-                state = "started",
-                "Protocol runtime started"
-            );
-            Some(tx)
-        }
-        Ok(None) => {
-            info!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftps",
-                state = "disabled",
-                "Protocol runtime disabled"
-            );
-            None
-        }
-        Err(e) => {
-            error!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "ftps",
-                state = "initialization_failed",
-                error = %e,
-                "Protocol runtime initialization failed"
-            );
-            return Err(Error::other(e));
-        }
-    };
-
-    #[cfg(not(feature = "ftps"))]
-    let ftps_shutdown_tx: Option<ShutdownHandle> = None;
-
-    // Initialize WebDAV system if enabled
-    #[cfg(feature = "webdav")]
-    let webdav_shutdown_tx = match init_webdav_system().await {
-        Ok(Some(tx)) => {
-            debug!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "webdav",
-                state = "started",
-                "Protocol runtime started"
-            );
-            Some(tx)
-        }
-        Ok(None) => {
-            info!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "webdav",
-                state = "disabled",
-                "Protocol runtime disabled"
-            );
-            None
-        }
-        Err(e) => {
-            error!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "webdav",
-                state = "initialization_failed",
-                error = %e,
-                "Protocol runtime initialization failed"
-            );
-            return Err(Error::other(e));
-        }
-    };
-
-    #[cfg(not(feature = "webdav"))]
-    let webdav_shutdown_tx: Option<ShutdownHandle> = None;
-
-    // Initialize SFTP system if enabled
-    #[cfg(feature = "sftp")]
-    let sftp_shutdown_tx = match init_sftp_system().await {
-        Ok(Some(tx)) => {
-            debug!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "sftp",
-                state = "started",
-                "Protocol runtime started"
-            );
-            Some(tx)
-        }
-        Ok(None) => {
-            info!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "sftp",
-                state = "disabled",
-                "Protocol runtime disabled"
-            );
-            None
-        }
-        Err(e) => {
-            error!(
-                event = EVENT_PROTOCOL_SYSTEM_STATE,
-                component = LOG_COMPONENT_MAIN,
-                subsystem = LOG_SUBSYSTEM_STARTUP,
-                protocol = "sftp",
-                state = "initialization_failed",
-                error = %e,
-                "Protocol runtime initialization failed"
-            );
-            return Err(Error::other(e));
-        }
-    };
-
-    #[cfg(not(feature = "sftp"))]
-    let sftp_shutdown_tx: Option<ShutdownHandle> = None;
+    let protocol_shutdowns = init_protocol_shutdown_senders().await?;
 
     // Initialize buffer profiling system
     init_buffer_profile_system(&config);
@@ -979,12 +805,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> {
         shutdown_signal,
         s3_shutdown_tx,
         console_shutdown_tx,
-        ProtocolShutdownSenders {
-            ftp: ftp_shutdown_tx,
-            ftps: ftps_shutdown_tx,
-            webdav: webdav_shutdown_tx,
-            sftp: sftp_shutdown_tx,
-        },
+        protocol_shutdowns,
         ctx.clone(),
     )
     .await;
@@ -1027,15 +848,6 @@ fn log_external_prefix_compat_report(report: &ExternalEnvCompatReport) {
     }
 }
 
-/// Shutdown channels for every protocol server. None means the protocol was
-/// disabled at startup.
-struct ProtocolShutdownSenders {
-    ftp: Option<ShutdownHandle>,
-    ftps: Option<ShutdownHandle>,
-    webdav: Option<ShutdownHandle>,
-    sftp: Option<ShutdownHandle>,
-}
-
 #[derive(Debug, Clone, Copy, PartialEq, Eq)]
 enum BackgroundShutdownStep {
     DataScanner,

rustfs/src/startup_protocols.rs

@@ -0,0 +1,153 @@
+// 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.
+
+#[cfg(feature = "sftp")]
+use crate::init::init_sftp_system;
+#[cfg(feature = "webdav")]
+use crate::init::init_webdav_system;
+#[cfg(feature = "ftps")]
+use crate::init::{init_ftp_system, init_ftps_system};
+use crate::server::ShutdownHandle;
+use std::future::Future;
+use std::io::{Error, Result};
+use tracing::{debug, error, info};
+
+const LOG_COMPONENT_MAIN: &str = "main";
+const LOG_SUBSYSTEM_STARTUP: &str = "startup";
+const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
+
+type ProtocolInitResult = std::result::Result<Option<ShutdownHandle>, Box<dyn std::error::Error + Send + Sync>>;
+
+/// Shutdown channels for every protocol server. None means the protocol was
+/// disabled at startup or not compiled in.
+pub struct ProtocolShutdownSenders {
+    pub ftp: Option<ShutdownHandle>,
+    pub ftps: Option<ShutdownHandle>,
+    pub webdav: Option<ShutdownHandle>,
+    pub sftp: Option<ShutdownHandle>,
+}
+
+pub async fn init_protocol_shutdown_senders() -> Result<ProtocolShutdownSenders> {
+    Ok(ProtocolShutdownSenders {
+        ftp: init_ftp_protocol().await?,
+        ftps: init_ftps_protocol().await?,
+        webdav: init_webdav_protocol().await?,
+        sftp: init_sftp_protocol().await?,
+    })
+}
+
+#[cfg(feature = "ftps")]
+async fn init_ftp_protocol() -> Result<Option<ShutdownHandle>> {
+    init_protocol("ftp", init_ftp_system).await
+}
+
+#[cfg(not(feature = "ftps"))]
+async fn init_ftp_protocol() -> Result<Option<ShutdownHandle>> {
+    Ok(None)
+}
+
+#[cfg(feature = "ftps")]
+async fn init_ftps_protocol() -> Result<Option<ShutdownHandle>> {
+    init_protocol("ftps", init_ftps_system).await
+}
+
+#[cfg(not(feature = "ftps"))]
+async fn init_ftps_protocol() -> Result<Option<ShutdownHandle>> {
+    Ok(None)
+}
+
+#[cfg(feature = "webdav")]
+async fn init_webdav_protocol() -> Result<Option<ShutdownHandle>> {
+    init_protocol("webdav", init_webdav_system).await
+}
+
+#[cfg(not(feature = "webdav"))]
+async fn init_webdav_protocol() -> Result<Option<ShutdownHandle>> {
+    Ok(None)
+}
+
+#[cfg(feature = "sftp")]
+async fn init_sftp_protocol() -> Result<Option<ShutdownHandle>> {
+    init_protocol("sftp", init_sftp_system).await
+}
+
+#[cfg(not(feature = "sftp"))]
+async fn init_sftp_protocol() -> Result<Option<ShutdownHandle>> {
+    Ok(None)
+}
+
+async fn init_protocol<InitFn, InitFuture>(protocol: &'static str, init: InitFn) -> Result<Option<ShutdownHandle>>
+where
+    InitFn: FnOnce() -> InitFuture,
+    InitFuture: Future<Output = ProtocolInitResult>,
+{
+    match init().await {
+        Ok(Some(tx)) => {
+            debug!(
+                event = EVENT_PROTOCOL_SYSTEM_STATE,
+                component = LOG_COMPONENT_MAIN,
+                subsystem = LOG_SUBSYSTEM_STARTUP,
+                protocol = protocol,
+                state = "started",
+                "Protocol runtime started"
+            );
+            Ok(Some(tx))
+        }
+        Ok(None) => {
+            info!(
+                event = EVENT_PROTOCOL_SYSTEM_STATE,
+                component = LOG_COMPONENT_MAIN,
+                subsystem = LOG_SUBSYSTEM_STARTUP,
+                protocol = protocol,
+                state = "disabled",
+                "Protocol runtime disabled"
+            );
+            Ok(None)
+        }
+        Err(err) => {
+            error!(
+                event = EVENT_PROTOCOL_SYSTEM_STATE,
+                component = LOG_COMPONENT_MAIN,
+                subsystem = LOG_SUBSYSTEM_STARTUP,
+                protocol = protocol,
+                state = "initialization_failed",
+                error = %err,
+                "Protocol runtime initialization failed"
+            );
+            Err(Error::other(err))
+        }
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::init_protocol;
+
+    #[tokio::test]
+    async fn init_protocol_returns_none_when_disabled() {
+        let shutdown = init_protocol("test", || async { Ok(None) }).await;
+
+        assert!(matches!(shutdown, Ok(None)));
+    }
+
+    #[tokio::test]
+    async fn init_protocol_maps_startup_error() {
+        let shutdown = init_protocol("test", || async {
+            Err(Box::<dyn std::error::Error + Send + Sync>::from("startup failed"))
+        })
+        .await;
+
+        assert!(shutdown.is_err());
+    }
+}