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/31 14:53:46 UTC
(opendal) branch main updated: fix: preserve conditional error semantics (#8204)
This is an automated email from the ASF dual-hosted git repository.
Xuanwo pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/opendal.git
The following commit(s) were added to refs/heads/main by this push:
new dc6ea03b5 fix: preserve conditional error semantics (#8204)
dc6ea03b5 is described below
commit dc6ea03b5563a490329b000b7d65206eb1702a35
Author: Xuanwo <gi...@xuanwo.io>
AuthorDate: Mon Aug 31 22:53:41 2026 +0800
fix: preserve conditional error semantics (#8204)
---
bindings/haskell/haskell-src/OpenDAL.hs | 6 ++
bindings/haskell/src/result.rs | 51 +++++++---
core/core/src/raw/ops.rs | 88 +++++++++++++++++
core/services/azblob/src/backend.rs | 3 +-
core/services/azblob/src/copier.rs | 6 +-
core/services/azblob/src/core.rs | 140 ++++++++++++++++-----------
core/services/azblob/src/deleter.rs | 6 +-
core/services/azblob/src/reader.rs | 5 +-
core/services/azblob/src/writer.rs | 10 +-
core/services/azdls/src/core.rs | 156 ++++++++++++++++++++++++------
core/services/azdls/src/deleter.rs | 3 +-
core/services/azdls/src/reader.rs | 3 +-
core/services/azdls/src/writer.rs | 24 +++--
core/services/azfile/src/core.rs | 150 ++++++++++++++++++++++++----
core/services/cloudflare-kv/src/reader.rs | 6 +-
core/services/cos/src/backend.rs | 6 +-
core/services/cos/src/core.rs | 112 +++++++++++++++++++--
core/services/cos/src/reader.rs | 3 +-
core/services/cos/src/writer.rs | 26 ++---
core/services/gcs-grpc/src/backend.rs | 6 +-
core/services/gcs-grpc/src/copier.rs | 10 +-
core/services/gcs-grpc/src/core.rs | 56 ++++++++++-
core/services/gcs-grpc/src/deleter.rs | 7 +-
core/services/gcs-grpc/src/lister.rs | 6 +-
core/services/gcs-grpc/src/reader.rs | 15 ++-
core/services/gcs-grpc/src/writer.rs | 35 +++++--
core/services/gcs/src/backend.rs | 5 +-
core/services/gcs/src/composer.rs | 17 ++--
core/services/gcs/src/copier.rs | 6 +-
core/services/gcs/src/core.rs | 119 +++++++++++++----------
core/services/gcs/src/deleter.rs | 18 ++--
core/services/gcs/src/reader.rs | 5 +-
core/services/gcs/src/writer.rs | 20 +---
core/services/hf/src/core.rs | 7 +-
core/services/onedrive/src/core.rs | 89 ++++++++++++++---
core/services/onedrive/src/reader.rs | 3 +-
core/services/onedrive/src/writer.rs | 6 +-
core/services/oss/src/backend.rs | 3 +-
core/services/oss/src/core.rs | 101 +++++++++++++++++--
core/services/oss/src/reader.rs | 3 +-
core/services/oss/src/writer.rs | 9 +-
core/services/s3/src/backend.rs | 5 +-
core/services/s3/src/copier.rs | 5 +-
core/services/s3/src/core.rs | 81 ++++++++--------
core/services/s3/src/deleter.rs | 5 +-
core/services/s3/src/reader.rs | 5 +-
core/services/s3/src/writer.rs | 10 +-
47 files changed, 1072 insertions(+), 389 deletions(-)
diff --git a/bindings/haskell/haskell-src/OpenDAL.hs b/bindings/haskell/haskell-src/OpenDAL.hs
index 86b1120e8..8d8f377ac 100644
--- a/bindings/haskell/haskell-src/OpenDAL.hs
+++ b/bindings/haskell/haskell-src/OpenDAL.hs
@@ -169,6 +169,10 @@ data ErrorCode
IsSameFile
| -- | The operation conflicts with the current or transitional state of the resource.
Conflict
+ | -- | A condition supplied through the OpenDAL operation evaluated to false.
+ ConditionNotMatch
+ | -- | The requested content range cannot be satisfied.
+ RangeNotSatisfied
deriving (Eq, Show)
-- | Represents an error that can occur when using OpenDAL.
@@ -315,6 +319,8 @@ parseErrorCode 9 = AlreadyExists
parseErrorCode 10 = RateLimited
parseErrorCode 11 = IsSameFile
parseErrorCode 12 = Conflict
+parseErrorCode 13 = ConditionNotMatch
+parseErrorCode 14 = RangeNotSatisfied
parseErrorCode _ = FFIError
parseEntryMode :: Int -> EntryMode
diff --git a/bindings/haskell/src/result.rs b/bindings/haskell/src/result.rs
index 33959981b..e9f12affe 100644
--- a/bindings/haskell/src/result.rs
+++ b/bindings/haskell/src/result.rs
@@ -32,19 +32,21 @@ pub struct FFIResult<T> {
#[repr(C)]
#[derive(Debug)]
pub enum FFIErrorCode {
- Ok,
- FFIError,
- Unexpected,
- Unsupported,
- ConfigInvalid,
- NotFound,
- PermissionDenied,
- IsADirectory,
- NotADirectory,
- AlreadyExists,
- RateLimited,
- IsSameFile,
- Conflict,
+ Ok = 0,
+ FFIError = 1,
+ Unexpected = 2,
+ Unsupported = 3,
+ ConfigInvalid = 4,
+ NotFound = 5,
+ PermissionDenied = 6,
+ IsADirectory = 7,
+ NotADirectory = 8,
+ AlreadyExists = 9,
+ RateLimited = 10,
+ IsSameFile = 11,
+ Conflict = 12,
+ ConditionNotMatch = 13,
+ RangeNotSatisfied = 14,
}
impl<T> FFIResult<T> {
@@ -90,7 +92,30 @@ impl From<od::ErrorKind> for FFIErrorCode {
od::ErrorKind::RateLimited => FFIErrorCode::RateLimited,
od::ErrorKind::IsSameFile => FFIErrorCode::IsSameFile,
od::ErrorKind::Conflict => FFIErrorCode::Conflict,
+ od::ErrorKind::ConditionNotMatch => FFIErrorCode::ConditionNotMatch,
+ od::ErrorKind::RangeNotSatisfied => FFIErrorCode::RangeNotSatisfied,
_ => FFIErrorCode::Unexpected,
}
}
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ #[test]
+ fn error_kind_conversion_preserves_public_error_kinds() {
+ assert!(matches!(
+ FFIErrorCode::from(od::ErrorKind::ConditionNotMatch),
+ FFIErrorCode::ConditionNotMatch
+ ));
+ assert!(matches!(
+ FFIErrorCode::from(od::ErrorKind::RangeNotSatisfied),
+ FFIErrorCode::RangeNotSatisfied
+ ));
+ assert!(matches!(
+ FFIErrorCode::from(od::ErrorKind::Conflict),
+ FFIErrorCode::Conflict
+ ));
+ }
+}
diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs
index 204b320d3..3d259ec0c 100644
--- a/core/core/src/raw/ops.rs
+++ b/core/core/src/raw/ops.rs
@@ -195,6 +195,15 @@ impl OpDelete {
string_value(&self.values, DeleteField::IfVersionNotMatch as usize)
}
+ /// Returns whether at least one value-bearing conditional option is set.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ }
+
pub(crate) fn set_recursive(&mut self, recursive: bool) {
if recursive {
self.flags |= OP_DELETE_RECURSIVE;
@@ -519,6 +528,17 @@ impl OpRead {
.map(decode_timestamp)
}
+ /// Returns whether at least one value-bearing conditional option is set.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ || self.if_modified_since().is_some()
+ || self.if_unmodified_since().is_some()
+ }
+
/// Get version from option
#[inline]
pub fn version(&self) -> Option<&str> {
@@ -722,6 +742,17 @@ impl OpStat {
.map(decode_timestamp)
}
+ /// Returns whether at least one value-bearing conditional option is set.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ || self.if_modified_since().is_some()
+ || self.if_unmodified_since().is_some()
+ }
+
/// Returns the content-disposition header that should be sent back by the remote read
/// operation.
#[inline]
@@ -967,6 +998,19 @@ impl OpWrite {
self.flags & OP_WRITE_IF_NOT_EXISTS != 0
}
+ /// Returns whether at least one conditional option is set.
+ ///
+ /// Value-bearing options are set when present. Boolean options are set only
+ /// when their value is `true`.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_not_exists()
+ || self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ }
+
/// Get the user defined metadata from the op
#[inline]
pub fn user_metadata(&self) -> Option<UserMetadata<'_>> {
@@ -1184,6 +1228,19 @@ impl OpCompose {
self.flags & OP_COMPOSE_IF_NOT_EXISTS != 0
}
+ /// Returns whether at least one conditional option is set.
+ ///
+ /// Value-bearing options are set when present. Boolean options are set only
+ /// when their value is `true`.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_not_exists()
+ || self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ }
+
#[doc(hidden)]
pub fn into_content_type(mut self, value: &str) -> Self {
self.values = self
@@ -1314,6 +1371,19 @@ impl OpCopy {
string_value(&self.values, CopyField::IfVersionNotMatch as usize)
}
+ /// Returns whether at least one conditional option is set.
+ ///
+ /// Value-bearing options are set when present. Boolean options are set only
+ /// when their value is `true`.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_not_exists()
+ || self.if_match().is_some()
+ || self.if_none_match().is_some()
+ || self.if_version_match().is_some()
+ || self.if_version_not_match().is_some()
+ }
+
/// Get source version from the operation.
#[inline]
pub fn source_version(&self) -> Option<&str> {
@@ -1371,6 +1441,12 @@ impl OpRename {
pub fn if_not_exists(&self) -> bool {
self.if_not_exists
}
+
+ /// Returns whether the boolean conditional option is set to `true`.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_not_exists()
+ }
}
impl From<options::RenameOptions> for OpRename {
@@ -1407,6 +1483,12 @@ impl OpRestore {
pub fn if_not_exists(&self) -> bool {
self.flags & OP_RESTORE_IF_NOT_EXISTS != 0
}
+
+ /// Returns whether the boolean conditional option is set to `true`.
+ #[inline]
+ pub fn is_conditional(&self) -> bool {
+ self.if_not_exists()
+ }
}
impl From<options::RestoreOptions> for OpRestore {
@@ -1480,6 +1562,7 @@ mod tests {
assert_eq!(args.override_content_type(), Some("text/plain"));
assert_eq!(args.override_cache_control(), Some("no-cache"));
assert_eq!(args.override_content_disposition(), Some("attachment"));
+ assert!(args.is_conditional());
}
#[test]
@@ -1516,6 +1599,7 @@ mod tests {
assert_eq!(args.if_version_not_match(), Some("version-not-match"));
assert!(args.if_not_exists());
assert_eq!(args.user_metadata().unwrap().get("owner"), Some("opendal"));
+ assert!(args.is_conditional());
}
#[test]
@@ -1550,6 +1634,7 @@ mod tests {
assert_eq!(args.if_version_not_match(), Some("version-not-match"));
assert!(args.if_not_exists());
assert_eq!(args.user_metadata().unwrap().get("owner"), Some("opendal"));
+ assert!(args.is_conditional());
}
#[test]
@@ -1647,6 +1732,7 @@ mod tests {
assert_eq!(delete.version(), Some("version"));
assert!(delete.recursive());
assert_eq!(delete.if_match(), Some("etag"));
+ assert!(delete.is_conditional());
let copy = OpCopy::from_options(
&Capability::default(),
@@ -1661,6 +1747,7 @@ mod tests {
assert!(copy.if_not_exists());
assert_eq!(copy.if_version_match(), Some("destination-version"));
assert_eq!(copy.source_version(), Some("source-version"));
+ assert!(copy.is_conditional());
let list: OpList = options::ListOptions {
limit: Some(100),
@@ -1683,5 +1770,6 @@ mod tests {
.into();
assert_eq!(restore.version(), Some("version"));
assert!(restore.if_not_exists());
+ assert!(restore.is_conditional());
}
}
diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs
index a2df1d84b..6c9bbb1af 100644
--- a/core/services/azblob/src/backend.rs
+++ b/core/services/azblob/src/backend.rs
@@ -492,8 +492,7 @@ impl Service for AzblobBackend {
async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
let error_ctx = ErrorContext::new(ServiceOperation("GetBlobProperties"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = self
.core
.azblob_get_blob_properties(ctx, path, &args)
diff --git a/core/services/azblob/src/copier.rs b/core/services/azblob/src/copier.rs
index f4e5679ef..590018e2f 100644
--- a/core/services/azblob/src/copier.rs
+++ b/core/services/azblob/src/copier.rs
@@ -122,8 +122,7 @@ impl oio::BlockCopy for AzblobCopier {
StatusCode::ACCEPTED => AzblobWriter::parse_metadata(resp.headers()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("CopyBlob"))
- .with_if_match(self.args.if_match().is_some())
- .with_if_none_match(self.args.if_none_match().is_some())
+ .with_caller_condition(self.args.is_conditional())
.with_if_not_exists(self.args.if_not_exists()),
resp,
)),
@@ -163,8 +162,7 @@ impl oio::BlockCopy for AzblobCopier {
StatusCode::CREATED | StatusCode::OK => Ok(meta),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("PutBlockList"))
- .with_if_match(self.args.if_match().is_some())
- .with_if_none_match(self.args.if_none_match().is_some())
+ .with_caller_condition(self.args.is_conditional())
.with_if_not_exists(self.args.if_not_exists()),
resp,
)),
diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs
index 68bb724d6..c0a5b1453 100644
--- a/core/services/azblob/src/core.rs
+++ b/core/services/azblob/src/core.rs
@@ -384,6 +384,10 @@ impl AzblobCore {
"AppendBlob",
);
+ if args.if_not_exists() {
+ req = req.header(IF_NONE_MATCH, "*");
+ }
+
if let Some(ty) = args.content_type() {
req = req.header(CONTENT_TYPE, ty)
}
@@ -940,6 +944,25 @@ mod tests {
use reqsign_azure_storage::RequestSigner;
use reqsign_azure_storage::StaticCredentialProvider;
+ fn test_core() -> AzblobCore {
+ AzblobCore {
+ info: ServiceInfo::new("azblob", "/", "c"),
+ capability: Capability::default(),
+ container: "c".to_string(),
+ root: "/".to_string(),
+ endpoint: "https://acc.blob.core.windows.net".to_string(),
+ encryption_key: None,
+ encryption_key_sha256: None,
+ encryption_algorithm: None,
+ skip_signature: true,
+ signer: Signer::new(
+ Context::new(),
+ StaticCredentialProvider::new_shared_key("acc", "a2V5"),
+ RequestSigner::new(),
+ ),
+ }
+ }
+
#[test]
fn test_parse_xml() {
let bs = bytes::Bytes::from(
@@ -1126,22 +1149,7 @@ mod tests {
/// have to be carried here as well as on the one-shot Put Blob path.
#[test]
fn test_put_block_list_carries_content_type_and_user_metadata() {
- let core = AzblobCore {
- info: ServiceInfo::new("azblob", "/", "c"),
- capability: Capability::default(),
- container: "c".to_string(),
- root: "/".to_string(),
- endpoint: "https://acc.blob.core.windows.net".to_string(),
- encryption_key: None,
- encryption_key_sha256: None,
- encryption_algorithm: None,
- skip_signature: true,
- signer: Signer::new(
- Context::new(),
- StaticCredentialProvider::new_shared_key("acc", "a2V5"),
- RequestSigner::new(),
- ),
- };
+ let core = test_core();
let mut meta = HashMap::new();
meta.insert("k".to_string(), "v".to_string());
@@ -1173,6 +1181,31 @@ mod tests {
);
}
+ #[test]
+ fn init_appendable_blob_carries_if_not_exists() {
+ let core = test_core();
+ let (args, _) = OpWrite::from_options(
+ &Capability {
+ write_with_if_not_exists: true,
+ ..Default::default()
+ },
+ options::WriteOptions {
+ if_not_exists: true,
+ ..Default::default()
+ },
+ )
+ .expect("options must lower");
+
+ let req = core
+ .azblob_init_appendable_blob_request("a.txt", &args)
+ .expect("request must build");
+
+ assert_eq!(
+ req.headers().get(IF_NONE_MATCH),
+ Some(&HeaderValue::from_static("*"))
+ );
+ }
+
#[test]
fn test_serialize_put_block_list_request() {
let req = PutBlockListRequest {
@@ -1246,10 +1279,7 @@ impl Debug for AzblobError {
#[derive(Clone, Copy, Debug)]
pub struct ErrorContext {
service_operation: ServiceOperation,
- if_match: bool,
- if_none_match: bool,
- if_modified_since: bool,
- if_unmodified_since: bool,
+ caller_condition: bool,
if_not_exists: bool,
is_append_blob_initialization: bool,
}
@@ -1258,36 +1288,19 @@ impl ErrorContext {
pub const fn new(service_operation: ServiceOperation) -> Self {
Self {
service_operation,
- if_match: false,
- if_none_match: false,
- if_modified_since: false,
- if_unmodified_since: false,
+ caller_condition: false,
if_not_exists: false,
is_append_blob_initialization: false,
}
}
- pub const fn with_if_match(mut self, if_match: bool) -> Self {
- self.if_match = if_match;
- self
- }
-
- pub const fn with_if_none_match(mut self, if_none_match: bool) -> Self {
- self.if_none_match = if_none_match;
- self
- }
-
- pub const fn with_if_modified_since(mut self, if_modified_since: bool) -> Self {
- self.if_modified_since = if_modified_since;
- self
- }
-
- pub const fn with_if_unmodified_since(mut self, if_unmodified_since: bool) -> Self {
- self.if_unmodified_since = if_unmodified_since;
+ pub const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
self
}
pub const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.caller_condition = self.caller_condition || if_not_exists;
self.if_not_exists = if_not_exists;
self
}
@@ -1299,14 +1312,6 @@ impl ErrorContext {
self.is_append_blob_initialization = is_append_blob_initialization;
self
}
-
- const fn has_condition(self) -> bool {
- self.if_match
- || self.if_none_match
- || self.if_modified_since
- || self.if_unmodified_since
- || self.if_not_exists
- }
}
/// Parse error response into Error.
@@ -1329,7 +1334,7 @@ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (mut kind, mut retryable) = match parts.status {
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.has_condition() => {
+ StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
@@ -1380,7 +1385,7 @@ fn parse_azblob_error_code(
) -> Option<(ErrorKind, bool)> {
match code {
"ConditionNotMet" | "SourceConditionNotMet" | "TargetConditionNotMet"
- if ctx.has_condition()
+ if ctx.caller_condition
&& matches!(
status,
StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED
@@ -1389,14 +1394,14 @@ fn parse_azblob_error_code(
Some((ErrorKind::ConditionNotMatch, false))
}
"BlobAlreadyExists" | "ContainerAlreadyExists" | "ResourceAlreadyExists" => {
- let kind = if ctx.if_not_exists {
- ErrorKind::ConditionNotMatch
+ let classification = if ctx.if_not_exists {
+ (ErrorKind::ConditionNotMatch, false)
} else if ctx.is_append_blob_initialization {
- ErrorKind::Conflict
+ (ErrorKind::Conflict, true)
} else {
- ErrorKind::AlreadyExists
+ (ErrorKind::AlreadyExists, false)
};
- Some((kind, false))
+ Some(classification)
}
"DirectorySasNotSupportedVersion"
| "FeatureVersionMismatch"
@@ -1461,3 +1466,26 @@ fn parse_azblob_error_code(
_ => None,
}
}
+
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+
+ #[test]
+ fn append_blob_initialization_preserves_condition_and_retryability() {
+ let conditional = ErrorContext::new(ServiceOperation("PutBlob"))
+ .with_if_not_exists(true)
+ .with_append_blob_initialization(true);
+ assert_eq!(
+ parse_azblob_error_code(conditional, "BlobAlreadyExists", StatusCode::CONFLICT),
+ Some((ErrorKind::ConditionNotMatch, false))
+ );
+
+ let raced =
+ ErrorContext::new(ServiceOperation("PutBlob")).with_append_blob_initialization(true);
+ assert_eq!(
+ parse_azblob_error_code(raced, "BlobAlreadyExists", StatusCode::CONFLICT),
+ Some((ErrorKind::Conflict, true))
+ );
+ }
+}
diff --git a/core/services/azblob/src/deleter.rs b/core/services/azblob/src/deleter.rs
index b3c2f7b04..43f6a98cc 100644
--- a/core/services/azblob/src/deleter.rs
+++ b/core/services/azblob/src/deleter.rs
@@ -54,8 +54,7 @@ impl oio::BatchDelete for AzblobDeleter {
StatusCode::NOT_FOUND => Ok(()),
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("DeleteBlob"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some()),
+ .with_caller_condition(args.is_conditional()),
resp,
)),
}
@@ -116,8 +115,7 @@ impl oio::BatchDelete for AzblobDeleter {
batched_result.succeeded.push((path, args));
} else {
let error_ctx = ErrorContext::new(ServiceOperation("BatchDeleteBlobs"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some());
+ .with_caller_condition(args.is_conditional());
let err = parse_error(error_ctx, resp);
batched_result.failed.push((path, args, err));
}
diff --git a/core/services/azblob/src/reader.rs b/core/services/azblob/src/reader.rs
index c48f7d18d..537405cc5 100644
--- a/core/services/azblob/src/reader.rs
+++ b/core/services/azblob/src/reader.rs
@@ -53,10 +53,7 @@ impl oio::StreamRead for AzblobReader {
let path = self.path.as_str();
let args = self.args.clone();
let error_ctx = ErrorContext::new(ServiceOperation("GetBlob"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some())
- .with_if_modified_since(args.if_modified_since().is_some())
- .with_if_unmodified_since(args.if_unmodified_since().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = backend
.core
.azblob_get_blob(&self.ctx, path, range, &args)
diff --git a/core/services/azblob/src/writer.rs b/core/services/azblob/src/writer.rs
index d62f3bc34..8449ad504 100644
--- a/core/services/azblob/src/writer.rs
+++ b/core/services/azblob/src/writer.rs
@@ -71,8 +71,7 @@ impl AzblobWriter {
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_caller_condition(self.op.is_conditional())
.with_if_not_exists(self.op.if_not_exists())
}
}
@@ -88,6 +87,12 @@ impl oio::AppendWrite for AzblobWriter {
match status {
StatusCode::OK => {
+ if self.op.if_not_exists() {
+ return Err(Error::new(
+ ErrorKind::ConditionNotMatch,
+ "the blob already exists",
+ ));
+ }
let headers = resp.headers();
let blob_type = headers.get(X_MS_BLOB_TYPE).and_then(|v| v.to_str().ok());
if blob_type != Some("AppendBlob") {
@@ -113,6 +118,7 @@ impl oio::AppendWrite for AzblobWriter {
_ => {
return Err(parse_error(
ErrorContext::new(ServiceOperation("PutBlob"))
+ .with_if_not_exists(self.op.if_not_exists())
.with_append_blob_initialization(true),
resp,
));
diff --git a/core/services/azdls/src/core.rs b/core/services/azdls/src/core.rs
index 1e344aa5c..106e5a19f 100644
--- a/core/services/azdls/src/core.rs
+++ b/core/services/azdls/src/core.rs
@@ -391,7 +391,8 @@ impl AzdlsCore {
if resp.status() != StatusCode::OK {
return Err(parse_error(
- ErrorContext::new(ServiceOperation("GetPathProperties")),
+ ErrorContext::new(ServiceOperation("GetPathProperties"))
+ .with_caller_condition(args.is_conditional()),
resp,
));
}
@@ -534,7 +535,8 @@ impl AzdlsCore {
StatusCode::OK | StatusCode::ACCEPTED | StatusCode::NOT_FOUND => {}
_ => {
return Err(parse_error(
- ErrorContext::new(ServiceOperation("RecursiveDeletePath")),
+ ErrorContext::new(ServiceOperation("RecursiveDeletePath"))
+ .with_delete_match_condition(args.if_match().is_some()),
resp,
));
}
@@ -691,14 +693,55 @@ mod tests {
parse_error(ctx, resp).kind()
};
- let read_ctx = ErrorContext::new(ServiceOperation("ReadFile")).with_if_match(true);
+ let read_ctx = ErrorContext::new(ServiceOperation("ReadFile")).with_caller_condition(true);
assert_eq!(parse_missing(read_ctx), ErrorKind::NotFound);
- let delete_ctx = ErrorContext::new(ServiceOperation("DeletePath"))
- .with_if_match(true)
- .with_delete();
+ let delete_ctx =
+ ErrorContext::new(ServiceOperation("DeletePath")).with_delete_match_condition(true);
assert_eq!(parse_missing(delete_ctx), ErrorKind::ConditionNotMatch);
}
+
+ #[test]
+ fn test_parse_conflict_uses_native_code_and_operation_context() {
+ let parse_conflict = |ctx, code: &str| {
+ let body = Buffer::from(format!(
+ "<Error><Code>{code}</Code><Message>test</Message></Error>"
+ ));
+ let resp = Response::builder()
+ .status(StatusCode::CONFLICT)
+ .body(body)
+ .expect("response must build");
+ parse_error(ctx, resp)
+ };
+
+ let conditional =
+ ErrorContext::new(ServiceOperation("CreateFile")).with_if_not_exists(true);
+ assert_eq!(
+ parse_conflict(conditional, "PathAlreadyExists").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+
+ let unconditional = ErrorContext::new(ServiceOperation("CreateFile"));
+ assert_eq!(
+ parse_conflict(unconditional, "PathAlreadyExists").kind(),
+ ErrorKind::AlreadyExists
+ );
+ for code in [
+ "DestinationPathIsBeingDeleted",
+ "PathConflict",
+ "ResourceTypeMismatch",
+ ] {
+ assert_eq!(
+ parse_conflict(unconditional, code).kind(),
+ ErrorKind::Conflict,
+ "native code {code}"
+ );
+ }
+ assert_eq!(
+ parse_conflict(unconditional, "UnknownConflict").kind(),
+ ErrorKind::Unexpected
+ );
+ }
}
use bytes::Buf;
@@ -743,26 +786,38 @@ impl Debug for AzdlsError {
#[derive(Clone, Copy, Debug)]
pub(crate) struct ErrorContext {
service_operation: ServiceOperation,
- if_match: bool,
- is_delete: bool,
+ caller_condition: bool,
+ if_not_exists: bool,
+ delete_match_condition: bool,
}
impl ErrorContext {
pub(crate) const fn new(service_operation: ServiceOperation) -> Self {
Self {
service_operation,
- if_match: false,
- is_delete: false,
+ caller_condition: false,
+ if_not_exists: false,
+ delete_match_condition: false,
}
}
- pub(crate) const fn with_if_match(mut self, if_match: bool) -> Self {
- self.if_match = if_match;
+ pub(crate) const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
self
}
- pub(crate) const fn with_delete(mut self) -> Self {
- self.is_delete = true;
+ pub(crate) const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.caller_condition = self.caller_condition || if_not_exists;
+ self.if_not_exists = if_not_exists;
+ self
+ }
+
+ pub(crate) const fn with_delete_match_condition(
+ mut self,
+ delete_match_condition: bool,
+ ) -> Self {
+ self.caller_condition = self.caller_condition || delete_match_condition;
+ self.delete_match_condition = delete_match_condition;
self
}
}
@@ -772,13 +827,23 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (parts, body) = resp.into_parts();
let bs = body.to_bytes();
- let (kind, retryable) = match parts.status {
- StatusCode::NOT_FOUND if ctx.is_delete && ctx.if_match => {
+ let mut azdls_error = de::from_reader::<_, AzdlsError>(bs.clone().reader()).ok();
+ if azdls_error.as_ref().is_none_or(|err| err.code.is_empty())
+ && let Some(code) = parts
+ .headers
+ .get("x-ms-error-code")
+ .and_then(|value| value.to_str().ok())
+ {
+ azdls_error.get_or_insert_with(AzdlsError::default).code = code.to_string();
+ }
+
+ let (mut kind, mut retryable) = match parts.status {
+ StatusCode::NOT_FOUND if ctx.delete_match_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED | StatusCode::CONFLICT => {
+ StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
@@ -789,24 +854,51 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
_ => (ErrorKind::Unexpected, false),
};
- let mut message = match de::from_reader::<_, AzdlsError>(bs.clone().reader()) {
- Ok(azdls_err) => format!("{azdls_err:?}"),
- Err(_) => String::from_utf8_lossy(&bs).into_owned(),
- };
- // If there is no body here, fill with error code.
- if message.is_empty()
- && let Some(v) = parts.headers.get("x-ms-error-code")
- && let Ok(code) = v.to_str()
- {
- message = format!(
- "{:?}",
- AzdlsError {
- code: code.to_string(),
- ..Default::default()
+ if let Some(azdls_error) = &azdls_error {
+ match azdls_error.code.as_str() {
+ "ConditionNotMet" if ctx.caller_condition => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
}
- )
+ "PathAlreadyExists" if ctx.if_not_exists => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ }
+ "PathAlreadyExists" => {
+ (kind, retryable) = (ErrorKind::AlreadyExists, false);
+ }
+ "DestinationPathIsBeingDeleted"
+ | "DirectoryNotEmpty"
+ | "FilesystemBeingDeleted"
+ | "InvalidDestinationPath"
+ | "InvalidFlushOperation"
+ | "LeaseAlreadyPresent"
+ | "LeaseIdMismatchWithLeaseOperation"
+ | "LeaseIdMismatchWithPathOperation"
+ | "LeaseIdMissing"
+ | "LeaseIsAlreadyBroken"
+ | "LeaseIsBreakingAndCannotBeAcquired"
+ | "LeaseIsBreakingAndCannotBeChanged"
+ | "LeaseIsBrokenAndCannotBeRenewed"
+ | "LeaseNotPresentWithLeaseOperation"
+ | "PathConflict"
+ | "ResourceTypeMismatch"
+ | "SourcePathIsBeingDeleted" => {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ }
+ _ if matches!(
+ parts.status,
+ StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+ ) =>
+ {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
+ _ => {}
+ }
}
+ let message = azdls_error
+ .map(|err| format!("{err:?}"))
+ .unwrap_or_else(|| String::from_utf8_lossy(&bs).into_owned());
+
let mut err = Error::new(kind, &message);
err = err.with_context("service_operation", ctx.service_operation.0);
diff --git a/core/services/azdls/src/deleter.rs b/core/services/azdls/src/deleter.rs
index 1d7864ff4..ade348278 100644
--- a/core/services/azdls/src/deleter.rs
+++ b/core/services/azdls/src/deleter.rs
@@ -51,8 +51,7 @@ impl oio::OneShotDelete for AzdlsDeleter {
} else {
ServiceOperation("DeletePath")
})
- .with_if_match(args.if_match().is_some())
- .with_delete();
+ .with_delete_match_condition(args.if_match().is_some());
match status {
StatusCode::OK => Ok(()),
diff --git a/core/services/azdls/src/reader.rs b/core/services/azdls/src/reader.rs
index c7eaba54c..ef148aeda 100644
--- a/core/services/azdls/src/reader.rs
+++ b/core/services/azdls/src/reader.rs
@@ -66,7 +66,8 @@ impl oio::StreamRead for AzdlsReader {
let (part, mut body) = resp.into_parts();
let buf = body.to_buffer().await?;
return Err(parse_error(
- ErrorContext::new(ServiceOperation("ReadFile")),
+ ErrorContext::new(ServiceOperation("ReadFile"))
+ .with_caller_condition(args.is_conditional()),
Response::from_parts(part, buf),
));
}
diff --git a/core/services/azdls/src/writer.rs b/core/services/azdls/src/writer.rs
index 80170c3db..5d16762ff 100644
--- a/core/services/azdls/src/writer.rs
+++ b/core/services/azdls/src/writer.rs
@@ -77,16 +77,20 @@ impl AzdlsWriter {
.await?;
match resp.status() {
StatusCode::CREATED | StatusCode::OK => Ok(()),
- StatusCode::CONFLICT if self.op.if_not_exists() => Err(parse_error(
- ErrorContext::new(ServiceOperation("CreateFile")),
- resp,
- )
- .with_operation("Backend::azdls_create_request")),
- StatusCode::CONFLICT => Ok(()),
- _ => Err(
- parse_error(ErrorContext::new(ServiceOperation("CreateFile")), resp)
- .with_operation("Backend::azdls_create_request"),
- ),
+ _ => {
+ let err = parse_error(
+ ErrorContext::new(ServiceOperation("CreateFile"))
+ .with_caller_condition(self.op.is_conditional())
+ .with_if_not_exists(self.op.if_not_exists()),
+ resp,
+ )
+ .with_operation("Backend::azdls_create_request");
+ if err.kind() == ErrorKind::AlreadyExists {
+ Ok(())
+ } else {
+ Err(err)
+ }
+ }
}
}
diff --git a/core/services/azfile/src/core.rs b/core/services/azfile/src/core.rs
index 6a0171b5f..9c8898ba9 100644
--- a/core/services/azfile/src/core.rs
+++ b/core/services/azfile/src/core.rs
@@ -552,12 +552,20 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (parts, body) = resp.into_parts();
let bs = body.to_bytes();
- let (kind, retryable) = match parts.status {
+ let mut azfile_error = de::from_reader::<_, AzfileError>(bs.clone().reader()).ok();
+ if azfile_error.as_ref().is_none_or(|err| err.code.is_empty())
+ && let Some(code) = parts
+ .headers
+ .get("x-ms-error-code")
+ .and_then(|value| value.to_str().ok())
+ {
+ azfile_error.get_or_insert_with(AzfileError::default).code = code.to_string();
+ }
+
+ let (mut kind, mut 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
@@ -565,25 +573,68 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
_ => (ErrorKind::Unexpected, false),
};
- let mut message = match de::from_reader::<_, AzfileError>(bs.clone().reader()) {
- Ok(azfile_err) => format!("{azfile_err:?}"),
- Err(_) => String::from_utf8_lossy(&bs).into_owned(),
- };
-
- // If there is no body here, fill with error code.
- if message.is_empty()
- && let Some(v) = parts.headers.get("x-ms-error-code")
- && let Ok(code) = v.to_str()
- {
- message = format!(
- "{:?}",
- AzfileError {
- code: code.to_string(),
- ..Default::default()
+ if let Some(azfile_error) = &azfile_error {
+ match azfile_error.code.as_str() {
+ "ParentNotFound" => {
+ (kind, retryable) = (ErrorKind::NotFound, false);
}
- )
+ "ResourceAlreadyExists" => {
+ (kind, retryable) = (ErrorKind::AlreadyExists, false);
+ }
+ "DeletePending" | "ShareBeingDeleted"
+ if ctx.service_operation == ServiceOperation("CreateDirectory") =>
+ {
+ (kind, retryable) = (ErrorKind::Conflict, true);
+ }
+ "CannotDeleteFileOrDirectory"
+ | "ContainerQuotaDowngradeNotAllowed"
+ | "DeletePending"
+ | "DeleteShareWhenSnapshotLeased"
+ | "DirectoryNotEmpty"
+ | "FileGenerationMismatch"
+ | "FileLockConflict"
+ | "FileOpenBySmbClient"
+ | "LeaseAcquireDuringShareDelete"
+ | "LeaseIdMismatchWithFileLeaseOperation"
+ | "LeaseIdMismatchWithFileOperation"
+ | "LeaseIdMismatchWithFileShareLeaseOperation"
+ | "LeaseIdMismatchWithFileShareOperation"
+ | "LeaseIdMissingWithFileOperation"
+ | "LeaseIdMissingWithFileShareOperation"
+ | "LeaseLostWithFileOperation"
+ | "LeaseLostWithFileShareOperation"
+ | "LeaseNotPresentWithFileLeaseOperation"
+ | "LeaseNotPresentWithFileOperation"
+ | "LeaseNotPresentWithFileShareLeaseOperation"
+ | "LeaseNotPresentWithFileShareOperation"
+ | "PreviousSnapshotNotFound"
+ | "PreviousSnapshotOperationNotSupported"
+ | "ReadOnlyAttribute"
+ | "RenameCannotOverwriteDestinationFile"
+ | "RenameCycle"
+ | "RenameDirectoryHasOpenFiles"
+ | "ShareAlreadyExists"
+ | "ShareBeingDeleted"
+ | "ShareHasSnapshots"
+ | "ShareSnapshotInProgress"
+ | "SharingViolation" => {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ }
+ _ if matches!(
+ parts.status,
+ StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+ ) =>
+ {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
+ _ => {}
+ }
}
+ let message = azfile_error
+ .map(|err| format!("{err:?}"))
+ .unwrap_or_else(|| String::from_utf8_lossy(&bs).into_owned());
+
let mut err = Error::new(kind, &message);
err = err.with_context("service_operation", ctx.service_operation.0);
@@ -595,3 +646,62 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
err
}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+
+ fn parse_native_error(operation: &'static str, status: StatusCode, code: &str) -> Error {
+ let resp = Response::builder()
+ .status(status)
+ .header("x-ms-error-code", code)
+ .body(Buffer::new())
+ .expect("response must build");
+ parse_error(ErrorContext::new(ServiceOperation(operation)), resp)
+ }
+
+ #[test]
+ fn test_parse_native_conflict_codes() {
+ let err = parse_native_error("CreateDirectory", StatusCode::CONFLICT, "DeletePending");
+ assert_eq!(err.kind(), ErrorKind::Conflict);
+ assert!(err.is_temporary());
+
+ let err = parse_native_error("PutRange", StatusCode::CONFLICT, "FileLockConflict");
+ assert_eq!(err.kind(), ErrorKind::Conflict);
+ assert!(!err.is_temporary());
+
+ let err = parse_native_error(
+ "GetFileProperties",
+ StatusCode::PRECONDITION_FAILED,
+ "LeaseIdMismatchWithFileOperation",
+ );
+ assert_eq!(err.kind(), ErrorKind::Conflict);
+ assert!(!err.is_temporary());
+ }
+
+ #[test]
+ fn test_parse_non_conflict_native_codes() {
+ assert_eq!(
+ parse_native_error(
+ "CreateDirectory",
+ StatusCode::CONFLICT,
+ "ResourceAlreadyExists"
+ )
+ .kind(),
+ ErrorKind::AlreadyExists
+ );
+ assert_eq!(
+ parse_native_error("CreateDirectory", StatusCode::NOT_FOUND, "ParentNotFound").kind(),
+ ErrorKind::NotFound
+ );
+ assert_eq!(
+ parse_native_error(
+ "GetFileProperties",
+ StatusCode::PRECONDITION_FAILED,
+ "UnknownPrecondition"
+ )
+ .kind(),
+ ErrorKind::Unexpected
+ );
+ }
+}
diff --git a/core/services/cloudflare-kv/src/reader.rs b/core/services/cloudflare-kv/src/reader.rs
index 6e7b39926..d14a3111a 100644
--- a/core/services/cloudflare-kv/src/reader.rs
+++ b/core/services/cloudflare-kv/src/reader.rs
@@ -66,11 +66,7 @@ impl oio::StreamRead for CloudflareKvReader {
let resp_body = resp.into_body();
- if args.if_match().is_some()
- || args.if_none_match().is_some()
- || args.if_modified_since().is_some()
- || args.if_unmodified_since().is_some()
- {
+ if args.is_conditional() {
let meta_resp = backend.core.metadata(&self.ctx, &path).await?;
if meta_resp.status() != StatusCode::OK {
diff --git a/core/services/cos/src/backend.rs b/core/services/cos/src/backend.rs
index e7e9ffc93..a6145d0e1 100644
--- a/core/services/cos/src/backend.rs
+++ b/core/services/cos/src/backend.rs
@@ -353,7 +353,8 @@ impl Service for CosBackend {
Ok(RpStat::new(meta.build()))
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("HeadObject")),
+ ErrorContext::new(ServiceOperation("HeadObject"))
+ .with_caller_condition(args.is_conditional()),
resp,
)),
}
@@ -468,7 +469,8 @@ impl Service for CosBackend {
match status {
StatusCode::OK => Ok(MetadataBuilder::file(source_size).build()),
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("CopyObject")),
+ ErrorContext::new(ServiceOperation("CopyObject"))
+ .with_if_not_exists(args.if_not_exists()),
resp,
)),
}
diff --git a/core/services/cos/src/core.rs b/core/services/cos/src/core.rs
index 86a63a7db..ae25ea245 100644
--- a/core/services/cos/src/core.rs
+++ b/core/services/cos/src/core.rs
@@ -888,11 +888,28 @@ struct CosError {
#[derive(Clone, Copy, Debug)]
pub(crate) struct ErrorContext {
service_operation: ServiceOperation,
+ caller_condition: bool,
+ if_not_exists: bool,
}
impl ErrorContext {
pub(crate) const fn new(service_operation: ServiceOperation) -> Self {
- Self { service_operation }
+ Self {
+ service_operation,
+ caller_condition: false,
+ if_not_exists: false,
+ }
+ }
+
+ pub(crate) const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
+ self
+ }
+
+ pub(crate) const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.caller_condition = self.caller_condition || if_not_exists;
+ self.if_not_exists = if_not_exists;
+ self
}
}
@@ -901,10 +918,12 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (parts, body) = resp.into_parts();
let bs = body.to_bytes();
- let (kind, retryable) = match parts.status {
+ let cos_error = de::from_reader::<_, CosError>(bs.clone().reader()).ok();
+
+ let (mut kind, mut retryable) = match parts.status {
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED | StatusCode::CONFLICT => {
+ StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::INTERNAL_SERVER_ERROR
@@ -917,9 +936,38 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
_ => (ErrorKind::Unexpected, false),
};
- let message = match de::from_reader::<_, CosError>(bs.clone().reader()) {
- Ok(cos_error) => format!("{cos_error:?}"),
- Err(_) => String::from_utf8_lossy(&bs).into_owned(),
+ if let Some(cos_error) = &cos_error {
+ match cos_error.code.as_str() {
+ "PreconditionFailed" if ctx.caller_condition => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ }
+ "FileAlreadyExists" if ctx.if_not_exists => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ }
+ "FileAlreadyExists"
+ | "PathConflict"
+ | "UploadConflict"
+ | "InvalidBucketState"
+ | "ObjectLocked"
+ | "InvalidObjectState"
+ | "RestoreAlreadyInProgress"
+ | "ObjectNotAppendable" => {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ }
+ _ if matches!(
+ parts.status,
+ StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+ ) =>
+ {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
+ _ => {}
+ }
+ }
+
+ let message = match cos_error {
+ Some(cos_error) => format!("{cos_error:?}"),
+ None => String::from_utf8_lossy(&bs).into_owned(),
};
let mut err = Error::new(kind, message);
@@ -932,3 +980,55 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
}
err
}
+
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+
+ fn parse_cos_error(ctx: ErrorContext, status: StatusCode, code: &str) -> Error {
+ let body = Buffer::from(format!(
+ "<Error><Code>{code}</Code><Message>test</Message></Error>"
+ ));
+ let resp = Response::builder()
+ .status(status)
+ .body(body)
+ .expect("response must build");
+ parse_error(ctx, resp)
+ }
+
+ #[test]
+ fn conflict_classification_uses_native_code_and_condition() {
+ let conditional = ErrorContext::new(ServiceOperation("PutObject")).with_if_not_exists(true);
+ assert_eq!(
+ parse_cos_error(conditional, StatusCode::CONFLICT, "FileAlreadyExists").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ assert_eq!(
+ parse_cos_error(
+ conditional,
+ StatusCode::PRECONDITION_FAILED,
+ "PreconditionFailed"
+ )
+ .kind(),
+ ErrorKind::ConditionNotMatch
+ );
+
+ let unconditional = ErrorContext::new(ServiceOperation("PutObject"));
+ for code in [
+ "PathConflict",
+ "UploadConflict",
+ "ObjectLocked",
+ "InvalidObjectState",
+ ] {
+ assert_eq!(
+ parse_cos_error(unconditional, StatusCode::CONFLICT, code).kind(),
+ ErrorKind::Conflict,
+ "native code {code}"
+ );
+ }
+ assert_eq!(
+ parse_cos_error(unconditional, StatusCode::CONFLICT, "UnknownConflict").kind(),
+ ErrorKind::Unexpected
+ );
+ }
+}
diff --git a/core/services/cos/src/reader.rs b/core/services/cos/src/reader.rs
index 473d9799d..559da7aac 100644
--- a/core/services/cos/src/reader.rs
+++ b/core/services/cos/src/reader.rs
@@ -67,7 +67,8 @@ impl oio::StreamRead for CosReader {
let (part, mut body) = resp.into_parts();
let buf = body.to_buffer().await?;
return Err(parse_error(
- ErrorContext::new(ServiceOperation("GetObject")),
+ ErrorContext::new(ServiceOperation("GetObject"))
+ .with_caller_condition(args.is_conditional()),
Response::from_parts(part, buf),
));
}
diff --git a/core/services/cos/src/writer.rs b/core/services/cos/src/writer.rs
index c670872b4..d972b6451 100644
--- a/core/services/cos/src/writer.rs
+++ b/core/services/cos/src/writer.rs
@@ -88,7 +88,8 @@ impl oio::MultipartWrite for CosWriter {
match status {
StatusCode::CREATED | StatusCode::OK => Ok(meta),
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("PutObject")),
+ ErrorContext::new(ServiceOperation("PutObject"))
+ .with_if_not_exists(self.op.if_not_exists()),
resp,
)),
}
@@ -113,7 +114,8 @@ impl oio::MultipartWrite for CosWriter {
Ok(result.upload_id)
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("InitiateMultipartUpload")),
+ ErrorContext::new(ServiceOperation("InitiateMultipartUpload"))
+ .with_if_not_exists(self.op.if_not_exists()),
resp,
)),
}
@@ -181,20 +183,20 @@ impl oio::MultipartWrite for CosWriter {
let mut meta = Self::parse_metadata(resp.headers())?.into_builder();
+ let status = resp.status();
+ if status != StatusCode::OK {
+ return Err(parse_error(
+ ErrorContext::new(ServiceOperation("CompleteMultipartUpload"))
+ .with_if_not_exists(self.op.if_not_exists()),
+ resp,
+ ));
+ }
+
let result: CompleteMultipartUploadResult =
quick_xml::de::from_reader(resp.body_mut().reader())
.map_err(new_xml_deserialize_error)?;
meta.etag(&result.etag);
-
- let status = resp.status();
-
- match status {
- StatusCode::OK => Ok(meta.build()),
- _ => Err(parse_error(
- ErrorContext::new(ServiceOperation("CompleteMultipartUpload")),
- resp,
- )),
- }
+ Ok(meta.build())
}
async fn abort_part(&self, upload_id: &str) -> Result<()> {
diff --git a/core/services/gcs-grpc/src/backend.rs b/core/services/gcs-grpc/src/backend.rs
index cea4be712..019940f56 100644
--- a/core/services/gcs-grpc/src/backend.rs
+++ b/core/services/gcs-grpc/src/backend.rs
@@ -34,7 +34,7 @@ use opendal_core::*;
use crate::GCS_GRPC_SCHEME;
use crate::config::GcsGrpcConfig;
use crate::copier::new_gcs_grpc_copier;
-use crate::core::{GcsGrpcCore, parse_generation, parse_object, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_generation, parse_object, parse_status};
use crate::deleter::GcsGrpcDeleter;
use crate::generated::google::storage::v2::GetObjectRequest;
use crate::lister::GcsGrpcLister;
@@ -347,7 +347,9 @@ impl Service for GcsGrpcBackend {
.client()
.get_object(request)
.await
- .map_err(parse_status)?;
+ .map_err(|status| {
+ parse_status(ErrorContext::new(ServiceOperation("GetObject")), status)
+ })?;
Ok(RpStat::new(parse_object(response.get_ref())))
}
diff --git a/core/services/gcs-grpc/src/copier.rs b/core/services/gcs-grpc/src/copier.rs
index 2d95a65a8..bbcda6438 100644
--- a/core/services/gcs-grpc/src/copier.rs
+++ b/core/services/gcs-grpc/src/copier.rs
@@ -20,7 +20,7 @@ use std::sync::Arc;
use opendal_core::raw::*;
use opendal_core::*;
-use crate::core::{GcsGrpcCore, parse_generation, parse_object, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_generation, parse_object, parse_status};
use crate::generated::google::storage::v2::RewriteObjectRequest;
pub(super) fn new_gcs_grpc_copier(
@@ -71,7 +71,13 @@ async fn copy_object(
.client()
.rewrite_object(tonic_request)
.await
- .map_err(parse_status)?
+ .map_err(|status| {
+ parse_status(
+ ErrorContext::new(ServiceOperation("RewriteObject"))
+ .with_if_not_exists(args.if_not_exists()),
+ status,
+ )
+ })?
.into_inner();
if response.done {
return response.resource.as_ref().map(parse_object).ok_or_else(|| {
diff --git a/core/services/gcs-grpc/src/core.rs b/core/services/gcs-grpc/src/core.rs
index 033c62263..ef10d0e3b 100644
--- a/core/services/gcs-grpc/src/core.rs
+++ b/core/services/gcs-grpc/src/core.rs
@@ -147,22 +147,43 @@ fn build_routing_header(parameters: &[(&str, &str)]) -> Option<String> {
})
}
-pub(crate) fn parse_status(status: tonic::Status) -> Error {
+#[derive(Clone, Copy)]
+pub(crate) struct ErrorContext {
+ service_operation: ServiceOperation,
+ if_not_exists: bool,
+}
+
+impl ErrorContext {
+ pub(crate) const fn new(service_operation: ServiceOperation) -> Self {
+ Self {
+ service_operation,
+ if_not_exists: false,
+ }
+ }
+
+ pub(crate) const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.if_not_exists = if_not_exists;
+ self
+ }
+}
+
+pub(crate) fn parse_status(ctx: ErrorContext, status: tonic::Status) -> Error {
use tonic::Code;
let kind = match status.code() {
Code::NotFound => ErrorKind::NotFound,
+ Code::AlreadyExists if ctx.if_not_exists => ErrorKind::ConditionNotMatch,
Code::AlreadyExists => ErrorKind::AlreadyExists,
Code::PermissionDenied | Code::Unauthenticated => ErrorKind::PermissionDenied,
- Code::FailedPrecondition => ErrorKind::ConditionNotMatch,
+ Code::FailedPrecondition if ctx.if_not_exists => ErrorKind::ConditionNotMatch,
+ Code::FailedPrecondition | Code::Aborted => ErrorKind::Conflict,
Code::OutOfRange => ErrorKind::RangeNotSatisfied,
Code::ResourceExhausted => ErrorKind::RateLimited,
_ => ErrorKind::Unexpected,
};
let temporary = matches!(
status.code(),
- Code::Aborted
- | Code::Cancelled
+ Code::Cancelled
| Code::DeadlineExceeded
| Code::Internal
| Code::ResourceExhausted
@@ -170,7 +191,8 @@ pub(crate) fn parse_status(status: tonic::Status) -> Error {
| Code::Unknown
);
let mut err = Error::new(kind, status.message().to_string())
- .with_context("grpc_code", status.code().description());
+ .with_context("grpc_code", status.code().description())
+ .with_context("service_operation", ctx.service_operation.0);
if temporary {
err = err.set_temporary();
}
@@ -249,4 +271,28 @@ mod tests {
)
);
}
+
+ #[test]
+ fn status_classification_uses_operation_context() {
+ let caller = ErrorContext::new(ServiceOperation("WriteObject")).with_if_not_exists(true);
+ let err = parse_status(
+ caller,
+ tonic::Status::failed_precondition("generation mismatch"),
+ );
+ assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert!(!err.is_temporary());
+
+ let state_conflict = ErrorContext::new(ServiceOperation("QueryWriteStatus"));
+ let err = parse_status(
+ state_conflict,
+ tonic::Status::failed_precondition("upload is not active"),
+ );
+ assert_eq!(err.kind(), ErrorKind::Conflict);
+ assert!(!err.is_temporary());
+
+ let concurrent = ErrorContext::new(ServiceOperation("RewriteObject"));
+ let err = parse_status(concurrent, tonic::Status::aborted("concurrent update"));
+ assert_eq!(err.kind(), ErrorKind::Conflict);
+ assert!(!err.is_temporary());
+ }
}
diff --git a/core/services/gcs-grpc/src/deleter.rs b/core/services/gcs-grpc/src/deleter.rs
index d9e8f097a..e666fe0d4 100644
--- a/core/services/gcs-grpc/src/deleter.rs
+++ b/core/services/gcs-grpc/src/deleter.rs
@@ -20,7 +20,7 @@ use std::sync::Arc;
use opendal_core::raw::*;
use opendal_core::*;
-use crate::core::{GcsGrpcCore, parse_generation, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_generation, parse_status};
use crate::generated::google::storage::v2::DeleteObjectRequest;
pub(super) struct GcsGrpcDeleter {
@@ -52,7 +52,10 @@ impl oio::Delete for GcsGrpcDeleter {
match self.core.client().delete_object(request).await {
Ok(_) => Ok(()),
Err(status) if status.code() == tonic::Code::NotFound => Ok(()),
- Err(status) => Err(parse_status(status)),
+ Err(status) => Err(parse_status(
+ ErrorContext::new(ServiceOperation("DeleteObject")),
+ status,
+ )),
}
}
diff --git a/core/services/gcs-grpc/src/lister.rs b/core/services/gcs-grpc/src/lister.rs
index 71df01c19..983ba4143 100644
--- a/core/services/gcs-grpc/src/lister.rs
+++ b/core/services/gcs-grpc/src/lister.rs
@@ -21,7 +21,7 @@ use std::sync::Arc;
use opendal_core::raw::*;
use opendal_core::*;
-use crate::core::{GcsGrpcCore, parse_object, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_object, parse_status};
use crate::generated::google::storage::v2::ListObjectsRequest;
pub(super) struct GcsGrpcLister {
@@ -83,7 +83,9 @@ impl GcsGrpcLister {
.client()
.list_objects(request)
.await
- .map_err(parse_status)?
+ .map_err(|status| {
+ parse_status(ErrorContext::new(ServiceOperation("ListObjects")), status)
+ })?
.into_inner();
self.page_token = response.next_page_token;
self.done = self.page_token.is_empty();
diff --git a/core/services/gcs-grpc/src/reader.rs b/core/services/gcs-grpc/src/reader.rs
index c167f0994..d37bfbc4d 100644
--- a/core/services/gcs-grpc/src/reader.rs
+++ b/core/services/gcs-grpc/src/reader.rs
@@ -20,7 +20,7 @@ use std::sync::Arc;
use opendal_core::raw::*;
use opendal_core::*;
-use crate::core::{GcsGrpcCore, parse_generation, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_generation, parse_status};
use crate::generated::google::storage::v2::{ReadObjectRequest, ReadObjectResponse};
pub(super) struct GcsGrpcReader {
@@ -70,12 +70,15 @@ impl oio::Read for GcsGrpcReader {
.client()
.read_object(request)
.await
- .map_err(parse_status)?;
+ .map_err(|status| {
+ parse_status(ErrorContext::new(ServiceOperation("ReadObject")), status)
+ })?;
Ok((
RpRead::default(),
Box::new(GcsGrpcReadStream {
stream: response.into_inner(),
done: false,
+ error_context: ErrorContext::new(ServiceOperation("ReadObject")),
}),
))
}
@@ -105,6 +108,7 @@ fn parse_read_range(range: BytesRange) -> Result<(i64, i64)> {
struct GcsGrpcReadStream {
stream: tonic::Streaming<ReadObjectResponse>,
done: bool,
+ error_context: ErrorContext,
}
impl oio::ReadStream for GcsGrpcReadStream {
@@ -113,7 +117,12 @@ impl oio::ReadStream for GcsGrpcReadStream {
return Ok(Buffer::new());
}
loop {
- match self.stream.message().await.map_err(parse_status)? {
+ match self
+ .stream
+ .message()
+ .await
+ .map_err(|status| parse_status(self.error_context, status))?
+ {
Some(response) => {
if let Some(data) = response.checksummed_data
&& !data.content.is_empty()
diff --git a/core/services/gcs-grpc/src/writer.rs b/core/services/gcs-grpc/src/writer.rs
index e401c3fb7..2729bde27 100644
--- a/core/services/gcs-grpc/src/writer.rs
+++ b/core/services/gcs-grpc/src/writer.rs
@@ -22,7 +22,7 @@ use futures::stream;
use opendal_core::raw::*;
use opendal_core::*;
-use crate::core::{GcsGrpcCore, parse_object, parse_status};
+use crate::core::{ErrorContext, GcsGrpcCore, parse_object, parse_status};
use crate::generated::google::storage::v2::*;
const WRITE_CHUNK_ALIGNMENT: usize = 256 * 1024;
@@ -60,6 +60,10 @@ impl GcsGrpcWriter {
}
}
+ fn error_context(&self, service_operation: ServiceOperation) -> ErrorContext {
+ ErrorContext::new(service_operation).with_if_not_exists(self.args.if_not_exists())
+ }
+
fn ensure_open(&self) -> Result<()> {
match self.state {
GcsGrpcWriterState::Open => Ok(()),
@@ -121,7 +125,12 @@ impl GcsGrpcWriter {
.client()
.start_resumable_write(request)
.await
- .map_err(parse_status)?
+ .map_err(|status| {
+ parse_status(
+ self.error_context(ServiceOperation("StartResumableWrite")),
+ status,
+ )
+ })?
.into_inner();
if response.upload_id.is_empty() {
return Err(Error::new(
@@ -146,7 +155,9 @@ impl GcsGrpcWriter {
.client()
.write_object(request)
.await
- .map_err(parse_status)?
+ .map_err(|status| {
+ parse_status(self.error_context(ServiceOperation("WriteObject")), status)
+ })?
.into_inner();
match response.write_status {
Some(write_object_response::WriteStatus::Resource(object)) => Ok(object),
@@ -254,7 +265,9 @@ impl GcsGrpcWriter {
.client()
.write_object(request)
.await
- .map_err(parse_status)?
+ .map_err(|status| {
+ parse_status(self.error_context(ServiceOperation("WriteObject")), status)
+ })?
.into_inner();
match response.write_status {
@@ -300,7 +313,12 @@ impl GcsGrpcWriter {
Err(status) if status.code() == tonic::Code::NotFound => {
return Ok(QueryOutcome::NotFound);
}
- Err(status) => return Err(parse_status(status)),
+ Err(status) => {
+ return Err(parse_status(
+ ErrorContext::new(ServiceOperation("QueryWriteStatus")),
+ status,
+ ));
+ }
};
match response.write_status {
Some(query_write_status_response::WriteStatus::PersistedSize(size)) => {
@@ -435,7 +453,12 @@ impl oio::Write for GcsGrpcWriter {
match self.core.client().cancel_resumable_write(request).await {
Ok(_) => {}
Err(status) if status.code() == tonic::Code::NotFound => {}
- Err(status) => return Err(parse_status(status)),
+ Err(status) => {
+ return Err(parse_status(
+ ErrorContext::new(ServiceOperation("CancelResumableWrite")),
+ status,
+ ));
+ }
}
}
self.buffer = None;
diff --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs
index c5857220f..1a498a245 100644
--- a/core/services/gcs/src/backend.rs
+++ b/core/services/gcs/src/backend.rs
@@ -476,10 +476,7 @@ 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_version_match(args.if_version_match().is_some())
- .with_if_version_not_match(args.if_version_not_match().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = self.core.gcs_get_object_metadata(ctx, path, &args).await?;
if !resp.status().is_success() {
diff --git a/core/services/gcs/src/composer.rs b/core/services/gcs/src/composer.rs
index ed1f06ea9..ece0a457c 100644
--- a/core/services/gcs/src/composer.rs
+++ b/core/services/gcs/src/composer.rs
@@ -143,11 +143,7 @@ impl GcsComposer {
if !resp.status().is_success() {
return Err(parse_error(
ErrorContext::new(ServiceOperation("ComposeObject"))
- .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()),
+ .with_caller_condition(self.args.is_conditional()),
resp,
));
}
@@ -297,11 +293,16 @@ impl GcsComposeTask {
if !resp.status().is_success() {
let err = parse_error(
ErrorContext::new(ServiceOperation("ComposeObject"))
- .with_if_not_exists(self.args.if_not_exists())
- .with_if_version_match(self.args.if_version_match().is_some()),
+ .with_caller_condition(self.args.is_conditional())
+ .with_internal_condition(self.token.is_some()),
resp,
);
- if err.kind() == ErrorKind::ConditionNotMatch
+ let recoverable_condition = if self.token.is_some() {
+ ErrorKind::Conflict
+ } else {
+ ErrorKind::ConditionNotMatch
+ };
+ if err.kind() == recoverable_condition
&& let Some(source) = self.recover_completed_intermediate().await?
{
return Ok(source);
diff --git a/core/services/gcs/src/copier.rs b/core/services/gcs/src/copier.rs
index 6172e7421..bec5347b5 100644
--- a/core/services/gcs/src/copier.rs
+++ b/core/services/gcs/src/copier.rs
@@ -91,11 +91,7 @@ impl oio::Copy for GcsCopier {
if !resp.status().is_success() {
return Err(parse_error(
ErrorContext::new(ServiceOperation("RewriteObject"))
- .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()),
+ .with_caller_condition(self.args.is_conditional()),
resp,
));
}
diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs
index 6be60d930..b2c3eefec 100644
--- a/core/services/gcs/src/core.rs
+++ b/core/services/gcs/src/core.rs
@@ -1628,16 +1628,11 @@ mod tests {
parse_error(ctx, resp).kind()
};
- for ctx in [
- ErrorContext::new(ServiceOperation("GetObject")).with_if_version_match(true),
- ErrorContext::new(ServiceOperation("GetObject")).with_if_version_not_match(true),
- ] {
- assert_eq!(parse_missing(ctx), ErrorKind::NotFound);
- }
+ let read_ctx = ErrorContext::new(ServiceOperation("GetObject")).with_caller_condition(true);
+ assert_eq!(parse_missing(read_ctx), ErrorKind::NotFound);
- let delete_ctx = ErrorContext::new(ServiceOperation("DeleteObject"))
- .with_if_version_match(true)
- .with_delete();
+ let delete_ctx =
+ ErrorContext::new(ServiceOperation("DeleteObject")).with_delete_match_condition(true);
assert_eq!(parse_missing(delete_ctx), ErrorKind::ConditionNotMatch);
}
@@ -1845,64 +1840,36 @@ mod tests {
#[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,
- is_delete: bool,
+ caller_condition: bool,
+ delete_match_condition: bool,
+ internal_condition: bool,
}
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,
- is_delete: false,
+ caller_condition: false,
+ delete_match_condition: false,
+ internal_condition: false,
}
}
- pub const fn with_if_match(mut self, if_match: bool) -> Self {
- self.if_match = if_match;
- self
- }
-
- pub const fn with_if_none_match(mut self, if_none_match: bool) -> Self {
- self.if_none_match = if_none_match;
- self
- }
-
- pub const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
- self.if_not_exists = if_not_exists;
- self
- }
-
- pub const fn with_if_version_match(mut self, if_version_match: bool) -> Self {
- self.if_version_match = if_version_match;
+ pub const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
self
}
- 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;
+ pub const fn with_delete_match_condition(mut self, delete_match_condition: bool) -> Self {
+ self.caller_condition = self.caller_condition || delete_match_condition;
+ self.delete_match_condition = delete_match_condition;
self
}
- pub const fn with_delete(mut self) -> Self {
- self.is_delete = true;
+ pub const fn with_internal_condition(mut self, internal_condition: bool) -> Self {
+ self.internal_condition = internal_condition;
self
}
-
- const fn has_condition(self) -> bool {
- self.if_match
- || self.if_none_match
- || self.if_not_exists
- || self.if_version_match
- || self.if_version_not_match
- }
}
#[derive(Default, Debug, Deserialize)]
@@ -1937,12 +1904,15 @@ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let gcs_error = de::from_slice::<GcsErrorResponse>(&bs).ok();
let (mut kind, mut retryable) = match parts.status {
- StatusCode::NOT_FOUND if ctx.is_delete && (ctx.if_match || ctx.if_version_match) => {
+ StatusCode::NOT_FOUND if ctx.delete_match_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() => {
+ StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.internal_condition => {
+ (ErrorKind::Conflict, false)
+ }
+ StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
@@ -1965,7 +1935,13 @@ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
.iter()
.any(|detail| detail.reason == "conditionNotMet")
{
- (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ if ctx.internal_condition {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ } else if ctx.caller_condition {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ } else {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
} else if gcs_error
.error
.errors
@@ -2000,6 +1976,43 @@ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
err
}
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+
+ fn condition_not_met(ctx: ErrorContext) -> Error {
+ let body = Buffer::from(
+ r#"{"error":{"code":412,"message":"condition failed","errors":[{"reason":"conditionNotMet"}]}}"#,
+ );
+ let resp = Response::builder()
+ .status(StatusCode::PRECONDITION_FAILED)
+ .body(body)
+ .expect("response must build");
+ parse_error(ctx, resp)
+ }
+
+ #[test]
+ fn condition_not_met_uses_operation_context() {
+ let caller =
+ ErrorContext::new(ServiceOperation("ComposeObject")).with_caller_condition(true);
+ assert_eq!(
+ condition_not_met(caller).kind(),
+ ErrorKind::ConditionNotMatch
+ );
+
+ let internal = ErrorContext::new(ServiceOperation("ComposeObject"))
+ .with_caller_condition(true)
+ .with_internal_condition(true);
+ assert_eq!(condition_not_met(internal).kind(), ErrorKind::Conflict);
+
+ let no_condition = ErrorContext::new(ServiceOperation("ComposeObject"));
+ assert_eq!(
+ condition_not_met(no_condition).kind(),
+ ErrorKind::Unexpected
+ );
+ }
+}
+
mod uri {
use percent_encoding::AsciiSet;
use percent_encoding::NON_ALPHANUMERIC;
diff --git a/core/services/gcs/src/deleter.rs b/core/services/gcs/src/deleter.rs
index ee5c14862..3d2273a5c 100644
--- a/core/services/gcs/src/deleter.rs
+++ b/core/services/gcs/src/deleter.rs
@@ -40,11 +40,10 @@ impl oio::BatchDelete for GcsDeleter {
async fn delete_once(&self, path: String, args: OpDelete) -> Result<()> {
let resp = self.core.gcs_delete_object(&self.ctx, &path, &args).await?;
let error_ctx = 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())
- .with_delete();
+ .with_caller_condition(args.is_conditional())
+ .with_delete_match_condition(
+ args.if_match().is_some() || args.if_version_match().is_some(),
+ );
if resp.status().is_success()
|| (resp.status() == StatusCode::NOT_FOUND
@@ -89,11 +88,10 @@ impl oio::BatchDelete for GcsDeleter {
// TODO: maybe we can take it directly?
let (path, op) = batch[i].clone();
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())
- .with_delete();
+ .with_caller_condition(op.is_conditional())
+ .with_delete_match_condition(
+ op.if_match().is_some() || op.if_version_match().is_some(),
+ );
if resp.status().is_success()
|| (resp.status() == StatusCode::NOT_FOUND
diff --git a/core/services/gcs/src/reader.rs b/core/services/gcs/src/reader.rs
index e2c71a609..50031dcc7 100644
--- a/core/services/gcs/src/reader.rs
+++ b/core/services/gcs/src/reader.rs
@@ -56,10 +56,7 @@ impl oio::StreamRead for GcsReader {
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_version_match(args.if_version_match().is_some())
- .with_if_version_not_match(args.if_version_not_match().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = backend
.core
.gcs_get_object(&self.ctx, path, range, &args)
diff --git a/core/services/gcs/src/writer.rs b/core/services/gcs/src/writer.rs
index a51b77b54..4e910421c 100644
--- a/core/services/gcs/src/writer.rs
+++ b/core/services/gcs/src/writer.rs
@@ -75,7 +75,7 @@ impl oio::MultipartWrite for GcsWriter {
}
_ => Err(parse_error(
ErrorContext::new(ServiceOperation("InsertObject"))
- .with_if_not_exists(self.op.if_not_exists()),
+ .with_caller_condition(self.op.is_conditional()),
resp,
)),
}
@@ -238,11 +238,7 @@ impl GcsConditionalWriter {
if !resp.status().is_success() {
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()),
+ .with_caller_condition(self.op.is_conditional()),
resp,
));
}
@@ -354,11 +350,7 @@ impl GcsConditionalWriter {
Err(parse_error(
ErrorContext::new(ServiceOperation("UploadResumableChunk"))
- .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()),
+ .with_caller_condition(self.op.is_conditional()),
resp,
))
}
@@ -450,11 +442,7 @@ impl oio::Write for GcsConditionalWriter {
let err = parse_error(
ErrorContext::new(ServiceOperation("UploadResumableChunk"))
- .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()),
+ .with_caller_condition(self.op.is_conditional()),
resp,
);
if !err.is_temporary() {
diff --git a/core/services/hf/src/core.rs b/core/services/hf/src/core.rs
index c8f247752..a8ccb3e5b 100644
--- a/core/services/hf/src/core.rs
+++ b/core/services/hf/src/core.rs
@@ -1534,7 +1534,7 @@ mod tests {
let err = parse_error(ErrorContext::new(ServiceOperation("Test")), parts);
- assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert_eq!(err.kind(), ErrorKind::Conflict);
assert!(err.is_temporary());
}
@@ -1549,7 +1549,7 @@ mod tests {
let err = parse_error(ErrorContext::new(ServiceOperation("Test")), parts);
- assert_eq!(err.kind(), ErrorKind::ConditionNotMatch);
+ assert_eq!(err.kind(), ErrorKind::Unexpected);
assert!(!err.is_temporary());
}
}
@@ -1591,7 +1591,8 @@ pub(crate) fn parse_error(ctx: ErrorContext, parts: http::response::Parts) -> Er
let (kind, retryable) = match parts.status {
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
StatusCode::UNAUTHORIZED | StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::PRECONDITION_FAILED => (ErrorKind::ConditionNotMatch, branch_updated_conflict),
+ StatusCode::PRECONDITION_FAILED if branch_updated_conflict => (ErrorKind::Conflict, true),
+ StatusCode::PRECONDITION_FAILED => (ErrorKind::Unexpected, false),
StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
StatusCode::INTERNAL_SERVER_ERROR
| StatusCode::BAD_GATEWAY
diff --git a/core/services/onedrive/src/core.rs b/core/services/onedrive/src/core.rs
index 81d3c7b38..acd2bb9f0 100644
--- a/core/services/onedrive/src/core.rs
+++ b/core/services/onedrive/src/core.rs
@@ -199,7 +199,8 @@ impl OneDriveCore {
let response = ctx.http_transport().send(request).await?;
if !response.status().is_success() {
return Err(parse_error(
- ErrorContext::new(ServiceOperation("GetItem")),
+ ErrorContext::new(ServiceOperation("GetItem"))
+ .with_caller_condition(args.is_conditional()),
response,
));
}
@@ -942,6 +943,55 @@ mod tests {
assert_eq!(err.kind(), ErrorKind::Unexpected);
assert!(err.is_temporary());
}
+
+ #[test]
+ fn conflict_classification_uses_native_code() {
+ let parse_conflict = |code: &str, operation| {
+ let body = Buffer::from(format!(r#"{{"error":{{"code":"{code}"}}}}"#));
+ let response = Response::builder()
+ .status(StatusCode::CONFLICT)
+ .body(body)
+ .expect("response must build");
+ parse_error(ErrorContext::new(operation), response)
+ };
+
+ let concurrent = parse_conflict(
+ "Directory_ConcurrencyViolation",
+ ServiceOperation("CreateFolder"),
+ );
+ assert_eq!(concurrent.kind(), ErrorKind::Conflict);
+ assert!(concurrent.is_temporary());
+
+ let name_conflict = parse_conflict("nameAlreadyExists", ServiceOperation("UploadFragment"));
+ assert_eq!(name_conflict.kind(), ErrorKind::Conflict);
+ assert!(!name_conflict.is_temporary());
+
+ assert_eq!(
+ parse_conflict("unknownConflict", ServiceOperation("MoveItem")).kind(),
+ ErrorKind::Unexpected
+ );
+ }
+
+ #[test]
+ fn precondition_failed_requires_caller_condition() {
+ let response = || {
+ Response::builder()
+ .status(StatusCode::PRECONDITION_FAILED)
+ .body(Buffer::new())
+ .expect("response must build")
+ };
+ let caller =
+ ErrorContext::new(ServiceOperation("UploadContent")).with_caller_condition(true);
+ assert_eq!(
+ parse_error(caller, response()).kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ let no_condition = ErrorContext::new(ServiceOperation("UploadFragment"));
+ assert_eq!(
+ parse_error(no_condition, response()).kind(),
+ ErrorKind::Unexpected
+ );
+ }
}
use crate::graph_model::GraphErrorResponse;
@@ -950,11 +1000,20 @@ use crate::graph_model::GraphErrorResponse;
#[derive(Clone, Copy, Debug)]
pub(crate) struct ErrorContext {
service_operation: ServiceOperation,
+ caller_condition: bool,
}
impl ErrorContext {
pub(crate) const fn new(service_operation: ServiceOperation) -> Self {
- Self { service_operation }
+ Self {
+ service_operation,
+ caller_condition: false,
+ }
+ }
+
+ pub(crate) const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
+ self
}
}
@@ -963,26 +1022,28 @@ pub(crate) fn parse_error(ctx: ErrorContext, response: Response<Buffer>) -> Erro
let (parts, body) = response.into_parts();
let bs = body.to_bytes();
- let consistency_lag = parts.status == StatusCode::BAD_REQUEST
- && serde_json::from_slice::<GraphErrorResponse>(&bs)
- .is_ok_and(|response| response.error.code == "invalidRequest");
+ let graph_error = serde_json::from_slice::<GraphErrorResponse>(&bs).ok();
+ let graph_code = graph_error
+ .as_ref()
+ .map(|response| response.error.code.as_str());
+ let consistency_lag =
+ parts.status == StatusCode::BAD_REQUEST && graph_code == Some("invalidRequest");
let (kind, mut retryable) = match parts.status {
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
- // The OneDrive service replaces resources.
- // However, the Onedrive doesn't have Strong Read-After-Write properties,
- // the concurrent requests to create directories might result in errors.
- //
- // Running behavior tests can yield HTTP 409 Conflict because of the consistency guarantee.
- //
- // Read more about `REPLACE_EXISTING_ITEM_WHEN_CONFLICT` in `graph_model.rs`.
- StatusCode::CONFLICT => (ErrorKind::AlreadyExists, true),
+ StatusCode::CONFLICT if graph_code == Some("Directory_ConcurrencyViolation") => {
+ (ErrorKind::Conflict, true)
+ }
+ StatusCode::CONFLICT if graph_code == Some("nameAlreadyExists") => {
+ (ErrorKind::Conflict, false)
+ }
+ StatusCode::CONFLICT => (ErrorKind::Unexpected, false),
StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
StatusCode::INTERNAL_SERVER_ERROR
| StatusCode::BAD_GATEWAY
| StatusCode::SERVICE_UNAVAILABLE
| StatusCode::GATEWAY_TIMEOUT => (ErrorKind::Unexpected, true),
- StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED => {
+ StatusCode::NOT_MODIFIED | StatusCode::PRECONDITION_FAILED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
_ => (ErrorKind::Unexpected, false),
diff --git a/core/services/onedrive/src/reader.rs b/core/services/onedrive/src/reader.rs
index 18429aac2..cb4635eae 100644
--- a/core/services/onedrive/src/reader.rs
+++ b/core/services/onedrive/src/reader.rs
@@ -75,7 +75,8 @@ impl oio::StreamRead for OnedriveReader {
let (part, mut body) = response.into_parts();
let buf = body.to_buffer().await?;
return Err(parse_error(
- ErrorContext::new(ServiceOperation("DownloadContent")),
+ ErrorContext::new(ServiceOperation("DownloadContent"))
+ .with_caller_condition(args.is_conditional()),
Response::from_parts(part, buf),
));
}
diff --git a/core/services/onedrive/src/writer.rs b/core/services/onedrive/src/writer.rs
index cc7a96bf8..6cff2e478 100644
--- a/core/services/onedrive/src/writer.rs
+++ b/core/services/onedrive/src/writer.rs
@@ -92,7 +92,8 @@ impl OneDriveWriter {
Ok(meta.build())
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("UploadContent")),
+ ErrorContext::new(ServiceOperation("UploadContent"))
+ .with_caller_condition(self.op.is_conditional()),
response,
)),
}
@@ -177,7 +178,8 @@ impl OneDriveWriter {
Ok(result)
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("CreateUploadSession")),
+ ErrorContext::new(ServiceOperation("CreateUploadSession"))
+ .with_caller_condition(self.op.is_conditional()),
response,
)),
}
diff --git a/core/services/oss/src/backend.rs b/core/services/oss/src/backend.rs
index 1dd9979f9..4e5187812 100644
--- a/core/services/oss/src/backend.rs
+++ b/core/services/oss/src/backend.rs
@@ -685,7 +685,8 @@ impl Service for OssBackend {
Ok(RpStat::new(meta.build()))
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("HeadObject")),
+ ErrorContext::new(ServiceOperation("HeadObject"))
+ .with_caller_condition(args.is_conditional()),
resp,
)),
}
diff --git a/core/services/oss/src/core.rs b/core/services/oss/src/core.rs
index 3c99cfa2c..256996130 100644
--- a/core/services/oss/src/core.rs
+++ b/core/services/oss/src/core.rs
@@ -1199,11 +1199,28 @@ struct OssError {
#[derive(Clone, Copy, Debug)]
pub(crate) struct ErrorContext {
service_operation: ServiceOperation,
+ caller_condition: bool,
+ if_not_exists: bool,
}
impl ErrorContext {
pub(crate) const fn new(service_operation: ServiceOperation) -> Self {
- Self { service_operation }
+ Self {
+ service_operation,
+ caller_condition: false,
+ if_not_exists: false,
+ }
+ }
+
+ pub(crate) const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
+ self
+ }
+
+ pub(crate) const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
+ self.caller_condition = self.caller_condition || if_not_exists;
+ self.if_not_exists = if_not_exists;
+ self
}
}
@@ -1212,10 +1229,12 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (parts, body) = resp.into_parts();
let bs = body.to_bytes();
- let (kind, retryable) = match parts.status {
+ let oss_error = de::from_reader::<_, OssError>(bs.clone().reader()).ok();
+
+ let (mut kind, mut retryable) = match parts.status {
StatusCode::NOT_FOUND => (ErrorKind::NotFound, false),
StatusCode::FORBIDDEN => (ErrorKind::PermissionDenied, false),
- StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED | StatusCode::CONFLICT => {
+ StatusCode::PRECONDITION_FAILED | StatusCode::NOT_MODIFIED if ctx.caller_condition => {
(ErrorKind::ConditionNotMatch, false)
}
StatusCode::TOO_MANY_REQUESTS => (ErrorKind::RateLimited, true),
@@ -1226,9 +1245,31 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
_ => (ErrorKind::Unexpected, false),
};
- let message = match de::from_reader::<_, OssError>(bs.clone().reader()) {
- Ok(oss_err) => format!("{oss_err:?}"),
- Err(_) => String::from_utf8_lossy(&bs).into_owned(),
+ if let Some(oss_error) = &oss_error {
+ match oss_error.code.as_str() {
+ "PreconditionFailed" if ctx.caller_condition => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ }
+ "FileAlreadyExists" if ctx.if_not_exists => {
+ (kind, retryable) = (ErrorKind::ConditionNotMatch, false);
+ }
+ "FileAlreadyExists" | "FileImmutable" => {
+ (kind, retryable) = (ErrorKind::Conflict, false);
+ }
+ _ if matches!(
+ parts.status,
+ StatusCode::CONFLICT | StatusCode::PRECONDITION_FAILED
+ ) =>
+ {
+ (kind, retryable) = (ErrorKind::Unexpected, false);
+ }
+ _ => {}
+ }
+ }
+
+ let message = match oss_error {
+ Some(oss_err) => format!("{oss_err:?}"),
+ None => String::from_utf8_lossy(&bs).into_owned(),
};
let mut err = Error::new(kind, message);
@@ -1242,3 +1283,51 @@ pub(crate) fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
err
}
+
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+
+ fn parse_oss_error(ctx: ErrorContext, status: StatusCode, code: &str) -> Error {
+ let body = Buffer::from(format!(
+ "<Error><Code>{code}</Code><Message>test</Message></Error>"
+ ));
+ let resp = Response::builder()
+ .status(status)
+ .body(body)
+ .expect("response must build");
+ parse_error(ctx, resp)
+ }
+
+ #[test]
+ fn conflict_classification_uses_native_code_and_condition() {
+ let conditional = ErrorContext::new(ServiceOperation("PutObject")).with_if_not_exists(true);
+ assert_eq!(
+ parse_oss_error(conditional, StatusCode::CONFLICT, "FileAlreadyExists").kind(),
+ ErrorKind::ConditionNotMatch
+ );
+ assert_eq!(
+ parse_oss_error(
+ conditional,
+ StatusCode::PRECONDITION_FAILED,
+ "PreconditionFailed"
+ )
+ .kind(),
+ ErrorKind::ConditionNotMatch
+ );
+
+ let unconditional = ErrorContext::new(ServiceOperation("PutObject"));
+ assert_eq!(
+ parse_oss_error(unconditional, StatusCode::CONFLICT, "FileAlreadyExists").kind(),
+ ErrorKind::Conflict
+ );
+ assert_eq!(
+ parse_oss_error(unconditional, StatusCode::CONFLICT, "FileImmutable").kind(),
+ ErrorKind::Conflict
+ );
+ assert_eq!(
+ parse_oss_error(unconditional, StatusCode::CONFLICT, "UnknownConflict").kind(),
+ ErrorKind::Unexpected
+ );
+ }
+}
diff --git a/core/services/oss/src/reader.rs b/core/services/oss/src/reader.rs
index a0c1890c5..bce1434fb 100644
--- a/core/services/oss/src/reader.rs
+++ b/core/services/oss/src/reader.rs
@@ -67,7 +67,8 @@ impl oio::StreamRead for OssReader {
let (part, mut body) = resp.into_parts();
let buf = body.to_buffer().await?;
return Err(parse_error(
- ErrorContext::new(ServiceOperation("GetObject")),
+ ErrorContext::new(ServiceOperation("GetObject"))
+ .with_caller_condition(args.is_conditional()),
Response::from_parts(part, buf),
));
}
diff --git a/core/services/oss/src/writer.rs b/core/services/oss/src/writer.rs
index 91593a0c6..071bc3fe2 100644
--- a/core/services/oss/src/writer.rs
+++ b/core/services/oss/src/writer.rs
@@ -77,7 +77,8 @@ impl oio::MultipartWrite for OssWriter {
match status {
StatusCode::CREATED | StatusCode::OK => Ok(meta),
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("PutObject")),
+ ErrorContext::new(ServiceOperation("PutObject"))
+ .with_if_not_exists(self.op.if_not_exists()),
resp,
)),
}
@@ -102,7 +103,8 @@ impl oio::MultipartWrite for OssWriter {
Ok(result.upload_id)
}
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("InitiateMultipartUpload")),
+ ErrorContext::new(ServiceOperation("InitiateMultipartUpload"))
+ .with_if_not_exists(self.op.if_not_exists()),
resp,
)),
}
@@ -184,7 +186,8 @@ impl oio::MultipartWrite for OssWriter {
match status {
StatusCode::OK => Ok(meta),
_ => Err(parse_error(
- ErrorContext::new(ServiceOperation("CompleteMultipartUpload")),
+ ErrorContext::new(ServiceOperation("CompleteMultipartUpload"))
+ .with_if_not_exists(self.op.if_not_exists()),
resp,
)),
}
diff --git a/core/services/s3/src/backend.rs b/core/services/s3/src/backend.rs
index aa0e74a45..14e0354a3 100644
--- a/core/services/s3/src/backend.rs
+++ b/core/services/s3/src/backend.rs
@@ -1201,10 +1201,7 @@ impl Service for S3Backend {
async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result<RpStat> {
let error_ctx = ErrorContext::new(ServiceOperation("HeadObject"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some())
- .with_if_modified_since(args.if_modified_since().is_some())
- .with_if_unmodified_since(args.if_unmodified_since().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = self.core.s3_head_object(ctx, path, args).await?;
let status = resp.status();
diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs
index 8e7344804..777b0fb2f 100644
--- a/core/services/s3/src/copier.rs
+++ b/core/services/s3/src/copier.rs
@@ -106,9 +106,8 @@ impl S3Copier {
source_if_match: bool,
) -> ErrorContext {
ErrorContext::new(service_operation)
- .with_if_match(self.args.if_match().is_some() || source_if_match)
- .with_if_none_match(self.args.if_none_match().is_some())
- .with_if_not_exists(self.args.if_not_exists())
+ .with_caller_condition(self.args.is_conditional())
+ .with_source_if_match(source_if_match)
}
fn copy_source(&self) -> Result<S3CopySource<'_>> {
diff --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs
index 3106141c5..d4eb10cb1 100644
--- a/core/services/s3/src/core.rs
+++ b/core/services/s3/src/core.rs
@@ -2151,57 +2151,28 @@ pub struct S3Error {
#[derive(Clone, Copy, Debug)]
pub struct ErrorContext {
service_operation: ServiceOperation,
- if_match: bool,
- if_none_match: bool,
- if_modified_since: bool,
- if_unmodified_since: bool,
- if_not_exists: bool,
+ caller_condition: bool,
+ source_if_match: bool,
}
impl ErrorContext {
pub const fn new(service_operation: ServiceOperation) -> Self {
Self {
service_operation,
- if_match: false,
- if_none_match: false,
- if_modified_since: false,
- if_unmodified_since: false,
- if_not_exists: false,
+ caller_condition: false,
+ source_if_match: false,
}
}
- pub const fn with_if_match(mut self, if_match: bool) -> Self {
- self.if_match = if_match;
+ pub const fn with_caller_condition(mut self, caller_condition: bool) -> Self {
+ self.caller_condition = caller_condition;
self
}
- pub const fn with_if_none_match(mut self, if_none_match: bool) -> Self {
- self.if_none_match = if_none_match;
+ pub const fn with_source_if_match(mut self, source_if_match: bool) -> Self {
+ self.source_if_match = source_if_match;
self
}
-
- pub const fn with_if_modified_since(mut self, if_modified_since: bool) -> Self {
- self.if_modified_since = if_modified_since;
- self
- }
-
- pub const fn with_if_unmodified_since(mut self, if_unmodified_since: bool) -> Self {
- self.if_unmodified_since = if_unmodified_since;
- self
- }
-
- pub const fn with_if_not_exists(mut self, if_not_exists: bool) -> Self {
- self.if_not_exists = if_not_exists;
- self
- }
-
- const fn has_condition(self) -> bool {
- self.if_match
- || self.if_none_match
- || self.if_modified_since
- || self.if_unmodified_since
- || self.if_not_exists
- }
}
/// Parse error response into Error.
@@ -2212,7 +2183,7 @@ pub fn parse_error(ctx: ErrorContext, resp: Response<Buffer>) -> Error {
let (mut kind, mut retryable) = match parts.status.as_u16() {
403 => (ErrorKind::PermissionDenied, false),
404 => (ErrorKind::NotFound, false),
- 304 | 412 if ctx.has_condition() => (ErrorKind::ConditionNotMatch, false),
+ 304 | 412 if ctx.caller_condition => (ErrorKind::ConditionNotMatch, false),
// Service like R2 could return 499 error with a message like:
// Client Disconnect, we should retry it.
499 => (ErrorKind::Unexpected, true),
@@ -2295,9 +2266,12 @@ pub fn parse_s3_error_code(ctx: ErrorContext, code: &str) -> Option<(ErrorKind,
| "ExceedBucketQPSLimit"
| "ExceedBucketRateLimit" => Some((ErrorKind::RateLimited, true)),
"InvalidRange" => Some((ErrorKind::RangeNotSatisfied, false)),
- "PreconditionFailed" | "412 Precondition Failed" => {
+ "PreconditionFailed" | "412 Precondition Failed" if ctx.caller_condition => {
Some((ErrorKind::ConditionNotMatch, false))
}
+ "PreconditionFailed" | "412 Precondition Failed" if ctx.source_if_match => {
+ Some((ErrorKind::Conflict, false))
+ }
"ConditionalRequestConflict" => Some((
ErrorKind::Conflict,
ctx.service_operation == ServiceOperation("PutObject"),
@@ -2305,3 +2279,32 @@ pub fn parse_s3_error_code(ctx: ErrorContext, code: &str) -> Option<(ErrorKind,
_ => None,
}
}
+
+#[cfg(test)]
+mod error_tests {
+ use super::*;
+
+ #[test]
+ fn precondition_failed_uses_operation_context() {
+ let caller_condition = ErrorContext::new(ServiceOperation("CopyObject"))
+ .with_caller_condition(true)
+ .with_source_if_match(true);
+ assert_eq!(
+ parse_s3_error_code(caller_condition, "PreconditionFailed"),
+ Some((ErrorKind::ConditionNotMatch, false))
+ );
+
+ let internal_source_condition =
+ ErrorContext::new(ServiceOperation("CopyObject")).with_source_if_match(true);
+ assert_eq!(
+ parse_s3_error_code(internal_source_condition, "PreconditionFailed"),
+ Some((ErrorKind::Conflict, false))
+ );
+
+ let no_condition = ErrorContext::new(ServiceOperation("CopyObject"));
+ assert_eq!(
+ parse_s3_error_code(no_condition, "PreconditionFailed"),
+ None
+ );
+ }
+}
diff --git a/core/services/s3/src/deleter.rs b/core/services/s3/src/deleter.rs
index 3789603a3..76e656232 100644
--- a/core/services/s3/src/deleter.rs
+++ b/core/services/s3/src/deleter.rs
@@ -46,7 +46,7 @@ impl oio::BatchDelete for S3Deleter {
}
let error_ctx = ErrorContext::new(ServiceOperation("DeleteObject"))
- .with_if_match(args.if_match().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = self.core.s3_delete_object(&self.ctx, &path, &args).await?;
let status = resp.status();
@@ -113,7 +113,8 @@ impl oio::BatchDelete for S3Deleter {
}
fn parse_delete_objects_result_error(err: DeleteObjectsResultError, conditional: bool) -> Error {
- let error_ctx = ErrorContext::new(ServiceOperation("DeleteObjects")).with_if_match(conditional);
+ let error_ctx =
+ ErrorContext::new(ServiceOperation("DeleteObjects")).with_caller_condition(conditional);
let (kind, retryable) =
parse_s3_error_code(error_ctx, err.code.as_str()).unwrap_or((ErrorKind::Unexpected, false));
let kind = if conditional && kind == ErrorKind::NotFound {
diff --git a/core/services/s3/src/reader.rs b/core/services/s3/src/reader.rs
index 9db1f3b43..43b19b28c 100644
--- a/core/services/s3/src/reader.rs
+++ b/core/services/s3/src/reader.rs
@@ -53,10 +53,7 @@ impl oio::StreamRead for S3Reader {
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_modified_since(args.if_modified_since().is_some())
- .with_if_unmodified_since(args.if_unmodified_since().is_some());
+ .with_caller_condition(args.is_conditional());
let resp = backend
.core
.s3_get_object(&self.ctx, path, range, &args)
diff --git a/core/services/s3/src/writer.rs b/core/services/s3/src/writer.rs
index 7cf70bcd1..f2e3facc0 100644
--- a/core/services/s3/src/writer.rs
+++ b/core/services/s3/src/writer.rs
@@ -70,10 +70,7 @@ impl S3Writer {
}
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())
+ ErrorContext::new(service_operation).with_caller_condition(self.op.is_conditional())
}
}
@@ -198,10 +195,7 @@ impl oio::MultipartWrite for S3Writer {
.expect("multipart writer copy range must be bounded");
let part_number = part_number + 1;
let error_context = ErrorContext::new(ServiceOperation("UploadPartCopy"))
- .with_if_match(args.if_match().is_some())
- .with_if_none_match(args.if_none_match().is_some())
- .with_if_modified_since(args.if_modified_since().is_some())
- .with_if_unmodified_since(args.if_unmodified_since().is_some());
+ .with_caller_condition(args.is_conditional());
let req = self
.core
.s3_upload_part_copy_request(S3UploadPartCopyRequest {