From 10cb40a0ae4bfdb6a98e69fe5ee0700b850d1294 Mon Sep 17 00:00:00 2001 From: Shani Elharrar Date: Thu, 30 Jul 2026 16:54:05 +0300 Subject: [PATCH 1/2] feat(services/azblob): support if_match writes and honor preconditions on block commit Azure Blob's Put Blob API supports If-Match with an arbitrary ETag, so wire it in alongside the existing If-None-Match support and declare write_with_if_match. Conditional args were only applied to the single-shot Put Blob request. Writing more than once before close() commits through Put Block + Put Block List instead, and that commit request carried no conditional headers, so if_match, if_none_match and if_not_exists were silently dropped and the write degraded to an unconditional overwrite. Put Block does not evaluate preconditions, so they are applied to Put Block List, which does. Add behavior tests that force the block path by writing twice before close(). They fail against Azurite on the previous code for if_none_match and if_not_exists, and pass with this change. --- core/services/azblob/src/backend.rs | 1 + core/services/azblob/src/core.rs | 18 +++++ core/tests/behavior/async_write.rs | 109 +++++++++++++++++++++++++++- 3 files changed, 127 insertions(+), 1 deletion(-) diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs index 9e6e946226ee..7f269ee3825d 100644 --- a/core/services/azblob/src/backend.rs +++ b/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, diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs index a57eab9aca5a..84a9a2b6abd5 100644 --- a/core/services/azblob/src/core.rs +++ b/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() diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs index d7a88cbada06..599f68f40313 100644 --- a/core/tests/behavior/async_write.rs +++ b/core/tests/behavior/async_write.rs @@ -57,7 +57,10 @@ pub fn tests(op: &Operator, tests: &mut Vec) { 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 { + 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 From 04c020925d5678505f0cfae140366cd973c2a0a8 Mon Sep 17 00:00:00 2001 From: tenzinplatter <143778894+TenzinPlatter@users.noreply.github.com> Date: Wed, 5 Aug 2026 14:42:42 +1000 Subject: [PATCH 2/2] chore: remove multiwrite helper --- core/tests/behavior/async_write.rs | 66 ++++++++++++++++-------------- 1 file changed, 35 insertions(+), 31 deletions(-) diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs index 599f68f40313..267bcd4c3150 100644 --- a/core/tests/behavior/async_write.rs +++ b/core/tests/behavior/async_write.rs @@ -813,22 +813,6 @@ 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 { - 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<()> { @@ -837,15 +821,23 @@ pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> { return Ok(()); } - let (path, content, _) = TEST_FIXTURE.new_file(op.clone()); + let path = TEST_FIXTURE.new_file_path(); + let content = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1)); 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()); + // Some services reject the precondition when the writer is created or on an early + // write rather than at commit time + let res: opendal::Result<()> = async { + let mut w = op.writer_with(&path).if_not_exists(true).await?; + w.write(content.clone()).await?; + w.write(content.clone()).await?; + w.close().await?; + Ok(()) + } + .await; assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch); Ok(()) @@ -859,7 +851,8 @@ pub async fn test_writer_write_with_if_none_match(op: Operator) -> Result<()> { return Ok(()); } - let (path, content, _) = TEST_FIXTURE.new_file(op.clone()); + let path = TEST_FIXTURE.new_file_path(); + let content = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1)); op.write(&path, content.clone()) .await @@ -868,9 +861,14 @@ pub async fn test_writer_write_with_if_none_match(op: Operator) -> Result<()> { 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()); + let res: opendal::Result<()> = async { + let mut w = op.writer_with(&path).if_none_match(etag).await?; + w.write(content.clone()).await?; + w.write(content.clone()).await?; + w.close().await?; + Ok(()) + } + .await; assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch); Ok(()) @@ -884,7 +882,8 @@ pub async fn test_writer_write_with_if_match(op: Operator) -> Result<()> { return Ok(()); } - let (path_a, content_a, _) = TEST_FIXTURE.new_file(op.clone()); + let path_a = TEST_FIXTURE.new_file_path(); + let content_a = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1)); let (path_b, content_b, _) = TEST_FIXTURE.new_file(op.clone()); op.write(&path_a, content_a.clone()).await?; @@ -903,15 +902,20 @@ pub async fn test_writer_write_with_if_match(op: Operator) -> Result<()> { .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()); + w.write(content_a.clone()).await?; + w.write(content_a.clone()).await?; + w.close().await.expect("close with own etag must succeed"); // 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()); + let res: opendal::Result<()> = async { + let mut w = op.writer_with(&path_a).if_match(&etag_b).await?; + w.write(content_a.clone()).await?; + w.write(content_a.clone()).await?; + w.close().await?; + Ok(()) + } + .await; assert_eq!(res.unwrap_err().kind(), ErrorKind::ConditionNotMatch); Ok(())