diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 32373dd69..40e4d6c99 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +5,16 @@ 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-startup-runtime-services` -- Baseline: `origin/main` at `3e723a2476692051cac1eab2cc31984c62063130` +- Branch: `overtrue/arch-startup-ready-shutdown-runtime` +- Baseline: `origin/main` at `e26eea7f3a5ef6c272a322a47f58f5a10cb2342b0` - PR type for this branch: `pure-move` -- Runtime behavior changes: no external behavior change expected; KMS, optional - protocols, diagnostics services, bucket metadata migration, IAM bootstrap, - auth integrations, notification runtime, heal/scanner setup, and metrics - initialization still run in the same relative order after storage runtime and - before the server-ready log. -- Rust code changes: add `startup_services::init_startup_runtime_services` and +- Runtime behavior changes: no external behavior change expected; server-ready + publication, global init time, scanner start, shutdown wait, and shutdown + ordering still run in the same relative order after runtime services. +- Rust code changes: add `startup_services::run_startup_runtime_lifecycle` and use it from binary startup. - CI/script changes: none. -- Docs changes: record `R-017` startup runtime service bootstrap progress and +- Docs changes: record `R-018` startup runtime lifecycle progress and verification. ## Phase 0 Tasks @@ -704,10 +702,29 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block formatting, migration guards, Rust risk scan, branch freshness check, and pre-commit quality gate. +- [x] `R-018` Centralize startup ready, scanner, and shutdown lifecycle. + - Do: move server-ready logging, IAM readiness publication, global init time, + scanner start, shutdown signal wait, background shutdown ordering, protocol + shutdown, notifier/audit/profiling shutdown, HTTP shutdown, and final stopped + state logging behind the `startup_services` boundary. + - Acceptance: `main.rs` still initializes listen/storage/runtime services in + the same order, then delegates lifecycle completion; `startup_services` + owns the shutdown handles, runtime token, readiness handle, store, and + service runtime needed for ready/scanner/shutdown orchestration. + - Must preserve: server-ready log fields, inline/deferred IAM readiness + behavior, global init time timing, scanner start timing, shutdown signal log, + runtime token cancellation before service-specific shutdown, scanner before + AHM shutdown order, protocol shutdown order, notifier/audit/profiling + shutdown order, HTTP shutdown order, stopped service state, and final stopped + logs. + - Verification: focused startup services tests, binary/lib compile checks, + formatting, migration guards, Rust risk scan, branch freshness check, and + pre-commit quality gate. + ## Next PRs -1. `pure-move`: continue extracting startup ready/scanner/shutdown wrappers while - preserving startup order, readiness ownership, and shutdown ownership. +1. `pure-move`: continue collapsing remaining binary startup glue around command + dispatch and server bootstrap without changing fatal stderr behavior. 2. `ci-gate`: finish `G-006` public re-export and storage trait coverage checks before the remaining cleanup slices. @@ -715,13 +732,13 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | Pure-move slice removes runtime service details from binary startup behind the existing startup services boundary. | -| Migration preservation | passed | Storage runtime, KMS, optional protocols, diagnostics, bucket metadata, IAM, auth, notification, heal/scanner, metrics, and server-ready order are preserved. | +| Quality/architecture | passed | Pure-move slice removes ready/scanner/shutdown lifecycle details from binary startup behind the existing startup services boundary. | +| Migration preservation | passed | Server-ready publication, scanner start, shutdown wait, background/protocol/notifier/audit/profiling/HTTP shutdown order, and stopped state are preserved. | | Testing/verification | passed | Focused startup services tests, compile checks, formatting, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. | ## Verification Notes -Passed on `3e723a2476692051cac1eab2cc31984c62063130`: +Passed on `e26eea7f3a5ef6c272a322a47f58f5a10cb2342b0`: - `cargo test -p rustfs startup_services --no-fail-fast`: passed. - `cargo check -p rustfs --lib`: passed. @@ -740,14 +757,15 @@ Passed on `3e723a2476692051cac1eab2cc31984c62063130`: Notes: -- This slice centralizes startup runtime services without changing startup - ordering, shutdown token ownership, readiness ownership, or endpoint pool +- This slice centralizes startup lifecycle completion without changing startup + ordering, shutdown token ownership, readiness ownership, or shutdown handle ownership. -- Runtime services remain after storage runtime and before the server-ready log. +- Ready publication and scanner start remain after runtime services; shutdown + wait remains after scanner start. ## Handoff Notes -- R-017 is implemented, locally verified, and current with `origin/main`. +- R-018 is implemented, locally verified, and current with `origin/main`. - Next startup slices can keep using larger pure moves, but must keep startup - ordering, fatal/non-fatal boundaries, shutdown ownership, and readiness - ownership explicit in tests and review notes. + ordering, fatal/non-fatal boundaries, shutdown ownership, readiness ownership, + and fatal stderr behavior explicit in tests and review notes. diff --git a/docs/architecture/startup-timeline.md b/docs/architecture/startup-timeline.md index d41e96624..240f932b1 100644 --- a/docs/architecture/startup-timeline.md +++ b/docs/architecture/startup-timeline.md @@ -44,9 +44,9 @@ new startup semantics. | `RUN-019` | `rustfs/src/startup_services.rs` | Add bucket notification config and initialize notification system. | Adds bucket notification configuration and publishes the global notification system. | Notification config add is non-fatal in this path; global notification init is fatal on error. | None | | `RUN-020` | `rustfs/src/startup_services.rs` | Create AHM cancellation token and initialize heal manager when scanner or heal is enabled. | Creates AHM cancellation token and starts heal manager for heal/scanner workflows. | Heal manager init is fatal when enabled. | None | | `RUN-021` | `rustfs/src/startup_services.rs` | Print server info, init update check, allocator reclaim, metrics, memory observability, and auto-tuner. | Starts informational/update/memory/metrics background tasks when enabled. | Non-fatal in this path. | None | -| `RUN-022` | `rustfs/src/main.rs:599` | Log successful startup and publish full readiness for inline IAM. | Logs version/address, checks runtime readiness, marks `FullReady`, and sets service state to `Ready` when IAM was ready inline. | Fatal if runtime readiness is not reached within the startup wait. | Marks `FullReady` only for inline IAM here | -| `RUN-023` | `rustfs/src/main.rs:609` | Publish global init time and start data scanner when enabled. | Sets global init time and starts scanner after the successful-startup log. | Scanner start is non-fatal in this path. | Full readiness may already be published or may await deferred IAM recovery | -| `RUN-024` | `rustfs/src/main.rs:616` | Wait for shutdown signal. | Blocks the main task until a shutdown signal is received. | Non-fatal. | Runtime remains in its current readiness state | +| `RUN-022` | `rustfs/src/startup_services.rs` and `rustfs/src/startup_iam.rs` | Log successful startup and publish full readiness for inline IAM. | Logs version/address, checks runtime readiness, marks `FullReady`, and sets service state to `Ready` when IAM was ready inline. | Fatal if runtime readiness is not reached within the startup wait. | Marks `FullReady` only for inline IAM here | +| `RUN-023` | `rustfs/src/startup_services.rs` | Publish global init time and start data scanner when enabled. | Sets global init time and starts scanner after the successful-startup log. | Scanner start is non-fatal in this path. | Full readiness may already be published or may await deferred IAM recovery | +| `RUN-024` | `rustfs/src/startup_services.rs` | Wait for shutdown signal. | Blocks the main task until a shutdown signal is received. | Non-fatal. | Runtime remains in its current readiness state | ## Deferred IAM Readiness @@ -68,11 +68,11 @@ new startup semantics. | Step | Source | Current action | Side effects | Fatal boundary | Ready stage | |---|---|---|---|---|---| -| `STOP-001` | `rustfs/src/main.rs:647` | 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/main.rs:675` | 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/main.rs:699` | Signal optional FTP/FTPS/WebDAV/SFTP protocol servers. | Collects protocol shutdown futures. | Non-fatal in this path. | No readiness-stage change | -| `STOP-004` | `rustfs/src/main.rs:735` | 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/main.rs:763` | 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-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-004` | `rustfs/src/startup_services.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 | ## Migration Rules diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index a09937192..e603775ca 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -13,40 +13,16 @@ // limitations under the License. // Ensure the correct path for parse_license is imported -use futures_util::future::join_all; -use rustfs::server::{ - ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, stop_audit_system, - wait_for_shutdown, -}; -use rustfs::startup_iam::publish_ready_for_iam_bootstrap; use rustfs::startup_preflight::{StartupServerPreflightError, bootstrap_external_prefix_compat, init_startup_server_preflight}; -use rustfs::startup_protocols::ProtocolShutdownSenders; use rustfs::startup_server::{StartupHttpServers, StartupListenContext, init_startup_http_servers, init_startup_listen_context}; -use rustfs::startup_services::{StartupServiceRuntime, init_startup_runtime_services}; +use rustfs::startup_services::{StartupRuntimeLifecycle, init_startup_runtime_services, run_startup_runtime_lifecycle}; use rustfs::startup_storage::{StartupStorageRuntime, init_startup_storage_foundation, init_startup_storage_runtime}; -use rustfs_ecstore::global::shutdown_background_services; -use rustfs_heal::shutdown_ahm_services; -use rustfs_scanner::init_data_scanner; -use rustfs_utils::get_env_bool_with_aliases; use std::io::{Error, Result}; -use tokio_util::sync::CancellationToken; -use tracing::{error, info, instrument}; +use tracing::{error, instrument}; -const ENV_SCANNER_ENABLED: &str = "RUSTFS_SCANNER_ENABLED"; -const ENV_SCANNER_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_SCANNER"; -const ENV_HEAL_ENABLED: &str = "RUSTFS_HEAL_ENABLED"; -const ENV_HEAL_ENABLED_DEPRECATED: &str = "RUSTFS_ENABLE_HEAL"; const LOG_COMPONENT_MAIN: &str = "main"; const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const EVENT_SERVER_RUNTIME_FAILED: &str = "server_runtime_failed"; -const EVENT_PROTOCOL_SYSTEM_STATE: &str = "protocol_system_state"; -const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; -const EVENT_SERVER_READY: &str = "server_ready"; -const EVENT_SHUTDOWN_SIGNAL_RECEIVED: &str = "shutdown_signal_received"; -const EVENT_BACKGROUND_SERVICE_SHUTDOWN: &str = "background_service_shutdown"; -const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown"; -const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown"; -const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state"; const OBSERVABILITY_INIT_FATAL_ALREADY_REPORTED: &str = "observability initialization failure already reported"; #[cfg(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))] #[global_allocator] @@ -147,11 +123,7 @@ async fn run(config: rustfs::config::Config) -> Result<()> { shutdown_token: ctx, } = init_startup_storage_runtime(server_addr, &endpoint_pools, readiness.clone()).await?; - let StartupServiceRuntime { - protocol_shutdowns, - iam_bootstrap, - enable_scanner, - } = init_startup_runtime_services( + let service_runtime = init_startup_runtime_services( &config, endpoint_pools, store.clone(), @@ -161,275 +133,17 @@ async fn run(config: rustfs::config::Config) -> Result<()> { ) .await?; - info!( - target: "rustfs::main::run", - event = EVENT_SERVER_READY, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - version = %rustfs::version::get_version(), - server_address = %server_address, - started_at = %jiff::Zoned::now(), - iam_bootstrap = ?iam_bootstrap, - "RustFS server ready" - ); - publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), Some(state_manager.as_ref())).await?; - // Set the global RustFS initialization time to now - rustfs_common::set_global_init_time_now().await; - - if enable_scanner { - init_data_scanner(ctx.clone(), store.clone()).await; - } - - // listen to the shutdown signal - let shutdown_signal = wait_for_shutdown().await; - handle_shutdown( - &state_manager, - shutdown_signal, + run_startup_runtime_lifecycle(StartupRuntimeLifecycle { + server_address, + state_manager, s3_shutdown_tx, console_shutdown_tx, - protocol_shutdowns, - ctx.clone(), - ) - .await; - - info!( - target: "rustfs::main::run", - event = EVENT_SERVER_SHUTDOWN_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = ?state_manager.current_state(), - result = "stopped", - "RustFS server stopped" - ); - Ok(()) -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -enum BackgroundShutdownStep { - DataScanner, - Ahm, -} - -fn background_shutdown_steps(enable_scanner: bool, enable_heal: bool) -> Vec { - let mut steps = Vec::with_capacity(2); - if enable_scanner { - steps.push(BackgroundShutdownStep::DataScanner); - } - if enable_heal || enable_scanner { - steps.push(BackgroundShutdownStep::Ahm); - } - steps -} - -/// Handles the shutdown process of the server -async fn handle_shutdown( - state_manager: &ServiceStateManager, - shutdown_signal: ShutdownSignal, - s3_shutdown_handle: Option, - console_shutdown_handle: Option, - protocols: ProtocolShutdownSenders, - ctx: CancellationToken, -) { - let ProtocolShutdownSenders { - ftp: ftp_shutdown_tx, - ftps: ftps_shutdown_tx, - webdav: webdav_shutdown_tx, - sftp: sftp_shutdown_tx, - } = protocols; - ctx.cancel(); - - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_SHUTDOWN_SIGNAL_RECEIVED, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - signal = shutdown_signal.log_label(), - "Shutdown signal received" - ); - // update the status to stopping first - state_manager.update(ServiceState::Stopping); - - // Check environment variables to determine what services need to be stopped - let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true); - let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true); - - // Stop background services based on what was enabled. - let background_steps = background_shutdown_steps(enable_scanner, enable_heal); - for step in &background_steps { - match step { - BackgroundShutdownStep::DataScanner => { - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - service = "data_scanner", - state = "stopping", - "Background service shutdown started" - ); - shutdown_background_services(); - } - BackgroundShutdownStep::Ahm => { - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - service = "ahm", - state = "stopping", - "Background service shutdown started" - ); - shutdown_ahm_services(); - } - } - } - - if background_steps.is_empty() { - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - service = "ahm", - state = "skipped", - reason = "disabled", - "Background service shutdown skipped" - ); - } - - // Shutdown FTP and FTPS servers - let mut protocol_shutdowns = Vec::new(); - 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()); - } - - // Shutdown WebDAV server - 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()); - } - - // Shutdown SFTP server - 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()); - } - - // Stop the notification system - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_EVENT_NOTIFIER_SHUTDOWN, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopping", - "Event notifier shutdown started" - ); - shutdown_event_notifier().await; - - // Stop the audit system - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_AUDIT_SYSTEM_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopping", - "Audit runtime stopping" - ); - match stop_audit_system().await { - Ok(_) => info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_AUDIT_SYSTEM_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopped", - "Audit runtime stopped" - ), - Err(e) => error!( - target: "rustfs::main::handle_shutdown", - event = EVENT_AUDIT_SYSTEM_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stop_failed", - error = %e, - "Audit runtime failed to stop" - ), - } - - // Stop profiling tasks - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_PROFILING_SHUTDOWN, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopping", - "Profiling shutdown started" - ); - rustfs::profiling::shutdown_profiling(); - - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_SERVER_SHUTDOWN_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopping", - "RustFS server stopping" - ); - if let Some(s3_shutdown_handle) = s3_shutdown_handle { - s3_shutdown_handle.shutdown().await; - } - if let Some(console_shutdown_handle) = console_shutdown_handle { - console_shutdown_handle.shutdown().await; - } - join_all(protocol_shutdowns).await; - - // the last updated status is stopped - state_manager.update(ServiceState::Stopped); - info!( - target: "rustfs::main::handle_shutdown", - event = EVENT_SERVER_SHUTDOWN_STATE, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - state = "stopped", - "RustFS server stopped" - ); + service_runtime, + store, + shutdown_token: ctx, + readiness, + }) + .await } #[cfg(test)] @@ -443,18 +157,4 @@ mod tests { "[FATAL] Observability initialization failed: collector unavailable" ); } - - #[test] - fn background_shutdown_plan_keeps_scanner_before_ahm() { - assert_eq!( - background_shutdown_steps(true, true), - vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm] - ); - assert_eq!( - background_shutdown_steps(true, false), - vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm] - ); - assert_eq!(background_shutdown_steps(false, true), vec![BackgroundShutdownStep::Ahm]); - assert!(background_shutdown_steps(false, false).is_empty()); - } } diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index db8136084..b6e858069 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -18,10 +18,14 @@ use crate::{ add_bucket_notification_configuration, init_auto_tuner, init_buffer_profile_system, init_kms_system, init_update_check, print_server_info, }, - server::{ServiceStateManager, init_event_notifier, start_audit_system}, - startup_iam::{IamBootstrapDisposition, bootstrap_or_defer_iam_init}, + server::{ + ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, shutdown_event_notifier, + 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}, }; +use futures_util::future::join_all; use rustfs_audit::AuditResult; use rustfs_common::GlobalReadiness; use rustfs_ecstore::{ @@ -31,13 +35,17 @@ use rustfs_ecstore::{ replication::get_global_replication_pool, }, endpoints::EndpointServerPools, + global::shutdown_background_services, notification_sys::new_global_notification_sys, store::ECStore, store_api::BucketOperations, }; -use rustfs_heal::{create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager}; +use rustfs_heal::{ + create_ahm_services_cancel_token, heal::storage::ECStoreHealStorage, init_heal_manager, shutdown_ahm_services, +}; use rustfs_iam::init_oidc_sys; use rustfs_obs::init_metrics_runtime; +use rustfs_scanner::init_data_scanner; use rustfs_storage_api::BucketOptions; use rustfs_utils::get_env_bool_with_aliases; use std::{ @@ -56,12 +64,19 @@ 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"; const EVENT_OIDC_INITIALIZATION_FAILED: &str = "oidc_initialization_failed"; const EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED: &str = "notification_system_initialization_failed"; const EVENT_BACKGROUND_SERVICES_CONFIGURED: &str = "background_services_configured"; +const EVENT_SERVER_READY: &str = "server_ready"; +const EVENT_SHUTDOWN_SIGNAL_RECEIVED: &str = "shutdown_signal_received"; +const EVENT_BACKGROUND_SERVICE_SHUTDOWN: &str = "background_service_shutdown"; +const EVENT_EVENT_NOTIFIER_SHUTDOWN: &str = "event_notifier_shutdown"; +const EVENT_PROFILING_SHUTDOWN: &str = "profiling_shutdown"; +const EVENT_SERVER_SHUTDOWN_STATE: &str = "server_shutdown_state"; pub struct StartupServiceRuntime { pub protocol_shutdowns: ProtocolShutdownSenders, @@ -69,6 +84,17 @@ pub struct StartupServiceRuntime { pub enable_scanner: bool, } +pub struct StartupRuntimeLifecycle { + pub server_address: String, + pub state_manager: Arc, + pub s3_shutdown_tx: Option, + pub console_shutdown_tx: Option, + pub service_runtime: StartupServiceRuntime, + pub store: Arc, + pub shutdown_token: CancellationToken, + pub readiness: Arc, +} + pub async fn init_startup_runtime_services( config: &Config, endpoint_pools: EndpointServerPools, @@ -99,6 +125,64 @@ pub async fn init_startup_runtime_services( }) } +pub async fn run_startup_runtime_lifecycle(lifecycle: StartupRuntimeLifecycle) -> Result<()> { + let StartupRuntimeLifecycle { + server_address, + state_manager, + s3_shutdown_tx, + console_shutdown_tx, + service_runtime, + store, + shutdown_token, + readiness, + } = lifecycle; + let StartupServiceRuntime { + protocol_shutdowns, + iam_bootstrap, + enable_scanner, + } = service_runtime; + + info!( + target: "rustfs::main::run", + event = EVENT_SERVER_READY, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + version = %crate::version::get_version(), + server_address = %server_address, + started_at = %jiff::Zoned::now(), + iam_bootstrap = ?iam_bootstrap, + "RustFS server ready" + ); + publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), Some(state_manager.as_ref())).await?; + rustfs_common::set_global_init_time_now().await; + + if enable_scanner { + init_data_scanner(shutdown_token.clone(), store).await; + } + + let shutdown_signal = wait_for_shutdown().await; + handle_shutdown( + &state_manager, + shutdown_signal, + s3_shutdown_tx, + console_shutdown_tx, + protocol_shutdowns, + shutdown_token, + ) + .await; + + info!( + target: "rustfs::main::run", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = ?state_manager.current_state(), + result = "stopped", + "RustFS server stopped" + ); + Ok(()) +} + async fn init_audit_runtime() { match init_event_notifier_and_audit().await { Ok(()) => info!( @@ -278,6 +362,223 @@ async fn init_observability_runtime(ctx: CancellationToken) { } } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum BackgroundShutdownStep { + DataScanner, + Ahm, +} + +fn background_shutdown_steps(enable_scanner: bool, enable_heal: bool) -> Vec { + let mut steps = Vec::with_capacity(2); + if enable_scanner { + steps.push(BackgroundShutdownStep::DataScanner); + } + if enable_heal || enable_scanner { + steps.push(BackgroundShutdownStep::Ahm); + } + steps +} + +async fn handle_shutdown( + state_manager: &ServiceStateManager, + shutdown_signal: ShutdownSignal, + s3_shutdown_handle: Option, + console_shutdown_handle: Option, + protocols: ProtocolShutdownSenders, + ctx: CancellationToken, +) { + let ProtocolShutdownSenders { + ftp: ftp_shutdown_tx, + ftps: ftps_shutdown_tx, + webdav: webdav_shutdown_tx, + sftp: sftp_shutdown_tx, + } = protocols; + ctx.cancel(); + + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_SHUTDOWN_SIGNAL_RECEIVED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + signal = shutdown_signal.log_label(), + "Shutdown signal received" + ); + state_manager.update(ServiceState::Stopping); + + let enable_scanner = get_env_bool_with_aliases(ENV_SCANNER_ENABLED, &[ENV_SCANNER_ENABLED_DEPRECATED], true); + let enable_heal = get_env_bool_with_aliases(ENV_HEAL_ENABLED, &[ENV_HEAL_ENABLED_DEPRECATED], true); + + let background_steps = background_shutdown_steps(enable_scanner, enable_heal); + for step in &background_steps { + match step { + BackgroundShutdownStep::DataScanner => { + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "data_scanner", + state = "stopping", + "Background service shutdown started" + ); + shutdown_background_services(); + } + BackgroundShutdownStep::Ahm => { + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "ahm", + state = "stopping", + "Background service shutdown started" + ); + shutdown_ahm_services(); + } + } + } + + if background_steps.is_empty() { + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_BACKGROUND_SERVICE_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + service = "ahm", + state = "skipped", + reason = "disabled", + "Background service shutdown skipped" + ); + } + + let mut protocol_shutdowns = Vec::new(); + 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", + event = EVENT_EVENT_NOTIFIER_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Event notifier shutdown started" + ); + shutdown_event_notifier().await; + + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Audit runtime stopping" + ); + match stop_audit_system().await { + Ok(_) => info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopped", + "Audit runtime stopped" + ), + Err(e) => error!( + target: "rustfs::main::handle_shutdown", + event = EVENT_AUDIT_SYSTEM_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stop_failed", + error = %e, + "Audit runtime failed to stop" + ), + } + + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_PROFILING_SHUTDOWN, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "Profiling shutdown started" + ); + crate::profiling::shutdown_profiling(); + + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopping", + "RustFS server stopping" + ); + if let Some(s3_shutdown_handle) = s3_shutdown_handle { + s3_shutdown_handle.shutdown().await; + } + if let Some(console_shutdown_handle) = console_shutdown_handle { + console_shutdown_handle.shutdown().await; + } + join_all(protocol_shutdowns).await; + + state_manager.update(ServiceState::Stopped); + info!( + target: "rustfs::main::handle_shutdown", + event = EVENT_SERVER_SHUTDOWN_STATE, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + state = "stopped", + "RustFS server stopped" + ); +} + pub async fn init_event_notifier_and_audit() -> AuditResult<()> { init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await } @@ -310,7 +611,9 @@ where #[cfg(test)] mod tests { - use super::{init_event_notifier_and_audit_with, init_notification_system_with}; + use super::{ + BackgroundShutdownStep, background_shutdown_steps, init_event_notifier_and_audit_with, init_notification_system_with, + }; use rustfs_audit::AuditError; use std::sync::{Arc, Mutex}; @@ -363,4 +666,18 @@ mod tests { assert!(result.is_err()); } + + #[test] + fn background_shutdown_plan_keeps_scanner_before_ahm() { + assert_eq!( + background_shutdown_steps(true, true), + vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm] + ); + assert_eq!( + background_shutdown_steps(true, false), + vec![BackgroundShutdownStep::DataScanner, BackgroundShutdownStep::Ahm] + ); + assert_eq!(background_shutdown_steps(false, true), vec![BackgroundShutdownStep::Ahm]); + assert!(background_shutdown_steps(false, false).is_empty()); + } }