From b5c4dbba620cc044270d9531887b9c4683164f82 Mon Sep 17 00:00:00 2001 From: Shani Elharrar Date: Thu, 30 Jul 2026 16:54:05 +0300 Subject: [PATCH] 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 62fb8f51ab04..45a9e50ae5e8 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 f714e9af9cc0..f1dc99797919 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 d3b5b05e788c..4a26ecff6dab 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