From 131e9dc804a371b3384028da5797ca32a91ac493 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 06:58:48 +0800 Subject: [PATCH] refactor: reuse embedded startup phase boundaries (#3640) --- docs/architecture/migration-progress.md | 53 +++++---- docs/architecture/runtime-lifecycle.md | 8 ++ rustfs/src/embedded.rs | 138 ++++++------------------ rustfs/src/startup_server.rs | 39 +++++++ rustfs/src/startup_storage.rs | 76 +++++++++++++ 5 files changed, 188 insertions(+), 126 deletions(-) diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index ab603df1b..09e4edbd9 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,17 +5,17 @@ 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-tls-material-boundary` +- Branch: `overtrue/arch-embedded-startup-phase-reuse` - Baseline: `origin/main` - (`46a02b6d03e2d8274222e7ef63ac2eb5d9124c69`). -- Stacked on: rustfs/rustfs#3638. + (`f8117eb46bb6ae21481bd744cfc245e88fbdb74c`). +- Stacked on: none. - PR type for this branch: `pure-move` - Runtime behavior changes: none. -- Rust code changes: add `startup_tls_material` and move outbound TLS material - loading, generation recording, and TLS metrics initialization out of - `startup_runtime`. +- 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. - CI/script changes: none. -- Docs changes: record the R-028 startup TLS material boundary slice. +- Docs changes: record the R-029 embedded startup phase reuse slice. ## Phase 0 Tasks @@ -2138,10 +2138,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-029` Reuse startup phase boundaries in embedded mode. + - Do: move embedded listen setup, endpoint/local disk setup, ECStore/global + config setup, storage readiness publication, and replication startup behind + startup server/storage helpers. + - Acceptance: embedded startup keeps its stable-port requirement, global + startup guard placement, S3-only HTTP startup, readiness publication, and + storage initialization order while sharing the same startup phase owners. + - Must preserve: embedded port 0 rejection, credential/region publication, + endpoint and unsupported filesystem validation, local disk and lock client + initialization, ECStore fatal shutdown behavior, global config retry limit, + and embedded-specific non-fatal KMS/audit/notification behavior. + - Verification: focused embedded/startup storage 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 larger lifecycle hook slices for startup profiling and - diagnostics while preserving startup and shutdown ordering. +1. `pure-move`: continue embedded lifecycle reuse for runtime services and + shutdown compatibility while preserving embedded-specific non-fatal service + behavior. 2. `contract`: continue extension contract coverage for future diagnostics and profiler handoff surfaces after runtime owners are stable. @@ -2149,26 +2165,27 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block | Expert | Status | Notes | |---|---|---| -| Quality/architecture | passed | R-028 moves outbound TLS material ownership into `startup_tls_material` while keeping `startup_runtime` as the BOOT-006 orchestrator. | -| Migration preservation | passed | BOOT-006 ordering, configured TLS fatal behavior, path trimming, generation saturation, publication, and metrics behavior remain unchanged. | -| Testing/verification | passed | Focused TLS material/runtime checks, guards, final hygiene, Rust risk scan, and full pre-commit passed. | +| 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. | ## Verification Notes Passed before push: -- Issue #660 R-028 current slice: - - `cargo test -p rustfs --lib startup_tls_material -- --nocapture`: passed. - - `cargo test -p rustfs --lib startup_runtime -- --nocapture`: passed. +- Issue #660 R-029 current slice: + - `cargo test -p rustfs --lib startup_storage -- --nocapture`: passed. + - `cargo test -p rustfs --lib startup_server -- --nocapture`: passed. + - `cargo test -p rustfs --lib embedded -- --nocapture`: passed; no + matching unit tests currently exist. - `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; only existing doc-example - `println!` in `rustfs/src/lib.rs`. - - `make pre-commit`: passed; 6304 tests passed, 111 skipped, doctests - passed. + `Box` / `println!` and existing test-only `panic!`. + - `make pre-commit`: passed. - Issue #660 X-012 current slice: - `cargo test -p rustfs-extension-schema`: passed. diff --git a/docs/architecture/runtime-lifecycle.md b/docs/architecture/runtime-lifecycle.md index 9158ea0f0..8abf6701d 100644 --- a/docs/architecture/runtime-lifecycle.md +++ b/docs/architecture/runtime-lifecycle.md @@ -81,6 +81,14 @@ outbound TLS publication, generation recording, and TLS metrics initialization. `startup_runtime` still owns BOOT-006 ordering and must preserve the fatal boundary when configured TLS material fails to load. +## Embedded Startup Phase Reuse + +Embedded startup should use the same startup server and storage phase owners for +listen context, endpoint/local disk setup, storage runtime setup, readiness +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. + ## 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 b4f1953eb..289279c82 100644 --- a/rustfs/src/embedded.rs +++ b/rustfs/src/embedded.rs @@ -49,28 +49,18 @@ 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_fs_guard::enforce_unsupported_fs_policy; use crate::startup_iam::{bootstrap_or_defer_iam_init, publish_ready_for_iam_bootstrap}; -use crate::storage_compat::init_lock_clients; -use crate::storage_compat::{ - ECStore, EndpointServerPools, init as init_ecstore_config, init_background_replication, init_bucket_metadata_sys, - init_global_config_sys, init_local_disks, set_global_endpoints, set_global_rustfs_port, try_migrate_bucket_metadata, - try_migrate_iam_config, try_migrate_server_config, update_erasure_type, -}; -use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; -use rustfs_credentials::init_global_action_credentials; +use crate::startup_server::init_embedded_startup_listen_context; +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 rustfs_utils::net::parse_and_resolve_address; use rustls::crypto::aws_lc_rs::default_provider; use std::net::{IpAddr, Ipv4Addr, Ipv6Addr, SocketAddr}; use std::path::{Path, PathBuf}; -use std::sync::{ - Arc, - atomic::{AtomicBool, Ordering}, -}; +use std::sync::atomic::{AtomicBool, Ordering}; use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, warn}; +use tracing::{debug, info, warn}; const LOG_COMPONENT_EMBEDDED: &str = "embedded"; const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded"; @@ -302,110 +292,41 @@ impl RustFSServerBuilder { // Trusted proxies. rustfs_trusted_proxies::init(); - // Resolve listen address before endpoint/global initialization. - let server_addr = - parse_and_resolve_address(config.address.as_str()).map_err(|e| ServerError::Init(format!("address: {e}")))?; - - if server_addr.port() == 0 { - return Err(ServerError::Init( - "port 0 is not supported in embedded mode because startup requires \ - a stable listen address and port before endpoint/global initialization. \ - Use `find_available_port()` to obtain a free port." - .to_string(), - )); - } - - // Credentials. - init_global_action_credentials(Some(config.access_key.clone()), Some(config.secret_key.clone())) - .map_err(|e| ServerError::Init(format!("credentials: {e:?}")))?; - - // Region. - if let Some(region_str) = &config.region { - let region = region_str - .parse() - .map_err(|e| ServerError::Init(format!("invalid region '{region_str}': {e}")))?; - crate::storage_compat::set_global_region(region); - } - - let server_port = server_addr.port(); - - set_global_rustfs_port(server_port); - set_global_addr(&config.address).await; + let listen_context = init_embedded_startup_listen_context(&config) + .await + .map_err(|e| ServerError::Init(e.to_string()))?; set_global_init_guard()?; - // Endpoints / erasure setup. - let server_addr_str = server_addr.to_string(); - let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_addr_str.as_str(), config.volumes.clone()) + let endpoint_pools = init_embedded_startup_storage_foundation(&listen_context.server_address, &config.volumes) .await - .map_err(|e| ServerError::Init(format!("endpoints: {e}")))?; - enforce_unsupported_fs_policy(&endpoint_pools).map_err(|e| ServerError::Init(format!("unsupported fs guard: {e}")))?; - - set_global_endpoints(endpoint_pools.as_ref().clone()); - update_erasure_type(setup_type).await; - - // Local disks. - init_local_disks(endpoint_pools.clone()) - .await - .map_err(|e| ServerError::Init(format!("local disks: {e}")))?; - init_lock_clients(endpoint_pools.clone()); - - // Service state. - let readiness = Arc::new(GlobalReadiness::new()); + .map_err(|e| ServerError::Init(e.to_string()))?; // Start HTTP server. let mut s3_config = config.clone(); s3_config.console_enable = false; - let (shutdown_handle, bound_addr) = start_http_server(&s3_config, readiness.clone()).await?; + let (shutdown_handle, bound_addr) = start_http_server(&s3_config, listen_context.readiness.clone()).await?; let ctx = CancellationToken::new(); let shutdown_embedded_server = || { shutdown_handle.signal(); ctx.cancel(); }; - // Storage engine. - let store = match ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()).await { - Ok(store) => store, + let storage_runtime = match init_embedded_startup_storage_runtime( + listen_context.server_addr, + &endpoint_pools, + listen_context.readiness.clone(), + ctx.clone(), + ) + .await + { + Ok(runtime) => runtime, Err(e) => { - error!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_storage_init_failed", - stage = "ecstore_new", - error = ?e, - "Embedded storage initialization failed" - ); shutdown_embedded_server(); - return Err(ServerError::Init(format!("ECStore: {e}"))); + return Err(ServerError::Init(e.to_string())); } }; - - init_ecstore_config(); - try_migrate_server_config(store.clone()).await; - - // Global config system (with retry). - let mut retry = 0; - while let Err(e) = init_global_config_sys(store.clone()).await { - retry += 1; - if retry > 15 { - shutdown_embedded_server(); - return Err(ServerError::Init(format!("init_global_config_sys failed after 15 retries: {e}"))); - } - debug!( - component = LOG_COMPONENT_EMBEDDED, - subsystem = LOG_SUBSYSTEM_EMBEDDED, - event = "embedded_storage_init_retry", - stage = "global_config_sys", - retry, - error = %e, - "Embedded storage initialization retry scheduled" - ); - tokio::time::sleep(std::time::Duration::from_secs(1)).await; - } - readiness.mark_stage(SystemStage::StorageReady); - - // Replication. - init_background_replication(store.clone()).await; + let store = storage_runtime.store; // KMS (optional, non-fatal for embedded). if let Err(e) = init_kms_system(&config).await { @@ -455,12 +376,13 @@ impl RustFSServerBuilder { // 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, readiness.clone(), None, Some(ctx.clone())) - .await - .map_err(|e| { - shutdown_embedded_server(); - ServerError::Init(format!("IAM bootstrap setup: {e}")) - })?; + let iam_bootstrap = + bootstrap_or_defer_iam_init(store.clone(), kms_interface, listen_context.readiness.clone(), None, Some(ctx.clone())) + .await + .map_err(|e| { + shutdown_embedded_server(); + ServerError::Init(format!("IAM bootstrap setup: {e}")) + })?; // Bucket notifications. add_bucket_notification_configuration(buckets.clone()).await; @@ -476,7 +398,7 @@ impl RustFSServerBuilder { ); } - publish_ready_for_iam_bootstrap(iam_bootstrap, readiness.as_ref(), None) + publish_ready_for_iam_bootstrap(iam_bootstrap, listen_context.readiness.as_ref(), None) .await .map_err(|e| { shutdown_embedded_server(); diff --git a/rustfs/src/startup_server.rs b/rustfs/src/startup_server.rs index bd85f1d73..6af4ff2f8 100644 --- a/rustfs/src/startup_server.rs +++ b/rustfs/src/startup_server.rs @@ -44,6 +44,12 @@ pub struct StartupListenContext { pub server_address: String, } +pub struct EmbeddedStartupListenContext { + pub readiness: Arc, + pub server_addr: SocketAddr, + pub server_address: String, +} + pub struct StartupHttpServers { pub state_manager: Arc, pub s3_shutdown_tx: Option, @@ -99,6 +105,39 @@ pub async fn init_startup_listen_context(config: &Config) -> Result Result { + let readiness = Arc::new(GlobalReadiness::new()); + + let server_addr = + parse_and_resolve_address(config.address.as_str()).map_err(|err| Error::other(format!("address: {err}")))?; + if server_addr.port() == 0 { + return Err(Error::other( + "port 0 is not supported in embedded mode because startup requires \ + a stable listen address and port before endpoint/global initialization. \ + Use `find_available_port()` to obtain a free port.", + )); + } + + init_global_action_credentials(Some(config.access_key.clone()), Some(config.secret_key.clone())) + .map_err(|err| Error::other(format!("credentials: {err:?}")))?; + + if let Some(region_str) = &config.region { + region_str + .parse::() + .map(crate::storage_compat::set_global_region) + .map_err(|err| Error::other(format!("invalid region '{region_str}': {err}")))?; + } + + set_global_rustfs_port(server_addr.port()); + set_global_addr(&config.address).await; + + Ok(EmbeddedStartupListenContext { + readiness, + server_addr, + server_address: server_addr.to_string(), + }) +} + pub async fn init_startup_http_servers(config: &Config, readiness: Arc) -> Result { init_capacity_management().await; let state_manager = Arc::new(ServiceStateManager::new()); diff --git a/rustfs/src/startup_storage.rs b/rustfs/src/startup_storage.rs index aa4c6dcaf..7739c8719 100644 --- a/rustfs/src/startup_storage.rs +++ b/rustfs/src/startup_storage.rs @@ -28,12 +28,16 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; const LOG_COMPONENT_MAIN: &str = "main"; +const LOG_COMPONENT_EMBEDDED: &str = "embedded"; const LOG_SUBSYSTEM_STORAGE: &str = "storage"; +const LOG_SUBSYSTEM_EMBEDDED: &str = "embedded"; const GLOBAL_CONFIG_INIT_MAX_RETRIES: usize = 15; const EVENT_ENDPOINT_PARSING_STARTED: &str = "endpoint_parsing_started"; const EVENT_STARTUP_STORAGE_STAGE: &str = "startup_storage_stage"; const EVENT_STORAGE_POOL_FORMATTING: &str = "storage_pool_formatting"; const EVENT_STORAGE_POOL_HOST_RISK: &str = "storage_pool_host_risk"; +const EVENT_EMBEDDED_STORAGE_INIT_FAILED: &str = "embedded_storage_init_failed"; +const EVENT_EMBEDDED_STORAGE_INIT_RETRY: &str = "embedded_storage_init_retry"; pub struct StartupStorageRuntime { pub store: Arc, @@ -102,6 +106,23 @@ pub async fn init_startup_storage_foundation(server_address: &str, volumes: &[St Ok(endpoint_pools) } +pub async fn init_embedded_startup_storage_foundation(server_address: &str, volumes: &[String]) -> Result { + let (endpoint_pools, setup_type) = EndpointServerPools::from_volumes(server_address, volumes.to_vec()) + .await + .map_err(|err| Error::other(format!("endpoints: {err}")))?; + enforce_unsupported_fs_policy(&endpoint_pools).map_err(|err| Error::other(format!("unsupported fs guard: {err}")))?; + + set_global_endpoints(endpoint_pools.as_ref().clone()); + update_erasure_type(setup_type).await; + + init_local_disks(endpoint_pools.clone()) + .await + .map_err(|err| Error::other(format!("local disks: {err}")))?; + init_lock_clients(endpoint_pools.clone()); + + Ok(endpoint_pools) +} + pub async fn init_startup_storage_runtime( server_addr: SocketAddr, endpoint_pools: &EndpointServerPools, @@ -143,6 +164,34 @@ pub async fn init_startup_storage_runtime( }) } +pub async fn init_embedded_startup_storage_runtime( + server_addr: SocketAddr, + endpoint_pools: &EndpointServerPools, + readiness: Arc, + shutdown_token: CancellationToken, +) -> Result { + let store = match ECStore::new(server_addr, endpoint_pools.clone(), shutdown_token.clone()).await { + Ok(store) => store, + Err(err) => { + error!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = EVENT_EMBEDDED_STORAGE_INIT_FAILED, + stage = "ecstore_new", + error = ?err, + "Embedded storage initialization failed" + ); + return Err(Error::other(format!("ECStore: {err}"))); + } + }; + + init_embedded_startup_storage_global_config(store.clone()).await?; + readiness.mark_stage(SystemStage::StorageReady); + init_background_replication(store.clone()).await; + + Ok(StartupStorageRuntime { store, shutdown_token }) +} + async fn init_startup_storage_global_config(store: Arc) -> Result<()> { init_ecstore_config(); try_migrate_server_config(store.clone()).await; @@ -172,6 +221,33 @@ async fn init_startup_storage_global_config(store: Arc) -> Result<()> { Ok(()) } +async fn init_embedded_startup_storage_global_config(store: Arc) -> Result<()> { + init_ecstore_config(); + try_migrate_server_config(store.clone()).await; + + let mut retry = 0; + while let Err(err) = init_global_config_sys(store.clone()).await { + retry += 1; + if retry > GLOBAL_CONFIG_INIT_MAX_RETRIES { + return Err(Error::other(format!( + "init_global_config_sys failed after {GLOBAL_CONFIG_INIT_MAX_RETRIES} retries: {err}" + ))); + } + debug!( + component = LOG_COMPONENT_EMBEDDED, + subsystem = LOG_SUBSYSTEM_EMBEDDED, + event = EVENT_EMBEDDED_STORAGE_INIT_RETRY, + stage = "global_config_sys", + retry, + error = %err, + "Embedded storage initialization retry scheduled" + ); + tokio::time::sleep(tokio::time::Duration::from_secs(1)).await; + } + + Ok(()) +} + fn log_storage_pool_layout(endpoint_pools: &EndpointServerPools) { for (i, eps) in endpoint_pools.as_ref().iter().enumerate() { info!(