feat(core): add if-match to OpWrite (#5360)

.github/services/s3/ceph_radios_s3_with_versioning/disable_action.yml

@@ -44,4 +44,5 @@ runs:
         OPENDAL_S3_SECRET_ACCESS_KEY=demo
         OPENDAL_S3_REGION=us-east-1
         OPENDAL_S3_ENABLE_VERSIONING=true
+        OPENDAL_S3_DISABLE_WRITE_WITH_IF_MATCH=on
         EOF

.github/services/s3/ceph_rados_s3/action.yml

@@ -41,4 +41,5 @@ runs:
         OPENDAL_S3_ACCESS_KEY_ID=demo
         OPENDAL_S3_SECRET_ACCESS_KEY=demo
         OPENDAL_S3_REGION=us-east-1
+        OPENDAL_S3_DISABLE_WRITE_WITH_IF_MATCH=on
         EOF

core/src/raw/ops.rs

@@ -577,6 +577,7 @@ pub struct OpWrite {
     content_disposition: Option<String>,
     cache_control: Option<String>,
     executor: Option<Executor>,
+    if_match: Option<String>,
     if_none_match: Option<String>,
     if_not_exists: bool,
     user_metadata: Option<HashMap<String, String>>,
@@ -664,6 +665,17 @@ impl OpWrite {
         self
     }
 
+    /// Set the If-Match of the option
+    pub fn with_if_match(mut self, s: &str) -> Self {
+        self.if_match = Some(s.to_string());
+        self
+    }
+
+    /// Get If-Match from option
+    pub fn if_match(&self) -> Option<&str> {
+        self.if_match.as_deref()
+    }
+
     /// Set the If-None-Match of the option
     pub fn with_if_none_match(mut self, s: &str) -> Self {
         self.if_none_match = Some(s.to_string());

core/src/services/s3/backend.rs

@@ -548,6 +548,12 @@ impl S3Builder {
         self
     }
 
+    /// Disable write with if match so that opendal will not send write request with if match headers.
+    pub fn disable_write_with_if_match(mut self) -> Self {
+        self.config.disable_write_with_if_match = true;
+        self
+    }
+
     /// Detect region of S3 bucket.
     ///
     /// # Args
@@ -878,6 +884,7 @@ impl Builder for S3Builder {
                 client,
                 batch_max_operations,
                 checksum_algorithm,
+                disable_write_with_if_match: self.config.disable_write_with_if_match,
             }),
         })
     }
@@ -924,6 +931,7 @@ impl Access for S3Backend {
                 write_can_multi: true,
                 write_with_cache_control: true,
                 write_with_content_type: true,
+                write_with_if_match: !self.core.disable_write_with_if_match,
                 write_with_if_not_exists: true,
                 write_with_user_metadata: true,
 

core/src/services/s3/config.rs

@@ -173,6 +173,10 @@ pub struct S3Config {
     /// Available options:
     /// - "crc32c"
     pub checksum_algorithm: Option<String>,
+    /// Disable write with if match so that opendal will not send write request with if match headers.
+    ///
+    /// For example, Ceph RADOS S3 doesn't support write with if match.
+    pub disable_write_with_if_match: bool,
 }
 
 impl Debug for S3Config {

core/src/services/s3/core.rs

@@ -98,6 +98,7 @@ pub struct S3Core {
     pub client: HttpClient,
     pub batch_max_operations: usize,
     pub checksum_algorithm: Option<ChecksumAlgorithm>,
+    pub disable_write_with_if_match: bool,
 }
 
 impl Debug for S3Core {
@@ -455,6 +456,10 @@ impl S3Core {
             req = req.header(CACHE_CONTROL, cache_control)
         }
 
+        if let Some(if_match) = args.if_match() {
+            req = req.header(IF_MATCH, if_match);
+        }
+
         if args.if_not_exists() {
             req = req.header(IF_NONE_MATCH, "*");
         }

core/src/services/s3/docs.md

@@ -28,8 +28,9 @@ This service can be used to:
 - `server_side_encryption_customer_algorithm`: Set the server_side_encryption_customer_algorithm for backend.
 - `server_side_encryption_customer_key`: Set the server_side_encryption_customer_key for backend.
 - `server_side_encryption_customer_key_md5`: Set the server_side_encryption_customer_key_md5 for backend.
-- `disable_config_load`: Disable aws config load from env
+- `disable_config_load`: Disable aws config load from env.
 - `enable_virtual_host_style`: Enable virtual host style.
+- `disable_write_with_if_match`: Disable write with if match.
 
 Refer to [`S3Builder`]'s public API docs for more information.
 

core/src/types/capability.rs

@@ -128,6 +128,8 @@ pub struct Capability {
     pub write_with_content_disposition: bool,
     /// Indicates if Cache-Control can be specified during write operations.
     pub write_with_cache_control: bool,
+    /// Indicates if conditional write operations using If-Match are supported.
+    pub write_with_if_match: bool,
     /// Indicates if conditional write operations using If-None-Match are supported.
     pub write_with_if_none_match: bool,
     /// Indicates if write operations can be conditional on object non-existence.

core/src/types/operator/operator.rs

@@ -1473,6 +1473,36 @@ impl Operator {
     /// # Ok(())
     /// # }
     /// ```
+    ///
+    /// ## `if_match`
+    ///
+    /// Sets an `if match` condition with specified ETag for this write request.
+    ///
+    /// ### Capability
+    ///
+    /// Check [`Capability::write_with_if_match`] before using this feature.
+    ///
+    /// ### Behavior
+    ///
+    /// - If the target file's ETag matches the specified one, proceeds with the write operation
+    /// - If the target file's ETag does not match the specified one, returns [`ErrorKind::ConditionNotMatch`]
+    ///
+    /// This operation will succeed when the target's ETag matches the specified one,
+    /// providing a way for conditional writes.
+    ///
+    /// ### Example
+    ///
+    /// ```no_run
+    /// # use opendal::{ErrorKind, Result};
+    /// use opendal::Operator;
+    /// # async fn test(op: Operator, incorrect_etag: &str) -> Result<()> {
+    /// let bs = b"hello, world!".to_vec();
+    /// let res = op.write_with("path/to/file", bs).if_match(incorrect_etag).await;
+    /// assert!(res.is_err());
+    /// assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+    /// # Ok(())
+    /// # }
+    /// ```
     pub fn write_with(
         &self,
         path: &str,

core/src/types/operator/operator_futures.rs

@@ -323,6 +323,11 @@ impl<F: Future<Output = Result<()>>> FutureWrite<F> {
         self.map(|(args, options, bs)| (args.with_executor(executor), options, bs))
     }
 
+    /// Set the If-Match for this operation.
+    pub fn if_match(self, s: &str) -> Self {
+        self.map(|(args, options, bs)| (args.with_if_match(s), options, bs))
+    }
+
     /// Set the If-None-Match for this operation.
     pub fn if_none_match(self, s: &str) -> Self {
         self.map(|(args, options, bs)| (args.with_if_none_match(s), options, bs))

core/tests/behavior/async_write.rs

@@ -46,6 +46,7 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_write_with_content_disposition,
             test_write_with_if_none_match,
             test_write_with_if_not_exists,
+            test_write_with_if_match,
             test_write_with_user_metadata,
             test_writer_write,
             test_writer_write_with_overwrite,
@@ -674,3 +675,41 @@ pub async fn test_write_with_if_not_exists(op: Operator) -> Result<()> {
 
     Ok(())
 }
+
+/// Write an file with if_match will get a ConditionNotMatch error if file's etag does not match.
+pub async fn test_write_with_if_match(op: Operator) -> Result<()> {
+    if !op.info().full_capability().write_with_if_match {
+        return Ok(());
+    }
+
+    // Create two different files with different content
+    let (path_a, content_a, _) = TEST_FIXTURE.new_file(op.clone());
+    let (path_b, content_b, _) = TEST_FIXTURE.new_file(op.clone());
+
+    // Write initial content to both files
+    op.write(&path_a, content_a.clone()).await?;
+    op.write(&path_b, content_b.clone()).await?;
+
+    // Get etags for both files
+    let meta_a = op.stat(&path_a).await?;
+    let etag_a = meta_a.etag().expect("etag must exist");
+    let meta_b = op.stat(&path_b).await?;
+    let etag_b = meta_b.etag().expect("etag must exist");
+
+    // Should succeed: Writing to path_a with its own etag
+    let res = op
+        .write_with(&path_a, content_a.clone())
+        .if_match(etag_a)
+        .await;
+    assert!(res.is_ok());
+
+    // Should fail: Writing to path_a with path_b's etag
+    let res = op
+        .write_with(&path_a, content_a.clone())
+        .if_match(etag_b)
+        .await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}