From 780dd849e4d7151ce884bc8ec431b6144f2cd607 Mon Sep 17 00:00:00 2001 From: Xuanwo Date: Sun, 9 Aug 2026 13:30:05 +0800 Subject: [PATCH] fix(services/oss): honor if_not_exists on multipart uploads OSS requires x-oss-forbid-overwrite on both InitiateMultipartUpload and CompleteMultipartUpload. Also skip the chunked-writer GCS coverage that the XML multipart API cannot satisfy, tracked in #8040. --- core/services/oss/src/core.rs | 35 ++++++++++++++++++++---------- core/services/oss/src/writer.rs | 14 ++++-------- core/tests/behavior/async_write.rs | 8 +++++++ 3 files changed, 35 insertions(+), 22 deletions(-) diff --git a/core/services/oss/src/core.rs b/core/services/oss/src/core.rs index 832ca3cc522a..2f794f0ff143 100644 --- a/core/services/oss/src/core.rs +++ b/core/services/oss/src/core.rs @@ -657,33 +657,36 @@ impl OssCore { } } - #[allow(clippy::too_many_arguments)] pub async fn oss_initiate_upload( &self, ctx: &OperationContext, path: &str, - content_type: Option<&str>, - content_disposition: Option<&str>, - cache_control: Option<&str>, - content_encoding: Option<&str>, + args: &OpWrite, is_presign: bool, ) -> Result> { let path = build_abs_path(&self.root, path); let endpoint = self.get_endpoint(is_presign); let url = format!("{}/{}?uploads", endpoint, percent_encode_path(&path)); let mut req = Request::post(&url); - if let Some(mime) = content_type { + if let Some(mime) = args.content_type() { req = req.header(CONTENT_TYPE, mime); } - if let Some(disposition) = content_disposition { + if let Some(disposition) = args.content_disposition() { req = req.header(CONTENT_DISPOSITION, disposition); } - if let Some(cache_control) = cache_control { + if let Some(cache_control) = args.cache_control() { req = req.header(CACHE_CONTROL, cache_control); } - if let Some(encoding) = content_encoding { + if let Some(encoding) = args.content_encoding() { req = req.header(CONTENT_ENCODING, encoding); } + // OSS evaluates x-oss-forbid-overwrite on both InitiateMultipartUpload and + // CompleteMultipartUpload. Setting it only on one of them is not enough. + // + // ref: https://www.alibabacloud.com/help/en/oss/developer-reference/initiatemultipartupload + if args.if_not_exists() { + req = req.header(X_OSS_FORBID_OVERWRITE, "true"); + } req = self.insert_sse_headers(req); let req = req @@ -737,6 +740,7 @@ impl OssCore { upload_id: &str, is_presign: bool, parts: Vec, + args: &OpWrite, ) -> Result> { let p = build_abs_path(&self.root, path); let endpoint = self.get_endpoint(is_presign); @@ -747,16 +751,23 @@ impl OssCore { percent_encode_path(upload_id) ); - let req = Request::post(&url); + let mut req = Request::post(&url); let content = quick_xml::se::to_string(&CompleteMultipartUploadRequest { part: parts.to_vec(), }) .map_err(new_xml_serialize_error)?; // Make sure content length has been set to avoid post with chunked encoding. - let req = req.header(CONTENT_LENGTH, content.len()); + req = req.header(CONTENT_LENGTH, content.len()); // Set content-type to `application/xml` to avoid mixed with form post. - let req = req.header(CONTENT_TYPE, "application/xml"); + req = req.header(CONTENT_TYPE, "application/xml"); + // CompleteMultipartUpload is the request that commits the object, so + // if_not_exists must also be enforced here. + // + // ref: https://www.alibabacloud.com/help/en/oss/developer-reference/completemultipartupload + if args.if_not_exists() { + req = req.header(X_OSS_FORBID_OVERWRITE, "true"); + } let req = req .extension(Operation::Write) diff --git a/core/services/oss/src/writer.rs b/core/services/oss/src/writer.rs index 9a150a3256a4..95f0bc5ff6dd 100644 --- a/core/services/oss/src/writer.rs +++ b/core/services/oss/src/writer.rs @@ -83,15 +83,7 @@ impl oio::MultipartWrite for OssWriter { async fn initiate_part(&self) -> Result { let resp = self .core - .oss_initiate_upload( - &self.ctx, - &self.path, - self.op.content_type(), - self.op.content_disposition(), - self.op.cache_control(), - self.op.content_encoding(), - false, - ) + .oss_initiate_upload(&self.ctx, &self.path, &self.op, false) .await?; let status = resp.status(); @@ -172,7 +164,9 @@ impl oio::MultipartWrite for OssWriter { let resp = self .core - .oss_complete_multipart_upload_request(&self.ctx, &self.path, upload_id, false, parts) + .oss_complete_multipart_upload_request( + &self.ctx, &self.path, upload_id, false, parts, &self.op, + ) .await?; let meta = Self::parse_metadata(resp.headers())?; diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs index 267bcd4c3150..5a17dc69cf82 100644 --- a/core/tests/behavior/async_write.rs +++ b/core/tests/behavior/async_write.rs @@ -821,6 +821,14 @@ pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> { return Ok(()); } + // GCS XML API multipart uploads do not support preconditions, so the multipart + // writer path cannot honor if_not_exists. Tracked in + // https://github.com/apache/opendal/issues/8040 + #[cfg(feature = "services-gcs")] + if op.info().scheme() == services::GCS_SCHEME { + return Ok(()); + } + let path = TEST_FIXTURE.new_file_path(); let content = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1));