You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/05/17 16:43:25 UTC
(opendal) 01/01: feat(services/s3): add multipart copier
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/copier-s3-multipart
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit f384e02aec8dfd77c9f23e8ca3b04926177c7403
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon May 18 00:43:10 2026 +0800
feat(services/s3): add multipart copier
---
core/core/src/raw/enum_utils.rs | 16 ++
core/services/s3/src/backend.rs | 41 +---
core/services/s3/src/copier.rs | 431 ++++++++++++++++++++++++++++++++++++++++
core/services/s3/src/core.rs | 200 +++++++++++++++++++
core/services/s3/src/lib.rs | 1 +
core/services/s3/src/mod.rs | 2 +-
6 files changed, 657 insertions(+), 34 deletions(-)
diff --git a/core/core/src/raw/enum_utils.rs b/core/core/src/raw/enum_utils.rs
index 93822aa44..59630a741 100644
--- a/core/core/src/raw/enum_utils.rs
+++ b/core/core/src/raw/enum_utils.rs
@@ -83,6 +83,22 @@ impl<ONE: oio::Write, TWO: oio::Write> oio::Write for TwoWays<ONE, TWO> {
}
}
+impl<ONE: oio::Copy, TWO: oio::Copy> oio::Copy for TwoWays<ONE, TWO> {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ match self {
+ Self::One(v) => v.next().await,
+ Self::Two(v) => v.next().await,
+ }
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ match self {
+ Self::One(v) => v.abort().await,
+ Self::Two(v) => v.abort().await,
+ }
+ }
+}
+
impl<ONE: oio::List, TWO: oio::List> oio::List for TwoWays<ONE, TWO> {
async fn next(&mut self) -> Result<Option<oio::Entry>> {
match self {
diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs
index c998267d2..2a1935c8c 100644
--- a/core/services/s3/src/backend.rs
+++ b/core/services/s3/src/backend.rs
@@ -45,6 +45,8 @@ use url::Url;
use crate::S3_SCHEME;
use crate::config::S3Config;
+use crate::copier::S3Copiers;
+use crate::copier::new_s3_copier;
use crate::core::*;
use crate::deleter::S3Deleter;
use crate::error::parse_error;
@@ -999,7 +1001,7 @@ impl Access for S3Backend {
type Writer = S3Writers;
type Lister = S3Listers;
type Deleter = oio::BatchDeleter<S3Deleter>;
- type Copier = ();
+ type Copier = S3Copiers;
fn info(&self) -> Arc<AccessorInfo> {
self.core.info.clone()
@@ -1100,39 +1102,12 @@ impl Access for S3Backend {
&self,
from: &str,
to: &str,
- _args: OpCopy,
- _opts: OpCopier,
+ args: OpCopy,
+ opts: OpCopier,
) -> Result<(RpCopy, Self::Copier)> {
- let resp = self.core.s3_copy_object(from, to).await?;
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => {
- // S3 CopyObject may return an error embedded in a 200 OK response body
- // when the error occurs during the copy (e.g., throttling, internal error).
- // We must parse the body to detect this.
- //
- // ref: https://docs.aws.amazon.com/AmazonS3/latest/API/API_CopyObject.html
- // ref: https://repost.aws/knowledge-center/s3-resolve-200-internalerror
- let body = resp.into_body().to_bytes();
-
- let result: CopyObjectResult =
- quick_xml::de::from_reader(body.as_ref()).map_err(new_xml_deserialize_error)?;
-
- // On success, ETag is always present. If it's empty, the body was not a
- // valid <CopyObjectResult> — it's an error response embedded in 200 OK.
- if result.etag.is_empty() {
- return Err(
- Error::new(ErrorKind::Unexpected, String::from_utf8_lossy(&body))
- .set_temporary(),
- );
- }
-
- Ok((RpCopy::default(), ()))
- }
- _ => Err(parse_error(resp)),
- }
+ let meta = self.stat(from, OpStat::default()).await?.into_metadata();
+ let copier = new_s3_copier(self.core.clone(), from, to, args, opts, meta)?;
+ Ok((RpCopy::default(), copier))
}
async fn presign(&self, path: &str, args: OpPresign) -> Result<RpPresign> {
diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs
new file mode 100644
index 000000000..443ee1048
--- /dev/null
+++ b/core/services/s3/src/copier.rs
@@ -0,0 +1,431 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+use std::sync::Arc;
+
+use bytes::Buf;
+use http::StatusCode;
+
+use crate::core::*;
+use crate::error::S3Error;
+use crate::error::from_s3_error;
+use crate::error::parse_error;
+use opendal_core::raw::*;
+use opendal_core::*;
+
+const S3_COPY_OBJECT_MAX_SIZE: u64 = 5 * 1024 * 1024 * 1024;
+const S3_MULTIPART_COPY_MIN_PART_SIZE: u64 = 5 * 1024 * 1024;
+const S3_MULTIPART_COPY_MAX_PARTS: u64 = 10_000;
+
+#[cfg(target_pointer_width = "64")]
+const S3_MULTIPART_COPY_MAX_PART_SIZE: u64 = 5 * 1024 * 1024 * 1024;
+#[cfg(not(target_pointer_width = "64"))]
+const S3_MULTIPART_COPY_MAX_PART_SIZE: u64 = usize::MAX as u64;
+
+pub type S3Copiers = TwoWays<oio::OneShotCopier, S3MultipartCopier>;
+
+pub fn new_s3_copier(
+ core: Arc<S3Core>,
+ from: &str,
+ to: &str,
+ args: OpCopy,
+ opts: OpCopier,
+ meta: Metadata,
+) -> Result<S3Copiers> {
+ let source_size = meta.content_length();
+
+ if source_size <= S3_COPY_OBJECT_MAX_SIZE {
+ let from = from.to_string();
+ let to = to.to_string();
+ return Ok(TwoWays::One(oio::OneShotCopier::new(async move {
+ copy_object_once(core, &from, &to).await?;
+ Ok(None)
+ })));
+ }
+
+ let part_size = multipart_copy_part_size(source_size)?;
+
+ Ok(TwoWays::Two(S3MultipartCopier::new(
+ S3MultipartCopy {
+ core,
+ from: from.to_string(),
+ to: to.to_string(),
+ args,
+ source_etag: meta.etag().map(ToOwned::to_owned),
+ source_version: meta.version().map(ToOwned::to_owned),
+ },
+ source_size,
+ part_size,
+ opts.concurrent(),
+ )))
+}
+
+async fn copy_object_once(core: Arc<S3Core>, from: &str, to: &str) -> Result<()> {
+ let resp = core.s3_copy_object(from, to).await?;
+
+ match resp.status() {
+ StatusCode::OK => {
+ let body = resp.into_body().to_bytes();
+
+ let result: CopyObjectResult =
+ quick_xml::de::from_reader(body.as_ref()).map_err(new_xml_deserialize_error)?;
+
+ if result.etag.is_empty() {
+ return Err(
+ Error::new(ErrorKind::Unexpected, String::from_utf8_lossy(&body))
+ .set_temporary(),
+ );
+ }
+
+ Ok(())
+ }
+ _ => Err(parse_error(resp)),
+ }
+}
+
+fn multipart_copy_part_size(source_size: u64) -> Result<u64> {
+ let part_size = source_size
+ .div_ceil(S3_MULTIPART_COPY_MAX_PARTS)
+ .max(S3_MULTIPART_COPY_MIN_PART_SIZE);
+
+ if part_size > S3_MULTIPART_COPY_MAX_PART_SIZE {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ "source object is too large for S3 multipart copy",
+ )
+ .with_context("source_size", source_size)
+ .with_context("part_size", part_size)
+ .with_context("max_part_size", S3_MULTIPART_COPY_MAX_PART_SIZE));
+ }
+
+ Ok(part_size)
+}
+
+struct S3MultipartCopy {
+ core: Arc<S3Core>,
+ from: String,
+ to: String,
+ args: OpCopy,
+ source_etag: Option<String>,
+ source_version: Option<String>,
+}
+
+#[derive(Clone)]
+struct CopyInput {
+ inner: Arc<S3MultipartCopy>,
+ upload_id: Arc<String>,
+ part_number: usize,
+ range: BytesRange,
+}
+
+struct CopiedPart {
+ part: oio::MultipartPart,
+ size: u64,
+}
+
+pub struct S3MultipartCopier {
+ inner: Arc<S3MultipartCopy>,
+
+ upload_id: Option<Arc<String>>,
+ parts: Vec<oio::MultipartPart>,
+ next_part_number: usize,
+ next_offset: u64,
+ source_size: u64,
+ part_size: u64,
+ concurrent: usize,
+ completed: bool,
+
+ tasks: ConcurrentTasks<CopyInput, CopiedPart>,
+}
+
+impl S3MultipartCopier {
+ fn new(inner: S3MultipartCopy, source_size: u64, part_size: u64, concurrent: usize) -> Self {
+ let executor = inner.core.info.executor();
+ Self {
+ inner: Arc::new(inner),
+ upload_id: None,
+ parts: Vec::new(),
+ next_part_number: 0,
+ next_offset: 0,
+ source_size,
+ part_size,
+ concurrent,
+ completed: false,
+
+ tasks: ConcurrentTasks::new(executor, concurrent, concurrent, |input| {
+ Box::pin(async move {
+ let result = input
+ .inner
+ .copy_part(&input.upload_id, input.part_number, input.range)
+ .await;
+ (input, result)
+ })
+ }),
+ }
+ }
+
+ async fn upload_id(&mut self) -> Result<Arc<String>> {
+ match self.upload_id.clone() {
+ Some(upload_id) => Ok(upload_id),
+ None => {
+ let upload_id = self.inner.initiate_copy().await?;
+ let upload_id = Arc::new(upload_id);
+ self.upload_id = Some(upload_id.clone());
+ Ok(upload_id)
+ }
+ }
+ }
+
+ async fn fill_tasks(&mut self, upload_id: Arc<String>) -> Result<()> {
+ let mut scheduled = 0;
+
+ while self.next_offset < self.source_size
+ && self.tasks.has_remaining()
+ && scheduled < self.concurrent
+ {
+ let size = self.part_size.min(self.source_size - self.next_offset);
+ let range = BytesRange::new(self.next_offset, Some(size));
+
+ let input = CopyInput {
+ inner: self.inner.clone(),
+ upload_id: upload_id.clone(),
+ part_number: self.next_part_number,
+ range,
+ };
+
+ loop {
+ match self.tasks.execute(input.clone()).await {
+ Ok(()) => break,
+ Err(err) if err.is_temporary() => continue,
+ Err(err) => return Err(err),
+ }
+ }
+
+ self.next_offset += size;
+ self.next_part_number += 1;
+ scheduled += 1;
+
+ if self.tasks.has_result() {
+ break;
+ }
+ }
+
+ Ok(())
+ }
+}
+
+impl S3MultipartCopy {
+ async fn initiate_copy(&self) -> Result<String> {
+ let resp = self.core.s3_initiate_multipart_copy(&self.to).await?;
+
+ match resp.status() {
+ StatusCode::OK => {
+ let bs = resp.into_body();
+
+ let result: InitiateMultipartUploadResult =
+ quick_xml::de::from_reader(bs.reader()).map_err(new_xml_deserialize_error)?;
+
+ Ok(result.upload_id)
+ }
+ _ => Err(parse_error(resp)),
+ }
+ }
+
+ async fn copy_part(
+ &self,
+ upload_id: &str,
+ part_number: usize,
+ range: BytesRange,
+ ) -> Result<CopiedPart> {
+ let size = range.size().expect("multipart copy range must be sized");
+ let part_number = part_number + 1;
+
+ let req = self
+ .core
+ .s3_upload_part_copy_request(S3UploadPartCopyRequest {
+ from: &self.from,
+ to: &self.to,
+ upload_id,
+ part_number,
+ range,
+ source_etag: self.source_etag.as_deref(),
+ source_version: self.source_version.as_deref(),
+ })?;
+
+ let resp = self.core.send(req).await?;
+
+ match resp.status() {
+ StatusCode::OK => {
+ let body = resp.into_body().to_bytes();
+ let result: CopyObjectResult =
+ quick_xml::de::from_reader(body.as_ref()).map_err(new_xml_deserialize_error)?;
+
+ if result.etag.is_empty() {
+ return Err(
+ Error::new(ErrorKind::Unexpected, String::from_utf8_lossy(&body))
+ .set_temporary(),
+ );
+ }
+
+ Ok(CopiedPart {
+ part: oio::MultipartPart {
+ part_number,
+ etag: result.etag,
+ checksum: None,
+ size: Some(size),
+ },
+ size,
+ })
+ }
+ _ => Err(parse_error(resp)),
+ }
+ }
+
+ async fn complete_copy(&self, upload_id: &str, parts: &[oio::MultipartPart]) -> Result<()> {
+ let parts = parts
+ .iter()
+ .map(|p| CompleteMultipartUploadRequestPart {
+ part_number: p.part_number,
+ etag: p.etag.clone(),
+ ..Default::default()
+ })
+ .collect();
+
+ let resp = self
+ .core
+ .s3_complete_multipart_copy(&self.to, upload_id, parts, &self.args)
+ .await?;
+
+ let status = resp.status();
+
+ match status {
+ StatusCode::OK => {
+ let (parts, body) = resp.into_parts();
+
+ let ret: CompleteMultipartUploadResult =
+ quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
+ if !ret.code.is_empty() {
+ return Err(from_s3_error(
+ S3Error {
+ code: ret.code,
+ message: ret.message,
+ resource: "".to_string(),
+ request_id: ret.request_id,
+ },
+ parts,
+ ));
+ }
+
+ Ok(())
+ }
+ _ => Err(parse_error(resp)),
+ }
+ }
+
+ async fn abort_copy(&self, upload_id: &str) -> Result<()> {
+ let resp = self
+ .core
+ .s3_abort_multipart_copy(&self.to, upload_id)
+ .await?;
+ match resp.status() {
+ StatusCode::NO_CONTENT => Ok(()),
+ _ => Err(parse_error(resp)),
+ }
+ }
+}
+
+impl oio::Copy for S3MultipartCopier {
+ async fn next(&mut self) -> Result<Option<usize>> {
+ if self.completed {
+ return Ok(None);
+ }
+
+ let upload_id = self.upload_id().await?;
+ self.fill_tasks(upload_id.clone()).await?;
+
+ loop {
+ match self.tasks.next().await {
+ Some(Ok(result)) => {
+ let size = result.size.try_into().map_err(|_| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "multipart copy part size exceeds usize",
+ )
+ })?;
+ self.parts.push(result.part);
+ return Ok(Some(size));
+ }
+ Some(Err(err)) if err.is_temporary() => continue,
+ Some(Err(err)) => return Err(err),
+ None => break,
+ }
+ }
+
+ if self.parts.len() != self.next_part_number {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "multipart copy part numbers mismatch, please report bug to opendal",
+ )
+ .with_context("expected", self.next_part_number)
+ .with_context("actual", self.parts.len())
+ .with_context("upload_id", upload_id.to_string()));
+ }
+
+ self.parts.sort_by_key(|part| part.part_number);
+ self.inner.complete_copy(&upload_id, &self.parts).await?;
+ self.completed = true;
+ Ok(None)
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.tasks.clear();
+ let Some(upload_id) = self.upload_id.take() else {
+ return Ok(());
+ };
+
+ self.inner.abort_copy(&upload_id).await?;
+ self.completed = true;
+ Ok(())
+ }
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn test_multipart_copy_part_size() {
+ assert_eq!(
+ multipart_copy_part_size(S3_COPY_OBJECT_MAX_SIZE + 1).unwrap(),
+ S3_MULTIPART_COPY_MIN_PART_SIZE
+ );
+
+ assert_eq!(
+ multipart_copy_part_size(S3_MULTIPART_COPY_MIN_PART_SIZE * S3_MULTIPART_COPY_MAX_PARTS)
+ .unwrap(),
+ S3_MULTIPART_COPY_MIN_PART_SIZE
+ );
+
+ assert_eq!(
+ multipart_copy_part_size(
+ S3_MULTIPART_COPY_MIN_PART_SIZE * S3_MULTIPART_COPY_MAX_PARTS + 1
+ )
+ .unwrap(),
+ S3_MULTIPART_COPY_MIN_PART_SIZE + 1
+ );
+ }
+}
diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs
index 21cc5903b..846e6c9bf 100644
--- a/core/services/s3/src/core.rs
+++ b/core/services/s3/src/core.rs
@@ -48,6 +48,8 @@ use opendal_core::*;
pub mod constants {
pub const X_AMZ_COPY_SOURCE: &str = "x-amz-copy-source";
+ pub const X_AMZ_COPY_SOURCE_IF_MATCH: &str = "x-amz-copy-source-if-match";
+ pub const X_AMZ_COPY_SOURCE_RANGE: &str = "x-amz-copy-source-range";
pub const X_AMZ_SERVER_SIDE_ENCRYPTION: &str = "x-amz-server-side-encryption";
pub const X_AMZ_SERVER_REQUEST_PAYER: (&str, &str) = ("x-amz-request-payer", "requester");
@@ -105,6 +107,16 @@ pub struct S3Core {
pub checksum_algorithm: Option<ChecksumAlgorithm>,
}
+pub(crate) struct S3UploadPartCopyRequest<'a> {
+ pub(crate) from: &'a str,
+ pub(crate) to: &'a str,
+ pub(crate) upload_id: &'a str,
+ pub(crate) part_number: usize,
+ pub(crate) range: BytesRange,
+ pub(crate) source_etag: Option<&'a str>,
+ pub(crate) source_version: Option<&'a str>,
+}
+
impl Debug for S3Core {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
f.debug_struct("S3Core")
@@ -866,6 +878,121 @@ impl S3Core {
self.send(req).await
}
+ pub async fn s3_initiate_multipart_copy(&self, path: &str) -> Result<Response<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+
+ let url = format!("{}/{}?uploads", self.endpoint, percent_encode_path(&p));
+
+ let mut req = Request::post(&url);
+
+ // Set storage class header.
+ if let Some(v) = &self.default_storage_class {
+ req = req.header(HeaderName::from_static(constants::X_AMZ_STORAGE_CLASS), v);
+ }
+
+ // Also set acl header if default_acl is set.
+ if let Some(acl) = &self.default_acl {
+ req = req.header(constants::X_AMZ_ACL, acl);
+ }
+
+ // Set request payer header if enabled.
+ req = self.insert_request_payer_header(req);
+
+ // Set SSE headers.
+ req = self.insert_sse_headers(req, true);
+
+ let req = req
+ .extension(Operation::Copy)
+ .extension(ServiceOperation("CreateMultipartUpload"))
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
+ pub(crate) fn s3_upload_part_copy_request(
+ &self,
+ input: S3UploadPartCopyRequest<'_>,
+ ) -> Result<Request<Buffer>> {
+ let from = build_abs_path(&self.root, input.from);
+ let to = build_abs_path(&self.root, input.to);
+
+ let mut source = format!("{}/{}", self.bucket, percent_encode_path(&from));
+ if let Some(version) = input.source_version {
+ source.push_str(&format!(
+ "?{}={}",
+ constants::S3_QUERY_VERSION_ID,
+ percent_encode_path(version)
+ ));
+ }
+
+ let url = format!(
+ "{}/{}?partNumber={}&uploadId={}",
+ self.endpoint,
+ percent_encode_path(&to),
+ input.part_number,
+ percent_encode_path(input.upload_id)
+ );
+
+ let mut req = Request::put(&url);
+
+ // Set SSE headers.
+ req = self.insert_sse_headers(req, true);
+
+ if let Some(v) = &self.server_side_encryption_customer_algorithm {
+ let mut v = v.clone();
+ v.set_sensitive(true);
+
+ req = req.header(
+ HeaderName::from_static(
+ constants::X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_ALGORITHM,
+ ),
+ v,
+ )
+ }
+
+ if let Some(v) = &self.server_side_encryption_customer_key {
+ let mut v = v.clone();
+ v.set_sensitive(true);
+
+ req = req.header(
+ HeaderName::from_static(
+ constants::X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY,
+ ),
+ v,
+ )
+ }
+
+ if let Some(v) = &self.server_side_encryption_customer_key_md5 {
+ let mut v = v.clone();
+ v.set_sensitive(true);
+
+ req = req.header(
+ HeaderName::from_static(
+ constants::X_AMZ_COPY_SOURCE_SERVER_SIDE_ENCRYPTION_CUSTOMER_KEY_MD5,
+ ),
+ v,
+ )
+ }
+
+ if let Some(etag) = input.source_etag {
+ req = req.header(constants::X_AMZ_COPY_SOURCE_IF_MATCH, etag);
+ }
+
+ // Set request payer header if enabled.
+ req = self.insert_request_payer_header(req);
+
+ let req = req
+ .extension(Operation::Copy)
+ .extension(ServiceOperation("UploadPartCopy"))
+ .header(constants::X_AMZ_COPY_SOURCE, source)
+ .header(constants::X_AMZ_COPY_SOURCE_RANGE, input.range.to_header())
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+
+ Ok(req)
+ }
+
pub fn s3_upload_part_request(
&self,
path: &str,
@@ -962,6 +1089,50 @@ impl S3Core {
self.send(req).await
}
+ pub async fn s3_complete_multipart_copy(
+ &self,
+ path: &str,
+ upload_id: &str,
+ parts: Vec<CompleteMultipartUploadRequestPart>,
+ args: &OpCopy,
+ ) -> Result<Response<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "{}/{}?uploadId={}",
+ self.endpoint,
+ percent_encode_path(&p),
+ percent_encode_path(upload_id)
+ );
+
+ let mut req = Request::post(&url);
+
+ // Set SSE headers.
+ req = self.insert_sse_headers(req, true);
+
+ let content = quick_xml::se::to_string(&CompleteMultipartUploadRequest { part: parts })
+ .map_err(new_xml_serialize_error)?;
+ // Make sure content length has been set to avoid post with chunked encoding.
+ req = req.header(CONTENT_LENGTH, content.len());
+ // Set content-type to `application/xml` to avoid mixed with form post.
+ req = req.header(CONTENT_TYPE, "application/xml");
+
+ if args.if_not_exists() {
+ req = req.header(IF_NONE_MATCH, "*");
+ }
+
+ // Set request payer header if enabled.
+ req = self.insert_request_payer_header(req);
+
+ let req = req
+ .extension(Operation::Copy)
+ .extension(ServiceOperation("CompleteMultipartUpload"))
+ .body(Buffer::from(Bytes::from(content)))
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
/// Abort an on-going multipart upload.
pub async fn s3_abort_multipart_upload(
&self,
@@ -992,6 +1163,35 @@ impl S3Core {
self.send(req).await
}
+ /// Abort an on-going multipart copy.
+ pub async fn s3_abort_multipart_copy(
+ &self,
+ path: &str,
+ upload_id: &str,
+ ) -> Result<Response<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+
+ let url = format!(
+ "{}/{}?uploadId={}",
+ self.endpoint,
+ percent_encode_path(&p),
+ percent_encode_path(upload_id)
+ );
+
+ let mut req = Request::delete(&url);
+
+ // Set request payer header if enabled.
+ req = self.insert_request_payer_header(req);
+
+ let req = req
+ .extension(Operation::Copy)
+ .extension(ServiceOperation("AbortMultipartUpload"))
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+
+ self.send(req).await
+ }
+
pub async fn s3_delete_objects(
&self,
paths: &[(String, OpDelete)],
diff --git a/core/services/s3/src/lib.rs b/core/services/s3/src/lib.rs
index de67ed161..4467cd3d4 100644
--- a/core/services/s3/src/lib.rs
+++ b/core/services/s3/src/lib.rs
@@ -21,6 +21,7 @@
mod backend;
mod config;
+mod copier;
mod core;
mod deleter;
mod error;
diff --git a/core/services/s3/src/mod.rs b/core/services/s3/src/mod.rs
index 69d3819b3..bfe9d905b 100644
--- a/core/services/s3/src/mod.rs
+++ b/core/services/s3/src/mod.rs
@@ -20,6 +20,7 @@ pub const S3_SCHEME: &str = "s3";
mod backend;
mod config;
+mod copier;
mod core;
mod deleter;
mod error;
@@ -28,4 +29,3 @@ mod writer;
pub use backend::S3Builder as S3;
pub use config::S3Config;
-