From 9e44c2dd223c59a93e7708a65f320b4cf4cdfcfe Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 21 Jul 2026 14:30:22 +0800 Subject: [PATCH 1/3] fix(phase-2): harden async command launcher stack --- crates/cli/src/main.rs | 179 ++++++++++++++++++++++++++++++++++++++++- 1 file changed, 176 insertions(+), 3 deletions(-) diff --git a/crates/cli/src/main.rs b/crates/cli/src/main.rs index e8b41688..32b36ea7 100644 --- a/crates/cli/src/main.rs +++ b/crates/cli/src/main.rs @@ -11,6 +11,24 @@ mod exit_code; mod output; 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 { @@ -20,8 +38,7 @@ fn build_env_filter(debug: bool, env_is_set: bool) -> EnvFilter { } } -#[tokio::main] -async fn main() { +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); @@ -32,18 +49,174 @@ async fn main() { .with(env_filter) .init(); - let exit_code = commands::execute(cli).await; + install_command_panic_hook(); + + 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 install_command_panic_hook() { + let default_hook = std::panic::take_hook(); + std::panic::set_hook(Box::new(move |panic_info| { + let current_thread = std::thread::current(); + if current_thread.name() != Some(COMMAND_THREAD_NAME) { + default_hook(panic_info); + } + })); +} + +fn run_command(cli: Cli) -> Result { + run_async_command( + move || commands::execute(cli), + || { + tokio::runtime::Builder::new_multi_thread() + .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)?; + Ok(runtime.block_on(future_factory())) + }) +} + +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}; + + 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_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_without_exposing_payload() { + let error = run_on_command_thread::<(), _>(|| { + panic!("sensitive panic payload must not be propagated") + }) + .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("sensitive panic payload")); + } } From 2ae14f8330a3a3f70fd914017bfce6dbaa4bf5db Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 21 Jul 2026 14:08:41 +0800 Subject: [PATCH 2/3] feat(phase-3): add global transfer planning controls --- Cargo.lock | 1 + README.md | 15 + crates/cli/src/commands/cp.rs | 1090 ++++++++++++++++++++++++-- crates/cli/src/commands/mod.rs | 44 +- crates/cli/src/commands/mv.rs | 61 +- crates/cli/tests/help_contract.rs | 24 +- crates/cli/tests/transfer_planner.rs | 167 ++++ crates/core/Cargo.toml | 1 + crates/core/src/lib.rs | 5 + crates/core/src/transfer.rs | 684 ++++++++++++++++ docs/reference/rc/cp.md | 37 +- 11 files changed, 2024 insertions(+), 105 deletions(-) create mode 100644 crates/cli/tests/transfer_planner.rs create mode 100644 crates/core/src/transfer.rs diff --git a/Cargo.lock b/Cargo.lock index c7437004..3d643ea7 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -2392,6 +2392,7 @@ dependencies = [ "async-trait", "dirs", "futures", + "glob", "humansize", "jiff", "mockall", diff --git a/README.md b/README.md index cbc6bc4f..db874b79 100644 --- a/README.md +++ b/README.md @@ -126,6 +126,21 @@ rc share download local/bucket/file.txt --expire 24h rc tree local/bucket -L 3 ``` +Bulk copies accept one or more sources followed by a directory or remote-prefix target: + +```bash +rc cp ./january.csv ./february.csv local/reports/ --concurrency 8 --summary +rc cp -r ./reports/ local/archive/ --include '*.csv' --exclude 'private-*' --newer-than 7d +rc cp -r ./reports/ local/archive/ --rate-limit 10MiB/s --retry-attempts 5 --continue-on-error +``` + +When include rules are present, a path must match at least one of them. Exclude rules are applied +after include rules and always win, regardless of flag order. Age filters compare source metadata in +UTC: newer/older boundaries are strict, while `--rewind` includes the specified boundary. Rate, +concurrency, and retry settings are shared across the full command. Any failed item leaves the +aggregate command exit code non-zero even when `--continue-on-error` is used. An empty selection +succeeds by default; pass `--fail-empty` when automation should receive a not-found exit code. + ### Admin Operations (IAM) ```bash diff --git a/crates/cli/src/commands/cp.rs b/crates/cli/src/commands/cp.rs index 0bf8339f..775a3d1d 100644 --- a/crates/cli/src/commands/cp.rs +++ b/crates/cli/src/commands/cp.rs @@ -3,12 +3,18 @@ //! Copies objects between local filesystem and S3, or between S3 locations. use clap::Args; +use jiff::Timestamp; +use rc_core::alias::RetryConfig; use rc_core::{ - AliasManager, ObjectEncryptionRequest, ObjectStore as _, ParsedPath, RemotePath, parse_path, + AliasManager, Error, ObjectEncryptionRequest, ObjectStore as _, ParsedPath, RemotePath, + TransferCandidate, TransferControls, TransferExecutor, TransferOutcomeState, TransferPlan, + TransferSelection, parse_path, }; use rc_s3::S3Client; use serde::Serialize; +use std::collections::HashSet; use std::path::{Path, PathBuf}; +use std::sync::Arc; use crate::exit_code::ExitCode; use crate::output::{Formatter, OutputConfig, ProgressBar}; @@ -23,11 +29,12 @@ const REMOTE_PATH_SUGGESTION: &str = "Use a local filesystem path or a remote path in the form alias/bucket[/key]."; /// Copy objects -#[derive(Args, Debug)] +#[derive(Args, Clone, Debug)] #[command(after_help = CP_AFTER_HELP)] pub struct CpArgs { - /// Source path (local path or alias/bucket/key) - pub source: String, + /// Source paths (local paths or alias/bucket/key) + #[arg(required = true, num_args = 1.., value_name = "SOURCE")] + pub sources: Vec, /// Destination path (local path or alias/bucket/key) pub target: String, @@ -67,6 +74,84 @@ pub struct CpArgs { /// Apply SSE-KMS to the remote destination path as TARGET=KMS_KEY_ID #[arg(long = "enc-kms")] pub enc_kms: Vec, + + /// Include source-relative paths matching this glob (repeatable) + #[arg(long)] + pub include: Vec, + + /// Exclude source-relative paths matching this glob; exclusions always win (repeatable) + #[arg(long)] + pub exclude: Vec, + + /// Select objects modified more recently than this age (for example 1h or 7d) + #[arg(long)] + pub newer_than: Option, + + /// Select objects modified less recently than this age (for example 1h or 7d) + #[arg(long)] + pub older_than: Option, + + /// Select object state at or before a UTC timestamp or age + #[arg(long)] + pub rewind: Option, + + /// Maximum number of transfers in flight across the command + #[arg(long, default_value_t = 4)] + pub concurrency: usize, + + /// Aggregate transfer start rate in bytes per second (for example 10MiB/s) + #[arg(long)] + pub rate_limit: Option, + + /// Maximum attempts for transient transfer failures + #[arg(long, default_value_t = 3)] + pub retry_attempts: u32, + + /// Initial transient retry backoff in milliseconds + #[arg(long, default_value_t = 100)] + pub retry_initial_backoff_ms: u64, + + /// Maximum transient retry backoff in milliseconds + #[arg(long, default_value_t = 10_000)] + pub retry_max_backoff_ms: u64, + + /// Return not-found when selection produces no transfer candidates + #[arg(long)] + pub fail_empty: bool, + + /// Print deterministic aggregate transfer counters (human output) + #[arg(long)] + pub summary: bool, +} + +impl CpArgs { + pub(crate) fn single(source: impl Into, target: impl Into) -> Self { + Self { + sources: vec![source.into()], + target: target.into(), + recursive: false, + preserve: false, + continue_on_error: false, + overwrite: true, + dry_run: false, + storage_class: None, + content_type: None, + enc_s3: Vec::new(), + enc_kms: Vec::new(), + include: Vec::new(), + exclude: Vec::new(), + newer_than: None, + older_than: None, + rewind: None, + concurrency: 4, + rate_limit: None, + retry_attempts: 3, + retry_initial_backoff_ms: 100, + retry_max_backoff_ms: 10_000, + fail_empty: false, + summary: false, + } + } } #[derive(Debug, Serialize)] @@ -97,19 +182,38 @@ pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode { ); } + if formatter.is_json() && uses_transfer_planner(&args) { + return formatter.fail( + ExitCode::UnsupportedFeature, + "Bulk transfer planning currently supports human output only; JSON batch output requires the versioned output contract", + ); + } + + let selection = match build_transfer_selection(&args, Timestamp::now()) { + Ok(selection) => selection, + Err(error) => return formatter.fail(ExitCode::UsageError, &error), + }; + let controls = match build_transfer_controls(&args) { + Ok(controls) => controls, + Err(error) => return formatter.fail(ExitCode::UsageError, &error), + }; + let alias_manager = AliasManager::new().ok(); - // Parse source and target paths - let source = match parse_cp_path(&args.source, alias_manager.as_ref()) { - Ok(p) => p, - Err(e) => { - return formatter.fail_with_suggestion( - ExitCode::UsageError, - &format!("Invalid source path: {e}"), - REMOTE_PATH_SUGGESTION, - ); + // Parse every operand before starting any transfer. + let mut sources = Vec::with_capacity(args.sources.len()); + for source in &args.sources { + match parse_cp_path(source, alias_manager.as_ref()) { + Ok(path) => sources.push(path), + Err(error) => { + return formatter.fail_with_suggestion( + ExitCode::UsageError, + &format!("Invalid source path '{source}': {error}"), + REMOTE_PATH_SUGGESTION, + ); + } } - }; + } let target = match parse_cp_path(&args.target, alias_manager.as_ref()) { Ok(p) => p, @@ -122,15 +226,83 @@ pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode { } }; - // Determine copy direction - match (&source, &target) { + let target_is_container = is_container_target(&args.target, &target); + if args.sources.len() > 1 && !target_is_container { + return formatter.fail( + ExitCode::UsageError, + "Multiple copy sources require a directory or remote prefix destination ending in '/'", + ); + } + + let target_encryption = match parse_destination_encryption(&args.enc_s3, &args.enc_kms, &target) + { + Ok(encryption) => encryption, + Err(error) => return formatter.fail(ExitCode::UsageError, &error), + }; + + let single_local_to_local = matches!( + (sources.as_slice(), &target), + ([ParsedPath::Local(_)], ParsedPath::Local(_)) + ); + if uses_transfer_planner(&args) && !single_local_to_local { + return execute_transfer_plan( + &args, + sources, + target, + target_is_container, + target_encryption, + selection, + controls, + &formatter, + ) + .await; + } + + let Some(source) = sources.first() else { + return formatter.fail(ExitCode::UsageError, "At least one copy source is required"); + }; + + execute_single_copy( + source, + &target, + &args, + &formatter, + target_encryption.as_ref(), + ) + .await +} + +fn uses_transfer_planner(args: &CpArgs) -> bool { + args.sources.len() > 1 + || args.recursive + || !args.include.is_empty() + || !args.exclude.is_empty() + || args.newer_than.is_some() + || args.older_than.is_some() + || args.rewind.is_some() + || args.concurrency != 4 + || args.rate_limit.is_some() + || args.retry_attempts != 3 + || args.retry_initial_backoff_ms != 100 + || args.retry_max_backoff_ms != 10_000 + || args.fail_empty + || args.summary +} + +async fn execute_single_copy( + source: &ParsedPath, + target: &ParsedPath, + args: &CpArgs, + formatter: &Formatter, + encryption: Option<&ObjectEncryptionRequest>, +) -> ExitCode { + // Determine copy direction. + match (source, target) { (ParsedPath::Local(src), ParsedPath::Remote(dst)) => { - // Local to S3 - copy_local_to_s3(src, dst, &args, &formatter).await + copy_local_to_s3_prepared(src, dst, args, formatter, encryption).await } (ParsedPath::Remote(src), ParsedPath::Local(dst)) => { - // S3 to Local - copy_s3_to_local(src, dst, &args, &formatter).await + copy_s3_to_local(src, dst, args, formatter).await } (ParsedPath::Remote(src), ParsedPath::Remote(dst)) => { if args.recursive { @@ -139,8 +311,7 @@ pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode { "Recursive S3-to-S3 copy is not implemented", ); } - // S3 to S3 - copy_s3_to_s3(src, dst, &args, &formatter).await + copy_s3_to_s3_prepared(src, dst, args, formatter, encryption).await } (ParsedPath::Local(_), ParsedPath::Local(_)) => formatter.fail_with_suggestion( ExitCode::UsageError, @@ -150,6 +321,766 @@ pub async fn execute(args: CpArgs, output_config: OutputConfig) -> ExitCode { } } +#[derive(Debug, Clone)] +enum CpOperation { + LocalToRemote { + source: PathBuf, + target: RemotePath, + encryption: Option, + }, + RemoteToLocal { + source: RemotePath, + target: PathBuf, + }, + RemoteToRemote { + source: RemotePath, + target: RemotePath, + encryption: Option, + }, +} + +#[allow(clippy::too_many_arguments)] +async fn execute_transfer_plan( + args: &CpArgs, + sources: Vec, + target: ParsedPath, + target_is_container: bool, + encryption: Option, + selection: TransferSelection, + controls: TransferControls, + formatter: &Formatter, +) -> ExitCode { + let alias_manager = match AliasManager::new() { + Ok(manager) => manager, + Err(error) => { + return formatter.fail( + ExitCode::UsageError, + &format!("Failed to load aliases: {error}"), + ); + } + }; + + let candidates = match build_transfer_candidates( + &sources, + &target, + target_is_container, + args.recursive, + encryption, + &alias_manager, + ) + .await + { + Ok(candidates) => candidates, + Err(error) => { + return formatter.fail( + exit_code_for_core_error(&error), + &format!("Failed to plan copy: {error}"), + ); + } + }; + + let plan = TransferPlan::build(candidates, &selection); + if let Err(error) = validate_plan_targets(&plan) { + return formatter.fail(ExitCode::UsageError, &error.to_string()); + } + if plan.items.is_empty() { + if args.fail_empty { + return formatter.fail( + ExitCode::NotFound, + "No copy sources matched the requested selection", + ); + } + if args.summary || args.recursive || args.sources.len() > 1 { + print_transfer_summary(formatter, &plan.summary); + } + return ExitCode::Success; + } + + if args.dry_run { + for item in &plan.items { + formatter.println(&format!( + "Would copy: {} -> {}", + formatter.style_file(&item.source), + formatter.style_file(&item.target) + )); + } + if args.summary || args.recursive || args.sources.len() > 1 { + print_transfer_summary(formatter, &plan.summary); + } + return ExitCode::Success; + } + + let executor = match TransferExecutor::new(controls) { + Ok(executor) => executor, + Err(error) => { + return formatter.fail(ExitCode::UsageError, &error.to_string()); + } + }; + let operation_args = Arc::new(args.clone()); + let report = executor + .execute(plan, { + let operation_args = Arc::clone(&operation_args); + move |item| { + let operation_args = Arc::clone(&operation_args); + async move { execute_planned_operation(item, &operation_args).await } + } + }) + .await; + + for outcome in &report.outcomes { + match &outcome.state { + TransferOutcomeState::Success { bytes_transferred } => { + print_planned_success(formatter, &outcome.item, *bytes_transferred); + } + TransferOutcomeState::Failed { error } => { + formatter.error_with_code(exit_code_for_core_error(error), &error.to_string()); + } + TransferOutcomeState::Cancelled => formatter.warning(&format!( + "Cancelled before transfer: {}", + outcome.item.source + )), + } + } + + if args.summary || args.recursive || args.sources.len() > 1 { + print_transfer_summary(formatter, &report.summary); + } + + report + .first_failure() + .map_or(ExitCode::Success, exit_code_for_core_error) +} + +async fn execute_planned_operation( + item: TransferCandidate, + args: &CpArgs, +) -> rc_core::Result { + match &item.payload { + CpOperation::LocalToRemote { + source, + target, + encryption, + } => perform_planned_upload(source, target, encryption.as_ref(), args).await, + CpOperation::RemoteToLocal { source, target } => { + perform_planned_download(source, target, args).await + } + CpOperation::RemoteToRemote { + source, + target, + encryption, + } => perform_planned_remote_copy(source, target, encryption.as_ref()).await, + } +} + +async fn perform_planned_upload( + source: &Path, + target: &RemotePath, + encryption: Option<&ObjectEncryptionRequest>, + args: &CpArgs, +) -> rc_core::Result { + let metadata = std::fs::metadata(source)?; + let file_size = metadata.len(); + let guessed_type = mime_guess::from_path(source) + .first() + .map(|mime| mime.essence_str().to_string()); + let content_type = select_upload_content_type( + args.content_type.as_deref(), + guessed_type.as_deref(), + file_size, + ); + let client = create_leaf_s3_client(&target.alias).await?; + let info = client + .put_object_from_path(target, source, content_type, encryption, |_| {}) + .await?; + Ok(info + .size_bytes + .and_then(|size| u64::try_from(size).ok()) + .unwrap_or(file_size)) +} + +async fn perform_planned_download( + source: &RemotePath, + target: &Path, + args: &CpArgs, +) -> rc_core::Result { + if target.exists() && !args.overwrite { + return Err(Error::Conflict(format!( + "Destination exists: {}. Use --overwrite to replace.", + target.display() + ))); + } + if let Some(parent) = target.parent() { + tokio::fs::create_dir_all(parent).await?; + } + let client = create_leaf_s3_client(&source.alias).await?; + client + .download_object_to_path(source, target, |_, _| {}) + .await +} + +async fn perform_planned_remote_copy( + source: &RemotePath, + target: &RemotePath, + encryption: Option<&ObjectEncryptionRequest>, +) -> rc_core::Result { + if source.alias != target.alias { + return Err(Error::UnsupportedFeature( + "Cross-alias S3-to-S3 copy is not supported".to_string(), + )); + } + let client = create_leaf_s3_client(&source.alias).await?; + let source_info = client.head_object(source).await?; + if source_info + .size_bytes + .is_some_and(|size| size > 5 * 1024 * 1024 * 1024) + { + return Err(Error::UnsupportedFeature( + "Objects larger than 5 GiB require multipart copy, which is not implemented" + .to_string(), + )); + } + let copied = client.copy_object(source, target, encryption).await?; + Ok(copied + .size_bytes + .or(source_info.size_bytes) + .and_then(|size| u64::try_from(size).ok()) + .unwrap_or_default()) +} + +async fn create_leaf_s3_client(alias_name: &str) -> rc_core::Result { + let alias_manager = AliasManager::new()?; + let mut alias = alias_manager + .get(alias_name) + .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?; + // The shared executor classifies failures and owns the exact attempt budget. Disabling adapter + // retries here prevents nested retries from exceeding the command-level policy. + alias.retry = Some(RetryConfig { + max_attempts: 1, + initial_backoff_ms: 1, + max_backoff_ms: 1, + }); + S3Client::new(alias).await +} + +fn print_planned_success( + formatter: &Formatter, + item: &TransferCandidate, + bytes_transferred: u64, +) { + let size_bytes = i64::try_from(bytes_transferred).ok(); + let size_human = size_bytes.map(|size| humansize::format_size(size as u64, humansize::BINARY)); + if formatter.is_json() { + formatter.json(&CpOutput { + status: "success", + source: item.source.clone(), + target: item.target.clone(), + size_bytes, + size_human, + }); + } else { + formatter.println(&format!( + "{} -> {} ({})", + formatter.style_file(&item.source), + formatter.style_file(&item.target), + formatter.style_size(size_human.as_deref().unwrap_or_default()) + )); + } +} + +fn print_transfer_summary(formatter: &Formatter, summary: &rc_core::TransferSummary) { + formatter.println(&format!( + "Summary: {} planned, {} skipped, {} succeeded, {} failed, {} cancelled, {} transferred", + summary.planned, + summary.skipped, + summary.successful, + summary.failed, + summary.cancelled, + humansize::format_size(summary.transferred_bytes, humansize::BINARY) + )); +} + +fn exit_code_for_core_error(error: &Error) -> ExitCode { + ExitCode::from_i32(error.exit_code()).unwrap_or(ExitCode::GeneralError) +} + +fn build_transfer_selection(args: &CpArgs, now: Timestamp) -> Result { + let newer_than = args + .newer_than + .as_deref() + .map(|value| parse_age_cutoff(value, now)) + .transpose()?; + let older_than = args + .older_than + .as_deref() + .map(|value| parse_age_cutoff(value, now)) + .transpose()?; + let rewind = args + .rewind + .as_deref() + .map(|value| parse_rewind_cutoff(value, now)) + .transpose()?; + + TransferSelection::new(&args.include, &args.exclude, newer_than, older_than, rewind) + .map_err(|error| error.to_string()) +} + +fn build_transfer_controls(args: &CpArgs) -> Result { + let bytes_per_second = args + .rate_limit + .as_deref() + .map(parse_byte_rate) + .transpose()?; + let controls = TransferControls { + concurrency: args.concurrency, + bytes_per_second, + retry: RetryConfig { + max_attempts: args.retry_attempts, + initial_backoff_ms: args.retry_initial_backoff_ms, + max_backoff_ms: args.retry_max_backoff_ms, + }, + continue_on_error: args.continue_on_error, + }; + controls.validate().map_err(|error| error.to_string())?; + Ok(controls) +} + +fn parse_age_cutoff(value: &str, now: Timestamp) -> Result { + let value = value.trim(); + if value.is_empty() { + return Err("Transfer age must not be empty".to_string()); + } + let suffix_index = value + .find(|character: char| character.is_ascii_alphabetic()) + .unwrap_or(value.len()); + let number = &value[..suffix_index]; + let suffix = &value[suffix_index..]; + let amount: i64 = number + .parse() + .map_err(|_| format!("Invalid transfer age number: {number}"))?; + if amount < 0 { + return Err("Transfer age must not be negative".to_string()); + } + let multiplier = match suffix.to_ascii_lowercase().as_str() { + "" | "s" => 1, + "m" => 60, + "h" => 3_600, + "d" => 86_400, + "w" => 604_800, + _ => return Err(format!("Unknown transfer age suffix: {suffix}")), + }; + let seconds = amount + .checked_mul(multiplier) + .ok_or_else(|| "Transfer age is too large".to_string())?; + now.checked_sub(jiff::Span::new().seconds(seconds)) + .map_err(|error| format!("Transfer age overflow: {error}")) +} + +fn parse_rewind_cutoff(value: &str, now: Timestamp) -> Result { + value + .parse::() + .or_else(|_| parse_age_cutoff(value, now)) + .map_err(|error| format!("Invalid rewind value '{value}': {error}")) +} + +fn parse_byte_rate(value: &str) -> Result { + let normalized = value.trim().to_ascii_lowercase(); + let normalized = normalized + .strip_suffix("/s") + .or_else(|| normalized.strip_suffix("ps")) + .unwrap_or(&normalized); + let suffix_index = normalized + .find(|character: char| character.is_ascii_alphabetic()) + .unwrap_or(normalized.len()); + let number = &normalized[..suffix_index]; + let suffix = &normalized[suffix_index..]; + let amount: u64 = number + .parse() + .map_err(|_| format!("Invalid transfer rate number: {number}"))?; + let multiplier = match suffix { + "" | "b" => 1u64, + "k" | "kb" => 1_000, + "ki" | "kib" => 1_024, + "m" | "mb" => 1_000_000, + "mi" | "mib" => 1_048_576, + "g" | "gb" => 1_000_000_000, + "gi" | "gib" => 1_073_741_824, + _ => return Err(format!("Unknown transfer rate suffix: {suffix}")), + }; + let rate = amount + .checked_mul(multiplier) + .ok_or_else(|| "Transfer rate is too large".to_string())?; + if rate == 0 { + return Err("Transfer rate must be greater than zero".to_string()); + } + Ok(rate) +} + +fn is_container_target(raw: &str, target: &ParsedPath) -> bool { + match target { + ParsedPath::Local(path) => path.is_dir() || raw.ends_with(['/', '\\']), + ParsedPath::Remote(path) => path.key.is_empty() || raw.ends_with('/'), + } +} + +async fn build_transfer_candidates( + sources: &[ParsedPath], + target: &ParsedPath, + target_is_container: bool, + recursive: bool, + encryption: Option, + alias_manager: &AliasManager, +) -> rc_core::Result>> { + let mut candidates = Vec::new(); + let multiple_sources = sources.len() > 1; + + for source in sources { + match source { + ParsedPath::Local(source) => build_local_candidates( + source, + target, + target_is_container, + recursive, + multiple_sources, + encryption.clone(), + &mut candidates, + )?, + ParsedPath::Remote(source) => { + build_remote_candidates( + source, + target, + target_is_container, + recursive, + multiple_sources, + encryption.clone(), + alias_manager, + &mut candidates, + ) + .await?; + } + } + } + + Ok(candidates) +} + +fn validate_plan_targets(plan: &TransferPlan) -> rc_core::Result<()> { + let mut targets = HashSet::with_capacity(plan.items.len()); + for candidate in &plan.items { + if !targets.insert(candidate.target.clone()) { + return Err(Error::InvalidPath(format!( + "Multiple sources resolve to the same destination '{}'", + candidate.target + ))); + } + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +fn build_local_candidates( + source: &Path, + target: &ParsedPath, + target_is_container: bool, + recursive: bool, + multiple_sources: bool, + encryption: Option, + candidates: &mut Vec>, +) -> rc_core::Result<()> { + let ParsedPath::Remote(target) = target else { + return Err(Error::InvalidPath( + "Cannot copy between two local paths. Use the system cp command.".to_string(), + )); + }; + let metadata = std::fs::metadata(source).map_err(|error| { + if error.kind() == std::io::ErrorKind::NotFound { + Error::NotFound(format!("Source not found: {}", source.display())) + } else { + Error::Io(error) + } + })?; + + if metadata.is_file() { + let name = local_file_name(source)?; + let destination = if target_is_container { + remote_child(target, &name) + } else { + target.clone() + }; + candidates.push(local_transfer_candidate( + source.to_path_buf(), + destination, + name, + metadata, + encryption, + )); + return Ok(()); + } + + if !recursive { + return Err(Error::InvalidPath( + "Source is a directory. Use -r/--recursive to copy directories.".to_string(), + )); + } + if !target_is_container { + return Err(Error::InvalidPath( + "Recursive copy requires a directory or remote prefix destination ending in '/'" + .to_string(), + )); + } + + let mut files = Vec::new(); + collect_local_files(source, source, &mut files)?; + files.sort_by(|left, right| left.1.cmp(&right.1)); + let source_root = multiple_sources + .then(|| local_file_name(source)) + .transpose()?; + for (path, relative, metadata) in files { + let relative = relative.replace('\\', "/"); + let target_relative = source_root + .as_deref() + .map_or_else(|| relative.clone(), |root| format!("{root}/{relative}")); + candidates.push(local_transfer_candidate( + path, + remote_child(target, &target_relative), + relative, + metadata, + encryption.clone(), + )); + } + Ok(()) +} + +fn local_file_name(path: &Path) -> rc_core::Result { + path.file_name() + .map(|name| name.to_string_lossy().to_string()) + .filter(|name| !name.is_empty()) + .ok_or_else(|| Error::InvalidPath(format!("Path has no file name: {}", path.display()))) +} + +fn local_transfer_candidate( + source: PathBuf, + target: RemotePath, + relative_path: String, + metadata: std::fs::Metadata, + encryption: Option, +) -> TransferCandidate { + TransferCandidate { + payload: CpOperation::LocalToRemote { + source: source.clone(), + target: target.clone(), + encryption, + }, + source: source.display().to_string(), + target: target.to_string(), + relative_path, + modified: metadata + .modified() + .ok() + .and_then(|time| time.try_into().ok()), + size_bytes: Some(metadata.len()), + } +} + +fn collect_local_files( + directory: &Path, + base: &Path, + files: &mut Vec<(PathBuf, String, std::fs::Metadata)>, +) -> rc_core::Result<()> { + let mut entries = std::fs::read_dir(directory)?.collect::>>()?; + entries.sort_by_key(std::fs::DirEntry::file_name); + for entry in entries { + let path = entry.path(); + let file_type = entry.file_type()?; + if file_type.is_symlink() { + return Err(Error::InvalidPath(format!( + "Symbolic links are not supported in recursive copy: {}", + path.display() + ))); + } + let metadata = entry.metadata()?; + if file_type.is_file() { + let relative = path + .strip_prefix(base) + .map_err(|error| Error::InvalidPath(error.to_string()))? + .to_string_lossy() + .to_string(); + files.push((path, relative, metadata)); + } else if file_type.is_dir() { + collect_local_files(&path, base, files)?; + } + } + Ok(()) +} + +#[allow(clippy::too_many_arguments)] +async fn build_remote_candidates( + source: &RemotePath, + target: &ParsedPath, + target_is_container: bool, + recursive: bool, + multiple_sources: bool, + encryption: Option, + alias_manager: &AliasManager, + candidates: &mut Vec>, +) -> rc_core::Result<()> { + let is_prefix = source.key.is_empty() || source.key.ends_with('/') || recursive; + let client = create_s3_client(alias_manager, &source.alias).await?; + + if is_prefix { + if !recursive { + return Err(Error::InvalidPath( + "Remote prefix copy requires -r/--recursive".to_string(), + )); + } + let ParsedPath::Local(target_root) = target else { + return Err(Error::UnsupportedFeature( + "Recursive S3-to-S3 copy is not implemented".to_string(), + )); + }; + if !target_is_container { + return Err(Error::InvalidPath( + "Recursive copy requires a directory destination ending in '/'".to_string(), + )); + } + + let source_root = if multiple_sources { + source + .key + .trim_end_matches('/') + .rsplit('/') + .next() + .filter(|value| !value.is_empty()) + .unwrap_or(&source.bucket) + } else { + "" + }; + let mut continuation_token = None; + loop { + let result = client + .list_objects( + source, + rc_core::ListOptions { + recursive: true, + max_keys: Some(1_000), + continuation_token: continuation_token.clone(), + ..Default::default() + }, + ) + .await?; + for object in result.items.into_iter().filter(|object| !object.is_dir) { + let relative = safe_download_relative_path(&object.key, &source.key) + .map_err(Error::InvalidPath)?; + let relative_string = relative.to_string_lossy().replace('\\', "/"); + let target_relative = if source_root.is_empty() { + relative.clone() + } else { + PathBuf::from(source_root).join(&relative) + }; + let destination = safe_download_destination(target_root, &target_relative) + .await + .map_err(Error::InvalidPath)?; + let object_source = RemotePath::new(&source.alias, &source.bucket, &object.key); + candidates.push(TransferCandidate { + payload: CpOperation::RemoteToLocal { + source: object_source.clone(), + target: destination.clone(), + }, + source: object_source.to_string(), + target: destination.display().to_string(), + relative_path: relative_string, + modified: object.last_modified, + size_bytes: object.size_bytes.and_then(|size| u64::try_from(size).ok()), + }); + } + if !result.truncated { + break; + } + continuation_token = result.continuation_token; + } + return Ok(()); + } + + let object = client.head_object(source).await?; + let name = source + .key + .rsplit('/') + .next() + .filter(|name| !name.is_empty()) + .ok_or_else(|| Error::InvalidPath(format!("Object key is empty: {source}")))?; + let size_bytes = object.size_bytes.and_then(|size| u64::try_from(size).ok()); + let modified = object.last_modified; + + match target { + ParsedPath::Local(target) => { + let destination = if target_is_container { + let relative = safe_download_relative_path(name, "").map_err(Error::InvalidPath)?; + safe_download_destination(target, &relative) + .await + .map_err(Error::InvalidPath)? + } else { + target.clone() + }; + candidates.push(TransferCandidate { + payload: CpOperation::RemoteToLocal { + source: source.clone(), + target: destination.clone(), + }, + source: source.to_string(), + target: destination.display().to_string(), + relative_path: name.to_string(), + modified, + size_bytes, + }); + } + ParsedPath::Remote(target) => { + if source.alias != target.alias { + return Err(Error::UnsupportedFeature( + "Cross-alias S3-to-S3 copy is not supported".to_string(), + )); + } + let destination = if target_is_container { + remote_child(target, name) + } else { + target.clone() + }; + candidates.push(TransferCandidate { + payload: CpOperation::RemoteToRemote { + source: source.clone(), + target: destination.clone(), + encryption, + }, + source: source.to_string(), + target: destination.to_string(), + relative_path: name.to_string(), + modified, + size_bytes, + }); + } + } + Ok(()) +} + +async fn create_s3_client( + alias_manager: &AliasManager, + alias_name: &str, +) -> rc_core::Result { + let alias = alias_manager + .get(alias_name) + .map_err(|_| Error::AliasNotFound(alias_name.to_string()))?; + S3Client::new(alias).await +} + +fn remote_child(parent: &RemotePath, relative: &str) -> RemotePath { + let key = if parent.key.is_empty() { + relative.to_string() + } else if parent.key.ends_with('/') { + format!("{}{}", parent.key, relative) + } else { + format!("{}/{}", parent.key, relative) + }; + RemotePath::new(&parent.alias, &parent.bucket, key) +} + fn parse_cp_path(path: &str, alias_manager: Option<&AliasManager>) -> rc_core::Result { let parsed = parse_path(path)?; @@ -170,20 +1101,13 @@ fn parse_cp_path(path: &str, alias_manager: Option<&AliasManager>) -> rc_core::R Ok(parsed) } -async fn copy_local_to_s3( +async fn copy_local_to_s3_prepared( src: &Path, dst: &RemotePath, args: &CpArgs, formatter: &Formatter, + encryption: Option<&ObjectEncryptionRequest>, ) -> ExitCode { - let target = ParsedPath::Remote(dst.clone()); - let encryption = match parse_destination_encryption(&args.enc_s3, &args.enc_kms, &target) { - Ok(encryption) => encryption, - Err(error) => { - return formatter.fail(ExitCode::UsageError, &error); - } - }; - // Check if source exists if !src.exists() { return formatter.fail_with_suggestion( @@ -221,7 +1145,6 @@ async fn copy_local_to_s3( ); } }; - let client = match S3Client::new(alias).await { Ok(c) => c, Err(e) => { @@ -234,10 +1157,10 @@ async fn copy_local_to_s3( if src.is_file() { // Single file upload - upload_file(&client, src, dst, args, formatter, encryption.as_ref()).await + upload_file(&client, src, dst, args, formatter, encryption).await } else { // Directory upload - upload_directory(&client, src, dst, args, formatter, encryption.as_ref()).await + upload_directory(&client, src, dst, args, formatter, encryption).await } } @@ -492,7 +1415,6 @@ async fn copy_s3_to_local( ); } }; - let client = match S3Client::new(alias).await { Ok(c) => c, Err(e) => { @@ -796,20 +1718,13 @@ pub(super) async fn safe_download_destination( Ok(destination) } -async fn copy_s3_to_s3( +async fn copy_s3_to_s3_prepared( src: &RemotePath, dst: &RemotePath, args: &CpArgs, formatter: &Formatter, + encryption: Option<&ObjectEncryptionRequest>, ) -> ExitCode { - let target = ParsedPath::Remote(dst.clone()); - let encryption = match parse_destination_encryption(&args.enc_s3, &args.enc_kms, &target) { - Ok(encryption) => encryption, - Err(error) => { - return formatter.fail(ExitCode::UsageError, &error); - } - }; - // For S3-to-S3, we need to handle same or different aliases let alias_manager = match AliasManager::new() { Ok(am) => am, @@ -838,7 +1753,6 @@ async fn copy_s3_to_s3( ); } }; - let client = match S3Client::new(alias).await { Ok(c) => c, Err(e) => { @@ -886,7 +1800,7 @@ async fn copy_s3_to_s3( } } - match client.copy_object(src, dst, encryption.as_ref()).await { + match client.copy_object(src, dst, encryption).await { Ok(info) => { if formatter.is_json() { let output = CpOutput { @@ -1172,6 +2086,22 @@ mod tests { assert!(result.is_err()); } + #[cfg(unix)] + #[test] + fn recursive_source_collection_rejects_symbolic_links() { + use std::os::unix::fs::symlink; + + let root = tempfile::tempdir().expect("create source root"); + let outside = tempfile::NamedTempFile::new().expect("create outside file"); + symlink(outside.path(), root.path().join("linked.txt")).expect("create source symlink"); + let mut files = Vec::new(); + + let result = collect_local_files(root.path(), root.path(), &mut files); + + assert!(result.is_err()); + assert!(files.is_empty()); + } + #[test] fn test_select_upload_content_type_uses_guess_for_small_files() { let selected = @@ -1243,22 +2173,64 @@ mod tests { #[test] fn test_cp_args_defaults() { - let args = CpArgs { - source: "src".to_string(), - target: "dst".to_string(), - recursive: false, - preserve: false, - continue_on_error: false, - overwrite: true, - dry_run: false, - storage_class: None, - content_type: None, - enc_s3: Vec::new(), - enc_kms: Vec::new(), - }; + let args = CpArgs::single("src", "dst"); assert!(args.overwrite); assert!(!args.recursive); assert!(!args.dry_run); + assert_eq!(args.concurrency, 4); + assert_eq!(args.retry_attempts, 3); + } + + #[test] + fn transfer_age_and_rewind_use_utc_cutoffs() { + let now: Timestamp = "2026-07-21T12:00:00Z".parse().expect("valid UTC timestamp"); + + assert_eq!( + parse_age_cutoff("1h", now).expect("valid age").to_string(), + "2026-07-21T11:00:00Z" + ); + assert_eq!( + parse_rewind_cutoff("2026-07-20T08:30:00+08:00", now) + .expect("valid offset timestamp") + .to_string(), + "2026-07-20T00:30:00Z" + ); + } + + #[test] + fn transfer_rate_parser_supports_decimal_and_binary_units() { + assert_eq!(parse_byte_rate("10MB/s").expect("decimal rate"), 10_000_000); + assert_eq!(parse_byte_rate("10MiB/s").expect("binary rate"), 10_485_760); + assert!(parse_byte_rate("0").is_err()); + assert!(parse_byte_rate("10widgets/s").is_err()); + } + + #[tokio::test] + async fn planner_rejects_two_sources_that_resolve_to_one_target() { + let (alias_manager, temp_dir) = temp_alias_manager(); + let first_dir = temp_dir.path().join("first"); + let second_dir = temp_dir.path().join("second"); + std::fs::create_dir_all(&first_dir).expect("create first source dir"); + std::fs::create_dir_all(&second_dir).expect("create second source dir"); + let first = first_dir.join("same.txt"); + let second = second_dir.join("same.txt"); + std::fs::write(&first, b"first").expect("write first source"); + std::fs::write(&second, b"second").expect("write second source"); + + let candidates = build_transfer_candidates( + &[ParsedPath::Local(first), ParsedPath::Local(second)], + &ParsedPath::Remote(RemotePath::new("target", "bucket", "prefix/")), + true, + false, + None, + &alias_manager, + ) + .await + .expect("sources can be expanded before selection"); + let plan = TransferPlan::build(candidates, &TransferSelection::default()); + let error = validate_plan_targets(&plan).expect_err("colliding destinations must fail"); + + assert!(error.to_string().contains("same destination")); } #[test] diff --git a/crates/cli/src/commands/mod.rs b/crates/cli/src/commands/mod.rs index 34606ca8..00eec68f 100644 --- a/crates/cli/src/commands/mod.rs +++ b/crates/cli/src/commands/mod.rs @@ -1092,7 +1092,7 @@ mod tests { match cli.command { Commands::Object(args) => match args.command { object::ObjectCommands::Copy(arg) => { - assert_eq!(arg.source, "./report.json"); + assert_eq!(arg.sources, ["./report.json"]); assert_eq!(arg.target, "local/my-bucket/reports/"); assert_eq!(arg.content_type.as_deref(), Some("application/json")); assert_eq!(arg.storage_class.as_deref(), Some("STANDARD_IA")); @@ -1104,6 +1104,48 @@ mod tests { } } + #[test] + fn cli_accepts_multiple_copy_sources_and_global_transfer_controls() { + let cli = Cli::try_parse_from([ + "rc", + "cp", + "./a.csv", + "./b.csv", + "local/reports/", + "--include", + "*.csv", + "--exclude", + "private-*", + "--newer-than", + "1h", + "--concurrency", + "8", + "--rate-limit", + "10MiB/s", + "--retry-attempts", + "5", + "--continue-on-error", + "--summary", + ]) + .expect("parse multi-source transfer controls"); + + match cli.command { + Commands::Cp(args) => { + assert_eq!(args.sources, ["./a.csv", "./b.csv"]); + assert_eq!(args.target, "local/reports/"); + assert_eq!(args.include, ["*.csv"]); + assert_eq!(args.exclude, ["private-*"]); + assert_eq!(args.newer_than.as_deref(), Some("1h")); + assert_eq!(args.concurrency, 8); + assert_eq!(args.rate_limit.as_deref(), Some("10MiB/s")); + assert_eq!(args.retry_attempts, 5); + assert!(args.continue_on_error); + assert!(args.summary); + } + other => panic!("expected cp command, got {other:?}"), + } + } + #[test] fn cli_accepts_object_move_with_recursive_dry_run() { let cli = Cli::try_parse_from([ diff --git a/crates/cli/src/commands/mv.rs b/crates/cli/src/commands/mv.rs index 16ae50a3..ade757e0 100644 --- a/crates/cli/src/commands/mv.rs +++ b/crates/cli/src/commands/mv.rs @@ -122,19 +122,15 @@ async fn move_local_to_s3( use crate::commands::cp; // First, copy local to S3 - let cp_args = cp::CpArgs { - source: src.to_string_lossy().to_string(), - target: format!("{}/{}/{}", dst.alias, dst.bucket, dst.key), - recursive: args.recursive, - preserve: false, - continue_on_error: args.continue_on_error, - overwrite: true, - dry_run: args.dry_run, - storage_class: None, - content_type: None, - enc_s3: args.enc_s3.clone(), - enc_kms: args.enc_kms.clone(), - }; + let mut cp_args = cp::CpArgs::single( + src.to_string_lossy().to_string(), + format!("{}/{}/{}", dst.alias, dst.bucket, dst.key), + ); + cp_args.recursive = args.recursive; + cp_args.continue_on_error = args.continue_on_error; + cp_args.dry_run = args.dry_run; + cp_args.enc_s3.clone_from(&args.enc_s3); + cp_args.enc_kms.clone_from(&args.enc_kms); let cp_result = cp::execute( cp_args, @@ -182,19 +178,15 @@ async fn move_s3_to_local( } // First, copy S3 to local - let cp_args = cp::CpArgs { - source: format!("{}/{}/{}", src.alias, src.bucket, src.key), - target: dst.to_string_lossy().to_string(), - recursive: args.recursive, - preserve: false, - continue_on_error: args.continue_on_error, - overwrite: true, - dry_run: args.dry_run, - storage_class: None, - content_type: None, - enc_s3: args.enc_s3.clone(), - enc_kms: args.enc_kms.clone(), - }; + let mut cp_args = cp::CpArgs::single( + format!("{}/{}/{}", src.alias, src.bucket, src.key), + dst.to_string_lossy().to_string(), + ); + cp_args.recursive = args.recursive; + cp_args.continue_on_error = args.continue_on_error; + cp_args.dry_run = args.dry_run; + cp_args.enc_s3.clone_from(&args.enc_s3); + cp_args.enc_kms.clone_from(&args.enc_kms); let cp_result = cp::execute( cp_args, @@ -304,19 +296,10 @@ async fn move_s3_prefix_to_local( continuation_token = result.continuation_token; } - let cp_args = cp::CpArgs { - source: src.to_string(), - target: dst.to_string_lossy().to_string(), - recursive: true, - preserve: false, - continue_on_error: args.continue_on_error, - overwrite: true, - dry_run: args.dry_run, - storage_class: None, - content_type: None, - enc_s3: Vec::new(), - enc_kms: Vec::new(), - }; + let mut cp_args = cp::CpArgs::single(src.to_string(), dst.to_string_lossy().to_string()); + cp_args.recursive = true; + cp_args.continue_on_error = args.continue_on_error; + cp_args.dry_run = args.dry_run; let mut errors = 0usize; for item in objects { diff --git a/crates/cli/tests/help_contract.rs b/crates/cli/tests/help_contract.rs index 1a0db130..1dc08820 100644 --- a/crates/cli/tests/help_contract.rs +++ b/crates/cli/tests/help_contract.rs @@ -287,7 +287,7 @@ fn top_level_command_help_contract() { }, HelpCase { args: &["cp"], - usage: "Usage: rc cp [OPTIONS] ", + usage: "Usage: rc cp [OPTIONS] ... ", expected_tokens: &[ "--recursive", "--preserve", @@ -298,6 +298,16 @@ fn top_level_command_help_contract() { "--content-type", "--enc-s3", "--enc-kms", + "--include", + "--exclude", + "--newer-than", + "--older-than", + "--rewind", + "--concurrency", + "--rate-limit", + "--retry-attempts", + "--fail-empty", + "--summary", "Examples:", "rc object copy ./report.json local/my-bucket/reports/", ], @@ -636,7 +646,7 @@ fn nested_subcommand_help_contract() { }, HelpCase { args: &["object", "copy"], - usage: "Usage: rc object copy [OPTIONS] ", + usage: "Usage: rc object copy [OPTIONS] ... ", expected_tokens: &[ "--recursive", "--preserve", @@ -645,6 +655,16 @@ fn nested_subcommand_help_contract() { "--dry-run", "--storage-class", "--content-type", + "--include", + "--exclude", + "--newer-than", + "--older-than", + "--rewind", + "--concurrency", + "--rate-limit", + "--retry-attempts", + "--fail-empty", + "--summary", "Examples:", "rc object copy ./report.json local/my-bucket/reports/", ], diff --git a/crates/cli/tests/transfer_planner.rs b/crates/cli/tests/transfer_planner.rs new file mode 100644 index 00000000..8ea6008d --- /dev/null +++ b/crates/cli/tests/transfer_planner.rs @@ -0,0 +1,167 @@ +//! Local-only exit-code coverage for bulk transfer preflight behavior. + +use std::path::{Path, PathBuf}; +use std::process::{Command, Output}; + +fn rc_binary() -> PathBuf { + if let Ok(path) = std::env::var("CARGO_BIN_EXE_rc") { + return PathBuf::from(path); + } + + let workspace_root = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .parent() + .expect("cli crate has parent directory") + .parent() + .expect("workspace root exists") + .to_path_buf(); + workspace_root.join("target/debug/rc") +} + +fn run_rc(args: &[&str], config_dir: &Path) -> Output { + let mut command = Command::new(rc_binary()); + for (key, _) in std::env::vars_os() { + if key.to_string_lossy().starts_with("RC_HOST_") { + command.env_remove(key); + } + } + command + .args(args) + .env("RC_CONFIG_DIR", config_dir) + .output() + .expect("execute rc") +} + +#[test] +fn multi_source_ambiguous_destination_fails_before_remote_access() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let sources = tempfile::tempdir().expect("create source dir"); + let first = sources.path().join("first.txt"); + let second = sources.path().join("second.txt"); + std::fs::write(&first, b"first").expect("write first source"); + std::fs::write(&second, b"second").expect("write second source"); + + let output = run_rc( + &[ + "cp", + first.to_str().expect("first path is UTF-8"), + second.to_str().expect("second path is UTF-8"), + "missing-alias/bucket/object.txt", + ], + config_dir.path(), + ); + + assert_eq!(output.status.code(), Some(2)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("Multiple copy sources require"), "{stderr}"); + assert!( + !stderr.contains("Alias 'missing-alias' not found"), + "{stderr}" + ); +} + +#[test] +fn empty_recursive_source_can_fail_with_not_found_exit_code() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::tempdir().expect("create empty source"); + + let output = run_rc( + &[ + "cp", + source.path().to_str().expect("source path is UTF-8"), + "missing-alias/bucket/", + "--recursive", + "--fail-empty", + ], + config_dir.path(), + ); + + assert_eq!(output.status.code(), Some(5)); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stderr.contains("No copy sources matched"), "{stderr}"); +} + +#[test] +fn empty_recursive_source_succeeds_and_reports_zero_plan() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::tempdir().expect("create empty source"); + + let output = run_rc( + &[ + "cp", + source.path().to_str().expect("source path is UTF-8"), + "missing-alias/bucket/", + "--recursive", + "--summary", + ], + config_dir.path(), + ); + + assert_eq!( + output.status.code(), + Some(0), + "stdout: {}\nstderr: {}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("0 planned"), "{stdout}"); + assert!(stdout.contains("0 B transferred"), "{stdout}"); +} + +#[test] +fn aggregate_failure_keeps_summary_and_returns_non_success_exit_code() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::NamedTempFile::new().expect("create source file"); + std::fs::write(source.path(), b"payload").expect("write source file"); + + let output = run_rc( + &[ + "cp", + source.path().to_str().expect("source path is UTF-8"), + "missing-alias/bucket/object.txt", + "--summary", + "--continue-on-error", + "--retry-attempts", + "1", + ], + config_dir.path(), + ); + + assert_eq!(output.status.code(), Some(5)); + let stdout = String::from_utf8_lossy(&output.stdout); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!(stdout.contains("1 planned"), "{stdout}"); + assert!(stdout.contains("1 failed"), "{stdout}"); + assert!( + stderr.contains("Alias not found: missing-alias"), + "{stderr}" + ); +} + +#[test] +fn planned_json_output_is_explicitly_deferred_to_versioned_contract() { + let config_dir = tempfile::tempdir().expect("create config dir"); + let source = tempfile::tempdir().expect("create empty source"); + + let output = run_rc( + &[ + "cp", + source.path().to_str().expect("source path is UTF-8"), + "missing-alias/bucket/", + "--recursive", + "--json", + ], + config_dir.path(), + ); + + assert_eq!(output.status.code(), Some(7)); + assert!(output.stdout.is_empty()); + let payload: serde_json::Value = + serde_json::from_slice(&output.stderr).expect("stderr is valid JSON"); + assert_eq!(payload["code"], 7); + assert!( + payload["error"] + .as_str() + .is_some_and(|message| message.contains("versioned output contract")) + ); +} diff --git a/crates/core/Cargo.toml b/crates/core/Cargo.toml index 97544358..bd4d31bb 100644 --- a/crates/core/Cargo.toml +++ b/crates/core/Cargo.toml @@ -35,6 +35,7 @@ jiff.workspace = true url.workspace = true urlencoding.workspace = true humansize.workspace = true +glob.workspace = true [dev-dependencies] tempfile.workspace = true diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index 906f091d..538fca5e 100644 --- a/crates/core/src/lib.rs +++ b/crates/core/src/lib.rs @@ -21,6 +21,7 @@ pub mod replication; pub mod retry; pub mod select; pub mod traits; +pub mod transfer; pub use alias::{ Alias, AliasManager, RequestHeader, global_request_headers, set_global_request_headers, @@ -50,3 +51,7 @@ pub use traits::{ BucketNotification, Capabilities, ListOptions, ListResult, NotificationTarget, ObjectInfo, ObjectStore, ObjectVersion, ObjectVersionListResult, }; +pub use transfer::{ + TransferCandidate, TransferControls, TransferExecutor, TransferOutcome, TransferOutcomeState, + TransferPlan, TransferReport, TransferSelection, TransferSummary, +}; diff --git a/crates/core/src/transfer.rs b/crates/core/src/transfer.rs new file mode 100644 index 00000000..911f6234 --- /dev/null +++ b/crates/core/src/transfer.rs @@ -0,0 +1,684 @@ +//! Shared planning and execution primitives for bulk transfers. +//! +//! Commands expand their own storage-specific inputs into [`TransferCandidate`] values, then use +//! this module for consistent selection, ordering, retry, rate, concurrency, cancellation, and +//! aggregate summary behavior. + +use std::collections::BTreeMap; +use std::future::Future; +use std::sync::Arc; +use std::time::Duration; + +use futures::stream::{FuturesUnordered, StreamExt as _}; +use glob::Pattern; +use jiff::Timestamp; +use serde::Serialize; +use tokio::sync::Mutex; +use tokio::time::{Instant, sleep_until}; + +use crate::alias::RetryConfig; +use crate::error::{Error, Result}; +use crate::retry::{is_retryable_error, retry_with_backoff}; + +/// A storage-independent item that can be selected and executed as part of a transfer. +#[derive(Debug, Clone)] +pub struct TransferCandidate { + /// Command-specific operation data. + pub payload: T, + /// Display-safe source identity. + pub source: String, + /// Display-safe destination identity. + pub target: String, + /// Source-relative path used by include and exclude rules. + pub relative_path: String, + /// UTC modification timestamp when the source provides one. + pub modified: Option, + /// Expected transfer size, used for aggregate rate pacing. + pub size_bytes: Option, +} + +/// Shared path and UTC timestamp selection rules. +#[derive(Debug, Clone, Default)] +pub struct TransferSelection { + include: Vec, + exclude: Vec, + newer_than: Option, + older_than: Option, + rewind: Option, +} + +impl TransferSelection { + /// Compile a deterministic selection policy. + /// + /// Include patterns restrict the candidate set when at least one is present. Exclude patterns + /// are applied afterwards and always win, independent of command-line flag order. `newer_than` + /// and `older_than` use strict comparisons; `rewind` includes the boundary itself. + pub fn new( + include: &[String], + exclude: &[String], + newer_than: Option, + older_than: Option, + rewind: Option, + ) -> Result { + let include = compile_patterns("include", include)?; + let exclude = compile_patterns("exclude", exclude)?; + + Ok(Self { + include, + exclude, + newer_than, + older_than, + rewind, + }) + } + + /// Return whether a candidate satisfies every configured rule. + pub fn matches(&self, candidate: &TransferCandidate) -> bool { + let path_matches = (self.include.is_empty() + || self + .include + .iter() + .any(|pattern| pattern.matches(&candidate.relative_path))) + && !self + .exclude + .iter() + .any(|pattern| pattern.matches(&candidate.relative_path)); + + if !path_matches { + return false; + } + + let has_time_filter = + self.newer_than.is_some() || self.older_than.is_some() || self.rewind.is_some(); + let Some(modified) = candidate.modified else { + return !has_time_filter; + }; + + self.newer_than.is_none_or(|cutoff| modified > cutoff) + && self.older_than.is_none_or(|cutoff| modified < cutoff) + && self.rewind.is_none_or(|cutoff| modified <= cutoff) + } +} + +fn compile_patterns(kind: &str, values: &[String]) -> Result> { + values + .iter() + .map(|value| { + Pattern::new(value).map_err(|error| { + Error::InvalidPath(format!("Invalid {kind} pattern '{value}': {error}")) + }) + }) + .collect() +} + +/// Deterministic aggregate counters for a transfer plan and its execution. +#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize)] +pub struct TransferSummary { + pub planned: usize, + pub skipped: usize, + pub successful: usize, + pub failed: usize, + pub cancelled: usize, + pub transferred_bytes: u64, +} + +/// Selected candidates in deterministic execution and output order. +#[derive(Debug)] +pub struct TransferPlan { + pub items: Vec>, + pub summary: TransferSummary, +} + +impl TransferPlan { + /// Filter and sort candidates without performing any transfer. + pub fn build(candidates: Vec>, selection: &TransferSelection) -> Self { + let candidate_count = candidates.len(); + let mut items: Vec<_> = candidates + .into_iter() + .filter(|candidate| selection.matches(candidate)) + .collect(); + items.sort_by(|left, right| { + left.relative_path + .cmp(&right.relative_path) + .then_with(|| left.source.cmp(&right.source)) + .then_with(|| left.target.cmp(&right.target)) + }); + + Self { + summary: TransferSummary { + planned: items.len(), + skipped: candidate_count.saturating_sub(items.len()), + ..TransferSummary::default() + }, + items, + } + } +} + +/// Controls applied across the entire transfer command. +#[derive(Debug, Clone)] +pub struct TransferControls { + pub concurrency: usize, + pub bytes_per_second: Option, + pub retry: RetryConfig, + pub continue_on_error: bool, +} + +impl TransferControls { + /// Validate global limits before any work starts. + pub fn validate(&self) -> Result<()> { + if !(1..=256).contains(&self.concurrency) { + return Err(Error::Config( + "Transfer concurrency must be between 1 and 256".to_string(), + )); + } + if self.bytes_per_second == Some(0) { + return Err(Error::Config( + "Transfer rate limit must be greater than zero".to_string(), + )); + } + if self.retry.max_attempts == 0 { + return Err(Error::Config( + "Transfer retry attempts must be greater than zero".to_string(), + )); + } + if self.retry.initial_backoff_ms > self.retry.max_backoff_ms { + return Err(Error::Config( + "Transfer retry initial backoff must not exceed its maximum backoff".to_string(), + )); + } + Ok(()) + } +} + +impl Default for TransferControls { + fn default() -> Self { + Self { + concurrency: 4, + bytes_per_second: None, + retry: RetryConfig::default(), + continue_on_error: false, + } + } +} + +/// Result state for one candidate. +#[derive(Debug)] +pub enum TransferOutcomeState { + Success { bytes_transferred: u64 }, + Failed { error: Error }, + Cancelled, +} + +/// Deterministically ordered result for one candidate. +#[derive(Debug)] +pub struct TransferOutcome { + pub item: TransferCandidate, + pub state: TransferOutcomeState, +} + +/// Aggregate execution result and per-candidate outcomes. +#[derive(Debug)] +pub struct TransferReport { + pub summary: TransferSummary, + pub outcomes: Vec>, +} + +impl TransferReport { + /// Return the first failure in deterministic plan order. + pub fn first_failure(&self) -> Option<&Error> { + self.outcomes + .iter() + .find_map(|outcome| match &outcome.state { + TransferOutcomeState::Failed { error } => Some(error), + TransferOutcomeState::Success { .. } | TransferOutcomeState::Cancelled => None, + }) + } +} + +/// Executes a transfer plan with shared controls. +#[derive(Debug, Clone)] +pub struct TransferExecutor { + controls: TransferControls, +} + +impl TransferExecutor { + pub fn new(controls: TransferControls) -> Result { + controls.validate()?; + Ok(Self { controls }) + } + + /// Execute candidates while bounding all in-flight work across the command. + pub async fn execute(&self, plan: TransferPlan, operation: F) -> TransferReport + where + T: Clone + Send + Sync + 'static, + F: Fn(TransferCandidate) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, + { + let mut summary = plan.summary; + let items = plan.items; + let operation = Arc::new(operation); + let rate_gate = Arc::new(RateGate::new(self.controls.bytes_per_second)); + let mut in_flight = FuturesUnordered::new(); + let mut completed = BTreeMap::new(); + let mut next_index = 0usize; + let mut stop_scheduling = false; + + while next_index < items.len() && in_flight.len() < self.controls.concurrency { + in_flight.push(execute_candidate( + next_index, + items[next_index].clone(), + Arc::clone(&operation), + Arc::clone(&rate_gate), + self.controls.retry.clone(), + )); + next_index += 1; + } + + while let Some((index, result)) = in_flight.next().await { + let failed = result.is_err(); + completed.insert(index, result); + + if failed && !self.controls.continue_on_error { + // Work that has already started is allowed to settle so the report never marks a + // completed side effect as cancelled. No additional candidates are scheduled. + stop_scheduling = true; + } + + if !stop_scheduling && next_index < items.len() { + in_flight.push(execute_candidate( + next_index, + items[next_index].clone(), + Arc::clone(&operation), + Arc::clone(&rate_gate), + self.controls.retry.clone(), + )); + next_index += 1; + } + } + drop(in_flight); + + let mut outcomes = Vec::with_capacity(items.len()); + for (index, item) in items.into_iter().enumerate() { + let state = match completed.remove(&index) { + Some(Ok(bytes_transferred)) => { + summary.successful += 1; + summary.transferred_bytes = + summary.transferred_bytes.saturating_add(bytes_transferred); + TransferOutcomeState::Success { bytes_transferred } + } + Some(Err(error)) => { + summary.failed += 1; + TransferOutcomeState::Failed { error } + } + None => { + summary.cancelled += 1; + TransferOutcomeState::Cancelled + } + }; + outcomes.push(TransferOutcome { item, state }); + } + + TransferReport { summary, outcomes } + } +} + +async fn execute_candidate( + index: usize, + item: TransferCandidate, + operation: Arc, + rate_gate: Arc, + retry: RetryConfig, +) -> (usize, Result) +where + T: Clone + Send + Sync + 'static, + F: Fn(TransferCandidate) -> Fut + Send + Sync + 'static, + Fut: Future> + Send + 'static, +{ + let result = retry_with_backoff( + &retry, + || { + let item = item.clone(); + let operation = Arc::clone(&operation); + let rate_gate = Arc::clone(&rate_gate); + async move { + rate_gate.wait(item.size_bytes.unwrap_or_default()).await; + operation(item).await + } + }, + is_retryable_error, + ) + .await; + (index, result) +} + +#[derive(Debug)] +struct RateGate { + bytes_per_second: Option, + next_slot: Mutex, +} + +impl RateGate { + fn new(bytes_per_second: Option) -> Self { + Self { + bytes_per_second, + next_slot: Mutex::new(Instant::now()), + } + } + + async fn wait(&self, bytes: u64) { + let Some(bytes_per_second) = self.bytes_per_second else { + return; + }; + if bytes == 0 || bytes_per_second == 0 { + return; + } + + let wait_until = { + let mut next_slot = self.next_slot.lock().await; + let now = Instant::now(); + let wait_until = (*next_slot).max(now); + let delay = rate_duration(bytes, bytes_per_second); + *next_slot = wait_until.checked_add(delay).unwrap_or(wait_until); + wait_until + }; + + if wait_until > Instant::now() { + sleep_until(wait_until).await; + } + } +} + +fn rate_duration(bytes: u64, bytes_per_second: u64) -> Duration { + let seconds = bytes / bytes_per_second; + let remaining_bytes = bytes % bytes_per_second; + let nanos = + (u128::from(remaining_bytes) * 1_000_000_000u128).div_ceil(u128::from(bytes_per_second)); + Duration::from_secs(seconds).saturating_add(Duration::from_nanos(nanos as u64)) +} + +#[cfg(test)] +mod tests { + use std::sync::atomic::{AtomicUsize, Ordering}; + + use super::*; + + fn candidate( + name: &str, + modified: Option, + size_bytes: u64, + ) -> TransferCandidate<()> { + TransferCandidate { + payload: (), + source: format!("source/{name}"), + target: format!("target/{name}"), + relative_path: name.to_string(), + modified, + size_bytes: Some(size_bytes), + } + } + + #[test] + fn selection_applies_includes_before_excludes_and_sorts_deterministically() { + let selection = TransferSelection::new( + &["reports/*.csv".to_string()], + &["**/private-*".to_string()], + None, + None, + None, + ) + .expect("valid selection"); + let plan = TransferPlan::build( + vec![ + candidate("reports/z.csv", None, 1), + candidate("reports/private-a.csv", None, 1), + candidate("notes.txt", None, 1), + candidate("reports/a.csv", None, 1), + ], + &selection, + ); + + assert_eq!( + plan.items + .iter() + .map(|item| item.relative_path.as_str()) + .collect::>(), + ["reports/a.csv", "reports/z.csv"] + ); + assert_eq!(plan.summary.planned, 2); + assert_eq!(plan.summary.skipped, 2); + } + + #[test] + fn time_selection_has_explicit_utc_boundary_semantics() { + let cutoff: Timestamp = "2026-07-21T12:00:00Z".parse().expect("valid UTC cutoff"); + let at_boundary = candidate("boundary", Some(cutoff), 1); + + let newer = TransferSelection::new(&[], &[], Some(cutoff), None, None) + .expect("valid newer selection"); + let older = TransferSelection::new(&[], &[], None, Some(cutoff), None) + .expect("valid older selection"); + let rewind = TransferSelection::new(&[], &[], None, None, Some(cutoff)) + .expect("valid rewind selection"); + + assert!(!newer.matches(&at_boundary)); + assert!(!older.matches(&at_boundary)); + assert!(rewind.matches(&at_boundary)); + assert!(!rewind.matches(&candidate("missing", None, 1))); + } + + #[tokio::test] + async fn executor_enforces_one_global_concurrency_limit() { + let controls = TransferControls { + concurrency: 2, + ..TransferControls::default() + }; + let executor = TransferExecutor::new(controls).expect("valid controls"); + let plan = TransferPlan::build( + (0..6) + .map(|index| candidate(&format!("{index}"), None, 1)) + .collect(), + &TransferSelection::default(), + ); + let active = Arc::new(AtomicUsize::new(0)); + let maximum = Arc::new(AtomicUsize::new(0)); + + let report = executor + .execute(plan, { + let active = Arc::clone(&active); + let maximum = Arc::clone(&maximum); + move |_| { + let active = Arc::clone(&active); + let maximum = Arc::clone(&maximum); + async move { + let current = active.fetch_add(1, Ordering::SeqCst) + 1; + maximum.fetch_max(current, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(10)).await; + active.fetch_sub(1, Ordering::SeqCst); + Ok(1) + } + } + }) + .await; + + assert_eq!(maximum.load(Ordering::SeqCst), 2); + assert_eq!(report.summary.successful, 6); + } + + #[tokio::test] + async fn executor_applies_rate_limit_across_all_workers() { + let controls = TransferControls { + concurrency: 3, + bytes_per_second: Some(1_000), + ..TransferControls::default() + }; + let executor = TransferExecutor::new(controls).expect("valid controls"); + let plan = TransferPlan::build( + vec![ + candidate("a", None, 10), + candidate("b", None, 10), + candidate("c", None, 10), + ], + &TransferSelection::default(), + ); + let starts = Arc::new(Mutex::new(Vec::new())); + + let report = executor + .execute(plan, { + let starts = Arc::clone(&starts); + move |_| { + let starts = Arc::clone(&starts); + async move { + starts.lock().await.push(Instant::now()); + Ok(10) + } + } + }) + .await; + + let starts = starts.lock().await; + let first = starts.iter().min().copied().expect("first start"); + let last = starts.iter().max().copied().expect("last start"); + assert!(last.duration_since(first) >= Duration::from_millis(15)); + assert_eq!(report.summary.transferred_bytes, 30); + } + + #[tokio::test] + async fn executor_retries_only_classified_transient_failures() { + let controls = TransferControls { + concurrency: 1, + retry: RetryConfig { + max_attempts: 3, + initial_backoff_ms: 1, + max_backoff_ms: 1, + }, + continue_on_error: true, + ..TransferControls::default() + }; + let executor = TransferExecutor::new(controls).expect("valid controls"); + let transient_calls = Arc::new(AtomicUsize::new(0)); + let permanent_calls = Arc::new(AtomicUsize::new(0)); + let plan = TransferPlan::build( + vec![ + candidate("transient", None, 1), + candidate("permanent", None, 1), + ], + &TransferSelection::default(), + ); + + let report = executor + .execute(plan, { + let transient_calls = Arc::clone(&transient_calls); + let permanent_calls = Arc::clone(&permanent_calls); + move |item| { + let transient_calls = Arc::clone(&transient_calls); + let permanent_calls = Arc::clone(&permanent_calls); + async move { + if item.relative_path == "transient" { + let attempt = transient_calls.fetch_add(1, Ordering::SeqCst) + 1; + if attempt < 3 { + return Err(Error::Network("503 Service Unavailable".to_string())); + } + return Ok(1); + } + + permanent_calls.fetch_add(1, Ordering::SeqCst); + Err(Error::Auth("denied".to_string())) + } + } + }) + .await; + + assert_eq!(transient_calls.load(Ordering::SeqCst), 3); + assert_eq!(permanent_calls.load(Ordering::SeqCst), 1); + assert_eq!(report.summary.successful, 1); + assert_eq!(report.summary.failed, 1); + } + + #[tokio::test] + async fn executor_cancels_remaining_work_after_first_failure() { + let controls = TransferControls { + concurrency: 1, + retry: RetryConfig { + max_attempts: 1, + ..RetryConfig::default() + }, + continue_on_error: false, + ..TransferControls::default() + }; + let executor = TransferExecutor::new(controls).expect("valid controls"); + let calls = Arc::new(AtomicUsize::new(0)); + let plan = TransferPlan::build( + vec![ + candidate("a", None, 1), + candidate("b", None, 1), + candidate("c", None, 1), + ], + &TransferSelection::default(), + ); + + let report = executor + .execute(plan, { + let calls = Arc::clone(&calls); + move |_| { + let calls = Arc::clone(&calls); + async move { + calls.fetch_add(1, Ordering::SeqCst); + Err(Error::Auth("stop".to_string())) + } + } + }) + .await; + + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(report.summary.failed, 1); + assert_eq!(report.summary.cancelled, 2); + } + + #[tokio::test] + async fn executor_drains_started_work_but_does_not_schedule_more_after_failure() { + let controls = TransferControls { + concurrency: 2, + retry: RetryConfig { + max_attempts: 1, + ..RetryConfig::default() + }, + continue_on_error: false, + ..TransferControls::default() + }; + let executor = TransferExecutor::new(controls).expect("valid controls"); + let calls = Arc::new(AtomicUsize::new(0)); + let plan = TransferPlan::build( + vec![ + candidate("a", None, 1), + candidate("b", None, 1), + candidate("c", None, 1), + ], + &TransferSelection::default(), + ); + + let report = executor + .execute(plan, { + let calls = Arc::clone(&calls); + move |item| { + let calls = Arc::clone(&calls); + async move { + calls.fetch_add(1, Ordering::SeqCst); + if item.relative_path == "a" { + tokio::time::sleep(Duration::from_millis(5)).await; + Err(Error::Auth("stop".to_string())) + } else { + tokio::time::sleep(Duration::from_millis(10)).await; + Ok(1) + } + } + } + }) + .await; + + assert_eq!(calls.load(Ordering::SeqCst), 2); + assert_eq!(report.summary.successful, 1); + assert_eq!(report.summary.failed, 1); + assert_eq!(report.summary.cancelled, 1); + } +} diff --git a/docs/reference/rc/cp.md b/docs/reference/rc/cp.md index 375a9896..c8cf2bd6 100644 --- a/docs/reference/rc/cp.md +++ b/docs/reference/rc/cp.md @@ -7,15 +7,15 @@ ## Syntax ```bash -rc [GLOBAL OPTIONS] cp [OPTIONS] +rc [GLOBAL OPTIONS] cp [OPTIONS] ... ``` ## Parameters | Parameter | Description | | --- | --- | -| `SOURCE` | Local file, local directory, or remote object/prefix path. This version accepts one source and one target. | -| `TARGET` | Local or remote destination path. | +| `SOURCE` | One or more local files/directories or remote objects/prefixes. | +| `TARGET` | Local directory or remote prefix for multiple sources; local or remote path for one source. | | `-r, --recursive` | Recursively copy a directory or prefix. | | `--overwrite` | Overwrite destination data where supported. | | `--dry-run` | Show planned copies without copying data. | @@ -24,6 +24,19 @@ rc [GLOBAL OPTIONS] cp [OPTIONS] | `--storage-class` | Set destination storage class for uploads where supported. | | `--enc-s3 ` | Apply `SSE-S3` to the named remote destination write. | | `--enc-kms =` | Apply `SSE-KMS` to the named remote destination write. | +| `--include ` | Include source-relative paths matching the glob. Repeatable. | +| `--exclude ` | Exclude matching paths after includes are evaluated. Repeatable. | +| `--newer-than ` | Select sources strictly newer than an age such as `1h` or `7d`. | +| `--older-than ` | Select sources strictly older than an age such as `1h` or `7d`. | +| `--rewind