feat(tiering): model provider version capabilities (#5372)

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,

crates/ecstore/src/client/api_put_object_streaming.rs

@@ -578,7 +578,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,

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()
         })
     }

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<Option<&str>, 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);

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<RemoteVersion, std::io::Error> {
+        self.provider_version_capabilities().remote_version(headers, versioning)
+    }
+
+    pub(crate) fn legacy_remote_version_id(&self, headers: &HeaderMap) -> Result<String, std::io::Error> {
+        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<RemoteVersion, std::io::Error> {
+        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")