feat(transfer): align cross-alias copy, move, and diff on source identity

crates/cli/src/commands/cp.rs

@@ -25,6 +25,7 @@ use crate::exit_code::ExitCode;
 use crate::output::{Formatter, OutputConfig, ProgressBar, V3SuccessEnvelope};
 use crate::secret_input::{SecretLocator, resolve_secret_locator};
 
+use super::object_identity::set_source_identity;
 use super::transfer_fidelity::{MetadataDirectiveArg, TaggingDirectiveArg, TransferFidelityArgs};
 
 const CP_AFTER_HELP: &str = "\
@@ -689,7 +690,10 @@ fn validate_fidelity_directions(
         && sources.iter().any(|source| {
             matches!(
                 source,
-                ParsedPath::Remote(source) if target.as_remote().is_some_and(|target| source.alias == target.alias)
+                ParsedPath::Remote(source)
+                    if target
+                        .as_remote()
+                        .is_some_and(|target| source.alias == target.alias)
             )
         });
     if any_remote && target_remote {
@@ -1245,6 +1249,7 @@ async fn perform_planned_download(
 struct PlannedRemoteCopyResult {
     bytes_copied: u64,
     source_version_id: Option<String>,
+    source_etag: Option<String>,
     destination_version_id: Option<String>,
     upload_id: Option<String>,
     object: ObjectInfo,
@@ -1312,6 +1317,7 @@ async fn perform_planned_remote_copy(
         return Ok(PlannedRemoteCopyResult {
             bytes_copied: copied.bytes_copied,
             source_version_id: options.source_version_id,
+            source_etag: current.etag.clone(),
             destination_version_id: copied.object.version_id.clone(),
             upload_id: Some(copied.upload_id),
             object: copied.object,
@@ -1332,6 +1338,7 @@ async fn perform_planned_remote_copy(
             .source_version_id
             .clone()
             .or_else(|| source_info.version_id.clone()),
+        source_etag: source_info.etag.clone(),
         destination_version_id: copied.version_id.clone(),
         upload_id: None,
         object: copied,
@@ -1427,6 +1434,7 @@ async fn perform_cross_alias_remote_copy(
         Ok(PlannedRemoteCopyResult {
             bytes_copied,
             source_version_id: current.version_id.clone(),
+            source_etag: current.etag.clone(),
             destination_version_id: object.version_id.clone(),
             upload_id: None,
             object,
@@ -1451,6 +1459,46 @@ fn source_identity_matches(planned: &ObjectInfo, current: &ObjectInfo) -> bool {
     }
 }
 
+/// Copy one object between two aliases by streaming through the client.
+///
+/// `rc mv` shares this path so a cross-alias move behaves exactly like a
+/// cross-alias copy followed by a source delete, rather than reimplementing the
+/// download/upload streaming and its source-change checks.
+#[derive(Debug, Clone)]
+pub(super) struct CrossAliasCopyResult {
+    pub(super) object: ObjectInfo,
+    pub(super) source_version_id: Option<String>,
+    pub(super) source_etag: Option<String>,
+}
+
+pub(super) async fn copy_object_across_aliases(
+    source_client: &S3Client,
+    target_client: &S3Client,
+    source: &RemotePath,
+    target: &RemotePath,
+    encryption: Option<&ObjectEncryptionRequest>,
+) -> rc_core::Result<CrossAliasCopyResult> {
+    let source_info = source_client.head_object(source).await?;
+    let args = CpArgs::single(source.to_string(), target.to_string());
+    let ignore_progress = |_: u64| {};
+    let result = perform_cross_alias_remote_copy(
+        source_client,
+        target_client,
+        source,
+        target,
+        &source_info,
+        encryption,
+        &ignore_progress,
+        &args,
+    )
+    .await?;
+    Ok(CrossAliasCopyResult {
+        object: result.object,
+        source_version_id: result.source_version_id,
+        source_etag: result.source_etag,
+    })
+}
+
 fn piped_copy_write_options(
     args: &CpArgs,
     source: &ObjectInfo,
@@ -1463,20 +1511,27 @@ fn piped_copy_write_options(
         args.destination_customer_key.as_ref(),
         args.storage_class.clone(),
     )?;
-    if matches!(
+    let replace_metadata = 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 !replace_metadata {
+        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.user_metadata.is_empty()
-        && let Some(metadata) = &source.metadata
-    {
-        attributes.user_metadata.clone_from(metadata);
+    // The destination computes its own ETag, so record the source ETag the same
+    // way `rc mirror` does. Without this a later `mirror --compare auto` cannot
+    // tell a faithful cross-alias copy from a changed object and recopies it.
+    // This is `rc` bookkeeping rather than user data, so it survives --metadata-directive replace.
+    if let Some(source_etag) = source.etag.as_deref() {
+        set_source_identity(&mut attributes, source_etag);
     }
     if attributes != ObjectAttributes::default() {
         options.attributes = Some(attributes);
@@ -1502,7 +1557,14 @@ fn multipart_options_from_source(source: &ObjectInfo) -> rc_core::Result<Multipa
     let mut options = MultipartCopyOptions::new(source_size, source_etag)?;
     options.source_version_id = source.version_id.clone();
     options.content_type = source.content_type.clone();
-    options.metadata = source.metadata.clone().unwrap_or_default();
+    let mut attributes = ObjectAttributes {
+        user_metadata: source.metadata.clone().unwrap_or_default(),
+        ..ObjectAttributes::default()
+    };
+    if let Some(source_etag) = source.etag.as_deref() {
+        set_source_identity(&mut attributes, source_etag);
+    }
+    options.metadata = attributes.user_metadata;
     Ok(options)
 }
 
@@ -3915,6 +3977,10 @@ mod tests {
             options.metadata.get("project").map(String::as_str),
             Some("archive")
         );
+        assert_eq!(
+            options.metadata.get("rc-source-etag").map(String::as_str),
+            Some("planned-etag")
+        );
     }
 
     #[test]
@@ -4124,7 +4190,7 @@ mod tests {
                 &cross_alias_target,
             )
             .is_ok(),
-            "metadata REPLACE is implemented by the cross-alias upload path"
+            "cross-alias metadata REPLACE uses the upload path"
         );
 
         let mut tags = CpArgs::single("test/source/report.json", "test/target/report.json");
@@ -4144,15 +4210,14 @@ mod tests {
     }
 
     #[test]
-    fn source_identity_validation_detects_same_size_etag_changes() {
+    fn source_identity_validation_rejects_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, &current));
 
         current.etag = Some("changed".to_string());
         assert!(!source_identity_matches(&planned, &current));
-
         current.etag = None;
         assert!(!source_identity_matches(&planned, &current));
 
@@ -4262,6 +4327,69 @@ mod tests {
         );
     }
 
+    #[test]
+    fn piped_copy_records_the_source_etag_as_identity_metadata() {
+        let mut source = ObjectInfo::file("file.txt", 4);
+        source.etag = Some("source-etag".to_string());
+        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
+
+        let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
+
+        let attributes = options.attributes.as_ref().expect("identity attributes");
+        assert_eq!(
+            super::super::object_identity::identity_etag_from_metadata(Some(
+                &attributes.user_metadata
+            ))
+            .as_deref(),
+            Some("source-etag"),
+            "a later mirror --compare auto must be able to skip this object"
+        );
+    }
+
+    #[test]
+    fn piped_copy_records_identity_even_when_metadata_is_replaced() {
+        let mut source = ObjectInfo::file("file.txt", 4);
+        source.etag = Some("source-etag".to_string());
+        source.metadata = Some(HashMap::from([(
+            "owner".to_string(),
+            "storage".to_string(),
+        )]));
+        let mut args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
+        args.metadata_directive = Some(MetadataDirectiveArg::Replace);
+
+        let options = piped_copy_write_options(&args, &source, None).expect("replace metadata");
+
+        let attributes = options.attributes.as_ref().expect("identity attributes");
+        assert_eq!(
+            super::super::object_identity::identity_etag_from_metadata(Some(
+                &attributes.user_metadata
+            ))
+            .as_deref(),
+            Some("source-etag"),
+            "identity is rc bookkeeping, not user metadata"
+        );
+        assert!(
+            !attributes.user_metadata.contains_key("owner"),
+            "replace must still drop source user metadata"
+        );
+    }
+
+    #[test]
+    fn piped_copy_omits_identity_when_the_source_has_no_etag() {
+        let source = ObjectInfo::file("file.txt", 4);
+        let args = CpArgs::single("alpha/source/file.txt", "beta/target/file.txt");
+
+        let options = piped_copy_write_options(&args, &source, None).expect("copy metadata");
+
+        assert!(
+            options
+                .attributes
+                .as_ref()
+                .is_none_or(|attributes| attributes.user_metadata.is_empty()),
+            "without a source ETag there is no identity to record"
+        );
+    }
+
     #[test]
     fn piped_copy_preserves_source_user_metadata_unless_replaced() {
         let mut source = ObjectInfo::file("file.txt", 4);