You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@opendal.apache.org by xu...@apache.org on 2026/08/26 09:55:17 UTC
(opendal) branch xuanwo/write-if-not-changed created (now c23bf44b8)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a change to branch xuanwo/write-if-not-changed
in repository https://gitbox.apache.org/repos/asf/opendal.git
at c23bf44b8 Merge remote-tracking branch 'origin/main' into xuanwo/write-if-not-changed
This branch includes the following new commits:
new baabd99b6 feat: support write if not changed
new c23bf44b8 Merge remote-tracking branch 'origin/main' into xuanwo/write-if-not-changed
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(opendal) 01/02: feat: support write if not changed
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/write-if-not-changed
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit baabd99b66a16f15bb4359df67dc29d7c7f1a24d
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Aug 26 17:45:25 2026 +0800
feat: support write if not changed
---
.../src/docs/rfcs/8147_write_if_not_changed.md | 26 +-
core/core/src/layers/correctness_check.rs | 247 ++++++++++++-
core/core/src/raw/accessor.rs | 2 +-
core/core/src/raw/ops.rs | 164 +++++++++
core/core/src/types/capability.rs | 24 ++
core/core/src/types/delete/deleter.rs | 75 +++-
core/core/src/types/delete/input.rs | 88 ++++-
core/core/src/types/operator/operator.rs | 226 +++++++++++-
core/core/src/types/operator/operator_futures.rs | 144 ++++++++
core/core/src/types/options.rs | 60 ++++
core/services/azblob/src/backend.rs | 8 +-
core/services/azblob/src/core.rs | 35 +-
core/services/azblob/src/deleter.rs | 40 ++-
core/services/gcs/src/backend.rs | 36 +-
core/services/gcs/src/core.rs | 395 +++++++++++++++++++--
core/services/gcs/src/deleter.rs | 48 ++-
core/services/gcs/src/lister.rs | 3 +
core/services/gcs/src/reader.rs | 20 +-
core/services/gcs/src/writer.rs | 210 ++++++++++-
core/services/s3/src/copier.rs | 48 ++-
core/services/s3/src/deleter.rs | 20 +-
core/services/s3/src/writer.rs | 36 +-
core/tests/behavior/async_copy.rs | 158 +++++++++
core/tests/behavior/async_delete.rs | 155 +++++++-
core/tests/behavior/async_read.rs | 60 ++++
core/tests/behavior/async_stat.rs | 58 +++
core/tests/behavior/async_write.rs | 170 ++++++++-
27 files changed, 2408 insertions(+), 148 deletions(-)
diff --git a/core/core/src/docs/rfcs/8147_write_if_not_changed.md b/core/core/src/docs/rfcs/8147_write_if_not_changed.md
index f76467210..9e79d8f83 100644
--- a/core/core/src/docs/rfcs/8147_write_if_not_changed.md
+++ b/core/core/src/docs/rfcs/8147_write_if_not_changed.md
@@ -69,9 +69,6 @@ let expected = Metadata::default()
Preserving both fields keeps externally stored metadata portable across
services. Both fields must describe the same observed object state.
-The new conditions are not supported by presign operations. Supplying them to
-a presign options API returns `Unsupported` before service dispatch.
-
## Condition semantics
All token values are opaque. A `version` option selects a historical object;
@@ -108,13 +105,13 @@ participate. OpenDAL does not validate provenance, path, or storage namespace,
so the caller must associate externally constructed metadata with the correct
target.
-`if_not_changed` is exclusive with every other target condition and, on
-delete, with the target `version` selector. Version match and non-match are
-mutually exclusive and cannot be combined with another target condition or a
-target version selector. Copy's `source_version` selects a different object
-and may be combined with destination conditions. Append writes cannot use
-version conditions or `if_not_changed` because append has no portable final
-replacement commit point. Invalid combinations return `ConfigInvalid`.
+OpenDAL preserves explicit target conditions and version selectors while
+lowering `if_not_changed`. If the selected primitive match field already
+contains the same opaque token, OpenDAL deduplicates it. If that field contains
+a different token, OpenDAL returns `ConditionNotMatch` because both equality
+conditions cannot hold. OpenDAL forwards every other representable combination
+unchanged. After OpenDAL validates each condition's capability, the service
+decides whether it accepts the combination and reports any invalid combination.
Missing-target results are part of the portable contract:
@@ -155,6 +152,9 @@ All new capabilities default to `false`. There is no independently advertised
`if_not_changed` capability; support is derived from the applicable version and
ETag match capabilities plus the supplied metadata.
+Capabilities describe support for individual conditions. They do not guarantee
+that a service accepts any particular combination of conditions or selectors.
+
A service advertising a version-match primitive must populate
`Metadata::version()` on stat, read, write, copy, and list whenever the native
response provides it. The same rule applies to `Metadata::etag()` for an
@@ -173,12 +173,6 @@ GCS maps version conditions to JSON API generation parameters:
| `if_not_exists` | `ifGenerationMatch=0` |
| `if_not_changed(meta)` | Lower to `if_version_match(meta.version())`. |
-GCS generation `0` is reserved for the absence condition and is never an
-object generation. GCS rejects `if_version_match("0")` and expected metadata
-whose selected version is `"0"` with `ConfigInvalid`.
-`if_version_not_match("0")` remains valid and succeeds only when a live object
-exists.
-
GCS applies these parameters to JSON API get, insert, delete, and destination
rewrite requests. `ifGenerationNotMatch` fails when no live object exists, and
OpenDAL maps that failure to `ConditionNotMatch`. Multi-request rewrite keeps
diff --git a/core/core/src/layers/correctness_check.rs b/core/core/src/layers/correctness_check.rs
index 603cfd75c..36ae29cc6 100644
--- a/core/core/src/layers/correctness_check.rs
+++ b/core/core/src/layers/correctness_check.rs
@@ -109,6 +109,20 @@ impl Service for CorrectnessService {
"if_none_match",
));
}
+ if !capability.read_with_if_version_match && args.if_version_match().is_some() {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Read,
+ "if_version_match",
+ ));
+ }
+ if !capability.read_with_if_version_not_match && args.if_version_not_match().is_some() {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Read,
+ "if_version_not_match",
+ ));
+ }
if !capability.read_with_if_modified_since && args.if_modified_since().is_some() {
return Err(new_unsupported_error(
scheme,
@@ -153,6 +167,20 @@ impl Service for CorrectnessService {
return Err(err);
}
+ if args.if_version_match().is_some() && !capability.write_with_if_version_match {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Write,
+ "if_version_match",
+ ));
+ }
+ if args.if_version_not_match().is_some() && !capability.write_with_if_version_not_match {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Write,
+ "if_version_not_match",
+ ));
+ }
self.inner.write(ctx, path, args)
}
@@ -173,6 +201,20 @@ impl Service for CorrectnessService {
"if_none_match",
));
}
+ if !capability.stat_with_if_version_match && args.if_version_match().is_some() {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Stat,
+ "if_version_match",
+ ));
+ }
+ if !capability.stat_with_if_version_not_match && args.if_version_not_match().is_some() {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Stat,
+ "if_version_not_match",
+ ));
+ }
if !capability.stat_with_if_modified_since && args.if_modified_since().is_some() {
return Err(new_unsupported_error(
scheme,
@@ -217,6 +259,27 @@ impl Service for CorrectnessService {
if args.if_match().is_some() && !capability.copy_with_if_match {
return Err(new_unsupported_error(scheme, Operation::Copy, "if_match"));
}
+ if args.if_none_match().is_some() && !capability.copy_with_if_none_match {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Copy,
+ "if_none_match",
+ ));
+ }
+ if args.if_version_match().is_some() && !capability.copy_with_if_version_match {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Copy,
+ "if_version_match",
+ ));
+ }
+ if args.if_version_not_match().is_some() && !capability.copy_with_if_version_not_match {
+ return Err(new_unsupported_error(
+ scheme,
+ Operation::Copy,
+ "if_version_not_match",
+ ));
+ }
if args.source_version().is_some() && !capability.copy_with_source_version {
return Err(new_unsupported_error(
scheme,
@@ -311,6 +374,32 @@ impl<T> CheckWrapper<T> {
));
}
+ if args.if_none_match().is_some() && !self.capability.delete_with_if_none_match {
+ return Err(new_unsupported_error(
+ self.scheme,
+ Operation::Delete,
+ "if_none_match",
+ ));
+ }
+
+ if args.if_version_match().is_some() && !self.capability.delete_with_if_version_match {
+ return Err(new_unsupported_error(
+ self.scheme,
+ Operation::Delete,
+ "if_version_match",
+ ));
+ }
+
+ if args.if_version_not_match().is_some()
+ && !self.capability.delete_with_if_version_not_match
+ {
+ return Err(new_unsupported_error(
+ self.scheme,
+ Operation::Delete,
+ "if_version_not_match",
+ ));
+ }
+
Ok(())
}
}
@@ -395,10 +484,7 @@ mod tests {
_: OpCopy,
_: OpCopier,
) -> Result<Self::Copier> {
- Err(Error::new(
- ErrorKind::Unsupported,
- "operation is not supported",
- ))
+ Ok(())
}
async fn rename(
@@ -578,6 +664,159 @@ mod tests {
assert!(res.is_ok())
}
+ #[tokio::test]
+ async fn test_version_preconditions_are_forwarded() -> Result<()> {
+ let op = new_test_operator(Capability {
+ stat: true,
+ stat_with_version: true,
+ stat_with_if_version_match: true,
+ read: true,
+ read_with_if_version_match: true,
+ write: true,
+ write_with_if_match: true,
+ write_with_if_version_match: true,
+ delete: true,
+ delete_with_if_version_match: true,
+ delete_with_if_version_not_match: true,
+ copy: true,
+ copy_with_if_version_match: true,
+ ..Default::default()
+ });
+
+ op.stat_with("path").if_version_match("version").await?;
+ op.read_with("path").if_version_match("version").await?;
+ op.write_with("path", "")
+ .if_version_match("version")
+ .await?;
+ op.delete_with("path").if_version_match("version").await?;
+ op.copy_with("from", "to")
+ .if_version_match("version")
+ .await?;
+
+ op.write_with("path", "")
+ .if_match("etag")
+ .if_version_match("version")
+ .await?;
+
+ op.stat_with("path")
+ .version("selected")
+ .if_version_match("current")
+ .await?;
+
+ op.delete_with("path")
+ .if_version_match("matched")
+ .if_version_not_match("not-matched")
+ .await?;
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_if_not_changed_merges_at_dispatch() -> Result<()> {
+ let op = new_test_operator(Capability {
+ write: true,
+ write_with_if_match: true,
+ write_with_if_version_match: true,
+ delete: true,
+ delete_with_if_match: true,
+ copy: true,
+ copy_with_if_match: true,
+ ..Default::default()
+ });
+ let metadata = Metadata::default()
+ .with_etag("etag".to_string())
+ .with_version("version".to_string());
+
+ op.write_with("path", "")
+ .if_match("other-etag")
+ .if_version_match("version")
+ .if_not_changed(&metadata)
+ .await?;
+
+ let err = op
+ .write_with("path", "")
+ .if_version_match("other-version")
+ .if_not_changed(&metadata)
+ .await
+ .expect_err("different selected version must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.copy_with("from", "to")
+ .if_match("etag")
+ .if_not_changed(&metadata)
+ .await?;
+ let err = op
+ .copy_with("from", "to")
+ .if_match("other-etag")
+ .if_not_changed(&metadata)
+ .await
+ .expect_err("different selected etag must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.delete_with("path")
+ .if_match("etag")
+ .if_not_changed(&metadata)
+ .await?;
+ let err = op
+ .delete_with("path")
+ .if_match("other-etag")
+ .if_not_changed(&metadata)
+ .await
+ .expect_err("different selected etag must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ Ok(())
+ }
+
+ #[tokio::test]
+ async fn test_presign_conditions_reach_service() {
+ let op = new_test_operator(Capability::default());
+
+ let err = op
+ .presign_stat_options(
+ "path",
+ std::time::Duration::from_secs(60),
+ options::StatOptions {
+ if_version_match: Some("version".to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect_err("mock service rejects presign");
+ assert!(err.to_string().contains("operation is not supported"));
+
+ let err = op
+ .presign_delete_options(
+ "path",
+ std::time::Duration::from_secs(60),
+ options::DeleteOptions {
+ if_none_match: Some("etag".to_string()),
+ if_version_not_match: Some("version".to_string()),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect_err("mock service rejects presign");
+ assert!(err.to_string().contains("operation is not supported"));
+
+ let op = new_test_operator(Capability {
+ write_with_if_match: true,
+ ..Default::default()
+ });
+ let err = op
+ .presign_write_options(
+ "path",
+ std::time::Duration::from_secs(60),
+ options::WriteOptions {
+ if_not_changed: Some(Metadata::default().with_etag("etag".to_string())),
+ ..Default::default()
+ },
+ )
+ .await
+ .expect_err("mock service rejects presign");
+ assert!(err.to_string().contains("operation is not supported"));
+ }
+
#[tokio::test]
async fn test_rename_with_if_not_exists() {
let op = new_test_operator(Capability {
diff --git a/core/core/src/raw/accessor.rs b/core/core/src/raw/accessor.rs
index 0036af785..6ad761b62 100644
--- a/core/core/src/raw/accessor.rs
+++ b/core/core/src/raw/accessor.rs
@@ -185,7 +185,7 @@ pub trait Service: Send + Sync + Debug + Unpin + 'static {
/// # Behavior
///
/// - The returned deleter handles one or more delete requests.
- /// - Deleting a missing path should succeed.
+ /// - Deleting a missing path should succeed unless a condition requires a live target.
fn delete(&self, ctx: &OperationContext) -> Result<Self::Deleter>;
/// Invoke the `list` operation on the specified path.
diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs
index 7f32031c6..d4e1150cd 100644
--- a/core/core/src/raw/ops.rs
+++ b/core/core/src/raw/ops.rs
@@ -51,6 +51,15 @@ pub struct OpDelete {
/// The ETag that the object must match before deletion.
if_match: Option<String>,
+
+ /// The ETag that the object must not match before deletion.
+ if_none_match: Option<String>,
+
+ /// The version that the current object must match before deletion.
+ if_version_match: Option<String>,
+
+ /// The version that the current object must not match before deletion.
+ if_version_not_match: Option<String>,
}
impl OpDelete {
@@ -81,6 +90,24 @@ impl OpDelete {
self
}
+ /// Set the ETag that the object must not match before deletion.
+ pub fn with_if_none_match(mut self, if_none_match: impl Into<String>) -> Self {
+ self.if_none_match = Some(if_none_match.into());
+ self
+ }
+
+ /// Set the version that the current object must match before deletion.
+ pub fn with_if_version_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_match = Some(version.into());
+ self
+ }
+
+ /// Set the version that the current object must not match before deletion.
+ pub fn with_if_version_not_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_not_match = Some(version.into());
+ self
+ }
+
/// Return the version of the object to delete.
pub fn version(&self) -> Option<&str> {
self.version.as_deref()
@@ -95,6 +122,21 @@ impl OpDelete {
pub fn if_match(&self) -> Option<&str> {
self.if_match.as_deref()
}
+
+ /// Return the ETag that the object must not match before deletion.
+ pub fn if_none_match(&self) -> Option<&str> {
+ self.if_none_match.as_deref()
+ }
+
+ /// Return the version that the current object must match before deletion.
+ pub fn if_version_match(&self) -> Option<&str> {
+ self.if_version_match.as_deref()
+ }
+
+ /// Return the version that the current object must not match before deletion.
+ pub fn if_version_not_match(&self) -> Option<&str> {
+ self.if_version_not_match.as_deref()
+ }
}
impl From<options::DeleteOptions> for OpDelete {
@@ -103,6 +145,9 @@ impl From<options::DeleteOptions> for OpDelete {
version: value.version,
recursive: value.recursive,
if_match: value.if_match,
+ if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
}
}
}
@@ -334,6 +379,8 @@ impl From<OpDelete> for PresignOperation {
pub struct OpRead {
if_match: Option<String>,
if_none_match: Option<String>,
+ if_version_match: Option<String>,
+ if_version_not_match: Option<String>,
if_modified_since: Option<Timestamp>,
if_unmodified_since: Option<Timestamp>,
override_content_type: Option<String>,
@@ -405,6 +452,28 @@ impl OpRead {
self.if_none_match.as_deref()
}
+ /// Set the version that the current object must match.
+ pub fn with_if_version_match(mut self, version: &str) -> Self {
+ self.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version match condition.
+ pub fn if_version_match(&self) -> Option<&str> {
+ self.if_version_match.as_deref()
+ }
+
+ /// Set the version that the current object must not match.
+ pub fn with_if_version_not_match(mut self, version: &str) -> Self {
+ self.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version non-match condition.
+ pub fn if_version_not_match(&self) -> Option<&str> {
+ self.if_version_not_match.as_deref()
+ }
+
/// Set the If-Modified-Since of the option
pub fn with_if_modified_since(mut self, v: Timestamp) -> Self {
self.if_modified_since = Some(v);
@@ -528,6 +597,8 @@ impl From<options::ReadOptions> for (BytesRange, OpRead, OpReader) {
OpRead {
if_match: value.if_match,
if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
if_modified_since: value.if_modified_since,
if_unmodified_since: value.if_unmodified_since,
override_content_type: value.override_content_type,
@@ -553,6 +624,8 @@ impl From<options::ReaderOptions> for (OpRead, OpReader) {
OpRead {
if_match: value.if_match,
if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
if_modified_since: value.if_modified_since,
if_unmodified_since: value.if_unmodified_since,
override_content_type: None,
@@ -577,6 +650,8 @@ impl From<options::ReaderOptions> for (OpRead, OpReader) {
pub struct OpStat {
if_match: Option<String>,
if_none_match: Option<String>,
+ if_version_match: Option<String>,
+ if_version_not_match: Option<String>,
if_modified_since: Option<Timestamp>,
if_unmodified_since: Option<Timestamp>,
override_content_type: Option<String>,
@@ -613,6 +688,28 @@ impl OpStat {
self.if_none_match.as_deref()
}
+ /// Set the version that the current object must match.
+ pub fn with_if_version_match(mut self, version: &str) -> Self {
+ self.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version match condition.
+ pub fn if_version_match(&self) -> Option<&str> {
+ self.if_version_match.as_deref()
+ }
+
+ /// Set the version that the current object must not match.
+ pub fn with_if_version_not_match(mut self, version: &str) -> Self {
+ self.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version non-match condition.
+ pub fn if_version_not_match(&self) -> Option<&str> {
+ self.if_version_not_match.as_deref()
+ }
+
/// Set the If-Modified-Since of the option
pub fn with_if_modified_since(mut self, v: Timestamp) -> Self {
self.if_modified_since = Some(v);
@@ -686,6 +783,8 @@ impl From<options::StatOptions> for OpStat {
Self {
if_match: value.if_match,
if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
if_modified_since: value.if_modified_since,
if_unmodified_since: value.if_unmodified_since,
override_content_type: value.override_content_type,
@@ -707,6 +806,8 @@ pub struct OpWrite {
cache_control: Option<String>,
if_match: Option<String>,
if_none_match: Option<String>,
+ if_version_match: Option<String>,
+ if_version_not_match: Option<String>,
if_not_exists: bool,
user_metadata: Option<HashMap<String, String>>,
}
@@ -815,6 +916,28 @@ impl OpWrite {
self.if_none_match.as_deref()
}
+ /// Set the version that the current object must match.
+ pub fn with_if_version_match(mut self, version: &str) -> Self {
+ self.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version match condition.
+ pub fn if_version_match(&self) -> Option<&str> {
+ self.if_version_match.as_deref()
+ }
+
+ /// Set the version that the current object must not match.
+ pub fn with_if_version_not_match(mut self, version: &str) -> Self {
+ self.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Get the version non-match condition.
+ pub fn if_version_not_match(&self) -> Option<&str> {
+ self.if_version_not_match.as_deref()
+ }
+
/// Set the If-Not-Exist of the option
pub fn with_if_not_exists(mut self, b: bool) -> Self {
self.if_not_exists = b;
@@ -885,6 +1008,8 @@ impl From<options::WriteOptions> for (OpWrite, OpWriter) {
cache_control: value.cache_control,
if_match: value.if_match,
if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
if_not_exists: value.if_not_exists,
user_metadata: value.user_metadata,
},
@@ -898,6 +1023,9 @@ impl From<options::WriteOptions> for (OpWrite, OpWriter) {
pub struct OpCopy {
if_not_exists: bool,
if_match: Option<String>,
+ if_none_match: Option<String>,
+ if_version_match: Option<String>,
+ if_version_not_match: Option<String>,
source_version: Option<String>,
}
@@ -935,6 +1063,39 @@ impl OpCopy {
self.if_match.as_deref()
}
+ /// Set the destination ETag non-match condition.
+ pub fn with_if_none_match(mut self, if_none_match: impl Into<String>) -> Self {
+ self.if_none_match = Some(if_none_match.into());
+ self
+ }
+
+ /// Get the destination ETag non-match condition.
+ pub fn if_none_match(&self) -> Option<&str> {
+ self.if_none_match.as_deref()
+ }
+
+ /// Set the current destination version match condition.
+ pub fn with_if_version_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_match = Some(version.into());
+ self
+ }
+
+ /// Get the current destination version match condition.
+ pub fn if_version_match(&self) -> Option<&str> {
+ self.if_version_match.as_deref()
+ }
+
+ /// Set the current destination version non-match condition.
+ pub fn with_if_version_not_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_not_match = Some(version.into());
+ self
+ }
+
+ /// Get the current destination version non-match condition.
+ pub fn if_version_not_match(&self) -> Option<&str> {
+ self.if_version_not_match.as_deref()
+ }
+
/// Set source version for the operation.
///
/// When set, the copy operation will copy from the specified source version.
@@ -1003,6 +1164,9 @@ impl From<options::CopyOptions> for (OpCopy, OpCopier) {
OpCopy {
if_not_exists: value.if_not_exists,
if_match: value.if_match,
+ if_none_match: value.if_none_match,
+ if_version_match: value.if_version_match,
+ if_version_not_match: value.if_version_not_match,
source_version: value.source_version,
},
OpCopier {
diff --git a/core/core/src/types/capability.rs b/core/core/src/types/capability.rs
index cca00a6e1..8849604c7 100644
--- a/core/core/src/types/capability.rs
+++ b/core/core/src/types/capability.rs
@@ -67,6 +67,10 @@ pub struct Capability {
pub stat_with_if_match: bool,
/// Indicates if conditional stat operations using If-None-Match are supported.
pub stat_with_if_none_match: bool,
+ /// Indicates if conditional stat operations using version match are supported.
+ pub stat_with_if_version_match: bool,
+ /// Indicates if conditional stat operations using version non-match are supported.
+ pub stat_with_if_version_not_match: bool,
/// Indicates if conditional stat operations using If-Modified-Since are supported.
pub stat_with_if_modified_since: bool,
/// Indicates if conditional stat operations using If-Unmodified-Since are supported.
@@ -86,6 +90,10 @@ pub struct Capability {
pub read_with_if_match: bool,
/// Indicates if conditional read operations using If-None-Match are supported.
pub read_with_if_none_match: bool,
+ /// Indicates if conditional read operations using version match are supported.
+ pub read_with_if_version_match: bool,
+ /// Indicates if conditional read operations using version non-match are supported.
+ pub read_with_if_version_not_match: bool,
/// Indicates if conditional read operations using If-Modified-Since are supported.
pub read_with_if_modified_since: bool,
/// Indicates if conditional read operations using If-Unmodified-Since are supported.
@@ -121,6 +129,10 @@ pub struct Capability {
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 conditional write operations using version match are supported.
+ pub write_with_if_version_match: bool,
+ /// Indicates if conditional write operations using version non-match are supported.
+ pub write_with_if_version_not_match: bool,
/// Indicates if write operations can be conditional on object non-existence.
pub write_with_if_not_exists: bool,
/// Indicates if custom user metadata can be attached during write operations.
@@ -146,6 +158,12 @@ pub struct Capability {
pub delete_with_recursive: bool,
/// Indicates if conditional delete operations using If-Match are supported.
pub delete_with_if_match: bool,
+ /// Indicates if conditional delete operations using If-None-Match are supported.
+ pub delete_with_if_none_match: bool,
+ /// Indicates if conditional delete operations using version match are supported.
+ pub delete_with_if_version_match: bool,
+ /// Indicates if conditional delete operations using version non-match are supported.
+ pub delete_with_if_version_not_match: bool,
/// Maximum size supported for single delete operations.
pub delete_max_size: Option<usize>,
@@ -155,6 +173,12 @@ pub struct Capability {
pub copy_with_if_not_exists: bool,
/// Indicates if conditional copy operations with if-match are supported.
pub copy_with_if_match: bool,
+ /// Indicates if conditional copy operations with if-none-match are supported.
+ pub copy_with_if_none_match: bool,
+ /// Indicates if conditional copy operations using version match are supported.
+ pub copy_with_if_version_match: bool,
+ /// Indicates if conditional copy operations using version non-match are supported.
+ pub copy_with_if_version_not_match: bool,
/// Indicates if copy operations from a specific source version are supported.
pub copy_with_source_version: bool,
/// Indicates if copy operations can be split into multiple server-side tasks.
diff --git a/core/core/src/types/delete/deleter.rs b/core/core/src/types/delete/deleter.rs
index bf83df06f..8e6023e60 100644
--- a/core/core/src/types/delete/deleter.rs
+++ b/core/core/src/types/delete/deleter.rs
@@ -83,25 +83,84 @@ use crate::*;
/// ```
pub struct Deleter {
deleter: oio::Deleter,
+ scheme: &'static str,
+ capability: Capability,
}
impl Deleter {
pub(crate) fn create(ctx: OperationContext, srv: Servicer) -> Result<Self> {
+ let scheme = srv.info().scheme();
+ let capability = srv.capability();
let deleter = srv.delete(&ctx)?;
- Ok(Self { deleter })
+ Ok(Self {
+ deleter,
+ scheme,
+ capability,
+ })
}
/// Delete a path.
pub async fn delete(&mut self, input: impl IntoDeleteInput) -> Result<()> {
let input = input.into_delete_input();
- let mut op = OpDelete::default();
- if let Some(version) = &input.version {
- op = op.with_version(version);
- }
- if input.recursive {
- op = op.with_recursive(true);
+ let mut opts = options::DeleteOptions {
+ version: input.version,
+ recursive: input.recursive,
+ if_match: input.if_match,
+ if_none_match: input.if_none_match,
+ if_version_match: input.if_version_match,
+ if_version_not_match: input.if_version_not_match,
+ if_not_changed: input.if_not_changed,
+ };
+ if let Some(metadata) = opts.if_not_changed.take() {
+ if self.capability.delete_with_if_version_match
+ && let Some(version) = metadata.version()
+ {
+ if let Some(explicit) = opts.if_version_match.as_deref() {
+ if explicit != version {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_version_match",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
+ } else {
+ opts.if_version_match = Some(version.to_string());
+ }
+ } else if self.capability.delete_with_if_match
+ && let Some(etag) = metadata.etag()
+ {
+ if let Some(explicit) = opts.if_match.as_deref() {
+ if explicit != etag {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_match",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
+ } else {
+ opts.if_match = Some(etag.to_string());
+ }
+ } else if !self.capability.delete_with_if_version_match
+ && !self.capability.delete_with_if_match
+ {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ format!(
+ "The service {} does not support the operation delete with if_not_changed",
+ self.scheme
+ ),
+ )
+ .with_operation(Operation::Delete.into_static()));
+ } else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "if_not_changed metadata does not contain an identity supported by delete",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
}
+ let op = opts.into();
self.deleter.delete(&input.path, op).await?;
Ok(())
@@ -235,6 +294,8 @@ mod tests {
let deleter = Deleter {
deleter: Box::new(mock),
+ scheme: "test",
+ capability: Capability::default(),
};
let mut sink = deleter.into_sink::<String>();
diff --git a/core/core/src/types/delete/input.rs b/core/core/src/types/delete/input.rs
index 62982921a..d3c4612ce 100644
--- a/core/core/src/types/delete/input.rs
+++ b/core/core/src/types/delete/input.rs
@@ -15,8 +15,8 @@
// specific language governing permissions and limitations
// under the License.
-use crate::Entry;
use crate::raw::OpDelete;
+use crate::{Entry, Metadata};
/// DeleteInput is the input for delete operations.
#[non_exhaustive]
@@ -28,6 +28,68 @@ pub struct DeleteInput {
pub version: Option<String>,
/// Whether to perform recursive deletion.
pub recursive: bool,
+ /// Delete only when the current ETag matches this value.
+ pub if_match: Option<String>,
+ /// Delete only when the current ETag does not match this value.
+ pub if_none_match: Option<String>,
+ /// Delete only when the current version matches this value.
+ pub if_version_match: Option<String>,
+ /// Delete only when the current version does not match this value.
+ pub if_version_not_match: Option<String>,
+ /// Delete only when the object still matches this metadata.
+ pub if_not_changed: Option<Metadata>,
+}
+
+impl DeleteInput {
+ /// Create a delete input for `path`.
+ pub fn new(path: impl Into<String>) -> Self {
+ Self {
+ path: path.into(),
+ ..Default::default()
+ }
+ }
+
+ /// Select the object version to delete.
+ pub fn with_version(mut self, version: impl Into<String>) -> Self {
+ self.version = Some(version.into());
+ self
+ }
+
+ /// Configure recursive deletion.
+ pub fn with_recursive(mut self, recursive: bool) -> Self {
+ self.recursive = recursive;
+ self
+ }
+
+ /// Delete only when the current ETag matches `etag`.
+ pub fn with_if_match(mut self, etag: impl Into<String>) -> Self {
+ self.if_match = Some(etag.into());
+ self
+ }
+
+ /// Delete only when the current ETag does not match `etag`.
+ pub fn with_if_none_match(mut self, etag: impl Into<String>) -> Self {
+ self.if_none_match = Some(etag.into());
+ self
+ }
+
+ /// Delete only when the current version matches `version`.
+ pub fn with_if_version_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_match = Some(version.into());
+ self
+ }
+
+ /// Delete only when the current version does not match `version`.
+ pub fn with_if_version_not_match(mut self, version: impl Into<String>) -> Self {
+ self.if_version_not_match = Some(version.into());
+ self
+ }
+
+ /// Delete only when the object still matches `metadata`.
+ pub fn with_if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.if_not_changed = Some(metadata.clone());
+ self
+ }
}
/// IntoDeleteInput is a helper trait that makes it easier for users to play with `Deleter`.
@@ -46,22 +108,14 @@ impl IntoDeleteInput for DeleteInput {
/// Implement `IntoDeleteInput` for `&str` so we can use `&str` as a DeleteInput.
impl IntoDeleteInput for &str {
fn into_delete_input(self) -> DeleteInput {
- DeleteInput {
- path: self.to_string(),
- recursive: false,
- ..Default::default()
- }
+ DeleteInput::new(self)
}
}
/// Implement `IntoDeleteInput` for `String` so we can use `Vec<String>` as a DeleteInput stream.
impl IntoDeleteInput for String {
fn into_delete_input(self) -> DeleteInput {
- DeleteInput {
- path: self,
- recursive: false,
- ..Default::default()
- }
+ DeleteInput::new(self)
}
}
@@ -80,6 +134,18 @@ impl IntoDeleteInput for (String, OpDelete) {
if let Some(version) = args.version() {
input.version = Some(version.to_string());
}
+ if let Some(etag) = args.if_match() {
+ input.if_match = Some(etag.to_string());
+ }
+ if let Some(etag) = args.if_none_match() {
+ input.if_none_match = Some(etag.to_string());
+ }
+ if let Some(version) = args.if_version_match() {
+ input.if_version_match = Some(version.to_string());
+ }
+ if let Some(version) = args.if_version_not_match() {
+ input.if_version_not_match = Some(version.to_string());
+ }
input
}
}
diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs
index c27efafb1..8f8ad83c8 100644
--- a/core/core/src/types/operator/operator.rs
+++ b/core/core/src/types/operator/operator.rs
@@ -24,7 +24,7 @@ use futures::TryStreamExt;
use crate::operator_futures::*;
use crate::raw::*;
-use crate::types::delete::Deleter;
+use crate::types::delete::{DeleteInput, Deleter};
use crate::*;
/// The `Operator` serves as the entry point for all public asynchronous APIs.
@@ -983,10 +983,7 @@ impl Operator {
);
}
- let (args, opts) = opts.into();
-
- let write_context = WriteContext::new(ctx, srv, path, args, opts);
- let mut w = Writer::new(write_context).await?;
+ let mut w = Self::writer_inner(ctx, srv, path, opts).await?;
w.write(bs).await?;
w.close().await
}
@@ -1153,6 +1150,54 @@ impl Operator {
);
}
+ let mut opts = opts;
+ if let Some(metadata) = opts.if_not_changed.take() {
+ let capability = srv.capability();
+ if capability.write_with_if_version_match
+ && let Some(version) = metadata.version()
+ {
+ if let Some(explicit) = opts.if_version_match.as_deref() {
+ if explicit != version {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_version_match",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ } else {
+ opts.if_version_match = Some(version.to_string());
+ }
+ } else if capability.write_with_if_match
+ && let Some(etag) = metadata.etag()
+ {
+ if let Some(explicit) = opts.if_match.as_deref() {
+ if explicit != etag {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_match",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ } else {
+ opts.if_match = Some(etag.to_string());
+ }
+ } else if !capability.write_with_if_version_match && !capability.write_with_if_match {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ format!(
+ "The service {} does not support the operation write with if_not_changed",
+ srv.info().scheme()
+ ),
+ )
+ .with_operation(Operation::Write.into_static()));
+ } else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "if_not_changed metadata does not contain an identity supported by write",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ }
let (args, opts) = opts.into();
let write_context = WriteContext::new(ctx, srv, path, args, opts);
let w = Writer::new(write_context).await?;
@@ -1399,6 +1444,54 @@ impl Operator {
);
}
+ let mut opts = opts;
+ if let Some(metadata) = opts.if_not_changed.take() {
+ let capability = srv.capability();
+ if capability.copy_with_if_version_match
+ && let Some(version) = metadata.version()
+ {
+ if let Some(explicit) = opts.if_version_match.as_deref() {
+ if explicit != version {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_version_match",
+ )
+ .with_operation(Operation::Copy.into_static()));
+ }
+ } else {
+ opts.if_version_match = Some(version.to_string());
+ }
+ } else if capability.copy_with_if_match
+ && let Some(etag) = metadata.etag()
+ {
+ if let Some(explicit) = opts.if_match.as_deref() {
+ if explicit != etag {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_match",
+ )
+ .with_operation(Operation::Copy.into_static()));
+ }
+ } else {
+ opts.if_match = Some(etag.to_string());
+ }
+ } else if !capability.copy_with_if_version_match && !capability.copy_with_if_match {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ format!(
+ "The service {} does not support the operation copy with if_not_changed",
+ srv.info().scheme()
+ ),
+ )
+ .with_operation(Operation::Copy.into_static()));
+ } else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "if_not_changed metadata does not contain an identity supported by copy",
+ )
+ .with_operation(Operation::Copy.into_static()));
+ }
+ }
let (args, opts) = opts.into();
std::future::ready(Copier::create(ctx, srv, &from, &to, args, opts)).await
}
@@ -1551,7 +1644,8 @@ impl Operator {
///
/// # Notes
///
- /// - Deleting a file that does not exist won't return errors.
+ /// - Deleting a file that does not exist won't return errors unless a condition requires a
+ /// live target.
///
/// # Examples
///
@@ -1572,7 +1666,8 @@ impl Operator {
///
/// # Notes
///
- /// - Deleting a file that does not exist won't return errors.
+ /// - Deleting a file that does not exist won't return errors unless a condition requires a
+ /// live target.
///
/// # Options
///
@@ -1607,7 +1702,8 @@ impl Operator {
///
/// # Notes
///
- /// - Deleting a file that does not exist won't return errors.
+ /// - Deleting a file that does not exist won't return errors unless a condition requires a
+ /// live target.
///
/// # Examples
///
@@ -1638,10 +1734,20 @@ impl Operator {
path: String,
opts: options::DeleteOptions,
) -> Result<()> {
- let mut deleter = srv.delete(&ctx)?;
- let args = opts.into();
- deleter.delete_dyn(&path, args).await?;
- deleter.close_dyn().await?;
+ let mut deleter = Deleter::create(ctx, srv)?;
+ deleter
+ .delete(DeleteInput {
+ path,
+ version: opts.version,
+ recursive: opts.recursive,
+ if_match: opts.if_match,
+ if_none_match: opts.if_none_match,
+ if_version_match: opts.if_version_match,
+ if_version_not_match: opts.if_version_not_match,
+ if_not_changed: opts.if_not_changed,
+ })
+ .await?;
+ deleter.close().await?;
Ok(())
}
@@ -2459,6 +2565,54 @@ impl Operator {
path: String,
(opts, expire): (options::WriteOptions, Duration),
) -> Result<PresignedRequest> {
+ let mut opts = opts;
+ if let Some(metadata) = opts.if_not_changed.take() {
+ let capability = srv.capability();
+ if capability.write_with_if_version_match
+ && let Some(version) = metadata.version()
+ {
+ if let Some(explicit) = opts.if_version_match.as_deref() {
+ if explicit != version {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_version_match",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ } else {
+ opts.if_version_match = Some(version.to_string());
+ }
+ } else if capability.write_with_if_match
+ && let Some(etag) = metadata.etag()
+ {
+ if let Some(explicit) = opts.if_match.as_deref() {
+ if explicit != etag {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_match",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ } else {
+ opts.if_match = Some(etag.to_string());
+ }
+ } else if !capability.write_with_if_version_match && !capability.write_with_if_match {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ format!(
+ "The service {} does not support the operation write with if_not_changed",
+ srv.info().scheme()
+ ),
+ )
+ .with_operation(Operation::Write.into_static()));
+ } else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "if_not_changed metadata does not contain an identity supported by write",
+ )
+ .with_operation(Operation::Write.into_static()));
+ }
+ }
let (op_write, _) = opts.into();
let op = OpPresign::new(op_write, expire);
let rp = srv.presign(&ctx, &path, op).await?;
@@ -2572,6 +2726,54 @@ impl Operator {
path: String,
(opts, expire): (options::DeleteOptions, Duration),
) -> Result<PresignedRequest> {
+ let mut opts = opts;
+ if let Some(metadata) = opts.if_not_changed.take() {
+ let capability = srv.capability();
+ if capability.delete_with_if_version_match
+ && let Some(version) = metadata.version()
+ {
+ if let Some(explicit) = opts.if_version_match.as_deref() {
+ if explicit != version {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_version_match",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
+ } else {
+ opts.if_version_match = Some(version.to_string());
+ }
+ } else if capability.delete_with_if_match
+ && let Some(etag) = metadata.etag()
+ {
+ if let Some(explicit) = opts.if_match.as_deref() {
+ if explicit != etag {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "if_not_changed conflicts with if_match",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
+ } else {
+ opts.if_match = Some(etag.to_string());
+ }
+ } else if !capability.delete_with_if_version_match && !capability.delete_with_if_match {
+ return Err(Error::new(
+ ErrorKind::Unsupported,
+ format!(
+ "The service {} does not support the operation delete with if_not_changed",
+ srv.info().scheme()
+ ),
+ )
+ .with_operation(Operation::Delete.into_static()));
+ } else {
+ return Err(Error::new(
+ ErrorKind::ConfigInvalid,
+ "if_not_changed metadata does not contain an identity supported by delete",
+ )
+ .with_operation(Operation::Delete.into_static()));
+ }
+ }
let op = OpPresign::new(OpDelete::from(opts), expire);
let rp = srv.presign(&ctx, &path, op).await?;
Ok(rp.into_presigned_request())
diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs
index b8b6c42f5..79788f749 100644
--- a/core/core/src/types/operator/operator_futures.rs
+++ b/core/core/src/types/operator/operator_futures.rs
@@ -101,6 +101,18 @@ impl<F: Future<Output = Result<Metadata>>> FutureStat<F> {
self
}
+ /// Refer to [`options::StatOptions::if_version_match`] for more details.
+ pub fn if_version_match(mut self, v: &str) -> Self {
+ self.args.if_version_match = Some(v.to_string());
+ self
+ }
+
+ /// Refer to [`options::StatOptions::if_version_not_match`] for more details.
+ pub fn if_version_not_match(mut self, v: &str) -> Self {
+ self.args.if_version_not_match = Some(v.to_string());
+ self
+ }
+
/// Set the If-Modified-Since for this operation.
///
/// Refer to [`options::StatOptions::if_modified_since`] for more details.
@@ -378,6 +390,18 @@ impl<F: Future<Output = Result<Buffer>>> FutureRead<F> {
self
}
+ /// Set the current object version that this read must match.
+ pub fn if_version_match(mut self, v: &str) -> Self {
+ self.args.if_version_match = Some(v.to_string());
+ self
+ }
+
+ /// Set the current object version that this read must not match.
+ pub fn if_version_not_match(mut self, v: &str) -> Self {
+ self.args.if_version_not_match = Some(v.to_string());
+ self
+ }
+
/// ## `if_modified_since`
///
/// Set `if_modified_since` for this `read` request.
@@ -580,6 +604,18 @@ impl<F: Future<Output = Result<Reader>>> FutureReader<F> {
self
}
+ /// Set the current object version that this reader must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current object version that this reader must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.if_version_not_match = Some(version.to_string());
+ self
+ }
+
/// Set `if-modified-since` for this `read` request.
///
/// This feature can be used to check if the file has been modified since the given timestamp.
@@ -872,6 +908,24 @@ impl<F: Future<Output = Result<Metadata>>> FutureWrite<F> {
self
}
+ /// Set the current object version that this write must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current object version that this write must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Write only when the object still matches `metadata`.
+ pub fn if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.args.0.if_not_changed = Some(metadata.clone());
+ self
+ }
+
/// Sets the condition that write operation will succeed only if target does not exist.
///
/// Refer to [`options::WriteOptions::if_not_exists`] for more details.
@@ -1209,6 +1263,24 @@ impl<F: Future<Output = Result<Writer>>> FutureWriter<F> {
self
}
+ /// Set the current object version that this writer must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current object version that this writer must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Write only when the object still matches `metadata`.
+ pub fn if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.args.if_not_changed = Some(metadata.clone());
+ self
+ }
+
/// Sets the condition that write operation will succeed only if target does not exist.
///
/// Refer to [`options::WriteOptions::if_not_exists`] for more details.
@@ -1290,6 +1362,30 @@ impl<F: Future<Output = Result<()>>> FutureDelete<F> {
self.args.if_match = Some(etag.to_string());
self
}
+
+ /// Set `if_none_match` for this delete operation.
+ pub fn if_none_match(mut self, etag: &str) -> Self {
+ self.args.if_none_match = Some(etag.to_string());
+ self
+ }
+
+ /// Set the current object version that this delete must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current object version that this delete must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Delete only when the object still matches `metadata`.
+ pub fn if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.args.if_not_changed = Some(metadata.clone());
+ self
+ }
}
/// This future creates a [`Deleter`].
@@ -1457,6 +1553,30 @@ impl<F: Future<Output = Result<Metadata>>> FutureCopy<F> {
self
}
+ /// Set the destination ETag that this copy must not match.
+ pub fn if_none_match(mut self, etag: &str) -> Self {
+ self.args.0.if_none_match = Some(etag.to_string());
+ self
+ }
+
+ /// Set the current destination version that this copy must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current destination version that this copy must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Copy only when the destination still matches `metadata`.
+ pub fn if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.args.0.if_not_changed = Some(metadata.clone());
+ self
+ }
+
/// Sets source version for this copy operation.
///
/// Refer to [`options::CopyOptions::source_version`] for more details.
@@ -1513,6 +1633,30 @@ impl<F: Future<Output = Result<Copier>>> FutureCopier<F> {
self
}
+ /// Set the destination ETag that this copier must not match.
+ pub fn if_none_match(mut self, etag: &str) -> Self {
+ self.args.0.if_none_match = Some(etag.to_string());
+ self
+ }
+
+ /// Set the current destination version that this copier must match.
+ pub fn if_version_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_match = Some(version.to_string());
+ self
+ }
+
+ /// Set the current destination version that this copier must not match.
+ pub fn if_version_not_match(mut self, version: &str) -> Self {
+ self.args.0.if_version_not_match = Some(version.to_string());
+ self
+ }
+
+ /// Copy only when the destination still matches `metadata`.
+ pub fn if_not_changed(mut self, metadata: &Metadata) -> Self {
+ self.args.0.if_not_changed = Some(metadata.clone());
+ self
+ }
+
/// Sets source version for this copier operation.
///
/// Refer to [`options::CopyOptions::source_version`] for more details.
diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs
index 15c8aa329..dc75181c2 100644
--- a/core/core/src/types/options.rs
+++ b/core/core/src/types/options.rs
@@ -17,6 +17,7 @@
//! Options module provides options definitions for operations.
+use crate::Metadata;
use crate::raw::Timestamp;
use crate::types::BytesRange;
use std::collections::HashMap;
@@ -44,6 +45,23 @@ pub struct DeleteOptions {
/// - If supported, the delete will only succeed when the existing object's
/// ETag matches the given value.
pub if_match: Option<String>,
+ /// Delete only when the current object's ETag does not match this value.
+ ///
+ /// Check [`crate::Capability::delete_with_if_none_match`] before using this feature.
+ pub if_none_match: Option<String>,
+ /// Delete only when the current object's version matches this value.
+ ///
+ /// Check [`crate::Capability::delete_with_if_version_match`] before using this feature.
+ pub if_version_match: Option<String>,
+ /// Delete only when the current object's version does not match this value.
+ ///
+ /// Check [`crate::Capability::delete_with_if_version_not_match`] before using this feature.
+ pub if_version_not_match: Option<String>,
+ /// Delete only when the object still matches the supplied metadata.
+ ///
+ /// OpenDAL prefers `version` when the service supports version preconditions and
+ /// otherwise uses `ETag`.
+ pub if_not_changed: Option<Metadata>,
}
/// Options for list operations.
@@ -103,6 +121,10 @@ pub struct ReadOptions {
/// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`]
/// will be returned.
pub if_none_match: Option<String>,
+ /// Read only when the current object's version matches this value.
+ pub if_version_match: Option<String>,
+ /// Read only when the current object's version does not match this value.
+ pub if_version_not_match: Option<String>,
/// Set `if_modified_since` for this operation.
///
/// This option can be used to check if the file has been modified since the given timestamp.
@@ -200,6 +222,10 @@ pub struct ReaderOptions {
/// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`]
/// will be returned.
pub if_none_match: Option<String>,
+ /// Read only when the current object's version matches this value.
+ pub if_version_match: Option<String>,
+ /// Read only when the current object's version does not match this value.
+ pub if_version_not_match: Option<String>,
/// Set `if_modified_since` for this operation.
///
/// This option can be used to check if the file has been modified since the given timestamp.
@@ -298,6 +324,10 @@ pub struct StatOptions {
/// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`]
/// will be returned.
pub if_none_match: Option<String>,
+ /// Stat only when the current object's version matches this value.
+ pub if_version_match: Option<String>,
+ /// Stat only when the current object's version does not match this value.
+ pub if_version_not_match: Option<String>,
/// Set `if_modified_since` for this operation.
///
/// This option can be used to check if the file has been modified since the given timestamp.
@@ -485,6 +515,14 @@ pub struct WriteOptions {
/// This operation provides conditional write functionality based on ETag non-matching,
/// useful for preventing overwriting existing resources or ensuring unique writes.
pub if_none_match: Option<String>,
+ /// Write only when the current object's version matches this value.
+ ///
+ /// Check [`crate::Capability::write_with_if_version_match`] before using this feature.
+ pub if_version_match: Option<String>,
+ /// Write only when the current object's version does not match this value.
+ ///
+ /// Check [`crate::Capability::write_with_if_version_not_match`] before using this feature.
+ pub if_version_not_match: Option<String>,
/// Sets the condition that write operation will succeed only if target does not exist.
///
/// ### Capability
@@ -500,6 +538,11 @@ pub struct WriteOptions {
/// This operation provides a way to ensure write operations only create new resources
/// without overwriting existing ones, useful for implementing "create if not exists" logic.
pub if_not_exists: bool,
+ /// Write only when the object still matches the supplied metadata.
+ ///
+ /// OpenDAL prefers `version` when the service supports version preconditions and
+ /// otherwise uses `ETag`.
+ pub if_not_changed: Option<Metadata>,
/// Sets concurrent write operations for this writer.
///
@@ -585,6 +628,23 @@ pub struct CopyOptions {
/// - If supported, the copy operation will only succeed when the existing
/// destination object's ETag matches the given value.
pub if_match: Option<String>,
+ /// Copy only when the destination ETag does not match this value.
+ ///
+ /// Check [`crate::Capability::copy_with_if_none_match`] before using this feature.
+ pub if_none_match: Option<String>,
+ /// Copy only when the current destination version matches this value.
+ ///
+ /// Check [`crate::Capability::copy_with_if_version_match`] before using this feature.
+ pub if_version_match: Option<String>,
+ /// Copy only when the current destination version does not match this value.
+ ///
+ /// Check [`crate::Capability::copy_with_if_version_not_match`] before using this feature.
+ pub if_version_not_match: Option<String>,
+ /// Copy only when the destination still matches the supplied metadata.
+ ///
+ /// OpenDAL prefers `version` when the service supports version preconditions and
+ /// otherwise uses `ETag`.
+ pub if_not_changed: Option<Metadata>,
/// Copy from a specific source object version.
///
diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs
index 7f269ee38..92c0566a6 100644
--- a/core/services/azblob/src/backend.rs
+++ b/core/services/azblob/src/backend.rs
@@ -413,10 +413,14 @@ impl Builder for AzblobBuilder {
write_with_user_metadata: true,
delete: true,
+ delete_with_if_match: true,
+ delete_with_if_none_match: true,
delete_max_size: Some(AZBLOB_BATCH_LIMIT),
copy: true,
copy_with_if_not_exists: true,
+ copy_with_if_match: true,
+ copy_with_if_none_match: true,
copy_can_multi: true,
copy_multi_min_size: Some(AZBLOB_COPY_MIN_BLOCK_SIZE),
copy_multi_max_size: Some(AZBLOB_COPY_MAX_BLOCK_SIZE),
@@ -612,9 +616,9 @@ impl Service for AzblobBackend {
let req = match args.operation() {
PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v),
PresignOperation::Read(range, v) => self.core.azblob_get_blob_request(path, *range, v),
- PresignOperation::Write(_) => {
+ PresignOperation::Write(v) => {
self.core
- .azblob_put_blob_request(path, None, &OpWrite::default(), Buffer::new())
+ .azblob_put_blob_request(path, None, v, Buffer::new())
}
PresignOperation::Delete(_) => Err(Error::new(
ErrorKind::Unsupported,
diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs
index c70ad3a36..1895a4dce 100644
--- a/core/services/azblob/src/core.rs
+++ b/core/services/azblob/src/core.rs
@@ -301,7 +301,6 @@ impl AzblobCore {
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);
}
@@ -651,6 +650,12 @@ impl AzblobCore {
if args.if_not_exists() {
req = req.header(IF_NONE_MATCH, "*");
}
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+ if let Some(if_none_match) = args.if_none_match() {
+ req = req.header(IF_NONE_MATCH, if_none_match);
+ }
let content = quick_xml::se::to_string(&PutBlockListRequest {
latest: block_ids
@@ -725,9 +730,16 @@ impl AzblobCore {
self.send(ctx, req).await
}
- fn azblob_delete_blob_request(&self, path: &str) -> Result<Request<Buffer>> {
- Request::delete(self.build_path_url(path))
- .header(CONTENT_LENGTH, 0)
+ fn azblob_delete_blob_request(&self, path: &str, args: &OpDelete) -> Result<Request<Buffer>> {
+ let mut req = Request::delete(self.build_path_url(path));
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+ if let Some(if_none_match) = args.if_none_match() {
+ req = req.header(IF_NONE_MATCH, if_none_match);
+ }
+
+ req.header(CONTENT_LENGTH, 0)
.extension(Operation::Delete)
.extension(ServiceOperation("DeleteBlob"))
.body(Buffer::new())
@@ -738,8 +750,9 @@ impl AzblobCore {
&self,
ctx: &OperationContext,
path: &str,
+ args: &OpDelete,
) -> Result<Response<Buffer>> {
- let req = self.azblob_delete_blob_request(path)?;
+ let req = self.azblob_delete_blob_request(path, args)?;
let req = self.sign(ctx, req).await?;
self.send(ctx, req).await
}
@@ -767,6 +780,12 @@ impl AzblobCore {
if args.if_not_exists() {
req = req.header(IF_NONE_MATCH, "*");
}
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+ if let Some(if_none_match) = args.if_none_match() {
+ req = req.header(IF_NONE_MATCH, if_none_match);
+ }
let req = req
.extension(Operation::Copy)
@@ -817,7 +836,7 @@ impl AzblobCore {
pub async fn azblob_batch_delete(
&self,
ctx: &OperationContext,
- paths: &[String],
+ paths: &[(String, OpDelete)],
) -> Result<Response<Buffer>> {
let url = format!(
"{}/{}?restype=container&comp=batch",
@@ -826,8 +845,8 @@ impl AzblobCore {
let mut multipart = Multipart::new();
- for (idx, path) in paths.iter().enumerate() {
- let req = self.azblob_delete_blob_request(path)?;
+ for (idx, (path, args)) in paths.iter().enumerate() {
+ let req = self.azblob_delete_blob_request(path, args)?;
let req = self.batch_sign(ctx, req).await?;
multipart = multipart.part(
diff --git a/core/services/azblob/src/deleter.rs b/core/services/azblob/src/deleter.rs
index 06e365381..d43c70fca 100644
--- a/core/services/azblob/src/deleter.rs
+++ b/core/services/azblob/src/deleter.rs
@@ -37,20 +37,28 @@ impl AzblobDeleter {
}
impl oio::BatchDelete for AzblobDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let resp = self.core.azblob_delete_blob(&self.ctx, &path).await?;
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ let resp = self
+ .core
+ .azblob_delete_blob(&self.ctx, &path, &args)
+ .await?;
let status = resp.status();
match status {
- StatusCode::ACCEPTED | StatusCode::NOT_FOUND => Ok(()),
+ StatusCode::ACCEPTED => Ok(()),
+ StatusCode::NOT_FOUND if args.if_match().is_some() => Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ )),
+ StatusCode::NOT_FOUND => Ok(()),
_ => Err(parse_error(resp)),
}
}
async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
// TODO: Add remove version support.
- let paths = batch.into_iter().map(|(p, _)| p).collect::<Vec<_>>();
+ let paths = batch;
// construct and complete batch request
let resp = self.core.azblob_batch_delete(&self.ctx, &paths).await?;
@@ -86,15 +94,23 @@ impl oio::BatchDelete for AzblobDeleter {
for (i, part) in parts.into_iter().enumerate() {
let resp = part.into_response();
- let path = paths[i].clone();
-
- // deleting not existing objects is ok
- if resp.status() == StatusCode::ACCEPTED || resp.status() == StatusCode::NOT_FOUND {
- batched_result.succeeded.push((path, OpDelete::default()));
+ let (path, op) = paths[i].clone();
+
+ if resp.status() == StatusCode::NOT_FOUND && op.if_match().is_some() {
+ batched_result.failed.push((
+ path,
+ op,
+ Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ ),
+ ));
+ } else if resp.status() == StatusCode::ACCEPTED
+ || resp.status() == StatusCode::NOT_FOUND
+ {
+ batched_result.succeeded.push((path, op));
} else {
- batched_result
- .failed
- .push((path, OpDelete::default(), parse_error(resp)));
+ batched_result.failed.push((path, op, parse_error(resp)));
}
}
diff --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs
index c038fb5de..e1cf9fde2 100644
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@ -45,6 +45,7 @@ use super::core::*;
use super::deleter::GcsDeleter;
use super::lister::GcsLister;
use super::reader::*;
+use super::writer::GcsConditionalWriter;
use super::writer::GcsWriter;
use super::writer::GcsWriters;
use opendal_core::raw::*;
@@ -343,12 +344,16 @@ impl Builder for GcsBuilder {
stat: true,
stat_with_if_match: true,
stat_with_if_none_match: true,
+ stat_with_if_version_match: true,
+ stat_with_if_version_not_match: true,
read: true,
read_with_suffix: true,
read_with_if_match: true,
read_with_if_none_match: true,
+ read_with_if_version_match: true,
+ read_with_if_version_not_match: true,
write: true,
write_can_empty: true,
@@ -358,6 +363,8 @@ impl Builder for GcsBuilder {
write_with_content_encoding: true,
write_with_user_metadata: true,
write_with_if_not_exists: true,
+ write_with_if_version_match: true,
+ write_with_if_version_not_match: true,
// The min multipart size of Gcs is 5 MiB.
//
@@ -373,9 +380,13 @@ impl Builder for GcsBuilder {
},
delete: true,
+ delete_with_if_version_match: true,
+ delete_with_if_version_not_match: true,
delete_max_size: Some(100),
copy: true,
+ copy_with_if_version_match: true,
+ copy_with_if_version_not_match: true,
copy_can_multi: true,
// GCS rewrite requires maxBytesRewrittenPerCall to be an
// integral multiple of 1 MiB if specified.
@@ -455,6 +466,14 @@ impl Service for GcsBackend {
let resp = self.core.gcs_get_object_metadata(ctx, path, &args).await?;
if !resp.status().is_success() {
+ if resp.status() == http::StatusCode::NOT_FOUND
+ && (args.if_version_match().is_some() || args.if_version_not_match().is_some())
+ {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "version precondition requires a live target",
+ ));
+ }
return Err(parse_error(resp));
}
@@ -477,15 +496,24 @@ impl Service for GcsBackend {
}
fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result<Self::Writer> {
- let output: GcsWriters = {
+ let conditional = args.if_not_exists()
+ || args.if_version_match().is_some()
+ || args.if_version_not_match().is_some();
+ let output: GcsWriters = if conditional {
+ TwoWays::Two(GcsConditionalWriter::new(
+ self.core.clone(),
+ ctx.clone(),
+ path,
+ args,
+ ))
+ } else {
let concurrent = args.concurrent();
let w = GcsWriter::new(self.core.clone(), ctx.clone(), path, args);
// Multipart uploads schedule work through the operation executor
// supplied by the caller.
let w = oio::MultipartWriter::new(ctx.executor().clone(), w, concurrent);
-
- Ok(w)
- }?;
+ TwoWays::One(w)
+ };
Ok(output)
}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 5faf9ac04..5d87f9513 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -43,6 +43,7 @@ use opendal_core::raw::*;
use opendal_core::*;
pub mod constants {
+ pub const X_GOOG_GENERATION: &str = "x-goog-generation";
pub const GCS_REWRITE_MIN_CHUNK_SIZE: usize = 1024 * 1024;
#[cfg(target_pointer_width = "64")]
pub const GCS_REWRITE_MAX_CHUNK_SIZE: usize =
@@ -162,12 +163,21 @@ impl GcsCore {
let p = build_abs_path(&self.root, path);
let url = format!(
- "{}/storage/v1/b/{}/o/{}?alt=media",
+ "{}/storage/v1/b/{}/o/{}",
self.endpoint,
self.bucket,
gcs_percent_encode_path(&p)
);
+ let mut url = QueryPairsWriter::new(&url).push("alt", "media");
+ if let Some(version) = args.if_version_match() {
+ url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version));
+ }
+ if let Some(version) = args.if_version_not_match() {
+ url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version));
+ }
+ let url = url.finish();
+
let mut req = Request::get(&url);
if let Some(if_match) = args.if_match() {
@@ -208,6 +218,9 @@ impl GcsCore {
if let Some(if_none_match) = args.if_none_match() {
req = req.header(IF_NONE_MATCH, if_none_match);
}
+ if let Some(version) = args.if_version_match() {
+ req = req.header("x-goog-if-generation-match", version);
+ }
if let Some(if_modified_since) = args.if_modified_since() {
req = req.header(IF_MODIFIED_SINCE, if_modified_since.format_http_date());
@@ -275,12 +288,24 @@ impl GcsCore {
write!(&mut url, "&predefinedAcl={acl}").unwrap();
}
- // Makes the operation conditional on whether the object's current generation
- // matches the given value. Setting to 0 makes the operation succeed only if
- // there are no live versions of the object.
- if op.if_not_exists() {
+ if let Some(version) = op.if_version_match() {
+ write!(
+ &mut url,
+ "&ifGenerationMatch={}",
+ gcs_percent_encode_path(version)
+ )
+ .unwrap();
+ } else if op.if_not_exists() {
write!(&mut url, "&ifGenerationMatch=0").unwrap();
}
+ if let Some(version) = op.if_version_not_match() {
+ write!(
+ &mut url,
+ "&ifGenerationNotMatch={}",
+ gcs_percent_encode_path(version)
+ )
+ .unwrap();
+ }
let mut req = Request::post(&url);
@@ -329,6 +354,105 @@ impl GcsCore {
}
}
+ pub fn gcs_initiate_resumable_upload_request(
+ &self,
+ path: &str,
+ op: &OpWrite,
+ ) -> Result<Request<Buffer>> {
+ let p = build_abs_path(&self.root, path);
+ let base = format!("{}/upload/storage/v1/b/{}/o", self.endpoint, self.bucket);
+ let mut url = QueryPairsWriter::new(&base)
+ .push("uploadType", "resumable")
+ .push("name", &gcs_percent_encode_path(&p));
+
+ if let Some(acl) = &self.predefined_acl {
+ url = url.push("predefinedAcl", acl);
+ }
+ if let Some(version) = op.if_version_match() {
+ url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version));
+ } else if op.if_not_exists() {
+ url = url.push("ifGenerationMatch", "0");
+ }
+ if let Some(version) = op.if_version_not_match() {
+ url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version));
+ }
+
+ let metadata = InsertRequestMetadata {
+ storage_class: self.default_storage_class.as_deref(),
+ cache_control: op.cache_control(),
+ content_type: op.content_type(),
+ content_encoding: op.content_encoding(),
+ metadata: op.user_metadata(),
+ };
+ let body = serde_json::to_vec(&metadata).map_err(new_json_serialize_error)?;
+
+ let mut req = Request::post(url.finish())
+ .header(CONTENT_TYPE, "application/json; charset=UTF-8")
+ .header(CONTENT_LENGTH, body.len());
+ if let Some(content_type) = op.content_type() {
+ req = req.header("x-upload-content-type", content_type);
+ }
+
+ req.extension(Operation::Write)
+ .extension(ServiceOperation("InitiateResumableUpload"))
+ .body(Buffer::from(Bytes::from(body)))
+ .map_err(new_request_build_error)
+ }
+
+ pub async fn gcs_initiate_resumable_upload(
+ &self,
+ ctx: &OperationContext,
+ path: &str,
+ op: &OpWrite,
+ ) -> Result<Response<Buffer>> {
+ let req = self.gcs_initiate_resumable_upload_request(path, op)?;
+ let req = self.sign(ctx, req).await?;
+ self.send(ctx, req).await
+ }
+
+ pub async fn gcs_upload_resumable_chunk(
+ &self,
+ ctx: &OperationContext,
+ session_uri: &str,
+ offset: u64,
+ body: Buffer,
+ total: Option<u64>,
+ ) -> Result<Response<Buffer>> {
+ let end = offset + body.len() as u64;
+ let content_range = if body.is_empty() {
+ format!("bytes */{}", total.unwrap_or_default())
+ } else {
+ format!(
+ "bytes {}-{}/{}",
+ offset,
+ end - 1,
+ total.map_or_else(|| "*".to_string(), |v| v.to_string())
+ )
+ };
+ let req = Request::put(session_uri)
+ .header(CONTENT_LENGTH, body.len())
+ .header("content-range", content_range)
+ .extension(Operation::Write)
+ .extension(ServiceOperation("UploadResumableChunk"))
+ .body(body)
+ .map_err(new_request_build_error)?;
+ self.send(ctx, req).await
+ }
+
+ pub async fn gcs_cancel_resumable_upload(
+ &self,
+ ctx: &OperationContext,
+ session_uri: &str,
+ ) -> Result<Response<Buffer>> {
+ let req = Request::delete(session_uri)
+ .header(CONTENT_LENGTH, 0)
+ .extension(Operation::Write)
+ .extension(ServiceOperation("CancelResumableUpload"))
+ .body(Buffer::new())
+ .map_err(new_request_build_error)?;
+ self.send(ctx, req).await
+ }
+
// It's for presign operation. Gcs only supports query sign over XML API.
pub fn gcs_insert_object_xml_request(
&self,
@@ -368,6 +492,18 @@ impl GcsCore {
req = req.header(X_GOOG_STORAGE_CLASS, storage_class);
}
+ if let Some(version) = args.if_version_match() {
+ req = req.header("x-goog-if-generation-match", version);
+ } else if args.if_not_exists() {
+ req = req.header("x-goog-if-generation-match", "0");
+ }
+ if let Some(if_match) = args.if_match() {
+ req = req.header(IF_MATCH, if_match);
+ }
+ if let Some(if_none_match) = args.if_none_match() {
+ req = req.header(IF_NONE_MATCH, if_none_match);
+ }
+
let req = req
.extension(Operation::Write)
.extension(ServiceOperation("InsertObject"));
@@ -387,6 +523,15 @@ impl GcsCore {
gcs_percent_encode_path(&p)
);
+ let mut url = QueryPairsWriter::new(&url);
+ if let Some(version) = args.if_version_match() {
+ url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version));
+ }
+ if let Some(version) = args.if_version_not_match() {
+ url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version));
+ }
+ let url = url.finish();
+
let mut req = Request::get(&url);
if let Some(if_none_match) = args.if_none_match() {
@@ -396,7 +541,6 @@ impl GcsCore {
if let Some(if_match) = args.if_match() {
req = req.header(IF_MATCH, if_match);
}
-
let req = req
.extension(Operation::Stat)
.extension(ServiceOperation("GetObject"));
@@ -425,6 +569,9 @@ impl GcsCore {
if let Some(if_match) = args.if_match() {
req = req.header(IF_MATCH, if_match);
}
+ if let Some(version) = args.if_version_match() {
+ req = req.header("x-goog-if-generation-match", version);
+ }
let req = req
.extension(Operation::Stat)
@@ -452,14 +599,19 @@ impl GcsCore {
&self,
ctx: &OperationContext,
path: &str,
+ args: &OpDelete,
) -> Result<Response<Buffer>> {
- let req = self.gcs_delete_object_request(path)?;
+ let req = self.gcs_delete_object_request(path, args)?;
let req = self.sign(ctx, req).await?;
self.send(ctx, req).await
}
- pub fn gcs_delete_object_request(&self, path: &str) -> Result<Request<Buffer>> {
+ pub fn gcs_delete_object_request(
+ &self,
+ path: &str,
+ args: &OpDelete,
+ ) -> Result<Request<Buffer>> {
let p = build_abs_path(&self.root, path);
let url = format!(
@@ -469,6 +621,18 @@ impl GcsCore {
gcs_percent_encode_path(&p)
);
+ let mut url = QueryPairsWriter::new(&url);
+ if let Some(version) = args.version() {
+ url = url.push("generation", &gcs_percent_encode_path(version));
+ }
+ if let Some(version) = args.if_version_match() {
+ url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version));
+ }
+ if let Some(version) = args.if_version_not_match() {
+ url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version));
+ }
+ let url = url.finish();
+
Request::delete(&url)
.extension(Operation::Delete)
.extension(ServiceOperation("DeleteObject"))
@@ -479,14 +643,14 @@ impl GcsCore {
pub async fn gcs_delete_objects(
&self,
ctx: &OperationContext,
- paths: Vec<String>,
+ paths: &[(String, OpDelete)],
) -> Result<Response<Buffer>> {
let uri = format!("{}/batch/storage/v1", self.endpoint);
let mut multipart = Multipart::new();
- for (idx, path) in paths.iter().enumerate() {
- let req = self.gcs_delete_object_request(path)?;
+ for (idx, (path, args)) in paths.iter().enumerate() {
+ let req = self.gcs_delete_object_request(path, args)?;
multipart = multipart.part(
MixedPart::from_request(req).part_header("content-id".parse().unwrap(), idx.into()),
@@ -502,15 +666,14 @@ impl GcsCore {
self.send(ctx, req).await
}
- pub async fn gcs_rewrite_object(
+ pub fn gcs_rewrite_object_request(
&self,
- ctx: &OperationContext,
from: &str,
to: &str,
args: &OpCopy,
max_bytes_rewritten_per_call: Option<usize>,
rewrite_token: Option<&str>,
- ) -> Result<Response<Buffer>> {
+ ) -> Result<Request<Buffer>> {
let source = build_abs_path(&self.root, from);
let dest = build_abs_path(&self.root, to);
@@ -528,9 +691,14 @@ impl GcsCore {
if let Some(version) = args.source_version() {
url = url.push("sourceGeneration", &gcs_percent_encode_path(version));
}
- if args.if_not_exists() {
+ if let Some(version) = args.if_version_match() {
+ url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version));
+ } else if args.if_not_exists() {
url = url.push("ifGenerationMatch", "0");
}
+ if let Some(version) = args.if_version_not_match() {
+ url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version));
+ }
if let Some(max_bytes) = max_bytes_rewritten_per_call {
url = url.push("maxBytesRewrittenPerCall", &max_bytes.to_string());
}
@@ -538,13 +706,30 @@ impl GcsCore {
url = url.push("rewriteToken", &gcs_percent_encode_path(token));
}
- let req = Request::post(url.finish())
+ Request::post(url.finish())
.header(CONTENT_LENGTH, 0)
.extension(Operation::Copy)
.extension(ServiceOperation("RewriteObject"))
.body(Buffer::new())
- .map_err(new_request_build_error)?;
+ .map_err(new_request_build_error)
+ }
+ pub async fn gcs_rewrite_object(
+ &self,
+ ctx: &OperationContext,
+ from: &str,
+ to: &str,
+ args: &OpCopy,
+ max_bytes_rewritten_per_call: Option<usize>,
+ rewrite_token: Option<&str>,
+ ) -> Result<Response<Buffer>> {
+ let req = self.gcs_rewrite_object_request(
+ from,
+ to,
+ args,
+ max_bytes_rewritten_per_call,
+ rewrite_token,
+ )?;
let req = self.sign(ctx, req).await?;
self.send(ctx, req).await
}
@@ -867,6 +1052,7 @@ pub struct ListResponseItem {
pub size: String,
// metadata
pub etag: String,
+ pub generation: String,
pub md5_hash: String,
pub updated: String,
pub content_type: String,
@@ -960,8 +1146,7 @@ mod tests {
use reqsign_google::RequestSigner;
use reqsign_google::TokenCredentialProvider;
- #[tokio::test]
- async fn test_insert_object_signing_preserves_wire_uri() {
+ fn test_core() -> GcsCore {
let sign_ctx = Context::new();
let signer = Signer::new(
sign_ctx.clone(),
@@ -969,7 +1154,7 @@ mod tests {
RequestSigner::new("storage"),
);
- let core = GcsCore {
+ GcsCore {
info: ServiceInfo::new("gcs", "/", "test-bucket"),
capability: Capability::default(),
endpoint: "https://storage.googleapis.com".to_string(),
@@ -980,7 +1165,12 @@ mod tests {
predefined_acl: None,
default_storage_class: None,
skip_signature: false,
- };
+ }
+ }
+
+ #[tokio::test]
+ async fn test_insert_object_signing_preserves_wire_uri() {
+ let core = test_core();
let req = core
.gcs_insert_object_request(
"nested/object #1.txt",
@@ -1006,6 +1196,168 @@ mod tests {
assert_eq!(signed.uri(), &original_uri);
}
+ #[test]
+ fn test_generation_preconditions_are_mapped_to_json_api() {
+ let core = test_core();
+
+ let read = core
+ .gcs_get_object_request(
+ "object",
+ BytesRange::default(),
+ &OpRead::default().with_if_version_match("123"),
+ )
+ .expect("read request must build");
+ assert!(
+ read.uri()
+ .query()
+ .unwrap()
+ .contains("ifGenerationMatch=123")
+ );
+
+ let read_zero = core
+ .gcs_get_object_request(
+ "object",
+ BytesRange::default(),
+ &OpRead::default().with_if_version_match("0"),
+ )
+ .expect("generation zero must be forwarded");
+ assert!(
+ read_zero
+ .uri()
+ .query()
+ .unwrap()
+ .contains("ifGenerationMatch=0")
+ );
+
+ let stat = core
+ .gcs_head_object_request(
+ "object",
+ &OpStat::default().with_if_version_not_match("456"),
+ )
+ .expect("stat request must build");
+ assert!(
+ stat.uri()
+ .query()
+ .unwrap()
+ .contains("ifGenerationNotMatch=456")
+ );
+
+ let write = core
+ .gcs_insert_object_request(
+ "object",
+ Some(0),
+ &OpWrite::default()
+ .with_if_not_exists(true)
+ .with_if_version_match("123"),
+ Buffer::new(),
+ )
+ .expect("write request must build");
+ let query = write.uri().query().unwrap();
+ assert!(query.contains("ifGenerationMatch=123"));
+ assert_eq!(query.matches("ifGenerationMatch=").count(), 1);
+
+ let delete = core
+ .gcs_delete_object_request(
+ "object",
+ &OpDelete::default().with_if_version_not_match("456"),
+ )
+ .expect("delete request must build");
+ assert!(
+ delete
+ .uri()
+ .query()
+ .unwrap()
+ .contains("ifGenerationNotMatch=456")
+ );
+
+ let resumable = core
+ .gcs_initiate_resumable_upload_request(
+ "object",
+ &OpWrite::default()
+ .with_if_not_exists(true)
+ .with_if_version_match("123"),
+ )
+ .expect("resumable request must build");
+ let query = resumable.uri().query().unwrap();
+ assert!(query.contains("ifGenerationMatch=123"));
+ assert_eq!(query.matches("ifGenerationMatch=").count(), 1);
+
+ let rewrite = core
+ .gcs_rewrite_object_request(
+ "source",
+ "target",
+ &OpCopy::default().with_if_version_not_match("456"),
+ Some(GCS_REWRITE_MIN_CHUNK_SIZE),
+ Some("token"),
+ )
+ .expect("rewrite request must build");
+ let query = rewrite.uri().query().unwrap();
+ assert!(query.contains("ifGenerationNotMatch=456"));
+ assert!(query.contains("rewriteToken=token"));
+
+ let rewrite = core
+ .gcs_rewrite_object_request(
+ "source",
+ "target",
+ &OpCopy::default()
+ .with_if_not_exists(true)
+ .with_if_version_match("123"),
+ None,
+ None,
+ )
+ .expect("rewrite request must build");
+ let query = rewrite.uri().query().unwrap();
+ assert!(query.contains("ifGenerationMatch=123"));
+ assert_eq!(query.matches("ifGenerationMatch=").count(), 1);
+ }
+
+ #[test]
+ fn test_generation_match_is_mapped_to_xml_api() {
+ let core = test_core();
+
+ let read = core
+ .gcs_get_object_xml_request(
+ "object",
+ BytesRange::default(),
+ &OpRead::default().with_if_version_match("123"),
+ )
+ .expect("read request must build");
+ assert_eq!(read.headers()["x-goog-if-generation-match"], "123");
+
+ let stat = core
+ .gcs_head_object_xml_request("object", &OpStat::default().with_if_version_match("123"))
+ .expect("stat request must build");
+ assert_eq!(stat.headers()["x-goog-if-generation-match"], "123");
+
+ let write = core
+ .gcs_insert_object_xml_request(
+ "object",
+ &OpWrite::default()
+ .with_if_not_exists(true)
+ .with_if_version_match("123"),
+ Buffer::new(),
+ )
+ .expect("write request must build");
+ assert_eq!(write.headers()["x-goog-if-generation-match"], "123");
+ assert_eq!(
+ write
+ .headers()
+ .get_all("x-goog-if-generation-match")
+ .iter()
+ .count(),
+ 1
+ );
+
+ let create = core
+ .gcs_insert_object_xml_request(
+ "object",
+ &OpWrite::default().with_if_not_exists(true),
+ Buffer::new(),
+ )
+ .expect("create request must build");
+ assert_eq!(create.headers()["x-goog-if-generation-match"], "0");
+ }
+
#[test]
fn test_deserialize_get_object_json_response() {
let content = r#"{
@@ -1118,6 +1470,7 @@ mod tests {
assert_eq!(output.items[0].size, "56535");
assert_eq!(output.items[0].md5_hash, "fHcEH1vPwA6eTPqxuasXcg==");
assert_eq!(output.items[0].etag, "CKWasoTgyPkCEAE=");
+ assert_eq!(output.items[0].generation, "1660563214863653");
assert_eq!(output.items[0].updated, "2022-08-15T11:33:34.866Z");
assert_eq!(output.items[1].name, "2.png");
assert_eq!(output.items[1].size, "45506");
diff --git a/core/services/gcs/src/deleter.rs b/core/services/gcs/src/deleter.rs
index 97abfdd14..44bc99fab 100644
--- a/core/services/gcs/src/deleter.rs
+++ b/core/services/gcs/src/deleter.rs
@@ -37,11 +37,19 @@ impl GcsDeleter {
}
impl oio::BatchDelete for GcsDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let resp = self.core.gcs_delete_object(&self.ctx, &path).await?;
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ let resp = self.core.gcs_delete_object(&self.ctx, &path, &args).await?;
- // deleting not existing objects is ok
- if resp.status().is_success() || resp.status() == StatusCode::NOT_FOUND {
+ if resp.status().is_success() {
+ Ok(())
+ } else if resp.status() == StatusCode::NOT_FOUND
+ && (args.if_version_match().is_some() || args.if_version_not_match().is_some())
+ {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ ))
+ } else if resp.status() == StatusCode::NOT_FOUND {
Ok(())
} else {
Err(parse_error(resp))
@@ -49,11 +57,7 @@ impl oio::BatchDelete for GcsDeleter {
}
async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result<BatchDeleteResult> {
- let paths: Vec<String> = batch.into_iter().map(|(p, _)| p).collect();
- let resp = self
- .core
- .gcs_delete_objects(&self.ctx, paths.clone())
- .await?;
+ let resp = self.core.gcs_delete_objects(&self.ctx, &batch).await?;
let status = resp.status();
@@ -79,15 +83,25 @@ impl oio::BatchDelete for GcsDeleter {
for (i, part) in parts.into_iter().enumerate() {
let resp = part.into_response();
// TODO: maybe we can take it directly?
- let path = paths[i].clone();
-
- // deleting not existing objects is ok
- if resp.status().is_success() || resp.status() == StatusCode::NOT_FOUND {
- batched_result.succeeded.push((path, OpDelete::default()));
+ let (path, op) = batch[i].clone();
+
+ if resp.status().is_success() {
+ batched_result.succeeded.push((path, op));
+ } else if resp.status() == StatusCode::NOT_FOUND
+ && (op.if_version_match().is_some() || op.if_version_not_match().is_some())
+ {
+ batched_result.failed.push((
+ path,
+ op,
+ Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ ),
+ ));
+ } else if resp.status() == StatusCode::NOT_FOUND {
+ batched_result.succeeded.push((path, op));
} else {
- batched_result
- .failed
- .push((path, OpDelete::default(), parse_error(resp)));
+ batched_result.failed.push((path, op, parse_error(resp)));
}
}
diff --git a/core/services/gcs/src/lister.rs b/core/services/gcs/src/lister.rs
index e0006a730..e71396581 100644
--- a/core/services/gcs/src/lister.rs
+++ b/core/services/gcs/src/lister.rs
@@ -118,6 +118,9 @@ impl oio::PageList for GcsLister {
// set metadata fields
meta.set_content_md5(object.md5_hash.as_str());
meta.set_etag(object.etag.as_str());
+ if !object.generation.is_empty() {
+ meta.set_version(&object.generation);
+ }
let size = object.size.parse().map_err(|e| {
Error::new(ErrorKind::Unexpected, "parse u64 from list response").set_source(e)
diff --git a/core/services/gcs/src/reader.rs b/core/services/gcs/src/reader.rs
index 64f883261..5d784e535 100644
--- a/core/services/gcs/src/reader.rs
+++ b/core/services/gcs/src/reader.rs
@@ -16,6 +16,7 @@
// under the License.
use super::backend::*;
+use super::core::constants::X_GOOG_GENERATION;
use super::core::parse_error;
use http::Response;
use http::StatusCode;
@@ -61,13 +62,24 @@ impl oio::StreamRead for GcsReader {
let status = resp.status();
let (rp, stream) = match status {
- StatusCode::OK | StatusCode::PARTIAL_CONTENT => (
- RpRead::new(parse_into_metadata(path, resp.headers())?),
- resp.into_body(),
- ),
+ StatusCode::OK | StatusCode::PARTIAL_CONTENT => {
+ let mut metadata = parse_into_metadata(path, resp.headers())?;
+ if let Some(generation) = parse_header_to_str(resp.headers(), X_GOOG_GENERATION)? {
+ metadata.set_version(generation);
+ }
+ (RpRead::new(metadata), resp.into_body())
+ }
_ => {
let (part, mut body) = resp.into_parts();
let buf = body.to_buffer().await?;
+ if part.status == StatusCode::NOT_FOUND
+ && (args.if_version_match().is_some() || args.if_version_not_match().is_some())
+ {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "version precondition requires a live target",
+ ));
+ }
return Err(parse_error(Response::from_parts(part, buf)));
}
};
diff --git a/core/services/gcs/src/writer.rs b/core/services/gcs/src/writer.rs
index 9e50e3b9e..aa95bc62c 100644
--- a/core/services/gcs/src/writer.rs
+++ b/core/services/gcs/src/writer.rs
@@ -19,15 +19,20 @@ use std::sync::Arc;
use bytes::Buf;
use http::StatusCode;
+use http::header::LOCATION;
+use http::header::RANGE;
use super::core::CompleteMultipartUploadRequestPart;
use super::core::GcsCore;
use super::core::InitiateMultipartUploadResult;
+use super::core::constants::X_GOOG_GENERATION;
use super::core::parse_error;
use opendal_core::raw::*;
use opendal_core::*;
-pub type GcsWriters = oio::MultipartWriter<GcsWriter>;
+pub type GcsWriters = TwoWays<oio::MultipartWriter<GcsWriter>, GcsConditionalWriter>;
+
+const RESUMABLE_CHUNK_SIZE: usize = 8 * 1024 * 1024;
pub struct GcsWriter {
core: Arc<GcsCore>,
@@ -144,10 +149,14 @@ impl oio::MultipartWrite for GcsWriter {
if !resp.status().is_success() {
return Err(parse_error(resp));
}
- // we don't extract metadata from `CompleteMultipartUploadResult`, since we only need the `ETag` from it.
- // However, the `ETag` differs from the `ETag` obtained through the `stat` operation.
- // refer to: https://cloud.google.com/storage/docs/metadata#etags
- Ok(Metadata::default())
+ let mut metadata = Metadata::new(EntryMode::from_path(&self.path));
+ if let Some(etag) = parse_etag(resp.headers())? {
+ metadata.set_etag(etag);
+ }
+ if let Some(generation) = parse_header_to_str(resp.headers(), X_GOOG_GENERATION)? {
+ metadata.set_version(generation);
+ }
+ Ok(metadata)
}
async fn abort_part(&self, upload_id: &str) -> Result<()> {
@@ -162,3 +171,194 @@ impl oio::MultipartWrite for GcsWriter {
}
}
}
+
+pub struct GcsConditionalWriter {
+ core: Arc<GcsCore>,
+ ctx: OperationContext,
+ path: String,
+ op: OpWrite,
+ session_uri: Option<String>,
+ offset: u64,
+ pending: oio::QueueBuf,
+}
+
+impl GcsConditionalWriter {
+ pub fn new(core: Arc<GcsCore>, ctx: OperationContext, path: &str, op: OpWrite) -> Self {
+ Self {
+ core,
+ ctx,
+ path: path.to_string(),
+ op,
+ session_uri: None,
+ offset: 0,
+ pending: oio::QueueBuf::new(),
+ }
+ }
+
+ async fn ensure_session(&mut self) -> Result<&str> {
+ if self.session_uri.is_none() {
+ let resp = self
+ .core
+ .gcs_initiate_resumable_upload(&self.ctx, &self.path, &self.op)
+ .await?;
+ if !resp.status().is_success() {
+ return Err(parse_error(resp));
+ }
+ let location = parse_header_to_str(resp.headers(), LOCATION)?
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "GCS resumable upload response is missing Location",
+ )
+ })?
+ .to_string();
+ self.session_uri = Some(location);
+ }
+ Ok(self.session_uri.as_deref().unwrap())
+ }
+
+ fn persisted_offset(resp: &http::Response<Buffer>) -> Result<u64> {
+ let Some(range) = parse_header_to_str(resp.headers(), RANGE)? else {
+ return Ok(0);
+ };
+ let end = range
+ .strip_prefix("bytes=")
+ .and_then(|v| v.rsplit_once('-'))
+ .map(|(_, end)| end)
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid Range header in GCS resumable upload response",
+ )
+ .with_context("range", range)
+ })?;
+ end.parse::<u64>().map(|v| v + 1).map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid Range header in GCS resumable upload response",
+ )
+ .with_context("range", range)
+ .set_source(err)
+ })
+ }
+
+ async fn upload(&mut self, mut body: Buffer, total: Option<u64>) -> Result<Option<Metadata>> {
+ loop {
+ let session_uri = self.ensure_session().await?.to_string();
+ let sent_offset = self.offset;
+ let sent_end = sent_offset + body.len() as u64;
+ let resp = self
+ .core
+ .gcs_upload_resumable_chunk(
+ &self.ctx,
+ &session_uri,
+ sent_offset,
+ body.clone(),
+ total,
+ )
+ .await?;
+
+ if resp.status() == StatusCode::PERMANENT_REDIRECT {
+ let persisted = Self::persisted_offset(&resp)?;
+ if persisted <= sent_offset || persisted > sent_end {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS resumable upload reported invalid persisted range",
+ )
+ .with_context("offset", sent_offset)
+ .with_context("persisted", persisted)
+ .set_temporary());
+ }
+ body.advance((persisted - sent_offset) as usize);
+ self.offset = persisted;
+ if body.is_empty() {
+ if total.is_some() {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS did not finalize the last resumable upload chunk",
+ )
+ .set_temporary());
+ }
+ return Ok(None);
+ }
+ continue;
+ }
+
+ if resp.status() == StatusCode::OK || resp.status() == StatusCode::CREATED {
+ if total.is_none() {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS finalized a resumable upload before the last chunk",
+ ));
+ }
+ self.offset = sent_end;
+ return GcsCore::build_metadata_from_object_response(&self.path, resp.into_body())
+ .map(Some);
+ }
+
+ return Err(parse_error(resp));
+ }
+ }
+
+ async fn write_once(&self, body: Buffer) -> Result<Metadata> {
+ let req = self.core.gcs_insert_object_request(
+ &self.path,
+ Some(body.len() as u64),
+ &self.op,
+ body,
+ )?;
+ let req = self.core.sign(&self.ctx, req).await?;
+ let resp = self.core.send(&self.ctx, req).await?;
+ match resp.status() {
+ StatusCode::OK | StatusCode::CREATED => {
+ GcsCore::build_metadata_from_object_response(&self.path, resp.into_body())
+ }
+ _ => Err(parse_error(resp)),
+ }
+ }
+}
+
+impl oio::Write for GcsConditionalWriter {
+ async fn write(&mut self, body: Buffer) -> Result<()> {
+ self.pending.push(body);
+ while self.pending.len() > RESUMABLE_CHUNK_SIZE {
+ let mut buffered = self.pending.take().collect();
+ let chunk = buffered.split_to(RESUMABLE_CHUNK_SIZE);
+ self.pending.push(buffered);
+ self.upload(chunk, None).await?;
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let body = self.pending.take().collect();
+ if self.session_uri.is_none() {
+ return self.write_once(body).await;
+ }
+
+ let total = self.offset + body.len() as u64;
+ self.upload(body, Some(total))
+ .await?
+ .ok_or_else(|| Error::new(ErrorKind::Unexpected, "GCS upload returned no metadata"))
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.pending.clear();
+ let Some(session_uri) = self.session_uri.take() else {
+ return Ok(());
+ };
+ let resp = self
+ .core
+ .gcs_cancel_resumable_upload(&self.ctx, &session_uri)
+ .await?;
+ if resp.status().is_success()
+ || resp.status().as_u16() == 499
+ || resp.status() == StatusCode::NOT_FOUND
+ || resp.status() == StatusCode::GONE
+ {
+ Ok(())
+ } else {
+ Err(parse_error(resp))
+ }
+ }
+}
diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs
index f9b369ca7..ec4a5115a 100644
--- a/core/services/s3/src/copier.rs
+++ b/core/services/s3/src/copier.rs
@@ -125,7 +125,7 @@ impl oio::MultipartCopy for S3Copier {
// S3 may return 200 OK with an <Error> body for CopyObject.
if result.etag.is_empty() {
- return Err(from_s3_error(
+ let err = from_s3_error(
S3Error {
code: result.code,
message: result.message,
@@ -133,7 +133,15 @@ impl oio::MultipartCopy for S3Copier {
request_id: result.request_id,
},
parts,
- ));
+ );
+ return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ };
}
let mut meta = Metadata::new(EntryMode::from_path(&self.to));
@@ -147,7 +155,17 @@ impl oio::MultipartCopy for S3Copier {
Ok(meta)
}
- _ => Err(parse_error(resp)),
+ _ => {
+ let err = parse_error(resp);
+ if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
@@ -257,7 +275,7 @@ impl oio::MultipartCopy for S3Copier {
quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
// S3 may return 200 OK with an <Error> body for CompleteMultipartUpload.
if ret.etag.is_empty() {
- return Err(from_s3_error(
+ let err = from_s3_error(
S3Error {
code: ret.code,
message: ret.message,
@@ -265,7 +283,15 @@ impl oio::MultipartCopy for S3Copier {
request_id: ret.request_id,
},
parts,
- ));
+ );
+ return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ };
}
let mut meta = Metadata::new(EntryMode::from_path(&self.to));
@@ -276,7 +302,17 @@ impl oio::MultipartCopy for S3Copier {
Ok(meta)
}
- _ => Err(parse_error(resp)),
+ _ => {
+ let err = parse_error(resp);
+ if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
diff --git a/core/services/s3/src/deleter.rs b/core/services/s3/src/deleter.rs
index 45fb15ae5..abe196898 100644
--- a/core/services/s3/src/deleter.rs
+++ b/core/services/s3/src/deleter.rs
@@ -54,6 +54,10 @@ impl oio::BatchDelete for S3Deleter {
// Allow 404 when deleting a non-existing object
// This is not a standard behavior, only some s3 alike service like GCS XML API do this.
// ref: <https://cloud.google.com/storage/docs/xml-api/delete-object>
+ StatusCode::NOT_FOUND if args.if_match().is_some() => Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ )),
StatusCode::NOT_FOUND => Ok(()),
_ => Err(parse_error(resp)),
}
@@ -88,9 +92,12 @@ impl oio::BatchDelete for S3Deleter {
.position(|e| e.key == abs_path && e.version_id.as_deref() == op.version())
{
let error = errors.swap_remove(idx);
- batched_result
- .failed
- .push((path, op, parse_delete_objects_result_error(error)));
+ let conditional = op.if_match().is_some();
+ batched_result.failed.push((
+ path,
+ op,
+ parse_delete_objects_result_error(error, conditional),
+ ));
} else {
batched_result.succeeded.push((path, op));
}
@@ -100,9 +107,14 @@ impl oio::BatchDelete for S3Deleter {
}
}
-fn parse_delete_objects_result_error(err: DeleteObjectsResultError) -> Error {
+fn parse_delete_objects_result_error(err: DeleteObjectsResultError, conditional: bool) -> Error {
let (kind, retryable) =
parse_s3_error_code(err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
+ let kind = if conditional && kind == ErrorKind::NotFound {
+ ErrorKind::ConditionNotMatch
+ } else {
+ kind
+ };
let mut err: Error = Error::new(kind, format!("{err:?}"));
if retryable {
err = err.set_temporary();
diff --git a/core/services/s3/src/writer.rs b/core/services/s3/src/writer.rs
index fb8207654..82f91b0b7 100644
--- a/core/services/s3/src/writer.rs
+++ b/core/services/s3/src/writer.rs
@@ -83,7 +83,17 @@ impl oio::MultipartWrite for S3Writer {
match status {
StatusCode::CREATED | StatusCode::OK => Ok(meta),
- _ => Err(parse_error(resp)),
+ _ => {
+ let err = parse_error(resp);
+ if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
@@ -204,7 +214,7 @@ impl oio::MultipartWrite for S3Writer {
let ret: CompleteMultipartUploadResult =
quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
if !ret.code.is_empty() {
- return Err(from_s3_error(
+ let err = from_s3_error(
S3Error {
code: ret.code,
message: ret.message,
@@ -212,13 +222,31 @@ impl oio::MultipartWrite for S3Writer {
request_id: ret.request_id,
},
parts,
- ));
+ );
+ return if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ };
}
meta.set_etag(&ret.etag);
Ok(meta)
}
- _ => Err(parse_error(resp)),
+ _ => {
+ let err = parse_error(resp);
+ if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
diff --git a/core/tests/behavior/async_copy.rs b/core/tests/behavior/async_copy.rs
index 490ad3f0a..47b2db282 100644
--- a/core/tests/behavior/async_copy.rs
+++ b/core/tests/behavior/async_copy.rs
@@ -65,6 +65,28 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
))
}
+ if cap.read && cap.write && cap.copy && cap.copy_with_if_none_match {
+ tests.extend(async_trials!(op, test_copy_with_if_none_match))
+ }
+
+ if cap.read
+ && cap.write
+ && cap.copy
+ && cap.copy_with_if_version_match
+ && cap.copy_with_if_version_not_match
+ {
+ tests.extend(async_trials!(op, test_copy_with_version_conditions))
+ }
+
+ if cap.read
+ && cap.write
+ && cap.stat
+ && cap.copy
+ && (cap.copy_with_if_match || cap.copy_with_if_version_match)
+ {
+ tests.extend(async_trials!(op, test_copy_if_not_changed))
+ }
+
if cap.read && cap.write && cap.stat && cap.copy && cap.copy_with_source_version {
tests.extend(async_trials!(
op,
@@ -437,11 +459,147 @@ pub async fn test_copy_with_if_match_mismatch(op: Operator) -> Result<()> {
sha256_digest(&target_content),
);
+ let missing_path = uuid::Uuid::new_v4().to_string();
+ let err = op
+ .copy_with(&source_path, &missing_path)
+ .if_match("\"00000000000000000000000000000000\"")
+ .await
+ .expect_err("missing destination must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
op.delete(&source_path).await.expect("delete must succeed");
op.delete(&target_path).await.expect("delete must succeed");
Ok(())
}
+/// Copy with `If-None-Match` should reject equality and accept inequality or absence.
+pub async fn test_copy_with_if_none_match(op: Operator) -> Result<()> {
+ let source_path = TEST_FIXTURE.new_file_path();
+ let target_path = TEST_FIXTURE.new_file_path();
+ let missing_path = TEST_FIXTURE.new_file_path();
+ let (source_content, _) = gen_bytes(op.info().capability());
+ let (target_content, _) = gen_bytes(op.info().capability());
+ assert_ne!(source_content, target_content);
+
+ op.write(&source_path, source_content.clone()).await?;
+ op.write(&target_path, target_content).await?;
+ let etag = op
+ .stat(&target_path)
+ .await?
+ .etag()
+ .expect("etag must exist")
+ .to_string();
+
+ let err = op
+ .copy_with(&source_path, &target_path)
+ .if_none_match(&etag)
+ .await
+ .expect_err("equal ETag non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.copy_with(&source_path, &target_path)
+ .if_none_match("\"different-etag\"")
+ .await?;
+ op.copy_with(&source_path, &missing_path)
+ .if_none_match(&etag)
+ .await?;
+
+ assert_eq!(op.read(&target_path).await?.to_bytes(), source_content);
+ Ok(())
+}
+
+/// Version preconditions should compare against the current destination version.
+pub async fn test_copy_with_version_conditions(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ let source_path = TEST_FIXTURE.new_file_path();
+ let target_path = TEST_FIXTURE.new_file_path();
+ let missing_path = TEST_FIXTURE.new_file_path();
+ let (source_content, _) = gen_bytes(cap);
+ let (first_target, _) = gen_bytes(cap);
+ let (second_target, _) = gen_bytes(cap);
+ assert_ne!(source_content, first_target);
+ assert_ne!(source_content, second_target);
+ assert_ne!(first_target, second_target);
+
+ op.write(&source_path, source_content).await?;
+ op.write(&target_path, first_target).await?;
+ let stale = op
+ .stat(&target_path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ op.write(&target_path, second_target).await?;
+ let current = op
+ .stat(&target_path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ let err = op
+ .copy_with(&source_path, &target_path)
+ .if_version_match(&stale)
+ .await
+ .expect_err("stale destination version must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ let err = op
+ .copy_with(&source_path, &target_path)
+ .if_version_not_match(¤t)
+ .await
+ .expect_err("equal destination version must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.copy_with(&source_path, &target_path)
+ .if_version_match(¤t)
+ .await?;
+ op.copy_with(&source_path, &target_path)
+ .if_version_not_match(¤t)
+ .await?;
+
+ for result in [
+ op.copy_with(&source_path, &missing_path)
+ .if_version_match(¤t)
+ .await,
+ op.copy_with(&source_path, &missing_path)
+ .if_version_not_match(¤t)
+ .await,
+ ] {
+ assert_eq!(
+ result.expect_err("missing destination must fail").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ }
+
+ Ok(())
+}
+
+/// `if_not_changed` should guard the destination against a stale observation.
+pub async fn test_copy_if_not_changed(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ let source_path = TEST_FIXTURE.new_file_path();
+ let target_path = TEST_FIXTURE.new_file_path();
+ let (source_content, _) = gen_bytes(cap);
+ let (target_content, _) = gen_bytes(cap);
+ assert_ne!(source_content, target_content);
+
+ op.write(&source_path, source_content).await?;
+ op.write(&target_path, target_content).await?;
+ let expected = op.stat(&target_path).await?;
+
+ op.copy_with(&source_path, &target_path)
+ .if_not_changed(&expected)
+ .await?;
+ let err = op
+ .copy_with(&source_path, &target_path)
+ .if_not_changed(&expected)
+ .await
+ .expect_err("stale destination metadata must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ Ok(())
+}
+
/// Copy with source_version should copy a specific source version to a new file.
pub async fn test_copy_with_source_version_to_new_file(op: Operator) -> Result<()> {
if !op.info().capability().copy_with_source_version {
diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs
index 9effe5d8f..651ac7fcb 100644
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@ -52,7 +52,21 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
tests.extend(async_trials!(
op,
test_delete_with_if_match_match,
- test_delete_with_if_match_mismatch
+ test_delete_with_if_match_mismatch,
+ test_delete_with_if_match_missing
+ ));
+ }
+ if cap.delete_with_if_none_match {
+ tests.extend(async_trials!(op, test_delete_with_if_none_match));
+ }
+ if cap.delete_with_if_version_match && cap.delete_with_if_version_not_match {
+ tests.extend(async_trials!(op, test_delete_with_version_conditions));
+ }
+ if cap.delete_with_if_match || cap.delete_with_if_version_match {
+ tests.extend(async_trials!(
+ op,
+ test_delete_if_not_changed,
+ test_deleter_if_not_changed
));
}
}
@@ -462,3 +476,142 @@ pub async fn test_delete_with_if_match_mismatch(op: Operator) -> Result<()> {
Ok(())
}
+
+/// A matching delete condition requires a live target.
+pub async fn test_delete_with_if_match_missing(op: Operator) -> Result<()> {
+ let path = TEST_FIXTURE.new_file_path();
+ let err = op
+ .delete_with(&path)
+ .if_match("\"missing-etag\"")
+ .await
+ .expect_err("missing target must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ Ok(())
+}
+
+/// Delete with `If-None-Match` should reject equality and accept inequality or absence.
+pub async fn test_delete_with_if_none_match(op: Operator) -> Result<()> {
+ let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+ op.write(&path, content).await?;
+ let etag = op
+ .stat(&path)
+ .await?
+ .etag()
+ .expect("etag must exist")
+ .to_string();
+
+ let err = op
+ .delete_with(&path)
+ .if_none_match(&etag)
+ .await
+ .expect_err("equal ETag non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert!(op.exists(&path).await?);
+
+ op.delete_with(&path)
+ .if_none_match("\"different-etag\"")
+ .await?;
+ op.delete_with(&path).if_none_match(&etag).await?;
+
+ Ok(())
+}
+
+/// Version preconditions should compare against the current live target version.
+pub async fn test_delete_with_version_conditions(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ let path = TEST_FIXTURE.new_file_path();
+ let (first, _) = gen_bytes(cap);
+ let (second, _) = gen_bytes(cap);
+ assert_ne!(first, second);
+
+ op.write(&path, first).await?;
+ let stale = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ op.write(&path, second.clone()).await?;
+ let current = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ let err = op
+ .delete_with(&path)
+ .if_version_match(&stale)
+ .await
+ .expect_err("stale version match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ let err = op
+ .delete_with(&path)
+ .if_version_not_match(¤t)
+ .await
+ .expect_err("equal version non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.delete_with(&path).if_version_not_match(&stale).await?;
+
+ for result in [
+ op.delete_with(&path).if_version_match(¤t).await,
+ op.delete_with(&path).if_version_not_match(¤t).await,
+ ] {
+ assert_eq!(
+ result.expect_err("missing target must fail").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ }
+
+ op.write(&path, second).await?;
+ let current = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ op.delete_with(&path).if_version_match(¤t).await?;
+
+ Ok(())
+}
+
+/// `if_not_changed` should reject stale metadata and accept the current state.
+pub async fn test_delete_if_not_changed(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ let path = TEST_FIXTURE.new_file_path();
+ let (first, _) = gen_bytes(cap);
+ let (second, _) = gen_bytes(cap);
+ assert_ne!(first, second);
+
+ op.write(&path, first).await?;
+ let stale = op.stat(&path).await?;
+ op.write(&path, second).await?;
+
+ let err = op
+ .delete_with(&path)
+ .if_not_changed(&stale)
+ .await
+ .expect_err("stale metadata must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ let current = op.stat(&path).await?;
+ op.delete_with(&path).if_not_changed(¤t).await?;
+ Ok(())
+}
+
+/// `Deleter` should lower and preserve `if_not_changed` for queued deletes.
+pub async fn test_deleter_if_not_changed(op: Operator) -> Result<()> {
+ let (path, content, _) = TEST_FIXTURE.new_file(op.clone());
+ op.write(&path, content).await?;
+ let expected = op.stat(&path).await?;
+
+ let mut deleter = op.deleter().await?;
+ deleter
+ .delete(DeleteInput::new(path.clone()).with_if_not_changed(&expected))
+ .await?;
+ deleter.close().await?;
+ assert!(!op.exists(&path).await?);
+
+ Ok(())
+}
diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs
index feee7c730..2335e2cc2 100644
--- a/core/tests/behavior/async_read.rs
+++ b/core/tests/behavior/async_read.rs
@@ -46,6 +46,7 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
test_read_not_exist,
test_read_with_if_match,
test_read_with_if_none_match,
+ test_read_with_version_conditions,
test_read_with_if_modified_since,
test_read_with_if_unmodified_since,
test_read_with_dir_path,
@@ -618,6 +619,65 @@ pub async fn test_read_with_if_none_match(op: Operator) -> anyhow::Result<()> {
Ok(())
}
+/// Version preconditions should compare against the current live object version.
+pub async fn test_read_with_version_conditions(op: Operator) -> anyhow::Result<()> {
+ let cap = op.info().capability();
+ if !cap.read_with_if_version_match || !cap.read_with_if_version_not_match {
+ return Ok(());
+ }
+
+ let path = TEST_FIXTURE.new_file_path();
+ let (first, _) = gen_bytes(cap);
+ let (second, _) = gen_bytes(cap);
+ assert_ne!(first, second);
+
+ op.write(&path, first).await?;
+ let stale = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ op.write(&path, second.clone()).await?;
+ let current = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ let reader = op.reader_with(&path).if_version_match(¤t).await?;
+ assert_eq!(reader.read(..).await?.to_bytes(), second);
+
+ let err = op
+ .read_with(&path)
+ .if_version_match(&stale)
+ .await
+ .expect_err("stale version match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.read_with(&path).if_version_not_match(&stale).await?;
+ let err = op
+ .read_with(&path)
+ .if_version_not_match(¤t)
+ .await
+ .expect_err("equal version non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ let missing = TEST_FIXTURE.new_file_path();
+ for result in [
+ op.read_with(&missing).if_version_match(¤t).await,
+ op.read_with(&missing).if_version_not_match(¤t).await,
+ ] {
+ assert_eq!(
+ result.expect_err("missing target must fail").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ }
+
+ Ok(())
+}
+
/// Read with dir path should return an error.
pub async fn test_read_with_dir_path(op: Operator) -> anyhow::Result<()> {
if !op.info().capability().create_dir {
diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs
index e2a595485..146c933a6 100644
--- a/core/tests/behavior/async_stat.rs
+++ b/core/tests/behavior/async_stat.rs
@@ -39,6 +39,7 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
test_stat_not_exist,
test_stat_with_if_match,
test_stat_with_if_none_match,
+ test_stat_with_version_conditions,
test_stat_with_if_modified_since,
test_stat_with_if_unmodified_since,
test_stat_with_override_cache_control,
@@ -236,6 +237,63 @@ pub async fn test_stat_with_if_none_match(op: Operator) -> Result<()> {
Ok(())
}
+/// Version preconditions should compare against the current live object version.
+pub async fn test_stat_with_version_conditions(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ if !cap.stat_with_if_version_match || !cap.stat_with_if_version_not_match {
+ return Ok(());
+ }
+
+ let path = TEST_FIXTURE.new_file_path();
+ let (first, _) = gen_bytes(cap);
+ let (second, _) = gen_bytes(cap);
+ assert_ne!(first, second);
+
+ op.write(&path, first).await?;
+ let stale = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ op.write(&path, second).await?;
+ let current = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ op.stat_with(&path).if_version_match(¤t).await?;
+ let err = op
+ .stat_with(&path)
+ .if_version_match(&stale)
+ .await
+ .expect_err("stale version match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.stat_with(&path).if_version_not_match(&stale).await?;
+ let err = op
+ .stat_with(&path)
+ .if_version_not_match(¤t)
+ .await
+ .expect_err("equal version non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ let missing = TEST_FIXTURE.new_file_path();
+ for result in [
+ op.stat_with(&missing).if_version_match(¤t).await,
+ op.stat_with(&missing).if_version_not_match(¤t).await,
+ ] {
+ assert_eq!(
+ result.expect_err("missing target must fail").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ }
+
+ Ok(())
+}
+
/// Stat file with if_modified_since should succeed, otherwise get a ConditionNotMatch error.
pub async fn test_stat_with_if_modified_since(op: Operator) -> Result<()> {
if !op.info().capability().stat_with_if_modified_since {
diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs
index 5a17dc69c..60df2813c 100644
--- a/core/tests/behavior/async_write.rs
+++ b/core/tests/behavior/async_write.rs
@@ -45,6 +45,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
test_write_with_if_none_match,
test_write_with_if_not_exists,
test_write_with_if_match,
+ test_write_with_version_conditions,
+ test_write_if_not_changed,
test_write_with_user_metadata,
test_write_returns_metadata,
test_writer_write,
@@ -60,7 +62,8 @@ pub fn tests(op: &Operator, tests: &mut Vec<Trial>) {
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
+ test_writer_write_with_if_match,
+ test_writer_write_with_version_conditions
))
}
@@ -813,6 +816,119 @@ pub async fn test_write_with_if_match(op: Operator) -> Result<()> {
Ok(())
}
+/// Version preconditions should compare against the current live object version.
+pub async fn test_write_with_version_conditions(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ if !cap.write_with_if_version_match || !cap.write_with_if_version_not_match {
+ return Ok(());
+ }
+
+ let path = TEST_FIXTURE.new_file_path();
+ let (initial, _) = gen_bytes(cap);
+ let (replacement, _) = gen_bytes(cap);
+ assert_ne!(initial, replacement);
+
+ op.write(&path, initial).await?;
+ let first_version = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ op.write_with(&path, replacement.clone())
+ .if_version_match(&first_version)
+ .await?;
+
+ let current_version = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ let err = op
+ .write_with(&path, replacement.clone())
+ .if_version_match(&first_version)
+ .await
+ .expect_err("stale version match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ let err = op
+ .write_with(&path, replacement.clone())
+ .if_version_not_match(¤t_version)
+ .await
+ .expect_err("equal version non-match must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ op.write_with(&path, replacement)
+ .if_version_not_match(&first_version)
+ .await?;
+
+ let missing = TEST_FIXTURE.new_file_path();
+ for result in [
+ op.write_with(&missing, Vec::<u8>::new())
+ .if_version_match(¤t_version)
+ .await,
+ op.write_with(&missing, Vec::<u8>::new())
+ .if_version_not_match(¤t_version)
+ .await,
+ ] {
+ assert_eq!(
+ result.expect_err("missing target must fail").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ }
+
+ Ok(())
+}
+
+/// `if_not_changed` should accept the observed state once and reject it after replacement.
+pub async fn test_write_if_not_changed(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ if !cap.write_with_if_version_match && !cap.write_with_if_match {
+ return Ok(());
+ }
+
+ let path = TEST_FIXTURE.new_file_path();
+ let (initial, _) = gen_bytes(cap);
+ let (replacement, _) = gen_bytes(cap);
+ assert_ne!(initial, replacement);
+
+ op.write(&path, initial).await?;
+ let expected = op.stat(&path).await?;
+
+ let mut conflicting = options::WriteOptions {
+ if_not_changed: Some(expected.clone()),
+ ..Default::default()
+ };
+ let mut matching = conflicting.clone();
+ if cap.write_with_if_version_match {
+ let version = expected.version().expect("version must exist");
+ conflicting.if_version_match = Some(format!("different-{version}"));
+ matching.if_version_match = Some(version.to_string());
+ } else {
+ let etag = expected.etag().expect("etag must exist");
+ conflicting.if_match = Some(format!("different-{etag}"));
+ matching.if_match = Some(etag.to_string());
+ }
+
+ let err = op
+ .write_options(&path, replacement.clone(), conflicting)
+ .await
+ .expect_err("conflicting explicit condition must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ op.write_options(&path, replacement.clone(), matching)
+ .await?;
+ let err = op
+ .write_with(&path, replacement)
+ .if_not_changed(&expected)
+ .await
+ .expect_err("stale metadata must fail");
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+
+ Ok(())
+}
+
/// 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<()> {
@@ -821,14 +937,6 @@ pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> {
return Ok(());
}
- // GCS XML API multipart uploads do not support preconditions, so the multipart
- // writer path cannot honor if_not_exists. Tracked in
- // https://github.com/apache/opendal/issues/8040
- #[cfg(feature = "services-gcs")]
- if op.info().scheme() == services::GCS_SCHEME {
- return Ok(());
- }
-
let path = TEST_FIXTURE.new_file_path();
let content = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1));
@@ -929,6 +1037,50 @@ pub async fn test_writer_write_with_if_match(op: Operator) -> Result<()> {
Ok(())
}
+/// Chunked writers should preserve version preconditions through final commit.
+pub async fn test_writer_write_with_version_conditions(op: Operator) -> Result<()> {
+ let cap = op.info().capability();
+ if !cap.write_can_multi
+ || !cap.write_with_if_version_match
+ || !cap.write_with_if_version_not_match
+ {
+ return Ok(());
+ }
+
+ let path = TEST_FIXTURE.new_file_path();
+ let body = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1));
+ op.write(&path, body.clone()).await?;
+ let expected = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+
+ let mut writer = op.writer_with(&path).if_version_match(&expected).await?;
+ writer.write(body.clone()).await?;
+ writer.write(body.clone()).await?;
+ writer.close().await?;
+
+ let current = op
+ .stat(&path)
+ .await?
+ .version()
+ .expect("version must exist")
+ .to_string();
+ let result: opendal::Result<()> = async {
+ let mut writer = op.writer_with(&path).if_version_not_match(¤t).await?;
+ writer.write(body.clone()).await?;
+ writer.write(body).await?;
+ writer.close().await?;
+ Ok(())
+ }
+ .await;
+ assert_eq!(result.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
(opendal) 02/02: Merge remote-tracking branch 'origin/main' into xuanwo/write-if-not-changed
Posted by xu...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch xuanwo/write-if-not-changed
in repository https://gitbox.apache.org/repos/asf/opendal.git
commit c23bf44b827f13d9c70c5d3391ab05fa1e637d53
Merge: baabd99b6 c27eee252
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Wed Aug 26 17:52:49 2026 +0800
Merge remote-tracking branch 'origin/main' into xuanwo/write-if-not-changed
# Conflicts:
# core/core/src/docs/rfcs/8147_write_if_not_changed.md
# core/core/src/types/delete/deleter.rs
# core/core/src/types/delete/input.rs
# core/services/azblob/src/backend.rs
# core/services/azblob/src/core.rs
# core/services/azblob/src/deleter.rs
# core/services/gcs/src/backend.rs
# core/services/gcs/src/deleter.rs
# core/services/gcs/src/reader.rs
# core/services/s3/src/copier.rs
# core/services/s3/src/deleter.rs
# core/services/s3/src/writer.rs
# core/tests/behavior/async_delete.rs
.github/services/s3/0_minio_s3/action.yml | 2 +-
.github/services/s3/aws_s3/action.yml | 2 +-
.../s3/aws_s3_with_list_objects_v1/action.yml | 2 +-
.github/services/s3/aws_s3_with_sse_c/action.yml | 2 +-
.../s3/aws_s3_with_virtual_host/action.yml | 2 +-
.../services/s3/ceph_rados_s3/disable_action.yml | 2 +-
.../services/s3/minio_s3_with_anonymous/action.yml | 2 +-
.../s3/minio_s3_with_list_objects_v1/action.yml | 2 +-
.../s3/minio_s3_with_versioning/action.yml | 2 +-
.github/services/s3/r2/disabled_action.yml | 2 +-
.github/services/s3/seaweedfs_s3/action.yml | 2 +-
.github/workflows/ci_odev.yml | 4 +
AGENTS.md | 31 +-
bindings/c/include/opendal.h | 6 +-
bindings/c/src/error.rs | 5 +-
bindings/dotnet/OpenDAL.Tests/OperatorInfoTest.cs | 16 +
bindings/dotnet/OpenDAL/Capability.cs | 163 +++--
.../Interop/NativeObject/OpenDALCapability.cs | 16 +-
bindings/dotnet/OpenDAL/OpenDALError.cs | 7 +-
bindings/dotnet/OpenDAL/OpenDALException.cs | 4 +-
.../ServiceConfig/AliyunDriveServiceConfig.cs | 10 +-
.../OpenDAL/ServiceConfig/AlluxioServiceConfig.cs | 6 +-
.../OpenDAL/ServiceConfig/AzblobServiceConfig.cs | 15 +-
.../OpenDAL/ServiceConfig/AzdlsServiceConfig.cs | 22 +-
.../OpenDAL/ServiceConfig/AzfileServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/B2ServiceConfig.cs | 6 +-
.../OpenDAL/ServiceConfig/CacacheServiceConfig.cs | 2 +-
.../ServiceConfig/CloudflareKvServiceConfig.cs | 6 +-
.../OpenDAL/ServiceConfig/CompfsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/CosServiceConfig.cs | 13 +-
.../OpenDAL/ServiceConfig/D1ServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/DashmapServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/DbfsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/DropboxServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/EtcdServiceConfig.cs | 4 +-
.../OpenDAL/ServiceConfig/FsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/GcsServiceConfig.cs | 24 +-
.../OpenDAL/ServiceConfig/GdriveServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/GhacServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/GithubServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/GoosefsServiceConfig.cs | 97 +++
.../OpenDAL/ServiceConfig/GridfsServiceConfig.cs | 2 +-
.../ServiceConfig/HdfsNativeServiceConfig.cs | 9 +-
.../OpenDAL/ServiceConfig/HttpServiceConfig.cs | 2 +-
.../ServiceConfig/HuggingfaceServiceConfig.cs | 22 +-
.../OpenDAL/ServiceConfig/IpfsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/IpmfsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/KoofrServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/LakefsServiceConfig.cs | 2 +-
.../ServiceConfig/MemcachedServiceConfig.cs | 4 +-
.../OpenDAL/ServiceConfig/MemoryServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/MiniMokaServiceConfig.cs | 8 +-
.../OpenDAL/ServiceConfig/MokaServiceConfig.cs | 8 +-
.../OpenDAL/ServiceConfig/MongodbServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/MonoiofsServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/MysqlServiceConfig.cs | 4 +-
.../OpenDAL/ServiceConfig/ObsServiceConfig.cs | 5 +-
.../OpenDAL/ServiceConfig/OnedriveServiceConfig.cs | 7 +-
.../OpenDAL/ServiceConfig/OpfsServiceConfig.cs | 11 +-
.../OpenDAL/ServiceConfig/OssServiceConfig.cs | 49 +-
.../OpenDAL/ServiceConfig/PcloudServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/PersyServiceConfig.cs | 2 +-
.../ServiceConfig/PostgresqlServiceConfig.cs | 6 +-
.../OpenDAL/ServiceConfig/RedbServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/RedisServiceConfig.cs | 4 +-
.../OpenDAL/ServiceConfig/S3ServiceConfig.cs | 98 ++-
.../OpenDAL/ServiceConfig/SeafileServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/SftpServiceConfig.cs | 5 +-
.../OpenDAL/ServiceConfig/SledServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/SqliteServiceConfig.cs | 8 +-
.../ServiceConfig/SurrealdbServiceConfig.cs | 2 +-
.../OpenDAL/ServiceConfig/SwiftServiceConfig.cs | 18 +-
...GdriveServiceConfig.cs => TikvServiceConfig.cs} | 48 +-
.../OpenDAL/ServiceConfig/TosServiceConfig.cs | 113 ++++
.../OpenDAL/ServiceConfig/UpyunServiceConfig.cs | 2 +-
.../ServiceConfig/VercelArtifactsServiceConfig.cs | 18 +-
.../ServiceConfig/VercelBlobServiceConfig.cs | 6 +-
.../OpenDAL/ServiceConfig/WebdavServiceConfig.cs | 48 +-
.../OpenDAL/ServiceConfig/WebhdfsServiceConfig.cs | 2 +-
.../ServiceConfig/YandexDiskServiceConfig.cs | 6 +-
bindings/dotnet/src/capability.rs | 98 ++-
bindings/dotnet/src/error.rs | 2 +
bindings/go/error.go | 15 +-
bindings/haskell/haskell-src/OpenDAL.hs | 3 +
bindings/haskell/src/result.rs | 2 +
bindings/java/src/error.rs | 1 +
.../java/org/apache/opendal/OpenDALException.java | 1 +
bindings/python/python/opendal/exceptions.pyi | 1 +
bindings/python/scripts/postprocess_stubs.py | 1 +
bindings/python/src/errors.rs | 7 +
bindings/python/src/lib.rs | 7 +-
bindings/python/tests/test_exceptions.py | 3 +
bindings/zig/src/opendal.zig | 6 +
core/core/src/blocking/operator.rs | 12 +
core/core/src/docs/rfcs/7182_restore_api.md | 696 ++++-----------------
core/core/src/docs/rfcs/7945_limited_read.md | 210 +++++++
.../core/src/docs/rfcs/8145_conflict_error_kind.md | 203 ++++++
core/core/src/docs/rfcs/mod.rs | 8 +
core/core/src/layers/capability_override.rs | 9 +
core/core/src/layers/complete.rs | 9 +
core/core/src/layers/correctness_check.rs | 32 +
core/core/src/layers/error_context.rs | 13 +
core/core/src/layers/simulate.rs | 9 +
core/core/src/raw/accessor.rs | 47 ++
core/core/src/raw/operation.rs | 3 +
core/core/src/raw/ops.rs | 45 ++
core/core/src/raw/rps.rs | 11 +
core/core/src/types/capability.rs | 7 +
core/core/src/types/delete/input.rs | 16 +
core/core/src/types/error.rs | 24 +-
core/core/src/types/operator/operator.rs | 88 +++
core/core/src/types/operator/operator_futures.rs | 19 +
core/core/src/types/options.rs | 23 +
core/core/src/types/write/writer.rs | 2 +-
core/layers/async-backtrace/src/lib.rs | 10 +
core/layers/await-tree/src/lib.rs | 12 +
core/layers/capability-check/src/lib.rs | 25 +
core/layers/chaos/src/lib.rs | 9 +
core/layers/concurrent-limit/src/lib.rs | 10 +
core/layers/dtrace/src/lib.rs | 9 +
core/layers/fastrace/src/lib.rs | 10 +
core/layers/foyer/src/lib.rs | 40 +-
core/layers/hotpath/src/lib.rs | 10 +
core/layers/immutable-index/src/lib.rs | 9 +
core/layers/logging/src/lib.rs | 12 +
core/layers/mime-guess/src/lib.rs | 9 +
core/layers/observe-metrics-common/src/lib.rs | 35 ++
core/layers/oteltrace/src/lib.rs | 14 +
core/layers/retry/src/lib.rs | 23 +
core/layers/route/src/lib.rs | 12 +
core/layers/tail-cut/src/lib.rs | 9 +
core/layers/throttle/src/lib.rs | 9 +
core/layers/timeout/src/lib.rs | 10 +
core/layers/tracing/src/lib.rs | 10 +
core/services/azblob/src/backend.rs | 6 +-
core/services/azblob/src/copier.rs | 23 +-
core/services/azblob/src/core.rs | 419 +++++++++----
core/services/azblob/src/deleter.rs | 35 +-
core/services/azblob/src/lister.rs | 6 +-
core/services/azblob/src/reader.rs | 8 +-
core/services/azblob/src/writer.rs | 41 +-
core/services/azdls/src/backend.rs | 29 +-
core/services/azdls/src/config.rs | 5 +-
core/services/azdls/src/core.rs | 18 +-
core/services/azdls/src/deleter.rs | 6 +-
core/services/gcs/src/backend.rs | 15 +-
core/services/gcs/src/copier.rs | 11 +-
core/services/gcs/src/core.rs | 223 ++++---
core/services/gcs/src/deleter.rs | 31 +-
core/services/gcs/src/lister.rs | 5 +-
core/services/gcs/src/reader.rs | 16 +-
core/services/gcs/src/writer.rs | 57 +-
core/services/ghac/src/writer.rs | 5 +-
core/services/hf/src/core.rs | 605 +++++++++++++++++-
core/services/hf/src/reader.rs | 435 +++++++++++--
core/services/memcached/src/backend.rs | 40 ++
core/services/memcached/src/core.rs | 12 +-
core/services/s3/src/backend.rs | 98 ++-
core/services/s3/src/copier.rs | 75 ++-
core/services/s3/src/core.rs | 312 +++++----
core/services/s3/src/deleter.rs | 12 +-
core/services/s3/src/docs.md | 25 +
core/services/s3/src/lister.rs | 15 +-
core/services/s3/src/reader.rs | 8 +-
core/services/s3/src/writer.rs | 58 +-
core/tests/behavior/async_delete.rs | 89 ++-
core/tests/behavior/async_restore.rs | 162 +++++
core/tests/behavior/main.rs | 2 +
dev/src/generate/capability.rs | 135 ++++
dev/src/generate/dotnet.rs | 268 ++++++++
dev/src/generate/dotnet_capability_rs.j2 | 48 ++
.../src/generate/dotnet_interop_cs.j2 | 30 +-
dev/src/generate/dotnet_public_cs.j2 | 52 ++
.../src/generate/dotnet_service_config.j2 | 35 +-
dev/src/generate/mod.rs | 3 +
dev/src/generate/options.rs | 2 +-
dev/src/release/bump.rs | 425 +++++++++++++
dev/src/release/mod.rs | 3 +
dev/src/release/package.rs | 54 +-
website/community/release/release.md | 6 +
180 files changed, 5409 insertions(+), 1517 deletions(-)
diff --cc core/services/azblob/src/deleter.rs
index d43c70fca,d1aa54c3d..b3c2f7b04
--- a/core/services/azblob/src/deleter.rs
+++ b/core/services/azblob/src/deleter.rs
@@@ -46,13 -46,12 +46,18 @@@ impl oio::BatchDelete for AzblobDelete
let status = resp.status();
match status {
- StatusCode::ACCEPTED | StatusCode::NOT_FOUND => Ok(()),
+ StatusCode::ACCEPTED => Ok(()),
+ StatusCode::NOT_FOUND if args.if_match().is_some() => Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ )),
+ StatusCode::NOT_FOUND => Ok(()),
- _ => Err(parse_error(resp)),
+ _ => Err(parse_error(
+ ErrorContext::new(ServiceOperation("DeleteBlob"))
- .with_if_match(args.if_match().is_some()),
++ .with_if_match(args.if_match().is_some())
++ .with_if_none_match(args.if_none_match().is_some()),
+ resp,
+ )),
}
}
@@@ -92,25 -92,17 +98,28 @@@
let mut batched_result = BatchDeleteResult::default();
- for (i, part) in parts.into_iter().enumerate() {
+ for (part, (path, args)) in parts.into_iter().zip(batch) {
let resp = part.into_response();
- let (path, op) = paths[i].clone();
- if resp.status() == StatusCode::NOT_FOUND && op.if_match().is_some() {
- // deleting not existing objects is ok
- if resp.status() == StatusCode::ACCEPTED || resp.status() == StatusCode::NOT_FOUND {
++ if resp.status() == StatusCode::NOT_FOUND && args.if_match().is_some() {
+ batched_result.failed.push((
+ path,
- op,
++ args,
+ Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ ),
+ ));
+ } else if resp.status() == StatusCode::ACCEPTED
+ || resp.status() == StatusCode::NOT_FOUND
+ {
- batched_result.succeeded.push((path, op));
+ batched_result.succeeded.push((path, args));
} else {
- batched_result.failed.push((path, op, parse_error(resp)));
+ let error_ctx = ErrorContext::new(ServiceOperation("BatchDeleteBlobs"))
- .with_if_match(args.if_match().is_some());
++ .with_if_match(args.if_match().is_some())
++ .with_if_none_match(args.if_none_match().is_some());
+ let err = parse_error(error_ctx, resp);
+ batched_result.failed.push((path, args, err));
}
}
diff --cc core/services/gcs/src/backend.rs
index e1cf9fde2,4469339f7..4f4175365
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@@ -463,6 -452,9 +463,11 @@@ impl Service for GcsBackend
}
async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
+ let error_ctx = ErrorContext::new(ServiceOperation("GetObject"))
+ .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some());
++ .with_if_none_match(args.if_none_match().is_some())
++ .with_if_version_match(args.if_version_match().is_some())
++ .with_if_version_not_match(args.if_version_not_match().is_some());
let resp = self.core.gcs_get_object_metadata(ctx, path, &args).await?;
if !resp.status().is_success() {
diff --cc core/services/gcs/src/copier.rs
index 87a75f74e,477f5c1c0..eb239309f
--- a/core/services/gcs/src/copier.rs
+++ b/core/services/gcs/src/copier.rs
@@@ -89,7 -90,11 +90,15 @@@ impl oio::Copy for GcsCopier
.await?;
if !resp.status().is_success() {
- return Err(parse_error(resp));
+ return Err(parse_error(
+ ErrorContext::new(ServiceOperation("RewriteObject"))
- .with_if_not_exists(self.args.if_not_exists()),
++ .with_if_not_exists(self.args.if_not_exists())
++ .with_if_match(self.args.if_match().is_some())
++ .with_if_none_match(self.args.if_none_match().is_some())
++ .with_if_version_match(self.args.if_version_match().is_some())
++ .with_if_version_not_match(self.args.if_version_not_match().is_some()),
+ resp,
+ ));
}
let result: RewriteResponse = serde_json::from_reader(resp.into_body().reader())
diff --cc core/services/gcs/src/core.rs
index 5d87f9513,c41b1322c..a0796336d
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@@ -1552,115 -1201,135 +1554,160 @@@ mod tests
}
}
- mod error {
- use http::Response;
- use http::StatusCode;
- use serde::Deserialize;
- use serde_json::de;
+ #[derive(Clone, Copy, Debug)]
+ pub struct ErrorContext {
+ service_operation: ServiceOperation,
+ if_match: bool,
+ if_none_match: bool,
+ if_not_exists: bool,
++ if_version_match: bool,
++ if_version_not_match: bool,
+ }
- use opendal_core::raw::*;
- use opendal_core::*;
+ impl ErrorContext {
+ pub const fn new(service_operation: ServiceOperation) -> Self {
+ Self {
+ service_operation,
+ if_match: false,
+ if_none_match: false,
+ if_not_exists: false,
++ if_version_match: false,
++ if_version_not_match: false,
+ }
+ }
- #[derive(Default, Debug, Deserialize)]
- #[serde(default, rename_all = "camelCase")]
- struct GcsErrorResponse {
- error: GcsError,
+ pub const fn with_if_match(mut self, if_match: bool) -> Self {
+ self.if_match = if_match;
+ self
}
- #[derive(Default, Debug, Deserialize)]
- #[serde(default, rename_all = "camelCase")]
- struct GcsError {
- code: usize,
- message: String,
- errors: Vec<GcsErrorDetail>,
+ pub const fn with_if_none_match(mut self, if_none_match: bool) -> Self {
+ self.if_none_match = if_none_match;
+ self
}
- #[derive(Default, Debug, Deserialize)]
- #[serde(default, rename_all = "camelCase")]
- struct GcsErrorDetail {
- domain: String,
- location: String,
- location_type: String,
- message: String,
- reason: String,
+ pub const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.if_not_exists = if_not_exists;
+ self
}
- /// Parse error response into Error.
- pub(crate) fn parse_error(resp: Response<Buffer>) -> Error {
- let (parts, body) = resp.into_parts();
- let bs = body.to_bytes();
++ pub const fn with_if_version_match(mut self, if_version_match: bool) -> Self {
++ self.if_version_match = if_version_match;
++ self
++ }
+
- let (kind, retryable) = match parts.status {
- StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
- StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED => {
- (ErrorKind::ConditionNotMatch, false)
- }
- StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
- StatusCode::INTERNAL_SERVER_ERROR
- | StatusCode::BAD_GATEWAY
- | StatusCode::SERVICE_UNAVAILABLE
- | StatusCode::GATEWAY_TIMEOUT => (ErrorKind::Unexpected, true),
- _ => (ErrorKind::Unexpected, false),
- };
++ pub const fn with_if_version_not_match(mut self, if_version_not_match: bool) -> Self {
++ self.if_version_not_match = if_version_not_match;
++ self
++ }
+
- let message = match de::from_slice::<GcsErrorResponse>(&bs) {
- Ok(gcs_err) => format!("{gcs_err:?}"),
- Err(_) => String::from_utf8_lossy(&bs).into_owned(),
- };
+ const fn has_condition(self) -> bool {
- self.if_match || self.if_none_match || self.if_not_exists
++ self.if_match
++ || self.if_none_match
++ || self.if_not_exists
++ || self.if_version_match
++ || self.if_version_not_match
++ }
+
- let mut err = Error::new(kind, message);
++ const fn has_version_condition(self) -> bool {
++ self.if_version_match || self.if_version_not_match
+ }
+ }
- err = with_error_response_context(err, parts);
+ #[derive(Default, Debug, Deserialize)]
+ #[serde(default, rename_all = "camelCase")]
+ struct GcsErrorResponse {
+ error: GcsError,
+ }
- if retryable {
- err = err.set_temporary();
- }
+ #[derive(Default, Debug, Deserialize)]
+ #[serde(default, rename_all = "camelCase")]
+ struct GcsError {
+ code: usize,
+ message: String,
+ errors: Vec<GcsErrorDetail>,
+ }
+
+ #[derive(Default, Debug, Deserialize)]
+ #[serde(default, rename_all = "camelCase")]
+ struct GcsErrorDetail {
+ domain: String,
+ location: String,
+ location_type: String,
+ message: String,
+ reason: String,
+ }
- err
+ /// Parse error response into Error.
+ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
+ let (parts, body) = resp.into_parts();
+ let bs = body.to_bytes();
+
+ let gcs_error = de::from_slice::<GcsErrorResponse>(&bs).ok();
+
+ let (mut kind, mut retryable) = match parts.status {
++ StatusCode::NOT_FOUND if ctx.has_version_condition() => {
++ (ErrorKind::ConditionNotMatch, false)
++ }
+ StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
+ StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
+ StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.has_condition() => {
+ (ErrorKind::ConditionNotMatch, false)
+ }
+ StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
+ StatusCode::INTERNAL_SERVER_ERROR
+ | StatusCode::BAD_GATEWAY
+ | StatusCode::SERVICE_UNAVAILABLE
+ | StatusCode::GATEWAY_TIMEOUT => (ErrorKind::Unexpected, true),
+ _ => (ErrorKind::Unexpected, false),
+ };
+
+ if let Some(gcs_error) = &gcs_error {
+ let has_reason = gcs_error
+ .error
+ .errors
+ .iter()
+ .any(|detail| !detail.reason.is_empty());
+ if gcs_error
+ .error
+ .errors
+ .iter()
+ .any(|detail| detail.reason == "conditionNotMet")
+ {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ } else if gcs_error
+ .error
+ .errors
+ .iter()
+ .any(|detail| detail.reason == "conflict")
+ {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ } else if has_reason
+ && matches!(
+ parts.status,
+ StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+ )
+ {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
}
- #[cfg(test)]
- mod tests {
- use super::*;
-
- #[test]
- fn test_parse_error() {
- let bs = bytes::Bytes::from(
- r#"
- {
- "error": {
- "errors": [
- {
- "domain": "global",
- "reason": "required",
- "message": "Login Required",
- "locationType": "header",
- "location": "Authorization"
- }
- ],
- "code": 401,
- "message": "Login Required"
- }
- }
- "#,
- );
+ let message = match gcs_error {
+ Some(gcs_err) => format!("{gcs_err:?}"),
+ None => String::from_utf8_lossy(&bs).into_owned(),
+ };
- let out: GcsErrorResponse = de::from_slice(&bs).expect("must success");
- println!("{out:?}");
+ let mut err =
+ Error::new(kind, message).with_context("service_operation", ctx.service_operation.0);
- assert_eq!(out.error.code, 401);
- assert_eq!(out.error.message, "Login Required");
- assert_eq!(out.error.errors[0].domain, "global");
- assert_eq!(out.error.errors[0].reason, "required");
- assert_eq!(out.error.errors[0].message, "Login Required");
- assert_eq!(out.error.errors[0].location_type, "header");
- assert_eq!(out.error.errors[0].location, "Authorization");
- }
+ err = with_error_response_context(err, parts);
+
+ if retryable {
+ err = err.set_temporary();
}
- }
- pub(super) use error::*;
+ err
+ }
mod uri {
use percent_encoding::AsciiSet;
diff --cc core/services/gcs/src/deleter.rs
index 44bc99fab,1da5adbe2..c6923c84e
--- a/core/services/gcs/src/deleter.rs
+++ b/core/services/gcs/src/deleter.rs
@@@ -37,22 -37,17 +37,31 @@@ impl GcsDeleter
}
impl oio::BatchDelete for GcsDeleter {
- async fn delete_once(&self, path: String, _: OpDelete) -> Result<()> {
- let resp = self.core.gcs_delete_object(&self.ctx, &path).await?;
+ async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
+ let resp = self.core.gcs_delete_object(&self.ctx, &path, &args).await?;
- // deleting not existing objects is ok
- if resp.status().is_success() || resp.status() == StatusCode::NOT_FOUND {
+ if resp.status().is_success() {
+ Ok(())
+ } else if resp.status() == StatusCode::NOT_FOUND
- && (args.if_version_match().is_some() || args.if_version_not_match().is_some())
++ && (args.if_match().is_some()
++ || args.if_version_match().is_some()
++ || args.if_version_not_match().is_some())
+ {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ ))
+ } else if resp.status() == StatusCode::NOT_FOUND {
Ok(())
} else {
- Err(parse_error(resp))
+ Err(parse_error(
- ErrorContext::new(ServiceOperation("DeleteObject")),
++ ErrorContext::new(ServiceOperation("DeleteObject"))
++ .with_if_match(args.if_match().is_some())
++ .with_if_none_match(args.if_none_match().is_some())
++ .with_if_version_match(args.if_version_match().is_some())
++ .with_if_version_not_match(args.if_version_not_match().is_some()),
+ resp,
+ ))
}
}
@@@ -83,25 -85,20 +95,34 @@@
for (i, part) in parts.into_iter().enumerate() {
let resp = part.into_response();
// TODO: maybe we can take it directly?
- let path = paths[i].clone();
-
- // deleting not existing objects is ok
- if resp.status().is_success() || resp.status() == StatusCode::NOT_FOUND {
- batched_result.succeeded.push((path, OpDelete::default()));
- } else {
+ let (path, op) = batch[i].clone();
+
+ if resp.status().is_success() {
+ batched_result.succeeded.push((path, op));
+ } else if resp.status() == StatusCode::NOT_FOUND
- && (op.if_version_match().is_some() || op.if_version_not_match().is_some())
++ && (op.if_match().is_some()
++ || op.if_version_match().is_some()
++ || op.if_version_not_match().is_some())
+ {
batched_result.failed.push((
path,
- OpDelete::default(),
- parse_error(
- ErrorContext::new(ServiceOperation("BatchDeleteObjects")),
- resp,
+ op,
+ Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
),
));
+ } else if resp.status() == StatusCode::NOT_FOUND {
+ batched_result.succeeded.push((path, op));
+ } else {
- batched_result.failed.push((path, op, parse_error(resp)));
++ let error_ctx = ErrorContext::new(ServiceOperation("BatchDeleteObjects"))
++ .with_if_match(op.if_match().is_some())
++ .with_if_none_match(op.if_none_match().is_some())
++ .with_if_version_match(op.if_version_match().is_some())
++ .with_if_version_not_match(op.if_version_not_match().is_some());
++ batched_result
++ .failed
++ .push((path, op, parse_error(error_ctx, resp)));
}
}
diff --cc core/services/gcs/src/reader.rs
index 5d784e535,1d8f08dd5..f57089d94
--- a/core/services/gcs/src/reader.rs
+++ b/core/services/gcs/src/reader.rs
@@@ -16,7 -16,7 +16,8 @@@
// under the License.
use super::backend::*;
+ use super::core::ErrorContext;
+use super::core::constants::X_GOOG_GENERATION;
use super::core::parse_error;
use http::Response;
use http::StatusCode;
@@@ -54,6 -54,9 +55,11 @@@ impl oio::StreamRead for GcsReader
let backend = &self.backend;
let path = self.path.as_str();
let args = self.args.clone();
+ let error_ctx = ErrorContext::new(ServiceOperation("GetObject"))
+ .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some());
++ .with_if_none_match(args.if_none_match().is_some())
++ .with_if_version_match(args.if_version_match().is_some())
++ .with_if_version_not_match(args.if_version_not_match().is_some());
let resp = backend
.core
.gcs_get_object(&self.ctx, path, range, &args)
diff --cc core/services/gcs/src/writer.rs
index aa95bc62c,9b6f40d88..05a871b01
--- a/core/services/gcs/src/writer.rs
+++ b/core/services/gcs/src/writer.rs
@@@ -19,13 -19,11 +19,14 @@@ use std::sync::Arc
use bytes::Buf;
use http::StatusCode;
+use http::header::LOCATION;
+use http::header::RANGE;
use super::core::CompleteMultipartUploadRequestPart;
+ use super::core::ErrorContext;
use super::core::GcsCore;
use super::core::InitiateMultipartUploadResult;
+use super::core::constants::X_GOOG_GENERATION;
use super::core::parse_error;
use opendal_core::raw::*;
use opendal_core::*;
@@@ -147,16 -153,15 +158,19 @@@ impl oio::MultipartWrite for GcsWriter
.await?;
if !resp.status().is_success() {
- return Err(parse_error(resp));
+ return Err(parse_error(
+ ErrorContext::new(ServiceOperation("CompleteMultipartUpload")),
+ resp,
+ ));
}
- // we don't extract metadata from `CompleteMultipartUploadResult`, since we only need the `ETag` from it.
- // However, the `ETag` differs from the `ETag` obtained through the `stat` operation.
- // refer to: https://cloud.google.com/storage/docs/metadata#etags
- Ok(Metadata::default())
+ let mut metadata = Metadata::new(EntryMode::from_path(&self.path));
+ if let Some(etag) = parse_etag(resp.headers())? {
+ metadata.set_etag(etag);
+ }
+ if let Some(generation) = parse_header_to_str(resp.headers(), X_GOOG_GENERATION)? {
+ metadata.set_version(generation);
+ }
+ Ok(metadata)
}
async fn abort_part(&self, upload_id: &str) -> Result<()> {
@@@ -171,194 -179,3 +188,216 @@@
}
}
}
+
+pub struct GcsConditionalWriter {
+ core: Arc<GcsCore>,
+ ctx: OperationContext,
+ path: String,
+ op: OpWrite,
+ session_uri: Option<String>,
+ offset: u64,
+ pending: oio::QueueBuf,
+}
+
+impl GcsConditionalWriter {
+ pub fn new(core: Arc<GcsCore>, ctx: OperationContext, path: &str, op: OpWrite) -> Self {
+ Self {
+ core,
+ ctx,
+ path: path.to_string(),
+ op,
+ session_uri: None,
+ offset: 0,
+ pending: oio::QueueBuf::new(),
+ }
+ }
+
+ async fn ensure_session(&mut self) -> Result<&str> {
+ if self.session_uri.is_none() {
+ let resp = self
+ .core
+ .gcs_initiate_resumable_upload(&self.ctx, &self.path, &self.op)
+ .await?;
+ if !resp.status().is_success() {
- return Err(parse_error(resp));
++ return Err(parse_error(
++ ErrorContext::new(ServiceOperation("InitiateResumableUpload"))
++ .with_if_not_exists(self.op.if_not_exists())
++ .with_if_match(self.op.if_match().is_some())
++ .with_if_none_match(self.op.if_none_match().is_some())
++ .with_if_version_match(self.op.if_version_match().is_some())
++ .with_if_version_not_match(self.op.if_version_not_match().is_some()),
++ resp,
++ ));
+ }
+ let location = parse_header_to_str(resp.headers(), LOCATION)?
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "GCS resumable upload response is missing Location",
+ )
+ })?
+ .to_string();
+ self.session_uri = Some(location);
+ }
+ Ok(self.session_uri.as_deref().unwrap())
+ }
+
+ fn persisted_offset(resp: &http::Response<Buffer>) -> Result<u64> {
+ let Some(range) = parse_header_to_str(resp.headers(), RANGE)? else {
+ return Ok(0);
+ };
+ let end = range
+ .strip_prefix("bytes=")
+ .and_then(|v| v.rsplit_once('-'))
+ .map(|(_, end)| end)
+ .ok_or_else(|| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid Range header in GCS resumable upload response",
+ )
+ .with_context("range", range)
+ })?;
+ end.parse::<u64>().map(|v| v + 1).map_err(|err| {
+ Error::new(
+ ErrorKind::Unexpected,
+ "invalid Range header in GCS resumable upload response",
+ )
+ .with_context("range", range)
+ .set_source(err)
+ })
+ }
+
+ async fn upload(&mut self, mut body: Buffer, total: Option<u64>) -> Result<Option<Metadata>> {
+ loop {
+ let session_uri = self.ensure_session().await?.to_string();
+ let sent_offset = self.offset;
+ let sent_end = sent_offset + body.len() as u64;
+ let resp = self
+ .core
+ .gcs_upload_resumable_chunk(
+ &self.ctx,
+ &session_uri,
+ sent_offset,
+ body.clone(),
+ total,
+ )
+ .await?;
+
+ if resp.status() == StatusCode::PERMANENT_REDIRECT {
+ let persisted = Self::persisted_offset(&resp)?;
+ if persisted <= sent_offset || persisted > sent_end {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS resumable upload reported invalid persisted range",
+ )
+ .with_context("offset", sent_offset)
+ .with_context("persisted", persisted)
+ .set_temporary());
+ }
+ body.advance((persisted - sent_offset) as usize);
+ self.offset = persisted;
+ if body.is_empty() {
+ if total.is_some() {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS did not finalize the last resumable upload chunk",
+ )
+ .set_temporary());
+ }
+ return Ok(None);
+ }
+ continue;
+ }
+
+ if resp.status() == StatusCode::OK || resp.status() == StatusCode::CREATED {
+ if total.is_none() {
+ return Err(Error::new(
+ ErrorKind::Unexpected,
+ "GCS finalized a resumable upload before the last chunk",
+ ));
+ }
+ self.offset = sent_end;
+ return GcsCore::build_metadata_from_object_response(&self.path, resp.into_body())
+ .map(Some);
+ }
+
- return Err(parse_error(resp));
++ return Err(parse_error(
++ ErrorContext::new(ServiceOperation("UploadResumableChunk")),
++ resp,
++ ));
+ }
+ }
+
+ async fn write_once(&self, body: Buffer) -> Result<Metadata> {
+ let req = self.core.gcs_insert_object_request(
+ &self.path,
+ Some(body.len() as u64),
+ &self.op,
+ body,
+ )?;
+ let req = self.core.sign(&self.ctx, req).await?;
+ let resp = self.core.send(&self.ctx, req).await?;
+ match resp.status() {
+ StatusCode::OK | StatusCode::CREATED => {
+ GcsCore::build_metadata_from_object_response(&self.path, resp.into_body())
+ }
- _ => Err(parse_error(resp)),
++ _ => Err(parse_error(
++ ErrorContext::new(ServiceOperation("InsertObject"))
++ .with_if_not_exists(self.op.if_not_exists())
++ .with_if_match(self.op.if_match().is_some())
++ .with_if_none_match(self.op.if_none_match().is_some())
++ .with_if_version_match(self.op.if_version_match().is_some())
++ .with_if_version_not_match(self.op.if_version_not_match().is_some()),
++ resp,
++ )),
+ }
+ }
+}
+
+impl oio::Write for GcsConditionalWriter {
+ async fn write(&mut self, body: Buffer) -> Result<()> {
+ self.pending.push(body);
+ while self.pending.len() > RESUMABLE_CHUNK_SIZE {
+ let mut buffered = self.pending.take().collect();
+ let chunk = buffered.split_to(RESUMABLE_CHUNK_SIZE);
+ self.pending.push(buffered);
+ self.upload(chunk, None).await?;
+ }
+ Ok(())
+ }
+
+ async fn close(&mut self) -> Result<Metadata> {
+ let body = self.pending.take().collect();
+ if self.session_uri.is_none() {
+ return self.write_once(body).await;
+ }
+
+ let total = self.offset + body.len() as u64;
+ self.upload(body, Some(total))
+ .await?
+ .ok_or_else(|| Error::new(ErrorKind::Unexpected, "GCS upload returned no metadata"))
+ }
+
+ async fn abort(&mut self) -> Result<()> {
+ self.pending.clear();
+ let Some(session_uri) = self.session_uri.take() else {
+ return Ok(());
+ };
+ let resp = self
+ .core
+ .gcs_cancel_resumable_upload(&self.ctx, &session_uri)
+ .await?;
+ if resp.status().is_success()
+ || resp.status().as_u16() == 499
+ || resp.status() == StatusCode::NOT_FOUND
+ || resp.status() == StatusCode::GONE
+ {
+ Ok(())
+ } else {
- Err(parse_error(resp))
++ Err(parse_error(
++ ErrorContext::new(ServiceOperation("CancelResumableUpload")),
++ resp,
++ ))
+ }
+ }
+}
diff --cc core/services/s3/src/copier.rs
index ec4a5115a,0461e4fb4..6724b822b
--- a/core/services/s3/src/copier.rs
+++ b/core/services/s3/src/copier.rs
@@@ -86,6 -85,14 +85,15 @@@ pub struct S3Copier
args: OpCopy,
}
+ impl S3Copier {
+ fn error_context(&self, service_operation: ServiceOperation) -> ErrorContext {
+ ErrorContext::new(service_operation)
+ .with_if_match(self.args.if_match().is_some())
++ .with_if_none_match(self.args.if_none_match().is_some())
+ .with_if_not_exists(self.args.if_not_exists())
+ }
+ }
+
impl oio::MultipartCopy for S3Copier {
async fn source_metadata(&self) -> Result<Metadata> {
let mut args = OpStat::default();
@@@ -125,23 -135,10 +136,18 @@@
// S3 may return 200 OK with an <Error> body for CopyObject.
if result.etag.is_empty() {
- let err = from_s3_error(
- S3Error {
- code: result.code,
- message: result.message,
- resource: String::new(),
- request_id: result.request_id,
- },
- parts,
- return Err(parse_error(
++ let err = parse_error(
+ self.error_context(ServiceOperation("CopyObject")),
+ Response::from_parts(parts, Buffer::from(bs)),
- ));
+ );
+ return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ };
}
let mut meta = Metadata::new(EntryMode::from_path(&self.to));
@@@ -155,17 -152,10 +161,17 @@@
Ok(meta)
}
- _ => Err(parse_error(
- self.error_context(ServiceOperation("CopyObject")),
- resp,
- )),
+ _ => {
- let err = parse_error(resp);
++ let err = parse_error(self.error_context(ServiceOperation("CopyObject")), resp);
+ if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
@@@ -272,26 -264,13 +280,21 @@@
.map(str::to_string);
let ret: CompleteMultipartUploadResult =
- quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
+ quick_xml::de::from_reader(bs.as_ref()).map_err(new_xml_deserialize_error)?;
// S3 may return 200 OK with an <Error> body for CompleteMultipartUpload.
if ret.etag.is_empty() {
- let err = from_s3_error(
- S3Error {
- code: ret.code,
- message: ret.message,
- resource: "".to_string(),
- request_id: ret.request_id,
- },
- parts,
- return Err(parse_error(
++ let err = parse_error(
+ self.error_context(ServiceOperation("CompleteMultipartUpload")),
+ Response::from_parts(parts, Buffer::from(bs)),
- ));
+ );
+ return if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ };
}
let mut meta = Metadata::new(EntryMode::from_path(&self.to));
@@@ -302,17 -281,10 +305,20 @@@
Ok(meta)
}
- _ => Err(parse_error(
- self.error_context(ServiceOperation("CompleteMultipartUpload")),
- resp,
- )),
+ _ => {
- let err = parse_error(resp);
++ let err = parse_error(
++ self.error_context(ServiceOperation("CompleteMultipartUpload")),
++ resp,
++ );
+ if self.args.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "copy precondition requires a live destination",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
diff --cc core/services/s3/src/deleter.rs
index abe196898,b5c44065a..3789603a3
--- a/core/services/s3/src/deleter.rs
+++ b/core/services/s3/src/deleter.rs
@@@ -54,12 -56,8 +56,12 @@@ impl oio::BatchDelete for S3Deleter
// Allow 404 when deleting a non-existing object
// This is not a standard behavior, only some s3 alike service like GCS XML API do this.
// ref: <https://cloud.google.com/storage/docs/xml-api/delete-object>
+ StatusCode::NOT_FOUND if args.if_match().is_some() => Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "delete precondition requires a live target",
+ )),
StatusCode::NOT_FOUND => Ok(()),
- _ => Err(parse_error(resp)),
+ _ => Err(parse_error(error_ctx, resp)),
}
}
@@@ -107,14 -105,12 +112,15 @@@
}
}
-fn parse_delete_objects_result_error(err: DeleteObjectsResultError) -> Error {
- let (kind, retryable) = parse_s3_error_code(
- ErrorContext::new(ServiceOperation("DeleteObjects")),
- err.code.as_str(),
- )
- .unwrap_or((ErrorKind::Unexpected, false));
+fn parse_delete_objects_result_error(err: DeleteObjectsResultError, conditional: bool) -> Error {
++ let error_ctx = ErrorContext::new(ServiceOperation("DeleteObjects")).with_if_match(conditional);
+ let (kind, retryable) =
- parse_s3_error_code(err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
++ parse_s3_error_code(error_ctx, err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
+ let kind = if conditional && kind == ErrorKind::NotFound {
+ ErrorKind::ConditionNotMatch
+ } else {
+ kind
+ };
let mut err: Error = Error::new(kind, format!("{err:?}"));
if retryable {
err = err.set_temporary();
diff --cc core/services/s3/src/writer.rs
index 82f91b0b7,6573e66ea..1cf260d22
--- a/core/services/s3/src/writer.rs
+++ b/core/services/s3/src/writer.rs
@@@ -64,6 -63,12 +63,13 @@@ impl S3Writer
}
Ok(meta)
}
+
+ fn error_context(&self, service_operation: ServiceOperation) -> ErrorContext {
+ ErrorContext::new(service_operation)
+ .with_if_match(self.op.if_match().is_some())
++ .with_if_none_match(self.op.if_none_match().is_some())
+ .with_if_not_exists(self.op.if_not_exists())
+ }
}
impl oio::MultipartWrite for S3Writer {
@@@ -83,17 -88,10 +89,17 @@@
match status {
StatusCode::CREATED | StatusCode::OK => Ok(meta),
- _ => Err(parse_error(
- self.error_context(ServiceOperation("PutObject")),
- resp,
- )),
+ _ => {
- let err = parse_error(resp);
++ let err = parse_error(self.error_context(ServiceOperation("PutObject")), resp);
+ if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
@@@ -210,43 -214,24 +222,42 @@@
// still check if there is any error because S3 might return error for status code 200
// https://docs.aws.amazon.com/AmazonS3/latest/API/API_CompleteMultipartUpload.html#API_CompleteMultipartUpload_Example_4
let (parts, body) = resp.into_parts();
+ let bs = body.to_bytes();
let ret: CompleteMultipartUploadResult =
- quick_xml::de::from_reader(body.reader()).map_err(new_xml_deserialize_error)?;
+ quick_xml::de::from_reader(bs.as_ref()).map_err(new_xml_deserialize_error)?;
if !ret.code.is_empty() {
- let err = from_s3_error(
- S3Error {
- code: ret.code,
- message: ret.message,
- resource: "".to_string(),
- request_id: ret.request_id,
- },
- parts,
- return Err(parse_error(
++ let err = parse_error(
+ self.error_context(ServiceOperation("CompleteMultipartUpload")),
+ Response::from_parts(parts, Buffer::from(bs)),
- ));
+ );
+ return if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ };
}
meta.set_etag(&ret.etag);
Ok(meta)
}
- _ => Err(parse_error(
- self.error_context(ServiceOperation("CompleteMultipartUpload")),
- resp,
- )),
+ _ => {
- let err = parse_error(resp);
++ let err = parse_error(
++ self.error_context(ServiceOperation("CompleteMultipartUpload")),
++ resp,
++ );
+ if self.op.if_match().is_some() && err.kind() == ErrorKind::NotFound {
+ Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "write precondition requires a live target",
+ ))
+ } else {
+ Err(err)
+ }
+ }
}
}
diff --cc core/tests/behavior/async_delete.rs
index 651ac7fcb,e074d0fbc..81c3a9079
--- a/core/tests/behavior/async_delete.rs
+++ b/core/tests/behavior/async_delete.rs
@@@ -53,22 -53,9 +53,23 @@@ pub fn tests(op: &Operator, tests: &mu
op,
test_delete_with_if_match_match,
test_delete_with_if_match_mismatch,
- test_delete_with_if_match_missing
++ test_delete_with_if_match_missing,
+ test_batch_delete_with_if_match
));
}
+ if cap.delete_with_if_none_match {
+ tests.extend(async_trials!(op, test_delete_with_if_none_match));
+ }
+ if cap.delete_with_if_version_match && cap.delete_with_if_version_not_match {
+ tests.extend(async_trials!(op, test_delete_with_version_conditions));
+ }
+ if cap.delete_with_if_match || cap.delete_with_if_version_match {
+ tests.extend(async_trials!(
+ op,
+ test_delete_if_not_changed,
+ test_deleter_if_not_changed
+ ));
+ }
}
}