diff --git a/crates/ecstore/src/client/api_put_object_multipart.rs b/crates/ecstore/src/client/api_put_object_multipart.rs index daebe97c2..dbb1b3c40 100644 --- a/crates/ecstore/src/client/api_put_object_multipart.rs +++ b/crates/ecstore/src/client/api_put_object_multipart.rs @@ -417,7 +417,7 @@ impl TransitionClient { bucket: complete_multipart_upload_result.bucket, key: complete_multipart_upload_result.key, etag: trim_etag(&complete_multipart_upload_result.etag), - version_id: self.raw_version_id(&h)?.unwrap_or_default().to_string(), + version_id: self.legacy_remote_version_id(&h)?, location: complete_multipart_upload_result.location, expiration: exp_time, expiration_rule_id: rule_id, diff --git a/crates/ecstore/src/client/api_put_object_streaming.rs b/crates/ecstore/src/client/api_put_object_streaming.rs index fdb1e820e..4bc5393a3 100644 --- a/crates/ecstore/src/client/api_put_object_streaming.rs +++ b/crates/ecstore/src/client/api_put_object_streaming.rs @@ -567,7 +567,7 @@ impl TransitionClient { key: object_name.to_string(), etag: trim_etag(h.get("ETag").and_then(|v| v.to_str().ok()).unwrap_or("")), - version_id: self.raw_version_id(h)?.unwrap_or_default().to_string(), + version_id: self.legacy_remote_version_id(h)?, size, expiration: exp_time, expiration_rule_id: rule_id, diff --git a/crates/ecstore/src/client/api_remove.rs b/crates/ecstore/src/client/api_remove.rs index 24575ff7c..25c71d80a 100644 --- a/crates/ecstore/src/client/api_remove.rs +++ b/crates/ecstore/src/client/api_remove.rs @@ -201,7 +201,7 @@ impl TransitionClient { object_name: object_name.to_string(), object_version_id: opts.version_id, delete_marker: resp.headers().get(X_AMZ_DELETE_MARKER).map_or(false, |v| v == "true"), - delete_marker_version_id: self.raw_version_id(resp.headers())?.unwrap_or_default().to_string(), + delete_marker_version_id: self.legacy_remote_version_id(resp.headers())?, ..Default::default() }) } diff --git a/crates/ecstore/src/client/provider_versions.rs b/crates/ecstore/src/client/provider_versions.rs index ddfff5910..0190bdfa2 100644 --- a/crates/ecstore/src/client/provider_versions.rs +++ b/crates/ecstore/src/client/provider_versions.rs @@ -46,11 +46,33 @@ impl RemoteVersion { Self::Unknown | Self::Disabled => None, } } + + pub(crate) fn exact_request_id(&self) -> Result, Error> { + match self { + Self::Unknown => Err(Error::new( + ErrorKind::InvalidData, + "remote object version is unknown; exact version routing is unsafe", + )), + Self::Disabled => Ok(None), + Self::SuspendedNull => Ok(Some("null")), + Self::Exact(version_id) => Ok(Some(version_id)), + } + } +} + +#[derive(Clone, Copy, Debug, Eq, PartialEq)] +pub(crate) enum ConditionalCreateCapability { + Unsupported, + IfNoneMatchStar, + GenerationMatchZero, } #[derive(Clone, Copy, Debug, Eq, PartialEq)] pub(crate) struct ProviderVersionCapabilities { raw_version_header: Option<&'static str>, + pub(crate) bucket_versioning_state: bool, + pub(crate) list_object_versions: bool, + pub(crate) conditional_create: ConditionalCreateCapability, pub(crate) exact_get_delete: bool, } @@ -62,28 +84,59 @@ impl ProviderVersionCapabilities { || tier_type.eq_ignore_ascii_case("r2") || tier_type.eq_ignore_ascii_case("wasabi") { + let list_object_versions = tier_type.eq_ignore_ascii_case("s3") + || tier_type.eq_ignore_ascii_case("rustfs") + || tier_type.eq_ignore_ascii_case("minio") + || tier_type.eq_ignore_ascii_case("r2"); Self { raw_version_header: Some(X_AMZ_VERSION_ID), + bucket_versioning_state: list_object_versions, + list_object_versions, + conditional_create: if tier_type.eq_ignore_ascii_case("s3") || tier_type.eq_ignore_ascii_case("r2") { + ConditionalCreateCapability::IfNoneMatchStar + } else { + ConditionalCreateCapability::Unsupported + }, exact_get_delete: true, } } else if tier_type.eq_ignore_ascii_case("aliyun") { Self { raw_version_header: Some(X_OSS_VERSION_ID), + bucket_versioning_state: false, + list_object_versions: false, + conditional_create: ConditionalCreateCapability::Unsupported, exact_get_delete: true, } } else if tier_type.eq_ignore_ascii_case("tencent") { Self { raw_version_header: Some(X_COS_VERSION_ID), + bucket_versioning_state: false, + list_object_versions: false, + conditional_create: ConditionalCreateCapability::Unsupported, exact_get_delete: true, } } else if tier_type.eq_ignore_ascii_case("huaweicloud") { Self { raw_version_header: Some(X_OBS_VERSION_ID), + bucket_versioning_state: false, + list_object_versions: false, + conditional_create: ConditionalCreateCapability::Unsupported, exact_get_delete: true, } + } else if tier_type.eq_ignore_ascii_case("gcs") { + Self { + raw_version_header: None, + bucket_versioning_state: false, + list_object_versions: false, + conditional_create: ConditionalCreateCapability::GenerationMatchZero, + exact_get_delete: false, + } } else { Self { raw_version_header: None, + bucket_versioning_state: false, + list_object_versions: false, + conditional_create: ConditionalCreateCapability::Unsupported, exact_get_delete: false, } } @@ -143,7 +196,7 @@ fn validate_remote_version_id(version_id: &str) -> Result<(), Error> { #[cfg(test)] mod tests { - use super::{BucketVersioningState, ProviderVersionCapabilities, RemoteVersion}; + use super::{BucketVersioningState, ConditionalCreateCapability, ProviderVersionCapabilities, RemoteVersion}; use http::{HeaderMap, HeaderValue}; #[test] @@ -219,6 +272,71 @@ mod tests { ); } + #[test] + fn provider_capability_matrix_is_conservative_and_provider_specific() { + for (tier_type, state, list, conditional_create, exact_get_delete) in [ + ("s3", true, true, ConditionalCreateCapability::IfNoneMatchStar, true), + ("rustfs", true, true, ConditionalCreateCapability::Unsupported, true), + ("minio", true, true, ConditionalCreateCapability::Unsupported, true), + ("r2", true, true, ConditionalCreateCapability::IfNoneMatchStar, true), + ("wasabi", false, false, ConditionalCreateCapability::Unsupported, true), + ("aliyun", false, false, ConditionalCreateCapability::Unsupported, true), + ("tencent", false, false, ConditionalCreateCapability::Unsupported, true), + ("huaweicloud", false, false, ConditionalCreateCapability::Unsupported, true), + ("gcs", false, false, ConditionalCreateCapability::GenerationMatchZero, false), + ("azure", false, false, ConditionalCreateCapability::Unsupported, false), + ("unsupported", false, false, ConditionalCreateCapability::Unsupported, false), + ] { + let capabilities = ProviderVersionCapabilities::for_tier_type(tier_type); + assert_eq!(capabilities.bucket_versioning_state, state, "{tier_type} versioning state"); + assert_eq!(capabilities.list_object_versions, list, "{tier_type} version listing"); + assert_eq!(capabilities.conditional_create, conditional_create, "{tier_type} conditional create"); + assert_eq!(capabilities.exact_get_delete, exact_get_delete, "{tier_type} exact routing"); + } + } + + #[test] + fn remote_version_states_preserve_unknown_disabled_suspended_and_exact() { + let capabilities = ProviderVersionCapabilities::for_tier_type("s3"); + let empty = HeaderMap::new(); + let mut null = HeaderMap::new(); + null.insert("x-amz-version-id", HeaderValue::from_static("null")); + let mut exact = HeaderMap::new(); + exact.insert("x-amz-version-id", HeaderValue::from_static("opaque.generation-7")); + + for (headers, state, expected) in [ + (&empty, BucketVersioningState::Unknown, RemoteVersion::Unknown), + (&empty, BucketVersioningState::Disabled, RemoteVersion::Disabled), + (&empty, BucketVersioningState::Suspended, RemoteVersion::Unknown), + (&empty, BucketVersioningState::Enabled, RemoteVersion::Unknown), + (&null, BucketVersioningState::Suspended, RemoteVersion::SuspendedNull), + ( + &exact, + BucketVersioningState::Enabled, + RemoteVersion::Exact("opaque.generation-7".to_string()), + ), + ] { + assert_eq!( + capabilities + .remote_version(headers, state) + .expect("version state should normalize"), + expected + ); + } + } + + #[test] + fn exact_request_routing_fails_closed_for_unknown_versions() { + for (version, expected) in [ + (RemoteVersion::Disabled, None), + (RemoteVersion::SuspendedNull, Some("null")), + (RemoteVersion::Exact("opaque-v1".to_string()), Some("opaque-v1")), + ] { + assert_eq!(version.exact_request_id().expect("known version state"), expected); + } + assert!(RemoteVersion::Unknown.exact_request_id().is_err()); + } + #[test] fn provider_version_rejects_empty_or_oversized_headers() { let oversized = "v".repeat(1025); diff --git a/crates/ecstore/src/client/transition_api.rs b/crates/ecstore/src/client/transition_api.rs index 008b9769b..53153c421 100644 --- a/crates/ecstore/src/client/transition_api.rs +++ b/crates/ecstore/src/client/transition_api.rs @@ -31,7 +31,7 @@ use crate::client::{ }, constants::{UNSIGNED_PAYLOAD, UNSIGNED_PAYLOAD_TRAILER}, credentials::{CredContext, Credentials, SignatureType, Static}, - provider_versions::{BucketVersioningState, ProviderVersionCapabilities}, + provider_versions::{BucketVersioningState, ProviderVersionCapabilities, RemoteVersion}, signer_error, }; use crate::{client::checksum::ChecksumMode, object_api::GetObjectReader}; @@ -332,6 +332,22 @@ impl TransitionClient { self.provider_version_capabilities().raw_version_id(headers) } + pub(crate) fn remote_version( + &self, + headers: &HeaderMap, + versioning: BucketVersioningState, + ) -> Result { + self.provider_version_capabilities().remote_version(headers, versioning) + } + + pub(crate) fn legacy_remote_version_id(&self, headers: &HeaderMap) -> Result { + Ok(self + .remote_version(headers, BucketVersioningState::Unknown)? + .exact_id() + .unwrap_or_default() + .to_string()) + } + fn trace_errors_only_off(&self) { if let Ok(mut trace_errors_only) = self.trace_errors_only.lock() { *trace_errors_only = false; @@ -1095,6 +1111,16 @@ impl Default for ObjectInfo { } } +impl ObjectInfo { + pub(crate) fn remote_version( + &self, + capabilities: ProviderVersionCapabilities, + versioning: BucketVersioningState, + ) -> Result { + capabilities.remote_version(&self.metadata, versioning) + } +} + #[derive(Serialize, Deserialize, Debug, Clone)] pub struct RestoreInfo { ongoing_restore: bool, @@ -1414,7 +1440,7 @@ mod tests { MAX_S3_CLIENT_RESPONSE_SIZE, MAX_S3_ERROR_RESPONSE_SIZE, SignatureType, build_tls_config, collect_response_body, signer_error_to_io_error, to_object_info_for_provider, validate_header_values, with_rustls_init_guard, }; - use crate::client::provider_versions::ProviderVersionCapabilities; + use crate::client::provider_versions::{BucketVersioningState, ProviderVersionCapabilities, RemoteVersion}; use http::{HeaderMap, HeaderValue}; use http_body_util::Full; use hyper::body::Bytes; @@ -1539,6 +1565,11 @@ mod tests { .expect("opaque provider version should parse"); assert_eq!(info.version_id, None); + assert_eq!( + info.remote_version(ProviderVersionCapabilities::for_tier_type("tencent"), BucketVersioningState::Enabled,) + .expect("opaque response version should remain available"), + RemoteVersion::Exact("opaque.version_01".to_string()) + ); assert_eq!( info.metadata.get("x-cos-version-id").and_then(|value| value.to_str().ok()), Some("opaque.version_01")