refactor: extract optional runtime shutdown boundary (#3632)

docs/architecture/migration-progress.md

@@ -5,15 +5,15 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 ## Current Context
 
 - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660)
-- Branch: `overtrue/arch-ops-profiler-capability-tests`
+- Branch: `overtrue/arch-optional-runtime-sidecar-boundary`
 - Baseline: latest `origin/main`
-  (`71809ba02ea405c3a2f0e4cb82ff608915dd519c`).
+  (`e6391598f0b62a66ca6eac405c26b53a9a77d5f1`).
-- PR type for this branch: `contract`
+- PR type for this branch: `pure-move`
 - Runtime behavior changes: none.
-- Rust code changes: add the `ops.profiler.v1` extension capability snapshot
+- Rust code changes: move optional protocol shutdown ownership from
-  contract for disabled, unsupported, and enabled profiler runtime states.
+  `startup_services` into the optional runtime boundary.
 - CI/script changes: none.
-- Docs changes: record the X-013 profiler capability snapshot slice.
+- Docs changes: record the R-021 optional runtime shutdown boundary slice.
 
 ## Phase 0 Tasks
 
@@ -2016,6 +2016,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
     guards, diff hygiene, Rust risk scan, branch freshness check, and
     pre-commit quality gate.
 
+- [x] `R-021` Extract optional runtime shutdown boundary.
+  - Do: add `startup_optional_runtimes` and move optional protocol shutdown
+    ownership/logging out of `startup_services`.
+  - Acceptance: optional protocol shutdown plan order stays FTP, FTPS, WebDAV,
+    SFTP; stopping logs remain before event notifier/audit/profiling shutdown;
+    signal/wait remains after S3/console HTTP shutdown; later optional
+    sidecars have an explicit owner without startup behavior changes.
+  - Must preserve: protocol initialization, protocol shutdown signaling and
+    waiting, shutdown order, profiling/audit/event notifier shutdown, HTTP
+    shutdown, readiness state, and fatal boundaries.
+  - Verification: focused startup optional runtime/service tests, RustFS lib
+    check, migration/layer guards, formatting, diff hygiene, Rust risk scan,
+    branch freshness check, pre-commit quality gate, and three-expert review.
+
 ## Next PRs
 
 1. `pure-move`: continue larger lifecycle hook slices for optional runtime
@@ -2027,9 +2041,9 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block
 
 | Expert | Status | Notes |
 |---|---|---|
-| Quality/architecture | passed | X-013 keeps the capability snapshot contract in `rustfs-extension-schema`, reuses explicit profiler backend/runtime DTOs, and adds no runtime dependency edges. |
+| Quality/architecture | passed | R-021 isolates optional runtime shutdown ownership in `startup_optional_runtimes` and keeps `startup_services` as the startup/shutdown orchestrator. |
-| Migration preservation | passed | Runtime profiling, admin pprof routes, exporters, storage paths, telemetry, sidecar startup, and plugin execution are untouched; validation only rejects unsafe snapshot declarations. |
+| Migration preservation | passed | Protocol startup, shutdown order, readiness, HTTP shutdown, event notifier, audit, and profiling behavior remain unchanged. |
-| Testing/verification | passed | Extension schema tests cover enabled, disabled, unsupported, snapshot JSON shape, disabled external runtime policy, and startup-fatal rejection; focused checks, guards, formatting, diff hygiene, and final pre-commit passed. |
+| Testing/verification | passed | Focused optional runtime and startup service tests pass; remaining checks and final pre-commit are tracked below before push. |
 
 ## Verification Notes
 
@@ -2057,6 +2071,19 @@ Passed before push:
   - `make pre-commit`: passed.
   - Three-expert review: passed.
 
+- Issue #660 R-021 current slice:
+  - `cargo test -p rustfs --lib startup_optional_runtimes -- --nocapture`:
+    passed.
+  - `cargo test -p rustfs --lib startup_services -- --nocapture`: passed.
+  - `cargo check -p rustfs --lib`: passed.
+  - `./scripts/check_architecture_migration_rules.sh`: passed.
+  - `./scripts/check_layer_dependencies.sh`: passed.
+  - `cargo fmt --all --check`: passed.
+  - `git diff --check`: passed.
+  - Rust risk scan on changed Rust files: passed.
+  - `make pre-commit`: passed.
+  - Three-expert review: passed.
+
 - Issue #660 R-020 current slice:
   - `cargo test -p rustfs --lib startup_profiling -- --nocapture`: passed.
   - `cargo check -p rustfs --lib`: passed.

docs/architecture/profiling-numa-capability-inventory.md

@@ -80,9 +80,9 @@ Future sidecars for profiling, eBPF, or NUMA must preserve these invariants:
 - Verify optional profiler sidecar and Wasm runtimes stay disabled by default
   and cannot declare a startup fatal boundary.
 
-`R-017`:
+`R-021`:
 
-- If a runtime service sidecar is added later, start it from the startup service
+- If a runtime service sidecar is added later, enter it through the optional
-  boundary with explicit shutdown ownership.
+  runtime boundary with explicit shutdown ownership.
 - Preserve current service order, KMS/audit/notification fatal boundaries, and
   scanner/heal startup semantics.

docs/architecture/runtime-lifecycle.md

@@ -31,6 +31,14 @@ Disallowed responsibilities:
 - Hide globals behind a service-locator API.
 - Change startup side effects while moving code.
 
+## Optional Runtime Boundary
+
+`startup_optional_runtimes` owns shutdown for optional runtime services that are
+not readiness or fatal startup boundaries. The current owner set is protocol
+servers only. Future optional sidecars must enter this boundary with explicit
+shutdown handles and status snapshots instead of adding ad hoc shutdown work to
+`startup_services`.
+
 ## AppContext Foundation
 
 Early AppContext work should split resolver files and add compatibility tests before

docs/architecture/startup-timeline.md

@@ -71,9 +71,9 @@ new startup semantics.
 |---|---|---|---|---|---|
 | `STOP-001` | `rustfs/src/startup_services.rs` | Cancel runtime token and move service state to `Stopping`. | Notifies cancellation-aware background tasks. | Non-fatal. | Service state moves to `Stopping`; readiness stages are not cleared here |
 | `STOP-002` | `rustfs/src/startup_services.rs` | Stop scanner/background services and AHM services according to enable flags. | Calls ECStore background shutdown and heal/scanner shutdown helpers. | Non-fatal in this path. | No readiness-stage change |
-| `STOP-003` | `rustfs/src/startup_services.rs` | Signal optional FTP/FTPS/WebDAV/SFTP protocol servers. | Collects protocol shutdown futures. | Non-fatal in this path. | No readiness-stage change |
+| `STOP-003` | `rustfs/src/startup_optional_runtimes.rs` | Plan optional runtime shutdown and log stopping state for FTP/FTPS/WebDAV/SFTP protocol servers. | Collects protocol shutdown handles. | Non-fatal in this path. | No readiness-stage change |
 | `STOP-004` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_profiling.rs` | Stop event notifier, audit system, and profiling tasks. | Stops notifier and profiling tasks; audit stop failures are logged. | Non-fatal in this path. | No readiness-stage change |
-| `STOP-005` | `rustfs/src/startup_services.rs` | Stop S3 and console HTTP servers, wait for protocol shutdowns, then mark service state `Stopped`. | HTTP shutdown happens after notifier/audit/profiling shutdown in current order. | Join failures are logged by shutdown handles; this path does not return errors. | Service state moves to `Stopped`; readiness stages are not cleared here |
+| `STOP-005` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_optional_runtimes.rs` | Stop S3 and console HTTP servers, signal and wait for optional protocol shutdowns, then mark service state `Stopped`. | HTTP shutdown happens after notifier/audit/profiling shutdown in current order. | Join failures are logged by shutdown handles; this path does not return errors. | Service state moves to `Stopped`; readiness stages are not cleared here |
 
 ## Migration Rules
 

rustfs/src/lib.rs

@@ -71,6 +71,7 @@ pub mod server;
 pub mod startup_entrypoint;
 pub mod startup_fs_guard;
 pub mod startup_iam;
+pub mod startup_optional_runtimes;
 pub mod startup_preflight;
 pub mod startup_profiling;
 pub mod startup_protocols;

rustfs/src/startup_optional_runtimes.rs

@@ -0,0 +1,174 @@
+// 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::{server::ShutdownHandle, startup_protocols::ProtocolShutdownSenders};
+use futures_util::future::join_all;
+use tracing::info;
+
+const LOG_COMPONENT_MAIN: &str = "main";
+const LOG_SUBSYSTEM_STARTUP: &str = "startup";
+const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
+
+pub struct OptionalRuntimeServices {
+    pub protocols: ProtocolShutdownSenders,
+}
+
+impl OptionalRuntimeServices {
+    pub fn new(protocols: ProtocolShutdownSenders) -> Self {
+        Self { protocols }
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+enum OptionalRuntimeShutdownStep {
+    Ftp,
+    Ftps,
+    Webdav,
+    Sftp,
+}
+
+impl OptionalRuntimeShutdownStep {
+    const fn protocol(self) -> &'static str {
+        match self {
+            Self::Ftp => "ftp",
+            Self::Ftps => "ftps",
+            Self::Webdav => "webdav",
+            Self::Sftp => "sftp",
+        }
+    }
+}
+
+#[cfg(test)]
+fn optional_runtime_shutdown_steps(protocols: &ProtocolShutdownSenders) -> Vec<OptionalRuntimeShutdownStep> {
+    let mut steps = Vec::with_capacity(4);
+    if protocols.ftp.is_some() {
+        steps.push(OptionalRuntimeShutdownStep::Ftp);
+    }
+    if protocols.ftps.is_some() {
+        steps.push(OptionalRuntimeShutdownStep::Ftps);
+    }
+    if protocols.webdav.is_some() {
+        steps.push(OptionalRuntimeShutdownStep::Webdav);
+    }
+    if protocols.sftp.is_some() {
+        steps.push(OptionalRuntimeShutdownStep::Sftp);
+    }
+    steps
+}
+
+pub fn prepare_optional_runtime_shutdowns(optional_runtimes: OptionalRuntimeServices) -> Vec<ShutdownHandle> {
+    let ProtocolShutdownSenders { ftp, ftps, webdav, sftp } = optional_runtimes.protocols;
+
+    let mut protocol_shutdowns = Vec::new();
+    if let Some(handle) = ftp {
+        log_protocol_stopping(OptionalRuntimeShutdownStep::Ftp);
+        protocol_shutdowns.push(handle);
+    }
+    if let Some(handle) = ftps {
+        log_protocol_stopping(OptionalRuntimeShutdownStep::Ftps);
+        protocol_shutdowns.push(handle);
+    }
+    if let Some(handle) = webdav {
+        log_protocol_stopping(OptionalRuntimeShutdownStep::Webdav);
+        protocol_shutdowns.push(handle);
+    }
+    if let Some(handle) = sftp {
+        log_protocol_stopping(OptionalRuntimeShutdownStep::Sftp);
+        protocol_shutdowns.push(handle);
+    }
+
+    protocol_shutdowns
+}
+
+pub async fn shutdown_optional_runtime_services(protocol_shutdowns: Vec<ShutdownHandle>) {
+    join_all(protocol_shutdowns.into_iter().map(ShutdownHandle::shutdown)).await;
+}
+
+fn log_protocol_stopping(step: OptionalRuntimeShutdownStep) {
+    info!(
+        target: "rustfs::main::handle_shutdown",
+        event = EVENT_PROTOCOL_SYSTEM_STATE,
+        component = LOG_COMPONENT_MAIN,
+        subsystem = LOG_SUBSYSTEM_STARTUP,
+        protocol = step.protocol(),
+        state = "stopping",
+        "Protocol runtime stopping"
+    );
+}
+
+#[cfg(test)]
+mod tests {
+    use super::{
+        OptionalRuntimeServices, OptionalRuntimeShutdownStep, optional_runtime_shutdown_steps,
+        prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
+    };
+    use crate::{server::ShutdownHandle, startup_protocols::ProtocolShutdownSenders};
+    use std::sync::{Arc, Mutex};
+    use tokio::sync::broadcast;
+
+    fn completed_shutdown_handle() -> ShutdownHandle {
+        let (tx, _rx) = broadcast::channel(1);
+        ShutdownHandle::new(tx, tokio::spawn(async {}))
+    }
+
+    fn observed_shutdown_handle(label: &'static str, events: Arc<Mutex<Vec<&'static str>>>) -> ShutdownHandle {
+        let (tx, mut rx) = broadcast::channel(1);
+        let task_handle = tokio::spawn(async move {
+            let _ = rx.recv().await;
+            events.lock().unwrap_or_else(|err| err.into_inner()).push(label);
+        });
+        ShutdownHandle::new(tx, task_handle)
+    }
+
+    #[tokio::test]
+    async fn optional_runtime_shutdown_plan_preserves_protocol_order() {
+        let protocols = ProtocolShutdownSenders {
+            ftp: Some(completed_shutdown_handle()),
+            ftps: Some(completed_shutdown_handle()),
+            webdav: Some(completed_shutdown_handle()),
+            sftp: Some(completed_shutdown_handle()),
+        };
+
+        assert_eq!(
+            optional_runtime_shutdown_steps(&protocols),
+            vec![
+                OptionalRuntimeShutdownStep::Ftp,
+                OptionalRuntimeShutdownStep::Ftps,
+                OptionalRuntimeShutdownStep::Webdav,
+                OptionalRuntimeShutdownStep::Sftp,
+            ]
+        );
+    }
+
+    #[tokio::test]
+    async fn optional_runtime_shutdown_signals_registered_protocols() {
+        let events = Arc::new(Mutex::new(Vec::new()));
+        let protocols = ProtocolShutdownSenders {
+            ftp: Some(observed_shutdown_handle("ftp", events.clone())),
+            ftps: None,
+            webdav: Some(observed_shutdown_handle("webdav", events.clone())),
+            sftp: None,
+        };
+
+        let protocol_shutdowns = prepare_optional_runtime_shutdowns(OptionalRuntimeServices::new(protocols));
+        tokio::task::yield_now().await;
+        assert!(events.lock().unwrap_or_else(|err| err.into_inner()).is_empty());
+
+        shutdown_optional_runtime_services(protocol_shutdowns).await;
+
+        let mut events = events.lock().unwrap_or_else(|err| err.into_inner()).clone();
+        events.sort_unstable();
+        assert_eq!(events, ["ftp", "webdav"]);
+    }
+}

rustfs/src/startup_services.rs

@@ -27,9 +27,11 @@ use crate::{
         start_audit_system, stop_audit_system, wait_for_shutdown,
     },
     startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap},
-    startup_protocols::{ProtocolShutdownSenders, init_protocol_shutdown_senders},
+    startup_optional_runtimes::{
+        OptionalRuntimeServices, prepare_optional_runtime_shutdowns, shutdown_optional_runtime_services,
+    },
+    startup_protocols::init_protocol_shutdown_senders,
 };
-use futures_util::future::join_all;
 use rustfs_audit::AuditResult;
 use rustfs_common::GlobalReadiness;
 use rustfs_heal::{
@@ -56,7 +58,6 @@ const LOG_COMPONENT_MAIN: &str = "main";
 const LOG_SUBSYSTEM_STARTUP: &str = "startup";
 const LOG_SUBSYSTEM_AUTH: &str = "auth";
 const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state";
-const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state";
 const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state";
 const EVENT_KEYSTONE_AUTH_INITIALIZED: &str = "keystone_auth_initialized";
 const EVENT_KEYSTONE_AUTH_INITIALIZATION_FAILED: &str = "keystone_auth_initialization_failed";
@@ -71,7 +72,7 @@ const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown";
 const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state";
 
 pub struct StartupServiceRuntime {
-    pub protocol_shutdowns: ProtocolShutdownSenders,
+    pub optional_runtimes: OptionalRuntimeServices,
     pub iam_bootstrap: IamBootstrapDisposition,
     pub enable_scanner: bool,
 }
@@ -98,6 +99,7 @@ pub async fn init_startup_runtime_services(
     init_kms_system(config).await?;
 
     let protocol_shutdowns = init_protocol_shutdown_senders().await?;
+    let optional_runtimes = OptionalRuntimeServices::new(protocol_shutdowns);
 
     init_buffer_profile_system(config);
     init_audit_runtime().await;
@@ -111,7 +113,7 @@ pub async fn init_startup_runtime_services(
     init_observability_runtime(ctx.clone()).await;
 
     Ok(StartupServiceRuntime {
-        protocol_shutdowns,
+        optional_runtimes,
         iam_bootstrap,
         enable_scanner,
     })
@@ -129,7 +131,7 @@ pub async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifecycle) -
         readiness,
     } = lifecycle;
     let StartupServiceRuntime {
-        protocol_shutdowns,
+        optional_runtimes,
         iam_bootstrap,
         enable_scanner,
     } = service_runtime;
@@ -158,7 +160,7 @@ pub async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifecycle) -
         shutdown_signal,
         s3_shutdown_tx,
         console_shutdown_tx,
-        protocol_shutdowns,
+        optional_runtimes,
         shutdown_token,
     )
     .await;
@@ -377,15 +379,9 @@ async fn handle_shutdown(
     shutdown_signal: ShutdownSignal,
     s3_shutdown_handle: Option<ShutdownHandle>,
     console_shutdown_handle: Option<ShutdownHandle>,
-    protocols: ProtocolShutdownSenders,
+    optional_runtimes: OptionalRuntimeServices,
     ctx: CancellationToken,
 ) {
-    let ProtocolShutdownSenders {
-        ftp: ftp_shutdown_tx,
-        ftps: ftps_shutdown_tx,
-        webdav: webdav_shutdown_tx,
-        sftp: sftp_shutdown_tx,
-    } = protocols;
     ctx.cancel();
 
     info!(
@@ -444,58 +440,7 @@ async fn handle_shutdown(
         );
     }
 
-    let mut protocol_shutdowns = Vec::new();
+    let optional_runtime_shutdowns = prepare_optional_runtime_shutdowns(optional_runtimes);
-    if let Some(ftp_shutdown_tx) = ftp_shutdown_tx {
-        info!(
-            target: "rustfs::main::handle_shutdown",
-            event = EVENT_PROTOCOL_SYSTEM_STATE,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            protocol = "ftp",
-            state = "stopping",
-            "Protocol runtime stopping"
-        );
-        protocol_shutdowns.push(ftp_shutdown_tx.shutdown());
-    }
-
-    if let Some(ftps_shutdown_tx) = ftps_shutdown_tx {
-        info!(
-            target: "rustfs::main::handle_shutdown",
-            event = EVENT_PROTOCOL_SYSTEM_STATE,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            protocol = "ftps",
-            state = "stopping",
-            "Protocol runtime stopping"
-        );
-        protocol_shutdowns.push(ftps_shutdown_tx.shutdown());
-    }
-
-    if let Some(webdav_shutdown_tx) = webdav_shutdown_tx {
-        info!(
-            target: "rustfs::main::handle_shutdown",
-            event = EVENT_PROTOCOL_SYSTEM_STATE,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            protocol = "webdav",
-            state = "stopping",
-            "Protocol runtime stopping"
-        );
-        protocol_shutdowns.push(webdav_shutdown_tx.shutdown());
-    }
-
-    if let Some(sftp_shutdown_tx) = sftp_shutdown_tx {
-        info!(
-            target: "rustfs::main::handle_shutdown",
-            event = EVENT_PROTOCOL_SYSTEM_STATE,
-            component = LOG_COMPONENT_MAIN,
-            subsystem = LOG_SUBSYSTEM_STARTUP,
-            protocol = "sftp",
-            state = "stopping",
-            "Protocol runtime stopping"
-        );
-        protocol_shutdowns.push(sftp_shutdown_tx.shutdown());
-    }
 
     info!(
         target: "rustfs::main::handle_shutdown",
@@ -559,8 +504,7 @@ async fn handle_shutdown(
     if let Some(console_shutdown_handle) = console_shutdown_handle {
         console_shutdown_handle.shutdown().await;
     }
-    join_all(protocol_shutdowns).await;
+    shutdown_optional_runtime_services(optional_runtime_shutdowns).await;
-
     state_manager.update(ServiceState::Stopped);
     info!(
         target: "rustfs::main::handle_shutdown",