From e791543c9feb35e2181528eb67b6a8b61e9f2c75 Mon Sep 17 00:00:00 2001 From: overtrue Date: Sun, 30 Aug 2026 08:43:54 +0800 Subject: [PATCH] fix(transfer): fence server-side copies by source ETag --- crates/cli/src/commands/cp.rs | 23 ++++++++--- crates/cli/src/commands/mv.rs | 6 ++- crates/cli/src/commands/undo.rs | 1 + crates/cli/tests/recursive_remote_copy.rs | 4 ++ crates/core/src/traits.rs | 25 ++++++++++-- crates/core/src/transfer_options.rs | 10 ++++- crates/core/tests/undo_planner.rs | 3 +- crates/s3/src/client.rs | 48 ++++++++++++++++++++--- 8 files changed, 102 insertions(+), 18 deletions(-) diff --git a/crates/cli/src/commands/cp.rs b/crates/cli/src/commands/cp.rs index 7b681cc1..5aeeb366 100644 --- a/crates/cli/src/commands/cp.rs +++ b/crates/cli/src/commands/cp.rs @@ -618,6 +618,7 @@ fn requested_metadata_directive(args: &CpArgs) -> Option { fn transfer_copy_options( args: &CpArgs, source_version_id: Option, + source_etag: Option, encryption: Option<&ObjectEncryptionRequest>, ) -> rc_core::Result { let metadata_directive = requested_metadata_directive(args); @@ -645,6 +646,7 @@ fn transfer_copy_options( customer_key: args.source_customer_key.clone(), ..TransferReadOptions::default() }, + source_etag, metadata_directive, tagging_directive, destination, @@ -697,7 +699,7 @@ fn validate_fidelity_directions( ) }); if any_remote && target_remote { - let copy_options = transfer_copy_options(args, None, None)?; + let copy_options = transfer_copy_options(args, None, None, None)?; if same_alias_remote_copy && matches!( copy_options.metadata_directive, @@ -1303,7 +1305,7 @@ async fn perform_planned_remote_copy( ))); } let options = multipart_options_from_source(¤t)?; - let transfer = transfer_copy_options(args, current.version_id.clone(), encryption)?; + let transfer = transfer_copy_options(args, current.version_id.clone(), None, encryption)?; let copied = source_client .multipart_copy_with_transfer_options( source, @@ -1323,7 +1325,12 @@ async fn perform_planned_remote_copy( object: copied.object, }); } - let options = transfer_copy_options(args, source_info.version_id.clone(), encryption)?; + let options = transfer_copy_options( + args, + source_info.version_id.clone(), + source_info.etag.clone(), + encryption, + )?; let copied = source_client .copy_object_with_transfer_options(source, target, &options) .await?; @@ -4233,11 +4240,17 @@ mod tests { args.fidelity.retain_until = Some("2099-01-02T03:04:05Z".to_string()); args.fidelity.legal_hold = Some("ON".to_string()); - let options = - transfer_copy_options(&args, Some("source-v1".to_string()), None).expect("copy policy"); + let options = transfer_copy_options( + &args, + Some("source-v1".to_string()), + Some("source-etag".to_string()), + None, + ) + .expect("copy policy"); assert_eq!(options.metadata_directive, Some(MetadataDirective::Copy)); assert!(options.destination.attributes.is_none()); assert_eq!(options.source.version_id.as_deref(), Some("source-v1")); + assert_eq!(options.source_etag.as_deref(), Some("source-etag")); assert!(options.destination.retention.is_some()); assert_eq!( options.destination.legal_hold, diff --git a/crates/cli/src/commands/mv.rs b/crates/cli/src/commands/mv.rs index 24cccb5c..c70aa551 100644 --- a/crates/cli/src/commands/mv.rs +++ b/crates/cli/src/commands/mv.rs @@ -406,8 +406,10 @@ async fn copy_for_move( } None => { let source_info = source_client.head_object(source).await?; - let copy_options = - CopyObjectOptions::for_source_version(source_info.version_id.clone())?; + let copy_options = CopyObjectOptions::for_source_identity( + source_info.version_id.clone(), + source_info.etag.clone(), + )?; let object = source_client .copy_object_with_options(source, target, ©_options, encryption) .await?; diff --git a/crates/cli/src/commands/undo.rs b/crates/cli/src/commands/undo.rs index b4bc5c78..c5b31acc 100644 --- a/crates/cli/src/commands/undo.rs +++ b/crates/cli/src/commands/undo.rs @@ -361,6 +361,7 @@ async fn execute_plan_item( UndoAction::RestoreVersion { source_version_id } => { let options = CopyObjectOptions { source_version_id: Some(source_version_id.clone()), + source_etag: None, }; store.copy_version(&path, &options).await.and_then(|info| { let created_version_id = info.version_id.ok_or_else(|| { diff --git a/crates/cli/tests/recursive_remote_copy.rs b/crates/cli/tests/recursive_remote_copy.rs index 96a25d84..ccc34fab 100644 --- a/crates/cli/tests/recursive_remote_copy.rs +++ b/crates/cli/tests/recursive_remote_copy.rs @@ -1686,6 +1686,10 @@ fn recursive_copy_executes_the_planned_server_side_copy() { copies[0].headers.get("x-amz-copy-source"), Some(&"source/src/a.txt".to_string()) ); + assert_eq!( + copies[0].headers.get("x-amz-copy-source-if-match"), + Some(&"\"source-etag\"".to_string()) + ); } #[test] diff --git a/crates/core/src/traits.rs b/crates/core/src/traits.rs index d00ea00f..8f60413b 100644 --- a/crates/core/src/traits.rs +++ b/crates/core/src/traits.rs @@ -135,17 +135,35 @@ pub struct ObjectReadOptions { pub struct CopyObjectOptions { /// Exact historical source version to copy. `None` selects the current source. pub source_version_id: Option, + /// Copy only when the selected source still has this ETag. + pub source_etag: Option, } impl CopyObjectOptions { /// Build copy options while rejecting an ambiguous empty source version identifier. pub fn for_source_version(source_version_id: Option) -> Result { + Self::for_source_identity(source_version_id, None) + } + + /// Build copy options pinned to an exact source version or ETag. + pub fn for_source_identity( + source_version_id: Option, + source_etag: Option, + ) -> Result { if source_version_id.as_deref().is_some_and(str::is_empty) { return Err(Error::InvalidPath( "Source version ID cannot be empty".to_string(), )); } - Ok(Self { source_version_id }) + if source_etag.as_deref().is_some_and(str::is_empty) { + return Err(Error::InvalidPath( + "Source ETag cannot be empty".to_string(), + )); + } + Ok(Self { + source_version_id, + source_etag, + }) } } @@ -793,10 +811,9 @@ pub trait ObjectStore: Send + Sync { options: &CopyObjectOptions, encryption: Option<&ObjectEncryptionRequest>, ) -> Result { - if options.source_version_id.is_some() { + if options.source_version_id.is_some() || options.source_etag.is_some() { return Err(Error::UnsupportedFeature( - "Exact-version server-side copy is not implemented by this object store" - .to_string(), + "Exact-source server-side copy is not implemented by this object store".to_string(), )); } self.copy_object(src, dst, encryption).await diff --git a/crates/core/src/transfer_options.rs b/crates/core/src/transfer_options.rs index e75a1a65..2da01aee 100644 --- a/crates/core/src/transfer_options.rs +++ b/crates/core/src/transfer_options.rs @@ -474,6 +474,8 @@ impl From for TransferReadOptions { pub struct TransferCopyOptions { /// Source version, checksum, and SSE-C selection. pub source: TransferReadOptions, + /// Copy only when the selected source still has this ETag. + pub source_etag: Option, /// Source metadata handling requested for the destination. pub metadata_directive: Option, /// Source tag handling requested for the destination. @@ -486,6 +488,11 @@ impl TransferCopyOptions { /// Validate directives and their replacement payloads before mutation. pub fn validate(&self) -> Result<()> { self.source.validate()?; + if self.source_etag.as_deref().is_some_and(str::is_empty) { + return Err(Error::InvalidPath( + "Source ETag cannot be empty".to_string(), + )); + } self.destination.validate()?; match self.metadata_directive { None | Some(MetadataDirective::Copy) if self.destination.attributes.is_some() => { @@ -529,7 +536,8 @@ impl TransferCopyOptions { } let source = self.source.legacy_read_options()?; let encryption = self.destination.legacy_copy_encryption()?; - let copy = CopyObjectOptions::for_source_version(source.version_id)?; + let copy = + CopyObjectOptions::for_source_identity(source.version_id, self.source_etag.clone())?; Ok((copy, encryption)) } diff --git a/crates/core/tests/undo_planner.rs b/crates/core/tests/undo_planner.rs index af8345aa..dd8bb15c 100644 --- a/crates/core/tests/undo_planner.rs +++ b/crates/core/tests/undo_planner.rs @@ -173,8 +173,9 @@ fn overwrite_after_delete_marker_is_refused_instead_of_guessing() { } #[test] -fn copy_options_reject_empty_source_version_ids() { +fn copy_options_reject_empty_source_identities() { assert!(CopyObjectOptions::for_source_version(Some(String::new())).is_err()); + assert!(CopyObjectOptions::for_source_identity(None, Some(String::new())).is_err()); assert_eq!( CopyObjectOptions::for_source_version(Some("data-v1".to_string())) .expect("valid source version") diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 8d692ffc..7e7cccc5 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -5501,17 +5501,18 @@ impl ObjectStore for S3Client { encryption: Option<&ObjectEncryptionRequest>, ) -> Result { let copy_source = encoded_copy_source(src, options.source_version_id.as_deref()); - let response = apply_object_encryption_to_copy_request( + let mut request = apply_object_encryption_to_copy_request( self.inner .copy_object() .copy_source(©_source) .bucket(&dst.bucket) .key(&dst.key), encryption, - ) - .send() - .await - .map_err(|error| { + ); + if let Some(source_etag) = options.source_etag.as_deref() { + request = request.copy_source_if_match(quoted_etag(source_etag)); + } + let response = request.send().await.map_err(|error| { Self::map_object_request_error(&error, src, options.source_version_id.as_deref()) })?; @@ -5555,6 +5556,9 @@ impl ObjectStore for S3Client { if matches!(options.metadata_directive, Some(MetadataDirective::Copy)) { request = request.metadata_directive(aws_sdk_s3::types::MetadataDirective::Copy); } + if let Some(source_etag) = options.source_etag.as_deref() { + request = request.copy_source_if_match(quoted_etag(source_etag)); + } request = apply_object_lock_to_copy_request(request, &options.destination)?; let response = request.send().await.map_err(|error| { if options.destination.retention.is_some() || options.destination.legal_hold.is_some() { @@ -11389,6 +11393,35 @@ mod tests { ); } + #[tokio::test] + async fn copy_object_with_options_conditions_on_source_etag() { + let response = http::Response::builder() + .status(412) + .header("x-amz-error-code", "PreconditionFailed") + .body(SdkBody::from( + "PreconditionFailedsource changed", + )) + .expect("build precondition response"); + let (client, request_receiver) = test_s3_client(Some(response)); + let src = RemotePath::new("test", "source-bucket", "src.txt"); + let dst = RemotePath::new("test", "destination-bucket", "dst.txt"); + let options = + CopyObjectOptions::for_source_identity(None, Some("planned-source-etag".to_string())) + .expect("valid source identity"); + + let error = client + .copy_object_with_options(&src, &dst, &options, None) + .await + .expect_err("a changed source must block the copy"); + + let request = request_receiver.expect_request(); + assert_eq!( + request.headers().get("x-amz-copy-source-if-match"), + Some("\"planned-source-etag\"") + ); + assert!(matches!(error, Error::Conflict(_))); + } + #[tokio::test] async fn transfer_copy_sends_explicit_metadata_copy_directive() { let response = http::Response::builder() @@ -11407,6 +11440,7 @@ mod tests { &src, &dst, &TransferCopyOptions { + source_etag: Some("planned-source-etag".to_string()), metadata_directive: Some(MetadataDirective::Copy), destination: ObjectWriteOptions { storage_class: Some("STANDARD".to_string()), @@ -11426,6 +11460,10 @@ mod tests { request.headers().get("x-amz-storage-class"), Some("STANDARD") ); + assert_eq!( + request.headers().get("x-amz-copy-source-if-match"), + Some("\"planned-source-etag\"") + ); } #[tokio::test]