From d72e7f6f288f88bac805409ceb2e247c1a9c8413 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=AE=89=E6=AD=A3=E8=B6=85?= Date: Tue, 7 Apr 2026 23:24:23 +0800 Subject: [PATCH] Fix remaining review follow-ups for #2361 (#2421) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: 安正超 Co-authored-by: jeadie Co-authored-by: Jack Eadie Co-authored-by: loverustfs Co-authored-by: Copilot <175728472+Copilot@users.noreply.github.com> Co-authored-by: houseme Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com> Co-authored-by: houseme <4829346+houseme@users.noreply.github.com> --- .gitignore | 5 +- AGENTS.md | 10 + rustfs/Cargo.toml | 5 + rustfs/src/admin/handlers/group.rs | 2 +- rustfs/src/config/config_struct.rs | 33 +- rustfs/src/config/config_test.rs | 31 + rustfs/src/config/workload_profiles.rs | 2 +- rustfs/src/embedded.rs | 617 ++++++++++++++++++ rustfs/src/init.rs | 10 +- rustfs/src/lib.rs | 73 +++ rustfs/src/main.rs | 75 +-- rustfs/src/server/audit.rs | 4 +- rustfs/src/server/event.rs | 4 +- rustfs/src/server/http.rs | 20 +- rustfs/src/server/mod.rs | 29 +- rustfs/src/server/prefix.rs | 2 +- rustfs/src/server/runtime.rs | 4 +- rustfs/src/server/service_state.rs | 10 +- rustfs/src/server/tls_material.rs | 4 +- rustfs/src/storage/concurrency/io_schedule.rs | 7 + rustfs/src/storage/ecfs.rs | 6 + rustfs/tests/embedded_test.rs | 101 +++ 22 files changed, 964 insertions(+), 90 deletions(-) create mode 100644 rustfs/src/embedded.rs create mode 100644 rustfs/src/lib.rs create mode 100644 rustfs/tests/embedded_test.rs diff --git a/.gitignore b/.gitignore index b188c5c80..18d03f8a6 100644 --- a/.gitignore +++ b/.gitignore @@ -49,4 +49,7 @@ result* rustfs-webdav.code-workspace .aiexclude -*.bak \ No newline at end of file +*.bak +# Local test/benchmark artifacts +benchmarks.logs +tmp/ diff --git a/AGENTS.md b/AGENTS.md index 7b52de3e1..63b96b251 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -50,6 +50,11 @@ For code changes, run and pass the following before opening a PR: make pre-commit ``` +Before pushing code changes, make sure formatting is clean: + +- Run `cargo fmt --all`. +- Run `cargo fmt --all --check` and ensure no files are modified unexpectedly. + If `make` is unavailable, run the equivalent checks defined under `.config/make/`. Documentation-only or instruction-only changes are exempt from the verification commands above (including the `.config/make/` equivalents), though any installed git pre-commit hooks (for example, from `make setup-hooks`) may still run on commit unless explicitly skipped. After build-based verification completes, clean generated build artifacts before wrapping up to avoid unnecessary disk usage. @@ -64,6 +69,11 @@ Do not open a PR with code changes when the required checks fail. - Use `N/A` for non-applicable template sections. - Include verification commands in the PR description. - When using `gh pr create`/`gh pr edit`, use `--body-file` instead of inline `--body` for multiline markdown. +- After fixing code review comments or CI findings, always mark corresponding review + comments/threads as resolved before returning to the user. +- In handling review comments, confirm the underlying issue before changing code. + If a suggested change is not appropriate for behavior or risk, reply with a + concise rationale instead of blindly applying it. ## Security Baseline diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 7c3653c30..3b314d08e 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -27,6 +27,10 @@ categories.workspace = true documentation = "https://docs.rustfs.com/" # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html +[lib] +name = "rustfs" +path = "src/lib.rs" + [[bin]] name = "rustfs" path = "src/main.rs" @@ -89,6 +93,7 @@ rustfs-io-metrics = { workspace = true } rustfs-object-io = { workspace = true } rustfs-concurrency = { workspace = true } rustfs-scanner = { workspace = true } +tempfile = { workspace = true } # Async Runtime and Networking async-trait = { workspace = true } diff --git a/rustfs/src/admin/handlers/group.rs b/rustfs/src/admin/handlers/group.rs index b99977e6e..276502085 100644 --- a/rustfs/src/admin/handlers/group.rs +++ b/rustfs/src/admin/handlers/group.rs @@ -181,7 +181,7 @@ impl Operation for GetGroup { /// - `500 Internal Server Error` - Server-side error /// /// # Example -/// ``` +/// ```text /// DELETE /rustfs/admin/v3/group/developers /// ``` pub struct DeleteGroup {} diff --git a/rustfs/src/config/config_struct.rs b/rustfs/src/config/config_struct.rs index 74fbe5183..23aee4edb 100644 --- a/rustfs/src/config/config_struct.rs +++ b/rustfs/src/config/config_struct.rs @@ -19,7 +19,9 @@ use super::Opt; use crate::apply_external_env_compat; -use rustfs_config::{ENV_RUSTFS_ROOT_PASSWORD, ENV_RUSTFS_ROOT_USER, RUSTFS_REGION}; +use rustfs_config::{ + DEFAULT_CONSOLE_ADDRESS, DEFAULT_CONSOLE_ENABLE, ENV_RUSTFS_ROOT_PASSWORD, ENV_RUSTFS_ROOT_USER, RUSTFS_REGION, +}; use rustfs_credentials::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, Masked}; /// Helper function to resolve credentials from multiple sources with precedence: @@ -111,6 +113,35 @@ pub struct Config { } impl Config { + /// Create a `Config` with sensible defaults for the given volumes and address. + /// + /// This is the programmatic alternative to [`Opt::parse_command`] which reads + /// from the CLI / environment. Useful for embedded / integration-test usage. + pub fn new(address: impl Into, volumes: Vec) -> Self { + Config { + volumes, + address: address.into(), + server_domains: Vec::new(), + access_key: DEFAULT_ACCESS_KEY.to_string(), + secret_key: DEFAULT_SECRET_KEY.to_string(), + console_enable: DEFAULT_CONSOLE_ENABLE, + console_address: DEFAULT_CONSOLE_ADDRESS.to_string(), + obs_endpoint: rustfs_config::DEFAULT_OBS_ENDPOINT.to_string(), + tls_path: None, + license: None, + region: Some(RUSTFS_REGION.to_string()), + kms_enable: false, + kms_backend: "local".to_string(), + kms_key_dir: None, + kms_vault_address: None, + kms_vault_token: None, + kms_vault_mount_path: None, + kms_default_key_id: None, + buffer_profile_disable: false, + buffer_profile: "GeneralPurpose".to_string(), + } + } + /// Create Config from Opt pub(super) fn from_opt(opt: Opt) -> std::io::Result { let Opt { diff --git a/rustfs/src/config/config_test.rs b/rustfs/src/config/config_test.rs index 63e908138..76de6d749 100644 --- a/rustfs/src/config/config_test.rs +++ b/rustfs/src/config/config_test.rs @@ -16,6 +16,8 @@ #[allow(unsafe_op_in_unsafe_fn)] mod tests { use crate::config::{Config, Opt}; + use rustfs_config::{DEFAULT_CONSOLE_ADDRESS, DEFAULT_CONSOLE_ENABLE, DEFAULT_OBS_ENDPOINT, RUSTFS_REGION}; + use rustfs_credentials::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY}; use rustfs_ecstore::disks_layout::DisksLayout; use serial_test::serial; use std::env; @@ -80,6 +82,35 @@ mod tests { assert_eq!(opt.address, ":9000"); } + #[test] + #[serial] + fn test_config_new_defaults() { + let volumes = vec!["/tmp/rustfs-vol1".to_string()]; + let address = "127.0.0.1:9100".to_string(); + let config = Config::new(&address, volumes.clone()); + + assert_eq!(config.volumes, volumes); + assert_eq!(config.address, address); + assert_eq!(config.server_domains, Vec::::new()); + assert_eq!(config.access_key, DEFAULT_ACCESS_KEY); + assert_eq!(config.secret_key, DEFAULT_SECRET_KEY); + assert_eq!(config.console_enable, DEFAULT_CONSOLE_ENABLE); + assert_eq!(config.console_address, DEFAULT_CONSOLE_ADDRESS); + assert_eq!(config.obs_endpoint, DEFAULT_OBS_ENDPOINT); + assert_eq!(config.tls_path, None); + assert_eq!(config.license, None); + assert_eq!(config.region, Some(RUSTFS_REGION.to_string())); + assert!(!config.kms_enable); + assert_eq!(config.kms_backend, "local"); + assert_eq!(config.kms_key_dir, None); + assert_eq!(config.kms_vault_address, None); + assert_eq!(config.kms_vault_token, None); + assert_eq!(config.kms_vault_mount_path, None); + assert_eq!(config.kms_default_key_id, None); + assert!(!config.buffer_profile_disable); + assert_eq!(config.buffer_profile, "GeneralPurpose"); + } + #[test] #[serial] fn test_custom_console_configuration() { diff --git a/rustfs/src/config/workload_profiles.rs b/rustfs/src/config/workload_profiles.rs index e5426ba89..404f7117b 100644 --- a/rustfs/src/config/workload_profiles.rs +++ b/rustfs/src/config/workload_profiles.rs @@ -122,7 +122,7 @@ impl WorkloadProfile { /// /// # Examples /// ``` - /// use rustfs::config::workload_profiles::WorkloadProfile; + /// use rustfs::config::WorkloadProfile; /// /// let profile = WorkloadProfile::from_name("AiTraining"); /// let profile2 = WorkloadProfile::from_name("aitraining"); // case-insensitive diff --git a/rustfs/src/embedded.rs b/rustfs/src/embedded.rs new file mode 100644 index 000000000..218cdbac2 --- /dev/null +++ b/rustfs/src/embedded.rs @@ -0,0 +1,617 @@ +// 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. + +//! Embedded RustFS server for integration testing. +//! +//! Start a fully-functional S3-compatible server in-process without Docker +//! or child processes. Perfect for integration tests that need a local S3 +//! endpoint. +//! +//! # Quick start +//! +//! ```rust,no_run +//! use rustfs::embedded::{find_available_port, RustFSServerBuilder}; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), Box> { +//! let port = find_available_port()?; +//! let server = RustFSServerBuilder::new() +//! .address(format!("127.0.0.1:{port}")) +//! .access_key("minioadmin") +//! .secret_key("minioadmin") +//! .build() +//! .await?; +//! +//! println!("S3 endpoint: {}", server.endpoint()); +//! // ... use any S3 client ... +//! server.shutdown().await; +//! Ok(()) +//! } +//! ``` +//! +//! # Limitations +//! +//! Only **one `RustFSServer`** may exist per process because the underlying +//! storage engine uses process-global singletons (`OnceLock`). Attempting to +//! start a second server will return an error. + +use crate::app::context::{AppContext, init_global_app_context}; +use crate::config::Config; +use crate::init::{add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system}; +use crate::server::{ + ServiceState, ServiceStateManager, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, + stop_audit_system, +}; +use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; +use rustfs_credentials::init_global_action_credentials; +use rustfs_ecstore::store::init_lock_clients; +use rustfs_ecstore::{ + bucket::replication::init_background_replication, + bucket::{ + metadata_sys::init_bucket_metadata_sys, + migration::{try_migrate_bucket_metadata, try_migrate_iam_config}, + }, + 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, + store_api::BucketOperations, + store_api::BucketOptions, + update_erasure_type, +}; +use rustfs_iam::init_iam_sys; +use rustfs_obs::{init_obs, set_global_guard}; +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 tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +/// Tracks whether a server has been started in this process. +static SERVER_STARTED: AtomicBool = AtomicBool::new(false); + +/// Error type for embedded server operations. +#[derive(Debug)] +pub enum ServerError { + /// A server has already been started in this process. + AlreadyStarted, + /// The server failed to initialize. + Init(String), + /// An I/O error occurred. + Io(std::io::Error), +} + +impl std::fmt::Display for ServerError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ServerError::AlreadyStarted => write!( + f, + "A RustFS server has already been started in this process. \ + Only one embedded server is supported due to global state." + ), + ServerError::Init(msg) => write!(f, "RustFS initialization failed: {msg}"), + ServerError::Io(e) => write!(f, "I/O error: {e}"), + } + } +} + +impl std::error::Error for ServerError { + fn source(&self) -> Option<&(dyn std::error::Error + 'static)> { + match self { + ServerError::Io(e) => Some(e), + _ => None, + } + } +} + +impl From for ServerError { + fn from(e: std::io::Error) -> Self { + ServerError::Io(e) + } +} + +/// Builder for configuring and starting an embedded RustFS server. +/// +/// # Examples +/// +/// ```rust,no_run +/// # async fn example() -> Result<(), Box> { +/// use rustfs::embedded::RustFSServerBuilder; +/// +/// let server = RustFSServerBuilder::new() +/// .address("127.0.0.1:9100") +/// .access_key("mykey") +/// .secret_key("mysecret") +/// .volume("/tmp/rustfs-data") +/// .build() +/// .await?; +/// # Ok(()) +/// # } +/// ``` +pub struct RustFSServerBuilder { + address: String, + access_key: String, + secret_key: String, + volumes: Vec, + region: String, +} + +impl Default for RustFSServerBuilder { + fn default() -> Self { + Self::new() + } +} + +impl RustFSServerBuilder { + /// Create a new builder with sensible defaults. + /// + /// Defaults: + /// - address: `"127.0.0.1:9000"` + /// - access_key / secret_key: `"minioadmin"` + /// - region: `"us-east-1"` + /// - A temporary directory is created automatically for data storage + /// + /// Use [`find_available_port`] to pick a free port when the default is + /// not suitable. + pub fn new() -> Self { + Self { + address: "127.0.0.1:9000".to_string(), + access_key: "minioadmin".to_string(), + secret_key: "minioadmin".to_string(), + volumes: Vec::new(), + region: "us-east-1".to_string(), + } + } + + /// Set the listen address (e.g. `"127.0.0.1:9000"`). + /// + /// Use [`find_available_port`] to obtain a free port when the default is + /// not suitable. Port `0` is **not** supported because startup requires + /// a concrete listen address and port during initialization. + /// + /// The bound address is available via [`RustFSServer::address`] after + /// [`build`](Self::build), but that is too late for the earlier + /// initialization that depends on the configured address. + pub fn address(mut self, addr: impl Into) -> Self { + self.address = addr.into(); + self + } + + /// Set the S3 access key (default: `"minioadmin"`). + pub fn access_key(mut self, key: impl Into) -> Self { + self.access_key = key.into(); + self + } + + /// Set the S3 secret key (default: `"minioadmin"`). + pub fn secret_key(mut self, key: impl Into) -> Self { + self.secret_key = key.into(); + self + } + + /// Set the AWS region (default: `"us-east-1"`). + pub fn region(mut self, region: impl Into) -> Self { + self.region = region.into(); + self + } + + /// Add a data volume path. + /// + /// If no volumes are added, a temporary directory with a single drive is + /// created automatically (and cleaned up on [`RustFSServer::shutdown`]). + pub fn volume(mut self, path: impl Into) -> Self { + self.volumes.push(path.into()); + self + } + + /// Set multiple volume paths at once, replacing any previously set volumes. + pub fn volumes(mut self, paths: Vec) -> Self { + self.volumes = paths; + self + } + + /// Build and start the embedded server. + /// + /// Returns a [`RustFSServer`] handle that provides the endpoint URL and + /// a [`shutdown`](RustFSServer::shutdown) method. + /// + /// # Errors + /// + /// Returns [`ServerError::AlreadyStarted`] if another server is already + /// running in this process, or if another startup attempt has already + /// entered irreversible global initialization. + pub async fn build(mut self) -> Result { + self.do_build().await + } + + /// Inner build implementation. Separated from [`build`] so the outer + /// method can enforce the one-shot process-global startup guard. + async fn do_build(&mut self) -> Result { + // Build is allowed to fail before irreversible global initialization + // (for example on temporary I/O or directory setup errors), and in that + // case callers can retry. + let mut global_init_started = false; + let mut set_global_init_guard = || -> Result<(), ServerError> { + if global_init_started { + return Ok(()); + } + if SERVER_STARTED + .compare_exchange(false, true, Ordering::SeqCst, Ordering::SeqCst) + .is_err() + { + return Err(ServerError::AlreadyStarted); + } + global_init_started = true; + Ok(()) + }; + + // Keep a TempDir guard alive so that if build fails the directory is + // cleaned up automatically. We disarm (keep) on success. + let mut temp_dir_guard: Option = None; + if self.volumes.is_empty() { + let dir = tempfile::tempdir().map_err(|e| ServerError::Init(format!("failed to create temp dir: {e}")))?; + self.volumes.push(dir.path().display().to_string()); + temp_dir_guard = Some(dir); + } + + // Ensure volume directories exist. + for v in &self.volumes { + let p = Path::new(v); + if !p.exists() { + tokio::fs::create_dir_all(p) + .await + .map_err(|e| ServerError::Init(format!("failed to create volume dir {v}: {e}")))?; + } + } + + // Build Config. + let mut config = Config::new(&self.address, self.volumes.clone()); + config.access_key = self.access_key.clone(); + config.secret_key = self.secret_key.clone(); + config.region = Some(self.region.clone()); + config.console_enable = false; + + // --- Initialization sequence (mirrors main.rs::run) --- + + // Observability (minimal / no-op endpoint for embedded use). + let guard = init_obs(Some(config.obs_endpoint.clone())) + .await + .map_err(|e| ServerError::Init(format!("init_obs: {e}")))?; + set_global_guard(guard).map_err(|e| ServerError::Init(format!("set_global_guard: {e}")))?; + + // Crypto provider. + if let Err(err) = default_provider().install_default() { + debug!("Ignoring crypto provider installation error: {err:?}"); + } + + // Trusted proxies. + rustfs_trusted_proxies::init(); + + // 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}")))?; + rustfs_ecstore::global::set_global_region(region); + } + + // Resolve listen address. + 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(), + )); + } + + let server_port = server_addr.port(); + + set_global_rustfs_port(server_port); + set_global_addr(&config.address).await; + + 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()) + .await + .map_err(|e| ServerError::Init(format!("endpoints: {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()); + let state_manager = ServiceStateManager::new(); + state_manager.update(ServiceState::Starting); + + // Start HTTP server. + let mut s3_config = config.clone(); + s3_config.console_enable = false; + let (shutdown_tx, bound_addr) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?; + let ctx = CancellationToken::new(); + let shutdown_embedded_server = || { + let _ = shutdown_tx.send(()); + ctx.cancel(); + }; + + // Storage engine. + let store = match ECStore::new(server_addr, endpoint_pools.clone(), ctx.clone()).await { + Ok(store) => store, + Err(e) => { + error!("ECStore::new {:?}", e); + shutdown_embedded_server(); + return Err(ServerError::Init(format!("ECStore: {e}"))); + } + }; + + ecconfig::init(); + ecconfig::try_migrate_server_config(store.clone()).await; + + // Global config system (with retry). + let mut retry = 0; + while let Err(e) = ecconfig::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!("init_global_config_sys retry {retry}: {e}"); + tokio::time::sleep(std::time::Duration::from_secs(1)).await; + } + readiness.mark_stage(SystemStage::StorageReady); + + // Replication. + init_background_replication(store.clone()).await; + + // KMS (optional, non-fatal for embedded). + if let Err(e) = init_kms_system(&config).await { + warn!("KMS initialization skipped: {e}"); + } + + // Buffer profiles. + init_buffer_profile_system(&config); + + // Event notifier. + init_event_notifier().await; + + // Audit (non-fatal). + if let Err(e) = start_audit_system().await { + warn!("Audit system: {e}"); + } + + // 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. + init_iam_sys(store.clone()).await.map_err(|e| { + shutdown_embedded_server(); + ServerError::Init(format!("IAM: {e}")) + })?; + readiness.mark_stage(SystemStage::IamReady); + + // App context. + let iam_interface = rustfs_iam::get().map_err(|e| { + shutdown_embedded_server(); + ServerError::Init(format!("IAM get: {e}")) + })?; + let kms_interface = + rustfs_kms::get_global_kms_service_manager().unwrap_or_else(rustfs_kms::init_global_kms_service_manager); + let _app_context = + init_global_app_context(AppContext::with_default_interfaces(store.clone(), iam_interface, kms_interface)); + + // Bucket notifications. + add_bucket_notification_configuration(buckets.clone()).await; + + // Notification system. + if let Err(e) = new_global_notification_sys(endpoint_pools.clone()).await { + warn!("notification system: {e}"); + } + + // Mark fully ready. + readiness.mark_stage(SystemStage::FullReady); + rustfs_common::set_global_init_time_now().await; + + let server = RustFSServer { + address: bound_addr, + access_key: self.access_key.clone(), + secret_key: self.secret_key.clone(), + region: self.region.clone(), + shutdown_tx: Some(shutdown_tx), + cancel_token: ctx, + temp_dir: temp_dir_guard.map(|g| g.keep()), + }; + + info!( + target: "rustfs::embedded", + "RustFS embedded server ready at http://{}", + server.endpoint_address() + ); + + Ok(server) + } +} + +/// A running embedded RustFS server. +/// +/// Use [`endpoint`](Self::endpoint) to get the HTTP URL for S3 clients. +/// Call [`shutdown`](Self::shutdown) to stop the server and clean up resources. +/// +/// Dropping the server performs best-effort synchronous cleanup and may leave +/// async or process-global subsystems running until process exit. +pub struct RustFSServer { + address: SocketAddr, + access_key: String, + secret_key: String, + region: String, + shutdown_tx: Option>, + cancel_token: CancellationToken, + temp_dir: Option, +} + +impl RustFSServer { + fn endpoint_address(&self) -> SocketAddr { + let ip = match self.address.ip() { + ip @ IpAddr::V4(v4) if !v4.is_unspecified() => ip, + IpAddr::V4(_) => IpAddr::V4(Ipv4Addr::LOCALHOST), + ip @ IpAddr::V6(v6) if !v6.is_unspecified() => ip, + IpAddr::V6(_) => IpAddr::V6(Ipv6Addr::LOCALHOST), + }; + + SocketAddr::new(ip, self.address.port()) + } + + /// The HTTP endpoint URL (e.g. `"http://127.0.0.1:54321"`). + /// + /// Pass this to your S3 client's `endpoint_url` setting. + pub fn endpoint(&self) -> String { + format!("http://{}", self.endpoint_address()) + } + + /// The bound socket address. + pub fn address(&self) -> SocketAddr { + self.address + } + + /// The configured access key. + pub fn access_key(&self) -> &str { + &self.access_key + } + + /// The configured secret key. + pub fn secret_key(&self) -> &str { + &self.secret_key + } + + /// The configured region. + pub fn region(&self) -> &str { + &self.region + } + + /// Gracefully stop the server and clean up resources. + pub async fn shutdown(mut self) { + self.do_shutdown().await; + } + + async fn do_shutdown(&mut self) { + info!(target: "rustfs::embedded", "Shutting down embedded RustFS server..."); + + // 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!("Failed to stop audit system during shutdown: {e}"); + } + + // Signal HTTP server to stop. + if let Some(tx) = self.shutdown_tx.take() { + let _ = tx.send(()); + } + + // Brief grace period for connections to drain. + tokio::time::sleep(std::time::Duration::from_secs(2)).await; + + // Clean up temp directory if we created it. + if let Some(ref dir) = self.temp_dir + && let Err(e) = tokio::fs::remove_dir_all(dir).await + { + warn!("Failed to clean up temp dir {}: {e}", dir.display()); + } + + info!(target: "rustfs::embedded", "Embedded RustFS server stopped."); + } +} + +impl Drop for RustFSServer { + fn drop(&mut self) { + // Best-effort synchronous cleanup. + self.cancel_token.cancel(); + if let Some(tx) = self.shutdown_tx.take() { + let _ = tx.send(()); + } + if let Some(ref dir) = self.temp_dir { + let _ = std::fs::remove_dir_all(dir); + } + } +} + +/// Find an available TCP port on localhost. +/// +/// Binds to port `0`, reads the OS-assigned port, then releases the socket. +/// The port is **best-effort**: another process could claim it before RustFS +/// binds (TOCTOU), but in practice this is reliable for testing. +/// +/// Use with [`RustFSServerBuilder::address`]: +/// +/// ```rust,no_run +/// use rustfs::embedded::{find_available_port, RustFSServerBuilder}; +/// +/// # async fn example() -> Result<(), Box> { +/// let port = find_available_port()?; +/// let server = RustFSServerBuilder::new() +/// .address(format!("127.0.0.1:{port}")) +/// .build() +/// .await?; +/// println!("Listening on port {port}"); +/// # Ok(()) +/// # } +/// ``` +pub fn find_available_port() -> Result { + let listener = std::net::TcpListener::bind("127.0.0.1:0")?; + let port = listener.local_addr()?.port(); + drop(listener); + Ok(port) +} diff --git a/rustfs/src/init.rs b/rustfs/src/init.rs index 172330b65..de6bf354f 100644 --- a/rustfs/src/init.rs +++ b/rustfs/src/init.rs @@ -28,7 +28,7 @@ use std::io::Error; use tracing::{debug, error, info, instrument, warn}; #[instrument] -pub(crate) fn print_server_info() { +pub fn print_server_info() { let current_year = jiff::Zoned::now().year(); // Use custom macros to print server information info!("RustFS Object Storage Server"); @@ -42,7 +42,7 @@ pub(crate) fn print_server_info() { /// This function checks if update checking is enabled via /// environment variable or default configuration. If enabled, /// it spawns an asynchronous task to check for updates with a timeout. -pub(crate) fn init_update_check() { +pub fn init_update_check() { let update_check_enable = env::var(ENV_UPDATE_CHECK) .unwrap_or_else(|_| DEFAULT_UPDATE_CHECK.to_string()) .parse::() @@ -104,7 +104,7 @@ fn arn_to_target_id(arn_str: &str) -> Result) { +pub async fn add_bucket_notification_configuration(buckets: Vec) { let global_region = rustfs_ecstore::global::get_global_region(); let region = global_region .as_ref() @@ -277,7 +277,7 @@ async fn configure_and_start_kms( /// /// Returns `std::io::Result<()>` indicating success or failure #[instrument(skip(config))] -pub(crate) async fn init_kms_system(config: &config::Config) -> std::io::Result<()> { +pub async fn init_kms_system(config: &config::Config) -> std::io::Result<()> { // Initialize global KMS service manager (starts in NotConfigured state) let service_manager = rustfs_kms::init_global_kms_service_manager(); @@ -331,7 +331,7 @@ pub(crate) async fn init_kms_system(config: &config::Config) -> std::io::Result< /// /// # Arguments /// * `config` - The application configuration options -pub(crate) fn init_buffer_profile_system(config: &config::Config) { +pub fn init_buffer_profile_system(config: &config::Config) { use crate::config::{RustFSBufferConfig, WorkloadProfile, init_global_buffer_config, set_buffer_profile_enabled}; // Whether buffer profiling is disabled or not, it is enabled by default, unless the user explicitly sets '--buffer-profile-disable' or 'RUSTFS_BUFFER_PROFILE_DISABLE=true' diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs new file mode 100644 index 000000000..44aef7b1c --- /dev/null +++ b/rustfs/src/lib.rs @@ -0,0 +1,73 @@ +// 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. + +//! RustFS — high-performance S3-compatible object storage. +//! +//! This library exposes the [`embedded`] module which lets you start a +//! fully-functional RustFS server **in-process** — ideal for integration +//! tests that need a local S3 endpoint without Docker or child processes. +//! +//! # Quick start +//! +//! ```rust,no_run +//! use rustfs::embedded::{find_available_port, RustFSServerBuilder}; +//! +//! #[tokio::main] +//! async fn main() -> Result<(), Box> { +//! let port = find_available_port()?; +//! let server = RustFSServerBuilder::new() +//! .address(format!("127.0.0.1:{port}")) +//! .access_key("minioadmin") +//! .secret_key("minioadmin") +//! .build() +//! .await?; +//! +//! println!("S3 endpoint: {}", server.endpoint()); +//! +//! // ... use any S3 client against server.endpoint() ... +//! +//! server.shutdown().await; +//! Ok(()) +//! } +//! ``` +//! +//! # Limitations +//! +//! Because the underlying storage engine uses process-global singletons, +//! **only one `RustFSServer` may exist per process**. Attempting to start +//! a second server will return an error. This is fine for integration +//! tests where you start one server in a background task, run all your +//! tests, and then shut it down. + +pub mod admin; +pub mod app; +pub mod auth; +pub mod auth_keystone; +pub mod capacity; +pub mod config; +pub mod embedded; +pub mod error; +pub mod init; +pub mod license; +pub mod profiling; +#[cfg(any(feature = "ftps", feature = "webdav"))] +pub mod protocols; +pub mod server; +pub mod storage; +pub mod update; +pub mod version; + +// Re-export from rustfs_utils so that config sub-modules can use +// `crate::apply_external_env_compat` without breaking. +pub use rustfs_utils::{ExternalEnvCompatReport, apply_external_env_compat}; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index d8a328df2..70095ad42 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -12,41 +12,24 @@ // See the License for the specific language governing permissions and // limitations under the License. -mod admin; -mod app; -mod auth; -mod auth_keystone; -mod capacity; -mod config; -mod error; -mod init; -mod license; -mod profiling; -#[cfg(any(feature = "ftps", feature = "webdav"))] -mod protocols; -mod server; -mod storage; -mod update; -mod version; - // Ensure the correct path for parse_license is imported -use crate::app::context::{AppContext, init_global_app_context}; -use crate::init::{ +use rustfs::app::context::{AppContext, init_global_app_context}; +use rustfs::init::{ add_bucket_notification_configuration, init_buffer_profile_system, init_kms_system, init_update_check, print_server_info, }; #[cfg(feature = "ftps")] -use crate::init::{init_ftp_system, init_ftps_system}; +use rustfs::init::{init_ftp_system, init_ftps_system}; #[cfg(feature = "webdav")] -use crate::init::init_webdav_system; +use rustfs::init::init_webdav_system; -use crate::capacity::capacity_integration::init_capacity_management; -use crate::server::{ +use rustfs::capacity::capacity_integration::init_capacity_management; +use rustfs::license::{current_license, init_license, license_status}; +use rustfs::server::{ SHUTDOWN_TIMEOUT, ServiceState, ServiceStateManager, ShutdownSignal, init_event_notifier, shutdown_event_notifier, start_audit_system, start_http_server, stop_audit_system, wait_for_shutdown, }; -use license::{current_license, init_license, license_status}; use rustfs_common::{GlobalReadiness, SystemStage, set_global_addr}; use rustfs_credentials::init_global_action_credentials; use rustfs_ecstore::store::init_lock_clients; @@ -95,8 +78,8 @@ static GLOBAL: tikv_jemallocator::Jemalloc = tikv_jemallocator::Jemalloc; not(all(target_os = "linux", target_env = "gnu", target_arch = "x86_64")) ))] #[global_allocator] -static GLOBAL: profiling::allocator::TracingAllocator = - profiling::allocator::TracingAllocator::new(mimalloc::MiMalloc); +static GLOBAL: rustfs::profiling::allocator::TracingAllocator = + rustfs::profiling::allocator::TracingAllocator::new(mimalloc::MiMalloc); #[cfg(target_os = "windows")] #[global_allocator] @@ -108,7 +91,7 @@ fn main() { } // Build Tokio runtime with optional dial9 telemetry support - let runtime = server::build_tokio_runtime().expect("Failed to build Tokio runtime"); + let runtime = rustfs::server::build_tokio_runtime().expect("Failed to build Tokio runtime"); let result = runtime.block_on(async_main()); if let Err(ref e) = result { // Use eprintln as tracing may not be initialized at this point @@ -154,7 +137,7 @@ fn format_external_prefix_mappings(report: &ExternalEnvCompatReport) -> String { async fn async_main() -> Result<()> { // Parse command line arguments let args: Vec = std::env::args().collect(); - let command_result = match config::Opt::parse_command(args) { + let command_result = match rustfs::config::Opt::parse_command(args) { Ok(result) => result, Err(e) => { eprintln!("Command parse failed, error: {}", e); @@ -163,19 +146,19 @@ async fn async_main() -> Result<()> { }; // Handle info command - if let config::CommandResult::Info(opts) = command_result { - config::execute_info(&opts); + if let rustfs::config::CommandResult::Info(opts) = command_result { + rustfs::config::execute_info(&opts); return Ok(()); } // Get config for server command let config = match command_result { - config::CommandResult::Server(cfg) => cfg, - config::CommandResult::Info(_) => unreachable!(), + rustfs::config::CommandResult::Server(cfg) => cfg, + rustfs::config::CommandResult::Info(_) => unreachable!(), }; // Initialize the global config snapshot for info command - config::init_config_snapshot(&config); + rustfs::config::init_config_snapshot(&config); // Initialize the configuration init_license(config.license.clone()); @@ -216,10 +199,10 @@ async fn async_main() -> Result<()> { } // print startup logo - info!("{}", server::LOGO); + info!("{}", rustfs::server::LOGO); // Initialize performance profiling if enabled - profiling::init_from_env().await; + rustfs::profiling::init_from_env().await; // Initialize trusted proxies system rustfs_trusted_proxies::init(); @@ -233,7 +216,7 @@ async fn async_main() -> Result<()> { // Server-side TLS acceptor is built separately inside start_http_server() // using the same TlsMaterialSnapshot loading logic. if let Some(tls_path) = &config.tls_path { - match server::tls_material::TlsMaterialSnapshot::load(tls_path).await { + match rustfs::server::tls_material::TlsMaterialSnapshot::load(tls_path).await { Ok(snapshot) => { snapshot.apply_outbound().await; info!(target: "rustfs::main", "TLS outbound material initialized from {}", tls_path); @@ -256,14 +239,14 @@ async fn async_main() -> Result<()> { } #[instrument(skip(config))] -async fn run(config: config::Config) -> Result<()> { +async fn run(config: rustfs::config::Config) -> Result<()> { debug!("config: {:?}", &config); // 1. Initialize global readiness tracker let readiness = Arc::new(GlobalReadiness::new()); if let Some(region_str) = &config.region { region_str - .parse() + .parse::() .map(rustfs_ecstore::global::set_global_region) .map_err(|e| Error::other(format!("invalid region '{}': {}", region_str, e)))?; } @@ -277,7 +260,7 @@ async fn run(config: config::Config) -> Result<()> { server_address = %server_address, ip = %server_addr.ip(), port = %server_port, - version = %version::get_version(), + version = %rustfs::version::get_version(), "Starting RustFS server at {}", &server_address ); @@ -353,14 +336,14 @@ async fn run(config: config::Config) -> Result<()> { let s3_shutdown_tx = { let mut s3_config = config.clone(); s3_config.console_enable = false; - let s3_shutdown_tx = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?; + let (s3_shutdown_tx, _) = start_http_server(&s3_config, state_manager.clone(), readiness.clone()).await?; Some(s3_shutdown_tx) }; let console_shutdown_tx = if config.console_enable && !config.console_address.is_empty() { let mut console_config = config.clone(); console_config.address = console_config.console_address.clone(); - let console_shutdown_tx = start_http_server(&console_config, state_manager.clone(), readiness.clone()).await?; + let (console_shutdown_tx, _) = start_http_server(&console_config, state_manager.clone(), readiness.clone()).await?; Some(console_shutdown_tx) } else { None @@ -468,7 +451,7 @@ async fn run(config: config::Config) -> Result<()> { } // Initialize deadlock detector if enabled - let detector = storage::deadlock_detector::get_deadlock_detector(); + let detector = rustfs::storage::deadlock_detector::get_deadlock_detector(); if detector.is_enabled() { detector.start(); info!(target: "rustfs::main::run","Deadlock detector started successfully."); @@ -504,7 +487,7 @@ async fn run(config: config::Config) -> Result<()> { // 3a. Initialize Keystone authentication if enabled let keystone_config = rustfs_keystone::KeystoneConfig::from_env().map_err(Error::other)?; if keystone_config.enable { - match auth_keystone::init_keystone_auth(keystone_config).await { + match rustfs::auth_keystone::init_keystone_auth(keystone_config).await { Ok(_) => info!("Keystone authentication initialized successfully"), Err(e) => { error!("Failed to initialize Keystone authentication: {}", e); @@ -569,13 +552,13 @@ async fn run(config: config::Config) -> Result<()> { init_metrics_system(ctx.clone()); // Initialize auto-tuner for performance optimization (optional) - crate::init::init_auto_tuner(ctx.clone()).await; + rustfs::init::init_auto_tuner(ctx.clone()).await; } info!( target: "rustfs::main::run", "RustFS server version: {} started successfully at {}, current time: {}", - version::get_version(), + rustfs::version::get_version(), &server_address, jiff::Zoned::now() ); @@ -718,7 +701,7 @@ async fn handle_shutdown( target: "rustfs::main::handle_shutdown", "Stopping profiling tasks..." ); - profiling::shutdown_profiling(); + rustfs::profiling::shutdown_profiling(); info!( target: "rustfs::main::handle_shutdown", diff --git a/rustfs/src/server/audit.rs b/rustfs/src/server/audit.rs index 7809684cd..660683cef 100644 --- a/rustfs/src/server/audit.rs +++ b/rustfs/src/server/audit.rs @@ -40,7 +40,7 @@ fn has_any_audit_targets(config: &rustfs_ecstore::config::Config) -> bool { /// If configured, it initializes and starts the audit system. /// If not configured, it skips the initialization. /// It also handles cases where the audit system is already running or if the global configuration is not loaded. -pub(crate) async fn start_audit_system() -> AuditResult<()> { +pub async fn start_audit_system() -> AuditResult<()> { info!( target: "rustfs::main::start_audit_system", "Initializing the audit system..." @@ -118,7 +118,7 @@ pub(crate) async fn start_audit_system() -> AuditResult<()> { /// This function checks if the audit system is initialized and running. /// If it is running, it prepares to stop the system, stops it, and records the stop time. /// If the system is already stopped or not initialized, it logs a warning and returns. -pub(crate) async fn stop_audit_system() -> AuditResult<()> { +pub async fn stop_audit_system() -> AuditResult<()> { if let Some(system) = audit_system() { let state = system.get_state().await; if state == AuditSystemState::Stopped { diff --git a/rustfs/src/server/event.rs b/rustfs/src/server/event.rs index 64c1be00f..73a0af17c 100644 --- a/rustfs/src/server/event.rs +++ b/rustfs/src/server/event.rs @@ -62,7 +62,7 @@ fn install_ecstore_event_dispatch_hook() { } /// Shuts down the event notifier system gracefully -pub(crate) async fn shutdown_event_notifier() { +pub async fn shutdown_event_notifier() { info!("Shutting down event notifier system..."); if !rustfs_notify::is_notification_system_initialized() { @@ -84,7 +84,7 @@ pub(crate) async fn shutdown_event_notifier() { } #[instrument] -pub(crate) async fn init_event_notifier() { +pub async fn init_event_notifier() { info!( target: "rustfs::main::init_event_notifier", "Initializing event notifier..." diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index 6e0243d9a..55ff6abdb 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -70,9 +70,8 @@ pub async fn start_http_server( config: &config::Config, worker_state_manager: ServiceStateManager, readiness: Arc, -) -> Result> { +) -> Result<(tokio::sync::broadcast::Sender<()>, SocketAddr)> { let server_addr = parse_and_resolve_address(config.address.as_str()).map_err(Error::other)?; - let server_port = server_addr.port(); // The listening address and port are obtained from the parameters let listener = { @@ -184,6 +183,7 @@ pub async fn start_http_server( local_addr.ip() } }; + let local_port = local_addr.port(); let local_ip_str = if local_ip.is_ipv6() { format!("[{local_ip}]") @@ -192,19 +192,19 @@ pub async fn start_http_server( }; // Detailed endpoint information (showing all API endpoints) - let api_endpoints = format!("{protocol}://{local_ip_str}:{server_port}"); - let localhost_endpoint = format!("{protocol}://127.0.0.1:{server_port}"); + let api_endpoints = format!("{protocol}://{local_ip_str}:{local_port}"); + let localhost_endpoint = format!("{protocol}://127.0.0.1:{local_port}"); let now_time = jiff::Zoned::now().strftime("%Y-%m-%d %H:%M:%S").to_string(); if config.console_enable { - admin::console::init_console_cfg(local_ip, server_port); + admin::console::init_console_cfg(local_ip, local_port); info!( target: "rustfs::console::startup", - "Console WebUI available at: {protocol}://{local_ip_str}:{server_port}/rustfs/console/index.html" + "Console WebUI available at: {protocol}://{local_ip_str}:{local_port}/rustfs/console/index.html" ); info!( target: "rustfs::console::startup", - "Console WebUI (localhost): {protocol}://127.0.0.1:{server_port}/rustfs/console/index.html", + "Console WebUI (localhost): {protocol}://127.0.0.1:{local_port}/rustfs/console/index.html", ); } else { @@ -245,9 +245,9 @@ pub async fn start_http_server( for domain in &config.server_domains { domain_sets.insert(domain.to_string()); if let Some((host, _)) = domain.split_once(':') { - domain_sets.insert(format!("{host}:{server_port}")); + domain_sets.insert(format!("{host}:{local_port}")); } else { - domain_sets.insert(format!("{domain}:{server_port}")); + domain_sets.insert(format!("{domain}:{local_port}")); } } @@ -488,7 +488,7 @@ pub async fn start_http_server( worker_state_manager.update(ServiceState::Stopped); }); - Ok(shutdown_tx) + Ok((shutdown_tx, local_addr)) } #[derive(Clone)] diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index c460626ca..62c5ee6ca 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -23,19 +23,26 @@ mod prefix; mod readiness; mod runtime; mod service_state; -pub(crate) mod tls_material; +pub mod tls_material; -pub(crate) use audit::{start_audit_system, stop_audit_system}; -pub(crate) use event::{init_event_notifier, shutdown_event_notifier}; -pub(crate) use http::start_http_server; -pub(crate) use prefix::*; +// Items used by main.rs (binary crate) and/or embedded.rs — must be fully pub. +pub use audit::{start_audit_system, stop_audit_system}; +pub use event::{init_event_notifier, shutdown_event_notifier}; +pub use http::start_http_server; +pub use prefix::LOGO; +pub use runtime::build_tokio_runtime; +pub use service_state::SHUTDOWN_TIMEOUT; +pub use service_state::ServiceState; +pub use service_state::ServiceStateManager; +pub use service_state::ShutdownSignal; +pub use service_state::wait_for_shutdown; + +// Items only used within the library crate (admin handlers, server/http.rs, etc.). +pub(crate) use prefix::{ + ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, + PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, +}; pub(crate) use readiness::ReadinessGateLayer; -pub(crate) use runtime::build_tokio_runtime; -pub(crate) use service_state::SHUTDOWN_TIMEOUT; -pub(crate) use service_state::ServiceState; -pub(crate) use service_state::ServiceStateManager; -pub(crate) use service_state::ShutdownSignal; -pub(crate) use service_state::wait_for_shutdown; #[derive(Clone, Copy, Debug)] pub struct RemoteAddr(pub std::net::SocketAddr); diff --git a/rustfs/src/server/prefix.rs b/rustfs/src/server/prefix.rs index 6a61c8f97..47ab182fe 100644 --- a/rustfs/src/server/prefix.rs +++ b/rustfs/src/server/prefix.rs @@ -62,7 +62,7 @@ pub(crate) const RPC_PREFIX: &str = "/rustfs/rpc"; pub(crate) const TONIC_PREFIX: &str = "/node_service.NodeService"; /// LOGO art for RustFS server. -pub(crate) const LOGO: &str = r#" +pub const LOGO: &str = r#" ░█▀▄░█░█░█▀▀░▀█▀░█▀▀░█▀▀ ░█▀▄░█░█░▀▀█░░█░░█▀▀░▀▀█ diff --git a/rustfs/src/server/runtime.rs b/rustfs/src/server/runtime.rs index d0e8102ba..aacbe711b 100644 --- a/rustfs/src/server/runtime.rs +++ b/rustfs/src/server/runtime.rs @@ -91,7 +91,7 @@ fn compute_default_max_blocking_threads() -> usize { /// // let builder = tokio_runtime_builder(); /// // let runtime = builder.build().unwrap(); /// ``` -pub(crate) fn tokio_runtime_builder() -> tokio::runtime::Builder { +pub fn tokio_runtime_builder() -> tokio::runtime::Builder { let mut builder = tokio::runtime::Builder::new_multi_thread(); // Worker threads(Default physical cores) @@ -190,7 +190,7 @@ fn print_tokio_thread_enable() -> bool { /// // let runtime = build_tokio_runtime().expect("Failed to build runtime"); /// // runtime.block_on(async { /* ... */ }) /// ``` -pub(crate) fn build_tokio_runtime() -> Result { +pub fn build_tokio_runtime() -> Result { let mut builder = tokio_runtime_builder(); // Check if dial9 is enabled diff --git a/rustfs/src/server/service_state.rs b/rustfs/src/server/service_state.rs index ed9e509f9..420ab3264 100644 --- a/rustfs/src/server/service_state.rs +++ b/rustfs/src/server/service_state.rs @@ -19,7 +19,7 @@ use std::time::Duration; use tracing::info; // a configurable shutdown timeout -pub(crate) const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1); +pub const SHUTDOWN_TIMEOUT: Duration = Duration::from_secs(1); #[cfg(target_os = "linux")] fn notify_systemd(state: &str) { @@ -58,7 +58,7 @@ pub enum ShutdownSignal { #[atomic_enum] #[derive(PartialEq)] -pub(crate) enum ServiceState { +pub enum ServiceState { Starting, Ready, Stopping, @@ -66,7 +66,7 @@ pub(crate) enum ServiceState { } #[cfg(unix)] -pub(crate) async fn wait_for_shutdown() -> ShutdownSignal { +pub async fn wait_for_shutdown() -> ShutdownSignal { use tokio::signal::unix::{SignalKind, signal}; let mut sigterm = signal(SignalKind::terminate()).expect("failed to create SIGTERM signal handler"); let mut sigint = signal(SignalKind::interrupt()).expect("failed to create SIGINT signal handler"); @@ -88,7 +88,7 @@ pub(crate) async fn wait_for_shutdown() -> ShutdownSignal { } #[cfg(not(unix))] -pub(crate) async fn wait_for_shutdown() -> ShutdownSignal { +pub async fn wait_for_shutdown() -> ShutdownSignal { tokio::select! { _ = tokio::signal::ctrl_c() => { info!("Received Ctrl-C signal"); @@ -98,7 +98,7 @@ pub(crate) async fn wait_for_shutdown() -> ShutdownSignal { } #[derive(Clone)] -pub(crate) struct ServiceStateManager { +pub struct ServiceStateManager { state: Arc, } diff --git a/rustfs/src/server/tls_material.rs b/rustfs/src/server/tls_material.rs index f6caf9ada..47b3e9425 100644 --- a/rustfs/src/server/tls_material.rs +++ b/rustfs/src/server/tls_material.rs @@ -20,7 +20,7 @@ //! Usage: //! 1. Call `TlsMaterialSnapshot::load(tls_path)` once at startup. //! 2. Call `snapshot.apply_outbound()` to set global root CAs and mTLS identity. -//! 3. Call `snapshot.build_tls_acceptor(tls_path)` to build the server TLS acceptor. +//! 3. TLS acceptor construction is handled internally during server startup. use rustfs_common::{MtlsIdentityPem, set_global_mtls_identity, set_global_root_cert}; use rustfs_config::{ @@ -108,7 +108,7 @@ impl TlsMaterialSnapshot { /// This is the single place that constructs the server `ServerConfig`, /// handling both multi-cert (SNI resolver) and single-cert fallback. /// Returns `None` if no TLS certificates are available. - pub async fn build_tls_acceptor(&self, tls_path: &str) -> Result>, TlsMaterialError> { + pub(crate) async fn build_tls_acceptor(&self, tls_path: &str) -> Result>, TlsMaterialError> { if tls_path.is_empty() || !self.has_server_certs { return Ok(None); } diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index 8d5d8fc25..53c74570b 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -1713,6 +1713,13 @@ pub struct IoPriorityMetrics { pub low_processed: AtomicU64, } +#[allow(dead_code)] +impl Default for IoPriorityMetrics { + fn default() -> Self { + Self::new() + } +} + #[allow(dead_code)] impl IoPriorityMetrics { /// Create a new metrics instance. diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index d488a0209..2be0022a0 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -43,6 +43,12 @@ pub(crate) struct ListObjectUnorderedQuery { pub(crate) allow_unordered: Option, } +impl Default for FS { + fn default() -> Self { + Self::new() + } +} + impl FS { pub fn new() -> Self { rustfs_s3_common::init_s3_metrics(); diff --git a/rustfs/tests/embedded_test.rs b/rustfs/tests/embedded_test.rs new file mode 100644 index 000000000..93ff183df --- /dev/null +++ b/rustfs/tests/embedded_test.rs @@ -0,0 +1,101 @@ +// Integration test demonstrating the embedded RustFS server API. +// +// This test starts a RustFS server in-process and exercises it via the +// standard AWS S3 SDK — exactly as you would in your own integration tests. + +use aws_sdk_s3::config::{Credentials, Region}; +use aws_sdk_s3::primitives::ByteStream; +use aws_sdk_s3::{Client, Config}; +use rustfs::embedded::{RustFSServerBuilder, find_available_port}; + +/// Helper: create an S3 client pointed at the embedded server. +fn s3_client(endpoint: &str, access_key: &str, secret_key: &str) -> Client { + let creds = Credentials::new(access_key, secret_key, None, None, "test"); + let config = Config::builder() + .credentials_provider(creds) + .region(Region::new("us-east-1")) + .endpoint_url(endpoint) + .force_path_style(true) + .behavior_version_latest() + .build(); + Client::from_conf(config) +} + +#[tokio::test] +async fn test_embedded_server_basic_s3_operations() { + // 1. Pick a free port and start the embedded server. + let port = find_available_port().expect("find free port"); + let server = RustFSServerBuilder::new() + .address(format!("127.0.0.1:{port}")) + .access_key("testaccesskey") + .secret_key("testsecretkey") + .build() + .await + .expect("start embedded server"); + + let endpoint = server.endpoint(); + assert!(endpoint.contains(&port.to_string())); + + // 2. Create an S3 client and perform basic operations. + let client = s3_client(&endpoint, server.access_key(), server.secret_key()); + + // Create bucket + client + .create_bucket() + .bucket("test-bucket") + .send() + .await + .expect("create bucket"); + + // Put object + let body = ByteStream::from_static(b"hello rustfs embedded!"); + client + .put_object() + .bucket("test-bucket") + .key("greeting.txt") + .body(body) + .send() + .await + .expect("put object"); + + // Get object + let resp = client + .get_object() + .bucket("test-bucket") + .key("greeting.txt") + .send() + .await + .expect("get object"); + + let data = resp.body.collect().await.expect("read body").into_bytes(); + assert_eq!(data.as_ref(), b"hello rustfs embedded!"); + + // List objects + let list = client + .list_objects_v2() + .bucket("test-bucket") + .send() + .await + .expect("list objects"); + assert_eq!(list.key_count(), Some(1)); + + // Delete object + client + .delete_object() + .bucket("test-bucket") + .key("greeting.txt") + .send() + .await + .expect("delete object"); + + // Delete bucket + client + .delete_bucket() + .bucket("test-bucket") + .send() + .await + .expect("delete bucket"); + + // 3. Shut down. + server.shutdown().await; +}