fix(storage): expose truthful storage class capabilities (#5172)

rustfs/src/admin/handlers/system.rs

@@ -24,6 +24,7 @@ use crate::admin::runtime_sources::{
 use crate::admin::storage_api::cluster::{
     CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider,
 };
+use crate::admin::storage_api::storageclass as storage_class_contract;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::runtime_capabilities::{EndpointTopologySnapshotProvider, RustFsObservabilitySnapshotProvider};
 use crate::server::{ADMIN_PREFIX, RemoteAddr};
@@ -635,6 +636,7 @@ pub struct RuntimeCapabilitiesSummary {
 #[derive(Debug, Clone, PartialEq, Eq, Serialize)]
 pub struct RuntimeCapabilitiesResponse {
     pub summary: RuntimeCapabilitiesSummary,
+    pub storage_classes: StorageClassCapabilities,
     pub cluster_snapshot_path: String,
     pub cluster_snapshot_summary: Option<CapabilityStatus>,
     pub observability: crate::admin::storage_api::cluster::ObservabilitySnapshot,
@@ -643,6 +645,25 @@ pub struct RuntimeCapabilitiesResponse {
     pub topology_status: CapabilityStatus,
 }
 
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub struct StorageClassCapabilities {
+    pub contract_version: u32,
+    pub supported_write_classes: [&'static str; 2],
+    pub unsupported_write_error: &'static str,
+    pub legacy_label_behavior: &'static str,
+}
+
+impl StorageClassCapabilities {
+    fn current() -> Self {
+        Self {
+            contract_version: storage_class_contract::CAPABILITY_CONTRACT_VERSION,
+            supported_write_classes: storage_class_contract::SUPPORTED_WRITE_CLASSES,
+            unsupported_write_error: storage_class_contract::UNSUPPORTED_WRITE_ERROR,
+            legacy_label_behavior: storage_class_contract::LEGACY_LABEL_BEHAVIOR,
+        }
+    }
+}
+
 pub struct RuntimeCapabilitiesHandler {}
 
 pub(crate) async fn build_runtime_capabilities_response()
@@ -669,6 +690,7 @@ pub(crate) async fn build_runtime_capabilities_response()
 
     Ok(RuntimeCapabilitiesResponse {
         summary,
+        storage_classes: StorageClassCapabilities::current(),
         cluster_snapshot_path: usecase.cluster_snapshot_route().to_string(),
         cluster_snapshot_summary: cluster_snapshot_discovery.summary,
         observability,
@@ -931,11 +953,22 @@ mod tests {
         assert_eq!(response.summary.site_replication_info.state, CapabilityState::Supported);
         assert_eq!(response.summary.site_replication_edit.state, CapabilityState::Supported);
         assert_eq!(response.summary.site_replication_resync.state, CapabilityState::Supported);
+        assert_eq!(response.storage_classes.contract_version, 1);
+        assert_eq!(response.storage_classes.supported_write_classes, ["STANDARD", "REDUCED_REDUNDANCY"]);
+        assert_eq!(response.storage_classes.unsupported_write_error, "InvalidStorageClass");
+        assert_eq!(response.storage_classes.legacy_label_behavior, "normalized_to_effective_class");
 
         let value = serde_json::to_value(response).expect("runtime capability response should serialize");
         assert_eq!(value["summary"]["site_replication_info"]["state"], "supported");
         assert_eq!(value["summary"]["site_replication_edit"]["state"], "supported");
         assert_eq!(value["summary"]["site_replication_resync"]["state"], "supported");
+        assert_eq!(value["storage_classes"]["contract_version"], 1);
+        assert_eq!(
+            value["storage_classes"]["supported_write_classes"],
+            json!(["STANDARD", "REDUCED_REDUNDANCY"])
+        );
+        assert_eq!(value["storage_classes"]["unsupported_write_error"], "InvalidStorageClass");
+        assert_eq!(value["storage_classes"]["legacy_label_behavior"], "normalized_to_effective_class");
     }
 
     #[test]

rustfs/src/admin/storage_api.rs

@@ -390,9 +390,11 @@ pub(crate) mod versioning_sys {
 }
 
 pub(crate) mod storageclass {
+    pub(crate) const CAPABILITY_CONTRACT_VERSION: u32 = super::ecstore_config::storageclass::CAPABILITY_CONTRACT_VERSION;
     #[cfg(test)]
     pub(crate) const CLASS_STANDARD: &str = super::ecstore_config::storageclass::CLASS_STANDARD;
     pub(crate) const INLINE_BLOCK_ENV: &str = super::ecstore_config::storageclass::INLINE_BLOCK_ENV;
+    pub(crate) const LEGACY_LABEL_BEHAVIOR: &str = super::ecstore_config::storageclass::LEGACY_LABEL_BEHAVIOR;
     pub(crate) const OPTIMIZE_ENV: &str = super::ecstore_config::storageclass::OPTIMIZE_ENV;
     #[cfg(test)]
     pub(crate) const RRS: &str = super::ecstore_config::storageclass::RRS;
@@ -400,6 +402,8 @@ pub(crate) mod storageclass {
     #[cfg(test)]
     pub(crate) const STANDARD: &str = super::ecstore_config::storageclass::STANDARD;
     pub(crate) const STANDARD_ENV: &str = super::ecstore_config::storageclass::STANDARD_ENV;
+    pub(crate) const SUPPORTED_WRITE_CLASSES: [&str; 2] = super::ecstore_config::storageclass::SUPPORTED_WRITE_CLASSES;
+    pub(crate) const UNSUPPORTED_WRITE_ERROR: &str = super::ecstore_config::storageclass::UNSUPPORTED_WRITE_ERROR;
 
     pub(crate) type Config = super::ecstore_config::storageclass::Config;
 

rustfs/src/app/object_usecase.rs

@@ -2778,11 +2778,16 @@ fn apply_put_request_metadata(
 }
 
 fn response_storage_class(info: &ObjectInfo, metadata: &HashMap<String, String>) -> Option<StorageClass> {
-    info.storage_class
-        .clone()
-        .or_else(|| metadata.get(AMZ_STORAGE_CLASS).cloned())
-        .filter(|storage_class| !storage_class.is_empty() && storage_class != storageclass::STANDARD)
-        .map(StorageClass::from)
+    let stored_class = info
+        .storage_class
+        .as_deref()
+        .or_else(|| metadata.get(AMZ_STORAGE_CLASS).map(String::as_str));
+    let transitioned_tier = (info.transitioned_object.status == rustfs_filemeta::TRANSITION_COMPLETE
+        && !info.transitioned_object.tier.is_empty())
+    .then_some(info.transitioned_object.tier.as_str());
+    let effective_class = storageclass::effective_class(stored_class, transitioned_tier);
+
+    (effective_class != storageclass::STANDARD).then(|| StorageClass::from(effective_class.to_string()))
 }
 
 fn response_storage_class_for_object_attributes(
@@ -2794,12 +2799,17 @@ fn response_storage_class_for_object_attributes(
         return None;
     }
 
-    info.storage_class
-        .clone()
-        .or_else(|| metadata.get(AMZ_STORAGE_CLASS).cloned())
-        .or_else(|| Some(storageclass::STANDARD.to_string()))
-        .filter(|storage_class| !storage_class.is_empty())
-        .map(StorageClass::from)
+    let stored_class = info
+        .storage_class
+        .as_deref()
+        .or_else(|| metadata.get(AMZ_STORAGE_CLASS).map(String::as_str));
+    let transitioned_tier = (info.transitioned_object.status == rustfs_filemeta::TRANSITION_COMPLETE
+        && !info.transitioned_object.tier.is_empty())
+    .then_some(info.transitioned_object.tier.as_str());
+
+    Some(StorageClass::from(
+        storageclass::effective_class(stored_class, transitioned_tier).to_string(),
+    ))
 }
 
 async fn apply_put_request_object_lock_opts(
@@ -12286,7 +12296,7 @@ mod tests {
     }
 
     #[test]
-    fn response_storage_class_omits_standard_and_keeps_non_default() {
+    fn response_storage_class_reports_effective_layout_and_preserves_transition_tier() {
         let metadata = HashMap::new();
         let standard_info = ObjectInfo {
             storage_class: Some(storageclass::STANDARD.to_string()),
@@ -12297,16 +12307,39 @@ mod tests {
 
         let mut metadata = HashMap::new();
         metadata.insert(AMZ_STORAGE_CLASS.to_string(), storageclass::STANDARD_IA.to_string());
-        let infrequent_access_info = ObjectInfo {
+        let label_only_info = ObjectInfo {
             storage_class: Some(storageclass::STANDARD_IA.to_string()),
             user_defined: Arc::new(metadata.clone()),
             ..Default::default()
         };
+        assert!(
+            response_storage_class(&label_only_info, &metadata).is_none(),
+            "historical STANDARD_IA labels must report the effective implicit STANDARD layout"
+        );
+
+        let rrs_info = ObjectInfo {
+            storage_class: Some(storageclass::RRS.to_string()),
+            ..Default::default()
+        };
         assert_eq!(
-            response_storage_class(&infrequent_access_info, &metadata)
+            response_storage_class(&rrs_info, &HashMap::new())
                 .as_ref()
                 .map(StorageClass::as_str),
-            Some(storageclass::STANDARD_IA)
+            Some(storageclass::RRS)
+        );
+
+        let mut transitioned_info = label_only_info;
+        transitioned_info.transitioned_object.tier = "WARM-TIER".to_string();
+        assert!(
+            response_storage_class(&transitioned_info, &metadata).is_none(),
+            "a tier name without a completed transition must not override the effective local class"
+        );
+        transitioned_info.transitioned_object.status = rustfs_filemeta::TRANSITION_COMPLETE.to_string();
+        assert_eq!(
+            response_storage_class(&transitioned_info, &metadata)
+                .as_ref()
+                .map(StorageClass::as_str),
+            Some("WARM-TIER")
         );
 
         let mut metadata = HashMap::new();
@@ -12337,6 +12370,17 @@ mod tests {
                 .map(StorageClass::as_str),
             Some(storageclass::STANDARD)
         );
+
+        let legacy_info = ObjectInfo {
+            storage_class: Some(storageclass::STANDARD_IA.to_string()),
+            ..Default::default()
+        };
+        assert_eq!(
+            response_storage_class_for_object_attributes(&legacy_info, &HashMap::new(), true)
+                .as_ref()
+                .map(StorageClass::as_str),
+            Some(storageclass::STANDARD)
+        );
     }
 
     #[test]
@@ -12575,7 +12619,7 @@ mod tests {
             })
             .bucket("test-bucket".to_string())
             .key("test-key".to_string())
-            .storage_class(Some(StorageClass::from_static(storageclass::STANDARD_IA)))
+            .storage_class(Some(StorageClass::from_static(storageclass::RRS)))
             .build()
             .unwrap();
 

rustfs/src/app/storage_api.rs

@@ -903,9 +903,9 @@ pub(crate) mod set_disk {
 }
 
 pub(crate) mod storage_class {
-    pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::STANDARD;
     #[cfg(test)]
-    pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::STANDARD_IA;
+    pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::{RRS, STANDARD_IA};
+    pub(crate) use crate::storage::storage_api::ecstore_config::storageclass::{STANDARD, effective_class};
 }
 
 pub(crate) mod timeout_wrapper {

rustfs/src/storage/s3_api/bucket.rs

@@ -449,6 +449,8 @@ mod tests {
     use crate::storage::s3_api::common::rustfs_owner;
     use crate::storage::storage_api::s3_api_consumer::bucket::StorageObjectInfo as ObjectInfo;
     use crate::storage::storage_api::s3_api_consumer::bucket::contract::bucket::BucketInfo;
+    use rustfs_filemeta::FileInfo;
+    use rustfs_utils::http::headers::AMZ_STORAGE_CLASS;
     use s3s::S3ErrorCode;
     use s3s::dto::{CommonPrefix, EncodingType, ListObjectsV2Output, Object};
     use time::OffsetDateTime;
@@ -654,6 +656,67 @@ mod tests {
         assert_eq!(output.common_prefixes.as_ref().map(std::vec::Vec::len), Some(2));
     }
 
+    #[test]
+    fn list_responses_report_standard_for_legacy_label_only_file_metadata() {
+        let version_id = Uuid::parse_str("11111111-2222-3333-4444-555555555555").expect("fixture version ID should be valid");
+        let file_info = FileInfo {
+            name: "legacy-object".to_string(),
+            version_id: Some(version_id),
+            metadata: std::collections::HashMap::from([(AMZ_STORAGE_CLASS.to_string(), "STANDARD_IA".to_string())]),
+            ..Default::default()
+        };
+        let object_info = ObjectInfo::from_file_info(&file_info, "bucket", "legacy-object", true);
+
+        let list_output = build_list_objects_v2_output(
+            ListObjectsV2Info {
+                objects: vec![object_info.clone()],
+                ..Default::default()
+            },
+            false,
+            1000,
+            "bucket".to_string(),
+            String::new(),
+            None,
+            None,
+            None,
+            None,
+        );
+        assert_eq!(
+            list_output
+                .contents
+                .as_ref()
+                .and_then(|objects| objects.first())
+                .and_then(|object| object.storage_class.as_ref())
+                .map(|storage_class| storage_class.as_str()),
+            Some("STANDARD")
+        );
+
+        let versions_output = build_list_object_versions_output(
+            ListObjectVersionsInfo {
+                objects: vec![object_info],
+                ..Default::default()
+            },
+            "bucket".to_string(),
+            &ListObjectVersionsParams {
+                prefix: String::new(),
+                delimiter: None,
+                key_marker: None,
+                version_id_marker: None,
+                max_keys: 1000,
+            },
+            None,
+        );
+        assert_eq!(
+            versions_output
+                .versions
+                .as_ref()
+                .and_then(|versions| versions.first())
+                .and_then(|version| version.storage_class.as_ref())
+                .map(|storage_class| storage_class.as_str()),
+            Some("STANDARD")
+        );
+    }
+
     #[test]
     fn test_list_objects_v2_url_encoding_preserves_slash() {
         let object_infos = ListObjectsV2Info {

rustfs/src/storage/s3_api/multipart.rs

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 use crate::storage::s3_api::common::{rustfs_initiator, rustfs_owner};
+use crate::storage::storage_api::effective_storage_class;
 use crate::storage::storage_api::s3_api_consumer::multipart::contract::multipart::{
     ListMultipartsInfo, ListPartsInfo, MAX_MULTIPART_PART_NUMBER,
 };
@@ -61,11 +62,11 @@ pub(crate) fn build_list_parts_output(res: ListPartsInfo) -> ListPartsOutput {
         next_part_number_marker: res.next_part_number_marker.try_into().ok(),
         max_parts: res.max_parts.try_into().ok(),
         part_number_marker: res.part_number_marker.try_into().ok(),
-        storage_class: if res.storage_class.is_empty() {
-            None
-        } else {
-            Some(res.storage_class.into())
-        },
+        storage_class: Some(
+            effective_storage_class((!res.storage_class.is_empty()).then_some(res.storage_class.as_str()), None)
+                .to_string()
+                .into(),
+        ),
         ..Default::default()
     }
 }
@@ -245,9 +246,9 @@ mod tests {
     }
 
     #[test]
-    fn test_list_parts_output_handles_empty_storage_class_and_overflow_markers() {
+    fn test_list_parts_output_normalizes_legacy_storage_class_and_handles_overflow_markers() {
         let input = ListPartsInfo {
-            storage_class: String::new(),
+            storage_class: "STANDARD_IA".to_string(),
             part_number_marker: usize::MAX,
             next_part_number_marker: usize::MAX,
             max_parts: usize::MAX,
@@ -262,13 +263,20 @@ mod tests {
         let output = build_list_parts_output(input);
         let parts = output.parts.as_ref().expect("parts should be present");
 
-        assert_eq!(output.storage_class, None);
+        assert_eq!(output.storage_class.as_ref().map(|value| value.as_str()), Some("STANDARD"));
         assert_eq!(output.part_number_marker, None);
         assert_eq!(output.next_part_number_marker, None);
         assert_eq!(output.max_parts, None);
         assert_eq!(parts.len(), 1);
         assert_eq!(parts[0].part_number, None);
         assert_eq!(parts[0].size, None);
+
+        let output = build_list_parts_output(ListPartsInfo::default());
+        assert_eq!(
+            output.storage_class.as_ref().map(|value| value.as_str()),
+            Some("STANDARD"),
+            "legacy uploads without a stored class must report their effective STANDARD layout"
+        );
     }
 
     #[test]

rustfs/src/storage/storage_api.rs

@@ -1512,6 +1512,10 @@ pub(crate) fn is_valid_storage_class(storage_class: &str) -> bool {
     ecstore_set_disk::is_valid_storage_class(storage_class)
 }
 
+pub(crate) fn effective_storage_class<'a>(stored_class: Option<&'a str>, completed_transition_tier: Option<&'a str>) -> &'a str {
+    ecstore_config::storageclass::effective_class(stored_class, completed_transition_tier)
+}
+
 pub(crate) fn register_event_dispatch_hook<F>(hook: F) -> bool
 where
     F: Fn(EventArgs) + Send + Sync + 'static,