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/08/31 04:17:48 UTC
(opendal) 01/02: fix(services): return authoritative copy metadata
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/compact-metadata
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit ed95de1f73a926824a27ac70217ddcaba7f9b851
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Aug 31 12:13:05 2026 +0800
fix(services): return authoritative copy metadata
---
core/core/src/docs/specs/metadata.md | 10 +-
core/core/src/types/options.rs | 18 ++-
core/services/aliyun-drive/src/backend.rs | 24 ++--
core/services/dropbox/src/backend.rs | 11 +-
core/services/koofr/src/backend.rs | 17 ++-
core/services/lakefs/src/backend.rs | 15 ++-
core/services/obs/src/backend.rs | 15 ++-
core/services/onedrive/src/backend.rs | 15 ++-
core/services/pcloud/src/backend.rs | 21 +++-
core/services/s3/src/backend.rs | 182 +++++++++++++++++++++++++++++-
core/services/s3/src/copier.rs | 97 ++++++++++++++--
core/services/s3/src/core.rs | 7 +-
core/services/upyun/src/backend.rs | 15 ++-
core/services/vercel-blob/src/backend.rs | 15 ++-
core/services/webdav/src/backend.rs | 18 ++-
core/services/webdav/src/core.rs | 7 +-
core/services/yandex-disk/src/backend.rs | 17 ++-
17 files changed, 439 insertions(+), 65 deletions(-)
diff --git a/core/core/src/docs/specs/metadata.md b/core/core/src/docs/specs/metadata.md
index 0c7257f15..e253c160b 100644
--- a/core/core/src/docs/specs/metadata.md
+++ b/core/core/src/docs/specs/metadata.md
@@ -40,10 +40,12 @@ boundary that owns an authoritative length:
files with the stat result's length.
- write completion uses the number of bytes written for replacements. Append
adapters use the final object offset instead of the number of appended bytes.
-- copy completion uses a service result length, the source-length hint, the
- source length discovered by a segmented copier, or successful copy progress.
- It does not issue a target `stat` and fails when none of these sources can
- provide an authoritative length.
+- copy completion uses an authoritative service result or successful copy
+ progress when the copy protocol provides one. Otherwise, it trusts a
+ caller-asserted source length without reading source metadata, or reads source
+ metadata before copying when no assertion is present. It does not issue a
+ target `stat` and fails when none of these sources can provide an authoritative
+ length.
Delete markers that do not carry an object length use a content length of zero.
diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs
index a13a63a25..a38a4e76f 100644
--- a/core/core/src/types/options.rs
+++ b/core/core/src/types/options.rs
@@ -932,11 +932,21 @@ pub struct CopyOptions {
/// this option.
pub source_version: Option<String>,
- /// Known content length of the source object.
+ /// Asserted complete content length of the source object.
///
- /// This is an execution hint that allows OpenDAL to avoid extra metadata
- /// requests while planning copy operations. It must not be used as an object
- /// identity or consistency condition.
+ /// OpenDAL may trust this value without reading source metadata. It can use
+ /// the value to plan copied ranges and to construct the returned metadata.
+ /// An incorrect value can therefore cause incomplete copy planning, failed
+ /// requests, or incorrect result metadata.
+ ///
+ /// A service can ignore this value when its copy operation already reports
+ /// an authoritative copied size.
+ ///
+ /// This option does not pin the source object or protect the copy from a
+ /// concurrent source change. Set it only when the caller can guarantee that
+ /// the value describes the object that the service will copy, for example by
+ /// selecting an immutable [`Self::source_version`]. Otherwise omit it if the
+ /// source can change.
pub source_content_length_hint: Option<u64>,
/// Sets concurrent copy operations for this copier.
diff --git a/core/services/aliyun-drive/src/backend.rs b/core/services/aliyun-drive/src/backend.rs
index 7b27560fb..7e5dbb31f 100644
--- a/core/services/aliyun-drive/src/backend.rs
+++ b/core/services/aliyun-drive/src/backend.rs
@@ -253,21 +253,29 @@ impl Service for AliyunDriveBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let res = core.get_by_path(&ctx, &from).await?;
+ let file: AliyunDriveFile =
+ serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
+ let source_content_length =
+ file.size.or(source_content_length_hint).ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "Aliyun Drive source file does not contain a size",
+ )
+ })?;
+
if from == to {
- Ok(MetadataBuilder::unknown().build())
+ Ok(MetadataBuilder::file(source_content_length).build())
} else {
- let res = core.get_by_path(&ctx, &from).await?;
- let file: AliyunDriveFile =
- serde_json::from_reader(res.reader()).map_err(new_json_serialize_error)?;
- let source_content_length = file.size;
// copy can overwrite.
match core.get_by_path(&ctx, &to).await {
Err(err) if err.kind() == ErrorKind::NotFound => {}
@@ -301,9 +309,7 @@ impl Service for AliyunDriveBackend {
core.update_path(&ctx, &file_id, to_name).await?;
}
- let metadata = source_content_length
- .map_or_else(MetadataBuilder::unknown, MetadataBuilder::file);
- Ok(metadata.build())
+ Ok(MetadataBuilder::file(source_content_length).build())
}
}))
}
diff --git a/core/services/dropbox/src/backend.rs b/core/services/dropbox/src/backend.rs
index 67572d1bc..44dffa9e5 100644
--- a/core/services/dropbox/src/backend.rs
+++ b/core/services/dropbox/src/backend.rs
@@ -352,13 +352,10 @@ impl Service for DropboxBackend {
.map_err(new_json_deserialize_error)?;
DropboxWriter::parse_metadata(decoded_response)
}
- _ => {
- let err = parse_error(ErrorContext::new(ServiceOperation("CopyFile")), resp);
- match err.kind() {
- ErrorKind::NotFound => Ok(MetadataBuilder::unknown().build()),
- _ => Err(err),
- }
- }
+ _ => Err(parse_error(
+ ErrorContext::new(ServiceOperation("CopyFile")),
+ resp,
+ )),
}
}))
}
diff --git a/core/services/koofr/src/backend.rs b/core/services/koofr/src/backend.rs
index d70e7fd62..dcb9bb570 100644
--- a/core/services/koofr/src/backend.rs
+++ b/core/services/koofr/src/backend.rs
@@ -298,17 +298,28 @@ impl Service for KoofrBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
core.ensure_dir_exists(&ctx, &to).await?;
if from == to {
- Ok(MetadataBuilder::unknown().build())
+ Ok(MetadataBuilder::file(source_size).build())
} else {
let resp = core.remove(&ctx, &to).await?;
@@ -325,7 +336,7 @@ impl Service for KoofrBackend {
let status = resp.status();
match status {
- StatusCode::OK => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("FilesCopy")),
resp,
diff --git a/core/services/lakefs/src/backend.rs b/core/services/lakefs/src/backend.rs
index 79a89b4eb..331b0a998 100644
--- a/core/services/lakefs/src/backend.rs
+++ b/core/services/lakefs/src/backend.rs
@@ -299,19 +299,30 @@ impl Service for LakefsBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
let resp = core.copy_object(&ctx, &from, &to).await?;
let status = resp.status();
match status {
- StatusCode::CREATED => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::CREATED => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyObject")),
resp,
diff --git a/core/services/obs/src/backend.rs b/core/services/obs/src/backend.rs
index b8f4c8136..7b45ae1e5 100644
--- a/core/services/obs/src/backend.rs
+++ b/core/services/obs/src/backend.rs
@@ -399,19 +399,30 @@ impl Service for ObsBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
let resp = core.obs_copy_object(&ctx, &from, &to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyObject")),
resp,
diff --git a/core/services/onedrive/src/backend.rs b/core/services/onedrive/src/backend.rs
index 629ec7b9b..317a6f9f2 100644
--- a/core/services/onedrive/src/backend.rs
+++ b/core/services/onedrive/src/backend.rs
@@ -300,17 +300,28 @@ impl Service for OnedriveBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
let monitor_url = core.initialize_copy(&ctx, &from, &to).await?;
core.wait_until_complete(&ctx, monitor_url).await?;
- Ok(MetadataBuilder::unknown().build())
+ Ok(MetadataBuilder::file(source_size).build())
}))
}
diff --git a/core/services/pcloud/src/backend.rs b/core/services/pcloud/src/backend.rs
index 897314b77..6dd55281d 100644
--- a/core/services/pcloud/src/backend.rs
+++ b/core/services/pcloud/src/backend.rs
@@ -279,14 +279,29 @@ impl Service for PcloudBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = if from.ends_with('/') {
+ None
+ } else {
+ Some(match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ })
+ };
+
core.ensure_dir_exists(&ctx, &to).await?;
let resp = if from.ends_with('/') {
@@ -308,7 +323,9 @@ impl Service for PcloudBackend {
} else if result != 0 {
Err(Error::new(ErrorKind::Unexpected, format!("{resp:?}")))
} else {
- Ok(MetadataBuilder::unknown().build())
+ let metadata =
+ source_size.map_or_else(MetadataBuilder::dir, MetadataBuilder::file);
+ Ok(metadata.build())
}
}
_ => Err(parse_error(
diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs
index b432b436c..858468385 100644
--- a/core/services/s3/src/backend.rs
+++ b/core/services/s3/src/backend.rs
@@ -1470,6 +1470,8 @@ mod tests {
#[derive(Clone)]
struct S3ExpressMockTransport {
outcome: CreateSessionOutcome,
+ source_content_length: u64,
+ source_version: Option<&'static str>,
requests: Arc<Mutex<Vec<CapturedRequest>>>,
}
@@ -1477,10 +1479,22 @@ mod tests {
fn new(outcome: CreateSessionOutcome) -> Self {
Self {
outcome,
+ source_content_length: 1,
+ source_version: None,
requests: Arc::new(Mutex::new(Vec::new())),
}
}
+ fn with_source_content_length(mut self, source_content_length: u64) -> Self {
+ self.source_content_length = source_content_length;
+ self
+ }
+
+ fn with_source_version(mut self, source_version: &'static str) -> Self {
+ self.source_version = Some(source_version);
+ self
+ }
+
fn requests(&self) -> Vec<CapturedRequest> {
self.requests.lock().expect("lock poisoned").clone()
}
@@ -1537,6 +1551,32 @@ mod tests {
};
}
+ if req.method() == Method::HEAD {
+ let mut response = Self::response(StatusCode::OK, Bytes::new());
+ response.headers_mut().insert(
+ http::header::CONTENT_LENGTH,
+ self.source_content_length
+ .to_string()
+ .parse()
+ .expect("source content length must be a valid header"),
+ );
+ response.headers_mut().insert(
+ http::header::ETAG,
+ "\"source-etag\""
+ .parse()
+ .expect("source ETag must be a valid header"),
+ );
+ if let Some(version) = self.source_version {
+ response.headers_mut().insert(
+ constants::X_AMZ_VERSION_ID,
+ version
+ .parse()
+ .expect("source version must be a valid header"),
+ );
+ }
+ return Ok(response);
+ }
+
let query = req.uri().query().unwrap_or_default();
if req.headers().contains_key(constants::X_AMZ_COPY_SOURCE) {
return Ok(Self::response(
@@ -1707,6 +1747,98 @@ mod tests {
}
}
+ #[tokio::test]
+ async fn test_s3_copy_hint_skips_source_snapshot() {
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
+ let op = s3_general_operator(transport.clone());
+
+ let metadata = op
+ .copy_with("source", "target")
+ .source_content_length_hint(7)
+ .await
+ .expect("copy must succeed");
+
+ assert_eq!(metadata.content_length(), 7);
+ let requests = transport.requests();
+ assert_eq!(requests.len(), 1);
+ assert_eq!(requests[0].method, Method::PUT);
+ assert!(
+ !requests[0]
+ .headers
+ .contains_key(constants::X_AMZ_COPY_SOURCE_IF_MATCH)
+ );
+ }
+
+ #[tokio::test]
+ async fn test_s3_copy_hint_preserves_explicit_source_version() {
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
+ let op = s3_general_operator(transport.clone());
+
+ op.copy_with("source", "target")
+ .source_version("source-version")
+ .source_content_length_hint(7)
+ .await
+ .expect("copy must succeed");
+
+ let requests = transport.requests();
+ assert_eq!(requests.len(), 1);
+ assert!(
+ header(&requests[0], constants::X_AMZ_COPY_SOURCE).contains("versionId=source-version")
+ );
+ }
+
+ #[tokio::test]
+ async fn test_s3_copy_without_hint_pins_source_etag() {
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success)
+ .with_source_content_length(7);
+ let op = s3_general_operator(transport.clone());
+
+ let metadata = op
+ .copy("source", "target")
+ .await
+ .expect("copy must succeed");
+
+ assert_eq!(metadata.content_length(), 7);
+ let requests = transport.requests();
+ assert_eq!(
+ requests
+ .iter()
+ .filter(|request| request.method == Method::HEAD)
+ .count(),
+ 1
+ );
+ let copy = requests
+ .iter()
+ .find(|request| request.headers.contains_key(constants::X_AMZ_COPY_SOURCE))
+ .expect("CopyObject request must be captured");
+ assert_eq!(
+ header(copy, constants::X_AMZ_COPY_SOURCE_IF_MATCH),
+ "\"source-etag\""
+ );
+ }
+
+ #[tokio::test]
+ async fn test_s3_copy_uses_version_from_source_snapshot() {
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success)
+ .with_source_version("source-version");
+ let op = s3_general_operator(transport.clone());
+
+ op.copy_with("source", "target")
+ .await
+ .expect("copy must succeed");
+
+ let requests = transport.requests();
+ let copy = requests
+ .iter()
+ .find(|request| request.headers.contains_key(constants::X_AMZ_COPY_SOURCE))
+ .expect("CopyObject request must be captured");
+ assert!(header(copy, constants::X_AMZ_COPY_SOURCE).contains("versionId=source-version"));
+ assert_eq!(
+ header(copy, constants::X_AMZ_COPY_SOURCE_IF_MATCH),
+ "\"source-etag\""
+ );
+ }
+
#[tokio::test]
async fn test_s3_express_copy_and_presign_use_iam_credentials() {
let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
@@ -1723,7 +1855,7 @@ mod tests {
.filter(|request| request.uri.query() == Some("session"))
.count(),
0,
- "CopyObject must not create an S3 Express session"
+ "a source length hint must skip HeadObject"
);
let copy = requests
.iter()
@@ -1732,6 +1864,11 @@ mod tests {
assert_eq!(header(copy, "x-amz-security-token"), "source-session-token");
assert!(!copy.headers.contains_key("x-amz-s3session-token"));
assert!(header(copy, "authorization").contains("source-access-key/"));
+ assert!(
+ !copy
+ .headers
+ .contains_key(constants::X_AMZ_COPY_SOURCE_IF_MATCH)
+ );
let presigned = op
.presign_read("source", Duration::from_secs(60))
@@ -1747,12 +1884,14 @@ mod tests {
#[tokio::test]
async fn test_s3_express_upload_part_copy_uses_iam_credentials() {
- let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
+ let source_content_length = 5 * 1024 * 1024 + 1;
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success)
+ .with_source_content_length(source_content_length);
let op = s3_express_operator(transport.clone());
op.copy_with("source", "target")
.chunk(5 * 1024 * 1024)
- .source_content_length_hint(5 * 1024 * 1024 + 1)
+ .source_content_length_hint(source_content_length)
.await
.expect("multipart copy must succeed");
@@ -1775,6 +1914,11 @@ mod tests {
);
assert!(!request.headers.contains_key("x-amz-s3session-token"));
assert!(header(request, "authorization").contains("source-access-key/"));
+ assert!(
+ !request
+ .headers
+ .contains_key(constants::X_AMZ_COPY_SOURCE_IF_MATCH)
+ );
}
assert_eq!(
requests
@@ -1786,6 +1930,38 @@ mod tests {
);
}
+ #[tokio::test]
+ async fn test_s3_multipart_copy_without_hint_pins_every_part() {
+ let source_content_length = 5 * 1024 * 1024 + 1;
+ let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success)
+ .with_source_content_length(source_content_length);
+ let op = s3_general_operator(transport.clone());
+
+ op.copy_with("source", "target")
+ .chunk(5 * 1024 * 1024)
+ .await
+ .expect("multipart copy must succeed");
+
+ let requests = transport.requests();
+ let part_copies = requests
+ .iter()
+ .filter(|request| {
+ request.headers.contains_key(constants::X_AMZ_COPY_SOURCE)
+ && request
+ .uri
+ .query()
+ .is_some_and(|query| query.contains("partNumber="))
+ })
+ .collect::<Vec<_>>();
+ assert_eq!(part_copies.len(), 2);
+ for request in part_copies {
+ assert_eq!(
+ header(request, constants::X_AMZ_COPY_SOURCE_IF_MATCH),
+ "\"source-etag\""
+ );
+ }
+ }
+
#[tokio::test]
async fn test_s3_express_writer_copy_from_uses_iam_credentials() {
let transport = S3ExpressMockTransport::new(CreateSessionOutcome::Success);
diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs
index c572cc79c..8e7344804 100644
--- a/core/services/s3/src/copier.rs
+++ b/core/services/s3/src/copier.rs
@@ -16,6 +16,7 @@
// under the License.
use std::sync::Arc;
+use std::sync::OnceLock;
use bytes::Buf;
use http::Response;
@@ -28,6 +29,10 @@ use opendal_core::*;
pub type S3Copiers = oio::MultipartCopier<S3Copier>;
+fn is_immutable_source_version(version: &str) -> bool {
+ !version.is_empty() && version != "null"
+}
+
pub fn new_s3_copier(
core: Arc<S3Core>,
ctx: &OperationContext,
@@ -70,6 +75,7 @@ pub fn new_s3_copier(
from: from.to_string(),
to: to.to_string(),
args,
+ source_snapshot: OnceLock::new(),
},
source_content_length_hint,
copy_once_threshold,
@@ -84,19 +90,57 @@ pub struct S3Copier {
from: String,
to: String,
args: OpCopy,
+ // An explicit source length hint intentionally skips loading this snapshot.
+ source_snapshot: OnceLock<Metadata>,
+}
+
+struct S3CopySource<'a> {
+ version: Option<&'a str>,
+ if_match: Option<&'a str>,
}
impl S3Copier {
- fn error_context(&self, service_operation: ServiceOperation) -> ErrorContext {
+ fn error_context(
+ &self,
+ service_operation: ServiceOperation,
+ source_if_match: bool,
+ ) -> ErrorContext {
ErrorContext::new(service_operation)
- .with_if_match(self.args.if_match().is_some())
+ .with_if_match(self.args.if_match().is_some() || source_if_match)
.with_if_none_match(self.args.if_none_match().is_some())
.with_if_not_exists(self.args.if_not_exists())
}
+
+ fn copy_source(&self) -> Result<S3CopySource<'_>> {
+ let snapshot = self.source_snapshot.get();
+ let version = self.args.source_version().or_else(|| {
+ snapshot
+ .and_then(Metadata::version)
+ .filter(|version| is_immutable_source_version(version))
+ });
+ let if_match = snapshot.and_then(Metadata::etag);
+
+ if version.is_none()
+ && if_match.is_none()
+ && self.args.source_content_length_hint().is_none()
+ {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "S3 copy source does not have an immutable version or ETag",
+ )
+ .with_operation("S3Copier::copy_source"));
+ }
+
+ Ok(S3CopySource { version, if_match })
+ }
}
impl oio::MultipartCopy for S3Copier {
async fn source_metadata(&self) -> Result<Metadata> {
+ if let Some(metadata) = self.source_snapshot.get() {
+ return Ok(metadata.clone());
+ }
+
let args = options::StatOptions {
version: self.args.source_version().map(str::to_owned),
..Default::default()
@@ -111,7 +155,15 @@ impl oio::MultipartCopy for S3Copier {
match resp.status() {
StatusCode::OK => {
let headers = resp.headers();
- parse_into_metadata(&self.from, headers)
+ let mut metadata = parse_into_metadata(&self.from, headers)?.into_builder();
+ if let Some(version) = parse_header_to_str(headers, constants::X_AMZ_VERSION_ID)?
+ .or(self.args.source_version())
+ {
+ metadata.version(version);
+ }
+ let metadata = metadata.build();
+ let _ = self.source_snapshot.set(metadata.clone());
+ Ok(self.source_snapshot.get().cloned().unwrap_or(metadata))
}
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("HeadObject")),
@@ -121,9 +173,17 @@ impl oio::MultipartCopy for S3Copier {
}
async fn copy_once(&self) -> Result<Metadata> {
+ let source = self.copy_source()?;
let resp = self
.core
- .s3_copy_object(&self.ctx, &self.from, &self.to, &self.args)
+ .s3_copy_object(
+ &self.ctx,
+ &self.from,
+ &self.to,
+ source.version,
+ source.if_match,
+ &self.args,
+ )
.await?;
match resp.status() {
@@ -139,7 +199,10 @@ impl oio::MultipartCopy for S3Copier {
// S3 may return 200 OK with an <Error> body for CopyObject.
if result.etag.is_empty() {
let err = parse_error(
- self.error_context(ServiceOperation("CopyObject")),
+ self.error_context(
+ ServiceOperation("CopyObject"),
+ source.if_match.is_some(),
+ ),
Response::from_parts(parts, Buffer::from(bs)),
);
return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
@@ -168,7 +231,10 @@ impl oio::MultipartCopy for S3Copier {
Ok(meta.build())
}
_ => {
- let err = parse_error(self.error_context(ServiceOperation("CopyObject")), resp);
+ let err = parse_error(
+ self.error_context(ServiceOperation("CopyObject"), source.if_match.is_some()),
+ resp,
+ );
if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
Err(Error::new(
ErrorKind::ConditionNotMatch,
@@ -211,14 +277,15 @@ impl oio::MultipartCopy for S3Copier {
) -> Result<oio::MultipartPart> {
let size = range.size().expect("multipart copy range must be sized");
let part_number = part_number + 1;
+ let source = self.copy_source()?;
let req = self
.core
.s3_upload_part_copy_request(S3UploadPartCopyRequest {
from: &self.from,
to: &self.to,
- source_version: self.args.source_version(),
- if_match: None,
+ source_version: source.version,
+ if_match: source.if_match,
if_none_match: None,
if_modified_since: None,
if_unmodified_since: None,
@@ -243,7 +310,10 @@ impl oio::MultipartCopy for S3Copier {
// S3 may return 200 OK with an <Error> body for UploadPartCopy.
if result.etag.is_empty() {
return Err(parse_error(
- ErrorContext::new(ServiceOperation("UploadPartCopy")),
+ self.error_context(
+ ServiceOperation("UploadPartCopy"),
+ source.if_match.is_some(),
+ ),
Response::from_parts(parts, Buffer::from(bs)),
));
}
@@ -256,7 +326,10 @@ impl oio::MultipartCopy for S3Copier {
})
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("UploadPartCopy")),
+ self.error_context(
+ ServiceOperation("UploadPartCopy"),
+ source.if_match.is_some(),
+ ),
resp,
)),
}
@@ -295,7 +368,7 @@ impl oio::MultipartCopy for S3Copier {
// S3 may return 200 OK with an <Error> body for CompleteMultipartUpload.
if ret.etag.is_empty() {
let err = parse_error(
- self.error_context(ServiceOperation("CompleteMultipartUpload")),
+ self.error_context(ServiceOperation("CompleteMultipartUpload"), false),
Response::from_parts(parts, Buffer::from(bs)),
);
return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
@@ -322,7 +395,7 @@ impl oio::MultipartCopy for S3Copier {
}
_ => {
let err = parse_error(
- self.error_context(ServiceOperation("CompleteMultipartUpload")),
+ self.error_context(ServiceOperation("CompleteMultipartUpload"), false),
resp,
);
if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs
index 4d007301d..3106141c5 100644
--- a/core/services/s3/src/core.rs
+++ b/core/services/s3/src/core.rs
@@ -776,13 +776,15 @@ impl S3Core {
ctx: &OperationContext,
from: &str,
to: &str,
+ source_version: Option<&str>,
+ source_if_match: Option<&str>,
args: &OpCopy,
) -> Result<Response<Buffer>> {
let from = build_abs_path(&self.root, from);
let to = build_abs_path(&self.root, to);
let source = format!("{}/{}", self.bucket, percent_encode_path(&from));
- let source = if let Some(version) = args.source_version() {
+ let source = if let Some(version) = source_version {
QueryPairsWriter::new(&source)
.push(
constants::S3_QUERY_VERSION_ID,
@@ -803,6 +805,9 @@ impl S3Core {
if let Some(if_match) = args.if_match() {
req = req.header(IF_MATCH, if_match);
}
+ if let Some(source_if_match) = source_if_match {
+ req = req.header(constants::X_AMZ_COPY_SOURCE_IF_MATCH, source_if_match);
+ }
// Set SSE headers.
req = self.insert_sse_headers(req, true);
diff --git a/core/services/upyun/src/backend.rs b/core/services/upyun/src/backend.rs
index 552935ad0..4e6e6f63c 100644
--- a/core/services/upyun/src/backend.rs
+++ b/core/services/upyun/src/backend.rs
@@ -281,19 +281,30 @@ impl Service for UpyunBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
let resp = core.copy(&ctx, &from, &to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyFile")),
resp,
diff --git a/core/services/vercel-blob/src/backend.rs b/core/services/vercel-blob/src/backend.rs
index 5b9083b95..ece424e75 100644
--- a/core/services/vercel-blob/src/backend.rs
+++ b/core/services/vercel-blob/src/backend.rs
@@ -211,19 +211,30 @@ impl Service for VercelBlobBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
let resp = core.copy(&ctx, &from, &to).await?;
let status = resp.status();
match status {
- StatusCode::OK => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyBlob")),
resp,
diff --git a/core/services/webdav/src/backend.rs b/core/services/webdav/src/backend.rs
index 0cf59f8f7..ddeed390b 100644
--- a/core/services/webdav/src/backend.rs
+++ b/core/services/webdav/src/backend.rs
@@ -351,26 +351,38 @@ impl Service for WebdavBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new_with(move || {
+ let backend = backend.clone();
let core = core.clone();
let ctx = ctx.clone();
let from = from.clone();
let to = to.clone();
async move {
- let (resp, source_metadata) = core.webdav_copy(&ctx, &from, &to).await?;
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
+ let resp = core.webdav_copy(&ctx, &from, &to).await?;
let status = resp.status();
match status {
StatusCode::CREATED | StatusCode::NO_CONTENT => {
- Ok(MetadataBuilder::file(source_metadata.content_length()).build())
+ Ok(MetadataBuilder::file(source_size).build())
}
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("Copy")),
diff --git a/core/services/webdav/src/core.rs b/core/services/webdav/src/core.rs
index 141d63e04..3d4ab2fc0 100644
--- a/core/services/webdav/src/core.rs
+++ b/core/services/webdav/src/core.rs
@@ -326,9 +326,7 @@ impl WebdavCore {
ctx: &OperationContext,
from: &str,
to: &str,
- ) -> Result<(Response<Buffer>, Metadata)> {
- // Check if source file exists.
- let source_metadata = self.webdav_stat(ctx, from).await?;
+ ) -> Result<Response<Buffer>> {
// Make sure target's dir is exist.
self.webdav_mkcol(ctx, get_parent(to)).await?;
@@ -353,8 +351,7 @@ impl WebdavCore {
.body(Buffer::new())
.map_err(new_request_build_error)?;
- let resp = ctx.http_transport().send(req).await?;
- Ok((resp, source_metadata))
+ ctx.http_transport().send(req).await
}
pub async fn webdav_move(
diff --git a/core/services/yandex-disk/src/backend.rs b/core/services/yandex-disk/src/backend.rs
index 3bf8ba709..a546ddc98 100644
--- a/core/services/yandex-disk/src/backend.rs
+++ b/core/services/yandex-disk/src/backend.rs
@@ -184,14 +184,25 @@ impl Service for YandexDiskBackend {
ctx: &OperationContext,
from: &str,
to: &str,
- _args: OpCopy,
+ args: OpCopy,
) -> Result<Self::Copier> {
+ let backend = self.clone();
let core = self.core.clone();
let ctx = ctx.clone();
let from = from.to_string();
let to = to.to_string();
+ let source_content_length_hint = args.source_content_length_hint();
Ok(oio::OneShotCopier::new(async move {
+ let source_size = match source_content_length_hint {
+ Some(size) => size,
+ None => backend
+ .stat(&ctx, &from, OpStat::default())
+ .await?
+ .into_metadata()
+ .content_length(),
+ };
+
core.ensure_dir_exists(&ctx, &to).await?;
let resp = core.copy(&ctx, &from, &to).await?;
@@ -199,7 +210,9 @@ impl Service for YandexDiskBackend {
let status = resp.status();
match status {
- StatusCode::OK | StatusCode::CREATED => Ok(MetadataBuilder::unknown().build()),
+ StatusCode::OK | StatusCode::CREATED => {
+ Ok(MetadataBuilder::file(source_size).build())
+ }
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyResource")),
resp,