diff --git a/crates/cli/src/commands/cp.rs b/crates/cli/src/commands/cp.rs index 4494e01c..070bc66a 100644 --- a/crates/cli/src/commands/cp.rs +++ b/crates/cli/src/commands/cp.rs @@ -7,10 +7,10 @@ use jiff::Timestamp; use rc_core::alias::RetryConfig; use rc_core::{ AliasManager, Error, MetadataDirective, MultipartCopyCancellation, MultipartCopyOptions, - ObjectEncryptionRequest, ObjectInfo, ObjectStore as _, ObjectWriteOptions, ParsedPath, - RemotePath, SseCustomerKey, TransferCancellation, TransferCandidate, TransferControls, - TransferCopyOptions, TransferExecutor, TransferOutcomeState, TransferPlan, TransferReadOptions, - TransferSelection, parse_path, + ObjectAttributes, ObjectEncryptionRequest, ObjectInfo, ObjectStore as _, ObjectWriteOptions, + ParsedPath, RemotePath, SseCustomerKey, TransferCancellation, TransferCandidate, + TransferControls, TransferCopyOptions, TransferExecutor, TransferOutcomeState, TransferPlan, + TransferReadOptions, TransferSelection, parse_path, }; use rc_s3::S3Client; use serde::Serialize; @@ -675,12 +675,21 @@ fn validate_fidelity_directions( "Destination transfer policies require a remote destination".to_string(), )); } + let same_alias_remote_copy = target_remote + && sources.iter().any(|source| { + matches!( + source, + ParsedPath::Remote(source) if target.as_remote().is_some_and(|target| source.alias == target.alias) + ) + }); if any_remote && target_remote { let copy_options = transfer_copy_options(args, None, None)?; - if matches!( - copy_options.metadata_directive, - Some(MetadataDirective::Replace) - ) { + if same_alias_remote_copy + && matches!( + copy_options.metadata_directive, + Some(MetadataDirective::Replace) + ) + { return Err(Error::UnsupportedFeature( "RustFS beta.10 does not preserve complete metadata REPLACE semantics; tracked by rustfs/backlog#1463" .to_string(), @@ -1091,6 +1100,9 @@ fn validate_storage_class_plan( })?; let multipart = match &item.payload { CpOperation::LocalToRemote { .. } => size > MULTIPART_THRESHOLD, + CpOperation::RemoteToRemote { source, target, .. } if source.alias != target.alias => { + size > MULTIPART_THRESHOLD + } CpOperation::RemoteToRemote { .. } => rc_core::requires_multipart_copy(size), CpOperation::RemoteToLocal { .. } => false, }; @@ -1128,10 +1140,13 @@ async fn execute_planned_operation( source_info, encryption, } => { + let source_client = planned_client(clients, &source.alias)?; + let target_client = planned_client(clients, &target.alias)?; let progress_key = (item.source.clone(), item.target.clone()); copy_progress.reset(&progress_key); let result = perform_planned_remote_copy( - client, + source_client, + target_client, source, target, source_info, @@ -1226,7 +1241,8 @@ struct PlannedRemoteCopyResult { #[allow(clippy::too_many_arguments)] async fn perform_planned_remote_copy( - client: &S3Client, + source_client: &S3Client, + target_client: &S3Client, source: &RemotePath, target: &RemotePath, source_info: &ObjectInfo, @@ -1236,9 +1252,17 @@ async fn perform_planned_remote_copy( args: &CpArgs, ) -> rc_core::Result { if source.alias != target.alias { - return Err(Error::UnsupportedFeature( - "Cross-alias S3-to-S3 copy is not supported".to_string(), - )); + return perform_cross_alias_remote_copy( + source_client, + target_client, + source, + target, + source_info, + encryption, + on_progress, + args, + ) + .await; } if args.source_customer_key.is_some() || args.destination_customer_key.is_some() { return Err(Error::UnsupportedFeature( @@ -1256,21 +1280,15 @@ async fn perform_planned_remote_copy( "RustFS beta.10 does not persist storage class for multipart copies".to_string(), )); } - let current = client.head_object(source).await?; - if current.size_bytes != source_info.size_bytes - || source_info - .etag - .as_ref() - .zip(current.etag.as_ref()) - .is_some_and(|(planned, current)| planned != current) - { + let current = source_client.head_object(source).await?; + if !source_identity_matches(source_info, ¤t) { return Err(Error::Conflict(format!( "Source changed after copy planning: {source}" ))); } let options = multipart_options_from_source(¤t)?; let transfer = transfer_copy_options(args, current.version_id.clone(), encryption)?; - let copied = client + let copied = source_client .multipart_copy_with_transfer_options( source, target, @@ -1289,7 +1307,7 @@ async fn perform_planned_remote_copy( }); } let options = transfer_copy_options(args, source_info.version_id.clone(), encryption)?; - let copied = client + let copied = source_client .copy_object_with_transfer_options(source, target, &options) .await?; let bytes_copied = copied @@ -1309,6 +1327,152 @@ async fn perform_planned_remote_copy( }) } +#[allow(clippy::too_many_arguments)] +async fn perform_cross_alias_remote_copy( + source_client: &S3Client, + target_client: &S3Client, + source: &RemotePath, + target: &RemotePath, + source_info: &ObjectInfo, + encryption: Option<&ObjectEncryptionRequest>, + on_progress: &(dyn Fn(u64) + Send + Sync), + args: &CpArgs, +) -> rc_core::Result { + // Server-side CopyObject cannot target a different alias/endpoint. Stream + // through a bounded temporary file so the destination write is a normal + // upload with the destination alias credentials. + if args.source_customer_key.is_some() || args.destination_customer_key.is_some() { + return Err(Error::UnsupportedFeature( + "RustFS beta.10 server-side SSE-C copy is not compatibility-proven; tracked by rustfs/backlog#1467" + .to_string(), + )); + } + let planned_size = source_info + .size_bytes + .and_then(|size| u64::try_from(size).ok()) + .ok_or_else(|| Error::InvalidPath(format!("Source size is unavailable: {source}")))?; + if args.storage_class.is_some() && planned_size > MULTIPART_THRESHOLD { + return Err(Error::UnsupportedFeature( + "RustFS beta.10 does not persist storage class for multipart uploads".to_string(), + )); + } + + let current = source_client.head_object(source).await?; + if !source_identity_matches(source_info, ¤t) { + return Err(Error::Conflict(format!( + "Source changed after copy planning: {source}" + ))); + } + + let staging = tempfile::Builder::new() + .prefix("rc-cp-cross-alias-") + .suffix(".part") + .tempfile()? + .into_temp_path(); + async { + let downloaded = source_client + .download_object_to_path_with_transfer_options( + source, + &staging, + &TransferReadOptions { + version_id: current.version_id.clone(), + customer_key: args.source_customer_key.clone(), + ..TransferReadOptions::default() + }, + |copied, _total| on_progress(copied), + ) + .await?; + if downloaded != planned_size { + return Err(Error::Conflict(format!( + "Source changed after copy planning: {source}" + ))); + } + let after = source_client + .head_object_with_transfer_options( + source, + &TransferReadOptions { + version_id: current.version_id.clone(), + customer_key: args.source_customer_key.clone(), + ..TransferReadOptions::default() + }, + ) + .await?; + if !source_identity_matches(¤t, &after) { + return Err(Error::Conflict(format!( + "Source changed after copy planning: {source}" + ))); + } + let options = piped_copy_write_options(args, ¤t, encryption)?; + let object = target_client + .put_object_from_path_with_options(target, &staging, &options, |copied| { + on_progress(copied) + }) + .await?; + let bytes_copied = object + .size_bytes + .and_then(|size| u64::try_from(size).ok()) + .unwrap_or(downloaded); + on_progress(bytes_copied); + Ok(PlannedRemoteCopyResult { + bytes_copied, + source_version_id: current.version_id.clone(), + destination_version_id: object.version_id.clone(), + upload_id: None, + object, + }) + } + .await +} + +fn source_identity_matches(planned: &ObjectInfo, current: &ObjectInfo) -> bool { + if planned.size_bytes != current.size_bytes { + return false; + } + match (&planned.etag, ¤t.etag) { + (Some(planned), Some(current)) => planned == current, + // Without an ETag an unversioned object has no stable read identity; + // only an identical explicit version can prove that it is unchanged. + (None, None) => { + planned.version_id.is_some() + && planned.version_id.as_ref() == current.version_id.as_ref() + } + _ => false, + } +} + +fn piped_copy_write_options( + args: &CpArgs, + source: &ObjectInfo, + encryption: Option<&ObjectEncryptionRequest>, +) -> rc_core::Result { + let mut options = object_write_options( + &args.fidelity, + args.content_type.as_deref(), + encryption, + args.destination_customer_key.as_ref(), + args.storage_class.clone(), + )?; + if matches!( + requested_metadata_directive(args), + Some(MetadataDirective::Replace) + ) { + return Ok(options); + } + let mut attributes = options.attributes.take().unwrap_or_default(); + if attributes.content_type.is_none() { + attributes.content_type = source.content_type.clone(); + } + if attributes.user_metadata.is_empty() + && let Some(metadata) = &source.metadata + { + attributes.user_metadata.clone_from(metadata); + } + if attributes != ObjectAttributes::default() { + options.attributes = Some(attributes); + } + Ok(options) +} + #[cfg(test)] fn requires_multipart_copy(planned_size: Option) -> bool { planned_size.is_some_and(rc_core::requires_multipart_copy) @@ -1340,10 +1504,28 @@ fn operation_alias(operation: &CpOperation) -> &str { } } +fn operation_client_aliases(operation: &CpOperation) -> Vec<&str> { + match operation { + CpOperation::LocalToRemote { target, .. } => vec![target.alias.as_str()], + CpOperation::RemoteToLocal { source, .. } => vec![source.alias.as_str()], + CpOperation::RemoteToRemote { source, target, .. } => { + if source.alias == target.alias { + vec![source.alias.as_str()] + } else { + vec![source.alias.as_str(), target.alias.as_str()] + } + } + } +} + fn planned_client_aliases(items: &[TransferCandidate]) -> BTreeSet { items .iter() - .map(|item| operation_alias(&item.payload).to_string()) + .flat_map(|item| { + operation_client_aliases(&item.payload) + .into_iter() + .map(ToOwned::to_owned) + }) .collect() } @@ -1862,18 +2044,13 @@ async fn build_remote_candidates( } let listing_source = recursive_listing_source(source); - if let ParsedPath::Remote(target) = target { - if source.alias != target.alias { - return Err(Error::UnsupportedFeature( - "Cross-alias S3-to-S3 copy is not supported".to_string(), - )); - } - if remote_copy_scopes_overlap(&listing_source, target) { - return Err(Error::Conflict(format!( - "Recursive source '{}' overlaps destination '{}'", - listing_source, target - ))); - } + if let ParsedPath::Remote(target) = target + && remote_copy_scopes_overlap(&listing_source, target) + { + return Err(Error::Conflict(format!( + "Recursive source '{}' overlaps destination '{}'", + listing_source, target + ))); } let source_root = recursive_source_root(&listing_source, multiple_sources); @@ -2004,11 +2181,6 @@ async fn build_remote_candidates( }); } 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 { @@ -2800,7 +2972,6 @@ async fn copy_s3_to_s3_prepared( ); } - // For S3-to-S3, we need to handle same or different aliases let alias_manager = match AliasManager::new() { Ok(am) => am, Err(e) => { @@ -2809,16 +2980,9 @@ async fn copy_s3_to_s3_prepared( } }; - // For now, only support same-alias copies (server-side copy) - if src.alias != dst.alias { - return formatter.fail_with_suggestion( - ExitCode::UnsupportedFeature, - "Cross-alias S3-to-S3 copy not yet supported. Use download + upload.", - "Copy via a local path or split the operation into download and upload steps.", - ); - } - - let alias = match alias_manager.get(&src.alias) { + // Same-alias copies use server-side CopyObject. Different aliases download + // through a temporary file and upload with the destination credentials. + let source_alias = match alias_manager.get(&src.alias) { Ok(a) => a, Err(_) => { return formatter.fail_with_suggestion( @@ -2828,7 +2992,7 @@ async fn copy_s3_to_s3_prepared( ); } }; - let client = match S3Client::new(alias).await { + let source_client = match S3Client::new(source_alias).await { Ok(c) => c, Err(e) => { return formatter.fail( @@ -2837,6 +3001,31 @@ async fn copy_s3_to_s3_prepared( ); } }; + let target_client; + let target_client_ref = if src.alias == dst.alias { + &source_client + } else { + let destination_alias = match alias_manager.get(&dst.alias) { + Ok(a) => a, + Err(_) => { + return formatter.fail_with_suggestion( + ExitCode::NotFound, + &format!("Alias '{}' not found", dst.alias), + "Run `rc alias list` to inspect configured aliases or add one with `rc alias set ...`.", + ); + } + }; + target_client = match S3Client::new(destination_alias).await { + Ok(c) => c, + Err(e) => { + return formatter.fail( + ExitCode::NetworkError, + &format!("Failed to create destination S3 client: {e}"), + ); + } + }; + &target_client + }; let src_display = format!("{}/{}/{}", src.alias, src.bucket, src.key); let dst_display = format!("{}/{}/{}", dst.alias, dst.bucket, dst.key); @@ -2851,7 +3040,7 @@ async fn copy_s3_to_s3_prepared( return ExitCode::Success; } - let source_info = match client.head_object(src).await { + let source_info = match source_client.head_object(src).await { Ok(info) => info, Err(Error::NotFound(_)) => { return formatter.fail_with_suggestion( @@ -2870,7 +3059,15 @@ async fn copy_s3_to_s3_prepared( let source_size = source_info .size_bytes .and_then(|size| u64::try_from(size).ok()); - if args.storage_class.is_some() && source_size.is_none_or(rc_core::requires_multipart_copy) { + if args.storage_class.is_some() + && source_size.is_none_or(|size| { + if src.alias == dst.alias { + rc_core::requires_multipart_copy(size) + } else { + size > MULTIPART_THRESHOLD + } + }) + { return formatter.fail( ExitCode::UnsupportedFeature, "RustFS beta.10 does not persist storage class for multipart or unknown-size copies", @@ -2900,7 +3097,8 @@ async fn copy_s3_to_s3_prepared( }); let ignore_progress = |_: u64| {}; let copy = perform_planned_remote_copy( - &client, + &source_client, + target_client_ref, src, dst, &source_info, @@ -3404,6 +3602,54 @@ mod tests { ); } + #[test] + fn planned_client_aliases_include_both_sides_of_a_cross_alias_copy() { + let candidate = TransferCandidate { + payload: CpOperation::RemoteToRemote { + source: RemotePath::new("alpha", "source", "file.txt"), + target: RemotePath::new("beta", "target", "file.txt"), + source_info: Box::new(ObjectInfo::file("file.txt", 4)), + encryption: None, + }, + source: "alpha/source/file.txt".to_string(), + target: "beta/target/file.txt".to_string(), + relative_path: "file.txt".to_string(), + modified: None, + size_bytes: Some(4), + }; + + assert_eq!( + planned_client_aliases(&[candidate]) + .into_iter() + .collect::>(), + ["alpha", "beta"] + ); + } + + #[test] + fn planned_client_aliases_keep_same_alias_remote_copy_on_one_alias() { + let candidate = TransferCandidate { + payload: CpOperation::RemoteToRemote { + source: RemotePath::new("shared", "source", "file.txt"), + target: RemotePath::new("shared", "target", "file.txt"), + source_info: Box::new(ObjectInfo::file("file.txt", 4)), + encryption: None, + }, + source: "shared/source/file.txt".to_string(), + target: "shared/target/file.txt".to_string(), + relative_path: "file.txt".to_string(), + modified: None, + size_bytes: Some(4), + }; + + assert_eq!( + planned_client_aliases(&[candidate]) + .into_iter() + .collect::>(), + ["shared"] + ); + } + #[tokio::test] async fn planning_client_reuses_one_connection_pool_per_alias() { let (alias_manager, _temp_dir) = temp_alias_manager(); @@ -3745,6 +3991,20 @@ mod tests { Err(Error::UnsupportedFeature(_)) )); + let cross_alias_source = + ParsedPath::Remote(RemotePath::new("source", "source", "report.json")); + let cross_alias_target = + ParsedPath::Remote(RemotePath::new("destination", "target", "report.json")); + assert!( + validate_fidelity_directions( + &replace, + std::slice::from_ref(&cross_alias_source), + &cross_alias_target, + ) + .is_ok(), + "metadata REPLACE is implemented by the cross-alias upload path" + ); + let mut tags = CpArgs::single("test/source/report.json", "test/target/report.json"); tags.tagging_directive = Some(TaggingDirectiveArg::Replace); tags.fidelity.tags = vec!["env=prod".to_string()]; @@ -3761,6 +4021,23 @@ mod tests { )); } + #[test] + fn source_identity_validation_detects_same_size_etag_changes() { + let mut planned = ObjectInfo::file("report.json", 4); + planned.etag = Some("planned".to_string()); + let mut current = planned.clone(); + assert!(source_identity_matches(&planned, ¤t)); + + current.etag = Some("changed".to_string()); + assert!(!source_identity_matches(&planned, ¤t)); + + current.etag = None; + assert!(!source_identity_matches(&planned, ¤t)); + + planned.etag = None; + assert!(!source_identity_matches(&planned, ¤t)); + } + #[test] fn preserve_builds_explicit_metadata_copy_without_replacement_payload() { let mut args = CpArgs::single("test/source/report.json", "test/target/report.json"); @@ -3809,6 +4086,90 @@ mod tests { )); } + #[test] + fn storage_class_plan_rejects_cross_alias_multipart_uploads() { + let plan = TransferPlan::build( + vec![TransferCandidate { + payload: CpOperation::RemoteToRemote { + source: RemotePath::new("alpha", "source", "medium.bin"), + target: RemotePath::new("beta", "target", "medium.bin"), + source_info: Box::new(ObjectInfo::file( + "medium.bin", + (MULTIPART_THRESHOLD + 1) as i64, + )), + encryption: None, + }, + source: "alpha/source/medium.bin".to_string(), + target: "beta/target/medium.bin".to_string(), + relative_path: "medium.bin".to_string(), + modified: None, + size_bytes: Some(MULTIPART_THRESHOLD + 1), + }], + &TransferSelection::default(), + ); + + assert!(matches!( + validate_storage_class_plan(&plan, Some("STANDARD")), + Err(Error::UnsupportedFeature(_)) + )); + } + + #[test] + fn piped_copy_preserves_source_content_type_unless_replaced() { + let mut source = ObjectInfo::file("file.txt", 4); + source.content_type = Some("text/plain".to_string()); + let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt"); + + let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata"); + assert_eq!( + copied + .attributes + .as_ref() + .and_then(|value| value.content_type.as_deref()), + Some("text/plain") + ); + + let mut replace = args; + replace.metadata_directive = Some(MetadataDirectiveArg::Replace); + let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata"); + assert!( + replaced + .attributes + .as_ref() + .is_none_or(|value| value.content_type.is_none()) + ); + } + + #[test] + fn piped_copy_preserves_source_user_metadata_unless_replaced() { + let mut source = ObjectInfo::file("file.txt", 4); + source.metadata = Some(HashMap::from([( + "owner".to_string(), + "storage".to_string(), + )])); + let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt"); + + let copied = piped_copy_write_options(&args, &source, None).expect("copy metadata"); + assert_eq!( + copied + .attributes + .as_ref() + .and_then(|value| value.user_metadata.get("owner")) + .map(String::as_str), + Some("storage") + ); + + let mut replace = args; + replace.metadata_directive = Some(MetadataDirectiveArg::Replace); + let replaced = piped_copy_write_options(&replace, &source, None).expect("replace metadata"); + assert!( + replaced + .attributes + .as_ref() + .is_none_or(|value| value.user_metadata.is_empty()) + ); + } + #[test] fn get_alias_accepts_only_one_remote_source_and_local_target() { let remote = ParsedPath::Remote(RemotePath::new("local", "reports", "report.json")); diff --git a/crates/cli/tests/recursive_remote_copy.rs b/crates/cli/tests/recursive_remote_copy.rs index 5e67f17c..5957a2fe 100644 --- a/crates/cli/tests/recursive_remote_copy.rs +++ b/crates/cli/tests/recursive_remote_copy.rs @@ -5,7 +5,7 @@ use std::io::{ErrorKind, Read, Write}; use std::net::{TcpListener, TcpStream}; use std::path::PathBuf; use std::process::{Command, Output, Stdio}; -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicUsize, Ordering}; use std::sync::{Arc, Mutex}; use std::thread::{self, JoinHandle}; use std::time::{Duration, Instant}; @@ -198,10 +198,14 @@ fn write_response(stream: &mut TcpStream, response: Response) { .headers .iter() .any(|(name, _)| name.eq_ignore_ascii_case("content-length")); - let mut head = format!( - "HTTP/1.1 {}\r\ncontent-type: application/xml\r\nconnection: keep-alive\r\n", - response.status - ); + let has_content_type = response + .headers + .iter() + .any(|(name, _)| name.eq_ignore_ascii_case("content-type")); + let mut head = format!("HTTP/1.1 {}\r\nconnection: keep-alive\r\n", response.status); + if !has_content_type { + head.push_str("content-type: application/xml\r\n"); + } if !has_content_length { head.push_str(&format!("content-length: {}\r\n", response.body.len())); } @@ -253,6 +257,29 @@ fn run_rc(mock: &S3Mock, args: &[&str]) -> Output { .expect("run rc command") } +fn run_rc_with_hosts(mock: &S3Mock, hosts: &[(&str, &str)], args: &[&str]) -> Output { + let config_dir = tempfile::tempdir().expect("create config directory"); + let authority = mock + .endpoint + .strip_prefix("http://") + .expect("mock endpoint has scheme"); + let alias = format!("http://ACCESS_KEY:SECRET_KEY@{authority}"); + 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.path()) + .env("AWS_EC2_METADATA_DISABLED", "true"); + for (name, _) in hosts { + command.env(format!("RC_HOST_{name}"), &alias); + } + command.output().expect("run rc command") +} + fn run_rc_with_stdin(mock: &S3Mock, args: &[&str], input: &[u8]) -> Output { let config_dir = tempfile::tempdir().expect("create config directory"); let authority = mock @@ -337,6 +364,32 @@ fn is_copy_request(request: &Request) -> bool { request.method == "PUT" && request.headers.contains_key("x-amz-copy-source") } +fn is_upload_request(request: &Request) -> bool { + request.method == "PUT" + && !request.headers.contains_key("x-amz-copy-source") + && !request.target.contains("uploads") +} + +fn object_get_result(body: &str) -> Response { + Response { + status: "200 OK", + headers: vec![ + ("content-length", body.len().to_string()), + ("etag", "\"source-etag\"".to_string()), + ("content-type", "text/plain".to_string()), + ], + body: body.to_string(), + } +} + +fn missing_object() -> Response { + Response { + status: "404 Not Found", + headers: Vec::new(), + body: "NoSuchKeymissing".to_string(), + } +} + #[test] fn single_copy_and_pipe_send_supported_storage_classes() { let mock = S3Mock::start(|request| { @@ -1517,3 +1570,369 @@ fn recursive_multipart_sigint_aborts_once_and_reports_cancelled_cleanup() { 1 ); } + +#[test] +fn cross_alias_recursive_dry_run_plans_without_server_side_copy() { + let mock = S3Mock::start(|request| { + if is_list_request(request) { + return one_object_list(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--recursive", + "--overwrite=true", + "--dry-run", + "--concurrency", + "1", + "alpha/source/src/", + "beta/destination/dst/", + ], + ); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}\nrequests: {:#?}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + mock.requests() + ); + let stdout = String::from_utf8_lossy(&output.stdout); + assert!(stdout.contains("Would copy: alpha/source/src/a.txt -> beta/destination/dst/a.txt")); + assert!( + mock.requests() + .iter() + .all(|request| request.method == "GET" && request.target.contains("list-type=2")), + "dry-run must only list the source: {:#?}", + mock.requests() + ); +} + +#[test] +fn cross_alias_copy_downloads_then_uploads_without_copy_source() { + let mock = S3Mock::start(|request| { + if request.method == "HEAD" && request.target == "/source/a.txt" { + return Response::head_with_etag(5, "source-etag"); + } + if request.method == "GET" && request.target.starts_with("/source/a.txt") { + return object_get_result("hello"); + } + if is_upload_request(request) && request.target.starts_with("/destination/b.txt") { + return Response { + status: "200 OK", + headers: vec![("etag", "\"dest-etag\"".to_string())], + body: String::new(), + }; + } + if request.method == "HEAD" && request.target.starts_with("/destination/") { + return missing_object(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--overwrite=true", + "alpha/source/a.txt", + "beta/destination/b.txt", + ], + ); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}\nrequests: {:#?}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + mock.requests() + ); + let requests = mock.requests(); + assert!( + requests.iter().any(|request| { + request.method == "GET" && request.target.starts_with("/source/a.txt") + }), + "cross-alias copy should download the source: {requests:#?}" + ); + assert!( + requests.iter().any(is_upload_request), + "cross-alias copy should upload to the destination: {requests:#?}" + ); + assert!( + requests.iter().all(|request| !is_copy_request(request)), + "cross-alias copy must not use CopyObject: {requests:#?}" + ); +} + +#[test] +fn cross_alias_copy_allows_metadata_replace() { + let mock = S3Mock::start(|request| { + if request.method == "HEAD" && request.target == "/source/a.txt" { + return Response::head_with_etag(5, "source-etag"); + } + if request.method == "GET" && request.target.starts_with("/source/a.txt") { + return object_get_result("hello"); + } + if is_upload_request(request) && request.target.starts_with("/destination/b.txt") { + return Response { + status: "200 OK", + headers: vec![("etag", "\"dest-etag\"".to_string())], + body: String::new(), + }; + } + if request.method == "HEAD" && request.target.starts_with("/destination/") { + return missing_object(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--overwrite=true", + "--metadata-directive", + "replace", + "--metadata", + "owner=analytics", + "alpha/source/a.txt", + "beta/destination/b.txt", + ], + ); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}\nrequests: {:#?}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + mock.requests() + ); + let requests = mock.requests(); + let upload = requests + .iter() + .find(|request| { + is_upload_request(request) && request.target.starts_with("/destination/b.txt") + }) + .expect("cross-alias replace should upload the destination"); + assert_eq!( + upload.headers.get("x-amz-meta-owner"), + Some(&"analytics".to_string()) + ); + assert!( + requests.iter().all(|request| !is_copy_request(request)), + "cross-alias replace must not use CopyObject: {requests:#?}" + ); +} + +#[test] +fn cross_alias_copy_rejects_same_size_source_replacement_after_download() { + let source_head_calls = Arc::new(AtomicUsize::new(0)); + let source_head_calls_for_handler = Arc::clone(&source_head_calls); + let mock = S3Mock::start(move |request| { + if request.method == "HEAD" && request.target == "/source/a.txt" { + let call = source_head_calls_for_handler.fetch_add(1, Ordering::SeqCst); + return if call < 2 { + Response::head_with_etag(5, "source-etag") + } else { + Response::head_with_etag(5, "replacement-etag") + }; + } + if request.method == "GET" && request.target.starts_with("/source/a.txt") { + return object_get_result("hello"); + } + if request.method == "HEAD" && request.target.starts_with("/destination/") { + return missing_object(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--overwrite=true", + "alpha/source/a.txt", + "beta/destination/b.txt", + ], + ); + + assert!(!output.status.success()); + let stderr = String::from_utf8_lossy(&output.stderr); + assert!( + stderr.contains("Source changed after copy planning"), + "stderr: {stderr}\nrequests: {:?}", + mock.requests() + ); + assert_eq!(source_head_calls.load(Ordering::SeqCst), 3); + assert!( + mock.requests() + .iter() + .all(|request| !is_upload_request(request)), + "source replacement must be detected before upload: {:?}", + mock.requests() + ); +} + +#[test] +fn cross_alias_recursive_copy_downloads_then_uploads_without_copy_source() { + let mock = S3Mock::start(|request| { + if is_list_request(request) { + return one_object_list(); + } + if request.method == "HEAD" && request.target.starts_with("/source/src/a.txt") { + return Response::head_with_etag(1, "source-etag"); + } + if request.method == "GET" && request.target.starts_with("/source/src/a.txt") { + return object_get_result("h"); + } + if is_upload_request(request) && request.target.starts_with("/destination/dst/a.txt") { + return Response { + status: "200 OK", + headers: vec![("etag", "\"dest-etag\"".to_string())], + body: String::new(), + }; + } + if request.method == "HEAD" && request.target.starts_with("/destination/") { + return missing_object(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--recursive", + "--overwrite=true", + "--concurrency", + "1", + "alpha/source/src/", + "beta/destination/dst/", + ], + ); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}\nrequests: {:#?}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + mock.requests() + ); + let requests = mock.requests(); + assert!( + requests.iter().any(is_list_request), + "recursive copy should list the source: {requests:#?}" + ); + assert!( + requests.iter().any(|request| { + request.method == "GET" && request.target.starts_with("/source/src/a.txt") + }), + "recursive cross-alias copy should download the source: {requests:#?}" + ); + assert!( + requests.iter().any(is_upload_request), + "recursive cross-alias copy should upload to the destination: {requests:#?}" + ); + assert!( + requests.iter().all(|request| !is_copy_request(request)), + "recursive cross-alias copy must not use CopyObject: {requests:#?}" + ); +} + +#[test] +fn cross_alias_copy_forwards_source_user_metadata_on_upload() { + let mock = S3Mock::start(|request| { + if request.method == "HEAD" && request.target.starts_with("/source/a.txt") { + return Response { + status: "200 OK", + headers: vec![ + ("content-length", "5".to_string()), + ("etag", "\"source-etag\"".to_string()), + ("content-type", "text/plain".to_string()), + ("x-amz-meta-owner", "storage".to_string()), + ], + body: String::new(), + }; + } + if request.method == "GET" && request.target.starts_with("/source/a.txt") { + return object_get_result("hello"); + } + if is_upload_request(request) && request.target.starts_with("/destination/b.txt") { + return Response { + status: "200 OK", + headers: vec![("etag", "\"dest-etag\"".to_string())], + body: String::new(), + }; + } + if request.method == "HEAD" && request.target.starts_with("/destination/") { + return missing_object(); + } + Response { + status: "500 Internal Server Error", + headers: Vec::new(), + body: "UnexpectedRequest".to_string(), + } + }); + + let output = run_rc_with_hosts( + &mock, + &[("alpha", ""), ("beta", "")], + &[ + "cp", + "--overwrite=true", + "alpha/source/a.txt", + "beta/destination/b.txt", + ], + ); + + assert!( + output.status.success(), + "stdout: {}\nstderr: {}\nrequests: {:#?}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr), + mock.requests() + ); + let requests = mock.requests(); + let upload = requests + .iter() + .find(|request| is_upload_request(request)) + .expect("cross-alias copy should upload"); + assert_eq!( + upload.headers.get("x-amz-meta-owner").map(String::as_str), + Some("storage"), + "upload should preserve source user metadata: {requests:#?}" + ); + assert!( + requests.iter().all(|request| !is_copy_request(request)), + "cross-alias copy must not use CopyObject: {requests:#?}" + ); +} diff --git a/crates/cli/tests/rm_purge.rs b/crates/cli/tests/rm_purge.rs index eeb4d438..bb52b2f9 100644 --- a/crates/cli/tests/rm_purge.rs +++ b/crates/cli/tests/rm_purge.rs @@ -49,6 +49,10 @@ impl TestServer { while !thread_stop.load(Ordering::SeqCst) { match listener.accept() { Ok((mut stream, _)) => { + // Accepted sockets can inherit nonblocking mode on some platforms. + stream + .set_nonblocking(false) + .expect("set accepted stream blocking"); if let Some(request) = read_request(&mut stream) { let response = response_for(&request); thread_requests diff --git a/crates/cli/tests/versioned_objects.rs b/crates/cli/tests/versioned_objects.rs index 6818ea02..18f82c9a 100644 --- a/crates/cli/tests/versioned_objects.rs +++ b/crates/cli/tests/versioned_objects.rs @@ -63,6 +63,10 @@ impl TestServer { while !thread_stop.load(Ordering::SeqCst) { match listener.accept() { Ok((mut stream, _)) => { + // Accepted sockets can inherit nonblocking mode on some platforms. + stream + .set_nonblocking(false) + .expect("set accepted stream blocking"); if let Some(request) = read_request(&mut stream) { let response = response_for(mode, &request); thread_requests diff --git a/docs/reference/rc/cp.md b/docs/reference/rc/cp.md index e650d33e..3c3bb34c 100644 --- a/docs/reference/rc/cp.md +++ b/docs/reference/rc/cp.md @@ -58,6 +58,13 @@ Copy between buckets on the same alias: rc cp local/reports/summary.json local/archive/summary.json ``` +Copy between aliases: + +```bash +rc cp --overwrite stage/data/report.json prod/archive/report.json +rc cp --recursive --overwrite stage/data/ prod/archive/ +``` + Copy multiple files with command-wide controls: ```bash @@ -84,7 +91,7 @@ rc cp ./reports/ local/archive/ --recursive --enc-kms local/archive/=alias/archi ## Behavior -The last path is always the target. Multiple sources require a local directory or remote prefix target, and ambiguous targets fail before any transfer starts. Sources can mix local and remote paths only where the command can infer a valid copy direction. S3-to-S3 copies are limited to paths under the same alias in the current implementation; use `rc mirror` for remote-to-remote synchronization across aliases. Recursive S3-to-S3 copy remains unsupported. Use trailing slashes consistently when copying directory-like prefixes. +The last path is always the target. Multiple sources require a local directory or remote prefix target, and ambiguous targets fail before any transfer starts. Sources can mix local and remote paths only where the command can infer a valid copy direction. Same-alias S3-to-S3 copies use server-side CopyObject, including recursive prefix copies. Cross-alias copies download through a temporary file and upload with the destination alias credentials. Use trailing slashes consistently when copying directory-like prefixes. Include rules restrict the candidate set when present. Exclude rules are evaluated afterwards and always win, regardless of flag order. `--newer-than` and `--older-than` use strict UTC comparisons; `--rewind` includes its boundary. Candidates without required source timestamps are skipped. Empty selections succeed unless `--fail-empty` is passed. @@ -99,6 +106,10 @@ The current implementation supports `SSE-S3` and `SSE-KMS`. It does not support When the server returns a source or destination object version ID, JSON copy output uses the output v3 `versioned_objects` envelope with `data.operation` set to `copy`. `data.source_version_id` identifies the copied source version and `data.version_id` identifies the created destination version. Copies for which the backend reports no version information retain the legacy JSON shape. +### BREAKING cross-alias copy contract migration + +Cross-alias `rc cp` is additive and does not change same-alias CopyObject behavior. Destinations on a different alias are copied by download then upload instead of failing as `unsupported_feature`. Source content type and user metadata are preserved unless `--metadata-directive replace` is set. This PR must be marked `BREAKING` because `docs/reference/rc/cp.md` is a protected CLI behavior contract. No JSON schema or config `schema_version` bump applies. + Global options shown in command syntax use the same meaning everywhere: | Option | Description | diff --git a/scripts/regression/cross-alias-copy.sh b/scripts/regression/cross-alias-copy.sh new file mode 100755 index 00000000..b823ccac --- /dev/null +++ b/scripts/regression/cross-alias-copy.sh @@ -0,0 +1,62 @@ +#!/usr/bin/env bash +# +# Regression tests for cross-alias S3-to-S3 copy. +# +# Usage: +# ./scripts/regression/cross-alias-copy.sh +# +# These tests do not require a running S3 backend. They cover: +# - Planning clients for both source and destination aliases +# - Same-alias remote copies still using one client alias +# - Recursive dry-run and live copies across aliases without CopyObject +# - Single-object overwrite using download then upload +# - Source user metadata forwarded on the destination upload +# - Storage-class rejection using the upload multipart threshold +# + +set -euo pipefail + +SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)" +PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)" +cd "$PROJECT_ROOT" + +RED='\033[0;31m' +GREEN='\033[0;32m' +BLUE='\033[0;34m' +NC='\033[0m' + +log_info() { echo -e "${BLUE}[INFO]${NC} $*"; } +log_success() { echo -e "${GREEN}[PASS]${NC} $*"; } +log_error() { echo -e "${RED}[FAIL]${NC} $*"; } + +run_tests() { + local description="$1" + shift + log_info "$description" + if "$@"; then + log_success "$description" + else + log_error "$description" + return 1 + fi +} + +log_info "Cross-alias copy regression suite" + +run_tests "copy planner unit tests" \ + cargo test -p rustfs-cli --lib -- \ + planned_client_aliases_include_both_sides_of_a_cross_alias_copy \ + planned_client_aliases_keep_same_alias_remote_copy_on_one_alias \ + storage_class_plan_rejects_cross_alias_multipart_uploads \ + piped_copy_preserves_source_content_type_unless_replaced \ + piped_copy_preserves_source_user_metadata_unless_replaced + +run_tests "cross-alias copy integration tests" \ + cargo test -p rustfs-cli --test recursive_remote_copy -- \ + cross_alias_recursive_dry_run_plans_without_server_side_copy \ + cross_alias_copy_downloads_then_uploads_without_copy_source \ + cross_alias_recursive_copy_downloads_then_uploads_without_copy_source \ + cross_alias_copy_forwards_source_user_metadata_on_upload \ + recursive_same_alias_copy_paginates_and_emits_deterministic_plan + +log_success "Cross-alias copy regression suite passed"