From e227c6e18ab18a219ee0c7dec9d2bfc40f100efc Mon Sep 17 00:00:00 2001 From: cxymds Date: Thu, 23 Jul 2026 13:28:52 +0800 Subject: [PATCH] feat(phase-6): add multipart server-side copy primitive --- .github/workflows/integration.yml | 6 + crates/cli/tests/integration.rs | 91 +++ crates/core/src/admin/kms_diagnostic.rs | 3 +- crates/core/src/error.rs | 84 +++ crates/core/src/lib.rs | 9 +- crates/core/src/multipart_copy.rs | 374 ++++++++++ crates/core/src/traits.rs | 21 + crates/s3/src/client.rs | 952 +++++++++++++++++++++++- 8 files changed, 1529 insertions(+), 11 deletions(-) create mode 100644 crates/core/src/multipart_copy.rs diff --git a/.github/workflows/integration.yml b/.github/workflows/integration.yml index 9a133559..7405ecd4 100644 --- a/.github/workflows/integration.yml +++ b/.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 \ diff --git a/crates/cli/tests/integration.rs b/crates/cli/tests/integration.rs index ac0c5fc0..f95c61dc 100644 --- a/crates/cli/tests/integration.rs +++ b/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::>(); + 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 { diff --git a/crates/core/src/admin/kms_diagnostic.rs b/crates/core/src/admin/kms_diagnostic.rs index d6049c3b..e8181378 100644 --- a/crates/core/src/admin/kms_diagnostic.rs +++ b/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 } diff --git a/crates/core/src/error.rs b/crates/core/src/error.rs index c3e3d814..6e2e0001 100644 --- a/crates/core/src/error.rs +++ b/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 = std::result::Result; @@ -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::(); + 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")); + } } diff --git a/crates/core/src/lib.rs b/crates/core/src/lib.rs index c63ba6ea..f7a4df7d 100644 --- a/crates/core/src/lib.rs +++ b/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, diff --git a/crates/core/src/multipart_copy.rs b/crates/core/src/multipart_copy.rs new file mode 100644 index 00000000..a940b6ec --- /dev/null +++ b/crates/core/src/multipart_copy.rs @@ -0,0 +1,374 @@ +//! Pure planning and request types for multipart server-side copies. + +use std::collections::HashMap; +use std::sync::Arc; +use std::sync::atomic::{AtomicBool, Ordering}; + +use crate::{Error, ObjectInfo, Result}; +use tokio::sync::Notify; + +/// Largest object that can be copied with a single S3 CopyObject request. +pub const S3_SINGLE_COPY_MAX_SIZE: u64 = 5 * 1024 * 1024 * 1024; + +/// Smallest non-final part accepted by S3 multipart uploads. +pub const S3_MULTIPART_COPY_MIN_PART_SIZE: u64 = 5 * 1024 * 1024; + +/// Largest part accepted by S3 multipart uploads. +pub const S3_MULTIPART_COPY_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024; + +/// Largest number of parts accepted by S3 multipart uploads. +pub const S3_MULTIPART_COPY_MAX_PARTS: usize = 10_000; + +/// Largest object accepted by S3. +pub const S3_MAX_OBJECT_SIZE: u64 = 5 * 1024 * 1024 * 1024 * 1024; + +/// Default multipart copy part size. +pub const DEFAULT_MULTIPART_COPY_PART_SIZE: u64 = 64 * 1024 * 1024; + +/// Cooperative cancellation control for multipart server-side copies. +/// +/// Callers must signal this control rather than dropping the copy future when +/// they need the backend to await multipart upload cleanup. +#[derive(Debug, Clone, Default)] +pub struct MultipartCopyCancellation { + inner: Arc, +} + +#[derive(Debug, Default)] +struct MultipartCopyCancellationInner { + cancelled: AtomicBool, + notify: Notify, +} + +impl MultipartCopyCancellation { + /// Create a cancellation control in the active state. + pub fn new() -> Self { + Self::default() + } + + /// Request cooperative cancellation. + pub fn cancel(&self) { + if !self.inner.cancelled.swap(true, Ordering::AcqRel) { + self.inner.notify.notify_waiters(); + } + } + + /// Return whether cancellation has been requested. + pub fn is_cancelled(&self) -> bool { + self.inner.cancelled.load(Ordering::Acquire) + } + + /// Wait until cancellation is requested without losing an early signal. + pub async fn cancelled(&self) { + let notified = self.inner.notify.notified(); + tokio::pin!(notified); + // Register before reading the flag so cancellation cannot occur in the + // gap between the state check and awaiting the notification. + notified.as_mut().enable(); + if self.is_cancelled() { + return; + } + notified.await; + } +} + +/// A byte range assigned to one UploadPartCopy request. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MultipartCopyPart { + /// One-based S3 part number. + pub part_number: i32, + /// Inclusive first byte offset. + pub start: u64, + /// Inclusive last byte offset. + pub end_inclusive: u64, + /// Number of logical bytes in the part. + pub size: u64, +} + +/// A validated multipart server-side copy plan. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MultipartCopyPlan { + /// Total logical source size. + pub object_size: u64, + /// Planned size of every non-final part. + pub part_size: u64, + /// Ordered, gap-free byte ranges. + pub parts: Vec, +} + +impl MultipartCopyPlan { + /// Build a plan that satisfies S3 part size and count limits. + pub fn new(object_size: u64, preferred_part_size: Option) -> Result { + if object_size == 0 { + return Err(Error::InvalidPath( + "Multipart copy source size must be greater than zero".to_string(), + )); + } + if object_size > S3_MAX_OBJECT_SIZE { + return Err(Error::InvalidPath(format!( + "Multipart copy source size exceeds the S3 object limit of {S3_MAX_OBJECT_SIZE} bytes" + ))); + } + + let preferred_part_size = preferred_part_size.unwrap_or(DEFAULT_MULTIPART_COPY_PART_SIZE); + if !(S3_MULTIPART_COPY_MIN_PART_SIZE..=S3_MULTIPART_COPY_MAX_PART_SIZE) + .contains(&preferred_part_size) + { + return Err(Error::InvalidPath(format!( + "Multipart copy part size must be between {S3_MULTIPART_COPY_MIN_PART_SIZE} and {S3_MULTIPART_COPY_MAX_PART_SIZE} bytes" + ))); + } + + // Raising a caller's preferred size is necessary when its requested size + // would exceed S3's 10,000-part ceiling. + let minimum_size_for_part_limit = object_size.div_ceil(S3_MULTIPART_COPY_MAX_PARTS as u64); + let part_size = preferred_part_size + .max(minimum_size_for_part_limit) + .max(S3_MULTIPART_COPY_MIN_PART_SIZE); + if part_size > S3_MULTIPART_COPY_MAX_PART_SIZE { + return Err(Error::InvalidPath(format!( + "Multipart copy requires a part larger than the S3 limit of {S3_MULTIPART_COPY_MAX_PART_SIZE} bytes" + ))); + } + + let part_count = object_size.div_ceil(part_size); + if part_count > S3_MULTIPART_COPY_MAX_PARTS as u64 { + return Err(Error::InvalidPath(format!( + "Multipart copy requires {part_count} parts, exceeding the S3 limit of {S3_MULTIPART_COPY_MAX_PARTS}" + ))); + } + + let mut parts = Vec::with_capacity(part_count as usize); + let mut start = 0; + for part_index in 0..part_count { + let end_exclusive = (start + part_size).min(object_size); + parts.push(MultipartCopyPart { + part_number: i32::try_from(part_index + 1) + .expect("S3 multipart copy plans contain at most 10,000 parts"), + start, + end_inclusive: end_exclusive - 1, + size: end_exclusive - start, + }); + start = end_exclusive; + } + + Ok(Self { + object_size, + part_size, + parts, + }) + } +} + +/// Return whether S3 requires multipart server-side copy for this object size. +pub const fn requires_multipart_copy(object_size: u64) -> bool { + object_size > S3_SINGLE_COPY_MAX_SIZE +} + +/// Inputs required for a consistent multipart server-side copy. +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct MultipartCopyOptions { + /// Logical source size used to build part ranges. + pub source_size: u64, + /// ETag sent as `x-amz-copy-source-if-match` on every part. + pub source_etag: String, + /// Exact historical source version, when selected. + pub source_version_id: Option, + /// Optional preferred part size. + pub preferred_part_size: Option, + /// Destination content type inherited from the source. + pub content_type: Option, + /// Destination user metadata inherited from the source. + pub metadata: HashMap, +} + +impl MultipartCopyOptions { + /// Build options with source identity and safe defaults. + pub fn new(source_size: u64, source_etag: impl Into) -> Result { + let source_etag = source_etag.into(); + if source_etag.is_empty() { + return Err(Error::InvalidPath( + "Multipart copy source ETag cannot be empty".to_string(), + )); + } + Ok(Self { + source_size, + source_etag, + source_version_id: None, + preferred_part_size: None, + content_type: None, + metadata: HashMap::new(), + }) + } + + /// Validate the request and return its pure copy plan. + pub fn plan(&self) -> Result { + if self.source_etag.is_empty() { + return Err(Error::InvalidPath( + "Multipart copy source ETag cannot be empty".to_string(), + )); + } + if self.source_version_id.as_deref().is_some_and(str::is_empty) { + return Err(Error::InvalidPath( + "Source version ID cannot be empty".to_string(), + )); + } + MultipartCopyPlan::new(self.source_size, self.preferred_part_size) + } +} + +/// Successful multipart copy details. +#[derive(Debug, Clone)] +pub struct MultipartCopyResult { + /// Destination object metadata. + pub object: ObjectInfo, + /// Service upload identifier used for the completed operation. + pub upload_id: String, + /// Number of successfully completed parts. + pub part_count: usize, + /// Logical bytes copied. + pub bytes_copied: u64, +} + +/// Callback invoked with cumulative successfully copied logical bytes. +pub type MultipartCopyProgress<'a> = dyn Fn(u64) + Send + Sync + 'a; + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn copy_threshold_is_strictly_greater_than_five_gibibytes() { + assert!(!requires_multipart_copy(S3_SINGLE_COPY_MAX_SIZE)); + assert!(requires_multipart_copy(S3_SINGLE_COPY_MAX_SIZE + 1)); + } + + #[tokio::test] + async fn cancellation_control_remembers_early_and_late_signals() { + let early = MultipartCopyCancellation::new(); + early.cancel(); + early.cancel(); + early.cancelled().await; + assert!(early.is_cancelled()); + + let late = MultipartCopyCancellation::new(); + let waiter = late.clone(); + let task = tokio::spawn(async move { + waiter.cancelled().await; + }); + late.cancel(); + task.await.expect("cancellation waiter"); + } + + #[tokio::test] + async fn cancellation_control_handles_registration_races() { + for _ in 0..1_000 { + let cancellation = MultipartCopyCancellation::new(); + let waiter = cancellation.clone(); + let task = tokio::spawn(async move { + waiter.cancelled().await; + }); + tokio::task::yield_now().await; + cancellation.cancel(); + tokio::time::timeout(std::time::Duration::from_secs(1), task) + .await + .expect("cancellation waiter must not lose notification") + .expect("cancellation waiter task"); + } + } + + #[test] + fn plan_uses_inclusive_gap_free_ranges() { + let object_size = S3_MULTIPART_COPY_MIN_PART_SIZE * 2 + 1; + let plan = MultipartCopyPlan::new(object_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE)) + .expect("build multipart copy plan"); + + assert_eq!(plan.parts.len(), 3); + assert_eq!( + plan.parts, + vec![ + MultipartCopyPart { + part_number: 1, + start: 0, + end_inclusive: S3_MULTIPART_COPY_MIN_PART_SIZE - 1, + size: S3_MULTIPART_COPY_MIN_PART_SIZE, + }, + MultipartCopyPart { + part_number: 2, + start: S3_MULTIPART_COPY_MIN_PART_SIZE, + end_inclusive: S3_MULTIPART_COPY_MIN_PART_SIZE * 2 - 1, + size: S3_MULTIPART_COPY_MIN_PART_SIZE, + }, + MultipartCopyPart { + part_number: 3, + start: S3_MULTIPART_COPY_MIN_PART_SIZE * 2, + end_inclusive: object_size - 1, + size: 1, + }, + ] + ); + assert_eq!( + plan.parts.iter().map(|part| part.size).sum::(), + object_size + ); + } + + #[test] + fn plan_increases_part_size_to_stay_within_ten_thousand_parts() { + let object_size = + S3_MULTIPART_COPY_MIN_PART_SIZE * (S3_MULTIPART_COPY_MAX_PARTS as u64 + 1); + let plan = MultipartCopyPlan::new(object_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE)) + .expect("adjust undersized preferred part"); + + assert_eq!(plan.parts.len(), S3_MULTIPART_COPY_MAX_PARTS); + assert_eq!(plan.part_size, object_size.div_ceil(10_000)); + assert_eq!( + plan.parts.last().map(|part| part.end_inclusive), + Some(object_size - 1) + ); + } + + #[test] + fn plan_accepts_exact_s3_boundaries() { + let exact_max_parts_size = + S3_MULTIPART_COPY_MIN_PART_SIZE * S3_MULTIPART_COPY_MAX_PARTS as u64; + let plan = + MultipartCopyPlan::new(exact_max_parts_size, Some(S3_MULTIPART_COPY_MIN_PART_SIZE)) + .expect("accept exactly ten thousand parts"); + assert_eq!(plan.parts.len(), S3_MULTIPART_COPY_MAX_PARTS); + + let max_object = MultipartCopyPlan::new(S3_MAX_OBJECT_SIZE, None) + .expect("accept maximum S3 object size"); + assert!(max_object.parts.len() <= S3_MULTIPART_COPY_MAX_PARTS); + assert!(max_object.part_size <= S3_MULTIPART_COPY_MAX_PART_SIZE); + } + + #[test] + fn plan_rejects_zero_oversized_objects_and_invalid_part_sizes() { + for (object_size, part_size) in [ + (0, None), + (S3_MAX_OBJECT_SIZE + 1, None), + (1, Some(S3_MULTIPART_COPY_MIN_PART_SIZE - 1)), + (1, Some(S3_MULTIPART_COPY_MAX_PART_SIZE + 1)), + ] { + assert!( + MultipartCopyPlan::new(object_size, part_size).is_err(), + "object_size={object_size}, part_size={part_size:?}" + ); + } + } + + #[test] + fn options_reject_empty_source_identity() { + assert!(MultipartCopyOptions::new(1, "").is_err()); + + let mut options = + MultipartCopyOptions::new(1, "etag").expect("valid multipart copy options"); + options.source_etag.clear(); + assert!(options.plan().is_err()); + + options.source_etag = "etag".to_string(); + options.source_version_id = Some(String::new()); + assert!(options.plan().is_err()); + } +} diff --git a/crates/core/src/traits.rs b/crates/core/src/traits.rs index a7f38823..fc6f902e 100644 --- a/crates/core/src/traits.rs +++ b/crates/core/src/traits.rs @@ -14,6 +14,9 @@ use crate::cors::CorsRule; use crate::encryption::{BucketEncryption, ObjectEncryptionRequest}; use crate::error::{Error, Result}; use crate::lifecycle::LifecycleRule; +use crate::multipart_copy::{ + MultipartCopyCancellation, MultipartCopyOptions, MultipartCopyProgress, MultipartCopyResult, +}; use crate::object_lock::{ BucketObjectLockConfiguration, LegalHoldStatus, ObjectLockOptions, ObjectRetention, }; @@ -603,6 +606,24 @@ pub trait ObjectStore: Send + Sync { self.copy_object(src, dst, encryption).await } + /// Copy an object with S3's multipart server-side copy lifecycle. + /// + /// Backends that do not implement multipart copy fail explicitly. The + /// callback receives cumulative logical bytes after each successful part. + async fn multipart_copy( + &self, + _src: &RemotePath, + _dst: &RemotePath, + _options: &MultipartCopyOptions, + _cancellation: &MultipartCopyCancellation, + _encryption: Option<&ObjectEncryptionRequest>, + _on_progress: &MultipartCopyProgress<'_>, + ) -> Result { + Err(Error::UnsupportedFeature( + "Multipart server-side copy is not implemented by this object store".to_string(), + )) + } + /// Generate a presigned URL for an object async fn presign_get(&self, path: &RemotePath, expires_secs: u64) -> Result; diff --git a/crates/s3/src/client.rs b/crates/s3/src/client.rs index 16e1da05..5d45a2bd 100644 --- a/crates/s3/src/client.rs +++ b/crates/s3/src/client.rs @@ -33,13 +33,14 @@ use rc_core::{ Alias, BucketEncryption, BucketNotification, BucketObjectLockConfiguration, Capabilities, CopyObjectOptions, CorsRule, CreateBucketOptions, DefaultRetention, DeleteObjectFailure, DeleteObjectsResult, DeletedObject, Error, LegalHoldStatus, LifecycleRule, - ListObjectVersionsOptions, ListOptions, ListResult, NotificationTarget, - ObjectEncryptionRequest, ObjectInfo, ObjectLockOptions, ObjectReadOptions, ObjectRetention, - ObjectStore, ObjectVersion, ObjectVersionIdentifier, ObjectVersionListResult, RemotePath, - ReplicationConfiguration, ReplicationResyncStartOptions, ReplicationResyncStartResult, - ReplicationResyncState, ReplicationResyncStatus, ReplicationResyncTargetStatus, RequestHeader, - Result, RetentionDuration, RetentionDurationUnit, RetentionMode, SelectOptions, - global_request_headers, + ListObjectVersionsOptions, ListOptions, ListResult, MultipartAbortStatus, + MultipartCopyCancellation, MultipartCopyOptions, MultipartCopyProgress, MultipartCopyResult, + NotificationTarget, ObjectEncryptionRequest, ObjectInfo, ObjectLockOptions, ObjectReadOptions, + ObjectRetention, ObjectStore, ObjectVersion, ObjectVersionIdentifier, ObjectVersionListResult, + RemotePath, ReplicationConfiguration, ReplicationResyncStartOptions, + ReplicationResyncStartResult, ReplicationResyncState, ReplicationResyncStatus, + ReplicationResyncTargetStatus, RequestHeader, Result, RetentionDuration, RetentionDurationUnit, + RetentionMode, SelectOptions, global_request_headers, }; use reqwest::Method; use reqwest::header::{CONTENT_TYPE, HeaderMap, HeaderName, HeaderValue}; @@ -598,6 +599,21 @@ fn apply_object_encryption_to_copy_request( } } +fn apply_object_encryption_to_multipart_create_request( + request: aws_sdk_s3::operation::create_multipart_upload::builders::CreateMultipartUploadFluentBuilder, + encryption: Option<&ObjectEncryptionRequest>, +) -> aws_sdk_s3::operation::create_multipart_upload::builders::CreateMultipartUploadFluentBuilder { + match encryption { + Some(ObjectEncryptionRequest::SseS3) => { + request.server_side_encryption(aws_sdk_s3::types::ServerSideEncryption::Aes256) + } + Some(ObjectEncryptionRequest::SseKms { key_id }) => request + .server_side_encryption(aws_sdk_s3::types::ServerSideEncryption::AwsKms) + .ssekms_key_id(key_id), + None => request, + } +} + fn encoded_copy_source(src: &RemotePath, source_version_id: Option<&str>) -> String { let encoded_key = src .key @@ -615,6 +631,10 @@ fn encoded_copy_source(src: &RemotePath, source_version_id: Option<&str>) -> Str copy_source } +fn quoted_etag(etag: &str) -> String { + format!("\"{}\"", etag.trim_matches('"')) +} + fn sdk_retention_mode(mode: RetentionMode) -> aws_sdk_s3::types::ObjectLockRetentionMode { match mode { RetentionMode::Governance => aws_sdk_s3::types::ObjectLockRetentionMode::Governance, @@ -1851,6 +1871,14 @@ impl S3Client { .and_then(|value| std::str::from_utf8(value.as_bytes()).ok()) }); let status = raw.status().as_u16(); + if matches!(status, 409 | 412) + || matches!( + code, + Some("ConditionalRequestConflict") | Some("PreconditionFailed") + ) + { + return Error::Conflict(formatted); + } if status == 401 || matches!(code, Some("Unauthorized")) { return Error::Auth(formatted); } @@ -2004,6 +2032,10 @@ impl S3Client { } Error::General(message) => Error::General(self.redact_sensitive_text(message)), Error::NotFound(message) => Error::NotFound(self.redact_sensitive_text(message)), + Error::RequestRejected(message) => { + Error::RequestRejected(self.redact_sensitive_text(message)) + } + Error::Interrupted(message) => Error::Interrupted(self.redact_sensitive_text(message)), other => other, } } @@ -2746,15 +2778,35 @@ impl S3Client { Ok(info) } - async fn abort_multipart_upload_best_effort(&self, path: &RemotePath, upload_id: &str) { - let _ = self + async fn abort_multipart_upload_best_effort( + &self, + path: &RemotePath, + upload_id: &str, + ) -> MultipartAbortStatus { + match self .inner .abort_multipart_upload() .bucket(&path.bucket) .key(&path.key) .upload_id(upload_id) .send() + .await + { + Ok(_) => MultipartAbortStatus::Succeeded, + Err(_) => MultipartAbortStatus::Failed, + } + } + + async fn abort_multipart_copy_with_error( + &self, + path: &RemotePath, + upload_id: &str, + error: Error, + ) -> Error { + let abort_status = self + .abort_multipart_upload_best_effort(path, upload_id) .await; + self.redact_sensitive_error(error.with_multipart_copy_context(upload_id, abort_status)) } async fn put_object_multipart_from_path( @@ -3638,6 +3690,181 @@ impl ObjectStore for S3Client { Ok(result) } + async fn multipart_copy( + &self, + src: &RemotePath, + dst: &RemotePath, + options: &MultipartCopyOptions, + cancellation: &MultipartCopyCancellation, + encryption: Option<&ObjectEncryptionRequest>, + on_progress: &MultipartCopyProgress<'_>, + ) -> Result { + use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart}; + + let plan = options + .plan() + .map_err(|error| self.redact_sensitive_error(error))?; + let copy_source = encoded_copy_source(src, options.source_version_id.as_deref()); + let source_etag = quoted_etag(&options.source_etag); + let metadata = (!options.metadata.is_empty()).then(|| options.metadata.clone()); + + let mut create_request = apply_object_encryption_to_multipart_create_request( + self.inner + .create_multipart_upload() + .bucket(&dst.bucket) + .key(&dst.key) + .set_metadata(metadata), + encryption, + ); + if let Some(content_type) = &options.content_type { + create_request = create_request.content_type(content_type); + } + let create_response = create_request.send().await.map_err(|error| { + self.redact_sensitive_error(Self::map_object_request_error(&error, dst, None)) + })?; + let upload_id = create_response + .upload_id() + .ok_or_else(|| { + self.redact_sensitive_error(Error::General( + "Multipart copy create response did not include an upload ID".to_string(), + )) + })? + .to_string(); + + let mut completed_parts = Vec::with_capacity(plan.parts.len()); + let mut bytes_copied = 0_u64; + let mut service_source_version = None; + + for part in &plan.parts { + let copy_request = self + .inner + .upload_part_copy() + .bucket(&dst.bucket) + .key(&dst.key) + .upload_id(&upload_id) + .part_number(part.part_number) + .copy_source(©_source) + .copy_source_range(format!("bytes={}-{}", part.start, part.end_inclusive)) + .copy_source_if_match(&source_etag) + .send(); + let copy_response = tokio::select! { + biased; + _ = cancellation.cancelled() => { + return Err(self + .abort_multipart_copy_with_error( + dst, + &upload_id, + Error::Interrupted("Multipart copy was cancelled".to_string()), + ) + .await); + } + response = copy_request => response, + }; + + let copy_response = match copy_response { + Ok(response) => response, + Err(error) => { + let primary = Self::map_object_request_error( + &error, + src, + options.source_version_id.as_deref(), + ); + return Err(self + .abort_multipart_copy_with_error(dst, &upload_id, primary) + .await); + } + }; + + if service_source_version.is_none() { + service_source_version = copy_response + .copy_source_version_id() + .map(ToString::to_string); + } + let etag = match copy_response + .copy_part_result() + .and_then(|result| result.e_tag()) + { + Some(etag) => etag.trim_matches('"').to_string(), + None => { + return Err(self + .abort_multipart_copy_with_error( + dst, + &upload_id, + Error::General(format!( + "UploadPartCopy response for part {} did not include an ETag", + part.part_number + )), + ) + .await); + } + }; + completed_parts.push( + CompletedPart::builder() + .part_number(part.part_number) + .e_tag(etag) + .build(), + ); + + bytes_copied += part.size; + on_progress(bytes_copied); + } + + let completed_upload = CompletedMultipartUpload::builder() + .set_parts(Some(completed_parts)) + .build(); + let complete_request = self + .inner + .complete_multipart_upload() + .bucket(&dst.bucket) + .key(&dst.key) + .upload_id(&upload_id) + .multipart_upload(completed_upload) + .send(); + let complete_response = tokio::select! { + biased; + // A successful completion is irreversible. Prefer an already-ready + // service response over a simultaneous cancellation signal so the + // caller is not told that a completed destination was interrupted. + response = complete_request => response, + _ = cancellation.cancelled() => { + return Err(self + .abort_multipart_copy_with_error( + dst, + &upload_id, + Error::Interrupted("Multipart copy was cancelled".to_string()), + ) + .await); + } + }; + let complete_response = match complete_response { + Ok(response) => response, + Err(error) => { + let primary = Self::map_object_request_error(&error, dst, None); + return Err(self + .abort_multipart_copy_with_error(dst, &upload_id, primary) + .await); + } + }; + + let mut object = ObjectInfo::file(&dst.key, plan.object_size as i64); + object.etag = complete_response + .e_tag() + .map(|etag| etag.trim_matches('"').to_string()); + object.version_id = complete_response.version_id().map(ToString::to_string); + object.source_version_id = + service_source_version.or_else(|| options.source_version_id.clone()); + object.content_type = options.content_type.clone(); + object.metadata = (!options.metadata.is_empty()).then(|| options.metadata.clone()); + object.last_modified = Some(jiff::Timestamp::now()); + + Ok(MultipartCopyResult { + object, + upload_id, + part_count: plan.parts.len(), + bytes_copied, + }) + } + async fn presign_get(&self, path: &RemotePath, expires_secs: u64) -> Result { let config = aws_sdk_s3::presigning::PresigningConfig::builder() .expires_in(std::time::Duration::from_secs(expires_secs)) @@ -4876,12 +5103,17 @@ mod tests { use aws_smithy_http_client::test_util::{ CaptureRequestReceiver, ReplayEvent, StaticReplayClient, capture_request, }; + use rc_core::S3_MULTIPART_COPY_MIN_PART_SIZE; use std::collections::HashMap; + use std::collections::VecDeque; use std::io::{Read, Write}; use std::net::{TcpListener, TcpStream}; + use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::mpsc; + use std::sync::{Arc, Mutex}; use std::thread; use std::time::{Duration, Instant}; + use tokio::sync::Notify; #[derive(Debug)] struct CapturedXmlRequest { @@ -4953,6 +5185,13 @@ mod tests { fn test_s3_client_with_response_sequence( responses: Vec>, + ) -> (S3Client, StaticReplayClient) { + test_s3_client_with_response_sequence_and_headers(responses, Vec::new()) + } + + fn test_s3_client_with_response_sequence_and_headers( + responses: Vec>, + request_headers: Vec, ) -> (S3Client, StaticReplayClient) { let events = responses .into_iter() @@ -4973,6 +5212,141 @@ mod tests { None, "rc-test-credentials", ); + let mut config_builder = aws_sdk_s3::config::Builder::new() + .credentials_provider(credentials) + .endpoint_url("https://example.com") + .region(aws_sdk_s3::config::Region::new("us-east-1")) + .force_path_style(true) + .retry_config(aws_smithy_types::retry::RetryConfig::disabled()) + .behavior_version_latest() + .http_client(replay.clone()); + let presign_config = config_builder.clone().build(); + if !request_headers.is_empty() { + config_builder = config_builder.interceptor(CustomHeaderInterceptor { + headers: request_headers.clone(), + }); + } + let config = config_builder.build(); + let alias = Alias::new("test", "https://example.com", "access-key", "secret-key"); + let client = S3Client { + inner: aws_sdk_s3::Client::from_conf(config), + presign_inner: aws_sdk_s3::Client::from_conf(presign_config), + xml_http_client: reqwest::Client::new(), + alias, + request_headers, + }; + (client, replay) + } + + #[derive(Debug)] + enum BlockingReplayEvent { + Response(Box), + Pending, + } + + #[derive(Debug, Clone)] + struct BlockingReplayClient { + events: Arc>>, + requests: Arc>>, + pending_started: Arc, + pending_notify: Arc, + } + + impl BlockingReplayClient { + fn new( + create_response: http::Response, + abort_response: http::Response, + ) -> Self { + let events = VecDeque::from([ + BlockingReplayEvent::Response(Box::new( + create_response.try_into().expect("valid create response"), + )), + BlockingReplayEvent::Pending, + BlockingReplayEvent::Response(Box::new( + abort_response.try_into().expect("valid abort response"), + )), + ]); + Self { + events: Arc::new(Mutex::new(events)), + requests: Arc::new(Mutex::new(Vec::new())), + pending_started: Arc::new(AtomicBool::new(false)), + pending_notify: Arc::new(Notify::new()), + } + } + + async fn wait_for_pending_request(&self) { + let notified = self.pending_notify.notified(); + tokio::pin!(notified); + notified.as_mut().enable(); + if self.pending_started.load(Ordering::Acquire) { + return; + } + notified.await; + } + + fn requests(&self) -> Vec<(String, String)> { + self.requests.lock().expect("request lock").clone() + } + } + + impl HttpConnector for BlockingReplayClient { + fn call(&self, request: HttpRequest) -> HttpConnectorFuture { + self.requests + .lock() + .expect("request lock") + .push((request.method().to_string(), request.uri().to_string())); + let event = self.events.lock().expect("event lock").pop_front(); + match event { + Some(BlockingReplayEvent::Response(response)) => { + HttpConnectorFuture::ready(Ok(*response)) + } + Some(BlockingReplayEvent::Pending) => { + self.pending_started.store(true, Ordering::Release); + self.pending_notify.notify_waiters(); + HttpConnectorFuture::new(async { + futures::future::pending::< + std::result::Result< + aws_smithy_runtime_api::client::orchestrator::HttpResponse, + ConnectorError, + >, + >() + .await + }) + } + None => HttpConnectorFuture::ready(Err(ConnectorError::other( + "BlockingReplayClient has no remaining response".into(), + None, + ))), + } + } + } + + impl HttpClient for BlockingReplayClient { + fn http_connector( + &self, + _settings: &HttpConnectorSettings, + _runtime_components: &RuntimeComponents, + ) -> SharedHttpConnector { + SharedHttpConnector::new(self.clone()) + } + } + + fn test_s3_client_with_pending_part() -> (S3Client, BlockingReplayClient) { + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let replay = BlockingReplayClient::new( + multipart_copy_create_response("cancel-upload-id"), + abort_response, + ); + let credentials = Credentials::new( + "access-key", + "secret-key", + None, + None, + "rc-test-credentials", + ); let config = aws_sdk_s3::config::Builder::new() .credentials_provider(credentials) .endpoint_url("https://example.com") @@ -7454,6 +7828,566 @@ mod tests { )); } + fn multipart_copy_create_response(upload_id: &str) -> http::Response { + http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from(format!( + r#"destination-bucketdst.txt{upload_id}"# + ))) + .expect("build multipart copy create response") + } + + fn multipart_copy_part_response( + etag: &str, + source_version_id: Option<&str>, + ) -> http::Response { + let mut builder = http::Response::builder() + .status(200) + .header("content-type", "application/xml"); + if let Some(source_version_id) = source_version_id { + builder = builder.header("x-amz-copy-source-version-id", source_version_id); + } + builder + .body(SdkBody::from(format!( + r#"2026-07-23T00:00:00Z"{etag}""# + ))) + .expect("build multipart copy part response") + } + + fn multipart_copy_complete_response() -> http::Response { + http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .header("x-amz-version-id", "destination-v2") + .body(SdkBody::from( + r#"destination-bucketdst.txt"complete-etag""#, + )) + .expect("build multipart copy complete response") + } + + fn multipart_copy_options(source_size: u64) -> MultipartCopyOptions { + let mut options = MultipartCopyOptions::new(source_size, "source-etag") + .expect("valid multipart copy options"); + options.source_version_id = Some("source v1+/?".to_string()); + options.preferred_part_size = Some(S3_MULTIPART_COPY_MIN_PART_SIZE); + options.content_type = Some("application/octet-stream".to_string()); + options + .metadata + .insert("owner".to_string(), "copy-test".to_string()); + options + } + + #[tokio::test] + async fn multipart_copy_sends_encoded_versioned_ranges_and_reports_progress() { + let source_size = S3_MULTIPART_COPY_MIN_PART_SIZE + 1; + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("copy-upload-id"), + multipart_copy_part_response("part-1", Some("source v1+/?")), + multipart_copy_part_response("part-2", Some("source v1+/?")), + multipart_copy_complete_response(), + ]); + let src = RemotePath::new("test", "source-bucket", "dir one/a+b?#.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.txt"); + let options = multipart_copy_options(source_size); + let progress = std::sync::Mutex::new(Vec::new()); + + let result = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + Some(&ObjectEncryptionRequest::SseKms { + key_id: "kms-key".to_string(), + }), + &|bytes| progress.lock().expect("progress lock").push(bytes), + ) + .await + .expect("multipart server-side copy"); + + assert_eq!(result.upload_id, "copy-upload-id"); + assert_eq!(result.part_count, 2); + assert_eq!(result.bytes_copied, source_size); + assert_eq!(result.object.size_bytes, Some(source_size as i64)); + assert_eq!(result.object.version_id.as_deref(), Some("destination-v2")); + assert_eq!( + result.object.source_version_id.as_deref(), + Some("source v1+/?") + ); + assert_eq!( + *progress.lock().expect("progress lock"), + vec![S3_MULTIPART_COPY_MIN_PART_SIZE, source_size] + ); + + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 4); + assert_eq!(requests[0].method(), "POST"); + assert!(requests[0].uri().ends_with("?uploads")); + assert_eq!( + requests[0].headers().get("content-type"), + Some("application/octet-stream") + ); + assert_eq!( + requests[0].headers().get("x-amz-meta-owner"), + Some("copy-test") + ); + assert_eq!( + requests[0].headers().get("x-amz-server-side-encryption"), + Some("aws:kms") + ); + assert_eq!( + requests[0] + .headers() + .get("x-amz-server-side-encryption-aws-kms-key-id"), + Some("kms-key") + ); + + for request in &requests[1..=2] { + assert_eq!(request.method(), "PUT"); + assert_eq!( + request.headers().get("x-amz-copy-source"), + Some("source-bucket/dir%20one/a%2Bb%3F%23.bin?versionId=source%20v1%2B%2F%3F") + ); + assert_eq!( + request.headers().get("x-amz-copy-source-if-match"), + Some("\"source-etag\"") + ); + } + assert_eq!( + requests[1].headers().get("x-amz-copy-source-range"), + Some("bytes=0-5242879") + ); + assert_eq!( + requests[2].headers().get("x-amz-copy-source-range"), + Some("bytes=5242880-5242880") + ); + assert_eq!(requests[3].method(), "POST"); + assert!(requests[3].uri().contains("uploadId=copy-upload-id")); + let completion_body = requests[3].body().bytes().expect("completion request body"); + let completion_body = std::str::from_utf8(completion_body).expect("completion body is XML"); + assert!(completion_body.contains("part-1")); + assert!(completion_body.contains("part-2")); + } + + #[tokio::test] + async fn multipart_copy_part_failure_aborts_once_and_preserves_missing_version() { + let missing_version_response = http::Response::builder() + .status(404) + .header("content-type", "application/xml") + .header("x-amz-error-code", "NoSuchVersion") + .body(SdkBody::from( + "NoSuchVersionmissing", + )) + .expect("build missing version response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("failed-upload-id"), + missing_version_response, + abort_response, + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("missing source version should fail"); + + assert!(matches!( + error, + Error::VersionNotFound { + path, + version_id, + } if version_id == "source v1+/?" + && path.contains("failed-upload-id") + && path.contains("abort: succeeded") + )); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert_eq!( + requests + .iter() + .filter(|request| request.method() == "DELETE") + .count(), + 1 + ); + } + + #[tokio::test] + async fn multipart_copy_access_denial_stays_typed_after_abort() { + let denied_response = http::Response::builder() + .status(403) + .header("content-type", "application/xml") + .header("x-amz-error-code", "AccessDenied") + .body(SdkBody::from("AccessDeniedaccess-key secret-key custom-header-secret")) + .expect("build access denied response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let (client, _) = test_s3_client_with_response_sequence_and_headers( + vec![ + multipart_copy_create_response("denied-upload-id"), + denied_response, + abort_response, + ], + vec![RequestHeader { + name: "x-test-secret".to_string(), + value: "custom-header-secret".to_string(), + }], + ); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("access denial should fail"); + + assert_eq!(error.exit_code(), 4); + let display = error.to_string(); + assert!(matches!(error, Error::Auth(_))); + assert!(display.contains("denied-upload-id")); + for secret in ["access-key", "secret-key", "custom-header-secret"] { + assert!(!display.contains(secret), "{display}"); + } + } + + #[tokio::test] + async fn multipart_copy_create_failure_redacts_all_configured_secrets() { + let missing_bucket_response = http::Response::builder() + .status(404) + .header("content-type", "application/xml") + .header("x-amz-error-code", "NoSuchBucket") + .body(SdkBody::from( + "NoSuchBucketmissing", + )) + .expect("build missing bucket response"); + let (client, replay) = test_s3_client_with_response_sequence_and_headers( + vec![missing_bucket_response], + vec![RequestHeader { + name: "x-test-secret".to_string(), + value: "custom-header-secret".to_string(), + }], + ); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new( + "test", + "access-key-secret-key-custom-header-secret", + "dst.bin", + ); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("missing destination bucket should fail"); + let display = error.to_string(); + + assert!(matches!(error, Error::NotFound(_))); + assert!(display.contains("[REDACTED]"), "{display}"); + for secret in ["access-key", "secret-key", "custom-header-secret"] { + assert!(!display.contains(secret), "{display}"); + } + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 1); + assert_eq!(requests[0].method(), "POST"); + } + + #[tokio::test] + async fn multipart_copy_part_precondition_failure_aborts_once_as_conflict() { + let conflict_response = http::Response::builder() + .status(412) + .header("content-type", "application/xml") + .header("x-amz-error-code", "PreconditionFailed") + .body(SdkBody::from( + "PreconditionFailedsource changed", + )) + .expect("build precondition response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("conflict-upload-id"), + conflict_response, + abort_response, + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("source precondition should fail"); + + assert_eq!(error.exit_code(), 6); + assert!(matches!(error, Error::Conflict(_))); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert_eq!( + requests + .iter() + .filter(|request| request.method() == "DELETE") + .count(), + 1 + ); + } + + #[tokio::test] + async fn multipart_copy_missing_part_etag_aborts_once() { + let missing_etag_response = http::Response::builder() + .status(200) + .header("content-type", "application/xml") + .body(SdkBody::from( + r#"2026-07-23T00:00:00Z"#, + )) + .expect("build missing ETag response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("missing-etag-upload-id"), + missing_etag_response, + abort_response, + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("missing ETag should fail"); + + assert!(matches!(error, Error::General(_))); + assert!(error.to_string().contains("missing-etag-upload-id")); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert_eq!( + requests + .iter() + .filter(|request| request.method() == "DELETE") + .count(), + 1 + ); + } + + #[tokio::test] + async fn multipart_copy_cooperative_cancellation_aborts_pending_part_once() { + let (client, replay) = test_s3_client_with_pending_part(); + let replay_for_wait = replay.clone(); + let cancellation = MultipartCopyCancellation::new(); + let cancellation_for_copy = cancellation.clone(); + let copy = tokio::spawn(async move { + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + client + .multipart_copy(&src, &dst, &options, &cancellation_for_copy, None, &|_| {}) + .await + }); + + tokio::time::timeout( + Duration::from_secs(2), + replay_for_wait.wait_for_pending_request(), + ) + .await + .expect("part request should become pending"); + cancellation.cancel(); + let error = tokio::time::timeout(Duration::from_secs(2), copy) + .await + .expect("cooperative cancellation must await abort") + .expect("copy task") + .expect_err("copy should be interrupted"); + + assert_eq!(error.exit_code(), 130); + assert!(matches!(error, Error::Interrupted(_))); + assert!(error.to_string().contains("cancel-upload-id")); + assert!(error.to_string().contains("abort: succeeded")); + let requests = replay.requests(); + assert_eq!(requests.len(), 3, "{requests:?}"); + assert_eq!( + requests + .iter() + .filter(|(method, _)| method == "DELETE") + .count(), + 1 + ); + assert!(requests[1].1.contains("partNumber=1")); + assert!(requests[2].1.contains("uploadId=cancel-upload-id")); + } + + #[tokio::test] + async fn multipart_copy_ready_completion_wins_simultaneous_cancellation() { + let source_size = S3_MULTIPART_COPY_MIN_PART_SIZE; + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("complete-race-id"), + multipart_copy_part_response("part-1", None), + multipart_copy_complete_response(), + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(source_size); + let cancellation = MultipartCopyCancellation::new(); + + let result = client + .multipart_copy(&src, &dst, &options, &cancellation, None, &|bytes| { + assert_eq!(bytes, source_size); + cancellation.cancel(); + }) + .await + .expect("ready completion should win cancellation"); + + assert_eq!(result.bytes_copied, source_size); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 3); + assert_eq!( + requests + .iter() + .filter(|request| request.method() == "DELETE") + .count(), + 0 + ); + } + + #[tokio::test] + async fn multipart_copy_complete_failure_aborts_exactly_once() { + let complete_failure = http::Response::builder() + .status(500) + .header("content-type", "application/xml") + .header("x-amz-error-code", "InternalError") + .body(SdkBody::from( + "InternalErrorcomplete failed", + )) + .expect("build complete failure response"); + let abort_response = http::Response::builder() + .status(204) + .body(SdkBody::empty()) + .expect("build abort response"); + let (client, replay) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("complete-failure-id"), + multipart_copy_part_response("part-1", None), + complete_failure, + abort_response, + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("completion failure should abort"); + + assert!(error.to_string().contains("complete-failure-id")); + assert!(error.to_string().contains("abort: succeeded")); + let requests = replay.actual_requests().collect::>(); + assert_eq!(requests.len(), 4); + assert_eq!( + requests + .iter() + .filter(|request| request.method() == "DELETE") + .count(), + 1 + ); + } + + #[tokio::test] + async fn multipart_copy_abort_failure_is_reported_without_service_details() { + let part_failure = http::Response::builder() + .status(500) + .header("content-type", "application/xml") + .header("x-amz-error-code", "InternalError") + .body(SdkBody::from( + "InternalErrorpart failed", + )) + .expect("build part failure response"); + let abort_failure = http::Response::builder() + .status(500) + .header("content-type", "application/xml") + .body(SdkBody::from( + "InternalErrorSECRET_ABORT_DETAIL", + )) + .expect("build abort failure response"); + let (client, _) = test_s3_client_with_response_sequence(vec![ + multipart_copy_create_response("abort-failure-id"), + part_failure, + abort_failure, + ]); + let src = RemotePath::new("test", "source-bucket", "src.bin"); + let dst = RemotePath::new("test", "destination-bucket", "dst.bin"); + let options = multipart_copy_options(S3_MULTIPART_COPY_MIN_PART_SIZE); + + let error = client + .multipart_copy( + &src, + &dst, + &options, + &MultipartCopyCancellation::new(), + None, + &|_| {}, + ) + .await + .expect_err("part and abort failures should be reported"); + let display = error.to_string(); + + assert!(matches!(error, Error::Network(_))); + assert!(display.contains("abort-failure-id")); + assert!(display.contains("abort: failed")); + assert!(!display.contains("SECRET_ABORT_DETAIL")); + } + #[tokio::test] async fn delete_object_wrapper_uses_default_options_without_rustfs_header() { let (client, request_receiver) = test_s3_client(None);