From ee21eebd94a89c4496f9b8c222c756453cf4347e Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 01:57:28 +0800 Subject: [PATCH 01/15] feat: add optional RDMA object data integration --- .gitignore | 4 + .gitmodules | 4 + Cargo.toml | 1 + README.md | 3 + crates/cli/src/commands/mod.rs | 20 +- crates/cli/src/commands/rb.rs | 3 + crates/cli/src/launcher.rs | 276 +++++ crates/cli/src/lib.rs | 2 + crates/cli/src/main.rs | 281 +---- crates/cli/tests/help_contract.rs | 1 + crates/core/src/admin/kms_diagnostic.rs | 9 + crates/core/src/data_transport.rs | 118 ++ crates/core/src/error.rs | 28 +- crates/core/src/lib.rs | 5 + crates/s3/src/client.rs | 1321 +++++++++++++++++++++-- crates/s3/src/data_transport.rs | 271 +++++ crates/s3/src/lib.rs | 1 + docs/ADR/0001-object-data-transport.md | 50 + docs/rdma-build.md | 48 + rdma/Cargo.toml | 33 + scripts/build-rdma.sh | 15 + scripts/tests/test_build_rdma.py | 84 ++ vendor/rustfs-ucx | 1 + 23 files changed, 2229 insertions(+), 350 deletions(-) create mode 100644 .gitmodules create mode 100644 crates/cli/src/launcher.rs create mode 100644 crates/core/src/data_transport.rs create mode 100644 crates/s3/src/data_transport.rs create mode 100644 docs/ADR/0001-object-data-transport.md create mode 100644 docs/rdma-build.md create mode 100644 rdma/Cargo.toml create mode 100755 scripts/build-rdma.sh create mode 100644 scripts/tests/test_build_rdma.py create mode 160000 vendor/rustfs-ucx diff --git a/.gitignore b/.gitignore index d31e730d..f0844ff5 100644 --- a/.gitignore +++ b/.gitignore @@ -27,3 +27,7 @@ coverage/ # Cargo lock for library crates (keep for binary) # Cargo.lock CLAUDE.md + +# The independent opt-in build owns its artifacts and lockfile. +/rdma/target/ +/rdma/Cargo.lock diff --git a/.gitmodules b/.gitmodules new file mode 100644 index 00000000..47b6db4a --- /dev/null +++ b/.gitmodules @@ -0,0 +1,4 @@ +[submodule "vendor/rustfs-ucx"] + path = vendor/rustfs-ucx + url = git@github.com:rustfs/rustfs-ucx.git + branch = s3-rdma diff --git a/Cargo.toml b/Cargo.toml index b12aadf0..d69fa11d 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,4 +1,5 @@ [workspace] +exclude = ["rdma", "vendor/rustfs-ucx"] resolver = "3" members = ["crates/cli", "crates/core", "crates/s3"] diff --git a/README.md b/README.md index feecf87d..6404444b 100644 --- a/README.md +++ b/README.md @@ -71,6 +71,9 @@ cd cli cargo build --release ``` +Optional object-data interfaces and the explicit private build are documented +in [RDMA builds](docs/rdma-build.md). The ordinary build requires no private source. + ## Quick Start ### Configure Aliases diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index c6355f07..20df0298 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -8,7 +8,10 @@ use std::io::{IsTerminal, stderr, stdout}; use clap::{Parser, Subcommand, ValueEnum}; use rc_core::config::Defaults; -use rc_core::{ConfigManager, RequestHeader, set_global_request_headers}; +use rc_core::{ + ConfigManager, DataTransportMode, RequestHeader, set_global_data_transport_mode, + set_global_request_headers, +}; use crate::exit_code::ExitCode; use crate::output::{Formatter, OutputConfig}; @@ -90,6 +93,10 @@ fn validate_version_selector(version_id: Option<&str>, rewind: Option<&str>) -> #[command(after_help = COMPATIBILITY_AFTER_HELP)] #[command(propagate_version = true)] pub struct Cli { + /// Object data transport: http, auto, or rdma (requires an installed provider) + #[arg(long, global = true, default_value = "http")] + pub data_transport: DataTransportMode, + /// Output format: auto-detect, human-readable, or JSON #[arg(long, global = true, value_enum)] pub format: Option, @@ -357,6 +364,7 @@ pub enum Commands { /// Execute the CLI command and return an exit code pub async fn execute(cli: Cli) -> ExitCode { + set_global_data_transport_mode(cli.data_transport); set_global_request_headers(cli.request_headers.clone()); let cli_output_options = GlobalOutputOptions::from_cli(&cli); let defaults = match ConfigManager::new() { @@ -564,6 +572,16 @@ pub async fn execute(cli: Cli) -> ExitCode { #[cfg(test)] mod tests { use super::*; + + #[test] + fn data_transport_selection_is_explicit_and_http_is_default() { + let cli = Cli::try_parse_from(["rc", "alias", "list"]).expect("default CLI"); + assert_eq!(cli.data_transport, DataTransportMode::Http); + let cli = Cli::try_parse_from(["rc", "--data-transport", "rdma", "alias", "list"]) + .expect("explicit external data transport"); + assert_eq!(cli.data_transport, DataTransportMode::Rdma); + assert!(Cli::try_parse_from(["rc", "--data-transport", "udp", "alias", "list"]).is_err()); + } use clap::Parser; #[test] diff --git a/crates/cli/src/commands/rb.rs b/crates/cli/src/commands/rb.rs index e32189f1..47d1f474 100644 --- a/crates/cli/src/commands/rb.rs +++ b/crates/cli/src/commands/rb.rs @@ -862,6 +862,9 @@ fn bucket_delete_message(error: &Error, bucket: &str) -> String { fn exit_code_for_error(error: &Error) -> ExitCode { match error { + Error::DataTransfer { .. } => { + ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError) + } Error::InvalidPath(_) | Error::Config(_) => ExitCode::UsageError, Error::Network(_) => ExitCode::NetworkError, Error::Auth(_) => ExitCode::AuthError, diff --git a/crates/cli/src/launcher.rs b/crates/cli/src/launcher.rs new file mode 100644 index 00000000..5305c5f5 --- /dev/null +++ b/crates/cli/src/launcher.rs @@ -0,0 +1,276 @@ +//! rc - Rust S3 CLI Client +//! +//! A command-line interface for S3-compatible object storage services. +//! Designed for RustFS and other S3-compatible backends. + +use clap::Parser; +use tracing_subscriber::{EnvFilter, fmt, prelude::*}; + +use crate::commands::{self, Cli}; +use crate::exit_code::ExitCode; + +const COMMAND_THREAD_NAME: &str = "rc-command"; +// AWS SDK command futures can exceed the Windows process main-thread stack. A 16 MiB +// worker stack provides headroom for the largest transfer commands observed in CI. +const COMMAND_THREAD_STACK_SIZE: usize = 16 * 1024 * 1024; + +#[derive(Debug, thiserror::Error)] +enum LauncherError { + #[error("Failed to start command thread: {0}")] + ThreadSpawn(#[source] std::io::Error), + + #[error("Failed to initialize async runtime: {0}")] + Runtime(#[source] std::io::Error), + + #[error("Command execution thread terminated unexpectedly")] + ThreadPanicked, +} + +fn build_env_filter(debug: bool, env_is_set: bool) -> EnvFilter { + if debug && !env_is_set { + EnvFilter::new("debug") + } else { + EnvFilter::from_default_env() + } +} + +/// Run the shared CLI launcher, preserving its worker-thread and exit behavior. +pub fn run() { + let cli = Cli::parse(); + let env_is_set = std::env::var_os(EnvFilter::DEFAULT_ENV).is_some(); + let env_filter = build_env_filter(cli.debug, env_is_set); + + // Initialize tracing subscriber for logging + tracing_subscriber::registry() + .with(fmt::layer()) + .with(env_filter) + .init(); + + let exit_code = match run_command(cli) { + Ok(exit_code) => exit_code, + Err(error) => { + eprintln!("{error}"); + ExitCode::GeneralError + } + }; + + std::process::exit(exit_code.as_i32()); +} + +fn run_command(cli: Cli) -> Result { + run_async_command(move || commands::execute(cli), command_runtime) +} + +fn command_runtime() -> std::io::Result { + tokio::runtime::Builder::new_multi_thread() + .thread_stack_size(COMMAND_THREAD_STACK_SIZE) + .enable_all() + .build() +} + +fn run_async_command( + future_factory: F, + runtime_factory: R, +) -> Result +where + T: Send + 'static, + F: FnOnce() -> Fut + Send + 'static, + Fut: Future + 'static, + R: FnOnce() -> std::io::Result + Send + 'static, +{ + run_on_command_thread(move || { + let runtime = runtime_factory().map_err(LauncherError::Runtime)?; + let output = runtime.block_on(future_factory()); + // The former async main exited the process without waiting for runtime workers. + // Preserve that bounded shutdown behavior if a background task is wedged. + runtime.shutdown_background(); + Ok(output) + }) +} + +fn run_on_command_thread(command: F) -> Result +where + T: Send + 'static, + F: FnOnce() -> Result + Send + 'static, +{ + // The Windows process main thread has a smaller stack than the async command graph + // requires. Constructing and polling the command future here keeps it off that stack. + let command_thread = std::thread::Builder::new() + .name(COMMAND_THREAD_NAME.to_string()) + .stack_size(COMMAND_THREAD_STACK_SIZE) + .spawn(command) + .map_err(LauncherError::ThreadSpawn)?; + + command_thread + .join() + .map_err(|_| LauncherError::ThreadPanicked)? +} + +#[cfg(test)] +mod tests { + use super::*; + use std::sync::Arc; + use std::sync::atomic::{AtomicBool, Ordering}; + use std::sync::mpsc::sync_channel; + use std::time::Duration; + + fn test_runtime() -> std::io::Result { + tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + } + + #[test] + fn debug_uses_debug_filter_when_env_missing() { + let filter = build_env_filter(true, false); + assert_eq!(filter.to_string(), "debug"); + } + + #[test] + fn command_launcher_preserves_command_exit_code() { + let exit_code = run_async_command(|| async { ExitCode::UnsupportedFeature }, test_runtime) + .expect("command launcher should return its exit code"); + + assert_eq!(exit_code, ExitCode::UnsupportedFeature); + } + + #[test] + fn command_launcher_uses_the_named_worker_thread() { + let thread_name = run_async_command( + || async { std::thread::current().name().map(str::to_string) }, + test_runtime, + ) + .expect("command launcher should return its worker thread name"); + + assert_eq!(thread_name.as_deref(), Some(COMMAND_THREAD_NAME)); + assert_eq!(COMMAND_THREAD_STACK_SIZE, 16 * 1024 * 1024); + } + + #[test] + fn command_launcher_constructs_and_polls_large_future_on_worker() { + const LARGE_FUTURE_SIZE: usize = 2 * 1024 * 1024; + + async fn large_future() -> usize { + let marker = std::hint::black_box(7_u8); + let payload = [marker; LARGE_FUTURE_SIZE]; + tokio::task::yield_now().await; + let payload = std::hint::black_box(&payload); + usize::from(payload[0]) + usize::from(payload[LARGE_FUTURE_SIZE - 1]) + } + + let result = run_async_command(large_future, test_runtime) + .expect("large command future should complete on the sized worker stack"); + + assert_eq!(result, 14); + } + + #[test] + fn command_runtime_polls_spawned_large_future_on_sized_worker() { + const LARGE_FUTURE_SIZE: usize = 1024 * 1024; + + async fn large_future() -> usize { + let marker = std::hint::black_box(11_u8); + let payload = [marker; LARGE_FUTURE_SIZE]; + tokio::task::yield_now().await; + let payload = std::hint::black_box(&payload); + usize::from(payload[0]) + usize::from(payload[LARGE_FUTURE_SIZE - 1]) + } + + let result = run_async_command( + || async { + tokio::spawn(large_future()) + .await + .expect("spawned large future should complete") + }, + command_runtime, + ) + .expect("command launcher should use the production runtime"); + + assert_eq!(result, 22); + } + + #[test] + fn command_launcher_does_not_wait_for_blocking_tasks_on_shutdown() { + let (started_tx, started_rx) = sync_channel(0); + let (release_tx, release_rx) = sync_channel(0); + let (result_tx, result_rx) = sync_channel(1); + let launcher = std::thread::spawn(move || { + let result = run_async_command( + move || async move { + let _task = tokio::task::spawn_blocking(move || { + started_tx + .send(()) + .expect("test should observe the blocking task start"); + release_rx + .recv() + .expect("test should release the blocking task"); + }); + started_rx + .recv() + .expect("blocking task should report that it started"); + ExitCode::Success + }, + command_runtime, + ); + result_tx + .send(result) + .expect("test should receive the launcher result"); + }); + + let result = result_rx.recv_timeout(Duration::from_secs(2)); + release_tx + .send(()) + .expect("blocking task should remain available for cleanup"); + launcher.join().expect("launcher test thread should finish"); + + let exit_code = result + .expect("command launcher should not join blocking runtime tasks") + .expect("command launcher should return a successful result"); + assert_eq!(exit_code, ExitCode::Success); + } + + #[test] + fn command_launcher_maps_runtime_creation_failure() { + let future_created = Arc::new(AtomicBool::new(false)); + let future_created_by_worker = Arc::clone(&future_created); + + let error = run_async_command( + move || { + future_created_by_worker.store(true, Ordering::SeqCst); + async {} + }, + || Err(std::io::Error::other("expected runtime failure")), + ) + .expect_err("runtime creation failure should be returned by the launcher"); + + assert!(matches!(error, LauncherError::Runtime(_))); + assert_eq!( + error.to_string(), + "Failed to initialize async runtime: expected runtime failure" + ); + assert!(!future_created.load(Ordering::SeqCst)); + } + + #[test] + fn command_launcher_has_stable_thread_spawn_failure_mapping() { + let error = LauncherError::ThreadSpawn(std::io::Error::other("expected spawn failure")); + + assert_eq!( + error.to_string(), + "Failed to start command thread: expected spawn failure" + ); + } + + #[test] + fn command_launcher_maps_worker_panic_to_stable_error() { + let error = run_on_command_thread::<(), _>(|| panic!("expected worker panic")) + .expect_err("worker panic should be returned by the launcher"); + + assert!(matches!(error, LauncherError::ThreadPanicked)); + assert_eq!( + error.to_string(), + "Command execution thread terminated unexpectedly" + ); + assert!(!error.to_string().contains("expected worker panic")); + } +} diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index f0f9efd5..21540f2d 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -5,6 +5,8 @@ pub mod commands; pub(crate) mod confirm; pub mod exit_code; +mod launcher; +pub use launcher::run; pub mod output; pub(crate) mod private_file; pub(crate) mod secret_input; diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index e74dffe1..72e08ff6 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -1,282 +1,5 @@ -//! rc - Rust S3 CLI Client -//! -//! A command-line interface for S3-compatible object storage services. -//! Designed for RustFS and other S3-compatible backends. - -use clap::Parser; -use tracing_subscriber::{EnvFilter, fmt, prelude::*}; - -mod commands; -mod confirm; -mod exit_code; -mod output; -mod private_file; -mod secret_input; - -use commands::Cli; -use exit_code::ExitCode; - -const COMMAND_THREAD_NAME: &str = "rc-command"; -// AWS SDK command futures can exceed the Windows process main-thread stack. A 16 MiB -// worker stack provides headroom for the largest transfer commands observed in CI. -const COMMAND_THREAD_STACK_SIZE: usize = 16 * 1024 * 1024; - -#[derive(Debug, thiserror::Error)] -enum LauncherError { - #[error("Failed to start command thread: {0}")] - ThreadSpawn(#[source] std::io::Error), - - #[error("Failed to initialize async runtime: {0}")] - Runtime(#[source] std::io::Error), - - #[error("Command execution thread terminated unexpectedly")] - ThreadPanicked, -} - -fn build_env_filter(debug: bool, env_is_set: bool) -> EnvFilter { - if debug && !env_is_set { - EnvFilter::new("debug") - } else { - EnvFilter::from_default_env() - } -} +//! rc executable entry point. fn main() { - let cli = Cli::parse(); - let env_is_set = std::env::var_os(EnvFilter::DEFAULT_ENV).is_some(); - let env_filter = build_env_filter(cli.debug, env_is_set); - - // Initialize tracing subscriber for logging - tracing_subscriber::registry() - .with(fmt::layer()) - .with(env_filter) - .init(); - - let exit_code = match run_command(cli) { - Ok(exit_code) => exit_code, - Err(error) => { - eprintln!("{error}"); - ExitCode::GeneralError - } - }; - - std::process::exit(exit_code.as_i32()); -} - -fn run_command(cli: Cli) -> Result { - run_async_command(move || commands::execute(cli), command_runtime) -} - -fn command_runtime() -> std::io::Result { - tokio::runtime::Builder::new_multi_thread() - .thread_stack_size(COMMAND_THREAD_STACK_SIZE) - .enable_all() - .build() -} - -fn run_async_command( - future_factory: F, - runtime_factory: R, -) -> Result -where - T: Send + 'static, - F: FnOnce() -> Fut + Send + 'static, - Fut: Future + 'static, - R: FnOnce() -> std::io::Result + Send + 'static, -{ - run_on_command_thread(move || { - let runtime = runtime_factory().map_err(LauncherError::Runtime)?; - let output = runtime.block_on(future_factory()); - // The former async main exited the process without waiting for runtime workers. - // Preserve that bounded shutdown behavior if a background task is wedged. - runtime.shutdown_background(); - Ok(output) - }) -} - -fn run_on_command_thread(command: F) -> Result -where - T: Send + 'static, - F: FnOnce() -> Result + Send + 'static, -{ - // The Windows process main thread has a smaller stack than the async command graph - // requires. Constructing and polling the command future here keeps it off that stack. - let command_thread = std::thread::Builder::new() - .name(COMMAND_THREAD_NAME.to_string()) - .stack_size(COMMAND_THREAD_STACK_SIZE) - .spawn(command) - .map_err(LauncherError::ThreadSpawn)?; - - command_thread - .join() - .map_err(|_| LauncherError::ThreadPanicked)? -} - -#[cfg(test)] -mod tests { - use super::*; - use std::sync::Arc; - use std::sync::atomic::{AtomicBool, Ordering}; - use std::sync::mpsc::sync_channel; - use std::time::Duration; - - fn test_runtime() -> std::io::Result { - tokio::runtime::Builder::new_current_thread() - .enable_all() - .build() - } - - #[test] - fn debug_uses_debug_filter_when_env_missing() { - let filter = build_env_filter(true, false); - assert_eq!(filter.to_string(), "debug"); - } - - #[test] - fn command_launcher_preserves_command_exit_code() { - let exit_code = run_async_command(|| async { ExitCode::UnsupportedFeature }, test_runtime) - .expect("command launcher should return its exit code"); - - assert_eq!(exit_code, ExitCode::UnsupportedFeature); - } - - #[test] - fn command_launcher_uses_the_named_worker_thread() { - let thread_name = run_async_command( - || async { std::thread::current().name().map(str::to_string) }, - test_runtime, - ) - .expect("command launcher should return its worker thread name"); - - assert_eq!(thread_name.as_deref(), Some(COMMAND_THREAD_NAME)); - assert_eq!(COMMAND_THREAD_STACK_SIZE, 16 * 1024 * 1024); - } - - #[test] - fn command_launcher_constructs_and_polls_large_future_on_worker() { - const LARGE_FUTURE_SIZE: usize = 2 * 1024 * 1024; - - async fn large_future() -> usize { - let marker = std::hint::black_box(7_u8); - let payload = [marker; LARGE_FUTURE_SIZE]; - tokio::task::yield_now().await; - let payload = std::hint::black_box(&payload); - usize::from(payload[0]) + usize::from(payload[LARGE_FUTURE_SIZE - 1]) - } - - let result = run_async_command(large_future, test_runtime) - .expect("large command future should complete on the sized worker stack"); - - assert_eq!(result, 14); - } - - #[test] - fn command_runtime_polls_spawned_large_future_on_sized_worker() { - const LARGE_FUTURE_SIZE: usize = 1024 * 1024; - - async fn large_future() -> usize { - let marker = std::hint::black_box(11_u8); - let payload = [marker; LARGE_FUTURE_SIZE]; - tokio::task::yield_now().await; - let payload = std::hint::black_box(&payload); - usize::from(payload[0]) + usize::from(payload[LARGE_FUTURE_SIZE - 1]) - } - - let result = run_async_command( - || async { - tokio::spawn(large_future()) - .await - .expect("spawned large future should complete") - }, - command_runtime, - ) - .expect("command launcher should use the production runtime"); - - assert_eq!(result, 22); - } - - #[test] - fn command_launcher_does_not_wait_for_blocking_tasks_on_shutdown() { - let (started_tx, started_rx) = sync_channel(0); - let (release_tx, release_rx) = sync_channel(0); - let (result_tx, result_rx) = sync_channel(1); - let launcher = std::thread::spawn(move || { - let result = run_async_command( - move || async move { - let _task = tokio::task::spawn_blocking(move || { - started_tx - .send(()) - .expect("test should observe the blocking task start"); - release_rx - .recv() - .expect("test should release the blocking task"); - }); - started_rx - .recv() - .expect("blocking task should report that it started"); - ExitCode::Success - }, - command_runtime, - ); - result_tx - .send(result) - .expect("test should receive the launcher result"); - }); - - let result = result_rx.recv_timeout(Duration::from_secs(2)); - release_tx - .send(()) - .expect("blocking task should remain available for cleanup"); - launcher.join().expect("launcher test thread should finish"); - - let exit_code = result - .expect("command launcher should not join blocking runtime tasks") - .expect("command launcher should return a successful result"); - assert_eq!(exit_code, ExitCode::Success); - } - - #[test] - fn command_launcher_maps_runtime_creation_failure() { - let future_created = Arc::new(AtomicBool::new(false)); - let future_created_by_worker = Arc::clone(&future_created); - - let error = run_async_command( - move || { - future_created_by_worker.store(true, Ordering::SeqCst); - async {} - }, - || Err(std::io::Error::other("expected runtime failure")), - ) - .expect_err("runtime creation failure should be returned by the launcher"); - - assert!(matches!(error, LauncherError::Runtime(_))); - assert_eq!( - error.to_string(), - "Failed to initialize async runtime: expected runtime failure" - ); - assert!(!future_created.load(Ordering::SeqCst)); - } - - #[test] - fn command_launcher_has_stable_thread_spawn_failure_mapping() { - let error = LauncherError::ThreadSpawn(std::io::Error::other("expected spawn failure")); - - assert_eq!( - error.to_string(), - "Failed to start command thread: expected spawn failure" - ); - } - - #[test] - fn command_launcher_maps_worker_panic_to_stable_error() { - let error = run_on_command_thread::<(), _>(|| panic!("expected worker panic")) - .expect_err("worker panic should be returned by the launcher"); - - assert!(matches!(error, LauncherError::ThreadPanicked)); - assert_eq!( - error.to_string(), - "Command execution thread terminated unexpectedly" - ); - assert!(!error.to_string().contains("expected worker panic")); - } + rustfs_cli::run(); } diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 9e51b2d2..3582481b 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -8,6 +8,7 @@ use std::path::PathBuf; use std::process::{Command, Output}; const GLOBAL_OPTIONS: &[&str] = &[ + "--data-transport", "--format", "--json", "--no-color", diff --git a/crates/core/src/admin/kms_diagnostic.rs b/crates/core/src/admin/kms_diagnostic.rs index e8181378..e8f32986 100644 --- a/crates/core/src/admin/kms_diagnostic.rs +++ b/crates/core/src/admin/kms_diagnostic.rs @@ -228,6 +228,15 @@ fn classify_and_discard_error(mut error: Error) -> KmsRoundTripErrorClass { message.zeroize(); KmsRoundTripErrorClass::Network } + Error::DataTransfer { + message, source, .. + } => { + message.zeroize(); + if let Some(source) = source.take() { + classify_and_discard_error(*source); + } + KmsRoundTripErrorClass::General + } Error::Conflict(message) | Error::AliasExists(message) => { message.zeroize(); KmsRoundTripErrorClass::Conflict diff --git a/crates/core/src/data_transport.rs b/crates/core/src/data_transport.rs new file mode 100644 index 00000000..0e56226f --- /dev/null +++ b/crates/core/src/data_transport.rs @@ -0,0 +1,118 @@ +//! Selection and failure classification for an optional object data transport. + +use std::str::FromStr; +use std::sync::atomic::{AtomicU8, Ordering}; + +/// Object payload transport selected independently of S3 management operations. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +#[repr(u8)] +pub enum DataTransportMode { + /// Use the existing S3 HTTP body implementation. + #[default] + Http, + /// Try an installed external transport, with safe HTTP fallback. + Auto, + /// Require an installed external transport for object payloads. + Rdma, +} + +impl FromStr for DataTransportMode { + type Err = String; + + fn from_str(value: &str) -> std::result::Result { + match value { + "http" => Ok(Self::Http), + "auto" => Ok(Self::Auto), + "rdma" => Ok(Self::Rdma), + _ => Err("data transport must be http, auto, or rdma".to_string()), + } + } +} + +/// Business outcome classification. Transport timeouts never imply rejection. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum DataTransferState { + /// The provider proves that the operation was not accepted and IO was untouched. + NotAccepted, + /// The operation failed with an authoritative business outcome. + ConfirmedFailure, + /// Acceptance or the final business outcome could not be established. + OutcomeUnknown, +} + +impl std::fmt::Display for DataTransferState { + fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + formatter.write_str(match self { + Self::NotAccepted => "not accepted", + Self::ConfirmedFailure => "confirmed failure", + Self::OutcomeUnknown => "outcome unknown", + }) + } +} + +static GLOBAL_MODE: AtomicU8 = AtomicU8::new(DataTransportMode::Http as u8); + +/// Set process-wide CLI policy before constructing any clients. +pub fn set_global_data_transport_mode(mode: DataTransportMode) { + GLOBAL_MODE.store(mode as u8, Ordering::Relaxed); +} + +/// Snapshot process-wide policy when constructing a client. +pub fn global_data_transport_mode() -> DataTransportMode { + match GLOBAL_MODE.load(Ordering::Relaxed) { + 1 => DataTransportMode::Auto, + 2 => DataTransportMode::Rdma, + _ => DataTransportMode::Http, + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn policy_is_explicit_and_defaults_to_http() { + assert_eq!(DataTransportMode::default(), DataTransportMode::Http); + assert_eq!("auto".parse(), Ok(DataTransportMode::Auto)); + assert_eq!("rdma".parse(), Ok(DataTransportMode::Rdma)); + assert!("udp".parse::().is_err()); + } + + #[test] + fn external_failures_never_enter_string_based_retry() { + for state in [ + DataTransferState::NotAccepted, + DataTransferState::ConfirmedFailure, + DataTransferState::OutcomeUnknown, + ] { + let error = crate::Error::DataTransfer { + state, + message: "temporary timeout".to_string(), + source: None, + }; + assert!(!crate::is_retryable_error(&error)); + } + } + + #[test] + fn authoritative_error_retains_exit_code_and_cleanup_without_retry() { + let error = crate::Error::DataTransfer { + state: DataTransferState::ConfirmedFailure, + message: "S3 request failed".to_string(), + source: Some(Box::new(crate::Error::Auth("access denied".to_string()))), + } + .with_multipart_copy_context("upload-id", crate::MultipartAbortStatus::Succeeded); + assert_eq!(error.exit_code(), 4); + assert!(!crate::is_retryable_error(&error)); + assert!(std::error::Error::source(&error).is_some()); + let error = crate::Error::DataTransfer { + state: DataTransferState::OutcomeUnknown, + message: "temporary timeout".to_string(), + source: Some(Box::new(crate::Error::Io(std::io::Error::new( + std::io::ErrorKind::TimedOut, + "temporary timeout", + )))), + }; + assert!(!crate::is_retryable_error(&error)); + } +} diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index 6e2e0001..d0b9e4c3 100644 --- a/crates/core/src/error.rs +++ b/crates/core/src/error.rs @@ -103,6 +103,17 @@ pub enum Error { #[error("Network error: {0}")] Network(String), + /// An external data operation with explicit acceptance and outcome state. + #[error("Object data transfer {state}: {message}")] + DataTransfer { + /// Acceptance and business outcome classification. + state: crate::DataTransferState, + /// A safe diagnostic that excludes credentials and protocol envelopes. + message: String, + /// Authoritative S3 error, when one was received. + source: Option>, + }, + /// Conflict error #[error("Conflict: {0}")] Conflict(String), @@ -139,8 +150,12 @@ impl Error { Error::Conflict(_) | Error::GovernanceDenied { .. } | Error::AliasExists(_) => 6, // Conflict Error::UnsupportedFeature(_) => 7, // UnsupportedFeature Error::Interrupted(_) => 130, // Interrupted - Error::RequestRejected(_) => 1, // GeneralError - _ => 1, // GeneralError + Error::DataTransfer { + source: Some(source), + .. + } => source.exit_code(), + Error::RequestRejected(_) => 1, // GeneralError + _ => 1, // GeneralError } } @@ -161,6 +176,15 @@ impl Error { match self { Self::Auth(message) => Self::Auth(format!("{message}; {context}")), Self::Network(message) => Self::Network(format!("{message}; {context}")), + Self::DataTransfer { + state, + message, + source, + } => Self::DataTransfer { + state, + message: format!("{message}; {context}"), + source, + }, Self::NotFound(message) => Self::NotFound(format!("{message}; {context}")), Self::VersionNotFound { path, version_id } => Self::VersionNotFound { path: format!("{path} ({context})"), diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index b215a418..3653e29c 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -13,6 +13,7 @@ pub mod admin; pub mod alias; pub mod config; pub mod cors; +pub mod data_transport; pub mod encryption; pub mod error; pub mod lifecycle; @@ -36,6 +37,10 @@ pub use alias::{ }; pub use config::{Config, ConfigManager}; pub use cors::{CorsConfiguration, CorsRule}; +pub use data_transport::{ + DataTransferState, DataTransportMode, global_data_transport_mode, + set_global_data_transport_mode, +}; pub use encryption::{BucketEncryption, ObjectEncryptionRequest}; pub use error::{Error, MultipartAbortStatus, Result}; pub use lifecycle::{ diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 7e7cccc5..50dc1d99 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -6,7 +6,8 @@ use async_trait::async_trait; use aws_credential_types::Credentials; use aws_sdk_s3::error::ProvideErrorMetadata; use aws_sigv4::http_request::{ - SignableBody, SignableRequest, SignatureLocation, SigningSettings, sign, + PayloadChecksumKind, PercentEncodingMode, SignableBody, SignableRequest, SignatureLocation, + SigningSettings, UriPathNormalizationMode, sign, }; use aws_sigv4::sign::v4; use aws_smithy_runtime_api::box_error::BoxError; @@ -57,7 +58,14 @@ use serde::Deserialize; use sha2::{Digest, Sha256}; use std::collections::HashMap; use std::str::FromStr; +use std::sync::Arc; use std::time::Duration; + +use crate::data_transport::{ + self, ControlExecutor, ControlRequest, ControlResponse, DataTransportContext, DownloadRequest, + ObjectDataTransport, ObservedReader, ObservedWriter, UploadOperation, UploadRequest, + UploadResult, WriteCondition, +}; use tokio::io::AsyncReadExt; use tokio::io::AsyncWrite; use tokio::io::AsyncWriteExt; @@ -1642,13 +1650,22 @@ impl HttpClient for ReqwestConnector { } } +struct RequestSigningOptions<'a> { + region: &'a str, + settings: SigningSettings, + time: std::time::SystemTime, +} + /// S3 client wrapper +#[derive(Clone)] pub struct S3Client { inner: aws_sdk_s3::Client, presign_inner: aws_sdk_s3::Client, xml_http_client: reqwest::Client, alias: Alias, request_headers: Vec, + data_transport_mode: rc_core::DataTransportMode, + data_transport: Option>, } #[derive(Debug, Clone)] @@ -1764,13 +1781,153 @@ impl S3Client { let client = aws_sdk_s3::Client::from_conf(s3_config); let presign_client = aws_sdk_s3::Client::from_conf(presign_s3_config); - Ok(Self { + let mut client = Self { inner: client, presign_inner: presign_client, xml_http_client, alias, request_headers, - }) + data_transport_mode: rc_core::global_data_transport_mode(), + data_transport: None, + }; + if client.data_transport_mode != rc_core::DataTransportMode::Http + && let Some(factory) = data_transport::installed_factory() + { + client.data_transport = Some(factory.create(DataTransportContext { + endpoint: client.alias.endpoint.clone(), + control: Arc::new(client.clone()), + })?); + } + Ok(client) + } + + /// Inject a transport into this client without changing ObjectStore signatures. + pub fn with_object_data_transport( + mut self, + mode: rc_core::DataTransportMode, + transport: Arc, + ) -> Self { + self.data_transport_mode = mode; + self.data_transport = Some(transport); + self + } + + fn selected_data_transport(&self) -> Result>> { + use rc_core::DataTransportMode; + match (self.data_transport_mode, self.data_transport.as_ref()) { + (DataTransportMode::Http, _) | (DataTransportMode::Auto, None) => Ok(None), + (_, Some(transport)) => Ok(Some(transport)), + (DataTransportMode::Rdma, None) => Err(Error::DataTransfer { + state: rc_core::DataTransferState::NotAccepted, + message: "This executable has no external object data provider".to_string(), + source: None, + }), + } + } + + async fn try_external_download( + &self, + path: &RemotePath, + options: &TransferReadOptions, + writer: &mut W, + max_bytes: Option, + on_progress: &mut (dyn FnMut(u64, Option) + Send), + ) -> Result> { + let Some(transport) = self.selected_data_transport()? else { + return Ok(None); + }; + let limit = max_bytes.unwrap_or(rc_core::S3_MAX_OBJECT_SIZE); + let mut observed = ObservedWriter { + inner: writer, + polled: false, + bytes: 0, + limit, + flushed: false, + }; + let result = transport + .download( + DownloadRequest { + path, + options, + max_bytes, + max_receive_bytes: limit, + on_progress, + }, + &mut observed, + ) + .await; + match result { + Ok(result) + if result.bytes_transferred == observed.bytes + && result.content_length == observed.bytes + && observed.flushed => + { + Ok(Some(observed.bytes)) + } + Ok(_) => Err(Error::DataTransfer { + state: rc_core::DataTransferState::ConfirmedFailure, + message: "External download did not complete its declared length and writer flush" + .to_string(), + source: None, + }), + Err(error) + if error.state == rc_core::DataTransferState::NotAccepted + && !observed.polled + && self.data_transport_mode == rc_core::DataTransportMode::Auto => + { + Ok(None) + } + Err(mut error) => { + if error.state == rc_core::DataTransferState::NotAccepted && observed.polled { + error.state = rc_core::DataTransferState::OutcomeUnknown; + } + Err(self.redact_sse_customer_error(error.into(), options.customer_key.as_ref())) + } + } + } + + async fn try_external_upload( + &self, + request: UploadRequest<'_>, + data: &[u8], + ) -> Result> { + let Some(transport) = self.selected_data_transport()? else { + return Ok(None); + }; + let length = request.content_length; + let customer_key = + destination_sse_customer_key(request.options.encryption.as_ref()).cloned(); + let mut source = data; + let mut observed = ObservedReader { + inner: &mut source, + polled: false, + bytes: 0, + eof: false, + }; + let result = transport.upload(request, &mut observed).await; + match result { + Ok(result) if observed.bytes == length && observed.eof => Ok(Some(result)), + Ok(_) => Err(Error::DataTransfer { + state: rc_core::DataTransferState::OutcomeUnknown, + message: + "External upload returned success without consuming the exact source and EOF" + .to_string(), + source: None, + }), + Err(error) + if error.state == rc_core::DataTransferState::NotAccepted + && !observed.polled + && self.data_transport_mode == rc_core::DataTransportMode::Auto => + { + Ok(None) + } + Err(mut error) => { + if error.state == rc_core::DataTransferState::NotAccepted && observed.polled { + error.state = rc_core::DataTransferState::OutcomeUnknown; + } + Err(self.redact_sse_customer_error(error.into(), customer_key.as_ref())) + } + } } fn ensure_sse_customer_transport(&self, key: Option<&SseCustomerKey>) -> Result<()> { @@ -1952,6 +2109,24 @@ impl S3Client { options: &ObjectReadOptions, mut on_progress: impl FnMut(u64, Option) + Send, ) -> Result> { + let mut external_data = Vec::new(); + let external_options = TransferReadOptions { + version_id: options.version_id.clone(), + ..TransferReadOptions::default() + }; + if self + .try_external_download( + path, + &external_options, + &mut external_data, + None, + &mut on_progress, + ) + .await? + .is_some() + { + return Ok(external_data); + } let mut request = self.inner.get_object().bucket(&path.bucket).key(&path.key); if let Some(version_id) = &options.version_id { request = request.version_id(version_id); @@ -2017,6 +2192,23 @@ impl S3Client { ) -> Result { options.validate()?; self.ensure_sse_customer_transport(options.customer_key.as_ref())?; + if self.selected_data_transport()?.is_some() { + let (mut file, temporary) = Self::temporary_download_file(destination)?; + if let Some(bytes) = self + .try_external_download(path, options, &mut file, None, &mut on_progress) + .await? + { + drop(file); + temporary.persist(destination).map_err(|error| { + Error::General(format!( + "atomically replace download destination '{}': {}", + destination.display(), + error.error + )) + })?; + return Ok(bytes); + } + } let mut request = apply_sse_customer_to_get_request( self.inner.get_object().bucket(&path.bucket).key(&path.key), options.customer_key.as_ref(), @@ -2036,29 +2228,7 @@ impl S3Client { let content_length = response .content_length() .and_then(|length| u64::try_from(length).ok()); - let parent = destination.parent().ok_or_else(|| { - Error::General(format!( - "download destination '{}' has no parent directory", - destination.display() - )) - })?; - let file_name = destination.file_name().ok_or_else(|| { - Error::General(format!( - "download destination '{}' has no file name", - destination.display() - )) - })?; - let temporary_file = tempfile::Builder::new() - .prefix(&format!(".{}.rc-part-", file_name.to_string_lossy())) - .tempfile_in(parent) - .map_err(|error| { - Error::General(format!( - "create temporary download in '{}': {error}", - parent.display() - )) - })?; - let (file, temporary) = temporary_file.into_parts(); - let mut file = tokio::fs::File::from_std(file); + let (mut file, temporary) = Self::temporary_download_file(destination)?; let mut body = response.body; let mut bytes_downloaded = 0u64; @@ -2100,6 +2270,35 @@ impl S3Client { Ok(bytes_downloaded) } + fn temporary_download_file( + destination: &std::path::Path, + ) -> Result<(tokio::fs::File, tempfile::TempPath)> { + let parent = destination.parent().ok_or_else(|| { + Error::General(format!( + "download destination '{}' has no parent directory", + destination.display() + )) + })?; + let file_name = destination.file_name().ok_or_else(|| { + Error::General(format!( + "download destination '{}' has no file name", + destination.display() + )) + })?; + let temporary_file = tempfile::Builder::new() + .prefix(&format!(".{}.rc-part-", file_name.to_string_lossy())) + .tempfile_in(parent) + .map_err(|error| { + Error::General(format!( + "create temporary download in '{}': {error}", + parent.display() + )) + })?; + let (file, temporary) = temporary_file.into_parts(); + let file = tokio::fs::File::from_std(file); + Ok((file, temporary)) + } + pub async fn write_object_to( &self, path: &RemotePath, @@ -2131,6 +2330,17 @@ impl S3Client { return Ok(0); } + let external_options = TransferReadOptions { + version_id: options.version_id.clone(), + ..TransferReadOptions::default() + }; + if let Some(bytes) = self + .try_external_download(path, &external_options, writer, max_bytes, &mut |_, _| {}) + .await? + { + return Ok(bytes); + } + let mut request = self.inner.get_object().bucket(&path.bucket).key(&path.key); if let Some(version_id) = &options.version_id { request = request.version_id(version_id); @@ -2180,6 +2390,30 @@ impl S3Client { data: Vec, content_type: Option<&str>, ) -> Result { + let options = ObjectWriteOptions { + attributes: Some(ObjectAttributes { + content_type: content_type.map(ToString::to_string), + ..ObjectAttributes::default() + }), + ..ObjectWriteOptions::default() + }; + if let Some(result) = self + .try_external_upload( + UploadRequest { + path, + operation: UploadOperation::PutObject { + condition: WriteCondition::IfAbsent, + }, + options: &options, + content_length: data.len() as u64, + checksum_sha256: None, + }, + &data, + ) + .await? + { + return Ok(result.object); + } let size = data.len() as i64; let mut request = self .inner @@ -2713,6 +2947,15 @@ impl S3Client { version_id: version_id.map(|value| self.redact_sensitive_text(value)), }, Error::Network(message) => Error::Network(self.redact_sensitive_text(message)), + Error::DataTransfer { + state, + message, + source, + } => Error::DataTransfer { + state, + message: self.redact_sensitive_text(message), + source: source.map(|error| Box::new(self.redact_sensitive_error(*error))), + }, Error::Conflict(message) => Error::Conflict(self.redact_sensitive_text(message)), Error::UnsupportedFeature(message) => { Error::UnsupportedFeature(self.redact_sensitive_text(message)) @@ -2764,6 +3007,17 @@ impl S3Client { version_id: version_id.map(redact), }, Error::Network(message) => Error::Network(redact(message)), + Error::DataTransfer { + state, + message, + source, + } => Error::DataTransfer { + state, + message: redact(message), + source: source.map(|error| { + Box::new(self.redact_sse_customer_error(*error, Some(customer_key))) + }), + }, Error::Conflict(message) => Error::Conflict(redact(message)), Error::UnsupportedFeature(message) => Error::UnsupportedFeature(redact(message)), Error::General(message) => Error::General(redact(message)), @@ -3286,6 +3540,114 @@ impl S3Client { headers: &HeaderMap, body: &[u8], region: &str, + ) -> Result { + self.sign_request_with_settings( + method, + url, + headers, + body, + RequestSigningOptions { + region, + settings: SigningSettings::default(), + time: std::time::SystemTime::now(), + }, + ) + .await + } + + fn external_control_url( + &self, + path: &RemotePath, + query: &[(String, String)], + ) -> Result { + let mut url = reqwest::Url::parse(&self.alias.endpoint)?; + if url.scheme() != "https" || self.alias.insecure { + return Err(Error::RequestRejected( + "External object control requires verified HTTPS".to_string(), + )); + } + if path + .key + .split('/') + .any(|segment| matches!(segment, "." | "..")) + { + return Err(Error::UnsupportedFeature( + "External control does not support dot path segments".to_string(), + )); + } + let mut object_path = url + .path() + .strip_suffix('/') + .unwrap_or(url.path()) + .to_string(); + if force_path_style_for_alias(&self.alias) { + object_path.push('/'); + object_path.push_str(&urlencoding::encode(&path.bucket)); + } else { + let host = url + .host_str() + .ok_or_else(|| Error::InvalidPath("Object endpoint has no host".to_string()))?; + url.set_host(Some(&format!("{}.{}", path.bucket, host))) + .map_err(|_| Error::InvalidPath("Invalid DNS-style bucket name".to_string()))?; + } + object_path.push('/'); + object_path.push_str( + &path + .key + .split('/') + .map(|segment| urlencoding::encode(segment).into_owned()) + .collect::>() + .join("/"), + ); + url.set_path(&object_path); + let encoded_query = query + .iter() + .map(|(name, value)| { + format!( + "{}={}", + urlencoding::encode(name), + urlencoding::encode(value) + ) + }) + .collect::>() + .join("&"); + url.set_query((!encoded_query.is_empty()).then_some(encoded_query.as_str())); + Ok(url) + } + + async fn sign_external_control_request_at( + &self, + method: &Method, + url: &str, + headers: &HeaderMap, + body: &[u8], + time: std::time::SystemTime, + ) -> Result { + let mut settings = SigningSettings::default(); + settings.percent_encoding_mode = PercentEncodingMode::Single; + settings.uri_path_normalization_mode = UriPathNormalizationMode::Disabled; + settings.payload_checksum_kind = PayloadChecksumKind::XAmzSha256; + self.sign_request_with_settings( + method, + url, + headers, + body, + RequestSigningOptions { + region: &self.alias.region, + settings, + time, + }, + ) + .await + } + + async fn sign_request_with_settings( + &self, + method: &Method, + url: &str, + headers: &HeaderMap, + body: &[u8], + options: RequestSigningOptions<'_>, ) -> Result { if self.alias.anonymous { return Ok(headers.clone()); @@ -3300,14 +3662,14 @@ impl S3Client { ); let identity = credentials.into(); - let mut signing_settings = SigningSettings::default(); + let mut signing_settings = options.settings; signing_settings.signature_location = SignatureLocation::Headers; let signing_params = v4::SigningParams::builder() .identity(&identity) - .region(region) + .region(options.region) .name(S3_SERVICE_NAME) - .time(std::time::SystemTime::now()) + .time(options.time) .settings(signing_settings) .build() .map_err(|e| Error::Auth(format!("Failed to build signing params: {e}")))?; @@ -3784,6 +4146,36 @@ impl S3Client { .await .map_err(|e| Error::General(format!("read file '{}': {e}", file_path.display())))?; let requested_checksum = requested_sha256_checksum(&data, options.write.checksum.as_ref())?; + let condition = match options.precondition { + ObjectWritePrecondition::None => WriteCondition::None, + ObjectWritePrecondition::IfAbsent => WriteCondition::IfAbsent, + ObjectWritePrecondition::IfMatch(etag) => WriteCondition::IfMatch(etag), + }; + if let Some(result) = self + .try_external_upload( + UploadRequest { + path, + operation: UploadOperation::PutObject { condition }, + options: options.write, + content_length: file_size, + checksum_sha256: requested_checksum.as_deref(), + }, + &data, + ) + .await? + { + if let Some(checksum) = &requested_checksum { + self.verify_persisted_sha256( + path, + result.object.version_id.clone(), + checksum, + destination_sse_customer_key(options.write.encryption.as_ref()), + ) + .await + .map_err(data_transport::committed_verification_failure)?; + } + return Ok(result.object); + } let body = aws_sdk_s3::primitives::ByteStream::from(data); let storage_class = rustfs_storage_class(options.write.storage_class.as_deref())?; @@ -3954,6 +4346,7 @@ impl S3Client { tracing::debug!(upload_id = %upload_id, "Multipart upload initiated"); let mut completed_parts = Vec::new(); + let mut has_external_parts = false; let mut part_number: i32 = 1; let mut bytes_uploaded: u64 = 0; let mut part_digests = Vec::new(); @@ -3981,43 +4374,72 @@ impl S3Client { part_digests.push(digest); BASE64_STANDARD.encode(digest) }); - let body = aws_sdk_s3::primitives::ByteStream::from(chunk[..bytes_read].to_vec()); - let mut upload_part_request = apply_sse_customer_to_upload_part_request( - self.inner - .upload_part() - .bucket(&path.bucket) - .key(&path.key) - .upload_id(&upload_id) - .part_number(part_number) - .body(body), - destination_sse_customer_key(options.write.encryption.as_ref()), - ); - if let Some(checksum) = &part_checksum { - upload_part_request = upload_part_request - .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256) - .checksum_sha256(checksum); - } - let upload_part_result = upload_part_request.send().await; - - let upload_part_response = match upload_part_result { - Ok(response) => response, - Err(e) => { - tracing::debug!( - upload_id = %upload_id, - part_number, - "Aborting multipart upload due to error" - ); - let primary = self.redact_sse_customer_error( - Self::map_object_request_error(&e, path, None), + let upload_part_result: Result<(Option, Option)> = match self + .try_external_upload( + UploadRequest { + path, + operation: UploadOperation::UploadPart { + upload_id: &upload_id, + part_number, + }, + options: options.write, + content_length: bytes_read as u64, + checksum_sha256: part_checksum.as_deref(), + }, + &chunk[..bytes_read], + ) + .await + { + Ok(Some(result)) => { + has_external_parts = true; + Ok((result.object.etag, result.checksum_sha256)) + } + Err(error) => Err(error), + Ok(None) => { + let body = + aws_sdk_s3::primitives::ByteStream::from(chunk[..bytes_read].to_vec()); + let mut upload_part_request = apply_sse_customer_to_upload_part_request( + self.inner + .upload_part() + .bucket(&path.bucket) + .key(&path.key) + .upload_id(&upload_id) + .part_number(part_number) + .body(body), destination_sse_customer_key(options.write.encryption.as_ref()), ); + if let Some(checksum) = &part_checksum { + upload_part_request = upload_part_request + .checksum_algorithm(aws_sdk_s3::types::ChecksumAlgorithm::Sha256) + .checksum_sha256(checksum); + } + upload_part_request + .send() + .await + .map(|response| { + ( + response.e_tag().map(ToString::to_string), + response.checksum_sha256().map(ToString::to_string), + ) + }) + .map_err(|error| { + self.redact_sse_customer_error( + Self::map_object_request_error(&error, path, None), + destination_sse_customer_key(options.write.encryption.as_ref()), + ) + }) + } + }; + let (part_etag, persisted_part_checksum) = match upload_part_result { + Ok(response) => response, + Err(primary) => { return Err(self .abort_multipart_upload_with_error(path, &upload_id, primary) .await); } }; if let Some(expected) = &part_checksum { - match upload_part_response.checksum_sha256() { + match persisted_part_checksum.as_deref() { Some(actual) if actual == expected => {} Some(_) => { let primary = Error::Conflict(format!( @@ -4038,7 +4460,7 @@ impl S3Client { } } - let etag = match upload_part_response.e_tag() { + let etag = match part_etag.as_deref() { Some(value) => value.trim_matches('"').to_string(), None => { let primary = @@ -4098,6 +4520,24 @@ impl S3Client { destination_sse_customer_key(options.write.encryption.as_ref()), ) }; + let primary = if has_external_parts { + let state = match &error { + aws_sdk_s3::error::SdkError::ServiceError(service_error) + if matches!(service_error.raw().status().as_u16(), 409 | 412) => + { + rc_core::DataTransferState::ConfirmedFailure + } + _ => rc_core::DataTransferState::OutcomeUnknown, + }; + Error::DataTransfer { + state, + message: "Multipart completion did not return a usable success result" + .to_string(), + source: Some(Box::new(primary)), + } + } else { + primary + }; return Err(self .abort_multipart_upload_with_error(path, &upload_id, primary) .await); @@ -4121,7 +4561,14 @@ impl S3Client { &expected_checksum, destination_sse_customer_key(options.write.encryption.as_ref()), ) - .await?; + .await + .map_err(|error| { + if has_external_parts { + data_transport::committed_verification_failure(error) + } else { + error + } + })?; } Ok(info) } @@ -5251,6 +5698,14 @@ impl ObjectStore for S3Client { ) -> Result> { options.validate()?; self.ensure_sse_customer_transport(options.customer_key.as_ref())?; + let mut external_data = Vec::new(); + if self + .try_external_download(path, options, &mut external_data, None, &mut |_, _| {}) + .await? + .is_some() + { + return Ok(external_data); + } let mut request = apply_sse_customer_to_get_request( self.inner.get_object().bucket(&path.bucket).key(&path.key), options.customer_key.as_ref(), @@ -5314,6 +5769,12 @@ impl ObjectStore for S3Client { .await?; return Ok(0); } + if let Some(bytes) = self + .try_external_download(path, options, writer, max_bytes, &mut |_, _| {}) + .await? + { + return Ok(bytes); + } let mut request = apply_sse_customer_to_get_request( self.inner.get_object().bucket(&path.bucket).key(&path.key), options.customer_key.as_ref(), @@ -5362,6 +5823,31 @@ impl ObjectStore for S3Client { content_type: Option<&str>, encryption: Option<&ObjectEncryptionRequest>, ) -> Result { + let external_options = ObjectWriteOptions { + attributes: Some(ObjectAttributes { + content_type: content_type.map(ToString::to_string), + ..ObjectAttributes::default() + }), + encryption: encryption.cloned().map(ObjectWriteEncryption::Managed), + ..ObjectWriteOptions::default() + }; + if let Some(result) = self + .try_external_upload( + UploadRequest { + path, + operation: UploadOperation::PutObject { + condition: WriteCondition::None, + }, + options: &external_options, + content_length: data.len() as u64, + checksum_sha256: None, + }, + &data, + ) + .await? + { + return Ok(result.object); + } let size = data.len() as i64; let body = aws_sdk_s3::primitives::ByteStream::from(data); @@ -5403,6 +5889,33 @@ impl ObjectStore for S3Client { self.ensure_write_sse_customer_transport(options.encryption.as_ref())?; let size = data.len() as i64; let requested_checksum = requested_sha256_checksum(&data, options.checksum.as_ref())?; + if let Some(result) = self + .try_external_upload( + UploadRequest { + path, + operation: UploadOperation::PutObject { + condition: WriteCondition::None, + }, + options, + content_length: data.len() as u64, + checksum_sha256: requested_checksum.as_deref(), + }, + &data, + ) + .await? + { + if let Some(checksum) = &requested_checksum { + self.verify_persisted_sha256( + path, + result.object.version_id.clone(), + checksum, + destination_sse_customer_key(options.encryption.as_ref()), + ) + .await + .map_err(data_transport::committed_verification_failure)?; + } + return Ok(result.object); + } let body = aws_sdk_s3::primitives::ByteStream::from(data); let storage_class = rustfs_storage_class(options.storage_class.as_deref())?; let request = apply_object_write_encryption_to_put_request( @@ -6813,14 +7326,126 @@ impl ObjectStore for S3Client { } } -#[cfg(test)] -mod tests { - use super::*; - use aws_smithy_http_client::test_util::{ - CaptureRequestReceiver, ReplayEvent, StaticReplayClient, capture_request, - }; - use rc_core::{ - ChecksumAlgorithm, ChecksumRequest, MetadataDirective, ObjectAttributes, +/// External control exchanges retain the host client's TLS and signing policy. +#[async_trait] +impl ControlExecutor for S3Client { + async fn execute(&self, request: ControlRequest) -> Result { + if self.alias.anonymous { + return Err(Error::RequestRejected( + "External object control requires an authenticated S3 principal".to_string(), + )); + } + const MAX_CONTROL_BYTES: usize = 1024 * 1024; + if request.body.len() > MAX_CONTROL_BYTES + || request.response_limit == 0 + || request.response_limit > MAX_CONTROL_BYTES + { + return Err(Error::RequestRejected( + "Control envelope exceeds the bounded exchange policy".to_string(), + )); + } + let url = self.external_control_url(&request.path, &request.query)?; + let mut headers = request.headers; + for name in [ + "authorization", + "host", + "content-length", + "x-amz-date", + "x-amz-content-sha256", + ] { + if headers.contains_key(name) { + return Err(Error::RequestRejected( + "Control requests cannot override signing headers".to_string(), + )); + } + } + for header in &self.request_headers { + if headers.contains_key(&header.name) { + return Err(Error::RequestRejected( + "A custom header conflicts with the external control request".to_string(), + )); + } + headers.insert( + HeaderName::from_bytes(header.name.as_bytes()).map_err(|_| { + Error::RequestRejected("Invalid custom header name".to_string()) + })?, + HeaderValue::from_str(&header.value).map_err(|_| { + Error::RequestRejected("Invalid custom header value".to_string()) + })?, + ); + } + headers.insert( + "host", + HeaderValue::from_str(&self.request_host(&url)?) + .map_err(|_| Error::RequestRejected("Invalid object host".to_string()))?, + ); + headers.insert( + "x-amz-content-sha256", + HeaderValue::from_str(&Self::sha256_hash(&request.body)) + .map_err(|_| Error::RequestRejected("Invalid payload digest".to_string()))?, + ); + if headers.values().any(|value| value.to_str().is_err()) { + return Err(Error::RequestRejected( + "Control request headers must have UTF-8 values".to_string(), + )); + } + let signed = self + .sign_external_control_request_at( + &request.method, + url.as_str(), + &headers, + &request.body, + std::time::SystemTime::now(), + ) + .await?; + let response = self + .xml_http_client + .request(request.method, url) + .headers(signed) + .body(request.body) + .send() + .await + .map_err(|_| Error::Network("External control exchange failed".to_string()))?; + let status = response.status(); + let headers = response.headers().clone(); + if response + .content_length() + .is_some_and(|length| length > request.response_limit as u64) + { + return Err(Error::Network( + "External control response exceeds its bound".to_string(), + )); + } + let mut body = Vec::new(); + let mut stream = response.bytes_stream(); + while let Some(chunk) = stream + .try_next() + .await + .map_err(|_| Error::Network("External control response was interrupted".to_string()))? + { + if chunk.len() > request.response_limit.saturating_sub(body.len()) { + return Err(Error::Network( + "External control response exceeds its bound".to_string(), + )); + } + body.extend_from_slice(&chunk); + } + Ok(ControlResponse { + status, + headers, + body: Bytes::from(body), + }) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use aws_smithy_http_client::test_util::{ + CaptureRequestReceiver, ReplayEvent, StaticReplayClient, capture_request, + }; + use rc_core::{ + ChecksumAlgorithm, ChecksumRequest, MetadataDirective, ObjectAttributes, ObjectWriteEncryption, ObjectWriteOptions, S3_MULTIPART_COPY_MIN_PART_SIZE, SseCustomerKey, TransferCopyOptions, TransferReadOptions, }; @@ -6899,6 +7524,8 @@ mod tests { xml_http_client, alias, request_headers, + data_transport_mode: rc_core::DataTransportMode::Http, + data_transport: None, }; (client, request_receiver) @@ -6955,6 +7582,8 @@ mod tests { xml_http_client: reqwest::Client::new(), alias, request_headers, + data_transport_mode: rc_core::DataTransportMode::Http, + data_transport: None, }; (client, replay) } @@ -7084,6 +7713,8 @@ mod tests { xml_http_client: reqwest::Client::new(), alias, request_headers: Vec::new(), + data_transport_mode: rc_core::DataTransportMode::Http, + data_transport: None, }; (client, replay) } @@ -13381,4 +14012,558 @@ mod tests { .expect("third read"); assert_eq!(third, 0); } + + #[derive(Clone, Copy)] + enum ExternalBehavior { + Reject, + PartialReject, + Success, + Unknown, + } + + struct TestExternalTransport { + behavior: ExternalBehavior, + operations: std::sync::Mutex>, + } + + #[async_trait] + impl ObjectDataTransport for TestExternalTransport { + async fn download( + &self, + request: DownloadRequest<'_>, + target: &mut (dyn AsyncWrite + Unpin + Send), + ) -> data_transport::DataTransportResult { + self.operations.lock().expect("record GET").push(format!( + "get:{:?}:{:?}", + request.options.version_id, request.max_bytes + )); + assert!(request.max_receive_bytes <= rc_core::S3_MAX_OBJECT_SIZE); + if matches!(self.behavior, ExternalBehavior::Reject) { + return Err(data_transport::DataTransportError { + state: rc_core::DataTransferState::NotAccepted, + message: "provider unavailable".to_string(), + source: None, + }); + } + target + .write_all(b"data") + .await + .expect("write external data"); + if matches!(self.behavior, ExternalBehavior::PartialReject) { + return Err(data_transport::DataTransportError { + state: rc_core::DataTransferState::NotAccepted, + message: "provider unavailable".to_string(), + source: None, + }); + } + target.flush().await.expect("flush external data"); + (request.on_progress)(4, Some(4)); + Ok(data_transport::DownloadResult { + bytes_transferred: 4, + content_length: 4, + }) + } + + async fn upload( + &self, + request: UploadRequest<'_>, + source: &mut (dyn tokio::io::AsyncRead + Unpin + Send), + ) -> data_transport::DataTransportResult { + self.operations + .lock() + .expect("record upload") + .push(format!("{:?}", request.operation)); + if matches!(self.behavior, ExternalBehavior::Reject) { + return Err(data_transport::DataTransportError { + state: rc_core::DataTransferState::NotAccepted, + message: "provider unavailable".to_string(), + source: None, + }); + } + let mut payload = Vec::new(); + source + .read_to_end(&mut payload) + .await + .expect("read external source"); + assert_eq!(payload.len() as u64, request.content_length); + if matches!(self.behavior, ExternalBehavior::Unknown) { + return Err(data_transport::DataTransportError { + state: rc_core::DataTransferState::OutcomeUnknown, + message: "temporary timeout".to_string(), + source: None, + }); + } + let mut object = ObjectInfo::file(&request.path.key, request.content_length as i64); + object.etag = Some("external-etag".to_string()); + object.content_type = request + .options + .attributes + .as_ref() + .and_then(|value| value.content_type.clone()); + Ok(UploadResult { + object, + checksum_sha256: request.checksum_sha256.map(ToString::to_string), + }) + } + } + + fn external_transport(behavior: ExternalBehavior) -> Arc { + Arc::new(TestExternalTransport { + behavior, + operations: std::sync::Mutex::new(Vec::new()), + }) + } + + #[tokio::test] + async fn external_auto_rejection_preserves_http_get_selection() { + let response = http::Response::builder() + .status(200) + .header("content-length", "4") + .body(SdkBody::from("http")) + .expect("HTTP GET response"); + let (client, receiver) = test_s3_client(Some(response)); + let provider = external_transport(ExternalBehavior::Reject); + let client = + client.with_object_data_transport(rc_core::DataTransportMode::Auto, provider.clone()); + let options = ObjectReadOptions { + version_id: Some("v1".to_string()), + }; + let result = client + .get_object_with_options(&RemotePath::new("test", "bucket", "key"), &options) + .await + .expect("safe HTTP fallback"); + assert_eq!(result, b"http"); + assert!(receiver.expect_request().uri().contains("versionId=v1")); + assert_eq!( + provider.operations.lock().expect("GET record")[0], + "get:Some(\"v1\"):None" + ); + } + + #[tokio::test] + async fn external_download_cannot_fallback_after_writer_poll() { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Auto, + external_transport(ExternalBehavior::PartialReject), + ); + let mut target = Vec::new(); + let error = client + .write_object_to(&RemotePath::new("test", "bucket", "key"), &mut target, None) + .await + .expect_err("partial output cannot replay"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::OutcomeUnknown, + .. + } + )); + assert_eq!(target, b"data"); + assert_eq!(replay.actual_requests().count(), 0); + } + + #[tokio::test] + async fn external_file_download_keeps_atomic_replacement() { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Rdma, + external_transport(ExternalBehavior::PartialReject), + ); + let directory = tempfile::tempdir().expect("destination directory"); + let destination = directory.path().join("object"); + std::fs::write(&destination, b"original").expect("existing destination"); + assert!( + client + .download_object_to_path( + &RemotePath::new("test", "bucket", "key"), + &destination, + |_, _| {} + ) + .await + .is_err() + ); + assert_eq!( + std::fs::read(&destination).expect("read destination"), + b"original" + ); + assert_eq!( + std::fs::read_dir(directory.path()) + .expect("list directory") + .count(), + 1 + ); + assert_eq!(replay.actual_requests().count(), 0); + } + + #[tokio::test] + async fn external_put_retains_options_and_unknown_outcome_classification() { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Auto, + external_transport(ExternalBehavior::Unknown), + ); + let error = client + .put_object( + &RemotePath::new("test", "bucket", "key"), + b"data".to_vec(), + Some("application/test"), + None, + ) + .await + .expect_err("unknown write cannot replay"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::OutcomeUnknown, + .. + } + )); + assert!(!rc_core::is_retryable_error(&error)); + assert_eq!(replay.actual_requests().count(), 0); + } + + #[tokio::test] + async fn external_conditional_file_upload_keeps_http_multipart_completion() { + let complete = http::Response::builder().status(200).header("content-type", "application/xml") + .body(SdkBody::from(r#"bucketkey"final-etag""#)) + .expect("multipart complete response"); + let (client, replay) = + test_s3_client_with_response_sequence(vec![multipart_create_response(), complete]); + let provider = external_transport(ExternalBehavior::Success); + let client = + client.with_object_data_transport(rc_core::DataTransportMode::Rdma, provider.clone()); + let mut source = tempfile::NamedTempFile::new().expect("file source"); + source.write_all(b"part data").expect("write source"); + client + .put_object_from_path_if_absent( + &RemotePath::new("test", "bucket", "key"), + source.path(), + None, + None, + |_| {}, + ) + .await + .expect("external part with HTTP completion"); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 2); + assert_eq!(requests[1].headers().get("if-none-match"), Some("*")); + assert!(provider.operations.lock().expect("part record")[0].starts_with("UploadPart")); + } + + #[tokio::test] + async fn external_rdmamode_without_provider_fails_only_data_operations() { + let (mut client, replay) = test_s3_client_with_response_sequence(Vec::new()); + client.data_transport_mode = rc_core::DataTransportMode::Rdma; + let error = client + .get_object(&RemotePath::new("test", "bucket", "key")) + .await + .expect_err("required provider missing"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::NotAccepted, + .. + } + )); + assert_eq!(replay.actual_requests().count(), 0); + } + #[test] + fn external_error_source_is_redacted_without_losing_business_class() { + let (client, _) = test_s3_client(None); + let error = Error::DataTransfer { + state: rc_core::DataTransferState::ConfirmedFailure, + message: "secret-key in response".to_string(), + source: Some(Box::new(Error::Auth("access-key secret-key".to_string()))), + }; + let error = client.redact_sensitive_error(error); + assert_eq!(error.exit_code(), 4); + let diagnostic = format!("{error:?}"); + assert!(!diagnostic.contains("secret-key")); + assert!(!diagnostic.contains("access-key")); + assert!(std::error::Error::source(&error).is_some()); + assert!(!rc_core::is_retryable_error(&error)); + } + + #[tokio::test] + async fn external_http_policy_does_not_invoke_installed_provider() { + let response = http::Response::builder() + .status(200) + .header("content-length", "4") + .body(SdkBody::from("http")) + .expect("HTTP GET response"); + let (client, receiver) = test_s3_client(Some(response)); + let provider = external_transport(ExternalBehavior::PartialReject); + let client = + client.with_object_data_transport(rc_core::DataTransportMode::Http, provider.clone()); + assert_eq!( + client + .get_object(&RemotePath::new("test", "bucket", "key")) + .await + .expect("unchanged HTTP path"), + b"http" + ); + assert!( + provider + .operations + .lock() + .expect("no provider calls") + .is_empty() + ); + receiver.expect_request(); + } + #[tokio::test] + async fn external_committed_put_checksum_failure_never_replays_upload() { + let response = http::Response::builder().status(503) + .header("x-amz-error-code", "ServiceUnavailable") + .body(SdkBody::from("ServiceUnavailabletemporary timeout")) + .expect("post-write HEAD failure"); + let (client, receiver) = test_s3_client(Some(response)); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Auto, + external_transport(ExternalBehavior::Success), + ); + let options = ObjectWriteOptions { + checksum: Some(ChecksumRequest::Calculate(ChecksumAlgorithm::Sha256)), + ..ObjectWriteOptions::default() + }; + let error = client + .put_object_with_options( + &RemotePath::new("test", "bucket", "key"), + b"data".to_vec(), + &options, + ) + .await + .expect_err("post-write verification failed"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::ConfirmedFailure, + .. + } + )); + assert!(!rc_core::is_retryable_error(&error)); + assert_eq!(receiver.expect_request().method(), "HEAD"); + } + #[tokio::test] + async fn external_multipart_completion_failure_does_not_replay_the_file() { + let complete = http::Response::builder().status(503) + .header("x-amz-error-code", "ServiceUnavailable") + .body(SdkBody::from("ServiceUnavailabletemporary timeout")) + .expect("uncertain completion response"); + let abort = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + complete, + abort, + ]); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Auto, + external_transport(ExternalBehavior::Success), + ); + let mut source = tempfile::NamedTempFile::new().expect("file source"); + source.write_all(b"data").expect("write source"); + let error = client + .put_object_from_path_if_absent( + &RemotePath::new("test", "bucket", "key"), + source.path(), + None, + None, + |_| {}, + ) + .await + .expect_err("uncertain completion"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::OutcomeUnknown, + .. + } + )); + assert!(!rc_core::is_retryable_error(&error)); + assert!(error.to_string().contains("upload-id")); + assert_eq!(replay.actual_requests().count(), 3); + } + + #[tokio::test] + async fn external_multipart_post_commit_checksum_failure_never_replays_the_file() { + let complete = http::Response::builder().status(200).header("content-type", "application/xml") + .body(SdkBody::from(r#"bucketkey"final-etag""#)) + .expect("multipart completion"); + let head = http::Response::builder().status(503).header("x-amz-error-code", "ServiceUnavailable") + .body(SdkBody::from("ServiceUnavailabletemporary timeout")) + .expect("post-completion HEAD failure"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_create_response(), + complete, + head, + ]); + let client = client.with_object_data_transport( + rc_core::DataTransportMode::Auto, + external_transport(ExternalBehavior::Success), + ); + let mut source = tempfile::NamedTempFile::new().expect("file source"); + source.write_all(b"data").expect("write source"); + let options = ObjectWriteOptions { + checksum: Some(ChecksumRequest::Calculate(ChecksumAlgorithm::Sha256)), + ..ObjectWriteOptions::default() + }; + let error = client + .put_object_multipart_from_path( + &RemotePath::new("test", "bucket", "key"), + source.path(), + 4, + PathUploadOptions { + write: &options, + precondition: ObjectWritePrecondition::None, + }, + |_| {}, + ) + .await + .expect_err("post-commit checksum failure"); + assert!(matches!( + error, + Error::DataTransfer { + state: rc_core::DataTransferState::ConfirmedFailure, + .. + } + )); + assert!(!rc_core::is_retryable_error(&error)); + assert_eq!(replay.actual_requests().count(), 3); + } + #[tokio::test] + async fn external_control_object_signing_matches_sdk_for_encoded_keys() { + for key in [ + "space key", + "percent%key", + "unicode-\u{1f680}", + "nested//key", + "symbols-:@!$&'()*+,;=[]", + ] { + let response = http::Response::builder() + .status(200) + .header("content-length", "0") + .body(SdkBody::empty()) + .expect("GET response"); + let (client, receiver) = + test_s3_client_with_endpoint("https://objects.example.test", Some(response)); + let version = "version +/% test"; + client + .get_object_with_options( + &RemotePath::new("test", "bucket", key), + &ObjectReadOptions { + version_id: Some(version.to_string()), + }, + ) + .await + .expect("capture SDK signature"); + let sdk_request = receiver.expect_request(); + let url = client + .external_control_url( + &RemotePath::new("test", "bucket", key), + &[ + ("versionId".to_string(), version.to_string()), + ("x-id".to_string(), "GetObject".to_string()), + ], + ) + .expect("control object URL"); + let sdk_url = reqwest::Url::parse(sdk_request.uri()).expect("SDK object URL"); + assert_eq!(url.path(), sdk_url.path(), "S3 object URL for {key}"); + let mut actual_query = url + .query() + .expect("control query") + .split('&') + .collect::>(); + let mut expected_query = sdk_url + .query() + .expect("SDK query") + .split('&') + .collect::>(); + actual_query.sort(); + expected_query.sort(); + assert_eq!(actual_query, expected_query, "S3 query escaping for {key}"); + let sdk_date = sdk_request + .headers() + .get("x-amz-date") + .expect("SDK signing date"); + let instant: Timestamp = format!( + "{}-{}-{}T{}:{}:{}Z", + &sdk_date[..4], + &sdk_date[4..6], + &sdk_date[6..8], + &sdk_date[9..11], + &sdk_date[11..13], + &sdk_date[13..15] + ) + .parse() + .expect("parse SDK signing instant"); + let time = std::time::UNIX_EPOCH + Duration::from_secs(instant.as_second() as u64); + let authorization = sdk_request + .headers() + .get("authorization") + .expect("SDK authorization"); + let signed_names = authorization + .split("SignedHeaders=") + .nth(1) + .expect("SDK signed header list") + .split(',') + .next() + .expect("SDK signed headers") + .split(';') + .collect::>(); + let mut headers = HeaderMap::new(); + for (name, value) in sdk_request.headers().iter() { + if signed_names.contains(&name) { + headers.insert( + HeaderName::from_bytes(name.as_bytes()).expect("SDK header name"), + HeaderValue::from_str(value).expect("SDK header value"), + ); + } + } + let signed = client + .sign_external_control_request_at( + &Method::GET, + sdk_request.uri(), + &headers, + &[], + time, + ) + .await + .expect("S3 control signature"); + assert_eq!( + signed + .get("authorization") + .expect("control authorization") + .to_str() + .expect("UTF-8 authorization"), + sdk_request + .headers() + .get("authorization") + .expect("SDK authorization"), + "S3 signature for {key}" + ); + } + } + + #[tokio::test] + async fn external_control_rejects_anonymous_principal_before_network_io() { + let (mut client, replay) = test_s3_client_with_response_sequence(Vec::new()); + client.alias.anonymous = true; + let request = ControlRequest { + method: Method::GET, + path: RemotePath::new("test", "bucket", "key"), + query: Vec::new(), + headers: HeaderMap::new(), + body: Bytes::new(), + response_limit: 1024, + }; + assert!(matches!( + ControlExecutor::execute(&client, request).await, + Err(Error::RequestRejected(_)) + )); + assert_eq!(replay.actual_requests().count(), 0); + } } diff --git a/crates/s3/src/data_transport.rs b/crates/s3/src/data_transport.rs new file mode 100644 index 00000000..207dbcc1 --- /dev/null +++ b/crates/s3/src/data_transport.rs @@ -0,0 +1,271 @@ +//! Transport-neutral ports for object payload delivery. +//! +//! The S3 client owns request policy, credentials, management operations, and +//! atomic file replacement. An injected provider owns only external payload IO. + +use std::sync::{Arc, OnceLock}; +use std::{ + pin::Pin, + task::{Context, Poll}, +}; + +use async_trait::async_trait; +use bytes::Bytes; +use http::{HeaderMap, Method, StatusCode}; +use rc_core::{ + DataTransferState, Error, ObjectInfo, ObjectWriteOptions, RemotePath, Result, + TransferReadOptions, +}; +use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; + +/// A bounded signed control exchange. It is executed once without HTTP retries. +pub struct ControlRequest { + pub method: Method, + pub path: RemotePath, + pub query: Vec<(String, String)>, + pub headers: HeaderMap, + pub body: Bytes, + pub response_limit: usize, +} + +/// Uninterpreted control response; duplicate and binary headers are preserved. +pub struct ControlResponse { + pub status: StatusCode, + pub headers: HeaderMap, + pub body: Bytes, +} + +/// Host-owned HTTPS, credentials, signing, custom headers, and TLS configuration. +#[async_trait] +pub trait ControlExecutor: Send + Sync { + async fn execute(&self, request: ControlRequest) -> Result; +} + +/// Configuration passed to a process-installed provider factory. +pub struct DataTransportContext { + pub endpoint: String, + pub control: Arc, +} + +/// External transport construction is separate from process-wide policy. +pub trait ObjectDataTransportFactory: Send + Sync { + fn create(&self, context: DataTransportContext) -> Result>; +} + +static FACTORY: OnceLock> = OnceLock::new(); + +/// Install a provider before running the CLI. Replacing a provider is rejected. +pub fn install_object_data_transport_factory( + factory: Arc, +) -> Result<()> { + FACTORY.set(factory).map_err(|_| { + Error::Config("An object data transport factory is already installed".to_string()) + }) +} + +pub(crate) fn installed_factory() -> Option> { + FACTORY.get().cloned() +} + +/// A GET's S3 selection and an independent finite receive bound. +pub struct DownloadRequest<'a> { + pub path: &'a RemotePath, + pub options: &'a TransferReadOptions, + /// Business Range selection; `None` requests the whole object. + pub max_bytes: Option, + /// Safety bound checked against the authenticated selected-object length. + pub max_receive_bytes: u64, + pub on_progress: &'a mut (dyn FnMut(u64, Option) + Send), +} + +/// Verified logical completion, including the selected S3 result metadata. +pub struct DownloadResult { + pub bytes_transferred: u64, + pub content_length: u64, +} + +/// S3 PUT conditions, independent of the external data protocol. +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub enum WriteCondition<'a> { + #[default] + None, + IfAbsent, + IfMatch(&'a str), +} + +/// Object write or one part of an existing HTTP-managed multipart upload. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum UploadOperation<'a> { + PutObject { + condition: WriteCondition<'a>, + }, + UploadPart { + upload_id: &'a str, + part_number: i32, + }, +} + +/// Known-length upload. All S3 options remain available to the provider. +pub struct UploadRequest<'a> { + pub path: &'a RemotePath, + pub operation: UploadOperation<'a>, + pub options: &'a ObjectWriteOptions, + pub content_length: u64, + /// The checksum already selected/calculated by the S3 operation planner. + pub checksum_sha256: Option<&'a str>, +} + +/// Authoritative S3 result after payload verification and business completion. +pub struct UploadResult { + pub object: ObjectInfo, + pub checksum_sha256: Option, +} + +/// A typed failure never eligible for generic timeout-based business retries. +#[derive(Debug, thiserror::Error)] +#[error("Object data transfer {state}: {message}")] +pub struct DataTransportError { + pub state: DataTransferState, + /// Safe diagnostics must exclude credentials, opaque offers, and raw bodies. + pub message: String, + /// Authoritative S3 error kept separate from replay classification. + pub source: Option>, +} + +impl From for Error { + fn from(error: DataTransportError) -> Self { + Self::DataTransfer { + state: error.state, + message: error.message, + source: error.source, + } + } +} + +pub type DataTransportResult = std::result::Result; + +pub(crate) fn committed_verification_failure(source: Error) -> Error { + Error::DataTransfer { + state: DataTransferState::ConfirmedFailure, + message: "The object was committed but its post-write verification failed".to_string(), + source: Some(Box::new(source)), + } +} + +/// One provider call performs one S3 data operation and its owned reconciliation. +/// +/// Success requires authenticated acceptance, exact logical length, data +/// verification, and an authoritative S3 result. `NotAccepted` guarantees that +/// neither supplied IO was polled nor any request accepted. All other failures +/// retain their classification; callers never infer replay safety from strings. +#[async_trait] +pub trait ObjectDataTransport: Send + Sync { + async fn download( + &self, + request: DownloadRequest<'_>, + target: &mut (dyn AsyncWrite + Unpin + Send), + ) -> DataTransportResult; + + async fn upload( + &self, + request: UploadRequest<'_>, + source: &mut (dyn AsyncRead + Unpin + Send), + ) -> DataTransportResult; +} + +pub(crate) struct ObservedWriter<'a, W: ?Sized> { + pub inner: &'a mut W, + pub polled: bool, + pub bytes: u64, + pub limit: u64, + pub flushed: bool, +} + +impl AsyncWrite for ObservedWriter<'_, W> { + fn poll_write( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + bytes: &[u8], + ) -> Poll> { + self.polled = true; + if bytes.len() as u64 > self.limit.saturating_sub(self.bytes) { + return Poll::Ready(Err(std::io::Error::other( + "External download exceeds the receive bound", + ))); + } + match Pin::new(&mut *self.inner).poll_write(cx, bytes) { + Poll::Ready(Ok(written)) => { + self.bytes += written as u64; + self.flushed = false; + Poll::Ready(Ok(written)) + } + result => result, + } + } + + fn poll_flush(mut self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll> { + self.polled = true; + let result = Pin::new(&mut *self.inner).poll_flush(cx); + if matches!(result, Poll::Ready(Ok(()))) { + self.flushed = true; + } + result + } + + fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll> { + Poll::Ready(Err(std::io::Error::other( + "External downloads must not shut down the caller's writer", + ))) + } +} + +pub(crate) struct ObservedReader<'a, R: ?Sized> { + pub inner: &'a mut R, + pub polled: bool, + pub bytes: u64, + pub eof: bool, +} + +impl AsyncRead for ObservedReader<'_, R> { + fn poll_read( + mut self: Pin<&mut Self>, + cx: &mut Context<'_>, + buffer: &mut ReadBuf<'_>, + ) -> Poll> { + self.polled = true; + let previous = buffer.filled().len(); + let had_capacity = buffer.remaining() > 0; + match Pin::new(&mut *self.inner).poll_read(cx, buffer) { + Poll::Ready(Ok(())) => { + let read = buffer.filled().len() - previous; + self.bytes += read as u64; + self.eof |= read == 0 && had_capacity; + Poll::Ready(Ok(())) + } + result => result, + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn failure_classification_survives_conversion() { + let error: Error = DataTransportError { + state: DataTransferState::OutcomeUnknown, + message: "timeout".to_string(), + source: None, + } + .into(); + assert!(matches!( + error, + Error::DataTransfer { + state: DataTransferState::OutcomeUnknown, + .. + } + )); + assert!(!rc_core::is_retryable_error(&error)); + } +} diff --git a/crates/s3/src/lib.rs b/crates/s3/src/lib.rs index 021879b3..5f005b50 100644 --- a/crates/s3/src/lib.rs +++ b/crates/s3/src/lib.rs @@ -6,6 +6,7 @@ pub mod admin; pub mod client; +pub mod data_transport; mod lifecycle_xml; pub mod multipart; mod ops; diff --git a/docs/ADR/0001-object-data-transport.md b/docs/ADR/0001-object-data-transport.md new file mode 100644 index 00000000..3f2f51f5 --- /dev/null +++ b/docs/ADR/0001-object-data-transport.md @@ -0,0 +1,50 @@ +# Object data transport ports + +The default executable uses HTTP for all S3 object payloads. The global +`--data-transport http|auto|rdma` option selects payload policy without changing +the existing `ObjectStore` interface. Management requests, multipart creation, +completion, and abort continue to use the established S3 implementation. + +`rc-core` owns transport selection and business outcome classification. +`rc-s3::data_transport` owns the optional provider ports, expressed with existing +S3 read/write options and borrowed asynchronous IO. A separately built executable +can install an `ObjectDataTransportFactory` before calling `rustfs_cli::run()`. +The common launcher preserves command dispatch, output, exit codes, and the +existing worker-thread configuration. + +The provider receives a host-owned `ControlExecutor` for bounded, signed HTTPS +exchanges. Credentials, endpoint addressing, custom headers, and TLS remain in +the S3 client. The executor sends each exchange once and preserves response +status and duplicate headers. The provider owns acceptance proof and any +reconciliation required after an ambiguous response. + +External control requires an authenticated principal and verified HTTPS. Object +paths and query values use the S3 encoding rules, with canonical path +normalization disabled and single URI encoding during SigV4 signing. The +encoding and signature are checked against actual AWS SDK requests for reserved +characters, Unicode, repeated slashes, and selected version IDs. + +Object reads to vectors, writers, and temporary files use the same provider +entry point. Files retain atomic replacement only after verified data and writer +flush. Existing version, checksum, encryption, and Range options are forwarded. +Object writes and multipart parts retain their existing option validation and +checksum verification; conditional file writes retain HTTP-managed multipart +completion. No existing S3 options may be silently omitted by a provider. + +`auto` permits HTTP fallback only when the provider proves nonacceptance and the +host observes that the supplied IO was untouched. Partial output and unknown +write results never trigger automatic replay. External failures keep typed +outcome state through redaction and multipart cleanup; generic timeout-string +retry logic does not retry them. `rdma` reports an explicit failure when no +provider or usable capability exists. + +After a successful external PUT, a failed persisted-checksum HEAD cannot cause +the object write to replay. Multipart transfers record whether an external part +succeeded. Uncertain completion responses retain an unknown outcome and upload +identity; post-completion verification failures retain a confirmed failure with +their original diagnostic cause. These failures remain outside generic business +retry logic. + +Native provider code is outside the default workspace dependency graph. The +independent opt-in build and private dependency instructions are described in +the project build documentation. diff --git a/docs/rdma-build.md b/docs/rdma-build.md new file mode 100644 index 00000000..d5e605c5 --- /dev/null +++ b/docs/rdma-build.md @@ -0,0 +1,48 @@ +# Optional external object data build + +The normal workspace builds without access to the private transport repository: + +```sh +cargo build --release -p rustfs-cli +``` + +The optional build uses an independent manifest. Cargo resolves optional +dependencies even when their feature is disabled, so the normal workspace does +not declare a private path or Git dependency. + +An authorized developer enables the private build with: + +```sh +git submodule update --init vendor/rustfs-ucx +./scripts/build-rdma.sh --release +``` + +The equivalent Cargo command is: + +```sh +cargo +1.98.1 build --manifest-path rdma/Cargo.toml --features rdma --release -j 2 +``` + +The default output is `rdma/target/release/rc`. `RUSTFS_RDMA_TOOLCHAIN` selects +another installed toolchain satisfying the private dependency's MSRV, and +`CARGO_BUILD_JOBS` selects build parallelism. The submodule commit is pinned by +the parent repository; updating it is an explicit dependency update. + +Missing source causes the build helper to stop before invoking Cargo. Fetching +the private submodule requires repository authorization. The ordinary build does +not initialize it. The private startup entry installs the object data provider +and then runs the same public CLI launcher. + +`--data-transport http|auto|rdma` selects object payload policy. Its default is +`http`; management and metadata requests retain their existing transport. + +The public S3 client retains request options, signing, multipart management, +file publication and retry decisions. `rc-s3::data_transport` exposes the typed +provider and bounded signed-control interfaces. Transport implementation source +is compiled from the private submodule. + +The current provider reports a typed pre-admission unavailability error because +the data engine is not connected. HTTP remains the default. Automatic selection +may use HTTP only before any request is accepted or supplied IO is polled; +explicit RDMA selection reports the error. Successful port tests do not establish +device transfer, an operational S3 extension or a throughput result. diff --git a/rdma/Cargo.toml b/rdma/Cargo.toml new file mode 100644 index 00000000..89708999 --- /dev/null +++ b/rdma/Cargo.toml @@ -0,0 +1,33 @@ +[package] +name = "rustfs-cli-rdma" +version = "0.1.0" +edition = "2024" +rust-version = "1.98.1" +license = "Apache-2.0" +publish = false + +[workspace] +resolver = "3" + +[features] +default = [] +rdma = ["rustfs-ucx/object"] + +[lib] +path = "../vendor/rustfs-ucx/integrations/cli/src/lib.rs" + +[[bin]] +name = "rc" +path = "../vendor/rustfs-ucx/integrations/cli/src/main.rs" +required-features = ["rdma"] + +[dependencies] +rustfs-cli = { path = "../crates/cli" } +rc-core = { path = "../crates/core" } +rc-s3 = { path = "../crates/s3" } +rustfs-ucx = { path = "../vendor/rustfs-ucx", default-features = false } +async-trait = "0.1" +bytes = "1" +http = "1" +thiserror = "2" +tokio = { version = "1", features = ["io-util", "rt", "sync", "time"] } diff --git a/scripts/build-rdma.sh b/scripts/build-rdma.sh new file mode 100755 index 00000000..3eabf8dc --- /dev/null +++ b/scripts/build-rdma.sh @@ -0,0 +1,15 @@ +#!/usr/bin/env bash +set -euo pipefail + +ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" +private="$ROOT_DIR/vendor/rustfs-ucx" +if [[ ! -f "$private/Cargo.toml" || ! -f "$private/integrations/cli/src/main.rs" || ! -f "$private/integrations/cli/src/lib.rs" ]]; then + printf '%s\n' 'RDMA requires the authorized private submodule vendor/rustfs-ucx.' >&2 + printf '%s\n' 'Run: git submodule update --init vendor/rustfs-ucx' >&2 + printf '%s\n' 'Authenticate with an account that can read rustfs/rustfs-ucx.' >&2 + exit 2 +fi + +exec cargo "+${RUSTFS_RDMA_TOOLCHAIN:-1.98.1}" build \ + --manifest-path "$ROOT_DIR/rdma/Cargo.toml" --features rdma \ + -j "${CARGO_BUILD_JOBS:-2}" "$@" diff --git a/scripts/tests/test_build_rdma.py b/scripts/tests/test_build_rdma.py new file mode 100644 index 00000000..fc360b79 --- /dev/null +++ b/scripts/tests/test_build_rdma.py @@ -0,0 +1,84 @@ +#!/usr/bin/env python3 +"""Exercise the opt-in build boundary without loading a private dependency.""" + +from pathlib import Path +import os +import shutil +import subprocess +import tempfile +import unittest + + +class RdmaBuildTests(unittest.TestCase): + def setUp(self): + self.temp = tempfile.TemporaryDirectory() + self.addCleanup(self.temp.cleanup) + self.root = Path(self.temp.name) + (self.root / "scripts").mkdir() + self.script = self.root / "scripts/build-rdma.sh" + shutil.copyfile(Path(__file__).parents[1] / "build-rdma.sh", self.script) + self.env = dict(os.environ) + self.env.pop("RUSTFS_RDMA_TOOLCHAIN", None) + self.env.pop("CARGO_BUILD_JOBS", None) + fake_bin = self.root / "bin" + fake_bin.mkdir() + cargo = fake_bin / "cargo" + cargo.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\"\n") + cargo.chmod(0o755) + self.env["PATH"] = str(fake_bin) + os.pathsep + self.env["PATH"] + + def run_build(self, *args): + return subprocess.run(["bash", str(self.script), *args], env=self.env, + capture_output=True, text=True, check=False) + + def prepare_private(self): + private = self.root / "vendor/rustfs-ucx" + private.mkdir(parents=True) + (private / "Cargo.toml").write_text('[package]\nname = "rustfs-ucx"\n') + entry = private / "integrations/cli/src/main.rs" + entry.parent.mkdir(parents=True) + entry.write_text("// Fixture: only the build preflight reads this path.\n") + (entry.parent / "lib.rs").touch() + + def test_missing_private_source_fails_before_cargo(self): + result = self.run_build() + self.assertEqual(result.returncode, 2) + self.assertIn("private submodule", result.stderr) + self.assertEqual(result.stdout, "") + + def test_partial_checkout_is_rejected(self): + private = self.root / "vendor/rustfs-ucx" + private.mkdir(parents=True) + (private / "Cargo.toml").write_text('[package]\nname = "rustfs-ucx"\n') + result = self.run_build() + self.assertEqual(result.returncode, 2) + self.assertIn("private submodule", result.stderr) + self.assertEqual(result.stdout, "") + + def test_entry_without_manifest_is_rejected(self): + self.prepare_private() + (self.root / "vendor/rustfs-ucx/Cargo.toml").unlink() + result = self.run_build() + self.assertEqual(result.returncode, 2) + self.assertEqual(result.stdout, "") + + def test_opt_in_preserves_arguments_and_uses_separate_manifest(self): + self.prepare_private() + result = self.run_build("--release", "--offline") + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.splitlines(), ["+1.98.1", "build", "--manifest-path", + str(self.root / "rdma/Cargo.toml"), "--features", "rdma", "-j", "2", + "--release", "--offline"]) + + def test_toolchain_and_parallelism_are_explicit(self): + self.prepare_private() + self.env["RUSTFS_RDMA_TOOLCHAIN"] = "stable" + self.env["CARGO_BUILD_JOBS"] = "3" + result = self.run_build() + self.assertEqual(result.returncode, 0, result.stderr) + self.assertEqual(result.stdout.splitlines()[0], "+stable") + self.assertEqual(result.stdout.splitlines()[-2:], ["-j", "3"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx new file mode 160000 index 00000000..08655081 --- /dev/null +++ b/vendor/rustfs-ucx @@ -0,0 +1 @@ +Subproject commit 08655081cc73b6a770231ea045fa689f89cb6719 From dcb6bdcfe6a22a5ffa125efa1f1e0f22c6275648 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 02:00:31 +0800 Subject: [PATCH 02/15] chore: update private integration submodule --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 08655081..3c28a068 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 08655081cc73b6a770231ea045fa689f89cb6719 +Subproject commit 3c28a0684246770dc1df2cd0c54b2d82a9e6d2e4 From 017517f4bc0856cd7b691b3487ba87c101fb04d4 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 02:24:26 +0800 Subject: [PATCH 03/15] chore: update private integration submodule --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 3c28a068..b808bea6 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 3c28a0684246770dc1df2cd0c54b2d82a9e6d2e4 +Subproject commit b808bea68fe56b9ab836229ffa68b9c4c48a782a From 68a5566e07f7da7665fd03216bbc3903d45af44e Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 11:35:28 +0800 Subject: [PATCH 04/15] feat: enable external object transfers with signed S3 control --- Cargo.lock | 1 + crates/s3/Cargo.toml | 1 + crates/s3/src/client.rs | 203 +++++++++++++++++++++--- crates/s3/src/data_transport.rs | 15 ++ crates/s3/src/lib.rs | 1 + crates/s3/src/payload_checksum.rs | 252 ++++++++++++++++++++++++++++++ docs/rdma-build.md | 22 ++- rdma/Cargo.toml | 7 +- scripts/build-rdma.sh | 4 +- scripts/tests/test_build_rdma.py | 11 +- vendor/rustfs-ucx | 2 +- 11 files changed, 489 insertions(+), 30 deletions(-) create mode 100644 crates/s3/src/payload_checksum.rs diff --git a/Cargo.lock b/Cargo.lock index 3d085e09..b20336de 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2885,6 +2885,7 @@ dependencies = [ "aws-credential-types", "aws-sdk-s3", "aws-sigv4", + "aws-smithy-checksums", "aws-smithy-http-client", "aws-smithy-runtime-api", "aws-smithy-types", diff --git a/crates/s3/Cargo.toml b/crates/s3/Cargo.toml index 8804eaf6..6a1ac2e6 100644 --- a/crates/s3/Cargo.toml +++ b/crates/s3/Cargo.toml @@ -21,6 +21,7 @@ aws-config.workspace = true aws-credential-types.workspace = true aws-smithy-types.workspace = true aws-smithy-runtime-api.workspace = true +aws-smithy-checksums = "0.64.8" # Async tokio.workspace = true diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 50dc1d99..43030af3 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -62,9 +62,9 @@ use std::sync::Arc; use std::time::Duration; use crate::data_transport::{ - self, ControlExecutor, ControlRequest, ControlResponse, DataTransportContext, DownloadRequest, - ObjectDataTransport, ObservedReader, ObservedWriter, UploadOperation, UploadRequest, - UploadResult, WriteCondition, + self, ControlExecutor, ControlPayload, ControlRequest, ControlResponse, DataTransportContext, + DownloadRequest, ObjectDataTransport, ObservedReader, ObservedWriter, UploadOperation, + UploadRequest, UploadResult, WriteCondition, }; use tokio::io::AsyncReadExt; use tokio::io::AsyncWrite; @@ -217,6 +217,25 @@ async fn build_reqwest_client( client_cert: Option<&str>, client_key: Option<&str>, timeout: Option<&rc_core::alias::TimeoutConfig>, +) -> Result { + build_reqwest_client_with_read_policy( + insecure, + ca_bundle, + client_cert, + client_key, + timeout, + true, + ) + .await +} + +async fn build_reqwest_client_with_read_policy( + insecure: bool, + ca_bundle: Option<&str>, + client_cert: Option<&str>, + client_key: Option<&str>, + timeout: Option<&rc_core::alias::TimeoutConfig>, + apply_read_timeout: bool, ) -> Result { // NOTE: When `insecure = true`, `danger_accept_invalid_certs` disables all TLS // certificate verification. Any CA bundle provided will still be added to the @@ -226,9 +245,10 @@ async fn build_reqwest_client( .redirect(reqwest::redirect::Policy::none()); if let Some(timeout) = timeout { - builder = builder - .connect_timeout(Duration::from_millis(timeout.connect_ms)) - .read_timeout(Duration::from_millis(timeout.read_ms)); + builder = builder.connect_timeout(Duration::from_millis(timeout.connect_ms)); + if apply_read_timeout { + builder = builder.read_timeout(Duration::from_millis(timeout.read_ms)); + } } if let Some(bundle_path) = ca_bundle { @@ -1662,6 +1682,7 @@ pub struct S3Client { inner: aws_sdk_s3::Client, presign_inner: aws_sdk_s3::Client, xml_http_client: reqwest::Client, + external_http_client: Arc>, alias: Alias, request_headers: Vec, data_transport_mode: rc_core::DataTransportMode, @@ -1785,6 +1806,7 @@ impl S3Client { inner: client, presign_inner: presign_client, xml_http_client, + external_http_client: Arc::new(tokio::sync::OnceCell::new()), alias, request_headers, data_transport_mode: rc_core::global_data_transport_mode(), @@ -3621,17 +3643,21 @@ impl S3Client { url: &str, headers: &HeaderMap, body: &[u8], + payload: ControlPayload, time: std::time::SystemTime, ) -> Result { let mut settings = SigningSettings::default(); settings.percent_encoding_mode = PercentEncodingMode::Single; settings.uri_path_normalization_mode = UriPathNormalizationMode::Disabled; settings.payload_checksum_kind = PayloadChecksumKind::XAmzSha256; - self.sign_request_with_settings( + self.sign_request_with_body( method, url, headers, - body, + match payload { + ControlPayload::Body => SignableBody::Bytes(body), + ControlPayload::External { .. } => SignableBody::UnsignedPayload, + }, RequestSigningOptions { region: &self.alias.region, settings, @@ -3648,6 +3674,18 @@ impl S3Client { headers: &HeaderMap, body: &[u8], options: RequestSigningOptions<'_>, + ) -> Result { + self.sign_request_with_body(method, url, headers, SignableBody::Bytes(body), options) + .await + } + + async fn sign_request_with_body( + &self, + method: &Method, + url: &str, + headers: &HeaderMap, + body: SignableBody<'_>, + options: RequestSigningOptions<'_>, ) -> Result { if self.alias.anonymous { return Ok(headers.clone()); @@ -3679,13 +3717,9 @@ impl S3Client { .filter_map(|(k, v)| v.to_str().ok().map(|v| (k.as_str(), v))) .collect(); - let signable_request = SignableRequest::new( - method.as_str(), - url, - header_pairs.into_iter(), - SignableBody::Bytes(body), - ) - .map_err(|e| Error::Auth(format!("Failed to create signable request: {e}")))?; + let signable_request = + SignableRequest::new(method.as_str(), url, header_pairs.into_iter(), body) + .map_err(|e| Error::Auth(format!("Failed to create signable request: {e}")))?; let (signing_instructions, _) = sign(signable_request, &signing_params.into()) .map_err(|e| Error::Auth(format!("Failed to sign request: {e}")))? @@ -7344,6 +7378,22 @@ impl ControlExecutor for S3Client { "Control envelope exceeds the bounded exchange policy".to_string(), )); } + if matches!(request.payload, ControlPayload::External { .. }) && !request.body.is_empty() { + return Err(Error::RequestRejected( + "An external payload control request must have an empty HTTPS body".to_string(), + )); + } + let operation_deadline = match request.payload { + ControlPayload::Body => None, + ControlPayload::External { deadline } => { + if deadline <= std::time::Instant::now() { + return Err(Error::RequestRejected( + "The external control operation deadline has elapsed".to_string(), + )); + } + Some(deadline) + } + }; let url = self.external_control_url(&request.path, &request.query)?; let mut headers = request.headers; for name in [ @@ -7381,8 +7431,11 @@ impl ControlExecutor for S3Client { ); headers.insert( "x-amz-content-sha256", - HeaderValue::from_str(&Self::sha256_hash(&request.body)) - .map_err(|_| Error::RequestRejected("Invalid payload digest".to_string()))?, + HeaderValue::from_str(&match request.payload { + ControlPayload::Body => Self::sha256_hash(&request.body), + ControlPayload::External { .. } => "UNSIGNED-PAYLOAD".to_string(), + }) + .map_err(|_| Error::RequestRejected("Invalid payload digest".to_string()))?, ); if headers.values().any(|value| value.to_str().is_err()) { return Err(Error::RequestRejected( @@ -7395,14 +7448,43 @@ impl ControlExecutor for S3Client { url.as_str(), &headers, &request.body, + request.payload, std::time::SystemTime::now(), ) .await?; - let response = self - .xml_http_client + let client = if operation_deadline.is_some() { + self.external_http_client + .get_or_try_init(|| async { + build_reqwest_client_with_read_policy( + self.alias.insecure, + self.alias.ca_bundle.as_deref(), + self.alias.client_cert.as_deref(), + self.alias.client_key.as_deref(), + self.alias.timeout.as_ref(), + false, + ) + .await + }) + .await? + } else { + &self.xml_http_client + }; + let mut outgoing = client .request(request.method, url) .headers(signed) - .body(request.body) + .body(request.body); + if let Some(deadline) = operation_deadline { + let remaining = deadline + .checked_duration_since(std::time::Instant::now()) + .filter(|remaining| !remaining.is_zero()) + .ok_or_else(|| { + Error::RequestRejected( + "The external control operation deadline has elapsed".to_string(), + ) + })?; + outgoing = outgoing.timeout(remaining); + } + let response = outgoing .send() .await .map_err(|_| Error::Network("External control exchange failed".to_string()))?; @@ -7522,6 +7604,7 @@ mod tests { inner: aws_sdk_s3::Client::from_conf(config), presign_inner: aws_sdk_s3::Client::from_conf(presign_config), xml_http_client, + external_http_client: Arc::new(tokio::sync::OnceCell::new()), alias, request_headers, data_transport_mode: rc_core::DataTransportMode::Http, @@ -7580,6 +7663,7 @@ mod tests { inner: aws_sdk_s3::Client::from_conf(config), presign_inner: aws_sdk_s3::Client::from_conf(presign_config), xml_http_client: reqwest::Client::new(), + external_http_client: Arc::new(tokio::sync::OnceCell::new()), alias, request_headers, data_transport_mode: rc_core::DataTransportMode::Http, @@ -7711,6 +7795,7 @@ mod tests { inner: aws_sdk_s3::Client::from_conf(config.clone()), presign_inner: aws_sdk_s3::Client::from_conf(config), xml_http_client: reqwest::Client::new(), + external_http_client: Arc::new(tokio::sync::OnceCell::new()), alias, request_headers: Vec::new(), data_transport_mode: rc_core::DataTransportMode::Http, @@ -14529,6 +14614,7 @@ mod tests { sdk_request.uri(), &headers, &[], + ControlPayload::Body, time, ) .await @@ -14558,6 +14644,7 @@ mod tests { query: Vec::new(), headers: HeaderMap::new(), body: Bytes::new(), + payload: ControlPayload::Body, response_limit: 1024, }; assert!(matches!( @@ -14566,4 +14653,80 @@ mod tests { )); assert_eq!(replay.actual_requests().count(), 0); } + + #[tokio::test] + async fn external_control_rejects_a_body_with_separately_delivered_payload() { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let request = ControlRequest { + method: Method::PUT, + path: RemotePath::new("test", "bucket", "key"), + query: Vec::new(), + headers: HeaderMap::new(), + body: Bytes::from_static(b"unexpected control body"), + payload: ControlPayload::External { + deadline: Instant::now() + Duration::from_secs(30), + }, + response_limit: 1024, + }; + assert!(matches!( + ControlExecutor::execute(&client, request).await, + Err(Error::RequestRejected(_)) + )); + assert_eq!(replay.actual_requests().count(), 0); + } + + #[tokio::test] + async fn external_control_rejects_an_elapsed_operation_deadline_before_network_io() { + let (client, replay) = test_s3_client_with_response_sequence(Vec::new()); + let request = ControlRequest { + method: Method::PUT, + path: RemotePath::new("test", "bucket", "key"), + query: Vec::new(), + headers: HeaderMap::new(), + body: Bytes::new(), + payload: ControlPayload::External { + deadline: Instant::now(), + }, + response_limit: 1024, + }; + assert!(matches!( + ControlExecutor::execute(&client, request).await, + Err(Error::RequestRejected(_)) + )); + assert_eq!(replay.actual_requests().count(), 0); + assert!(client.external_http_client.get().is_none()); + } + + #[tokio::test] + async fn external_control_signs_external_payload_and_all_binding_headers() { + let (client, _) = test_s3_client_with_response_sequence(Vec::new()); + let mut headers = HeaderMap::new(); + headers.insert("x-payload-session", HeaderValue::from_static("session-a")); + headers.insert("x-payload-length", HeaderValue::from_static("17")); + let sign = |headers: HeaderMap| { + let client = &client; + async move { + client + .sign_external_control_request_at( + &Method::PUT, + "https://objects.example.test/bucket/key", + &headers, + &[], + ControlPayload::External { + deadline: Instant::now() + Duration::from_secs(30), + }, + std::time::UNIX_EPOCH + Duration::from_secs(1_700_000_000), + ) + .await + .expect("signed control request") + } + }; + let first = sign(headers.clone()).await; + assert_eq!(first["x-amz-content-sha256"], "UNSIGNED-PAYLOAD"); + let authorization = first["authorization"].to_str().expect("authorization text"); + assert!(authorization.contains("x-payload-length;x-payload-session")); + headers.insert("x-payload-session", HeaderValue::from_static("session-b")); + let second = sign(headers).await; + assert_ne!(first["authorization"], second["authorization"]); + } } diff --git a/crates/s3/src/data_transport.rs b/crates/s3/src/data_transport.rs index 207dbcc1..41cbe2be 100644 --- a/crates/s3/src/data_transport.rs +++ b/crates/s3/src/data_transport.rs @@ -18,6 +18,8 @@ use rc_core::{ }; use tokio::io::{AsyncRead, AsyncWrite, ReadBuf}; +pub use crate::payload_checksum::{PayloadChecksumStatus, PayloadChecksumVerifier}; + /// A bounded signed control exchange. It is executed once without HTTP retries. pub struct ControlRequest { pub method: Method, @@ -25,9 +27,22 @@ pub struct ControlRequest { pub query: Vec<(String, String)>, pub headers: HeaderMap, pub body: Bytes, + pub payload: ControlPayload, pub response_limit: usize, } +/// Signing policy for a bounded control body or separately delivered payload. +#[derive(Clone, Copy, Debug, Default, Eq, PartialEq)] +pub enum ControlPayload { + /// Sign the exact HTTPS control body with SHA256. + #[default] + Body, + /// Use S3 UNSIGNED-PAYLOAD for a logical payload delivered by the provider. + /// The HTTPS request body must be empty; logical integrity is verified by + /// the provider and the authenticated S3 operation before publication. + External { deadline: std::time::Instant }, +} + /// Uninterpreted control response; duplicate and binary headers are preserved. pub struct ControlResponse { pub status: StatusCode, diff --git a/crates/s3/src/lib.rs b/crates/s3/src/lib.rs index 5f005b50..e008014c 100644 --- a/crates/s3/src/lib.rs +++ b/crates/s3/src/lib.rs @@ -10,6 +10,7 @@ pub mod data_transport; mod lifecycle_xml; pub mod multipart; mod ops; +mod payload_checksum; mod select; mod watch; diff --git a/crates/s3/src/payload_checksum.rs b/crates/s3/src/payload_checksum.rs new file mode 100644 index 00000000..9da684e8 --- /dev/null +++ b/crates/s3/src/payload_checksum.rs @@ -0,0 +1,252 @@ +//! Streaming verification of logical S3 response checksum headers and trailers. + +use std::collections::HashSet; + +use aws_smithy_checksums::{ + ChecksumAlgorithm as SdkAlgorithm, + http::{CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER, HttpChecksum}, +}; +use base64::{Engine, engine::general_purpose::STANDARD}; +use http::HeaderMap; +use rc_core::{ChecksumAlgorithm, Error, ObjectChecksum, Result}; + +/// Verification result distinguishes composite metadata from a verified body. +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub enum PayloadChecksumStatus { + NotDeclared, + Verified(ChecksumAlgorithm), + Composite(ChecksumAlgorithm), +} + +/// Selects the same preferred supported checksum as the AWS S3 SDK. +/// +/// Construct this value from the selected response head before receiving bytes. +/// Finish with the complete logical headers after authenticated trailers arrive. +pub struct PayloadChecksumVerifier { + selected: Option, +} + +struct Selected { + algorithm: ChecksumAlgorithm, + header: &'static str, + hasher: Box, + composite: bool, + expected: Option, +} + +impl PayloadChecksumVerifier { + pub fn new(headers: &HeaderMap) -> Result { + let declared = declared_trailers(headers)?; + let composite_type = scalar(headers, "x-amz-checksum-type")? == Some("COMPOSITE"); + for name in CHECKSUM_ALGORITHMS_IN_PRIORITY_ORDER { + let algorithm: SdkAlgorithm = name.parse().map_err(|_| { + Error::General("The S3 checksum algorithm registry is invalid".to_string()) + })?; + let hasher = algorithm.into_impl(); + let header = hasher.header_name(); + let expected = scalar(headers, header)?; + if expected.is_none() && !declared.contains(header) { + continue; + } + let algorithm = core_algorithm(algorithm)?; + if let Some(value) = expected { + ObjectChecksum::new_persisted(algorithm, value)?; + } + let composite = composite_type || expected.is_some_and(is_composite); + return Ok(Self { + selected: Some(Selected { + algorithm, + header, + hasher, + composite, + expected: expected.map(str::to_string), + }), + }); + } + Ok(Self { selected: None }) + } + + pub fn update(&mut self, bytes: &[u8]) { + if let Some(selected) = &mut self.selected + && !selected.composite + { + selected.hasher.update(bytes); + } + } + + pub fn verify(self, headers: &HeaderMap) -> Result { + let Some(selected) = self.selected else { + return Ok(PayloadChecksumStatus::NotDeclared); + }; + let expected = scalar(headers, selected.header)?.ok_or_else(|| { + Error::General("The declared S3 checksum trailer is missing".to_string()) + })?; + if selected + .expected + .as_deref() + .is_some_and(|original| original != expected) + { + return Err(Error::General( + "The declared S3 checksum changed during payload delivery".to_string(), + )); + } + ObjectChecksum::new_persisted(selected.algorithm, expected)?; + if selected.composite + || is_composite(expected) + || scalar(headers, "x-amz-checksum-type")? == Some("COMPOSITE") + { + return Ok(PayloadChecksumStatus::Composite(selected.algorithm)); + } + let actual = STANDARD.encode(selected.hasher.finalize()); + if actual != expected { + return Err(Error::General( + "The selected S3 response checksum does not match its payload".to_string(), + )); + } + Ok(PayloadChecksumStatus::Verified(selected.algorithm)) + } +} + +fn core_algorithm(algorithm: SdkAlgorithm) -> Result { + match algorithm { + SdkAlgorithm::Crc32 => Ok(ChecksumAlgorithm::Crc32), + SdkAlgorithm::Crc32c => Ok(ChecksumAlgorithm::Crc32c), + SdkAlgorithm::Crc64Nvme => Ok(ChecksumAlgorithm::Crc64Nvme), + SdkAlgorithm::Sha1 => Ok(ChecksumAlgorithm::Sha1), + SdkAlgorithm::Sha256 => Ok(ChecksumAlgorithm::Sha256), + _ => Err(Error::UnsupportedFeature( + "The S3 response checksum algorithm is unsupported".to_string(), + )), + } +} + +fn scalar<'a>(headers: &'a HeaderMap, name: &str) -> Result> { + let mut values = headers.get_all(name).iter(); + let result = values + .next() + .map(|value| { + value + .to_str() + .map_err(|_| Error::General("An S3 checksum header is invalid".to_string())) + }) + .transpose()?; + if values.next().is_some() { + return Err(Error::General( + "An S3 checksum header is ambiguous".to_string(), + )); + } + Ok(result) +} + +fn declared_trailers(headers: &HeaderMap) -> Result> { + let mut declared = HashSet::new(); + for name in ["trailer", "x-amz-trailer"] { + for value in headers.get_all(name) { + let value = value + .to_str() + .map_err(|_| Error::General("The S3 trailer declaration is invalid".to_string()))?; + for name in value.split(',').map(str::trim) { + if name.is_empty() { + return Err(Error::General( + "The S3 trailer declaration is empty".to_string(), + )); + } + declared.insert(name.to_ascii_lowercase()); + } + } + } + Ok(declared) +} + +fn is_composite(value: &str) -> bool { + value + .rsplit_once('-') + .is_some_and(|(_, count)| count.parse::().is_ok_and(|count| count > 0)) +} + +#[cfg(test)] +mod tests { + use super::*; + use http::HeaderValue; + + #[test] + fn verifies_split_payload_and_rejects_mismatching_bytes() { + let mut headers = HeaderMap::new(); + headers.insert( + "x-amz-checksum-sha256", + HeaderValue::from_static("LPJNul+wow4m6DsqxbninhsWHlwfp0JecwQzYpOLmCQ="), + ); + let mut verifier = PayloadChecksumVerifier::new(&headers).unwrap(); + verifier.update(b"he"); + verifier.update(b"llo"); + assert_eq!( + verifier.verify(&headers).unwrap(), + PayloadChecksumStatus::Verified(ChecksumAlgorithm::Sha256) + ); + let mut verifier = PayloadChecksumVerifier::new(&headers).unwrap(); + verifier.update(b"other"); + assert!(verifier.verify(&headers).is_err()); + } + + #[test] + fn selects_the_sdk_priority_and_validates_a_declared_trailer() { + let mut headers = HeaderMap::new(); + headers.insert("trailer", HeaderValue::from_static("x-amz-checksum-crc32")); + let mut verifier = PayloadChecksumVerifier::new(&headers).unwrap(); + verifier.update(b"hello"); + headers.insert("x-amz-checksum-crc32", HeaderValue::from_static("NhCmhg==")); + headers.insert( + "x-amz-checksum-sha256", + HeaderValue::from_static("AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="), + ); + assert_eq!( + verifier.verify(&headers).unwrap(), + PayloadChecksumStatus::Verified(ChecksumAlgorithm::Crc32) + ); + } + + #[test] + fn missing_declared_checksum_and_ambiguous_headers_fail() { + let mut headers = HeaderMap::new(); + headers.insert( + "x-amz-trailer", + HeaderValue::from_static("x-amz-checksum-crc32"), + ); + assert!( + PayloadChecksumVerifier::new(&headers) + .unwrap() + .verify(&headers) + .is_err() + ); + headers.append("x-amz-checksum-crc32", HeaderValue::from_static("NhCmhg==")); + headers.append("x-amz-checksum-crc32", HeaderValue::from_static("NhCmhg==")); + assert!(PayloadChecksumVerifier::new(&headers).is_err()); + } + + #[test] + fn composite_metadata_is_explicitly_distinct_from_verified_bytes() { + let mut headers = HeaderMap::new(); + headers.insert( + "x-amz-checksum-sha256", + HeaderValue::from_static("LPJNul+wow4m6DsqxbninhsWHlwfp0JecwQzYpOLmCQ=-2"), + ); + let mut verifier = PayloadChecksumVerifier::new(&headers).unwrap(); + verifier.update(b"unrelated data"); + assert_eq!( + verifier.verify(&headers).unwrap(), + PayloadChecksumStatus::Composite(ChecksumAlgorithm::Sha256) + ); + } + + #[test] + fn absent_checksum_retains_an_explicit_result() { + let headers = HeaderMap::new(); + assert_eq!( + PayloadChecksumVerifier::new(&headers) + .unwrap() + .verify(&headers) + .unwrap(), + PayloadChecksumStatus::NotDeclared + ); + } +} diff --git a/docs/rdma-build.md b/docs/rdma-build.md index d5e605c5..c100d4ba 100644 --- a/docs/rdma-build.md +++ b/docs/rdma-build.md @@ -41,8 +41,20 @@ file publication and retry decisions. `rc-s3::data_transport` exposes the typed provider and bounded signed-control interfaces. Transport implementation source is compiled from the private submodule. -The current provider reports a typed pre-admission unavailability error because -the data engine is not connected. HTTP remains the default. Automatic selection -may use HTTP only before any request is accepted or supplied IO is polled; -explicit RDMA selection reports the error. Successful port tests do not establish -device transfer, an operational S3 extension or a throughput result. +The private provider discovers a matching gateway through a signed HTTPS HEAD +request before allocating transport resources or submitting an object write. +An endpoint which does not advertise the supported protocol version receives no +external-payload write. The provider then coordinates one signed S3 operation, +its external data, and its verified final result. + +Automatic selection may use HTTP only before the data operation is accepted or +supplied IO is polled. Explicit RDMA selection reports unavailable capability. +An unknown upload outcome cannot trigger a generic business retry. Multipart +creation, completion, and abort retain their existing S3 HTTPS implementation. + +The optional executable also defaults to HTTP. Use `--data-transport rdma` to +require the provider, or `--data-transport auto` to allow pre-admission fallback. +`RUSTFS_RDMA_OPERATION_TIMEOUT_SECONDS` sets the finite external operation +deadline, with a default of 3600 seconds. The private repository documents its +platform and runtime installation requirements. Successful interface tests do +not establish hardware throughput or latency. diff --git a/rdma/Cargo.toml b/rdma/Cargo.toml index 89708999..d227ab0e 100644 --- a/rdma/Cargo.toml +++ b/rdma/Cargo.toml @@ -11,7 +11,7 @@ resolver = "3" [features] default = [] -rdma = ["rustfs-ucx/object"] +rdma = ["rustfs-ucx/object", "rustfs-ucx/native"] [lib] path = "../vendor/rustfs-ucx/integrations/cli/src/lib.rs" @@ -27,7 +27,10 @@ rc-core = { path = "../crates/core" } rc-s3 = { path = "../crates/s3" } rustfs-ucx = { path = "../vendor/rustfs-ucx", default-features = false } async-trait = "0.1" +base64 = "0.22" bytes = "1" http = "1" +md-5 = "0.10" thiserror = "2" -tokio = { version = "1", features = ["io-util", "rt", "sync", "time"] } +tokio = { version = "1", features = ["io-util", "macros", "rt", "sync", "time"] } +zeroize = "1" diff --git a/scripts/build-rdma.sh b/scripts/build-rdma.sh index 3eabf8dc..3311d7dd 100755 --- a/scripts/build-rdma.sh +++ b/scripts/build-rdma.sh @@ -3,13 +3,15 @@ set -euo pipefail ROOT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)" private="$ROOT_DIR/vendor/rustfs-ucx" -if [[ ! -f "$private/Cargo.toml" || ! -f "$private/integrations/cli/src/main.rs" || ! -f "$private/integrations/cli/src/lib.rs" ]]; then +if [[ ! -f "$private/Cargo.toml" || ! -f "$private/integrations/cli/src/main.rs" || ! -f "$private/integrations/cli/src/lib.rs" || ! -f "$private/scripts/build-env.sh" ]]; then printf '%s\n' 'RDMA requires the authorized private submodule vendor/rustfs-ucx.' >&2 printf '%s\n' 'Run: git submodule update --init vendor/rustfs-ucx' >&2 printf '%s\n' 'Authenticate with an account that can read rustfs/rustfs-ucx.' >&2 exit 2 fi +source "$private/scripts/build-env.sh" + exec cargo "+${RUSTFS_RDMA_TOOLCHAIN:-1.98.1}" build \ --manifest-path "$ROOT_DIR/rdma/Cargo.toml" --features rdma \ -j "${CARGO_BUILD_JOBS:-2}" "$@" diff --git a/scripts/tests/test_build_rdma.py b/scripts/tests/test_build_rdma.py index fc360b79..272a2ec7 100644 --- a/scripts/tests/test_build_rdma.py +++ b/scripts/tests/test_build_rdma.py @@ -23,7 +23,7 @@ def setUp(self): fake_bin = self.root / "bin" fake_bin.mkdir() cargo = fake_bin / "cargo" - cargo.write_text("#!/bin/sh\nprintf '%s\\n' \"$@\"\n") + cargo.write_text("#!/bin/sh\n[ \"${RUSTFS_PRIVATE_BUILD_PROBE:-}\" = enabled ] || exit 7\nprintf '%s\\n' \"$@\"\n") cargo.chmod(0o755) self.env["PATH"] = str(fake_bin) + os.pathsep + self.env["PATH"] @@ -39,6 +39,8 @@ def prepare_private(self): entry.parent.mkdir(parents=True) entry.write_text("// Fixture: only the build preflight reads this path.\n") (entry.parent / "lib.rs").touch() + (private / "scripts").mkdir() + (private / "scripts/build-env.sh").write_text("export RUSTFS_PRIVATE_BUILD_PROBE=enabled\n") def test_missing_private_source_fails_before_cargo(self): result = self.run_build() @@ -62,6 +64,13 @@ def test_entry_without_manifest_is_rejected(self): self.assertEqual(result.returncode, 2) self.assertEqual(result.stdout, "") + def test_missing_private_build_environment_is_rejected(self): + self.prepare_private() + (self.root / "vendor/rustfs-ucx/scripts/build-env.sh").unlink() + result = self.run_build() + self.assertEqual(result.returncode, 2) + self.assertEqual(result.stdout, "") + def test_opt_in_preserves_arguments_and_uses_separate_manifest(self): self.prepare_private() result = self.run_build("--release", "--offline") diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index b808bea6..540325ec 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit b808bea68fe56b9ab836229ffa68b9c4c48a782a +Subproject commit 540325ec3bb94fda3c8d3f5a999cd2fc70cbc040 From a25e31125febfbfc9b4099852f0646e66af22c69 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 13:12:25 +0800 Subject: [PATCH 05/15] build: update private object transport integration Pin the unified private runtime and stream implementation. Seed first-time opt-in dependency resolution from the public CLI lockfile while preserving an existing independent lockfile. --- docs/rdma-build.md | 5 +++++ scripts/build-rdma.sh | 4 ++++ vendor/rustfs-ucx | 2 +- 3 files changed, 10 insertions(+), 1 deletion(-) diff --git a/docs/rdma-build.md b/docs/rdma-build.md index c100d4ba..c9e17ec8 100644 --- a/docs/rdma-build.md +++ b/docs/rdma-build.md @@ -20,9 +20,14 @@ git submodule update --init vendor/rustfs-ucx The equivalent Cargo command is: ```sh +source vendor/rustfs-ucx/scripts/build-env.sh +cp -n Cargo.lock rdma/Cargo.lock cargo +1.98.1 build --manifest-path rdma/Cargo.toml --features rdma --release -j 2 ``` +The helper seeds a missing opt-in lockfile from the reviewed public CLI lockfile. +Cargo then adds private dependencies; an existing opt-in lockfile is preserved. + The default output is `rdma/target/release/rc`. `RUSTFS_RDMA_TOOLCHAIN` selects another installed toolchain satisfying the private dependency's MSRV, and `CARGO_BUILD_JOBS` selects build parallelism. The submodule commit is pinned by diff --git a/scripts/build-rdma.sh b/scripts/build-rdma.sh index 3311d7dd..dcbf2a4f 100755 --- a/scripts/build-rdma.sh +++ b/scripts/build-rdma.sh @@ -12,6 +12,10 @@ fi source "$private/scripts/build-env.sh" +if [[ ! -f "$ROOT_DIR/rdma/Cargo.lock" ]]; then + cp "$ROOT_DIR/Cargo.lock" "$ROOT_DIR/rdma/Cargo.lock" +fi + exec cargo "+${RUSTFS_RDMA_TOOLCHAIN:-1.98.1}" build \ --manifest-path "$ROOT_DIR/rdma/Cargo.toml" --features rdma \ -j "${CARGO_BUILD_JOBS:-2}" "$@" diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 540325ec..a3b9d845 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 540325ec3bb94fda3c8d3f5a999cd2fc70cbc040 +Subproject commit a3b9d84572a1883bcb9b145f8552f53432e6e17d From 7617230796a16c3d1f768bbc159d2ef1b96ff3cb Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 14:17:09 +0800 Subject: [PATCH 06/15] feat(cli): integrate shared transport lifecycle --- crates/cli/src/launcher.rs | 76 +++++++++++++++++++++++++++++++++++--- crates/cli/src/lib.rs | 2 +- vendor/rustfs-ucx | 2 +- 3 files changed, 73 insertions(+), 7 deletions(-) diff --git a/crates/cli/src/launcher.rs b/crates/cli/src/launcher.rs index 5305c5f5..97df5670 100644 --- a/crates/cli/src/launcher.rs +++ b/crates/cli/src/launcher.rs @@ -36,6 +36,16 @@ fn build_env_filter(debug: bool, env_is_set: bool) -> EnvFilter { /// Run the shared CLI launcher, preserving its worker-thread and exit behavior. pub fn run() { + run_with_shutdown(|| async {}); +} + +/// Runs host cleanup after the command while its async runtime is still available. +/// The host bounds its own cleanup and reports retained resources as appropriate. +pub fn run_with_shutdown(shutdown: S) +where + S: FnOnce() -> F + Send + 'static, + F: Future + 'static, +{ let cli = Cli::parse(); let env_is_set = std::env::var_os(EnvFilter::DEFAULT_ENV).is_some(); let env_filter = build_env_filter(cli.debug, env_is_set); @@ -46,7 +56,7 @@ pub fn run() { .with(env_filter) .init(); - let exit_code = match run_command(cli) { + let exit_code = match run_command(cli, shutdown) { Ok(exit_code) => exit_code, Err(error) => { eprintln!("{error}"); @@ -57,8 +67,30 @@ pub fn run() { std::process::exit(exit_code.as_i32()); } -fn run_command(cli: Cli) -> Result { - run_async_command(move || commands::execute(cli), command_runtime) +fn run_command(cli: Cli, shutdown: S) -> Result +where + S: FnOnce() -> F + Send + 'static, + F: Future + 'static, +{ + run_async_command( + move || finish_with_shutdown(commands::execute(cli), shutdown), + command_runtime, + ) +} + +fn finish_with_shutdown(command: C, shutdown: S) -> impl Future +where + C: Future, + S: FnOnce() -> F, + F: Future, +{ + // Keep the cleanup future small even when an SDK command has a large state. + let command = Box::pin(command); + async move { + let output = command.await; + shutdown().await; + output + } } fn command_runtime() -> std::io::Result { @@ -134,6 +166,37 @@ mod tests { assert_eq!(exit_code, ExitCode::UnsupportedFeature); } + #[test] + fn host_cleanup_runs_after_a_failed_command_before_runtime_shutdown() { + let state = Arc::new(std::sync::atomic::AtomicUsize::new(0)); + let observed = Arc::clone(&state); + let exit_code = run_async_command( + move || { + let shutdown_state = Arc::clone(&state); + finish_with_shutdown( + async move { + state.store(1, Ordering::Release); + ExitCode::GeneralError + }, + move || async move { + assert_eq!(shutdown_state.load(Ordering::Acquire), 1); + assert_eq!(std::thread::current().name(), Some(COMMAND_THREAD_NAME)); + tokio::spawn(async move { + tokio::task::yield_now().await; + shutdown_state.store(2, Ordering::Release); + }) + .await + .expect("cleanup task completes in the command runtime"); + }, + ) + }, + test_runtime, + ) + .expect("host cleanup completes"); + assert_eq!(exit_code, ExitCode::GeneralError); + assert_eq!(observed.load(Ordering::Acquire), 2); + } + #[test] fn command_launcher_uses_the_named_worker_thread() { let thread_name = run_async_command( @@ -158,8 +221,11 @@ mod tests { usize::from(payload[0]) + usize::from(payload[LARGE_FUTURE_SIZE - 1]) } - let result = run_async_command(large_future, test_runtime) - .expect("large command future should complete on the sized worker stack"); + let result = run_async_command( + || finish_with_shutdown(large_future(), || async {}), + test_runtime, + ) + .expect("large command future should complete on the sized worker stack"); assert_eq!(result, 14); } diff --git a/crates/cli/src/lib.rs b/crates/cli/src/lib.rs index 21540f2d..9902e51a 100644 --- a/crates/cli/src/lib.rs +++ b/crates/cli/src/lib.rs @@ -6,7 +6,7 @@ pub mod commands; pub(crate) mod confirm; pub mod exit_code; mod launcher; -pub use launcher::run; +pub use launcher::{run, run_with_shutdown}; pub mod output; pub(crate) mod private_file; pub(crate) mod secret_input; diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index a3b9d845..55d02c95 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit a3b9d84572a1883bcb9b145f8552f53432e6e17d +Subproject commit 55d02c953366642d9aa986098079a9f9966101b7 From e18a48f357ec402a2f0028faa35141fec4909e49 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 14:52:09 +0800 Subject: [PATCH 07/15] build: pin reviewed transport preparation --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 55d02c95..b4865c99 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 55d02c953366642d9aa986098079a9f9966101b7 +Subproject commit b4865c99d2adf2a9d45a296871485a454d4f7330 From c23aa2014bfec08f61c8f73b86c4ba709b225bcb Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 17:36:40 +0800 Subject: [PATCH 08/15] chore(rdma): update private transfer implementation --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index b4865c99..d4c75ebe 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit b4865c99d2adf2a9d45a296871485a454d4f7330 +Subproject commit d4c75ebe98e033c7d5d904b1482dac822bec71af From f8c9dc71381c19a7576fb30714a11a8c85ce4e6b Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 17:54:18 +0800 Subject: [PATCH 09/15] fix(rdma): update transfer lifecycle and recovery --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index d4c75ebe..07ca814e 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit d4c75ebe98e033c7d5d904b1482dac822bec71af +Subproject commit 07ca814edaed6d8baf743e3add5adc43ae62a994 From 45c5b5e22938871af628f156fffef3b76dca1101 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 19:27:30 +0800 Subject: [PATCH 10/15] feat(phase-2): pin bounded RDMA stream transport --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 07ca814e..62c4dfe0 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 07ca814edaed6d8baf743e3add5adc43ae62a994 +Subproject commit 62c4dfe07c61fb3f2eaf8b42e3c339784f7f56d5 From 21f74e94b7745a376ec2e2bbcb4688f9a2f7a578 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Fri, 16 Oct 2026 19:52:57 +0800 Subject: [PATCH 11/15] feat(phase-2): pin reviewed RDMA resource fixes --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 62c4dfe0..ddc32ada 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 62c4dfe07c61fb3f2eaf8b42e3c339784f7f56d5 +Subproject commit ddc32ada6e2f96152f1d29c630af8d9721502ed2 From a0d8a8d08c0b33bd61146600c0588a7fe552d5e4 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Sun, 18 Oct 2026 22:45:16 +0800 Subject: [PATCH 12/15] fix(rdma): use bounded transport and recover write results --- docs/rdma-build.md | 5 +++++ vendor/rustfs-ucx | 2 +- 2 files changed, 6 insertions(+), 1 deletion(-) diff --git a/docs/rdma-build.md b/docs/rdma-build.md index c9e17ec8..2e131f45 100644 --- a/docs/rdma-build.md +++ b/docs/rdma-build.md @@ -63,3 +63,8 @@ require the provider, or `--data-transport auto` to allow pre-admission fallback deadline, with a default of 3600 seconds. The private repository documents its platform and runtime installation requirements. Successful interface tests do not establish hardware throughput or latency. + +Install the matching native dependencies from the pinned private engine's +guide on both endpoints. Capability negotiation precedes external transfer +admission. Ambiguous write results use authenticated queries for the original +operation within its deadline; they do not resubmit the object or part. diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index ddc32ada..e4efecae 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit ddc32ada6e2f96152f1d29c630af8d9721502ed2 +Subproject commit e4efecae458d6a1f37cc68d1f7ae5ab9faab5c0e From 377240091f0e89541203073a1fb0ad2418245f56 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Sun, 18 Oct 2026 23:50:46 +0800 Subject: [PATCH 13/15] fix(rdma): use event-driven native progress --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index e4efecae..a2cc2f6c 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit e4efecae458d6a1f37cc68d1f7ae5ab9faab5c0e +Subproject commit a2cc2f6c2903adb3e7b39ac9bb5bc8319b3a3895 From b2c51437a33db7430812ad9eb18c764b443e1af8 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Mon, 19 Oct 2026 04:11:29 +0800 Subject: [PATCH 14/15] chore(rdma): update private transport engine --- vendor/rustfs-ucx | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index a2cc2f6c..79f82d2e 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit a2cc2f6c2903adb3e7b39ac9bb5bc8319b3a3895 +Subproject commit 79f82d2e48945f420c02ffd3402f3a6d7a1d96a2 From 7f2f64b802afa4e0f19a9bad96c5a493f6429419 Mon Sep 17 00:00:00 2001 From: gatewayJ <835269233@qq.com> Date: Mon, 19 Oct 2026 16:58:49 +0800 Subject: [PATCH 15/15] fix(rdma): update private operation ownership and tracing --- rdma/Cargo.toml | 1 + vendor/rustfs-ucx | 2 +- 2 files changed, 2 insertions(+), 1 deletion(-) diff --git a/rdma/Cargo.toml b/rdma/Cargo.toml index d227ab0e..f01a0f3c 100644 --- a/rdma/Cargo.toml +++ b/rdma/Cargo.toml @@ -34,3 +34,4 @@ md-5 = "0.10" thiserror = "2" tokio = { version = "1", features = ["io-util", "macros", "rt", "sync", "time"] } zeroize = "1" +tracing = { version = "0.1", default-features = false, features = ["std"] } diff --git a/vendor/rustfs-ucx b/vendor/rustfs-ucx index 79f82d2e..addf084a 160000 --- a/vendor/rustfs-ucx +++ b/vendor/rustfs-ucx @@ -1 +1 @@ -Subproject commit 79f82d2e48945f420c02ffd3402f3a6d7a1d96a2 +Subproject commit addf084a6d56d6702f98b62cf313787a672b644f