diff --git a/bindings/c/src/operator.rs b/bindings/c/src/operator.rs index 0a9d77b51227..65d2503c1241 100644 --- a/bindings/c/src/operator.rs +++ b/bindings/c/src/operator.rs @@ -820,6 +820,7 @@ pub unsafe extern "C" fn opendal_operator_delete_with( version, recursive: o.recursive, if_match: None, + ..Default::default() } }; match op.deref().delete_options(path, delete_opts) { diff --git a/bindings/c/src/presign.rs b/bindings/c/src/presign.rs index 4b18cbe8ecce..6fee7d628f5c 100644 --- a/bindings/c/src/presign.rs +++ b/bindings/c/src/presign.rs @@ -242,6 +242,7 @@ pub unsafe extern "C" fn opendal_operator_presign_delete_with( version, recursive: opts.recursive, if_match: None, + ..Default::default() } }; diff --git a/bindings/c/src/types.rs b/bindings/c/src/types.rs index df1fad2a66f4..8ae899a6dde5 100644 --- a/bindings/c/src/types.rs +++ b/bindings/c/src/types.rs @@ -601,6 +601,7 @@ impl From<&opendal_write_options> for options::WriteOptions { if_not_exists: value.if_not_exists, concurrent: value.concurrent, chunk: value.has_chunk.then_some(value.chunk), + ..Default::default() } } } @@ -777,6 +778,7 @@ impl From<&opendal_stat_options> for options::StatOptions { override_content_disposition: unsafe { optional_cstr(value.override_content_disposition) }, + ..Default::default() } } } @@ -1078,6 +1080,7 @@ impl From<&opendal_read_options> for options::ReadOptions { override_content_disposition: unsafe { optional_cstr(value.override_content_disposition) }, + ..Default::default() } } } @@ -1299,6 +1302,7 @@ impl From<&opendal_reader_options> for options::ReaderOptions { chunk: value.has_chunk.then_some(value.chunk), gap: value.has_gap.then_some(value.gap), prefetch: value.prefetch, + ..Default::default() } } } @@ -1437,6 +1441,7 @@ impl From<&opendal_copy_options> for options::CopyOptions { .then_some(value.source_content_length_hint), concurrent: value.concurrent, chunk: value.has_chunk.then_some(value.chunk), + ..Default::default() } } } diff --git a/bindings/cpp/src/lib.rs b/bindings/cpp/src/lib.rs index 0a06516efbbd..c807d7e0b4a7 100644 --- a/bindings/cpp/src/lib.rs +++ b/bindings/cpp/src/lib.rs @@ -462,6 +462,7 @@ fn stat_options(opts: ffi::FfiStatOptions) -> Result { override_content_type: optional_string(opts.override_content_type), override_cache_control: optional_string(opts.override_cache_control), override_content_disposition: optional_string(opts.override_content_disposition), + ..Default::default() }) } @@ -480,6 +481,7 @@ fn read_options(opts: ffi::FfiReadOptions) -> Result { override_content_type: optional_string(opts.override_content_type), override_cache_control: optional_string(opts.override_cache_control), override_content_disposition: optional_string(opts.override_content_disposition), + ..Default::default() }) } @@ -495,6 +497,7 @@ fn reader_options(opts: ffi::FfiReaderOptions) -> Result od::options::WriteOptions { if_not_exists: opts.if_not_exists, concurrent: opts.concurrent, chunk: optional_usize(opts.chunk), + ..Default::default() } } @@ -522,6 +526,7 @@ fn copy_options(opts: ffi::FfiCopyOptions) -> od::options::CopyOptions { source_content_length_hint: optional_u64(opts.source_content_length_hint), concurrent: opts.concurrent, chunk: optional_usize(opts.chunk), + ..Default::default() } } @@ -536,6 +541,7 @@ fn delete_options(opts: ffi::FfiDeleteOptions) -> od::options::DeleteOptions { version: optional_string(opts.version), recursive: opts.recursive, if_match: None, + ..Default::default() } } diff --git a/bindings/dotnet/OpenDAL/Capability.cs b/bindings/dotnet/OpenDAL/Capability.cs index b14e0e48e192..efe8ca754e58 100644 --- a/bindings/dotnet/OpenDAL/Capability.cs +++ b/bindings/dotnet/OpenDAL/Capability.cs @@ -38,6 +38,8 @@ internal Capability(OpenDALCapability native) Stat = native.stat != 0; StatWithIfMatch = native.statWithIfMatch != 0; StatWithIfNoneMatch = native.statWithIfNoneMatch != 0; + StatWithIfVersionMatch = native.statWithIfVersionMatch != 0; + StatWithIfVersionNotMatch = native.statWithIfVersionNotMatch != 0; StatWithIfModifiedSince = native.statWithIfModifiedSince != 0; StatWithIfUnmodifiedSince = native.statWithIfUnmodifiedSince != 0; StatWithOverrideCacheControl = native.statWithOverrideCacheControl != 0; @@ -47,6 +49,8 @@ internal Capability(OpenDALCapability native) Read = native.read != 0; ReadWithIfMatch = native.readWithIfMatch != 0; ReadWithIfNoneMatch = native.readWithIfNoneMatch != 0; + ReadWithIfVersionMatch = native.readWithIfVersionMatch != 0; + ReadWithIfVersionNotMatch = native.readWithIfVersionNotMatch != 0; ReadWithIfModifiedSince = native.readWithIfModifiedSince != 0; ReadWithIfUnmodifiedSince = native.readWithIfUnmodifiedSince != 0; ReadWithOverrideCacheControl = native.readWithOverrideCacheControl != 0; @@ -64,6 +68,8 @@ internal Capability(OpenDALCapability native) WriteWithCacheControl = native.writeWithCacheControl != 0; WriteWithIfMatch = native.writeWithIfMatch != 0; WriteWithIfNoneMatch = native.writeWithIfNoneMatch != 0; + WriteWithIfVersionMatch = native.writeWithIfVersionMatch != 0; + WriteWithIfVersionNotMatch = native.writeWithIfVersionNotMatch != 0; WriteWithIfNotExists = native.writeWithIfNotExists != 0; WriteWithUserMetadata = native.writeWithUserMetadata != 0; WriteMultiMaxSize = native.writeMultiMaxSize == nuint.MaxValue ? null : native.writeMultiMaxSize; @@ -74,14 +80,23 @@ internal Capability(OpenDALCapability native) DeleteWithVersion = native.deleteWithVersion != 0; DeleteWithRecursive = native.deleteWithRecursive != 0; DeleteWithIfMatch = native.deleteWithIfMatch != 0; + DeleteWithIfNoneMatch = native.deleteWithIfNoneMatch != 0; + DeleteWithIfVersionMatch = native.deleteWithIfVersionMatch != 0; + DeleteWithIfVersionNotMatch = native.deleteWithIfVersionNotMatch != 0; DeleteMaxSize = native.deleteMaxSize == nuint.MaxValue ? null : native.deleteMaxSize; Copy = native.copy != 0; CopyWithIfNotExists = native.copyWithIfNotExists != 0; CopyWithIfMatch = native.copyWithIfMatch != 0; + CopyWithIfNoneMatch = native.copyWithIfNoneMatch != 0; + CopyWithIfVersionMatch = native.copyWithIfVersionMatch != 0; + CopyWithIfVersionNotMatch = native.copyWithIfVersionNotMatch != 0; CopyWithSourceVersion = native.copyWithSourceVersion != 0; CopyCanMulti = native.copyCanMulti != 0; CopyMultiMaxSize = native.copyMultiMaxSize == nuint.MaxValue ? null : native.copyMultiMaxSize; CopyMultiMinSize = native.copyMultiMinSize == nuint.MaxValue ? null : native.copyMultiMinSize; + Restore = native.restore != 0; + RestoreWithVersion = native.restoreWithVersion != 0; + RestoreWithIfNotExists = native.restoreWithIfNotExists != 0; Rename = native.rename != 0; RenameWithIfNotExists = native.renameWithIfNotExists != 0; List = native.list != 0; @@ -113,6 +128,16 @@ internal Capability(OpenDALCapability native) /// public bool StatWithIfNoneMatch { get; private set; } + /// + /// Indicates if conditional stat operations using version match are supported. + /// + public bool StatWithIfVersionMatch { get; private set; } + + /// + /// Indicates if conditional stat operations using version non-match are supported. + /// + public bool StatWithIfVersionNotMatch { get; private set; } + /// /// Indicates if conditional stat operations using If-Modified-Since are supported. /// @@ -158,6 +183,16 @@ internal Capability(OpenDALCapability native) /// public bool ReadWithIfNoneMatch { get; private set; } + /// + /// Indicates if conditional read operations using version match are supported. + /// + public bool ReadWithIfVersionMatch { get; private set; } + + /// + /// Indicates if conditional read operations using version non-match are supported. + /// + public bool ReadWithIfVersionNotMatch { get; private set; } + /// /// Indicates if conditional read operations using If-Modified-Since are supported. /// @@ -243,6 +278,16 @@ internal Capability(OpenDALCapability native) /// public bool WriteWithIfNoneMatch { get; private set; } + /// + /// Indicates if conditional write operations using version match are supported. + /// + public bool WriteWithIfVersionMatch { get; private set; } + + /// + /// Indicates if conditional write operations using version non-match are supported. + /// + public bool WriteWithIfVersionNotMatch { get; private set; } + /// /// Indicates if write operations can be conditional on object non-existence. /// @@ -293,6 +338,21 @@ internal Capability(OpenDALCapability native) /// public bool DeleteWithIfMatch { get; private set; } + /// + /// Indicates if conditional delete operations using If-None-Match are supported. + /// + public bool DeleteWithIfNoneMatch { get; private set; } + + /// + /// Indicates if conditional delete operations using version match are supported. + /// + public bool DeleteWithIfVersionMatch { get; private set; } + + /// + /// Indicates if conditional delete operations using version non-match are supported. + /// + public bool DeleteWithIfVersionNotMatch { get; private set; } + /// /// Maximum size supported for single delete operations. /// @@ -313,6 +373,21 @@ internal Capability(OpenDALCapability native) /// public bool CopyWithIfMatch { get; private set; } + /// + /// Indicates if conditional copy operations with if-none-match are supported. + /// + public bool CopyWithIfNoneMatch { get; private set; } + + /// + /// Indicates if conditional copy operations using version match are supported. + /// + public bool CopyWithIfVersionMatch { get; private set; } + + /// + /// Indicates if conditional copy operations using version non-match are supported. + /// + public bool CopyWithIfVersionNotMatch { get; private set; } + /// /// Indicates if copy operations from a specific source version are supported. /// @@ -333,6 +408,21 @@ internal Capability(OpenDALCapability native) /// public ulong? CopyMultiMinSize { get; private set; } + /// + /// Indicates if restore operations are supported. + /// + public bool Restore { get; private set; } + + /// + /// Indicates if restoring a specific version is supported. + /// + public bool RestoreWithVersion { get; private set; } + + /// + /// Indicates if conditional restore operations using if-not-exists are supported. + /// + public bool RestoreWithIfNotExists { get; private set; } + /// /// Indicates if rename operations are supported. /// diff --git a/bindings/dotnet/OpenDAL/Interop/NativeObject/OpenDALCapability.cs b/bindings/dotnet/OpenDAL/Interop/NativeObject/OpenDALCapability.cs index 9781be7ff359..ba0028a52b7e 100644 --- a/bindings/dotnet/OpenDAL/Interop/NativeObject/OpenDALCapability.cs +++ b/bindings/dotnet/OpenDAL/Interop/NativeObject/OpenDALCapability.cs @@ -29,6 +29,8 @@ internal struct OpenDALCapability [MarshalAs(UnmanagedType.U1)] internal byte stat; [MarshalAs(UnmanagedType.U1)] internal byte statWithIfMatch; [MarshalAs(UnmanagedType.U1)] internal byte statWithIfNoneMatch; + [MarshalAs(UnmanagedType.U1)] internal byte statWithIfVersionMatch; + [MarshalAs(UnmanagedType.U1)] internal byte statWithIfVersionNotMatch; [MarshalAs(UnmanagedType.U1)] internal byte statWithIfModifiedSince; [MarshalAs(UnmanagedType.U1)] internal byte statWithIfUnmodifiedSince; [MarshalAs(UnmanagedType.U1)] internal byte statWithOverrideCacheControl; @@ -38,6 +40,8 @@ internal struct OpenDALCapability [MarshalAs(UnmanagedType.U1)] internal byte read; [MarshalAs(UnmanagedType.U1)] internal byte readWithIfMatch; [MarshalAs(UnmanagedType.U1)] internal byte readWithIfNoneMatch; + [MarshalAs(UnmanagedType.U1)] internal byte readWithIfVersionMatch; + [MarshalAs(UnmanagedType.U1)] internal byte readWithIfVersionNotMatch; [MarshalAs(UnmanagedType.U1)] internal byte readWithIfModifiedSince; [MarshalAs(UnmanagedType.U1)] internal byte readWithIfUnmodifiedSince; [MarshalAs(UnmanagedType.U1)] internal byte readWithOverrideCacheControl; @@ -55,6 +59,8 @@ internal struct OpenDALCapability [MarshalAs(UnmanagedType.U1)] internal byte writeWithCacheControl; [MarshalAs(UnmanagedType.U1)] internal byte writeWithIfMatch; [MarshalAs(UnmanagedType.U1)] internal byte writeWithIfNoneMatch; + [MarshalAs(UnmanagedType.U1)] internal byte writeWithIfVersionMatch; + [MarshalAs(UnmanagedType.U1)] internal byte writeWithIfVersionNotMatch; [MarshalAs(UnmanagedType.U1)] internal byte writeWithIfNotExists; [MarshalAs(UnmanagedType.U1)] internal byte writeWithUserMetadata; internal nuint writeMultiMaxSize; @@ -65,14 +71,23 @@ internal struct OpenDALCapability [MarshalAs(UnmanagedType.U1)] internal byte deleteWithVersion; [MarshalAs(UnmanagedType.U1)] internal byte deleteWithRecursive; [MarshalAs(UnmanagedType.U1)] internal byte deleteWithIfMatch; + [MarshalAs(UnmanagedType.U1)] internal byte deleteWithIfNoneMatch; + [MarshalAs(UnmanagedType.U1)] internal byte deleteWithIfVersionMatch; + [MarshalAs(UnmanagedType.U1)] internal byte deleteWithIfVersionNotMatch; internal nuint deleteMaxSize; [MarshalAs(UnmanagedType.U1)] internal byte copy; [MarshalAs(UnmanagedType.U1)] internal byte copyWithIfNotExists; [MarshalAs(UnmanagedType.U1)] internal byte copyWithIfMatch; + [MarshalAs(UnmanagedType.U1)] internal byte copyWithIfNoneMatch; + [MarshalAs(UnmanagedType.U1)] internal byte copyWithIfVersionMatch; + [MarshalAs(UnmanagedType.U1)] internal byte copyWithIfVersionNotMatch; [MarshalAs(UnmanagedType.U1)] internal byte copyWithSourceVersion; [MarshalAs(UnmanagedType.U1)] internal byte copyCanMulti; internal nuint copyMultiMaxSize; internal nuint copyMultiMinSize; + [MarshalAs(UnmanagedType.U1)] internal byte restore; + [MarshalAs(UnmanagedType.U1)] internal byte restoreWithVersion; + [MarshalAs(UnmanagedType.U1)] internal byte restoreWithIfNotExists; [MarshalAs(UnmanagedType.U1)] internal byte rename; [MarshalAs(UnmanagedType.U1)] internal byte renameWithIfNotExists; [MarshalAs(UnmanagedType.U1)] internal byte list; diff --git a/bindings/dotnet/src/capability.rs b/bindings/dotnet/src/capability.rs index df4d586fbc62..f5d405ed3ca6 100644 --- a/bindings/dotnet/src/capability.rs +++ b/bindings/dotnet/src/capability.rs @@ -36,6 +36,10 @@ pub struct Capability { pub stat_with_if_match: bool, /// Indicates if conditional stat operations using If-None-Match are supported. pub stat_with_if_none_match: bool, + /// Indicates if conditional stat operations using version match are supported. + pub stat_with_if_version_match: bool, + /// Indicates if conditional stat operations using version non-match are supported. + pub stat_with_if_version_not_match: bool, /// Indicates if conditional stat operations using If-Modified-Since are supported. pub stat_with_if_modified_since: bool, /// Indicates if conditional stat operations using If-Unmodified-Since are supported. @@ -54,6 +58,10 @@ pub struct Capability { pub read_with_if_match: bool, /// Indicates if conditional read operations using If-None-Match are supported. pub read_with_if_none_match: bool, + /// Indicates if conditional read operations using version match are supported. + pub read_with_if_version_match: bool, + /// Indicates if conditional read operations using version non-match are supported. + pub read_with_if_version_not_match: bool, /// Indicates if conditional read operations using If-Modified-Since are supported. pub read_with_if_modified_since: bool, /// Indicates if conditional read operations using If-Unmodified-Since are supported. @@ -88,6 +96,10 @@ pub struct Capability { pub write_with_if_match: bool, /// Indicates if conditional write operations using If-None-Match are supported. pub write_with_if_none_match: bool, + /// Indicates if conditional write operations using version match are supported. + pub write_with_if_version_match: bool, + /// Indicates if conditional write operations using version non-match are supported. + pub write_with_if_version_not_match: bool, /// Indicates if write operations can be conditional on object non-existence. pub write_with_if_not_exists: bool, /// Indicates if custom user metadata can be attached during write operations. @@ -108,6 +120,12 @@ pub struct Capability { pub delete_with_recursive: bool, /// Indicates if conditional delete operations using If-Match are supported. pub delete_with_if_match: bool, + /// Indicates if conditional delete operations using If-None-Match are supported. + pub delete_with_if_none_match: bool, + /// Indicates if conditional delete operations using version match are supported. + pub delete_with_if_version_match: bool, + /// Indicates if conditional delete operations using version non-match are supported. + pub delete_with_if_version_not_match: bool, /// Maximum size supported for single delete operations. pub delete_max_size: usize, /// Indicates if copy operations are supported. @@ -116,6 +134,12 @@ pub struct Capability { pub copy_with_if_not_exists: bool, /// Indicates if conditional copy operations with if-match are supported. pub copy_with_if_match: bool, + /// Indicates if conditional copy operations with if-none-match are supported. + pub copy_with_if_none_match: bool, + /// Indicates if conditional copy operations using version match are supported. + pub copy_with_if_version_match: bool, + /// Indicates if conditional copy operations using version non-match are supported. + pub copy_with_if_version_not_match: bool, /// Indicates if copy operations from a specific source version are supported. pub copy_with_source_version: bool, /// Indicates if copy operations can be split into multiple server-side tasks. @@ -124,6 +148,12 @@ pub struct Capability { pub copy_multi_max_size: usize, /// Minimum size required for segmented copy tasks. pub copy_multi_min_size: usize, + /// Indicates if restore operations are supported. + pub restore: bool, + /// Indicates if restoring a specific version is supported. + pub restore_with_version: bool, + /// Indicates if conditional restore operations using if-not-exists are supported. + pub restore_with_if_not_exists: bool, /// Indicates if rename operations are supported. pub rename: bool, /// Indicates if conditional rename operations with if-not-exists are supported. @@ -161,6 +191,8 @@ impl Capability { stat: cap.stat, stat_with_if_match: cap.stat_with_if_match, stat_with_if_none_match: cap.stat_with_if_none_match, + stat_with_if_version_match: cap.stat_with_if_version_match, + stat_with_if_version_not_match: cap.stat_with_if_version_not_match, stat_with_if_modified_since: cap.stat_with_if_modified_since, stat_with_if_unmodified_since: cap.stat_with_if_unmodified_since, stat_with_override_cache_control: cap.stat_with_override_cache_control, @@ -170,6 +202,8 @@ impl Capability { read: cap.read, read_with_if_match: cap.read_with_if_match, read_with_if_none_match: cap.read_with_if_none_match, + read_with_if_version_match: cap.read_with_if_version_match, + read_with_if_version_not_match: cap.read_with_if_version_not_match, read_with_if_modified_since: cap.read_with_if_modified_since, read_with_if_unmodified_since: cap.read_with_if_unmodified_since, read_with_override_cache_control: cap.read_with_override_cache_control, @@ -187,6 +221,8 @@ impl Capability { write_with_cache_control: cap.write_with_cache_control, write_with_if_match: cap.write_with_if_match, write_with_if_none_match: cap.write_with_if_none_match, + write_with_if_version_match: cap.write_with_if_version_match, + write_with_if_version_not_match: cap.write_with_if_version_not_match, write_with_if_not_exists: cap.write_with_if_not_exists, write_with_user_metadata: cap.write_with_user_metadata, write_multi_max_size: cap.write_multi_max_size.unwrap_or(usize::MAX), @@ -197,14 +233,23 @@ impl Capability { delete_with_version: cap.delete_with_version, delete_with_recursive: cap.delete_with_recursive, delete_with_if_match: cap.delete_with_if_match, + delete_with_if_none_match: cap.delete_with_if_none_match, + delete_with_if_version_match: cap.delete_with_if_version_match, + delete_with_if_version_not_match: cap.delete_with_if_version_not_match, delete_max_size: cap.delete_max_size.unwrap_or(usize::MAX), copy: cap.copy, copy_with_if_not_exists: cap.copy_with_if_not_exists, copy_with_if_match: cap.copy_with_if_match, + copy_with_if_none_match: cap.copy_with_if_none_match, + copy_with_if_version_match: cap.copy_with_if_version_match, + copy_with_if_version_not_match: cap.copy_with_if_version_not_match, copy_with_source_version: cap.copy_with_source_version, copy_can_multi: cap.copy_can_multi, copy_multi_max_size: cap.copy_multi_max_size.unwrap_or(usize::MAX), copy_multi_min_size: cap.copy_multi_min_size.unwrap_or(usize::MAX), + restore: cap.restore, + restore_with_version: cap.restore_with_version, + restore_with_if_not_exists: cap.restore_with_if_not_exists, rename: cap.rename, rename_with_if_not_exists: cap.rename_with_if_not_exists, list: cap.list, diff --git a/bindings/dotnet/src/options.rs b/bindings/dotnet/src/options.rs index b19f8bea8c63..82c3a5aec413 100644 --- a/bindings/dotnet/src/options.rs +++ b/bindings/dotnet/src/options.rs @@ -181,6 +181,7 @@ pub fn parse_stat_options( override_content_type: parse_string(values, "override_content_type"), override_cache_control: parse_string(values, "override_cache_control"), override_content_disposition: parse_string(values, "override_content_disposition"), + ..Default::default() }) } @@ -218,5 +219,6 @@ pub fn parse_delete_options( version: parse_string(values, "version"), recursive: parse_bool(values, "recursive")?.unwrap_or(false), if_match: parse_string(values, "if_match"), + ..Default::default() }) } diff --git a/bindings/java/src/lib.rs b/bindings/java/src/lib.rs index 7b6952ca9c88..e95057856796 100644 --- a/bindings/java/src/lib.rs +++ b/bindings/java/src/lib.rs @@ -236,6 +236,7 @@ fn make_write_options<'a>( user_metadata: convert::read_map_field(env, options, "userMetadata")?, concurrent, chunk: convert::read_jlong_field_to_usize(env, options, "chunk")?, + ..Default::default() }) } @@ -274,6 +275,7 @@ fn make_stat_options(env: &mut Env, options: &JObject) -> Result for opendal::options::StatOptions { override_content_type: value.override_content_type, override_cache_control: value.override_cache_control, override_content_disposition: value.override_content_disposition, + ..Default::default() } } } @@ -235,6 +236,7 @@ impl From for opendal::options::ReadOptions { override_content_type: value.content_type, override_cache_control: value.cache_control, override_content_disposition: value.content_disposition, + ..Default::default() } } } @@ -347,6 +349,7 @@ impl From for opendal::options::ReaderOptions { if_modified_since, if_unmodified_since, content_length_hint: None, + ..Default::default() } } } @@ -503,6 +506,7 @@ impl From for opendal::options::WriteOptions { if_none_match: value.if_none_match, if_not_exists: value.if_not_exists.unwrap_or_default(), concurrent: value.concurrent.unwrap_or_default() as usize, + ..Default::default() } } } @@ -527,6 +531,7 @@ impl From for opendal::options::DeleteOptions { version: value.version, recursive: value.recursive.unwrap_or_default(), if_match: value.if_match, + ..Default::default() } } } diff --git a/bindings/python/src/operator.rs b/bindings/python/src/operator.rs index 47de79e0a539..7f017f81cf15 100644 --- a/bindings/python/src/operator.rs +++ b/bindings/python/src/operator.rs @@ -725,6 +725,7 @@ impl Operator { version, recursive: recursive.unwrap_or(false), if_match, + ..Default::default() }; self.core.delete_options(&path, opts).map_err(format_pyerr) } else { @@ -1553,6 +1554,7 @@ impl AsyncOperator { version, recursive: recursive.unwrap_or(false), if_match, + ..Default::default() }; this.delete_options(&path, opts).await.map_err(format_pyerr) } else { diff --git a/bindings/python/src/options.rs b/bindings/python/src/options.rs index 01025021de01..96f37376e4b2 100644 --- a/bindings/python/src/options.rs +++ b/bindings/python/src/options.rs @@ -187,6 +187,7 @@ impl From for ocore::options::ReadOptions { override_content_type: opts.content_type, override_cache_control: opts.cache_control, override_content_disposition: opts.content_disposition, + ..Default::default() } } } @@ -204,6 +205,7 @@ impl From for ocore::options::ReaderOptions { chunk: opts.chunk, gap: opts.gap, prefetch: opts.prefetch.unwrap_or_default(), + ..Default::default() } } } @@ -222,6 +224,7 @@ impl From for ocore::options::WriteOptions { if_match: opts.if_match, if_none_match: opts.if_none_match, if_not_exists: opts.if_not_exists.unwrap_or(false), + ..Default::default() } } } @@ -322,6 +325,7 @@ impl From for ocore::options::StatOptions { override_content_type: opts.content_type, override_cache_control: opts.cache_control, override_content_disposition: opts.content_disposition, + ..Default::default() } } } @@ -360,6 +364,7 @@ impl From for ocore::options::DeleteOptions { version: opts.version, recursive: opts.recursive.unwrap_or(false), if_match: opts.if_match, + ..Default::default() } } } diff --git a/core/core/src/docs/rfcs/8147_write_if_not_changed.md b/core/core/src/docs/rfcs/8147_write_if_not_changed.md index f76467210585..9e79d8f83f2f 100644 --- a/core/core/src/docs/rfcs/8147_write_if_not_changed.md +++ b/core/core/src/docs/rfcs/8147_write_if_not_changed.md @@ -69,9 +69,6 @@ let expected = Metadata::default() Preserving both fields keeps externally stored metadata portable across services. Both fields must describe the same observed object state. -The new conditions are not supported by presign operations. Supplying them to -a presign options API returns `Unsupported` before service dispatch. - ## Condition semantics All token values are opaque. A `version` option selects a historical object; @@ -108,13 +105,13 @@ participate. OpenDAL does not validate provenance, path, or storage namespace, so the caller must associate externally constructed metadata with the correct target. -`if_not_changed` is exclusive with every other target condition and, on -delete, with the target `version` selector. Version match and non-match are -mutually exclusive and cannot be combined with another target condition or a -target version selector. Copy's `source_version` selects a different object -and may be combined with destination conditions. Append writes cannot use -version conditions or `if_not_changed` because append has no portable final -replacement commit point. Invalid combinations return `ConfigInvalid`. +OpenDAL preserves explicit target conditions and version selectors while +lowering `if_not_changed`. If the selected primitive match field already +contains the same opaque token, OpenDAL deduplicates it. If that field contains +a different token, OpenDAL returns `ConditionNotMatch` because both equality +conditions cannot hold. OpenDAL forwards every other representable combination +unchanged. After OpenDAL validates each condition's capability, the service +decides whether it accepts the combination and reports any invalid combination. Missing-target results are part of the portable contract: @@ -155,6 +152,9 @@ All new capabilities default to `false`. There is no independently advertised `if_not_changed` capability; support is derived from the applicable version and ETag match capabilities plus the supplied metadata. +Capabilities describe support for individual conditions. They do not guarantee +that a service accepts any particular combination of conditions or selectors. + A service advertising a version-match primitive must populate `Metadata::version()` on stat, read, write, copy, and list whenever the native response provides it. The same rule applies to `Metadata::etag()` for an @@ -173,12 +173,6 @@ GCS maps version conditions to JSON API generation parameters: | `if_not_exists` | `ifGenerationMatch=0` | | `if_not_changed(meta)` | Lower to `if_version_match(meta.version())`. | -GCS generation `0` is reserved for the absence condition and is never an -object generation. GCS rejects `if_version_match("0")` and expected metadata -whose selected version is `"0"` with `ConfigInvalid`. -`if_version_not_match("0")` remains valid and succeeds only when a live object -exists. - GCS applies these parameters to JSON API get, insert, delete, and destination rewrite requests. `ifGenerationNotMatch` fails when no live object exists, and OpenDAL maps that failure to `ConditionNotMatch`. Multi-request rewrite keeps diff --git a/core/core/src/layers/correctness_check.rs b/core/core/src/layers/correctness_check.rs index e9ac92004403..0d25b3aa29e4 100644 --- a/core/core/src/layers/correctness_check.rs +++ b/core/core/src/layers/correctness_check.rs @@ -109,6 +109,20 @@ impl Service for CorrectnessService { "if_none_match", )); } + if !capability.read_with_if_version_match && args.if_version_match().is_some() { + return Err(new_unsupported_error( + scheme, + Operation::Read, + "if_version_match", + )); + } + if !capability.read_with_if_version_not_match && args.if_version_not_match().is_some() { + return Err(new_unsupported_error( + scheme, + Operation::Read, + "if_version_not_match", + )); + } if !capability.read_with_if_modified_since && args.if_modified_since().is_some() { return Err(new_unsupported_error( scheme, @@ -153,6 +167,20 @@ impl Service for CorrectnessService { return Err(err); } + if args.if_version_match().is_some() && !capability.write_with_if_version_match { + return Err(new_unsupported_error( + scheme, + Operation::Write, + "if_version_match", + )); + } + if args.if_version_not_match().is_some() && !capability.write_with_if_version_not_match { + return Err(new_unsupported_error( + scheme, + Operation::Write, + "if_version_not_match", + )); + } self.inner.write(ctx, path, args) } @@ -173,6 +201,20 @@ impl Service for CorrectnessService { "if_none_match", )); } + if !capability.stat_with_if_version_match && args.if_version_match().is_some() { + return Err(new_unsupported_error( + scheme, + Operation::Stat, + "if_version_match", + )); + } + if !capability.stat_with_if_version_not_match && args.if_version_not_match().is_some() { + return Err(new_unsupported_error( + scheme, + Operation::Stat, + "if_version_not_match", + )); + } if !capability.stat_with_if_modified_since && args.if_modified_since().is_some() { return Err(new_unsupported_error( scheme, @@ -217,6 +259,27 @@ impl Service for CorrectnessService { if args.if_match().is_some() && !capability.copy_with_if_match { return Err(new_unsupported_error(scheme, Operation::Copy, "if_match")); } + if args.if_none_match().is_some() && !capability.copy_with_if_none_match { + return Err(new_unsupported_error( + scheme, + Operation::Copy, + "if_none_match", + )); + } + if args.if_version_match().is_some() && !capability.copy_with_if_version_match { + return Err(new_unsupported_error( + scheme, + Operation::Copy, + "if_version_match", + )); + } + if args.if_version_not_match().is_some() && !capability.copy_with_if_version_not_match { + return Err(new_unsupported_error( + scheme, + Operation::Copy, + "if_version_not_match", + )); + } if args.source_version().is_some() && !capability.copy_with_source_version { return Err(new_unsupported_error( scheme, @@ -343,6 +406,32 @@ impl CheckWrapper { )); } + if args.if_none_match().is_some() && !self.capability.delete_with_if_none_match { + return Err(new_unsupported_error( + self.scheme, + Operation::Delete, + "if_none_match", + )); + } + + if args.if_version_match().is_some() && !self.capability.delete_with_if_version_match { + return Err(new_unsupported_error( + self.scheme, + Operation::Delete, + "if_version_match", + )); + } + + if args.if_version_not_match().is_some() + && !self.capability.delete_with_if_version_not_match + { + return Err(new_unsupported_error( + self.scheme, + Operation::Delete, + "if_version_not_match", + )); + } + Ok(()) } } @@ -427,10 +516,7 @@ mod tests { _: OpCopy, _: OpCopier, ) -> Result { - Err(Error::new( - ErrorKind::Unsupported, - "operation is not supported", - )) + Ok(()) } async fn rename( @@ -610,6 +696,159 @@ mod tests { assert!(res.is_ok()) } + #[tokio::test] + async fn test_version_preconditions_are_forwarded() -> Result<()> { + let op = new_test_operator(Capability { + stat: true, + stat_with_version: true, + stat_with_if_version_match: true, + read: true, + read_with_if_version_match: true, + write: true, + write_with_if_match: true, + write_with_if_version_match: true, + delete: true, + delete_with_if_version_match: true, + delete_with_if_version_not_match: true, + copy: true, + copy_with_if_version_match: true, + ..Default::default() + }); + + op.stat_with("path").if_version_match("version").await?; + op.read_with("path").if_version_match("version").await?; + op.write_with("path", "") + .if_version_match("version") + .await?; + op.delete_with("path").if_version_match("version").await?; + op.copy_with("from", "to") + .if_version_match("version") + .await?; + + op.write_with("path", "") + .if_match("etag") + .if_version_match("version") + .await?; + + op.stat_with("path") + .version("selected") + .if_version_match("current") + .await?; + + op.delete_with("path") + .if_version_match("matched") + .if_version_not_match("not-matched") + .await?; + + Ok(()) + } + + #[tokio::test] + async fn test_if_not_changed_merges_at_dispatch() -> Result<()> { + let op = new_test_operator(Capability { + write: true, + write_with_if_match: true, + write_with_if_version_match: true, + delete: true, + delete_with_if_match: true, + copy: true, + copy_with_if_match: true, + ..Default::default() + }); + let metadata = Metadata::default() + .with_etag("etag".to_string()) + .with_version("version".to_string()); + + op.write_with("path", "") + .if_match("other-etag") + .if_version_match("version") + .if_not_changed(&metadata) + .await?; + + let err = op + .write_with("path", "") + .if_version_match("other-version") + .if_not_changed(&metadata) + .await + .expect_err("different selected version must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.copy_with("from", "to") + .if_match("etag") + .if_not_changed(&metadata) + .await?; + let err = op + .copy_with("from", "to") + .if_match("other-etag") + .if_not_changed(&metadata) + .await + .expect_err("different selected etag must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.delete_with("path") + .if_match("etag") + .if_not_changed(&metadata) + .await?; + let err = op + .delete_with("path") + .if_match("other-etag") + .if_not_changed(&metadata) + .await + .expect_err("different selected etag must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + Ok(()) + } + + #[tokio::test] + async fn test_presign_conditions_reach_service() { + let op = new_test_operator(Capability::default()); + + let err = op + .presign_stat_options( + "path", + std::time::Duration::from_secs(60), + options::StatOptions { + if_version_match: Some("version".to_string()), + ..Default::default() + }, + ) + .await + .expect_err("mock service rejects presign"); + assert!(err.to_string().contains("operation is not supported")); + + let err = op + .presign_delete_options( + "path", + std::time::Duration::from_secs(60), + options::DeleteOptions { + if_none_match: Some("etag".to_string()), + if_version_not_match: Some("version".to_string()), + ..Default::default() + }, + ) + .await + .expect_err("mock service rejects presign"); + assert!(err.to_string().contains("operation is not supported")); + + let op = new_test_operator(Capability { + write_with_if_match: true, + ..Default::default() + }); + let err = op + .presign_write_options( + "path", + std::time::Duration::from_secs(60), + options::WriteOptions { + if_not_changed: Some(Metadata::default().with_etag("etag".to_string())), + ..Default::default() + }, + ) + .await + .expect_err("mock service rejects presign"); + assert!(err.to_string().contains("operation is not supported")); + } + #[tokio::test] async fn test_rename_with_if_not_exists() { let op = new_test_operator(Capability { diff --git a/core/core/src/raw/accessor.rs b/core/core/src/raw/accessor.rs index 6902daa5d988..3b5fb044857a 100644 --- a/core/core/src/raw/accessor.rs +++ b/core/core/src/raw/accessor.rs @@ -185,7 +185,7 @@ pub trait Service: Send + Sync + Debug + Unpin + 'static { /// # Behavior /// /// - The returned deleter handles one or more delete requests. - /// - Deleting a missing path should succeed. + /// - Deleting a missing path should succeed unless a condition requires a live target. fn delete(&self, ctx: &OperationContext) -> Result; /// Invoke the `list` operation on the specified path. diff --git a/core/core/src/raw/ops.rs b/core/core/src/raw/ops.rs index 6b3cc53bb795..1356618b9952 100644 --- a/core/core/src/raw/ops.rs +++ b/core/core/src/raw/ops.rs @@ -51,6 +51,15 @@ pub struct OpDelete { /// The ETag that the object must match before deletion. if_match: Option, + + /// The ETag that the object must not match before deletion. + if_none_match: Option, + + /// The version that the current object must match before deletion. + if_version_match: Option, + + /// The version that the current object must not match before deletion. + if_version_not_match: Option, } impl OpDelete { @@ -81,6 +90,24 @@ impl OpDelete { self } + /// Set the ETag that the object must not match before deletion. + pub fn with_if_none_match(mut self, if_none_match: impl Into) -> Self { + self.if_none_match = Some(if_none_match.into()); + self + } + + /// Set the version that the current object must match before deletion. + pub fn with_if_version_match(mut self, version: impl Into) -> Self { + self.if_version_match = Some(version.into()); + self + } + + /// Set the version that the current object must not match before deletion. + pub fn with_if_version_not_match(mut self, version: impl Into) -> Self { + self.if_version_not_match = Some(version.into()); + self + } + /// Return the version of the object to delete. pub fn version(&self) -> Option<&str> { self.version.as_deref() @@ -95,6 +122,21 @@ impl OpDelete { pub fn if_match(&self) -> Option<&str> { self.if_match.as_deref() } + + /// Return the ETag that the object must not match before deletion. + pub fn if_none_match(&self) -> Option<&str> { + self.if_none_match.as_deref() + } + + /// Return the version that the current object must match before deletion. + pub fn if_version_match(&self) -> Option<&str> { + self.if_version_match.as_deref() + } + + /// Return the version that the current object must not match before deletion. + pub fn if_version_not_match(&self) -> Option<&str> { + self.if_version_not_match.as_deref() + } } impl From for OpDelete { @@ -103,6 +145,9 @@ impl From for OpDelete { version: value.version, recursive: value.recursive, if_match: value.if_match, + if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, } } } @@ -334,6 +379,8 @@ impl From for PresignOperation { pub struct OpRead { if_match: Option, if_none_match: Option, + if_version_match: Option, + if_version_not_match: Option, if_modified_since: Option, if_unmodified_since: Option, override_content_type: Option, @@ -405,6 +452,28 @@ impl OpRead { self.if_none_match.as_deref() } + /// Set the version that the current object must match. + pub fn with_if_version_match(mut self, version: &str) -> Self { + self.if_version_match = Some(version.to_string()); + self + } + + /// Get the version match condition. + pub fn if_version_match(&self) -> Option<&str> { + self.if_version_match.as_deref() + } + + /// Set the version that the current object must not match. + pub fn with_if_version_not_match(mut self, version: &str) -> Self { + self.if_version_not_match = Some(version.to_string()); + self + } + + /// Get the version non-match condition. + pub fn if_version_not_match(&self) -> Option<&str> { + self.if_version_not_match.as_deref() + } + /// Set the If-Modified-Since of the option pub fn with_if_modified_since(mut self, v: Timestamp) -> Self { self.if_modified_since = Some(v); @@ -528,6 +597,8 @@ impl From for (BytesRange, OpRead, OpReader) { OpRead { if_match: value.if_match, if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, if_modified_since: value.if_modified_since, if_unmodified_since: value.if_unmodified_since, override_content_type: value.override_content_type, @@ -553,6 +624,8 @@ impl From for (OpRead, OpReader) { OpRead { if_match: value.if_match, if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, if_modified_since: value.if_modified_since, if_unmodified_since: value.if_unmodified_since, override_content_type: None, @@ -577,6 +650,8 @@ impl From for (OpRead, OpReader) { pub struct OpStat { if_match: Option, if_none_match: Option, + if_version_match: Option, + if_version_not_match: Option, if_modified_since: Option, if_unmodified_since: Option, override_content_type: Option, @@ -613,6 +688,28 @@ impl OpStat { self.if_none_match.as_deref() } + /// Set the version that the current object must match. + pub fn with_if_version_match(mut self, version: &str) -> Self { + self.if_version_match = Some(version.to_string()); + self + } + + /// Get the version match condition. + pub fn if_version_match(&self) -> Option<&str> { + self.if_version_match.as_deref() + } + + /// Set the version that the current object must not match. + pub fn with_if_version_not_match(mut self, version: &str) -> Self { + self.if_version_not_match = Some(version.to_string()); + self + } + + /// Get the version non-match condition. + pub fn if_version_not_match(&self) -> Option<&str> { + self.if_version_not_match.as_deref() + } + /// Set the If-Modified-Since of the option pub fn with_if_modified_since(mut self, v: Timestamp) -> Self { self.if_modified_since = Some(v); @@ -686,6 +783,8 @@ impl From for OpStat { Self { if_match: value.if_match, if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, if_modified_since: value.if_modified_since, if_unmodified_since: value.if_unmodified_since, override_content_type: value.override_content_type, @@ -707,6 +806,8 @@ pub struct OpWrite { cache_control: Option, if_match: Option, if_none_match: Option, + if_version_match: Option, + if_version_not_match: Option, if_not_exists: bool, user_metadata: Option>, } @@ -815,6 +916,28 @@ impl OpWrite { self.if_none_match.as_deref() } + /// Set the version that the current object must match. + pub fn with_if_version_match(mut self, version: &str) -> Self { + self.if_version_match = Some(version.to_string()); + self + } + + /// Get the version match condition. + pub fn if_version_match(&self) -> Option<&str> { + self.if_version_match.as_deref() + } + + /// Set the version that the current object must not match. + pub fn with_if_version_not_match(mut self, version: &str) -> Self { + self.if_version_not_match = Some(version.to_string()); + self + } + + /// Get the version non-match condition. + pub fn if_version_not_match(&self) -> Option<&str> { + self.if_version_not_match.as_deref() + } + /// Set the If-Not-Exist of the option pub fn with_if_not_exists(mut self, b: bool) -> Self { self.if_not_exists = b; @@ -885,6 +1008,8 @@ impl From for (OpWrite, OpWriter) { cache_control: value.cache_control, if_match: value.if_match, if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, if_not_exists: value.if_not_exists, user_metadata: value.user_metadata, }, @@ -898,6 +1023,9 @@ impl From for (OpWrite, OpWriter) { pub struct OpCopy { if_not_exists: bool, if_match: Option, + if_none_match: Option, + if_version_match: Option, + if_version_not_match: Option, source_version: Option, } @@ -935,6 +1063,39 @@ impl OpCopy { self.if_match.as_deref() } + /// Set the destination ETag non-match condition. + pub fn with_if_none_match(mut self, if_none_match: impl Into) -> Self { + self.if_none_match = Some(if_none_match.into()); + self + } + + /// Get the destination ETag non-match condition. + pub fn if_none_match(&self) -> Option<&str> { + self.if_none_match.as_deref() + } + + /// Set the current destination version match condition. + pub fn with_if_version_match(mut self, version: impl Into) -> Self { + self.if_version_match = Some(version.into()); + self + } + + /// Get the current destination version match condition. + pub fn if_version_match(&self) -> Option<&str> { + self.if_version_match.as_deref() + } + + /// Set the current destination version non-match condition. + pub fn with_if_version_not_match(mut self, version: impl Into) -> Self { + self.if_version_not_match = Some(version.into()); + self + } + + /// Get the current destination version non-match condition. + pub fn if_version_not_match(&self) -> Option<&str> { + self.if_version_not_match.as_deref() + } + /// Set source version for the operation. /// /// When set, the copy operation will copy from the specified source version. @@ -1003,6 +1164,9 @@ impl From for (OpCopy, OpCopier) { OpCopy { if_not_exists: value.if_not_exists, if_match: value.if_match, + if_none_match: value.if_none_match, + if_version_match: value.if_version_match, + if_version_not_match: value.if_version_not_match, source_version: value.source_version, }, OpCopier { diff --git a/core/core/src/types/capability.rs b/core/core/src/types/capability.rs index 43ff7432a037..d07306b18353 100644 --- a/core/core/src/types/capability.rs +++ b/core/core/src/types/capability.rs @@ -67,6 +67,10 @@ pub struct Capability { pub stat_with_if_match: bool, /// Indicates if conditional stat operations using If-None-Match are supported. pub stat_with_if_none_match: bool, + /// Indicates if conditional stat operations using version match are supported. + pub stat_with_if_version_match: bool, + /// Indicates if conditional stat operations using version non-match are supported. + pub stat_with_if_version_not_match: bool, /// Indicates if conditional stat operations using If-Modified-Since are supported. pub stat_with_if_modified_since: bool, /// Indicates if conditional stat operations using If-Unmodified-Since are supported. @@ -86,6 +90,10 @@ pub struct Capability { pub read_with_if_match: bool, /// Indicates if conditional read operations using If-None-Match are supported. pub read_with_if_none_match: bool, + /// Indicates if conditional read operations using version match are supported. + pub read_with_if_version_match: bool, + /// Indicates if conditional read operations using version non-match are supported. + pub read_with_if_version_not_match: bool, /// Indicates if conditional read operations using If-Modified-Since are supported. pub read_with_if_modified_since: bool, /// Indicates if conditional read operations using If-Unmodified-Since are supported. @@ -121,6 +129,10 @@ pub struct Capability { pub write_with_if_match: bool, /// Indicates if conditional write operations using If-None-Match are supported. pub write_with_if_none_match: bool, + /// Indicates if conditional write operations using version match are supported. + pub write_with_if_version_match: bool, + /// Indicates if conditional write operations using version non-match are supported. + pub write_with_if_version_not_match: bool, /// Indicates if write operations can be conditional on object non-existence. pub write_with_if_not_exists: bool, /// Indicates if custom user metadata can be attached during write operations. @@ -146,6 +158,12 @@ pub struct Capability { pub delete_with_recursive: bool, /// Indicates if conditional delete operations using If-Match are supported. pub delete_with_if_match: bool, + /// Indicates if conditional delete operations using If-None-Match are supported. + pub delete_with_if_none_match: bool, + /// Indicates if conditional delete operations using version match are supported. + pub delete_with_if_version_match: bool, + /// Indicates if conditional delete operations using version non-match are supported. + pub delete_with_if_version_not_match: bool, /// Maximum size supported for single delete operations. pub delete_max_size: Option, @@ -155,6 +173,12 @@ pub struct Capability { pub copy_with_if_not_exists: bool, /// Indicates if conditional copy operations with if-match are supported. pub copy_with_if_match: bool, + /// Indicates if conditional copy operations with if-none-match are supported. + pub copy_with_if_none_match: bool, + /// Indicates if conditional copy operations using version match are supported. + pub copy_with_if_version_match: bool, + /// Indicates if conditional copy operations using version non-match are supported. + pub copy_with_if_version_not_match: bool, /// Indicates if copy operations from a specific source version are supported. pub copy_with_source_version: bool, /// Indicates if copy operations can be split into multiple server-side tasks. diff --git a/core/core/src/types/delete/deleter.rs b/core/core/src/types/delete/deleter.rs index 5296614812ff..8e6023e60475 100644 --- a/core/core/src/types/delete/deleter.rs +++ b/core/core/src/types/delete/deleter.rs @@ -83,28 +83,84 @@ use crate::*; /// ``` pub struct Deleter { deleter: oio::Deleter, + scheme: &'static str, + capability: Capability, } impl Deleter { pub(crate) fn create(ctx: OperationContext, srv: Servicer) -> Result { + let scheme = srv.info().scheme(); + let capability = srv.capability(); let deleter = srv.delete(&ctx)?; - Ok(Self { deleter }) + Ok(Self { + deleter, + scheme, + capability, + }) } /// Delete a path. pub async fn delete(&mut self, input: impl IntoDeleteInput) -> Result<()> { let input = input.into_delete_input(); - let mut op = OpDelete::default(); - if let Some(version) = &input.version { - op = op.with_version(version); - } - if input.recursive { - op = op.with_recursive(true); - } - if let Some(if_match) = &input.if_match { - op = op.with_if_match(if_match); + let mut opts = options::DeleteOptions { + version: input.version, + recursive: input.recursive, + if_match: input.if_match, + if_none_match: input.if_none_match, + if_version_match: input.if_version_match, + if_version_not_match: input.if_version_not_match, + if_not_changed: input.if_not_changed, + }; + if let Some(metadata) = opts.if_not_changed.take() { + if self.capability.delete_with_if_version_match + && let Some(version) = metadata.version() + { + if let Some(explicit) = opts.if_version_match.as_deref() { + if explicit != version { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_version_match", + ) + .with_operation(Operation::Delete.into_static())); + } + } else { + opts.if_version_match = Some(version.to_string()); + } + } else if self.capability.delete_with_if_match + && let Some(etag) = metadata.etag() + { + if let Some(explicit) = opts.if_match.as_deref() { + if explicit != etag { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_match", + ) + .with_operation(Operation::Delete.into_static())); + } + } else { + opts.if_match = Some(etag.to_string()); + } + } else if !self.capability.delete_with_if_version_match + && !self.capability.delete_with_if_match + { + return Err(Error::new( + ErrorKind::Unsupported, + format!( + "The service {} does not support the operation delete with if_not_changed", + self.scheme + ), + ) + .with_operation(Operation::Delete.into_static())); + } else { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "if_not_changed metadata does not contain an identity supported by delete", + ) + .with_operation(Operation::Delete.into_static())); + } } + let op = opts.into(); self.deleter.delete(&input.path, op).await?; Ok(()) @@ -238,6 +294,8 @@ mod tests { let deleter = Deleter { deleter: Box::new(mock), + scheme: "test", + capability: Capability::default(), }; let mut sink = deleter.into_sink::(); diff --git a/core/core/src/types/delete/input.rs b/core/core/src/types/delete/input.rs index e975ce0c04c2..892f56a6a6dc 100644 --- a/core/core/src/types/delete/input.rs +++ b/core/core/src/types/delete/input.rs @@ -15,8 +15,8 @@ // specific language governing permissions and limitations // under the License. -use crate::Entry; use crate::raw::OpDelete; +use crate::{Entry, Metadata}; /// DeleteInput is the input for delete operations. #[non_exhaustive] @@ -28,8 +28,68 @@ pub struct DeleteInput { pub version: Option, /// Whether to perform recursive deletion. pub recursive: bool, - /// Delete the path only when its ETag matches this value. + /// Delete only when the current ETag matches this value. pub if_match: Option, + /// Delete only when the current ETag does not match this value. + pub if_none_match: Option, + /// Delete only when the current version matches this value. + pub if_version_match: Option, + /// Delete only when the current version does not match this value. + pub if_version_not_match: Option, + /// Delete only when the object still matches this metadata. + pub if_not_changed: Option, +} + +impl DeleteInput { + /// Create a delete input for `path`. + pub fn new(path: impl Into) -> Self { + Self { + path: path.into(), + ..Default::default() + } + } + + /// Select the object version to delete. + pub fn with_version(mut self, version: impl Into) -> Self { + self.version = Some(version.into()); + self + } + + /// Configure recursive deletion. + pub fn with_recursive(mut self, recursive: bool) -> Self { + self.recursive = recursive; + self + } + + /// Delete only when the current ETag matches `etag`. + pub fn with_if_match(mut self, etag: impl Into) -> Self { + self.if_match = Some(etag.into()); + self + } + + /// Delete only when the current ETag does not match `etag`. + pub fn with_if_none_match(mut self, etag: impl Into) -> Self { + self.if_none_match = Some(etag.into()); + self + } + + /// Delete only when the current version matches `version`. + pub fn with_if_version_match(mut self, version: impl Into) -> Self { + self.if_version_match = Some(version.into()); + self + } + + /// Delete only when the current version does not match `version`. + pub fn with_if_version_not_match(mut self, version: impl Into) -> Self { + self.if_version_not_match = Some(version.into()); + self + } + + /// Delete only when the object still matches `metadata`. + pub fn with_if_not_changed(mut self, metadata: &Metadata) -> Self { + self.if_not_changed = Some(metadata.clone()); + self + } } /// IntoDeleteInput is a helper trait that makes it easier for users to play with `Deleter`. @@ -48,22 +108,14 @@ impl IntoDeleteInput for DeleteInput { /// Implement `IntoDeleteInput` for `&str` so we can use `&str` as a DeleteInput. impl IntoDeleteInput for &str { fn into_delete_input(self) -> DeleteInput { - DeleteInput { - path: self.to_string(), - recursive: false, - ..Default::default() - } + DeleteInput::new(self) } } /// Implement `IntoDeleteInput` for `String` so we can use `Vec` as a DeleteInput stream. impl IntoDeleteInput for String { fn into_delete_input(self) -> DeleteInput { - DeleteInput { - path: self, - recursive: false, - ..Default::default() - } + DeleteInput::new(self) } } @@ -82,8 +134,17 @@ impl IntoDeleteInput for (String, OpDelete) { if let Some(version) = args.version() { input.version = Some(version.to_string()); } - if let Some(if_match) = args.if_match() { - input.if_match = Some(if_match.to_string()); + if let Some(etag) = args.if_match() { + input.if_match = Some(etag.to_string()); + } + if let Some(etag) = args.if_none_match() { + input.if_none_match = Some(etag.to_string()); + } + if let Some(version) = args.if_version_match() { + input.if_version_match = Some(version.to_string()); + } + if let Some(version) = args.if_version_not_match() { + input.if_version_not_match = Some(version.to_string()); } input } diff --git a/core/core/src/types/operator/operator.rs b/core/core/src/types/operator/operator.rs index 3f6961eb0bc0..7c03753a8013 100644 --- a/core/core/src/types/operator/operator.rs +++ b/core/core/src/types/operator/operator.rs @@ -24,7 +24,7 @@ use futures::TryStreamExt; use crate::operator_futures::*; use crate::raw::*; -use crate::types::delete::Deleter; +use crate::types::delete::{DeleteInput, Deleter}; use crate::*; /// The `Operator` serves as the entry point for all public asynchronous APIs. @@ -983,10 +983,7 @@ impl Operator { ); } - let (args, opts) = opts.into(); - - let write_context = WriteContext::new(ctx, srv, path, args, opts); - let mut w = Writer::new(write_context).await?; + let mut w = Self::writer_inner(ctx, srv, path, opts).await?; w.write(bs).await?; w.close().await } @@ -1153,6 +1150,54 @@ impl Operator { ); } + let mut opts = opts; + if let Some(metadata) = opts.if_not_changed.take() { + let capability = srv.capability(); + if capability.write_with_if_version_match + && let Some(version) = metadata.version() + { + if let Some(explicit) = opts.if_version_match.as_deref() { + if explicit != version { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_version_match", + ) + .with_operation(Operation::Write.into_static())); + } + } else { + opts.if_version_match = Some(version.to_string()); + } + } else if capability.write_with_if_match + && let Some(etag) = metadata.etag() + { + if let Some(explicit) = opts.if_match.as_deref() { + if explicit != etag { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_match", + ) + .with_operation(Operation::Write.into_static())); + } + } else { + opts.if_match = Some(etag.to_string()); + } + } else if !capability.write_with_if_version_match && !capability.write_with_if_match { + return Err(Error::new( + ErrorKind::Unsupported, + format!( + "The service {} does not support the operation write with if_not_changed", + srv.info().scheme() + ), + ) + .with_operation(Operation::Write.into_static())); + } else { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "if_not_changed metadata does not contain an identity supported by write", + ) + .with_operation(Operation::Write.into_static())); + } + } let (args, opts) = opts.into(); let write_context = WriteContext::new(ctx, srv, path, args, opts); let w = Writer::new(write_context).await?; @@ -1399,6 +1444,54 @@ impl Operator { ); } + let mut opts = opts; + if let Some(metadata) = opts.if_not_changed.take() { + let capability = srv.capability(); + if capability.copy_with_if_version_match + && let Some(version) = metadata.version() + { + if let Some(explicit) = opts.if_version_match.as_deref() { + if explicit != version { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_version_match", + ) + .with_operation(Operation::Copy.into_static())); + } + } else { + opts.if_version_match = Some(version.to_string()); + } + } else if capability.copy_with_if_match + && let Some(etag) = metadata.etag() + { + if let Some(explicit) = opts.if_match.as_deref() { + if explicit != etag { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_match", + ) + .with_operation(Operation::Copy.into_static())); + } + } else { + opts.if_match = Some(etag.to_string()); + } + } else if !capability.copy_with_if_version_match && !capability.copy_with_if_match { + return Err(Error::new( + ErrorKind::Unsupported, + format!( + "The service {} does not support the operation copy with if_not_changed", + srv.info().scheme() + ), + ) + .with_operation(Operation::Copy.into_static())); + } else { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "if_not_changed metadata does not contain an identity supported by copy", + ) + .with_operation(Operation::Copy.into_static())); + } + } let (args, opts) = opts.into(); std::future::ready(Copier::create(ctx, srv, &from, &to, args, opts)).await } @@ -1639,7 +1732,8 @@ impl Operator { /// /// # Notes /// - /// - Deleting a file that does not exist won't return errors. + /// - Deleting a file that does not exist won't return errors unless a condition requires a + /// live target. /// /// # Examples /// @@ -1660,7 +1754,8 @@ impl Operator { /// /// # Notes /// - /// - Deleting a file that does not exist won't return errors. + /// - Deleting a file that does not exist won't return errors unless a condition requires a + /// live target. /// /// # Options /// @@ -1695,7 +1790,8 @@ impl Operator { /// /// # Notes /// - /// - Deleting a file that does not exist won't return errors. + /// - Deleting a file that does not exist won't return errors unless a condition requires a + /// live target. /// /// # Examples /// @@ -1726,10 +1822,20 @@ impl Operator { path: String, opts: options::DeleteOptions, ) -> Result<()> { - let mut deleter = srv.delete(&ctx)?; - let args = opts.into(); - deleter.delete_dyn(&path, args).await?; - deleter.close_dyn().await?; + let mut deleter = Deleter::create(ctx, srv)?; + deleter + .delete(DeleteInput { + path, + version: opts.version, + recursive: opts.recursive, + if_match: opts.if_match, + if_none_match: opts.if_none_match, + if_version_match: opts.if_version_match, + if_version_not_match: opts.if_version_not_match, + if_not_changed: opts.if_not_changed, + }) + .await?; + deleter.close().await?; Ok(()) } @@ -2547,6 +2653,54 @@ impl Operator { path: String, (opts, expire): (options::WriteOptions, Duration), ) -> Result { + let mut opts = opts; + if let Some(metadata) = opts.if_not_changed.take() { + let capability = srv.capability(); + if capability.write_with_if_version_match + && let Some(version) = metadata.version() + { + if let Some(explicit) = opts.if_version_match.as_deref() { + if explicit != version { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_version_match", + ) + .with_operation(Operation::Write.into_static())); + } + } else { + opts.if_version_match = Some(version.to_string()); + } + } else if capability.write_with_if_match + && let Some(etag) = metadata.etag() + { + if let Some(explicit) = opts.if_match.as_deref() { + if explicit != etag { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_match", + ) + .with_operation(Operation::Write.into_static())); + } + } else { + opts.if_match = Some(etag.to_string()); + } + } else if !capability.write_with_if_version_match && !capability.write_with_if_match { + return Err(Error::new( + ErrorKind::Unsupported, + format!( + "The service {} does not support the operation write with if_not_changed", + srv.info().scheme() + ), + ) + .with_operation(Operation::Write.into_static())); + } else { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "if_not_changed metadata does not contain an identity supported by write", + ) + .with_operation(Operation::Write.into_static())); + } + } let (op_write, _) = opts.into(); let op = OpPresign::new(op_write, expire); let rp = srv.presign(&ctx, &path, op).await?; @@ -2660,6 +2814,54 @@ impl Operator { path: String, (opts, expire): (options::DeleteOptions, Duration), ) -> Result { + let mut opts = opts; + if let Some(metadata) = opts.if_not_changed.take() { + let capability = srv.capability(); + if capability.delete_with_if_version_match + && let Some(version) = metadata.version() + { + if let Some(explicit) = opts.if_version_match.as_deref() { + if explicit != version { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_version_match", + ) + .with_operation(Operation::Delete.into_static())); + } + } else { + opts.if_version_match = Some(version.to_string()); + } + } else if capability.delete_with_if_match + && let Some(etag) = metadata.etag() + { + if let Some(explicit) = opts.if_match.as_deref() { + if explicit != etag { + return Err(Error::new( + ErrorKind::ConditionNotMatch, + "if_not_changed conflicts with if_match", + ) + .with_operation(Operation::Delete.into_static())); + } + } else { + opts.if_match = Some(etag.to_string()); + } + } else if !capability.delete_with_if_version_match && !capability.delete_with_if_match { + return Err(Error::new( + ErrorKind::Unsupported, + format!( + "The service {} does not support the operation delete with if_not_changed", + srv.info().scheme() + ), + ) + .with_operation(Operation::Delete.into_static())); + } else { + return Err(Error::new( + ErrorKind::ConfigInvalid, + "if_not_changed metadata does not contain an identity supported by delete", + ) + .with_operation(Operation::Delete.into_static())); + } + } let op = OpPresign::new(OpDelete::from(opts), expire); let rp = srv.presign(&ctx, &path, op).await?; Ok(rp.into_presigned_request()) diff --git a/core/core/src/types/operator/operator_futures.rs b/core/core/src/types/operator/operator_futures.rs index c92c2b01f1f0..080f57f6e117 100644 --- a/core/core/src/types/operator/operator_futures.rs +++ b/core/core/src/types/operator/operator_futures.rs @@ -101,6 +101,18 @@ impl>> FutureStat { self } + /// Refer to [`options::StatOptions::if_version_match`] for more details. + pub fn if_version_match(mut self, v: &str) -> Self { + self.args.if_version_match = Some(v.to_string()); + self + } + + /// Refer to [`options::StatOptions::if_version_not_match`] for more details. + pub fn if_version_not_match(mut self, v: &str) -> Self { + self.args.if_version_not_match = Some(v.to_string()); + self + } + /// Set the If-Modified-Since for this operation. /// /// Refer to [`options::StatOptions::if_modified_since`] for more details. @@ -378,6 +390,18 @@ impl>> FutureRead { self } + /// Set the current object version that this read must match. + pub fn if_version_match(mut self, v: &str) -> Self { + self.args.if_version_match = Some(v.to_string()); + self + } + + /// Set the current object version that this read must not match. + pub fn if_version_not_match(mut self, v: &str) -> Self { + self.args.if_version_not_match = Some(v.to_string()); + self + } + /// ## `if_modified_since` /// /// Set `if_modified_since` for this `read` request. @@ -580,6 +604,18 @@ impl>> FutureReader { self } + /// Set the current object version that this reader must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.if_version_match = Some(version.to_string()); + self + } + + /// Set the current object version that this reader must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.if_version_not_match = Some(version.to_string()); + self + } + /// Set `if-modified-since` for this `read` request. /// /// This feature can be used to check if the file has been modified since the given timestamp. @@ -872,6 +908,24 @@ impl>> FutureWrite { self } + /// Set the current object version that this write must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.0.if_version_match = Some(version.to_string()); + self + } + + /// Set the current object version that this write must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.0.if_version_not_match = Some(version.to_string()); + self + } + + /// Write only when the object still matches `metadata`. + pub fn if_not_changed(mut self, metadata: &Metadata) -> Self { + self.args.0.if_not_changed = Some(metadata.clone()); + self + } + /// Sets the condition that write operation will succeed only if target does not exist. /// /// Refer to [`options::WriteOptions::if_not_exists`] for more details. @@ -1209,6 +1263,24 @@ impl>> FutureWriter { self } + /// Set the current object version that this writer must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.if_version_match = Some(version.to_string()); + self + } + + /// Set the current object version that this writer must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.if_version_not_match = Some(version.to_string()); + self + } + + /// Write only when the object still matches `metadata`. + pub fn if_not_changed(mut self, metadata: &Metadata) -> Self { + self.args.if_not_changed = Some(metadata.clone()); + self + } + /// Sets the condition that write operation will succeed only if target does not exist. /// /// Refer to [`options::WriteOptions::if_not_exists`] for more details. @@ -1290,6 +1362,30 @@ impl>> FutureDelete { self.args.if_match = Some(etag.to_string()); self } + + /// Set `if_none_match` for this delete operation. + pub fn if_none_match(mut self, etag: &str) -> Self { + self.args.if_none_match = Some(etag.to_string()); + self + } + + /// Set the current object version that this delete must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.if_version_match = Some(version.to_string()); + self + } + + /// Set the current object version that this delete must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.if_version_not_match = Some(version.to_string()); + self + } + + /// Delete only when the object still matches `metadata`. + pub fn if_not_changed(mut self, metadata: &Metadata) -> Self { + self.args.if_not_changed = Some(metadata.clone()); + self + } } /// This future creates a [`Deleter`]. @@ -1457,6 +1553,30 @@ impl>> FutureCopy { self } + /// Set the destination ETag that this copy must not match. + pub fn if_none_match(mut self, etag: &str) -> Self { + self.args.0.if_none_match = Some(etag.to_string()); + self + } + + /// Set the current destination version that this copy must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.0.if_version_match = Some(version.to_string()); + self + } + + /// Set the current destination version that this copy must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.0.if_version_not_match = Some(version.to_string()); + self + } + + /// Copy only when the destination still matches `metadata`. + pub fn if_not_changed(mut self, metadata: &Metadata) -> Self { + self.args.0.if_not_changed = Some(metadata.clone()); + self + } + /// Sets source version for this copy operation. /// /// Refer to [`options::CopyOptions::source_version`] for more details. @@ -1513,6 +1633,30 @@ impl>> FutureCopier { self } + /// Set the destination ETag that this copier must not match. + pub fn if_none_match(mut self, etag: &str) -> Self { + self.args.0.if_none_match = Some(etag.to_string()); + self + } + + /// Set the current destination version that this copier must match. + pub fn if_version_match(mut self, version: &str) -> Self { + self.args.0.if_version_match = Some(version.to_string()); + self + } + + /// Set the current destination version that this copier must not match. + pub fn if_version_not_match(mut self, version: &str) -> Self { + self.args.0.if_version_not_match = Some(version.to_string()); + self + } + + /// Copy only when the destination still matches `metadata`. + pub fn if_not_changed(mut self, metadata: &Metadata) -> Self { + self.args.0.if_not_changed = Some(metadata.clone()); + self + } + /// Sets source version for this copier operation. /// /// Refer to [`options::CopyOptions::source_version`] for more details. diff --git a/core/core/src/types/options.rs b/core/core/src/types/options.rs index 7569312d1f43..5249dd185f35 100644 --- a/core/core/src/types/options.rs +++ b/core/core/src/types/options.rs @@ -17,6 +17,7 @@ //! Options module provides options definitions for operations. +use crate::Metadata; use crate::raw::Timestamp; use crate::types::BytesRange; use std::collections::HashMap; @@ -44,6 +45,23 @@ pub struct DeleteOptions { /// - If supported, the delete will only succeed when the existing object's /// ETag matches the given value. pub if_match: Option, + /// Delete only when the current object's ETag does not match this value. + /// + /// Check [`crate::Capability::delete_with_if_none_match`] before using this feature. + pub if_none_match: Option, + /// Delete only when the current object's version matches this value. + /// + /// Check [`crate::Capability::delete_with_if_version_match`] before using this feature. + pub if_version_match: Option, + /// Delete only when the current object's version does not match this value. + /// + /// Check [`crate::Capability::delete_with_if_version_not_match`] before using this feature. + pub if_version_not_match: Option, + /// Delete only when the object still matches the supplied metadata. + /// + /// OpenDAL prefers `version` when the service supports version preconditions and + /// otherwise uses `ETag`. + pub if_not_changed: Option, } /// Options for list operations. @@ -103,6 +121,10 @@ pub struct ReadOptions { /// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`] /// will be returned. pub if_none_match: Option, + /// Read only when the current object's version matches this value. + pub if_version_match: Option, + /// Read only when the current object's version does not match this value. + pub if_version_not_match: Option, /// Set `if_modified_since` for this operation. /// /// This option can be used to check if the file has been modified since the given timestamp. @@ -200,6 +222,10 @@ pub struct ReaderOptions { /// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`] /// will be returned. pub if_none_match: Option, + /// Read only when the current object's version matches this value. + pub if_version_match: Option, + /// Read only when the current object's version does not match this value. + pub if_version_not_match: Option, /// Set `if_modified_since` for this operation. /// /// This option can be used to check if the file has been modified since the given timestamp. @@ -298,6 +324,10 @@ pub struct StatOptions { /// If file exists and it's etag match, an error with kind [`crate::ErrorKind::ConditionNotMatch`] /// will be returned. pub if_none_match: Option, + /// Stat only when the current object's version matches this value. + pub if_version_match: Option, + /// Stat only when the current object's version does not match this value. + pub if_version_not_match: Option, /// Set `if_modified_since` for this operation. /// /// This option can be used to check if the file has been modified since the given timestamp. @@ -485,6 +515,14 @@ pub struct WriteOptions { /// This operation provides conditional write functionality based on ETag non-matching, /// useful for preventing overwriting existing resources or ensuring unique writes. pub if_none_match: Option, + /// Write only when the current object's version matches this value. + /// + /// Check [`crate::Capability::write_with_if_version_match`] before using this feature. + pub if_version_match: Option, + /// Write only when the current object's version does not match this value. + /// + /// Check [`crate::Capability::write_with_if_version_not_match`] before using this feature. + pub if_version_not_match: Option, /// Sets the condition that write operation will succeed only if target does not exist. /// /// ### Capability @@ -500,6 +538,11 @@ pub struct WriteOptions { /// This operation provides a way to ensure write operations only create new resources /// without overwriting existing ones, useful for implementing "create if not exists" logic. pub if_not_exists: bool, + /// Write only when the object still matches the supplied metadata. + /// + /// OpenDAL prefers `version` when the service supports version preconditions and + /// otherwise uses `ETag`. + pub if_not_changed: Option, /// Sets concurrent write operations for this writer. /// @@ -585,6 +628,23 @@ pub struct CopyOptions { /// - If supported, the copy operation will only succeed when the existing /// destination object's ETag matches the given value. pub if_match: Option, + /// Copy only when the destination ETag does not match this value. + /// + /// Check [`crate::Capability::copy_with_if_none_match`] before using this feature. + pub if_none_match: Option, + /// Copy only when the current destination version matches this value. + /// + /// Check [`crate::Capability::copy_with_if_version_match`] before using this feature. + pub if_version_match: Option, + /// Copy only when the current destination version does not match this value. + /// + /// Check [`crate::Capability::copy_with_if_version_not_match`] before using this feature. + pub if_version_not_match: Option, + /// Copy only when the destination still matches the supplied metadata. + /// + /// OpenDAL prefers `version` when the service supports version preconditions and + /// otherwise uses `ETag`. + pub if_not_changed: Option, /// Copy from a specific source object version. /// diff --git a/core/services/azblob/src/backend.rs b/core/services/azblob/src/backend.rs index 6cece64c2beb..c94807cdde62 100644 --- a/core/services/azblob/src/backend.rs +++ b/core/services/azblob/src/backend.rs @@ -415,10 +415,13 @@ impl Builder for AzblobBuilder { delete: true, delete_with_if_match: true, + delete_with_if_none_match: true, delete_max_size: Some(AZBLOB_BATCH_LIMIT), copy: true, copy_with_if_not_exists: true, + copy_with_if_match: true, + copy_with_if_none_match: true, copy_can_multi: true, copy_multi_min_size: Some(AZBLOB_COPY_MIN_BLOCK_SIZE), copy_multi_max_size: Some(AZBLOB_COPY_MAX_BLOCK_SIZE), @@ -617,9 +620,9 @@ impl Service for AzblobBackend { let req = match args.operation() { PresignOperation::Stat(v) => self.core.azblob_head_blob_request(path, v), PresignOperation::Read(range, v) => self.core.azblob_get_blob_request(path, *range, v), - PresignOperation::Write(_) => { + PresignOperation::Write(v) => { self.core - .azblob_put_blob_request(path, None, &OpWrite::default(), Buffer::new()) + .azblob_put_blob_request(path, None, v, Buffer::new()) } PresignOperation::Delete(_) => Err(Error::new( ErrorKind::Unsupported, diff --git a/core/services/azblob/src/core.rs b/core/services/azblob/src/core.rs index 89d04da5779e..457e6b9ca818 100644 --- a/core/services/azblob/src/core.rs +++ b/core/services/azblob/src/core.rs @@ -306,7 +306,6 @@ impl AzblobCore { if args.if_not_exists() { req = req.header(IF_NONE_MATCH, "*"); } - if let Some(v) = args.if_none_match() { req = req.header(IF_NONE_MATCH, v); } @@ -667,6 +666,12 @@ impl AzblobCore { if args.if_not_exists() { req = req.header(IF_NONE_MATCH, "*"); } + if let Some(if_match) = args.if_match() { + req = req.header(IF_MATCH, if_match); + } + if let Some(if_none_match) = args.if_none_match() { + req = req.header(IF_NONE_MATCH, if_none_match); + } let content = quick_xml::se::to_string(&PutBlockListRequest { latest: block_ids @@ -743,10 +748,12 @@ impl AzblobCore { fn azblob_delete_blob_request(&self, path: &str, args: &OpDelete) -> Result> { let mut req = Request::delete(self.build_path_url(path)); - if let Some(if_match) = args.if_match() { req = req.header(IF_MATCH, if_match); } + if let Some(if_none_match) = args.if_none_match() { + req = req.header(IF_NONE_MATCH, if_none_match); + } req.header(CONTENT_LENGTH, 0) .extension(Operation::Delete) @@ -789,6 +796,12 @@ impl AzblobCore { if args.if_not_exists() { req = req.header(IF_NONE_MATCH, "*"); } + if let Some(if_match) = args.if_match() { + req = req.header(IF_MATCH, if_match); + } + if let Some(if_none_match) = args.if_none_match() { + req = req.header(IF_NONE_MATCH, if_none_match); + } let req = req .extension(Operation::Copy) diff --git a/core/services/azblob/src/deleter.rs b/core/services/azblob/src/deleter.rs index d1aa54c3ddfe..b3c2f7b04692 100644 --- a/core/services/azblob/src/deleter.rs +++ b/core/services/azblob/src/deleter.rs @@ -46,10 +46,16 @@ impl oio::BatchDelete for AzblobDeleter { 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( 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, )), } @@ -95,12 +101,23 @@ impl oio::BatchDelete for AzblobDeleter { for (part, (path, args)) in parts.into_iter().zip(batch) { let resp = part.into_response(); - // 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, + 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, args)); } else { 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 --git a/core/services/gcs/src/backend.rs b/core/services/gcs/src/backend.rs index 4469339f7d67..4f4175365159 100644 --- a/core/services/gcs/src/backend.rs +++ b/core/services/gcs/src/backend.rs @@ -45,6 +45,7 @@ use super::core::*; use super::deleter::GcsDeleter; use super::lister::GcsLister; use super::reader::*; +use super::writer::GcsConditionalWriter; use super::writer::GcsWriter; use super::writer::GcsWriters; use opendal_core::raw::*; @@ -343,12 +344,16 @@ impl Builder for GcsBuilder { stat: true, stat_with_if_match: true, stat_with_if_none_match: true, + stat_with_if_version_match: true, + stat_with_if_version_not_match: true, read: true, read_with_suffix: true, read_with_if_match: true, read_with_if_none_match: true, + read_with_if_version_match: true, + read_with_if_version_not_match: true, write: true, write_can_empty: true, @@ -358,6 +363,8 @@ impl Builder for GcsBuilder { write_with_content_encoding: true, write_with_user_metadata: true, write_with_if_not_exists: true, + write_with_if_version_match: true, + write_with_if_version_not_match: true, // The min multipart size of Gcs is 5 MiB. // @@ -373,9 +380,13 @@ impl Builder for GcsBuilder { }, delete: true, + delete_with_if_version_match: true, + delete_with_if_version_not_match: true, delete_max_size: Some(100), copy: true, + copy_with_if_version_match: true, + copy_with_if_version_not_match: true, copy_can_multi: true, // GCS rewrite requires maxBytesRewrittenPerCall to be an // integral multiple of 1 MiB if specified. @@ -454,7 +465,9 @@ impl Service for GcsBackend { async fn stat(&self, ctx: &OperationContext, path: &str, args: OpStat) -> Result { 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() { @@ -480,15 +493,24 @@ impl Service for GcsBackend { } fn write(&self, ctx: &OperationContext, path: &str, args: OpWrite) -> Result { - let output: GcsWriters = { + let conditional = args.if_not_exists() + || args.if_version_match().is_some() + || args.if_version_not_match().is_some(); + let output: GcsWriters = if conditional { + TwoWays::Two(GcsConditionalWriter::new( + self.core.clone(), + ctx.clone(), + path, + args, + )) + } else { let concurrent = args.concurrent(); let w = GcsWriter::new(self.core.clone(), ctx.clone(), path, args); // Multipart uploads schedule work through the operation executor // supplied by the caller. let w = oio::MultipartWriter::new(ctx.executor().clone(), w, concurrent); - - Ok(w) - }?; + TwoWays::One(w) + }; Ok(output) } diff --git a/core/services/gcs/src/copier.rs b/core/services/gcs/src/copier.rs index 477f5c1c06b9..eb239309ff6f 100644 --- a/core/services/gcs/src/copier.rs +++ b/core/services/gcs/src/copier.rs @@ -92,7 +92,11 @@ 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_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, )); } diff --git a/core/services/gcs/src/core.rs b/core/services/gcs/src/core.rs index c41b1322c130..a0796336d137 100644 --- a/core/services/gcs/src/core.rs +++ b/core/services/gcs/src/core.rs @@ -45,6 +45,7 @@ use opendal_core::raw::*; use opendal_core::*; pub mod constants { + pub const X_GOOG_GENERATION: &str = "x-goog-generation"; pub const GCS_REWRITE_MIN_CHUNK_SIZE: usize = 1024 * 1024; #[cfg(target_pointer_width = "64")] pub const GCS_REWRITE_MAX_CHUNK_SIZE: usize = @@ -164,12 +165,21 @@ impl GcsCore { let p = build_abs_path(&self.root, path); let url = format!( - "{}/storage/v1/b/{}/o/{}?alt=media", + "{}/storage/v1/b/{}/o/{}", self.endpoint, self.bucket, gcs_percent_encode_path(&p) ); + let mut url = QueryPairsWriter::new(&url).push("alt", "media"); + if let Some(version) = args.if_version_match() { + url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version)); + } + if let Some(version) = args.if_version_not_match() { + url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version)); + } + let url = url.finish(); + let mut req = Request::get(&url); if let Some(if_match) = args.if_match() { @@ -210,6 +220,9 @@ impl GcsCore { if let Some(if_none_match) = args.if_none_match() { req = req.header(IF_NONE_MATCH, if_none_match); } + if let Some(version) = args.if_version_match() { + req = req.header("x-goog-if-generation-match", version); + } if let Some(if_modified_since) = args.if_modified_since() { req = req.header(IF_MODIFIED_SINCE, if_modified_since.format_http_date()); @@ -277,12 +290,24 @@ impl GcsCore { write!(&mut url, "&predefinedAcl={acl}").unwrap(); } - // Makes the operation conditional on whether the object's current generation - // matches the given value. Setting to 0 makes the operation succeed only if - // there are no live versions of the object. - if op.if_not_exists() { + if let Some(version) = op.if_version_match() { + write!( + &mut url, + "&ifGenerationMatch={}", + gcs_percent_encode_path(version) + ) + .unwrap(); + } else if op.if_not_exists() { write!(&mut url, "&ifGenerationMatch=0").unwrap(); } + if let Some(version) = op.if_version_not_match() { + write!( + &mut url, + "&ifGenerationNotMatch={}", + gcs_percent_encode_path(version) + ) + .unwrap(); + } let mut req = Request::post(&url); @@ -331,6 +356,105 @@ impl GcsCore { } } + pub fn gcs_initiate_resumable_upload_request( + &self, + path: &str, + op: &OpWrite, + ) -> Result> { + let p = build_abs_path(&self.root, path); + let base = format!("{}/upload/storage/v1/b/{}/o", self.endpoint, self.bucket); + let mut url = QueryPairsWriter::new(&base) + .push("uploadType", "resumable") + .push("name", &gcs_percent_encode_path(&p)); + + if let Some(acl) = &self.predefined_acl { + url = url.push("predefinedAcl", acl); + } + if let Some(version) = op.if_version_match() { + url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version)); + } else if op.if_not_exists() { + url = url.push("ifGenerationMatch", "0"); + } + if let Some(version) = op.if_version_not_match() { + url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version)); + } + + let metadata = InsertRequestMetadata { + storage_class: self.default_storage_class.as_deref(), + cache_control: op.cache_control(), + content_type: op.content_type(), + content_encoding: op.content_encoding(), + metadata: op.user_metadata(), + }; + let body = serde_json::to_vec(&metadata).map_err(new_json_serialize_error)?; + + let mut req = Request::post(url.finish()) + .header(CONTENT_TYPE, "application/json; charset=UTF-8") + .header(CONTENT_LENGTH, body.len()); + if let Some(content_type) = op.content_type() { + req = req.header("x-upload-content-type", content_type); + } + + req.extension(Operation::Write) + .extension(ServiceOperation("InitiateResumableUpload")) + .body(Buffer::from(Bytes::from(body))) + .map_err(new_request_build_error) + } + + pub async fn gcs_initiate_resumable_upload( + &self, + ctx: &OperationContext, + path: &str, + op: &OpWrite, + ) -> Result> { + let req = self.gcs_initiate_resumable_upload_request(path, op)?; + let req = self.sign(ctx, req).await?; + self.send(ctx, req).await + } + + pub async fn gcs_upload_resumable_chunk( + &self, + ctx: &OperationContext, + session_uri: &str, + offset: u64, + body: Buffer, + total: Option, + ) -> Result> { + let end = offset + body.len() as u64; + let content_range = if body.is_empty() { + format!("bytes */{}", total.unwrap_or_default()) + } else { + format!( + "bytes {}-{}/{}", + offset, + end - 1, + total.map_or_else(|| "*".to_string(), |v| v.to_string()) + ) + }; + let req = Request::put(session_uri) + .header(CONTENT_LENGTH, body.len()) + .header("content-range", content_range) + .extension(Operation::Write) + .extension(ServiceOperation("UploadResumableChunk")) + .body(body) + .map_err(new_request_build_error)?; + self.send(ctx, req).await + } + + pub async fn gcs_cancel_resumable_upload( + &self, + ctx: &OperationContext, + session_uri: &str, + ) -> Result> { + let req = Request::delete(session_uri) + .header(CONTENT_LENGTH, 0) + .extension(Operation::Write) + .extension(ServiceOperation("CancelResumableUpload")) + .body(Buffer::new()) + .map_err(new_request_build_error)?; + self.send(ctx, req).await + } + // It's for presign operation. Gcs only supports query sign over XML API. pub fn gcs_insert_object_xml_request( &self, @@ -370,6 +494,18 @@ impl GcsCore { req = req.header(X_GOOG_STORAGE_CLASS, storage_class); } + if let Some(version) = args.if_version_match() { + req = req.header("x-goog-if-generation-match", version); + } else if args.if_not_exists() { + req = req.header("x-goog-if-generation-match", "0"); + } + if let Some(if_match) = args.if_match() { + req = req.header(IF_MATCH, if_match); + } + if let Some(if_none_match) = args.if_none_match() { + req = req.header(IF_NONE_MATCH, if_none_match); + } + let req = req .extension(Operation::Write) .extension(ServiceOperation("InsertObject")); @@ -389,6 +525,15 @@ impl GcsCore { gcs_percent_encode_path(&p) ); + let mut url = QueryPairsWriter::new(&url); + if let Some(version) = args.if_version_match() { + url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version)); + } + if let Some(version) = args.if_version_not_match() { + url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version)); + } + let url = url.finish(); + let mut req = Request::get(&url); if let Some(if_none_match) = args.if_none_match() { @@ -398,7 +543,6 @@ impl GcsCore { if let Some(if_match) = args.if_match() { req = req.header(IF_MATCH, if_match); } - let req = req .extension(Operation::Stat) .extension(ServiceOperation("GetObject")); @@ -427,6 +571,9 @@ impl GcsCore { if let Some(if_match) = args.if_match() { req = req.header(IF_MATCH, if_match); } + if let Some(version) = args.if_version_match() { + req = req.header("x-goog-if-generation-match", version); + } let req = req .extension(Operation::Stat) @@ -454,14 +601,19 @@ impl GcsCore { &self, ctx: &OperationContext, path: &str, + args: &OpDelete, ) -> Result> { - let req = self.gcs_delete_object_request(path)?; + let req = self.gcs_delete_object_request(path, args)?; let req = self.sign(ctx, req).await?; self.send(ctx, req).await } - pub fn gcs_delete_object_request(&self, path: &str) -> Result> { + pub fn gcs_delete_object_request( + &self, + path: &str, + args: &OpDelete, + ) -> Result> { let p = build_abs_path(&self.root, path); let url = format!( @@ -471,6 +623,18 @@ impl GcsCore { gcs_percent_encode_path(&p) ); + let mut url = QueryPairsWriter::new(&url); + if let Some(version) = args.version() { + url = url.push("generation", &gcs_percent_encode_path(version)); + } + if let Some(version) = args.if_version_match() { + url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version)); + } + if let Some(version) = args.if_version_not_match() { + url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version)); + } + let url = url.finish(); + Request::delete(&url) .extension(Operation::Delete) .extension(ServiceOperation("DeleteObject")) @@ -481,14 +645,14 @@ impl GcsCore { pub async fn gcs_delete_objects( &self, ctx: &OperationContext, - paths: Vec, + paths: &[(String, OpDelete)], ) -> Result> { let uri = format!("{}/batch/storage/v1", self.endpoint); let mut multipart = Multipart::new(); - for (idx, path) in paths.iter().enumerate() { - let req = self.gcs_delete_object_request(path)?; + for (idx, (path, args)) in paths.iter().enumerate() { + let req = self.gcs_delete_object_request(path, args)?; multipart = multipart.part( MixedPart::from_request(req).part_header("content-id".parse().unwrap(), idx.into()), @@ -504,15 +668,14 @@ impl GcsCore { self.send(ctx, req).await } - pub async fn gcs_rewrite_object( + pub fn gcs_rewrite_object_request( &self, - ctx: &OperationContext, from: &str, to: &str, args: &OpCopy, max_bytes_rewritten_per_call: Option, rewrite_token: Option<&str>, - ) -> Result> { + ) -> Result> { let source = build_abs_path(&self.root, from); let dest = build_abs_path(&self.root, to); @@ -530,9 +693,14 @@ impl GcsCore { if let Some(version) = args.source_version() { url = url.push("sourceGeneration", &gcs_percent_encode_path(version)); } - if args.if_not_exists() { + if let Some(version) = args.if_version_match() { + url = url.push("ifGenerationMatch", &gcs_percent_encode_path(version)); + } else if args.if_not_exists() { url = url.push("ifGenerationMatch", "0"); } + if let Some(version) = args.if_version_not_match() { + url = url.push("ifGenerationNotMatch", &gcs_percent_encode_path(version)); + } if let Some(max_bytes) = max_bytes_rewritten_per_call { url = url.push("maxBytesRewrittenPerCall", &max_bytes.to_string()); } @@ -540,13 +708,30 @@ impl GcsCore { url = url.push("rewriteToken", &gcs_percent_encode_path(token)); } - let req = Request::post(url.finish()) + Request::post(url.finish()) .header(CONTENT_LENGTH, 0) .extension(Operation::Copy) .extension(ServiceOperation("RewriteObject")) .body(Buffer::new()) - .map_err(new_request_build_error)?; + .map_err(new_request_build_error) + } + pub async fn gcs_rewrite_object( + &self, + ctx: &OperationContext, + from: &str, + to: &str, + args: &OpCopy, + max_bytes_rewritten_per_call: Option, + rewrite_token: Option<&str>, + ) -> Result> { + let req = self.gcs_rewrite_object_request( + from, + to, + args, + max_bytes_rewritten_per_call, + rewrite_token, + )?; let req = self.sign(ctx, req).await?; self.send(ctx, req).await } @@ -869,6 +1054,7 @@ pub struct ListResponseItem { pub size: String, // metadata pub etag: String, + pub generation: String, pub md5_hash: String, pub updated: String, pub content_type: String, @@ -962,8 +1148,7 @@ mod tests { use reqsign_google::RequestSigner; use reqsign_google::TokenCredentialProvider; - #[tokio::test] - async fn test_insert_object_signing_preserves_wire_uri() { + fn test_core() -> GcsCore { let sign_ctx = Context::new(); let signer = Signer::new( sign_ctx.clone(), @@ -971,7 +1156,7 @@ mod tests { RequestSigner::new("storage"), ); - let core = GcsCore { + GcsCore { info: ServiceInfo::new("gcs", "/", "test-bucket"), capability: Capability::default(), endpoint: "https://storage.googleapis.com".to_string(), @@ -982,7 +1167,12 @@ mod tests { predefined_acl: None, default_storage_class: None, skip_signature: false, - }; + } + } + + #[tokio::test] + async fn test_insert_object_signing_preserves_wire_uri() { + let core = test_core(); let req = core .gcs_insert_object_request( "nested/object #1.txt", @@ -1008,6 +1198,168 @@ mod tests { assert_eq!(signed.uri(), &original_uri); } + #[test] + fn test_generation_preconditions_are_mapped_to_json_api() { + let core = test_core(); + + let read = core + .gcs_get_object_request( + "object", + BytesRange::default(), + &OpRead::default().with_if_version_match("123"), + ) + .expect("read request must build"); + assert!( + read.uri() + .query() + .unwrap() + .contains("ifGenerationMatch=123") + ); + + let read_zero = core + .gcs_get_object_request( + "object", + BytesRange::default(), + &OpRead::default().with_if_version_match("0"), + ) + .expect("generation zero must be forwarded"); + assert!( + read_zero + .uri() + .query() + .unwrap() + .contains("ifGenerationMatch=0") + ); + + let stat = core + .gcs_head_object_request( + "object", + &OpStat::default().with_if_version_not_match("456"), + ) + .expect("stat request must build"); + assert!( + stat.uri() + .query() + .unwrap() + .contains("ifGenerationNotMatch=456") + ); + + let write = core + .gcs_insert_object_request( + "object", + Some(0), + &OpWrite::default() + .with_if_not_exists(true) + .with_if_version_match("123"), + Buffer::new(), + ) + .expect("write request must build"); + let query = write.uri().query().unwrap(); + assert!(query.contains("ifGenerationMatch=123")); + assert_eq!(query.matches("ifGenerationMatch=").count(), 1); + + let delete = core + .gcs_delete_object_request( + "object", + &OpDelete::default().with_if_version_not_match("456"), + ) + .expect("delete request must build"); + assert!( + delete + .uri() + .query() + .unwrap() + .contains("ifGenerationNotMatch=456") + ); + + let resumable = core + .gcs_initiate_resumable_upload_request( + "object", + &OpWrite::default() + .with_if_not_exists(true) + .with_if_version_match("123"), + ) + .expect("resumable request must build"); + let query = resumable.uri().query().unwrap(); + assert!(query.contains("ifGenerationMatch=123")); + assert_eq!(query.matches("ifGenerationMatch=").count(), 1); + + let rewrite = core + .gcs_rewrite_object_request( + "source", + "target", + &OpCopy::default().with_if_version_not_match("456"), + Some(GCS_REWRITE_MIN_CHUNK_SIZE), + Some("token"), + ) + .expect("rewrite request must build"); + let query = rewrite.uri().query().unwrap(); + assert!(query.contains("ifGenerationNotMatch=456")); + assert!(query.contains("rewriteToken=token")); + + let rewrite = core + .gcs_rewrite_object_request( + "source", + "target", + &OpCopy::default() + .with_if_not_exists(true) + .with_if_version_match("123"), + None, + None, + ) + .expect("rewrite request must build"); + let query = rewrite.uri().query().unwrap(); + assert!(query.contains("ifGenerationMatch=123")); + assert_eq!(query.matches("ifGenerationMatch=").count(), 1); + } + + #[test] + fn test_generation_match_is_mapped_to_xml_api() { + let core = test_core(); + + let read = core + .gcs_get_object_xml_request( + "object", + BytesRange::default(), + &OpRead::default().with_if_version_match("123"), + ) + .expect("read request must build"); + assert_eq!(read.headers()["x-goog-if-generation-match"], "123"); + + let stat = core + .gcs_head_object_xml_request("object", &OpStat::default().with_if_version_match("123")) + .expect("stat request must build"); + assert_eq!(stat.headers()["x-goog-if-generation-match"], "123"); + + let write = core + .gcs_insert_object_xml_request( + "object", + &OpWrite::default() + .with_if_not_exists(true) + .with_if_version_match("123"), + Buffer::new(), + ) + .expect("write request must build"); + assert_eq!(write.headers()["x-goog-if-generation-match"], "123"); + assert_eq!( + write + .headers() + .get_all("x-goog-if-generation-match") + .iter() + .count(), + 1 + ); + + let create = core + .gcs_insert_object_xml_request( + "object", + &OpWrite::default().with_if_not_exists(true), + Buffer::new(), + ) + .expect("create request must build"); + assert_eq!(create.headers()["x-goog-if-generation-match"], "0"); + } + #[test] fn test_deserialize_get_object_json_response() { let content = r#"{ @@ -1120,6 +1472,7 @@ mod tests { assert_eq!(output.items[0].size, "56535"); assert_eq!(output.items[0].md5_hash, "fHcEH1vPwA6eTPqxuasXcg=="); assert_eq!(output.items[0].etag, "CKWasoTgyPkCEAE="); + assert_eq!(output.items[0].generation, "1660563214863653"); assert_eq!(output.items[0].updated, "2022-08-15T11:33:34.866Z"); assert_eq!(output.items[1].name, "2.png"); assert_eq!(output.items[1].size, "45506"); @@ -1207,6 +1560,8 @@ pub struct ErrorContext { if_match: bool, if_none_match: bool, if_not_exists: bool, + if_version_match: bool, + if_version_not_match: bool, } impl ErrorContext { @@ -1216,6 +1571,8 @@ impl ErrorContext { if_match: false, if_none_match: false, if_not_exists: false, + if_version_match: false, + if_version_not_match: false, } } @@ -1234,8 +1591,26 @@ impl ErrorContext { self } + pub const fn with_if_version_match(mut self, if_version_match: bool) -> Self { + self.if_version_match = if_version_match; + 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; + self + } + 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 + } + + const fn has_version_condition(self) -> bool { + self.if_version_match || self.if_version_not_match } } @@ -1271,6 +1646,9 @@ pub fn parse_error(ctx: ErrorContext, resp: Response) -> Error { let gcs_error = de::from_slice::(&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() => { diff --git a/core/services/gcs/src/deleter.rs b/core/services/gcs/src/deleter.rs index 1da5adbe2ec0..c6923c84e39a 100644 --- a/core/services/gcs/src/deleter.rs +++ b/core/services/gcs/src/deleter.rs @@ -37,26 +37,36 @@ 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_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( - 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, )) } } async fn delete_batch(&self, batch: Vec<(String, OpDelete)>) -> Result { - let paths: Vec = batch.into_iter().map(|(p, _)| p).collect(); - let resp = self - .core - .gcs_delete_objects(&self.ctx, paths.clone()) - .await?; + let resp = self.core.gcs_delete_objects(&self.ctx, &batch).await?; let status = resp.status(); @@ -85,20 +95,34 @@ impl oio::BatchDelete for GcsDeleter { for (i, part) in parts.into_iter().enumerate() { let resp = part.into_response(); // TODO: maybe we can take it directly? - let path = paths[i].clone(); - - // deleting not existing objects is ok - if resp.status().is_success() || resp.status() == StatusCode::NOT_FOUND { - batched_result.succeeded.push((path, OpDelete::default())); - } 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_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 { + 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 --git a/core/services/gcs/src/lister.rs b/core/services/gcs/src/lister.rs index 34785520e305..edec6a502a88 100644 --- a/core/services/gcs/src/lister.rs +++ b/core/services/gcs/src/lister.rs @@ -121,6 +121,9 @@ impl oio::PageList for GcsLister { // set metadata fields meta.set_content_md5(object.md5_hash.as_str()); meta.set_etag(object.etag.as_str()); + if !object.generation.is_empty() { + meta.set_version(&object.generation); + } let size = object.size.parse().map_err(|e| { Error::new(ErrorKind::Unexpected, "parse u64 from list response").set_source(e) diff --git a/core/services/gcs/src/reader.rs b/core/services/gcs/src/reader.rs index 1d8f08dd5fa0..f57089d945dc 100644 --- a/core/services/gcs/src/reader.rs +++ b/core/services/gcs/src/reader.rs @@ -17,6 +17,7 @@ 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; @@ -56,7 +57,9 @@ impl oio::StreamRead for GcsReader { 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) @@ -65,10 +68,13 @@ impl oio::StreamRead for GcsReader { let status = resp.status(); let (rp, stream) = match status { - StatusCode::OK | StatusCode::PARTIAL_CONTENT => ( - RpRead::new(parse_into_metadata(path, resp.headers())?), - resp.into_body(), - ), + StatusCode::OK | StatusCode::PARTIAL_CONTENT => { + let mut metadata = parse_into_metadata(path, resp.headers())?; + if let Some(generation) = parse_header_to_str(resp.headers(), X_GOOG_GENERATION)? { + metadata.set_version(generation); + } + (RpRead::new(metadata), resp.into_body()) + } _ => { let (part, mut body) = resp.into_parts(); let buf = body.to_buffer().await?; diff --git a/core/services/gcs/src/writer.rs b/core/services/gcs/src/writer.rs index 9b6f40d88844..5f3a150525ee 100644 --- a/core/services/gcs/src/writer.rs +++ b/core/services/gcs/src/writer.rs @@ -19,16 +19,21 @@ 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::*; -pub type GcsWriters = oio::MultipartWriter; +pub type GcsWriters = TwoWays, GcsConditionalWriter>; + +const RESUMABLE_CHUNK_SIZE: usize = 8 * 1024 * 1024; pub struct GcsWriter { core: Arc, @@ -158,10 +163,14 @@ impl oio::MultipartWrite for GcsWriter { 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<()> { @@ -179,3 +188,221 @@ impl oio::MultipartWrite for GcsWriter { } } } + +pub struct GcsConditionalWriter { + core: Arc, + ctx: OperationContext, + path: String, + op: OpWrite, + session_uri: Option, + offset: u64, + pending: oio::QueueBuf, +} + +impl GcsConditionalWriter { + pub fn new(core: Arc, 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( + 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) -> Result { + 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::().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) -> Result> { + 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( + 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()), + resp, + )); + } + } + + async fn write_once(&self, body: Buffer) -> Result { + 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( + 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 { + 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( + ErrorContext::new(ServiceOperation("CancelResumableUpload")), + resp, + )) + } + } +} diff --git a/core/services/s3/src/copier.rs b/core/services/s3/src/copier.rs index 0461e4fb4ddf..6724b822b100 100644 --- a/core/services/s3/src/copier.rs +++ b/core/services/s3/src/copier.rs @@ -89,6 +89,7 @@ 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()) } } @@ -135,10 +136,18 @@ impl oio::MultipartCopy for S3Copier { // S3 may return 200 OK with an body for CopyObject. if result.etag.is_empty() { - 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)); @@ -152,10 +161,17 @@ impl oio::MultipartCopy for S3Copier { Ok(meta) } - _ => Err(parse_error( - self.error_context(ServiceOperation("CopyObject")), - 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) + } + } } } @@ -267,10 +283,18 @@ impl oio::MultipartCopy for S3Copier { quick_xml::de::from_reader(bs.as_ref()).map_err(new_xml_deserialize_error)?; // S3 may return 200 OK with an body for CompleteMultipartUpload. if ret.etag.is_empty() { - 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)); @@ -281,10 +305,20 @@ impl oio::MultipartCopy for S3Copier { Ok(meta) } - _ => Err(parse_error( - self.error_context(ServiceOperation("CompleteMultipartUpload")), - 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 --git a/core/services/s3/src/core.rs b/core/services/s3/src/core.rs index b8824c78f07b..bf5e1087ba74 100644 --- a/core/services/s3/src/core.rs +++ b/core/services/s3/src/core.rs @@ -2224,6 +2224,10 @@ pub fn parse_s3_error_code(ctx: ErrorContext, code: &str) -> Option<(ErrorKind, // Although the status code is 404, NoSuchBucket is // a config invalid error, and it's not retryable from OpenDAL. "NoSuchBucket" => Some((ErrorKind::ConfigInvalid, false)), + // S3 Express can return 200 OK with an embedded NoSuchKey error for + // CopyObject. Classify the error code explicitly because the HTTP + // status alone cannot preserve the NotFound semantics in this case. + "NoSuchKey" => Some((ErrorKind::NotFound, false)), // > Your socket connection to the server was not read from // > or written to within the timeout period." // diff --git a/core/services/s3/src/deleter.rs b/core/services/s3/src/deleter.rs index b5c44065aff1..3789603a306b 100644 --- a/core/services/s3/src/deleter.rs +++ b/core/services/s3/src/deleter.rs @@ -56,6 +56,10 @@ impl oio::BatchDelete for S3Deleter { // Allow 404 when deleting a non-existing object // This is not a standard behavior, only some s3 alike service like GCS XML API do this. // ref: + 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(error_ctx, resp)), } @@ -93,9 +97,12 @@ impl oio::BatchDelete for S3Deleter { .position(|e| e.key == abs_path && e.version_id.as_deref() == op.version()) { let error = errors.swap_remove(idx); - batched_result - .failed - .push((path, op, parse_delete_objects_result_error(error))); + let conditional = op.if_match().is_some(); + batched_result.failed.push(( + path, + op, + parse_delete_objects_result_error(error, conditional), + )); } else { batched_result.succeeded.push((path, op)); } @@ -105,12 +112,15 @@ impl oio::BatchDelete for S3Deleter { } } -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(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 --git a/core/services/s3/src/writer.rs b/core/services/s3/src/writer.rs index 6573e66eac6e..1cf260d225ee 100644 --- a/core/services/s3/src/writer.rs +++ b/core/services/s3/src/writer.rs @@ -67,6 +67,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()) } } @@ -88,10 +89,17 @@ impl oio::MultipartWrite for S3Writer { match status { StatusCode::CREATED | StatusCode::OK => Ok(meta), - _ => Err(parse_error( - self.error_context(ServiceOperation("PutObject")), - 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) + } + } } } @@ -219,19 +227,37 @@ impl oio::MultipartWrite for S3Writer { let ret: CompleteMultipartUploadResult = quick_xml::de::from_reader(bs.as_ref()).map_err(new_xml_deserialize_error)?; if !ret.code.is_empty() { - 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( + 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 --git a/core/tests/behavior/async_copy.rs b/core/tests/behavior/async_copy.rs index 490ad3f0a3f5..47b2db282c73 100644 --- a/core/tests/behavior/async_copy.rs +++ b/core/tests/behavior/async_copy.rs @@ -65,6 +65,28 @@ pub fn tests(op: &Operator, tests: &mut Vec) { )) } + if cap.read && cap.write && cap.copy && cap.copy_with_if_none_match { + tests.extend(async_trials!(op, test_copy_with_if_none_match)) + } + + if cap.read + && cap.write + && cap.copy + && cap.copy_with_if_version_match + && cap.copy_with_if_version_not_match + { + tests.extend(async_trials!(op, test_copy_with_version_conditions)) + } + + if cap.read + && cap.write + && cap.stat + && cap.copy + && (cap.copy_with_if_match || cap.copy_with_if_version_match) + { + tests.extend(async_trials!(op, test_copy_if_not_changed)) + } + if cap.read && cap.write && cap.stat && cap.copy && cap.copy_with_source_version { tests.extend(async_trials!( op, @@ -437,11 +459,147 @@ pub async fn test_copy_with_if_match_mismatch(op: Operator) -> Result<()> { sha256_digest(&target_content), ); + let missing_path = uuid::Uuid::new_v4().to_string(); + let err = op + .copy_with(&source_path, &missing_path) + .if_match("\"00000000000000000000000000000000\"") + .await + .expect_err("missing destination must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + op.delete(&source_path).await.expect("delete must succeed"); op.delete(&target_path).await.expect("delete must succeed"); Ok(()) } +/// Copy with `If-None-Match` should reject equality and accept inequality or absence. +pub async fn test_copy_with_if_none_match(op: Operator) -> Result<()> { + let source_path = TEST_FIXTURE.new_file_path(); + let target_path = TEST_FIXTURE.new_file_path(); + let missing_path = TEST_FIXTURE.new_file_path(); + let (source_content, _) = gen_bytes(op.info().capability()); + let (target_content, _) = gen_bytes(op.info().capability()); + assert_ne!(source_content, target_content); + + op.write(&source_path, source_content.clone()).await?; + op.write(&target_path, target_content).await?; + let etag = op + .stat(&target_path) + .await? + .etag() + .expect("etag must exist") + .to_string(); + + let err = op + .copy_with(&source_path, &target_path) + .if_none_match(&etag) + .await + .expect_err("equal ETag non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.copy_with(&source_path, &target_path) + .if_none_match("\"different-etag\"") + .await?; + op.copy_with(&source_path, &missing_path) + .if_none_match(&etag) + .await?; + + assert_eq!(op.read(&target_path).await?.to_bytes(), source_content); + Ok(()) +} + +/// Version preconditions should compare against the current destination version. +pub async fn test_copy_with_version_conditions(op: Operator) -> Result<()> { + let cap = op.info().capability(); + let source_path = TEST_FIXTURE.new_file_path(); + let target_path = TEST_FIXTURE.new_file_path(); + let missing_path = TEST_FIXTURE.new_file_path(); + let (source_content, _) = gen_bytes(cap); + let (first_target, _) = gen_bytes(cap); + let (second_target, _) = gen_bytes(cap); + assert_ne!(source_content, first_target); + assert_ne!(source_content, second_target); + assert_ne!(first_target, second_target); + + op.write(&source_path, source_content).await?; + op.write(&target_path, first_target).await?; + let stale = op + .stat(&target_path) + .await? + .version() + .expect("version must exist") + .to_string(); + op.write(&target_path, second_target).await?; + let current = op + .stat(&target_path) + .await? + .version() + .expect("version must exist") + .to_string(); + + let err = op + .copy_with(&source_path, &target_path) + .if_version_match(&stale) + .await + .expect_err("stale destination version must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + let err = op + .copy_with(&source_path, &target_path) + .if_version_not_match(¤t) + .await + .expect_err("equal destination version must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.copy_with(&source_path, &target_path) + .if_version_match(¤t) + .await?; + op.copy_with(&source_path, &target_path) + .if_version_not_match(¤t) + .await?; + + for result in [ + op.copy_with(&source_path, &missing_path) + .if_version_match(¤t) + .await, + op.copy_with(&source_path, &missing_path) + .if_version_not_match(¤t) + .await, + ] { + assert_eq!( + result.expect_err("missing destination must fail").kind(), + ErrorKind::ConditionNotMatch + ); + } + + Ok(()) +} + +/// `if_not_changed` should guard the destination against a stale observation. +pub async fn test_copy_if_not_changed(op: Operator) -> Result<()> { + let cap = op.info().capability(); + let source_path = TEST_FIXTURE.new_file_path(); + let target_path = TEST_FIXTURE.new_file_path(); + let (source_content, _) = gen_bytes(cap); + let (target_content, _) = gen_bytes(cap); + assert_ne!(source_content, target_content); + + op.write(&source_path, source_content).await?; + op.write(&target_path, target_content).await?; + let expected = op.stat(&target_path).await?; + + op.copy_with(&source_path, &target_path) + .if_not_changed(&expected) + .await?; + let err = op + .copy_with(&source_path, &target_path) + .if_not_changed(&expected) + .await + .expect_err("stale destination metadata must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + Ok(()) +} + /// Copy with source_version should copy a specific source version to a new file. pub async fn test_copy_with_source_version_to_new_file(op: Operator) -> Result<()> { if !op.info().capability().copy_with_source_version { diff --git a/core/tests/behavior/async_delete.rs b/core/tests/behavior/async_delete.rs index e074d0fbc57c..81c3a90794a4 100644 --- a/core/tests/behavior/async_delete.rs +++ b/core/tests/behavior/async_delete.rs @@ -53,9 +53,23 @@ pub fn tests(op: &Operator, tests: &mut Vec) { op, test_delete_with_if_match_match, test_delete_with_if_match_mismatch, + 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 + )); + } } } @@ -539,3 +553,142 @@ pub async fn test_batch_delete_with_if_match(op: Operator) -> Result<()> { Ok(()) } + +/// A matching delete condition requires a live target. +pub async fn test_delete_with_if_match_missing(op: Operator) -> Result<()> { + let path = TEST_FIXTURE.new_file_path(); + let err = op + .delete_with(&path) + .if_match("\"missing-etag\"") + .await + .expect_err("missing target must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + Ok(()) +} + +/// Delete with `If-None-Match` should reject equality and accept inequality or absence. +pub async fn test_delete_with_if_none_match(op: Operator) -> Result<()> { + let (path, content, _) = TEST_FIXTURE.new_file(op.clone()); + op.write(&path, content).await?; + let etag = op + .stat(&path) + .await? + .etag() + .expect("etag must exist") + .to_string(); + + let err = op + .delete_with(&path) + .if_none_match(&etag) + .await + .expect_err("equal ETag non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + assert!(op.exists(&path).await?); + + op.delete_with(&path) + .if_none_match("\"different-etag\"") + .await?; + op.delete_with(&path).if_none_match(&etag).await?; + + Ok(()) +} + +/// Version preconditions should compare against the current live target version. +pub async fn test_delete_with_version_conditions(op: Operator) -> Result<()> { + let cap = op.info().capability(); + let path = TEST_FIXTURE.new_file_path(); + let (first, _) = gen_bytes(cap); + let (second, _) = gen_bytes(cap); + assert_ne!(first, second); + + op.write(&path, first).await?; + let stale = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + op.write(&path, second.clone()).await?; + let current = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + + let err = op + .delete_with(&path) + .if_version_match(&stale) + .await + .expect_err("stale version match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + let err = op + .delete_with(&path) + .if_version_not_match(¤t) + .await + .expect_err("equal version non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.delete_with(&path).if_version_not_match(&stale).await?; + + for result in [ + op.delete_with(&path).if_version_match(¤t).await, + op.delete_with(&path).if_version_not_match(¤t).await, + ] { + assert_eq!( + result.expect_err("missing target must fail").kind(), + ErrorKind::ConditionNotMatch + ); + } + + op.write(&path, second).await?; + let current = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + op.delete_with(&path).if_version_match(¤t).await?; + + Ok(()) +} + +/// `if_not_changed` should reject stale metadata and accept the current state. +pub async fn test_delete_if_not_changed(op: Operator) -> Result<()> { + let cap = op.info().capability(); + let path = TEST_FIXTURE.new_file_path(); + let (first, _) = gen_bytes(cap); + let (second, _) = gen_bytes(cap); + assert_ne!(first, second); + + op.write(&path, first).await?; + let stale = op.stat(&path).await?; + op.write(&path, second).await?; + + let err = op + .delete_with(&path) + .if_not_changed(&stale) + .await + .expect_err("stale metadata must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + let current = op.stat(&path).await?; + op.delete_with(&path).if_not_changed(¤t).await?; + Ok(()) +} + +/// `Deleter` should lower and preserve `if_not_changed` for queued deletes. +pub async fn test_deleter_if_not_changed(op: Operator) -> Result<()> { + let (path, content, _) = TEST_FIXTURE.new_file(op.clone()); + op.write(&path, content).await?; + let expected = op.stat(&path).await?; + + let mut deleter = op.deleter().await?; + deleter + .delete(DeleteInput::new(path.clone()).with_if_not_changed(&expected)) + .await?; + deleter.close().await?; + assert!(!op.exists(&path).await?); + + Ok(()) +} diff --git a/core/tests/behavior/async_read.rs b/core/tests/behavior/async_read.rs index feee7c73079c..2335e2cc239d 100644 --- a/core/tests/behavior/async_read.rs +++ b/core/tests/behavior/async_read.rs @@ -46,6 +46,7 @@ pub fn tests(op: &Operator, tests: &mut Vec) { test_read_not_exist, test_read_with_if_match, test_read_with_if_none_match, + test_read_with_version_conditions, test_read_with_if_modified_since, test_read_with_if_unmodified_since, test_read_with_dir_path, @@ -618,6 +619,65 @@ pub async fn test_read_with_if_none_match(op: Operator) -> anyhow::Result<()> { Ok(()) } +/// Version preconditions should compare against the current live object version. +pub async fn test_read_with_version_conditions(op: Operator) -> anyhow::Result<()> { + let cap = op.info().capability(); + if !cap.read_with_if_version_match || !cap.read_with_if_version_not_match { + return Ok(()); + } + + let path = TEST_FIXTURE.new_file_path(); + let (first, _) = gen_bytes(cap); + let (second, _) = gen_bytes(cap); + assert_ne!(first, second); + + op.write(&path, first).await?; + let stale = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + op.write(&path, second.clone()).await?; + let current = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + + let reader = op.reader_with(&path).if_version_match(¤t).await?; + assert_eq!(reader.read(..).await?.to_bytes(), second); + + let err = op + .read_with(&path) + .if_version_match(&stale) + .await + .expect_err("stale version match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.read_with(&path).if_version_not_match(&stale).await?; + let err = op + .read_with(&path) + .if_version_not_match(¤t) + .await + .expect_err("equal version non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + let missing = TEST_FIXTURE.new_file_path(); + for result in [ + op.read_with(&missing).if_version_match(¤t).await, + op.read_with(&missing).if_version_not_match(¤t).await, + ] { + assert_eq!( + result.expect_err("missing target must fail").kind(), + ErrorKind::ConditionNotMatch + ); + } + + Ok(()) +} + /// Read with dir path should return an error. pub async fn test_read_with_dir_path(op: Operator) -> anyhow::Result<()> { if !op.info().capability().create_dir { diff --git a/core/tests/behavior/async_stat.rs b/core/tests/behavior/async_stat.rs index e2a595485355..146c933a6740 100644 --- a/core/tests/behavior/async_stat.rs +++ b/core/tests/behavior/async_stat.rs @@ -39,6 +39,7 @@ pub fn tests(op: &Operator, tests: &mut Vec) { test_stat_not_exist, test_stat_with_if_match, test_stat_with_if_none_match, + test_stat_with_version_conditions, test_stat_with_if_modified_since, test_stat_with_if_unmodified_since, test_stat_with_override_cache_control, @@ -236,6 +237,63 @@ pub async fn test_stat_with_if_none_match(op: Operator) -> Result<()> { Ok(()) } +/// Version preconditions should compare against the current live object version. +pub async fn test_stat_with_version_conditions(op: Operator) -> Result<()> { + let cap = op.info().capability(); + if !cap.stat_with_if_version_match || !cap.stat_with_if_version_not_match { + return Ok(()); + } + + let path = TEST_FIXTURE.new_file_path(); + let (first, _) = gen_bytes(cap); + let (second, _) = gen_bytes(cap); + assert_ne!(first, second); + + op.write(&path, first).await?; + let stale = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + op.write(&path, second).await?; + let current = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + + op.stat_with(&path).if_version_match(¤t).await?; + let err = op + .stat_with(&path) + .if_version_match(&stale) + .await + .expect_err("stale version match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.stat_with(&path).if_version_not_match(&stale).await?; + let err = op + .stat_with(&path) + .if_version_not_match(¤t) + .await + .expect_err("equal version non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + let missing = TEST_FIXTURE.new_file_path(); + for result in [ + op.stat_with(&missing).if_version_match(¤t).await, + op.stat_with(&missing).if_version_not_match(¤t).await, + ] { + assert_eq!( + result.expect_err("missing target must fail").kind(), + ErrorKind::ConditionNotMatch + ); + } + + Ok(()) +} + /// Stat file with if_modified_since should succeed, otherwise get a ConditionNotMatch error. pub async fn test_stat_with_if_modified_since(op: Operator) -> Result<()> { if !op.info().capability().stat_with_if_modified_since { diff --git a/core/tests/behavior/async_write.rs b/core/tests/behavior/async_write.rs index 5a17dc69cf82..60df2813c0d7 100644 --- a/core/tests/behavior/async_write.rs +++ b/core/tests/behavior/async_write.rs @@ -45,6 +45,8 @@ pub fn tests(op: &Operator, tests: &mut Vec) { test_write_with_if_none_match, test_write_with_if_not_exists, test_write_with_if_match, + test_write_with_version_conditions, + test_write_if_not_changed, test_write_with_user_metadata, test_write_returns_metadata, test_writer_write, @@ -60,7 +62,8 @@ pub fn tests(op: &Operator, tests: &mut Vec) { test_writer_write_non_contiguous_data, test_writer_write_with_if_not_exists, test_writer_write_with_if_none_match, - test_writer_write_with_if_match + test_writer_write_with_if_match, + test_writer_write_with_version_conditions )) } @@ -813,6 +816,119 @@ pub async fn test_write_with_if_match(op: Operator) -> Result<()> { Ok(()) } +/// Version preconditions should compare against the current live object version. +pub async fn test_write_with_version_conditions(op: Operator) -> Result<()> { + let cap = op.info().capability(); + if !cap.write_with_if_version_match || !cap.write_with_if_version_not_match { + return Ok(()); + } + + let path = TEST_FIXTURE.new_file_path(); + let (initial, _) = gen_bytes(cap); + let (replacement, _) = gen_bytes(cap); + assert_ne!(initial, replacement); + + op.write(&path, initial).await?; + let first_version = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + + op.write_with(&path, replacement.clone()) + .if_version_match(&first_version) + .await?; + + let current_version = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + let err = op + .write_with(&path, replacement.clone()) + .if_version_match(&first_version) + .await + .expect_err("stale version match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + let err = op + .write_with(&path, replacement.clone()) + .if_version_not_match(¤t_version) + .await + .expect_err("equal version non-match must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + op.write_with(&path, replacement) + .if_version_not_match(&first_version) + .await?; + + let missing = TEST_FIXTURE.new_file_path(); + for result in [ + op.write_with(&missing, Vec::::new()) + .if_version_match(¤t_version) + .await, + op.write_with(&missing, Vec::::new()) + .if_version_not_match(¤t_version) + .await, + ] { + assert_eq!( + result.expect_err("missing target must fail").kind(), + ErrorKind::ConditionNotMatch + ); + } + + Ok(()) +} + +/// `if_not_changed` should accept the observed state once and reject it after replacement. +pub async fn test_write_if_not_changed(op: Operator) -> Result<()> { + let cap = op.info().capability(); + if !cap.write_with_if_version_match && !cap.write_with_if_match { + return Ok(()); + } + + let path = TEST_FIXTURE.new_file_path(); + let (initial, _) = gen_bytes(cap); + let (replacement, _) = gen_bytes(cap); + assert_ne!(initial, replacement); + + op.write(&path, initial).await?; + let expected = op.stat(&path).await?; + + let mut conflicting = options::WriteOptions { + if_not_changed: Some(expected.clone()), + ..Default::default() + }; + let mut matching = conflicting.clone(); + if cap.write_with_if_version_match { + let version = expected.version().expect("version must exist"); + conflicting.if_version_match = Some(format!("different-{version}")); + matching.if_version_match = Some(version.to_string()); + } else { + let etag = expected.etag().expect("etag must exist"); + conflicting.if_match = Some(format!("different-{etag}")); + matching.if_match = Some(etag.to_string()); + } + + let err = op + .write_options(&path, replacement.clone(), conflicting) + .await + .expect_err("conflicting explicit condition must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + op.write_options(&path, replacement.clone(), matching) + .await?; + let err = op + .write_with(&path, replacement) + .if_not_changed(&expected) + .await + .expect_err("stale metadata must fail"); + assert_eq!(err.kind(), ErrorKind::ConditionNotMatch); + + Ok(()) +} + /// Write an existing file through a chunked writer with if_not_exists should get a /// ConditionNotMatch error. pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> { @@ -821,14 +937,6 @@ pub async fn test_writer_write_with_if_not_exists(op: Operator) -> Result<()> { return Ok(()); } - // GCS XML API multipart uploads do not support preconditions, so the multipart - // writer path cannot honor if_not_exists. Tracked in - // https://github.com/apache/opendal/issues/8040 - #[cfg(feature = "services-gcs")] - if op.info().scheme() == services::GCS_SCHEME { - return Ok(()); - } - let path = TEST_FIXTURE.new_file_path(); let content = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1)); @@ -929,6 +1037,50 @@ pub async fn test_writer_write_with_if_match(op: Operator) -> Result<()> { Ok(()) } +/// Chunked writers should preserve version preconditions through final commit. +pub async fn test_writer_write_with_version_conditions(op: Operator) -> Result<()> { + let cap = op.info().capability(); + if !cap.write_can_multi + || !cap.write_with_if_version_match + || !cap.write_with_if_version_not_match + { + return Ok(()); + } + + let path = TEST_FIXTURE.new_file_path(); + let body = gen_fixed_bytes(cap.write_multi_min_size.unwrap_or(1)); + op.write(&path, body.clone()).await?; + let expected = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + + let mut writer = op.writer_with(&path).if_version_match(&expected).await?; + writer.write(body.clone()).await?; + writer.write(body.clone()).await?; + writer.close().await?; + + let current = op + .stat(&path) + .await? + .version() + .expect("version must exist") + .to_string(); + let result: opendal::Result<()> = async { + let mut writer = op.writer_with(&path).if_version_not_match(¤t).await?; + writer.write(body.clone()).await?; + writer.write(body).await?; + writer.close().await?; + Ok(()) + } + .await; + assert_eq!(result.unwrap_err().kind(), ErrorKind::ConditionNotMatch); + + Ok(()) +} + pub async fn test_writer_write_non_contiguous_data(op: Operator) -> Result<()> { let path = TEST_FIXTURE.new_file_path(); let size = 1024 * 1024; // write file with 1 MiB