From 77d842ae3457f902b550c7b0662de1e78c9974fc Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Sat, 20 Jun 2026 08:15:44 +0800 Subject: [PATCH] refactor: reuse embedded runtime service boundaries (#3641) --- docs/architecture/migration-progress.md | 49 ++++++++---- docs/architecture/runtime-lifecycle.md | 8 ++ rustfs/src/embedded.rs | 94 +++--------------------- rustfs/src/startup_service_components.rs | 67 ++++++++++++++++- rustfs/src/startup_services.rs | 25 ++++++- rustfs/src/startup_shutdown.rs | 20 ++++- 6 files changed, 159 insertions(+), 104 deletions(-) diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index 09e4edbd9..d55cbffd9 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,17 +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-embedded-startup-phase-reuse` +- Branch: `overtrue/arch-embedded-runtime-service-reuse` - Baseline: `origin/main` - (`f8117eb46bb6ae21481bd744cfc245e88fbdb74c`). -- Stacked on: none. + (`131e9dc804a371b3384028da5797ca32a91ac493`). - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: route embedded listen and storage startup phases through - the shared startup server/storage boundaries while preserving embedded-only - fatal and non-fatal behavior. +- Rust code changes: route embedded runtime services and shutdown cleanup + through startup service/shutdown helpers while preserving embedded-only fatal + and non-fatal behavior. - CI/script changes: none. -- Docs changes: record the R-029 embedded startup phase reuse slice. +- Docs changes: record the R-030 embedded runtime service reuse slice. ## Phase 0 Tasks @@ -2153,11 +2152,26 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block migration/layer guards, formatting, diff hygiene, Rust risk scan, branch freshness check, pre-commit quality gate, and three-expert review. +- [x] `R-030` Reuse runtime service boundaries in embedded mode. + - Do: move embedded KMS/buffer/audit setup, bucket metadata migration, IAM + bootstrap, notification setup, and event/audit shutdown cleanup behind + startup service/shutdown helpers. + - Acceptance: embedded startup keeps KMS/audit/notification failures + non-fatal, preserves bucket metadata and IAM initialization order, and + keeps shutdown cleanup behavior unchanged. + - Must preserve: KMS warning-only behavior, buffer profile initialization, + audit warning-only behavior, bucket listing failure shutdown, bucket + metadata migration before IAM migration, IAM bootstrap fatal behavior, + notification warning-only behavior, readiness publication, event notifier + shutdown, audit stop warning behavior, and temp directory cleanup. + - Verification: focused embedded/service/shutdown checks, 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 embedded lifecycle reuse for runtime services and - shutdown compatibility while preserving embedded-specific non-fatal service - behavior. +1. `pure-move`: continue embedded lifecycle reuse for final server-ready and + lifecycle publication edges. 2. `contract`: continue extension contract coverage for future diagnostics and profiler handoff surfaces after runtime owners are stable. @@ -2165,17 +2179,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | R-029 moves embedded listen/storage phase ownership into startup server/storage helpers without routing embedded through binary-only service policy. | -| Migration preservation | passed | Embedded stable-port rejection, global init guard placement, S3-only listener, storage readiness, retry limit, and non-fatal service policy remain unchanged. | -| Testing/verification | passed | Focused startup server/storage/embedded checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | +| Quality/architecture | passed | R-030 moves embedded runtime service and shutdown cleanup ownership into startup service/shutdown helpers while keeping binary-only runtime services separate. | +| Migration preservation | passed | Embedded non-fatal KMS/audit/notification policy, bucket metadata/IAM ordering, readiness publication, and shutdown cleanup behavior remain unchanged. | +| Testing/verification | passed | Focused embedded/service/shutdown checks, guards, formatting, diff hygiene, Rust risk scan, and full pre-commit passed. | ## Verification Notes Passed before push: -- Issue #660 R-029 current slice: - - `cargo test -p rustfs --lib startup_storage -- --nocapture`: passed. - - `cargo test -p rustfs --lib startup_server -- --nocapture`: passed. +- Issue #660 R-030 current slice: + - `cargo test -p rustfs --lib startup_service_components -- --nocapture`: + passed. + - `cargo test -p rustfs --lib startup_services -- --nocapture`: passed; no + matching unit tests currently exist. + - `cargo test -p rustfs --lib startup_shutdown -- --nocapture`: passed. - `cargo test -p rustfs --lib embedded -- --nocapture`: passed; no matching unit tests currently exist. - `cargo check -p rustfs --lib`: passed. diff --git a/docs/architecture/runtime-lifecycle.md b/docs/architecture/runtime-lifecycle.md index 8abf6701d..446e5aa66 100644 --- a/docs/architecture/runtime-lifecycle.md +++ b/docs/architecture/runtime-lifecycle.md @@ -89,6 +89,14 @@ publication, and replication startup. Embedded-specific behavior still owns its stable-port requirement, one-shot global initialization guard placement, S3-only HTTP listener, and non-fatal KMS/audit/notification policy. +## Embedded Runtime Service Reuse + +Embedded runtime service setup should share startup service helpers for optional +service initialization, bucket metadata/IAM setup, notification setup, and +shutdown cleanup. Embedded-specific behavior still owns warning-only +KMS/audit/notification failures, no binary-only background sidecars, no state +manager, and the one-shot server handle cleanup used by embedded shutdown. + ## AppContext Foundation Early AppContext work should split resolver files and add compatibility tests before diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs index 289279c82..1b89e9976 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -47,14 +47,13 @@ //! start a second server will return an error. use crate::config::Config; -use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system}; -use crate::server::{ShutdownHandle, shutdown_event_notifier, start_http_server, stop_audit_system}; -use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap}; +use crate::server::{ShutdownHandle, start_http_server}; +use crate::startup_iam::publish_ready_for_iam_bootstrap; use crate::startup_server::init_embedded_startup_listen_context; +use crate::startup_services::init_embedded_startup_runtime_services; +use crate::startup_shutdown::run_embedded_shutdown_cleanup; use crate::startup_storage::{init_embedded_startup_storage_foundation, init_embedded_startup_storage_runtime}; -use crate::storage_compat::{init_bucket_metadata_sys, try_migrate_bucket_metadata, try_migrate_iam_config}; use rustfs_obs::{init_obs, set_global_guard}; -use rustfs_storage_api::{BucketOperations, BucketOptions}; use rustls::crypto::aws_lc_rs::default_provider; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::path::{Path, PathBuf}; @@ -328,77 +327,15 @@ impl RustFSServerBuilder { }; let store = storage_runtime.store; - // KMS (optional, non-fatal for embedded). - if let Err(e) = init_kms_system(&config).await { - warn!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_optional_service_skipped", - service = "kms", - error = %e, - "Embedded optional service initialization skipped" - ); - } - - // Buffer profiles. - init_buffer_profile_system(&config); - - if let Err(e) = crate::startup_service_components::init_event_notifier_and_audit().await { - warn!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_optional_service_skipped", - service = "audit", - error = %e, - "Embedded optional service initialization skipped" - ); - } - - // Bucket listing for metadata + notification init. - let buckets: Vec = store - .list_bucket(&BucketOptions { - no_metadata: true, - ..Default::default() - }) - .await - .map_err(|e| { - shutdown_embedded_server(); - ServerError::Init(format!("list_bucket: {e}")) - })? - .into_iter() - .map(|v| v.name) - .collect(); - - try_migrate_bucket_metadata(store.clone()).await; - init_bucket_metadata_sys(store.clone(), buckets.clone()).await; - try_migrate_iam_config(store.clone()).await; - - // IAM. - let kms_interface = - rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); - let iam_bootstrap = - bootstrap_or_defer_iam_init(store.clone(), kms_interface, listen_context.readiness.clone(), None, Some(ctx.clone())) + let service_runtime = + init_embedded_startup_runtime_services(&config, endpoint_pools, store, ctx.clone(), listen_context.readiness.clone()) .await .map_err(|e| { shutdown_embedded_server(); - ServerError::Init(format!("IAM bootstrap setup: {e}")) + ServerError::Init(e.to_string()) })?; - // Bucket notifications. - add_bucket_notification_configuration(buckets.clone()).await; - - if let Err(e) = crate::startup_service_components::init_notification_system(endpoint_pools.clone()).await { - warn!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_optional_service_skipped", - service = "notification", - error = %e, - "Embedded optional service initialization skipped" - ); - } - - publish_ready_for_iam_bootstrap(iam_bootstrap, listen_context.readiness.as_ref(), None) + publish_ready_for_iam_bootstrap(service_runtime.iam_bootstrap, listen_context.readiness.as_ref(), None) .await .map_err(|e| { shutdown_embedded_server(); @@ -505,20 +442,7 @@ impl RustFSServer { // Cancel background services. self.cancel_token.cancel(); - // Shutdown event notifier. - shutdown_event_notifier().await; - - // Stop the audit system. - if let Err(e) = stop_audit_system().await { - warn!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_shutdown_cleanup_failed", - service = "audit", - error = %e, - "Embedded shutdown cleanup failed" - ); - } + run_embedded_shutdown_cleanup().await; // Signal HTTP server to stop. if let Some(shutdown_handle) = self.shutdown_handle.take() { diff --git a/rustfs/src/startup_service_components.rs b/rustfs/src/startup_service_components.rs index bab83f623..68d3171e4 100644 --- a/rustfs/src/startup_service_components.rs +++ b/rustfs/src/startup_service_components.rs @@ -17,7 +17,11 @@ use crate::storage_compat::{ try_migrate_bucket_metadata, try_migrate_iam_config, }; use crate::{ - init::{add_bucket_notification_configuration, init_auto_tuner, init_update_check, print_server_info}, + config::Config, + init::{ + 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}, }; @@ -41,10 +45,13 @@ 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_COMPONENT_EMBEDDED: &str = "embedded"; const LOG_SUBSYSTEM_STARTUP: &str = "startup"; const LOG_SUBSYSTEM_AUTH: &str = "auth"; +const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded"; const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; const EVENT_DEADLOCK_DETECTOR_STATE: &str = "deadlock_detector_state"; +const EVENT_EMBEDDED_OPTIONAL_SERVICE_SKIPPED: &str = "embedded_optional_service_skipped"; 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"; @@ -73,6 +80,18 @@ pub(crate) async fn init_audit_runtime() { } } +pub(crate) async fn init_embedded_optional_service_runtime(config: &Config) { + if let Err(err) = init_kms_system(config).await { + log_embedded_optional_service_skipped("kms", err); + } + + init_buffer_profile_system(config); + + if let Err(err) = init_event_notifier_and_audit().await { + log_embedded_optional_service_skipped("audit", err); + } +} + pub(crate) fn init_deadlock_detector_runtime() { let detector = crate::storage::deadlock_detector::get_deadlock_detector(); if detector.is_enabled() { @@ -97,6 +116,24 @@ pub(crate) fn init_deadlock_detector_runtime() { } } +pub(crate) async fn init_embedded_bucket_metadata_runtime(store: Arc) -> Result> { + let buckets_list = store + .list_bucket(&BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .map_err(|err| Error::other(format!("list_bucket: {err}")))?; + + let buckets: Vec = buckets_list.into_iter().map(|v| v.name).collect(); + + try_migrate_bucket_metadata(store.clone()).await; + init_bucket_metadata_sys(store.clone(), buckets.clone()).await; + try_migrate_iam_config(store).await; + + Ok(buckets) +} + pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: CancellationToken) -> Result> { let buckets_list = store .list_bucket(&BucketOptions { @@ -120,6 +157,15 @@ pub(crate) async fn init_bucket_metadata_runtime(store: Arc, ctx: Cance Ok(buckets) } +pub(crate) async fn init_embedded_iam_runtime( + store: Arc, + ctx: CancellationToken, + readiness: Arc, +) -> Result { + let kms_interface = rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); + bootstrap_or_defer_iam_init(store, kms_interface, readiness, None, Some(ctx)).await +} + pub(crate) async fn init_iam_runtime( store: Arc, ctx: CancellationToken, @@ -130,6 +176,14 @@ pub(crate) async fn init_iam_runtime( bootstrap_or_defer_iam_init(store, kms_interface, readiness, Some(state_manager), Some(ctx)).await } +pub(crate) async fn init_embedded_notification_runtime(endpoint_pools: EndpointServerPools, buckets: Vec) { + add_bucket_notification_configuration(buckets).await; + + if let Err(err) = init_notification_system(endpoint_pools).await { + log_embedded_optional_service_skipped("notification", err); + } +} + pub(crate) async fn init_auth_integrations() -> Result<()> { let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?; if keystone_config.enable { @@ -261,6 +315,17 @@ where init_notification().await } +fn log_embedded_optional_service_skipped(service: &str, err: impl std::fmt::Display) { + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = EVENT_EMBEDDED_OPTIONAL_SERVICE_SKIPPED, + service, + error = %err, + "Embedded optional service initialization skipped" + ); +} + #[cfg(test)] mod tests { use super::{init_event_notifier_and_audit_with, init_notification_system_with}; diff --git a/rustfs/src/startup_services.rs b/rustfs/src/startup_services.rs index 3b5ec4d58..df9cb92cb 100644 --- a/rustfs/src/startup_services.rs +++ b/rustfs/src/startup_services.rs @@ -21,7 +21,9 @@ use crate::{ startup_optional_runtime_sidecars::{OptionalRuntimeServices, init_optional_runtime_services}, startup_service_components::{ init_audit_runtime, init_auth_integrations, init_background_service_runtime, init_bucket_metadata_runtime, - init_deadlock_detector_runtime, init_iam_runtime, init_notification_runtime, init_observability_runtime, + init_deadlock_detector_runtime, init_embedded_bucket_metadata_runtime, init_embedded_iam_runtime, + init_embedded_notification_runtime, init_embedded_optional_service_runtime, init_iam_runtime, init_notification_runtime, + init_observability_runtime, }, }; use rustfs_common::GlobalReadiness; @@ -34,6 +36,27 @@ pub struct StartupServiceRuntime { pub enable_scanner: bool, } +pub struct EmbeddedStartupServiceRuntime { + pub iam_bootstrap: IamBootstrapDisposition, +} + +pub async fn init_embedded_startup_runtime_services( + config: &Config, + endpoint_pools: EndpointServerPools, + store: Arc, + ctx: CancellationToken, + readiness: Arc, +) -> Result { + init_embedded_optional_service_runtime(config).await; + let buckets = init_embedded_bucket_metadata_runtime(store.clone()).await?; + let iam_bootstrap = init_embedded_iam_runtime(store, ctx, readiness) + .await + .map_err(|err| std::io::Error::other(format!("IAM bootstrap setup: {err}")))?; + init_embedded_notification_runtime(endpoint_pools, buckets).await; + + Ok(EmbeddedStartupServiceRuntime { iam_bootstrap }) +} + pub async fn init_startup_runtime_services( config: &Config, endpoint_pools: EndpointServerPools, diff --git a/rustfs/src/startup_shutdown.rs b/rustfs/src/startup_shutdown.rs index add16ebd1..fa697d1c2 100644 --- a/rustfs/src/startup_shutdown.rs +++ b/rustfs/src/startup_shutdown.rs @@ -22,15 +22,18 @@ use crate::{ use rustfs_heal::shutdown_ahm_services; use rustfs_utils::get_env_bool_with_aliases; use tokio_util::sync::CancellationToken; -use tracing::{error, info}; +use tracing::{error, info, warn}; 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_COMPONENT_EMBEDDED: &str = "embedded"; const LOG_SUBSYSTEM_STARTUP: &str = "startup"; +const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded"; const EVENT_AUDIT_SYSTEM_STATE: &str = "audit_system_state"; +const EVENT_EMBEDDED_SHUTDOWN_CLEANUP_FAILED: &str = "embedded_shutdown_cleanup_failed"; 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"; @@ -196,6 +199,21 @@ pub async fn run_startup_shutdown_sequence( ); } +pub async fn run_embedded_shutdown_cleanup() { + shutdown_event_notifier().await; + + if let Err(err) = stop_audit_system().await { + warn!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = EVENT_EMBEDDED_SHUTDOWN_CLEANUP_FAILED, + service = "audit", + error = %err, + "Embedded shutdown cleanup failed" + ); + } +} + #[cfg(test)] mod tests { use super::{BackgroundShutdownStep, background_shutdown_steps};