feat(copy-core): add multipart server-side copy primitive

.github/workflows/integration.yml

@@ -109,6 +109,12 @@ jobs:
             "option_behavior_operations::test_mirror_parallel_zero_returns_usage_error"
           )
 
+          if [[ "${{ matrix.version }}" == "1.0.0-beta.10" ]]; then
+            TESTS+=(
+              "multipart_operations::test_multipart_server_side_copy_direct_primitive"
+            )
+          fi
+
           for test_name in "${TESTS[@]}"; do
             echo "==> Running $test_name"
             cargo test \

crates/cli/tests/integration.rs

@@ -1216,6 +1216,11 @@ mod presigned_urls {
 
 mod multipart_operations {
     use super::*;
+    use rc_core::{
+        Alias, MultipartCopyCancellation, MultipartCopyOptions, ObjectStore, RemotePath,
+        S3_MULTIPART_COPY_MIN_PART_SIZE,
+    };
+    use rc_s3::S3Client;
     use std::io::Write;
 
     #[test]
@@ -1337,6 +1342,92 @@ mod multipart_operations {
         // Cleanup
         cleanup_bucket(config_dir.path(), &bucket_name);
     }
+
+    #[test]
+    fn test_multipart_server_side_copy_direct_primitive() {
+        let (endpoint, access_key, secret_key) = match get_test_config() {
+            Some(config) => config,
+            None => panic!("S3 integration test setup failed"),
+        };
+        let runtime = tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("build multipart copy test runtime");
+
+        runtime.block_on(async {
+            let bucket = format!("test-multipart-copy-{}", uuid_suffix());
+            let source = RemotePath::new("test", &bucket, "source.bin");
+            let destination = RemotePath::new("test", &bucket, "destination.bin");
+            let mut alias = Alias::new("test", endpoint, access_key, secret_key);
+            alias.bucket_lookup = "path".to_string();
+            let client = S3Client::new(alias).await.expect("create direct S3 client");
+            client
+                .create_bucket(&bucket)
+                .await
+                .expect("create multipart copy bucket");
+
+            let source_size = S3_MULTIPART_COPY_MIN_PART_SIZE + 1;
+            let source_data = (0..source_size)
+                .map(|index| (index % 251) as u8)
+                .collect::<Vec<_>>();
+            client
+                .put_object(
+                    &source,
+                    source_data.clone(),
+                    Some("application/octet-stream"),
+                    None,
+                )
+                .await
+                .expect("upload multipart copy source");
+            let source_info = client
+                .head_object(&source)
+                .await
+                .expect("head multipart copy source");
+            let source_etag = source_info
+                .etag
+                .as_deref()
+                .expect("source ETag")
+                .to_string();
+            let mut options = MultipartCopyOptions::new(source_size, source_etag)
+                .expect("multipart copy options");
+            options.preferred_part_size = Some(S3_MULTIPART_COPY_MIN_PART_SIZE);
+            options.content_type = source_info.content_type.clone();
+            options.metadata = source_info.metadata.unwrap_or_default();
+
+            let copy = client
+                .multipart_copy(
+                    &source,
+                    &destination,
+                    &options,
+                    &MultipartCopyCancellation::new(),
+                    None,
+                    &|_| {},
+                )
+                .await
+                .expect("multipart server-side copy");
+            let destination_info = client
+                .head_object(&destination)
+                .await
+                .expect("head multipart copy destination");
+            let destination_data = client
+                .get_object(&destination)
+                .await
+                .expect("download multipart copy destination");
+
+            let _ = client.delete_object(&destination).await;
+            let _ = client.delete_object(&source).await;
+            let _ = client.delete_bucket(&bucket).await;
+
+            assert!(copy.part_count > 1);
+            assert_eq!(copy.bytes_copied, source_size);
+            assert_eq!(destination_info.size_bytes, Some(source_size as i64));
+            assert_eq!(
+                destination_info.content_type.as_deref(),
+                Some("application/octet-stream")
+            );
+            assert_eq!(destination_data, source_data);
+        });
+    }
 }
 
 mod recursive_operations {

crates/core/src/admin/kms_diagnostic.rs

@@ -241,7 +241,8 @@ fn classify_and_discard_error(mut error: Error) -> KmsRoundTripErrorClass {
         | Error::InvalidPath(message)
         | Error::UnsupportedFeature(message)
         | Error::General(message)
-        | Error::RequestRejected(message) => {
+        | Error::RequestRejected(message)
+        | Error::Interrupted(message) => {
             message.zeroize();
             KmsRoundTripErrorClass::General
         }

crates/core/src/error.rs

@@ -4,6 +4,24 @@
 
 use thiserror::Error;
 
+/// Outcome of cleaning up a multipart upload after a copy failure.
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum MultipartAbortStatus {
+    /// The service accepted the abort request.
+    Succeeded,
+    /// The abort request failed; the upload may require later cleanup.
+    Failed,
+}
+
+impl std::fmt::Display for MultipartAbortStatus {
+    fn fmt(&self, formatter: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            Self::Succeeded => formatter.write_str("succeeded"),
+            Self::Failed => formatter.write_str("failed"),
+        }
+    }
+}
+
 /// Result type alias for rc-core operations
 pub type Result<T> = std::result::Result<T, Error>;
 
@@ -100,6 +118,10 @@ pub enum Error {
     /// Request rejected locally before any network operation was attempted
     #[error("{0}")]
     RequestRejected(String),
+
+    /// Operation stopped through an explicit cooperative cancellation request.
+    #[error("Operation interrupted: {0}")]
+    Interrupted(String),
 }
 
 impl Error {
@@ -116,10 +138,57 @@ impl Error {
             | Error::AliasNotFound(_) => 5, // NotFound
             Error::Conflict(_) | Error::GovernanceDenied { .. } | Error::AliasExists(_) => 6, // Conflict
             Error::UnsupportedFeature(_) => 7, // UnsupportedFeature
+            Error::Interrupted(_) => 130,      // Interrupted
             Error::RequestRejected(_) => 1,    // GeneralError
             _ => 1,                            // GeneralError
         }
     }
+
+    /// Add multipart cleanup diagnostics without changing the primary error class.
+    pub fn with_multipart_copy_context(
+        self,
+        upload_id: &str,
+        abort_status: MultipartAbortStatus,
+    ) -> Self {
+        // Upload IDs originate at the service boundary. Escaping and bounding
+        // them prevents terminal control sequences or oversized diagnostics.
+        let safe_upload_id = upload_id
+            .chars()
+            .flat_map(char::escape_default)
+            .take(256)
+            .collect::<String>();
+        let context = format!("multipart upload ID: {safe_upload_id}, abort: {abort_status}");
+        match self {
+            Self::Auth(message) => Self::Auth(format!("{message}; {context}")),
+            Self::Network(message) => Self::Network(format!("{message}; {context}")),
+            Self::NotFound(message) => Self::NotFound(format!("{message}; {context}")),
+            Self::VersionNotFound { path, version_id } => Self::VersionNotFound {
+                path: format!("{path} ({context})"),
+                version_id,
+            },
+            Self::DeleteMarker { path, version_id } => Self::DeleteMarker {
+                path: format!("{path} ({context})"),
+                version_id,
+            },
+            Self::GovernanceDenied { path, version_id } => Self::GovernanceDenied {
+                path: format!("{path} ({context})"),
+                version_id,
+            },
+            Self::Conflict(message) => Self::Conflict(format!("{message}; {context}")),
+            Self::UnsupportedFeature(message) => {
+                Self::UnsupportedFeature(format!("{message}; {context}"))
+            }
+            Self::InvalidPath(message) => Self::InvalidPath(format!("{message}; {context}")),
+            Self::Config(message) => Self::Config(format!("{message}; {context}")),
+            Self::AliasNotFound(message) => Self::AliasNotFound(format!("{message}; {context}")),
+            Self::AliasExists(message) => Self::AliasExists(format!("{message}; {context}")),
+            Self::RequestRejected(message) => {
+                Self::RequestRejected(format!("{message}; {context}"))
+            }
+            Self::Interrupted(message) => Self::Interrupted(format!("{message}; {context}")),
+            source => Self::General(format!("{source}; {context}")),
+        }
+    }
 }
 
 #[cfg(test)]
@@ -172,4 +241,19 @@ mod tests {
         assert_eq!(governance.exit_code(), 6);
         assert!(governance.to_string().contains("Governance retention"));
     }
+
+    #[test]
+    fn multipart_copy_failure_preserves_primary_classification() {
+        let error = Error::Auth("access denied".to_string())
+            .with_multipart_copy_context("upload-123", MultipartAbortStatus::Succeeded);
+
+        assert_eq!(error.exit_code(), 4);
+        assert!(error.to_string().contains("upload-123"));
+        assert!(error.to_string().contains("abort: succeeded"));
+
+        let interrupted = Error::Interrupted("cancelled".to_string())
+            .with_multipart_copy_context("upload-456", MultipartAbortStatus::Succeeded);
+        assert_eq!(interrupted.exit_code(), 130);
+        assert!(interrupted.to_string().contains("upload-456"));
+    }
 }

crates/core/src/lib.rs

@@ -16,6 +16,7 @@ pub mod cors;
 pub mod encryption;
 pub mod error;
 pub mod lifecycle;
+pub mod multipart_copy;
 pub mod object_lock;
 pub mod ops;
 pub mod path;
@@ -34,11 +35,17 @@ pub use alias::{
 pub use config::{Config, ConfigManager};
 pub use cors::{CorsConfiguration, CorsRule};
 pub use encryption::{BucketEncryption, ObjectEncryptionRequest};
-pub use error::{Error, Result};
+pub use error::{Error, MultipartAbortStatus, Result};
 pub use lifecycle::{
     LifecycleConfiguration, LifecycleExpiration, LifecycleRule, LifecycleRuleStatus,
     LifecycleTransition, NoncurrentVersionExpiration, NoncurrentVersionTransition,
 };
+pub use multipart_copy::{
+    DEFAULT_MULTIPART_COPY_PART_SIZE, MultipartCopyCancellation, MultipartCopyOptions,
+    MultipartCopyPart, MultipartCopyPlan, MultipartCopyProgress, MultipartCopyResult,
+    S3_MAX_OBJECT_SIZE, S3_MULTIPART_COPY_MAX_PART_SIZE, S3_MULTIPART_COPY_MAX_PARTS,
+    S3_MULTIPART_COPY_MIN_PART_SIZE, S3_SINGLE_COPY_MAX_SIZE, requires_multipart_copy,
+};
 pub use object_lock::{
     BucketObjectLockConfiguration, DefaultRetention, LegalHoldStatus, ObjectLockOptions,
     ObjectRetention, RetentionDuration, RetentionDurationUnit, RetentionMode,