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:19 UTC
(opendal) 02/02: Merge remote-tracking branch 'origin/main' into xuanwo/write-if-not-changed
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
+ ));
+ }
}
}