From acc8cc41b291f4eb866001066da1672182a449dc Mon Sep 17 00:00:00 2001 From: James Smith Date: Tue, 2 Dec 2025 14:54:43 +1100 Subject: [PATCH 1/8] AWS: support copies >5GB using multipart copy --- src/aws/builder.rs | 59 +++++++++++++++++++++ src/aws/client.rs | 24 ++++++++- src/aws/mod.rs | 124 +++++++++++++++++++++++++++++---------------- src/config.rs | 9 ++++ 4 files changed, 172 insertions(+), 44 deletions(-) diff --git a/src/aws/builder.rs b/src/aws/builder.rs index e49145a4..71752c46 100644 --- a/src/aws/builder.rs +++ b/src/aws/builder.rs @@ -42,6 +42,10 @@ use url::Url; /// Default metadata endpoint static DEFAULT_METADATA_ENDPOINT: &str = "http://169.254.169.254"; +/// AWS S3 does not support copy operations larger than 5 GiB in a single request. +/// https://docs.aws.amazon.com/AmazonS3/latest/userguide/copy-object.html +const MAX_SINGLE_REQUEST_COPY_SIZE: u64 = 5 * 1024 * 1024 * 1024; + /// A specialized `Error` for object store-related errors #[derive(Debug, thiserror::Error)] enum Error { @@ -189,6 +193,10 @@ pub struct AmazonS3Builder { request_payer: ConfigValue, /// The [`HttpConnector`] to use http_connector: Option>, + /// Threshold (bytes) above which copy uses multipart copy. If not set, defaults to 5 GiB. + multipart_copy_threshold: Option>, + /// Preferred multipart copy part size (bytes). If not set, defaults to 5 GiB. + multipart_copy_part_size: Option>, } /// Configuration keys for [`AmazonS3Builder`] @@ -423,6 +431,10 @@ pub enum AmazonS3ConfigKey { /// Encryption options Encryption(S3EncryptionConfigKey), + /// Threshold (bytes) to switch to multipart copy + MultipartCopyThreshold, + /// Preferred multipart copy part size (bytes) + MultipartCopyPartSize, } impl AsRef for AmazonS3ConfigKey { @@ -455,6 +467,8 @@ impl AsRef for AmazonS3ConfigKey { Self::RequestPayer => "aws_request_payer", Self::Client(opt) => opt.as_ref(), Self::Encryption(opt) => opt.as_ref(), + Self::MultipartCopyThreshold => "aws_multipart_copy_threshold", + Self::MultipartCopyPartSize => "aws_multipart_copy_part_size", } } } @@ -499,6 +513,12 @@ impl FromStr for AmazonS3ConfigKey { "aws_conditional_put" | "conditional_put" => Ok(Self::ConditionalPut), "aws_disable_tagging" | "disable_tagging" => Ok(Self::DisableTagging), "aws_request_payer" | "request_payer" => Ok(Self::RequestPayer), + "aws_multipart_copy_threshold" | "multipart_copy_threshold" => { + Ok(Self::MultipartCopyThreshold) + } + "aws_multipart_copy_part_size" | "multipart_copy_part_size" => { + Ok(Self::MultipartCopyPartSize) + } // Backwards compatibility "aws_allow_http" => Ok(Self::Client(ClientConfigKey::AllowHttp)), "aws_server_side_encryption" | "server_side_encryption" => Ok(Self::Encryption( @@ -666,6 +686,12 @@ impl AmazonS3Builder { self.encryption_customer_key_base64 = Some(value.into()) } }, + AmazonS3ConfigKey::MultipartCopyThreshold => { + self.multipart_copy_threshold = Some(ConfigValue::Deferred(value.into())) + } + AmazonS3ConfigKey::MultipartCopyPartSize => { + self.multipart_copy_part_size = Some(ConfigValue::Deferred(value.into())) + } }; self } @@ -733,6 +759,14 @@ impl AmazonS3Builder { self.encryption_customer_key_base64.clone() } }, + AmazonS3ConfigKey::MultipartCopyThreshold => self + .multipart_copy_threshold + .as_ref() + .map(|x| x.to_string()), + AmazonS3ConfigKey::MultipartCopyPartSize => self + .multipart_copy_part_size + .as_ref() + .map(|x| x.to_string()), } } @@ -1029,6 +1063,18 @@ impl AmazonS3Builder { self } + /// Set threshold (bytes) above which copy uses multipart copy + pub fn with_multipart_copy_threshold(mut self, threshold_bytes: u64) -> Self { + self.multipart_copy_threshold = Some(ConfigValue::Parsed(threshold_bytes)); + self + } + + /// Set preferred multipart copy part size (bytes) + pub fn with_multipart_copy_part_size(mut self, part_size_bytes: u64) -> Self { + self.multipart_copy_part_size = Some(ConfigValue::Parsed(part_size_bytes)); + self + } + /// Create a [`AmazonS3`] instance from the provided values, /// consuming `self`. pub fn build(mut self) -> Result { @@ -1185,6 +1231,17 @@ impl AmazonS3Builder { S3EncryptionHeaders::default() }; + let multipart_copy_threshold = self + .multipart_copy_threshold + .map(|val| val.get()) + .transpose()? + .unwrap_or(MAX_SINGLE_REQUEST_COPY_SIZE); + let multipart_copy_part_size = self + .multipart_copy_part_size + .map(|val| val.get()) + .transpose()? + .unwrap_or(MAX_SINGLE_REQUEST_COPY_SIZE); + let config = S3Config { region, bucket, @@ -1201,6 +1258,8 @@ impl AmazonS3Builder { conditional_put: self.conditional_put.get()?, encryption_headers, request_payer: self.request_payer.get()?, + multipart_copy_threshold, + multipart_copy_part_size, }; let http_client = http.connect(&config.client_options)?; diff --git a/src/aws/client.rs b/src/aws/client.rs index 63371875..a45d13c8 100644 --- a/src/aws/client.rs +++ b/src/aws/client.rs @@ -138,6 +138,7 @@ impl From for crate::Error { pub(crate) enum PutPartPayload<'a> { Part(PutPayload), Copy(&'a Path), + CopyRange(&'a Path, std::ops::Range), } impl Default for PutPartPayload<'_> { @@ -207,6 +208,10 @@ pub(crate) struct S3Config { pub conditional_put: S3ConditionalPut, pub request_payer: bool, pub(super) encryption_headers: S3EncryptionHeaders, + /// Threshold in bytes above which copy will use multipart copy + pub multipart_copy_threshold: u64, + /// Preferred multipart copy part size in bytes (None => auto) + pub multipart_copy_part_size: u64, } impl S3Config { @@ -676,7 +681,10 @@ impl S3Client { part_idx: usize, data: PutPartPayload<'_>, ) -> Result { - let is_copy = matches!(data, PutPartPayload::Copy(_)); + let is_copy = matches!( + data, + PutPartPayload::Copy(_) | PutPartPayload::CopyRange(_, _) + ); let part = (part_idx + 1).to_string(); let mut request = self @@ -690,6 +698,18 @@ impl S3Client { "x-amz-copy-source", &format!("{}/{}", self.config.bucket, encode_path(path)), ), + PutPartPayload::CopyRange(path, range) => { + // AWS expects inclusive end for copy range header + let start = range.start; + let end_inclusive = range.end.saturating_sub(1); + let range_value = format!("bytes={}-{}", start, end_inclusive); + request + .header( + "x-amz-copy-source", + &format!("{}/{}", self.config.bucket, encode_path(path)), + ) + .header("x-amz-copy-source-range", &range_value) + } }; if self @@ -995,6 +1015,8 @@ mod tests { conditional_put: Default::default(), encryption_headers: Default::default(), request_payer: false, + multipart_copy_threshold: 5 * 1024 * 1024 * 1024, + multipart_copy_part_size: 5 * 1024 * 1024 * 1024, }; let client = S3Client::new(config, HttpClient::new(reqwest::Client::new())); diff --git a/src/aws/mod.rs b/src/aws/mod.rs index 3e658afc..8966c0d6 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -101,6 +101,56 @@ impl AmazonS3 { fn path_url(&self, path: &Path) -> String { self.client.config.path_url(path) } + + /// Perform a multipart copy operation + async fn copy_multipart( + &self, + from: &Path, + to: &Path, + size: u64, + mode: CompleteMultipartMode, + ) -> Result<()> { + // Perform multipart copy using UploadPartCopy + let upload_id = self + .client + .create_multipart(to, PutMultipartOptions::default()) + .await?; + + // S3 requires minimum 5 MiB per part (except final) and max 10,000 parts + let part_size = self.client.config.multipart_copy_part_size; + + let mut parts = Vec::new(); + let mut offset: u64 = 0; + let mut idx: usize = 0; + let res = async { + while offset < size { + let end = std::cmp::min(offset + part_size, size); + let payload = if offset == 0 && end == size { + PutPartPayload::Copy(from) + } else { + PutPartPayload::CopyRange(from, offset..end) + }; + let part = self.client.put_part(to, &upload_id, idx, payload).await?; + parts.push(part); + idx += 1; + offset = end; + } + self.client + .complete_multipart(to, &upload_id, parts, mode) + .await + .map(|_| ()) + } + .await; + + // If the multipart upload failed, make a best effort attempt to + // clean it up. It's the caller's responsibility to add a + // lifecycle rule if guaranteed cleanup is required, as we + // cannot protect against an ill-timed process crash. + if res.is_err() { + let _ = self.client.abort_multipart(to, &upload_id).await; + } + res + } } #[async_trait] @@ -310,14 +360,31 @@ impl ObjectStore for AmazonS3 { mode, extensions: _, } = options; + // Determine source size to decide between single CopyObject and multipart copy + let head_meta = self + .client + .get_opts( + from, + GetOptions { + head: true, + ..Default::default() + }, + ) + .await? + .meta; match mode { CopyMode::Overwrite => { - self.client - .copy_request(from, to) - .idempotent(true) - .send() - .await?; + if head_meta.size <= self.client.config.multipart_copy_threshold { + self.client + .copy_request(from, to) + .idempotent(true) + .send() + .await?; + } else { + self.copy_multipart(from, to, head_meta.size, CompleteMultipartMode::Overwrite) + .await?; + } Ok(()) } CopyMode::Create => { @@ -327,45 +394,16 @@ impl ObjectStore for AmazonS3 { } Some(S3CopyIfNotExists::HeaderWithStatus(k, v, status)) => (k, v, *status), Some(S3CopyIfNotExists::Multipart) => { - let upload_id = self - .client - .create_multipart(to, PutMultipartOptions::default()) - .await?; - - let res = async { - let part_id = self - .client - .put_part(to, &upload_id, 0, PutPartPayload::Copy(from)) - .await?; - match self - .client - .complete_multipart( - to, - &upload_id, - vec![part_id], - CompleteMultipartMode::Create, - ) - .await - { - Err(e @ Error::Precondition { .. }) => Err(Error::AlreadyExists { + return self + .copy_multipart(from, to, head_meta.size, CompleteMultipartMode::Create) + .await + .map_err(|err| match err { + Error::Precondition { .. } => Error::AlreadyExists { path: to.to_string(), - source: Box::new(e), - }), - Ok(_) => Ok(()), - Err(e) => Err(e), - } - } - .await; - - // If the multipart upload failed, make a best effort attempt to - // clean it up. It's the caller's responsibility to add a - // lifecycle rule if guaranteed cleanup is required, as we - // cannot protect against an ill-timed process crash. - if res.is_err() { - let _ = self.client.abort_multipart(to, &upload_id).await; - } - - return res; + source: Box::new(err), + }, + other => other, + }); } None => { return Err(Error::NotSupported { diff --git a/src/config.rs b/src/config.rs index 29a389d4..b042e209 100644 --- a/src/config.rs +++ b/src/config.rs @@ -112,6 +112,15 @@ impl Parse for u32 { } } +impl Parse for u64 { + fn parse(v: &str) -> Result { + Self::from_str(v).map_err(|_| Error::Generic { + store: "Config", + source: format!("failed to parse \"{v}\" as u64").into(), + }) + } +} + impl Parse for HeaderValue { fn parse(v: &str) -> Result { Self::from_str(v).map_err(|_| Error::Generic { From f4ab0182badf31271358cfdb9ac7321874e90ece Mon Sep 17 00:00:00 2001 From: James Smith Date: Wed, 17 Dec 2025 12:47:35 +1100 Subject: [PATCH 2/8] move comments --- src/aws/builder.rs | 5 +++-- src/aws/mod.rs | 8 ++++---- 2 files changed, 7 insertions(+), 6 deletions(-) diff --git a/src/aws/builder.rs b/src/aws/builder.rs index 71752c46..0393a3b4 100644 --- a/src/aws/builder.rs +++ b/src/aws/builder.rs @@ -42,8 +42,9 @@ use url::Url; /// Default metadata endpoint static DEFAULT_METADATA_ENDPOINT: &str = "http://169.254.169.254"; -/// AWS S3 does not support copy operations larger than 5 GiB in a single request. -/// https://docs.aws.amazon.com/AmazonS3/latest/userguide/copy-object.html +/// AWS S3 does not support copy operations larger than 5 GiB in a single request. See +/// [CopyObject](https://docs.aws.amazon.com/AmazonS3/latest/userguide/copy-object.html) for more +/// details. const MAX_SINGLE_REQUEST_COPY_SIZE: u64 = 5 * 1024 * 1024 * 1024; /// A specialized `Error` for object store-related errors diff --git a/src/aws/mod.rs b/src/aws/mod.rs index 8966c0d6..ded88103 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -103,6 +103,10 @@ impl AmazonS3 { } /// Perform a multipart copy operation + /// + /// If the multipart upload fails, this function makes a best effort attempt to clean it up. + /// It's the caller's responsibility to add a lifecycle rule if guaranteed cleanup is required, + /// as we cannot protect against an ill-timed process crash. async fn copy_multipart( &self, from: &Path, @@ -142,10 +146,6 @@ impl AmazonS3 { } .await; - // If the multipart upload failed, make a best effort attempt to - // clean it up. It's the caller's responsibility to add a - // lifecycle rule if guaranteed cleanup is required, as we - // cannot protect against an ill-timed process crash. if res.is_err() { let _ = self.client.abort_multipart(to, &upload_id).await; } From 788237db8afb5d5e2e69129fa499cfd3e91301ae Mon Sep 17 00:00:00 2001 From: James Smith Date: Wed, 17 Dec 2025 12:50:02 +1100 Subject: [PATCH 3/8] use let if instead of min() to avoid overflow --- src/aws/mod.rs | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/src/aws/mod.rs b/src/aws/mod.rs index ded88103..1b3b745f 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -128,7 +128,11 @@ impl AmazonS3 { let mut idx: usize = 0; let res = async { while offset < size { - let end = std::cmp::min(offset + part_size, size); + let end = if size - offset <= part_size { + size + } else { + offset + part_size + }; let payload = if offset == 0 && end == size { PutPartPayload::Copy(from) } else { From b5e738537d0501d0d74751ca4a6e16fdcdc67d33 Mon Sep 17 00:00:00 2001 From: James Smith Date: Wed, 17 Dec 2025 13:08:32 +1100 Subject: [PATCH 4/8] abort if part upload fails --- src/aws/mod.rs | 48 ++++++++++++++++++++++++++---------------------- 1 file changed, 26 insertions(+), 22 deletions(-) diff --git a/src/aws/mod.rs b/src/aws/mod.rs index be66c552..ff3df681 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -126,29 +126,33 @@ impl AmazonS3 { let mut parts = Vec::new(); let mut offset: u64 = 0; let mut idx: usize = 0; - let res = async { - while offset < size { - let end = if size - offset <= part_size { - size - } else { - offset + part_size - }; - let payload = if offset == 0 && end == size { - PutPartPayload::Copy(from) - } else { - PutPartPayload::CopyRange(from, offset..end) - }; - let part = self.client.put_part(to, &upload_id, idx, payload).await?; - parts.push(part); - idx += 1; - offset = end; - } - self.client - .complete_multipart(to, &upload_id, parts, mode) - .await - .map(|_| ()) + while offset < size { + let end = if size - offset <= part_size { + size + } else { + offset + part_size + }; + let payload = if offset == 0 && end == size { + PutPartPayload::Copy(from) + } else { + PutPartPayload::CopyRange(from, offset..end) + }; + let part = match self.client.put_part(to, &upload_id, idx, payload).await { + Ok(part) => part, + Err(e) => { + let _ = self.client.abort_multipart(to, &upload_id).await; + return Err(e); + } + }; + parts.push(part); + idx += 1; + offset = end; } - .await; + let res = self + .client + .complete_multipart(to, &upload_id, parts, mode) + .await + .map(|_| ()); if res.is_err() { let _ = self.client.abort_multipart(to, &upload_id).await; From 1fdf7d6f7746e58e59a04894fb41990af7e9fbeb Mon Sep 17 00:00:00 2001 From: James Smith Date: Fri, 19 Dec 2025 21:15:50 +1100 Subject: [PATCH 5/8] test internals --- .github/workflows/ci.yml | 2 + src/aws/mod.rs | 154 ++++++++++++++++++++++++++++++++------- 2 files changed, 130 insertions(+), 26 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 8eb49035..592c690c 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -131,6 +131,8 @@ jobs: aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-spawn aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-checksum aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-copy-if-not-exists + aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-multipart-copy-large + aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-multipart-copy-small aws --endpoint-url=http://localhost:4566 s3api create-bucket --bucket test-object-lock --object-lock-enabled-for-bucket KMS_KEY=$(aws --endpoint-url=http://localhost:4566 kms create-key --description "test key") diff --git a/src/aws/mod.rs b/src/aws/mod.rs index ff3df681..afa92e75 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -102,6 +102,26 @@ impl AmazonS3 { self.client.config.path_url(path) } + /// Construct the payloads for a multipart copy operation. + fn multipart_copy_payloads<'a>(&self, from: &'a Path, size: u64) -> Vec> { + let part_size = self.client.config.multipart_copy_part_size; + if size <= part_size { + return vec![PutPartPayload::Copy(from)]; + } + let mut payloads = Vec::new(); + let mut offset = 0; + while offset < size { + let end = if size - offset <= part_size { + size + } else { + offset + part_size + }; + payloads.push(PutPartPayload::CopyRange(from, offset..end)); + offset = end; + } + payloads + } + /// Perform a multipart copy operation /// /// If the multipart upload fails, this function makes a best effort attempt to clean it up. @@ -120,44 +140,29 @@ impl AmazonS3 { .create_multipart(to, PutMultipartOptions::default()) .await?; - // S3 requires minimum 5 MiB per part (except final) and max 10,000 parts - let part_size = self.client.config.multipart_copy_part_size; - let mut parts = Vec::new(); - let mut offset: u64 = 0; - let mut idx: usize = 0; - while offset < size { - let end = if size - offset <= part_size { - size - } else { - offset + part_size - }; - let payload = if offset == 0 && end == size { - PutPartPayload::Copy(from) - } else { - PutPartPayload::CopyRange(from, offset..end) - }; - let part = match self.client.put_part(to, &upload_id, idx, payload).await { - Ok(part) => part, + for (idx, payload) in self + .multipart_copy_payloads(from, size) + .into_iter() + .enumerate() + { + match self.client.put_part(to, &upload_id, idx, payload).await { + Ok(part) => parts.push(part), Err(e) => { let _ = self.client.abort_multipart(to, &upload_id).await; return Err(e); } }; - parts.push(part); - idx += 1; - offset = end; } - let res = self + if let Err(err) = self .client .complete_multipart(to, &upload_id, parts, mode) .await - .map(|_| ()); - - if res.is_err() { + { let _ = self.client.abort_multipart(to, &upload_id).await; + return Err(err); } - res + Ok(()) } } @@ -559,6 +564,7 @@ mod tests { use crate::tests::*; use base64::Engine; use base64::prelude::BASE64_STANDARD; + use bytes::BytesMut; use http::HeaderMap; const NON_EXISTENT_NAME: &str = "nonexistentname"; @@ -620,6 +626,63 @@ mod tests { store.delete(&dst).await.unwrap(); } + #[tokio::test] + async fn large_file_copy_multipart() { + maybe_skip_integration!(); + + let bucket = "test-bucket-for-multipart-copy-large"; + let store = AmazonS3Builder::from_env() + .with_bucket_name(bucket) + .with_multipart_copy_threshold(5 * 1024 * 1024) + .with_multipart_copy_part_size(5 * 1024 * 1024) + .build() + .unwrap(); + + let mut payload = BytesMut::zeroed(10 * 1024 * 1024); + rand::fill(&mut payload[..]); + + let src = Path::parse("src.bin").unwrap(); + let dst = Path::parse("dst.bin").unwrap(); + store + .put(&src, PutPayload::from(payload.clone().freeze())) + .await + .unwrap(); + store.copy(&src, &dst).await.unwrap(); + let copied = store.get(&dst).await.unwrap(); + let content = copied.bytes().await.unwrap(); + assert_eq!(content, payload); + store.delete(&src).await.unwrap(); + store.delete(&dst).await.unwrap(); + } + + #[tokio::test] + async fn small_file_copy_single_part() { + maybe_skip_integration!(); + + let bucket = "test-bucket-for-multipart-copy-small"; + let store = AmazonS3Builder::from_env() + .with_bucket_name(bucket) + .with_multipart_copy_threshold(5 * 1024 * 1024) // trigger multipart copy + .with_multipart_copy_part_size(50 * 1024 * 1024) // but only use one part + .build() + .unwrap(); + + let src = Path::parse("src.bin").unwrap(); + let dst = Path::parse("dst.bin").unwrap(); + let mut payload = BytesMut::zeroed(10 * 1024 * 1024); + rand::fill(&mut payload[..]); + store + .put(&src, PutPayload::from(payload.clone().freeze())) + .await + .unwrap(); + store.copy(&src, &dst).await.unwrap(); + let copied = store.get(&dst).await.unwrap(); + let content = copied.bytes().await.unwrap(); + assert_eq!(content, payload); + store.delete(&src).await.unwrap(); + store.delete(&dst).await.unwrap(); + } + #[tokio::test] async fn write_multipart_file_with_signature_object_lock() { maybe_skip_integration!(); @@ -960,4 +1023,43 @@ mod tests { shutdown_tx.send(()).ok(); thread_handle.join().expect("runtime thread panicked"); } + + #[test] + fn test_multipart_copy_payloads_single() { + let store = AmazonS3Builder::default() + .with_bucket_name(NON_EXISTENT_NAME) + .with_multipart_copy_part_size(1024) + .build() + .unwrap(); + let path = Path::from("test.txt"); + let payloads = store.multipart_copy_payloads(&path, 1024); + assert_eq!(payloads.len(), 1); + let PutPartPayload::Copy(payload_path) = payloads[0] else { + panic!("expected Copy payload"); + }; + assert_eq!(payload_path, &path); + } + #[test] + fn test_multipart_copy_payloads_multiple() { + let store = AmazonS3Builder::default() + .with_bucket_name(NON_EXISTENT_NAME) + .with_multipart_copy_part_size(1024) + .build() + .unwrap(); + let path = Path::from("test.txt"); + let payloads = store.multipart_copy_payloads(&path, 2000); + assert_eq!(payloads.len(), 2); + let mut payloads = payloads.into_iter(); + let Some(PutPartPayload::CopyRange(payload_path, range)) = payloads.next() else { + panic!("expected CopyRange payload"); + }; + assert_eq!(payload_path, &path); + assert_eq!(range, 0..1024); + let Some(PutPartPayload::CopyRange(payload_path, range)) = payloads.next() else { + panic!("expected CopyRange payload"); + }; + assert_eq!(payload_path, &path); + assert_eq!(range, 1024..2000); + assert!(payloads.next().is_none()); + } } From 231868b59d6469b81a43d168d1ac742000b43f4c Mon Sep 17 00:00:00 2001 From: James Smith Date: Thu, 15 Jan 2026 20:00:32 +1100 Subject: [PATCH 6/8] default to single request only --- src/aws/builder.rs | 6 ++-- src/aws/client.rs | 6 ++-- src/aws/mod.rs | 74 ++++++++++++++++++++++++++++++++-------------- 3 files changed, 57 insertions(+), 29 deletions(-) diff --git a/src/aws/builder.rs b/src/aws/builder.rs index 0393a3b4..a467c1d2 100644 --- a/src/aws/builder.rs +++ b/src/aws/builder.rs @@ -194,7 +194,8 @@ pub struct AmazonS3Builder { request_payer: ConfigValue, /// The [`HttpConnector`] to use http_connector: Option>, - /// Threshold (bytes) above which copy uses multipart copy. If not set, defaults to 5 GiB. + /// Threshold (bytes) above which copy uses multipart copy. If not set, all copies are performed + /// as single requests. multipart_copy_threshold: Option>, /// Preferred multipart copy part size (bytes). If not set, defaults to 5 GiB. multipart_copy_part_size: Option>, @@ -1235,8 +1236,7 @@ impl AmazonS3Builder { let multipart_copy_threshold = self .multipart_copy_threshold .map(|val| val.get()) - .transpose()? - .unwrap_or(MAX_SINGLE_REQUEST_COPY_SIZE); + .transpose()?; let multipart_copy_part_size = self .multipart_copy_part_size .map(|val| val.get()) diff --git a/src/aws/client.rs b/src/aws/client.rs index dd0c8679..456db44d 100644 --- a/src/aws/client.rs +++ b/src/aws/client.rs @@ -211,8 +211,8 @@ pub(crate) struct S3Config { pub request_payer: bool, pub(super) encryption_headers: S3EncryptionHeaders, /// Threshold in bytes above which copy will use multipart copy - pub multipart_copy_threshold: u64, - /// Preferred multipart copy part size in bytes (None => auto) + pub multipart_copy_threshold: Option, + /// Preferred multipart copy part size in bytes (None => 5GiB) pub multipart_copy_part_size: u64, } @@ -1020,7 +1020,7 @@ mod tests { conditional_put: Default::default(), encryption_headers: Default::default(), request_payer: false, - multipart_copy_threshold: 5 * 1024 * 1024 * 1024, + multipart_copy_threshold: None, multipart_copy_part_size: 5 * 1024 * 1024 * 1024, }; diff --git a/src/aws/mod.rs b/src/aws/mod.rs index afa92e75..ac814b83 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -166,6 +166,11 @@ impl AmazonS3 { } } +enum CopyMethod { + SingleRequest, + Multipart(u64), +} + #[async_trait] impl Signer for AmazonS3 { /// Create a URL containing the relevant [AWS SigV4] query parameters that authorize a request @@ -368,30 +373,37 @@ impl ObjectStore for AmazonS3 { mode, extensions: _, } = options; - // Determine source size to decide between single CopyObject and multipart copy - let head_meta = self - .client - .get_opts( - from, - GetOptions { - head: true, - ..Default::default() - }, - ) - .await? - .meta; + + let copy_method = if let Some(limit) = self.client.config.multipart_copy_threshold { + let size = self + .client + .get_opts(from, GetOptions::new().with_head(true)) + .await? + .meta + .size; + if size < limit { + CopyMethod::SingleRequest + } else { + CopyMethod::Multipart(size) + } + } else { + CopyMethod::SingleRequest + }; match mode { CopyMode::Overwrite => { - if head_meta.size <= self.client.config.multipart_copy_threshold { - self.client - .copy_request(from, to) - .idempotent(true) - .send() - .await?; - } else { - self.copy_multipart(from, to, head_meta.size, CompleteMultipartMode::Overwrite) - .await?; + match copy_method { + CopyMethod::SingleRequest => { + self.client + .copy_request(from, to) + .idempotent(true) + .send() + .await?; + } + CopyMethod::Multipart(size) => { + self.copy_multipart(from, to, size, CompleteMultipartMode::Overwrite) + .await?; + } } Ok(()) } @@ -402,8 +414,17 @@ impl ObjectStore for AmazonS3 { } Some(S3CopyIfNotExists::HeaderWithStatus(k, v, status)) => (k, v, *status), Some(S3CopyIfNotExists::Multipart) => { + let size = if let CopyMethod::Multipart(size) = copy_method { + size + } else { + self.client + .get_opts(from, GetOptions::new().with_head(true)) + .await? + .meta + .size + }; return self - .copy_multipart(from, to, head_meta.size, CompleteMultipartMode::Create) + .copy_multipart(from, to, size, CompleteMultipartMode::Create) .await .map_err(|err| match err { Error::Precondition { .. } => Error::AlreadyExists { @@ -419,9 +440,17 @@ impl ObjectStore for AmazonS3 { }); } }; + if matches!(copy_method, CopyMethod::Multipart(_)) { + return Err(Error::NotSupported { + source: "Object size is above multipart copy threshold, but \ + CopyIfNotExists headers are not supported for multipart copies" + .into(), + }); + } let req = self.client.copy_request(from, to); match req.header(k, v).send().await { + Ok(_) => Ok(()), Err(RequestError::Retry { source, path }) if source.status() == Some(status) => { @@ -431,7 +460,6 @@ impl ObjectStore for AmazonS3 { }) } Err(e) => Err(e.into()), - Ok(_) => Ok(()), } } } From dbd4724d07ab6131e03c798a2b9df6bc9a42c48d Mon Sep 17 00:00:00 2001 From: james-rms Date: Thu, 15 Jan 2026 20:08:22 +1100 Subject: [PATCH 7/8] Apply suggestion from @tustvold Co-authored-by: Raphael Taylor-Davies <1781103+tustvold@users.noreply.github.com> --- src/aws/mod.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/aws/mod.rs b/src/aws/mod.rs index ac814b83..dfa5d7e9 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -662,7 +662,7 @@ mod tests { let store = AmazonS3Builder::from_env() .with_bucket_name(bucket) .with_multipart_copy_threshold(5 * 1024 * 1024) - .with_multipart_copy_part_size(5 * 1024 * 1024) + .with_multipart_copy_part_size(1337 * 1024) .build() .unwrap(); From 6fc900eccd0a79157203dab64213598b3a826eda Mon Sep 17 00:00:00 2001 From: James Smith Date: Mon, 19 Jan 2026 11:23:06 +1100 Subject: [PATCH 8/8] fix tests, return error for <5MB part size --- .github/workflows/ci.yml | 2 +- src/aws/builder.rs | 12 ++++++++++++ src/aws/mod.rs | 4 ++-- 3 files changed, 15 insertions(+), 3 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 592c690c..be13fb23 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -132,7 +132,7 @@ jobs: aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-checksum aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-copy-if-not-exists aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-multipart-copy-large - aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-multipart-copy-small + aws --endpoint-url=http://localhost:4566 s3 mb s3://test-bucket-for-multipart-copy-single aws --endpoint-url=http://localhost:4566 s3api create-bucket --bucket test-object-lock --object-lock-enabled-for-bucket KMS_KEY=$(aws --endpoint-url=http://localhost:4566 kms create-key --description "test key") diff --git a/src/aws/builder.rs b/src/aws/builder.rs index a467c1d2..83735d88 100644 --- a/src/aws/builder.rs +++ b/src/aws/builder.rs @@ -46,6 +46,11 @@ static DEFAULT_METADATA_ENDPOINT: &str = "http://169.254.169.254"; /// [CopyObject](https://docs.aws.amazon.com/AmazonS3/latest/userguide/copy-object.html) for more /// details. const MAX_SINGLE_REQUEST_COPY_SIZE: u64 = 5 * 1024 * 1024 * 1024; +/// AWS S3 rejects multipart upload parts smaller than 5 MiB, unless they are the last part of the +/// upload. See +/// [CompleteMultipartUpload](https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html) +/// for more details. +const MIN_PART_SIZE_BYTES: u64 = 5 * 1024 * 1024; /// A specialized `Error` for object store-related errors #[derive(Debug, thiserror::Error)] @@ -1243,6 +1248,13 @@ impl AmazonS3Builder { .transpose()? .unwrap_or(MAX_SINGLE_REQUEST_COPY_SIZE); + if multipart_copy_part_size < MIN_PART_SIZE_BYTES { + return Err(crate::Error::Generic { + source: "Multipart copy part size must be >= 5MB".into(), + store: STORE, + }); + } + let config = S3Config { region, bucket, diff --git a/src/aws/mod.rs b/src/aws/mod.rs index dfa5d7e9..17495b4c 100644 --- a/src/aws/mod.rs +++ b/src/aws/mod.rs @@ -662,7 +662,7 @@ mod tests { let store = AmazonS3Builder::from_env() .with_bucket_name(bucket) .with_multipart_copy_threshold(5 * 1024 * 1024) - .with_multipart_copy_part_size(1337 * 1024) + .with_multipart_copy_part_size(6 * 1024 * 1024) .build() .unwrap(); @@ -687,7 +687,7 @@ mod tests { async fn small_file_copy_single_part() { maybe_skip_integration!(); - let bucket = "test-bucket-for-multipart-copy-small"; + let bucket = "test-bucket-for-multipart-copy-single"; let store = AmazonS3Builder::from_env() .with_bucket_name(bucket) .with_multipart_copy_threshold(5 * 1024 * 1024) // trigger multipart copy