From 949939137066435e590c8bd0f029ed1500e0e1cd Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sun, 14 Jun 2026 21:59:17 +0800 Subject: [PATCH] refactor: centralize startup service bootstrap (#3448) --- docs/architecture/migration-progress.md | 66 +++++++++------ rustfs/src/embedded.rs | 14 +--- rustfs/src/lib.rs | 1 + rustfs/src/main.rs | 36 ++++---- rustfs/src/startup_services.rs | 106 ++++++++++++++++++++++++ 5 files changed, 165 insertions(+), 58 deletions(-) create mode 100644 rustfs/src/startup_services.rs diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 210477ad7..76410933e 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,18 +5,18 @@ 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-readiness-bootstrap` -- Baseline: `origin/main` at `8d23ce06c6dba11f50f656af4c30b63036cef92f` +- Branch: `overtrue/arch-startup-service-bootstrap` +- Baseline: `origin/main` at `b49057c8e3e0b1a0ed828912806ee67b49e404c4` - PR type for this branch: `pure-move` -- Runtime behavior changes: no external behavior change expected; inline IAM - bootstrap still publishes runtime readiness after runtime dependencies are - ready, and deferred IAM bootstrap still leaves readiness publication to the - recovery loop. -- Rust code changes: centralize the IAM bootstrap readiness publication decision - in `startup_iam`, use it from binary and embedded startup, and add - wrapper-level coverage for inline, deferred, and failure paths. +- Runtime behavior changes: no external behavior change expected; binary startup + still treats notification initialization failure as fatal, while embedded + startup still treats audit and notification failures as non-fatal warnings. +- Rust code changes: centralize event notifier, audit startup, and notification + system bootstrap in `startup_services` helpers with caller-owned + logging/error policy, then use those helpers from binary and embedded + startup. - CI/script changes: none. -- Docs changes: record `R-009` startup readiness bootstrap wrapper progress and +- Docs changes: record `R-010` startup service bootstrap wrapper progress and verification. ## Phase 0 Tasks @@ -565,6 +565,19 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block - Verification: focused startup IAM tests, binary/lib compile checks, formatting, migration guards, Rust risk scan, and pre-commit quality gate. +- [x] `R-010` Centralize startup optional service bootstrap. + - Do: move event notifier, audit startup, and notification system startup + behind `startup_services` helpers with caller-owned logging/error policy. + - Acceptance: binary still initializes the event notifier before audit, logs + audit start/failure through the same startup target, and treats notification + init failure as fatal; embedded still treats audit and notification failures + as non-fatal warnings. + - Must preserve: startup order, audit non-fatal behavior, notification fatal + boundary in binary, embedded warn-and-continue behavior, and event notifier + initialization. + - Verification: focused startup service tests, binary/lib compile checks, + formatting, migration guards, Rust risk scan, and pre-commit quality gate. + ## Next PRs 1. `pure-move`: continue extracting startup boot wrappers in larger slices while @@ -576,38 +589,37 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | Pure-move slice centralizes IAM bootstrap readiness publication without moving runtime readiness collection or startup dependency checks. | -| Migration preservation | passed | Main and embedded keep ReadyInline publication behavior, Deferred remains recovery-loop owned, and embedded readiness errors still map through shutdown. | -| Testing/verification | passed | Focused startup IAM tests, compile checks, formatting, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. | +| Quality/architecture | passed | Pure-move slice centralizes optional startup service bootstrap while keeping caller-specific logging and error policy at the binary and embedded boundaries. | +| Migration preservation | passed | Binary keeps notification init fatal, embedded keeps audit and notification failures warn-and-continue, and event notifier still starts before audit. | +| Testing/verification | passed | Focused startup service tests, compile checks, formatting, migration/layer guards, Rust risk scan, branch freshness check, and full `make pre-commit` passed. | ## Verification Notes -Passed on `8d23ce06c6dba11f50f656af4c30b63036cef92f`: +Passed on `b49057c8e3e0b1a0ed828912806ee67b49e404c4`: -- `cargo fmt --all --check`. -- `cargo test -p rustfs startup_iam --no-fail-fast`. +- `cargo test -p rustfs startup_services --no-fail-fast`. - `cargo check -p rustfs --bin rustfs`. - `cargo check -p rustfs --lib`. +- `cargo fmt --all --check`. - `git diff --check`. - `./scripts/check_architecture_migration_rules.sh`. - `./scripts/check_layer_dependencies.sh`. - `git rev-list --left-right --count HEAD...origin/main` returned `0 0`. - Rust risk scan for changed Rust files: full-file matches were existing docs - examples, existing startup error output, existing test expectations, and the - existing boxed recovery future; added-line scan returned no unwrap/expect, - numeric cast, string error, boxed error, print macro, relaxed-ordering, or - unsafe match. -- `make pre-commit`: all checks passed, including nextest with 5966 passed and + examples, existing startup error output, and existing startup imports; added + lines introduced no unwrap/expect, numeric cast, string error, boxed error, + print macro, relaxed-ordering, or unsafe match. +- `make pre-commit`: all checks passed, including nextest with 5969 passed and 111 skipped, plus doctests. Notes: -- This slice centralizes startup IAM readiness publication without changing the - runtime readiness checks themselves. -- Deferred IAM bootstrap readiness remains owned by the recovery loop. +- This slice centralizes startup optional service bootstrap without moving + logging or error-policy ownership out of binary and embedded startup. +- Binary and embedded keep their existing notification failure semantics. ## Handoff Notes -- R-009 is complete. -- Next startup slices can be larger pure moves, but must keep startup ordering - and readiness ownership explicit in tests. +- R-010 is complete. +- Next startup slices can be larger pure moves, but must keep startup ordering, + fatal/non-fatal boundaries, and readiness ownership explicit in tests. diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index c47b24823..a99817bf7 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -48,9 +48,7 @@ use crate::config::Config; use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system}; -use crate::server::{ - ShutdownHandle, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system, -}; +use crate::server::{ShutdownHandle, shutdown_event_notifier, start_http_server, stop_audit_system}; use crate::startup_fs_guard::enforce_unsupported_fs_policy; use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap}; use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; @@ -65,7 +63,6 @@ use rustfs_ecstore::{ config as ecconfig, endpoints::EndpointServerPools, global::set_global_rustfs_port, - notification_sys::new_global_notification_sys, set_global_endpoints, store::ECStore, store::init_local_disks, @@ -435,11 +432,7 @@ impl RustFSServerBuilder { // Buffer profiles. init_buffer_profile_system(&config); - // Event notifier. - init_event_notifier().await; - - // Audit (non-fatal). - if let Err(e) = start_audit_system().await { + if let Err(e) = crate::startup_services::init_event_notifier_and_audit().await { warn!( component = LOG_COMPONENT_EMBEDDED, subsystem = LOG_SUBSYSTEM_EMBEDDED, @@ -482,8 +475,7 @@ impl RustFSServerBuilder { // Bucket notifications. add_bucket_notification_configuration(buckets.clone()).await; - // Notification system. - if let Err(e) = new_global_notification_sys(endpoint_pools.clone()).await { + if let Err(e) = crate::startup_services::init_notification_system(endpoint_pools.clone()).await { warn!( component = LOG_COMPONENT_EMBEDDED, subsystem = LOG_SUBSYSTEM_EMBEDDED, diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index ece2610b5..b6c1ae9b9 100644 --- a/rustfs/src/lib.rs +++ b/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_services; pub mod storage; pub(crate) mod table_catalog; pub mod tls; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 90cbc2d93..2c8436d26 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -30,8 +30,8 @@ use futures_util::future::join_all; use rustfs::capacity::capacity_integration::init_capacity_management; use rustfs::license::{init_license, license_status}; use rustfs::server::{ - ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, init_event_notifier, shutdown_event_notifier, - start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown, + ServiceState, ServiceStateManager, ShutdownHandle, ShutdownSignal, shutdown_event_notifier, start_http_server, + stop_audit_system, wait_for_shutdown, }; use rustfs::startup_fs_guard::enforce_unsupported_fs_policy; use rustfs::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap}; @@ -45,7 +45,6 @@ use rustfs_ecstore::{ config as ecconfig, endpoints::EndpointServerPools, global::{set_global_rustfs_port, shutdown_background_services}, - notification_sys::new_global_notification_sys, set_global_endpoints, store::ECStore, store::init_local_disks, @@ -778,12 +777,8 @@ async fn run(config: rustfs::config::Config) -> Result<()> { // Initialize buffer profiling system init_buffer_profile_system(&config); - // Initialize event notifier - init_event_notifier().await; - - // Start the audit system - match start_audit_system().await { - Ok(_) => info!( + match rustfs::startup_services::init_event_notifier_and_audit().await { + Ok(()) => info!( target: "rustfs::main::run", event = EVENT_AUDIT_SYSTEM_STATE, component = LOG_COMPONENT_MAIN, @@ -893,17 +888,18 @@ async fn run(config: rustfs::config::Config) -> Result<()> { add_bucket_notification_configuration(buckets.clone()).await; - // Initialize the global notification system - new_global_notification_sys(endpoint_pools.clone()).await.map_err(|err| { - error!( - event = EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED, - component = LOG_COMPONENT_MAIN, - subsystem = LOG_SUBSYSTEM_STARTUP, - error = ?err, - "Failed to initialize notification system" - ); - Error::other(err) - })?; + rustfs::startup_services::init_notification_system(endpoint_pools.clone()) + .await + .map_err(|err| { + error!( + event = EVENT_NOTIFICATION_SYSTEM_INITIALIZATION_FAILED, + component = LOG_COMPONENT_MAIN, + subsystem = LOG_SUBSYSTEM_STARTUP, + error = ?err, + "Failed to initialize notification system" + ); + Error::other(err) + })?; // Create a cancellation token for AHM services let _ = create_ahm_services_cancel_token(); diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs new file mode 100644 index 000000000..7ab65ad91 --- /dev/null +++ b/rustfs/src/startup_services.rs @@ -0,0 +1,106 @@ +// 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::{init_event_notifier, start_audit_system}; +use rustfs_audit::AuditResult; +use rustfs_ecstore::endpoints::EndpointServerPools; +use rustfs_ecstore::notification_sys::new_global_notification_sys; +use std::future::Future; + +pub async fn init_event_notifier_and_audit() -> AuditResult<()> { + init_event_notifier_and_audit_with(init_event_notifier, start_audit_system).await +} + +async fn init_event_notifier_and_audit_with( + notify: NotifyFn, + start_audit: AuditFn, +) -> AuditResult<()> +where + NotifyFn: FnOnce() -> NotifyFuture, + NotifyFuture: Future, + AuditFn: FnOnce() -> AuditFuture, + AuditFuture: Future>, +{ + notify().await; + start_audit().await +} + +pub async fn init_notification_system(endpoint_pools: EndpointServerPools) -> rustfs_ecstore::error::Result<()> { + init_notification_system_with(|| new_global_notification_sys(endpoint_pools)).await +} + +async fn init_notification_system_with(init_notification: InitFn) -> rustfs_ecstore::error::Result<()> +where + InitFn: FnOnce() -> InitFuture, + InitFuture: Future>, +{ + init_notification().await +} + +#[cfg(test)] +mod tests { + use super::{init_event_notifier_and_audit_with, init_notification_system_with}; + use rustfs_audit::AuditError; + use std::sync::{Arc, Mutex}; + + #[tokio::test] + async fn event_notifier_runs_before_successful_audit_start() { + let events = Arc::new(Mutex::new(Vec::new())); + let notify_events = events.clone(); + let audit_events = events.clone(); + let result = init_event_notifier_and_audit_with( + move || async move { + notify_events.lock().unwrap_or_else(|err| err.into_inner()).push("notify"); + }, + move || async move { + audit_events.lock().unwrap_or_else(|err| err.into_inner()).push("audit"); + Ok(()) + }, + ) + .await; + + assert!(result.is_ok()); + let events = events.lock().unwrap_or_else(|err| err.into_inner()).clone(); + assert_eq!(events, ["notify", "audit"]); + } + + #[tokio::test] + async fn event_notifier_runs_before_failed_audit_result() { + let events = Arc::new(Mutex::new(Vec::new())); + let notify_events = events.clone(); + let audit_events = events.clone(); + + let result = init_event_notifier_and_audit_with( + move || async move { + notify_events.lock().unwrap_or_else(|err| err.into_inner()).push("notify"); + }, + move || async move { + audit_events.lock().unwrap_or_else(|err| err.into_inner()).push("audit"); + Err(AuditError::ConfigNotLoaded) + }, + ) + .await; + + assert!(result.is_err()); + let events = events.lock().unwrap_or_else(|err| err.into_inner()).clone(); + assert_eq!(events, ["notify", "audit"]); + } + + #[tokio::test] + async fn notification_system_returns_source_error() { + let result = init_notification_system_with(|| async { Err(rustfs_ecstore::error::Error::FaultyDisk) }).await; + + assert!(result.is_err()); + } +}