feat(services/azblob): support if_match writes and honor preconditions on block commit

core/services/azblob/src/backend.rs

@@ -407,6 +407,7 @@ impl Builder for AzblobBuilder {
             write_can_multi: true,
             write_with_cache_control: true,
             write_with_content_type: true,
+            write_with_if_match: true,
             write_with_if_not_exists: true,
             write_with_if_none_match: true,
             write_with_user_metadata: true,

core/services/azblob/src/core.rs

@@ -306,6 +306,10 @@ impl AzblobCore {
             req = req.header(IF_NONE_MATCH, v);
         }
 
+        if let Some(v) = args.if_match() {
+            req = req.header(IF_MATCH, v);
+        }
+
         if let Some(cache_control) = args.cache_control() {
             req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
         }
@@ -586,6 +590,20 @@ impl AzblobCore {
             req = req.header(constants::X_MS_BLOB_CACHE_CONTROL, cache_control);
         }
 
+        // Put Block List is the request that actually commits a blocked write, so the
+        // write's preconditions have to be evaluated here rather than on Put Block.
+        if args.if_not_exists() {
+            req = req.header(IF_NONE_MATCH, "*");
+        }
+
+        if let Some(v) = args.if_none_match() {
+            req = req.header(IF_NONE_MATCH, v);
+        }
+
+        if let Some(v) = args.if_match() {
+            req = req.header(IF_MATCH, v);
+        }
+
         let content = quick_xml::se::to_string(&PutBlockListRequest {
             latest: block_ids
                 .into_iter()

core/tests/behavior/async_write.rs

@@ -57,7 +57,10 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
             test_writer_futures_copy,
             test_writer_futures_copy_with_concurrent,
             test_writer_return_metadata,
-            test_writer_write_non_contiguous_data
+            test_writer_write_non_contiguous_data,
+            test_writer_write_with_if_not_exists,
+            test_writer_write_with_if_none_match,
+            test_writer_write_with_if_match
         ))
     }
 
@@ -810,6 +813,110 @@ pub async fn test_write_with_if_match(op: Operator) -> Result<()> {
     Ok(())
 }
 
+/// Writing more than once before `close()` commits through the service's multi-part
+/// completion request instead of a single-shot upload. Preconditions must still be honored
+/// on that path, otherwise a conditional write silently degrades to an unconditional
+/// overwrite.
+///
+/// Services normally evaluate the precondition at commit time, but some may reject earlier,
+/// so an error from either `write()` or `close()` is accepted.
+async fn write_conditionally_in_chunks(
+    w: &mut Writer,
+    content: &[u8],
+) -> opendal::Result<Metadata> {
+    w.write(content.to_vec()).await?;
+    w.write(content.to_vec()).await?;
+    w.close().await
+}
+
+/// Write an existing file through a chunked writer with if_not_exists should get a
+/// ConditionNotMatch error.
+pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> {
+    let cap = op.info().capability();
+    if !cap.write_with_if_not_exists || !cap.write_can_multi {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+
+    let mut w = op.writer_with(&path).if_not_exists(true).await?;
+    let res = write_conditionally_in_chunks(&mut w, &content).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
+/// Write an existing file through a chunked writer with its own etag as if_none_match
+/// should get a ConditionNotMatch error.
+pub async fn test_writer_write_with_if_none_match(op: Operator) -> Result<()> {
+    let cap = op.info().capability();
+    if !cap.write_with_if_none_match || !cap.write_can_multi {
+        return Ok(());
+    }
+
+    let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path, content.clone())
+        .await
+        .expect("write must succeed");
+
+    let meta = op.stat(&path).await?;
+    let etag = meta.etag().expect("etag must exist");
+
+    let mut w = op.writer_with(&path).if_none_match(etag).await?;
+    let res = write_conditionally_in_chunks(&mut w, &content).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
+/// Write a file through a chunked writer with if_match should succeed with the file's own
+/// etag and get a ConditionNotMatch error with a stale one.
+pub async fn test_writer_write_with_if_match(op: Operator) -> Result<()> {
+    let cap = op.info().capability();
+    if !cap.write_with_if_match || !cap.write_can_multi {
+        return Ok(());
+    }
+
+    let (path_a, content_a, _) = TEST_FIXTURE.new_file(op.clone());
+    let (path_b, content_b, _) = TEST_FIXTURE.new_file(op.clone());
+
+    op.write(&path_a, content_a.clone()).await?;
+    op.write(&path_b, content_b.clone()).await?;
+
+    let etag_a = op
+        .stat(&path_a)
+        .await?
+        .etag()
+        .expect("etag must exist")
+        .to_string();
+    let etag_b = op
+        .stat(&path_b)
+        .await?
+        .etag()
+        .expect("etag must exist")
+        .to_string();
+
+    // Should succeed: writing to path_a with its own etag.
+    let mut w = op.writer_with(&path_a).if_match(&etag_a).await?;
+    let res = write_conditionally_in_chunks(&mut w, &content_a).await;
+    assert!(res.is_ok());
+
+    // Should fail: writing to path_a with path_b's etag.
+    let mut w = op.writer_with(&path_a).if_match(&etag_b).await?;
+    let res = write_conditionally_in_chunks(&mut w, &content_a).await;
+    assert!(res.is_err());
+    assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch);
+
+    Ok(())
+}
+
 pub async fn test_writer_write_non_contiguous_data(op: Operator) -> Result<()> {
     let path = TEST_FIXTURE.new_file_path();
     let size = 1024 * 1024; // write file with 1 MiB