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