fix(services/oss): honor if_not_exists on multipart uploads

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<Response<Buffer>> {
         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<MultipartUploadPart>,
+        args: &OpWrite,
     ) -> Result<Response<Buffer>> {
         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)

core/services/oss/src/writer.rs

@@ -83,15 +83,7 @@ impl oio::MultipartWrite for OssWriter {
     async fn initiate_part(&self) -> Result<String> {
         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())?;

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));