Merge remote-tracking branch 'origin/cxymds/fix-1358-opaque-transition-metadata' into cxymds/fix-1358-gcs-exact-generation

crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs

@@ -554,6 +554,7 @@ async fn delete_free_version_remote_object(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
 ) -> Result<(), std::io::Error> {
+    let version_id_exact = validate_transition_remote_version(oi)?;
     let identity = tier_destination_id_from_metadata(&oi.user_defined)?
         .ok_or_else(|| std::io::Error::other("tier free-version has no durable backend identity"))?;
     delete_object_from_remote_tier_idempotent_with_manager_and_identity(
@@ -562,7 +563,7 @@ async fn delete_free_version_remote_object(
         &oi.transitioned_object.tier,
         identity,
         tier_config_mgr,
-        false,
+        version_id_exact,
     )
     .await?;
     Ok(())
@@ -4135,6 +4136,23 @@ pub async fn get_transitioned_object_reader(
     get_transitioned_object_reader_with_tier_manager(bucket, object, rs, h, oi, opts, &tier_config_mgr).await
 }
 
+fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::Error> {
+    let version = oi.transitioned_object.version_id.as_str();
+    match oi.transition_version_state {
+        rustfs_filemeta::TransitionVersionState::Unknown => Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier object version state is unknown",
+        )),
+        rustfs_filemeta::TransitionVersionState::KnownDisabled if version.is_empty() => Ok(false),
+        rustfs_filemeta::TransitionVersionState::SuspendedNull if version == "null" => Ok(true),
+        rustfs_filemeta::TransitionVersionState::Exact if !version.is_empty() && version != "null" => Ok(true),
+        _ => Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier object version state conflicts with its version ID",
+        )),
+    }
+}
+
 pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     bucket: &str,
     object: &str,
@@ -4144,6 +4162,7 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     opts: &ObjectOptions,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
 ) -> Result<GetObjectReader, std::io::Error> {
+    validate_transition_remote_version(oi)?;
     let expected_identity = tier_destination_id_from_metadata(&oi.user_defined)?;
     let lease = match expected_identity {
         Some(identity) => {
@@ -5438,6 +5457,7 @@ mod tests {
                 tier: tier.clone(),
                 ..Default::default()
             },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
             ..Default::default()
         };
 
@@ -5501,6 +5521,7 @@ mod tests {
                 tier,
                 ..Default::default()
             },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
             ..Default::default()
         };
 
@@ -5523,6 +5544,70 @@ mod tests {
         assert_eq!(backend.get_count().await, 0);
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn transitioned_get_rejects_unknown_version_state_before_backend_io() {
+        let manager = TierConfigMgr::new();
+        let tier = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&manager, &tier).await;
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: 1,
+            transitioned_object: TransitionedObject {
+                name: "remote/object".to_string(),
+                version_id: String::new(),
+                status: crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(),
+                tier,
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            ..Default::default()
+        };
+
+        let err = match get_transitioned_object_reader_with_tier_manager(
+            &object_info.bucket,
+            &object_info.name,
+            &None,
+            &HeaderMap::new(),
+            &object_info,
+            &ObjectOptions::default(),
+            &manager,
+        )
+        .await
+        {
+            Ok(_) => panic!("unknown remote version state must fail before backend IO"),
+            Err(err) => err,
+        };
+
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(backend.get_count().await, 0);
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn free_version_delete_rejects_unknown_version_state_before_backend_io() {
+        let manager = TierConfigMgr::new();
+        let backend = register_mock_tier(&manager, "WARM").await;
+        let object_info = ObjectInfo {
+            transitioned_object: TransitionedObject {
+                name: "remote/object".to_string(),
+                version_id: "legacy-version".to_string(),
+                tier: "WARM".to_string(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            ..Default::default()
+        };
+
+        let err = super::delete_free_version_remote_object(&object_info, &manager)
+            .await
+            .expect_err("unknown remote version state must fail before backend IO");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(backend.remove_count().await, 0);
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     async fn free_version_remote_delete_requires_persisted_destination_identity() {
@@ -5834,7 +5919,8 @@ mod tests {
             version_id: "remote-version".to_string(),
             tier_name: "WARM".to_string(),
             backend_identity: Some([1; 32]),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         };
 
         let err = state
@@ -5945,7 +6031,8 @@ mod tests {
             version_id: "remote-version".to_string(),
             tier_name: "WARM".to_string(),
             backend_identity: Some([1; 32]),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         };
 
         state
@@ -9866,6 +9953,32 @@ mod tests {
         (backend, identity_hex)
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn journal_replay_rejects_unknown_version_state_before_backend_io() {
+        let (_disk_paths, ecstore) = setup_test_env().await;
+        let (backend, _) = register_recovery_mock_tier(&ecstore).await;
+        let identity = TierConfigMgr::acquire_operation_lease(&ecstore.tier_config_mgr(), "WARM")
+            .await
+            .expect("mock tier lease should be available")
+            .backend_identity();
+        let je = Jentry {
+            obj_name: "remote/object".to_string(),
+            version_id: "legacy-version".to_string(),
+            tier_name: "WARM".to_string(),
+            backend_identity: Some(identity),
+            version_id_exact: false,
+            version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+        };
+
+        let err = crate::bucket::lifecycle::tier_delete_journal::process_tier_delete_journal_entry(ecstore, &je)
+            .await
+            .expect_err("unknown journal state must fail before backend IO");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(backend.remove_count().await, 0);
+    }
+
     async fn seed_recoverable_free_version(
         disk_paths: &[PathBuf],
         bucket: &str,
@@ -9885,6 +9998,7 @@ mod tests {
                 identity,
             );
         }
+        let transition_version_id = Uuid::new_v4();
         let mut metadata = FileMeta::new();
         metadata
             .add_version(FileInfo {
@@ -9893,7 +10007,9 @@ mod tests {
                 version_id: Some(object_version_id),
                 transition_status: crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(),
                 transitioned_objname: format!("remote/{bucket}/{object}"),
-                transition_version_id: Some(Uuid::new_v4()),
+                transition_version_id: Some(transition_version_id),
+                transition_version: Some(transition_version_id.to_string()),
+                transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
                 transition_tier: "WARM".to_string(),
                 mod_time: Some(OffsetDateTime::now_utc()),
                 metadata: transitioned_metadata,

crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs

@@ -42,6 +42,7 @@ const TIER_DELETE_JOURNAL_RECOVERY_INTERVAL: Duration = Duration::from_secs(60);
 const TIER_DELETE_JOURNAL_RECOVERY_TIMEOUT: Duration = Duration::from_secs(300);
 const TIER_DELETE_JOURNAL_VERSION: u8 = 2;
 const TIER_DELETE_JOURNAL_EXACT_VERSION: u8 = 3;
+const TIER_DELETE_JOURNAL_STATE_VERSION: u8 = 4;
 pub(crate) const TIER_DELETE_JOURNAL_PREFIX: &str = "ilm/tier-delete-journal/";
 
 #[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
@@ -55,24 +56,35 @@ struct PersistedTierDeleteJournalEntry {
     backend_identity: Option<[u8; 32]>,
     #[serde(default, skip_serializing_if = "Option::is_none")]
     version_id_exact: Option<bool>,
+    #[serde(default, skip_serializing_if = "Option::is_none")]
+    version_state: Option<rustfs_filemeta::TransitionVersionState>,
 }
 
 impl PersistedTierDeleteJournalEntry {
-    fn from_jentry(je: &Jentry) -> Self {
+    fn from_jentry(je: &Jentry) -> Result<Self> {
-        Self {
+        validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?;
-            version: if je.version_id_exact {
+        let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown;
-                TIER_DELETE_JOURNAL_EXACT_VERSION
+        let version = if legacy_unknown {
-            } else if je.backend_identity.is_some() {
+            if je.backend_identity.is_some() {
                 TIER_DELETE_JOURNAL_VERSION
             } else {
                 1
-            },
+            }
+        } else {
+            if je.backend_identity.is_none() {
+                return Err(Error::other("new tier delete journal entry is missing its backend identity"));
+            }
+            TIER_DELETE_JOURNAL_STATE_VERSION
+        };
+        Ok(Self {
+            version,
             obj_name: je.obj_name.clone(),
             version_id: je.version_id.clone(),
             tier_name: je.tier_name.clone(),
             backend_identity: je.backend_identity,
             version_id_exact: je.version_id_exact.then_some(true),
-        }
+            version_state: (!legacy_unknown).then_some(je.version_state),
+        })
     }
 
     fn into_jentry(self) -> Result<Jentry> {
@@ -84,19 +96,23 @@ impl PersistedTierDeleteJournalEntry {
         if self.obj_name.is_empty() || self.tier_name.is_empty() {
             return Err(Error::other("tier delete journal entry is incomplete"));
         }
-        if self.version != TIER_DELETE_JOURNAL_EXACT_VERSION && self.version_id_exact.unwrap_or(false) {
+        if self.version != TIER_DELETE_JOURNAL_EXACT_VERSION
+            && self.version != TIER_DELETE_JOURNAL_STATE_VERSION
+            && self.version_id_exact.unwrap_or(false)
+        {
             return Err(Error::other(
                 "legacy tier delete journal entry has an unsupported exact version constraint",
             ));
         }
-        let (backend_identity, version_id_exact) = match self.version {
+        let (backend_identity, version_id_exact, version_state) = match self.version {
-            1 => (None, false),
+            1 => (None, false, rustfs_filemeta::TransitionVersionState::Unknown),
             TIER_DELETE_JOURNAL_VERSION => (
                 Some(
                     self.backend_identity
                         .ok_or_else(|| Error::other("tier delete journal v2 entry is missing its backend identity"))?,
                 ),
                 false,
+                rustfs_filemeta::TransitionVersionState::Unknown,
             ),
             TIER_DELETE_JOURNAL_EXACT_VERSION => {
                 if self.version_id.is_empty() || self.version_id_exact != Some(true) {
@@ -108,6 +124,22 @@ impl PersistedTierDeleteJournalEntry {
                             .ok_or_else(|| Error::other("tier delete journal v3 entry is missing its backend identity"))?,
                     ),
                     true,
+                    rustfs_filemeta::TransitionVersionState::Exact,
+                )
+            }
+            TIER_DELETE_JOURNAL_STATE_VERSION => {
+                let state = self
+                    .version_state
+                    .ok_or_else(|| Error::other("tier delete journal v4 entry is missing its version state"))?;
+                let exact = self.version_id_exact.unwrap_or(false);
+                validate_version_state(state, &self.version_id, exact)?;
+                (
+                    Some(
+                        self.backend_identity
+                            .ok_or_else(|| Error::other("tier delete journal v4 entry is missing its backend identity"))?,
+                    ),
+                    exact,
+                    state,
                 )
             }
             version => return Err(Error::other(format!("unsupported tier delete journal version {version}"))),
@@ -118,10 +150,30 @@ impl PersistedTierDeleteJournalEntry {
             tier_name: self.tier_name,
             backend_identity,
             version_id_exact,
+            version_state,
         })
     }
 }
 
+fn validate_version_state(
+    state: rustfs_filemeta::TransitionVersionState,
+    version_id: &str,
+    version_id_exact: bool,
+) -> Result<()> {
+    use rustfs_filemeta::TransitionVersionState::{Exact, KnownDisabled, SuspendedNull, Unknown};
+
+    let valid = match state {
+        Unknown => !version_id_exact,
+        KnownDisabled => version_id.is_empty() && !version_id_exact,
+        SuspendedNull => version_id == "null" && version_id_exact,
+        Exact => !version_id.is_empty() && version_id != "null" && version_id_exact,
+    };
+    if !valid {
+        return Err(Error::other("tier delete journal version state conflicts with its version id"));
+    }
+    Ok(())
+}
+
 #[derive(Debug, Clone, PartialEq, Eq)]
 pub struct TierDeleteJournalRecoveryStats {
     pub scanned: usize,
@@ -159,7 +211,7 @@ pub(crate) fn decode_tier_delete_journal_entry(data: &[u8]) -> Result<Jentry> {
 }
 
 pub(crate) fn encode_tier_delete_journal_entry(je: &Jentry) -> Result<Vec<u8>> {
-    serde_json::to_vec(&PersistedTierDeleteJournalEntry::from_jentry(je))
+    serde_json::to_vec(&PersistedTierDeleteJournalEntry::from_jentry(je)?)
         .map_err(|err| Error::other(format!("encode tier delete journal failed: {err}")))
 }
 
@@ -209,6 +261,12 @@ where
 }
 
 pub async fn process_tier_delete_journal_entry(api: Arc<ECStore>, je: &Jentry) -> std::io::Result<()> {
+    if je.version_state == rustfs_filemeta::TransitionVersionState::Unknown {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "tier delete journal remote version state is unknown",
+        ));
+    }
     let backend_identity = je
         .backend_identity
         .ok_or_else(|| std::io::Error::other("legacy tier delete journal has no durable backend identity"))?;
@@ -406,8 +464,9 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        TIER_DELETE_JOURNAL_EXACT_VERSION, await_tier_delete_journal_recovery, decode_tier_delete_journal_entry,
+        TIER_DELETE_JOURNAL_EXACT_VERSION, TIER_DELETE_JOURNAL_STATE_VERSION, await_tier_delete_journal_recovery,
-        encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity, tier_delete_journal_object_name,
+        decode_tier_delete_journal_entry, encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity,
+        tier_delete_journal_object_name,
     };
     use crate::bucket::lifecycle::tier_sweeper::Jentry;
     use crate::error::Result;
@@ -420,7 +479,8 @@ mod tests {
             version_id: "remote-version".to_string(),
             tier_name: "WARM".to_string(),
             backend_identity: Some([7; 32]),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         }
     }
 
@@ -436,6 +496,7 @@ mod tests {
         assert_eq!(decoded.tier_name, je.tier_name);
         assert_eq!(decoded.backend_identity, je.backend_identity);
         assert_eq!(decoded.version_id_exact, je.version_id_exact);
+        assert_eq!(decoded.version_state, je.version_state);
     }
 
     #[test]
@@ -450,7 +511,7 @@ mod tests {
         let persisted: serde_json::Value = serde_json::from_slice(&encoded).expect("exact journal JSON should decode");
         let decoded = decode_tier_delete_journal_entry(&encoded).expect("exact journal entry should decode");
 
-        assert_eq!(persisted["version"], TIER_DELETE_JOURNAL_EXACT_VERSION);
+        assert_eq!(persisted["version"], TIER_DELETE_JOURNAL_STATE_VERSION);
         assert_eq!(persisted["version_id_exact"], true);
         assert!(decoded.version_id_exact);
         assert_ne!(tier_delete_journal_object_name(&exact), tier_delete_journal_object_name(&normalized));
@@ -513,6 +574,46 @@ mod tests {
         }
     }
 
+    #[test]
+    fn tier_delete_journal_rejects_conflicting_v4_version_states() {
+        let identity = vec![7_u8; 32];
+        let invalid = [
+            ("known-disabled", "unexpected", false),
+            ("suspended-null", "", true),
+            ("suspended-null", "null", false),
+            ("exact", "", true),
+            ("exact", "null", true),
+            ("exact", "version", false),
+            ("unknown", "version", true),
+        ];
+
+        for (state, version_id, exact) in invalid {
+            let persisted = serde_json::json!({
+                "version": TIER_DELETE_JOURNAL_STATE_VERSION,
+                "obj_name": "remote/object",
+                "version_id": version_id,
+                "tier_name": "WARM",
+                "backend_identity": identity,
+                "version_id_exact": exact.then_some(true),
+                "version_state": state,
+            });
+            let encoded = serde_json::to_vec(&persisted).expect("invalid journal fixture should encode");
+            decode_tier_delete_journal_entry(&encoded).expect_err("conflicting v4 version state must fail closed");
+        }
+    }
+
+    #[test]
+    fn legacy_journals_decode_with_unknown_version_state() {
+        let v1 = br#"{"version":1,"obj_name":"remote/object","version_id":"opaque","tier_name":"WARM"}"#;
+        let v2 = br#"{"version":2,"obj_name":"remote/object","version_id":"opaque","tier_name":"WARM","backend_identity":[7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7,7]}"#;
+
+        for payload in [v1.as_slice(), v2.as_slice()] {
+            let decoded = decode_tier_delete_journal_entry(payload).expect("legacy journal should decode");
+            assert_eq!(decoded.version_state, rustfs_filemeta::TransitionVersionState::Unknown);
+            assert!(!decoded.version_id_exact);
+        }
+    }
+
     #[test]
     fn tier_delete_journal_path_is_stable_and_sanitized() {
         let je = journal_entry();
@@ -530,6 +631,8 @@ mod tests {
     fn tier_delete_journal_paths_separate_legacy_and_backend_identities() {
         let mut legacy = journal_entry();
         legacy.backend_identity = None;
+        legacy.version_id_exact = false;
+        legacy.version_state = rustfs_filemeta::TransitionVersionState::Unknown;
         let mut backend_a = journal_entry();
         backend_a.backend_identity = Some([1; 32]);
         let mut backend_b = journal_entry();
@@ -575,6 +678,8 @@ mod tests {
     fn tier_delete_journal_without_transition_identity_stays_legacy() {
         let mut je = journal_entry();
         je.backend_identity = None;
+        je.version_id_exact = false;
+        je.version_state = rustfs_filemeta::TransitionVersionState::Unknown;
 
         let encoded = encode_tier_delete_journal_entry(&je).expect("legacy journal should remain encodable");
         let persisted: serde_json::Value = serde_json::from_slice(&encoded).expect("journal JSON should decode");

crates/ecstore/src/bucket/lifecycle/tier_sweeper.rs

@@ -185,6 +185,7 @@ struct ObjSweeper {
     transition_status: String,
     transition_tier: String,
     transition_version_id: String,
+    transition_version_state: rustfs_filemeta::TransitionVersionState,
     remote_object: String,
 }
 
@@ -231,7 +232,9 @@ impl ObjSweeper {
     }
 
     pub fn should_remove_remote_object(&self) -> Option<Jentry> {
-        if self.transition_status != lifecycle::TRANSITION_COMPLETE {
+        if self.transition_status != lifecycle::TRANSITION_COMPLETE
+            || self.transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
+        {
             return None;
         }
 
@@ -250,6 +253,7 @@ impl ObjSweeper {
                 tier_name: self.transition_tier.clone(),
                 backend_identity: None,
                 version_id_exact: false,
+                version_state: self.transition_version_state,
             });
         }
         None
@@ -286,6 +290,7 @@ pub struct Jentry {
     pub(crate) tier_name: String,
     pub(crate) backend_identity: Option<TierDestinationId>,
     pub(crate) version_id_exact: bool,
+    pub(crate) version_state: rustfs_filemeta::TransitionVersionState,
 }
 
 impl ExpiryOp for Jentry {
@@ -486,6 +491,7 @@ pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject
         tier_name: transitioned.tier.clone(),
         backend_identity: None,
         version_id_exact: false,
+        version_state: rustfs_filemeta::TransitionVersionState::Unknown,
     })
 }
 

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

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

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")

crates/ecstore/src/config/com.rs

@@ -2543,6 +2543,7 @@ mod tests {
                 data_dir: None,
                 delete_marker: false,
                 transitioned_object: Default::default(),
+                transition_version_state: Default::default(),
                 restore_ongoing: false,
                 restore_expires: None,
                 user_tags: Arc::new(String::new()),

crates/ecstore/src/object_api/types.rs

@@ -180,6 +180,7 @@ pub struct ObjectInfo {
     pub data_dir: Option<Uuid>,
     pub delete_marker: bool,
     pub transitioned_object: TransitionedObject,
+    pub transition_version_state: rustfs_filemeta::TransitionVersionState,
     pub restore_ongoing: bool,
     pub restore_expires: Option<OffsetDateTime>,
     pub user_tags: Arc<String>,
@@ -220,6 +221,7 @@ impl Clone for ObjectInfo {
             data_dir: self.data_dir,
             delete_marker: self.delete_marker,
             transitioned_object: self.transitioned_object.clone(),
+            transition_version_state: self.transition_version_state,
             restore_ongoing: self.restore_ongoing,
             restore_expires: self.restore_expires,
             user_tags: self.user_tags.clone(),
@@ -537,6 +539,7 @@ impl ObjectInfo {
             inlined,
             user_defined: Arc::new(metadata),
             transitioned_object,
+            transition_version_state: fi.transition_version_state,
             checksum: fi.checksum.clone(),
             storage_class,
             restore_ongoing,

crates/ecstore/src/services/tier/tier.rs

@@ -9274,7 +9274,8 @@ mod tests {
             version_id: "v1".to_string(),
             tier_name: "COLD-A".to_string(),
             backend_identity: Some(current_identity),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         };
         journal_store
             .insert_config_object(

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -545,6 +545,7 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
         && left.transition_tier == right.transition_tier
         && left.transition_version_id == right.transition_version_id
         && left.transition_version == right.transition_version
+        && left.transition_version_state == right.transition_version_state
         && left.expire_restored == right.expire_restored
         && left.size == right.size
         && left.mod_time == right.mod_time

crates/ecstore/src/set_disk/metadata.rs

@@ -579,6 +579,12 @@ impl SetDisks {
         Self::update_hash_str(hasher, &meta.transitioned_objname);
         Self::update_hash_optional_uuid(hasher, meta.transition_version_id);
         Self::update_hash_optional_str(hasher, meta.transition_version.as_deref());
+        hasher.update([match meta.transition_version_state {
+            rustfs_filemeta::TransitionVersionState::Unknown => 0,
+            rustfs_filemeta::TransitionVersionState::KnownDisabled => 1,
+            rustfs_filemeta::TransitionVersionState::SuspendedNull => 2,
+            rustfs_filemeta::TransitionVersionState::Exact => 3,
+        }]);
         Self::update_hash_optional_u32(hasher, meta.mode);
         Self::update_hash_optional_u64(hasher, meta.written_by_version);
 

crates/ecstore/src/set_disk/ops/object.rs

@@ -1692,6 +1692,13 @@ async fn cleanup_rejected_transition_upload_durably(
         tier_name: lease.tier_name().to_string(),
         backend_identity: Some(lease.backend_identity()),
         version_id_exact,
+        version_state: if !version_id_exact {
+            rustfs_filemeta::TransitionVersionState::KnownDisabled
+        } else if cleanup_version == "null" {
+            rustfs_filemeta::TransitionVersionState::SuspendedNull
+        } else {
+            rustfs_filemeta::TransitionVersionState::Exact
+        },
     };
 
     let journal_error = if let Some(api) = api.as_ref() {
@@ -2107,12 +2114,28 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
     }
 }
 
-fn parse_transition_version_id(remote_version: &str) -> Option<String> {
+fn persisted_transition_version(
-    if remote_version.is_empty() || Uuid::parse_str(remote_version).is_ok_and(|version_id| version_id.is_nil()) {
+    remote_version: &str,
-        None
+) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
-    } else {
+    if remote_version.is_empty() {
-        Some(remote_version.to_string())
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "a missing remote tier object version remains unknown until the cluster capability gate is active",
+        ));
     }
+    let version_id = Uuid::parse_str(remote_version).map_err(|_| {
+        std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "opaque remote tier versions require the cluster capability gate",
+        )
+    })?;
+    if version_id.is_nil() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier returned a nil object version ID",
+        ));
+    }
+    Ok((Some(remote_version.to_string()), rustfs_filemeta::TransitionVersionState::Exact))
 }
 
 #[cfg(test)]
@@ -2274,13 +2297,14 @@ mod transition_upload_completion_tests {
 
 #[cfg(test)]
 mod transition_version_id_tests {
-    use super::{TransitionUploadCandidate, parse_transition_version_id};
+    use super::{TransitionUploadCandidate, persisted_transition_version};
+    use rustfs_filemeta::TransitionVersionState;
     use uuid::Uuid;
 
     #[test]
     fn normalizes_persisted_unversioned_ids_and_preserves_put_constraints() {
-        assert_eq!(parse_transition_version_id(""), None);
+        assert!(persisted_transition_version("").is_err());
-        assert_eq!(parse_transition_version_id(&Uuid::nil().to_string()), None);
+        assert!(persisted_transition_version(&Uuid::nil().to_string()).is_err());
         let nil_put_response = Uuid::nil().to_string();
         let nil_candidate = TransitionUploadCandidate::from_put_response(nil_put_response.clone());
         assert_eq!(nil_candidate.cleanup_version(), nil_put_response);
@@ -2292,13 +2316,14 @@ mod transition_version_id_tests {
     }
 
     #[test]
-    fn preserves_uuid_and_opaque_remote_ids() {
+    fn preserves_uuid_and_gates_opaque_remote_ids() {
         let version_id = Uuid::new_v4();
-        assert_eq!(parse_transition_version_id(&version_id.to_string()), Some(version_id.to_string()));
         assert_eq!(
-            parse_transition_version_id("opaque-version-token"),
+            persisted_transition_version(&version_id.to_string()).expect("UUID remote version"),
-            Some("opaque-version-token".to_string())
+            (Some(version_id.to_string()), TransitionVersionState::Exact)
         );
+        assert!(persisted_transition_version("null").is_err());
+        assert!(persisted_transition_version("opaque-version-token").is_err());
         assert_eq!(
             TransitionUploadCandidate::from_put_response(version_id.to_string()).cleanup_version(),
             version_id.to_string()
@@ -3552,7 +3577,20 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object).await;
             return Err(err);
         }
-        let transition_version_id = parse_transition_version_id(candidate.remote_version());
+        let (transition_version_id, transition_version_state) = match persisted_transition_version(candidate.remote_version()) {
+            Ok(version) => version,
+            Err(err) => {
+                let cleanup_api = transition_cleanup_store(&self.ctx).await;
+                if let Err(cleanup_err) = upload_cleanup.cleanup_rejected_upload(cleanup_api).await {
+                    return Err(StorageError::Io(std::io::Error::other(format!(
+                        "{err}; rejected remote upload cleanup failed: {cleanup_err}"
+                    ))));
+                }
+                delete_transition_transaction_after_remote_cleanup(transaction_api.as_ref(), transaction_id, bucket, object)
+                    .await;
+                return Err(err.into());
+            }
+        };
 
         let mut commit_opts = opts.clone();
         commit_opts.no_lock = true;
@@ -3614,6 +3652,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             .as_deref()
             .and_then(|version_id| Uuid::parse_str(version_id).ok());
         current_fi.transition_version = transition_version_id;
+        current_fi.transition_version_state = transition_version_state;
         rustfs_utils::http::metadata_compat::insert_str(
             &mut current_fi.metadata,
             rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
@@ -4429,6 +4468,34 @@ mod transition_commit_failure_tests {
         metadata
     }
 
+    #[tokio::test]
+    async fn rejected_unsupported_remote_versions_are_cleaned_up() {
+        for remote_version in ["", "null", "opaque-version-token"] {
+            let manager = TierConfigMgr::new();
+            let backend = register_mock_tier(&manager, "WARM").await;
+            let lease = TierConfigMgr::acquire_operation_lease(&manager, "WARM")
+                .await
+                .expect("mock tier lease should be available");
+            let candidate = TransitionUploadCandidate::from_put_response(remote_version.to_string());
+
+            persisted_transition_version(candidate.remote_version()).expect_err("unsupported writer version must fail closed");
+            cleanup_rejected_transition_upload_durably(
+                &lease,
+                "remote/object",
+                candidate.cleanup_version(),
+                candidate.cleanup_version_is_exact(),
+                None,
+            )
+            .await
+            .expect("rejected remote upload must be cleaned up");
+
+            assert_eq!(
+                backend.remove_versions().await,
+                vec![("remote/object".to_string(), candidate.cleanup_version().to_string())]
+            );
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn local_commit_failure_returns_error_and_preserves_remote_candidate() {

crates/ecstore/src/store/init.rs

@@ -1311,14 +1311,16 @@ mod tests {
             version_id: "version-a".to_string(),
             tier_name: tier_a.to_string(),
             backend_identity: Some(identity_a),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         };
         let entry_b = Jentry {
             obj_name: "remote-b".to_string(),
             version_id: "version-b".to_string(),
             tier_name: tier_b.to_string(),
             backend_identity: Some(identity_b),
-            version_id_exact: false,
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
         };
         let remove_a = backend_a.arm_failing_remove_barrier().await;
         persist_tier_delete_journal_entry(store_a.clone(), &entry_a)

crates/filemeta/src/fileinfo.rs

@@ -219,6 +219,16 @@ impl ErasureInfo {
 }
 
 // #[derive(Debug, Clone)]
+#[derive(Serialize, Deserialize, Debug, PartialEq, Eq, Clone, Copy, Default)]
+#[serde(rename_all = "kebab-case")]
+pub enum TransitionVersionState {
+    #[default]
+    Unknown,
+    KnownDisabled,
+    SuspendedNull,
+    Exact,
+}
+
 #[derive(Serialize, Deserialize, Debug, PartialEq, Clone, Default)]
 pub struct FileInfo {
     pub volume: String,
@@ -232,6 +242,8 @@ pub struct FileInfo {
     pub transition_version_id: Option<Uuid>,
     #[serde(default)]
     pub transition_version: Option<String>,
+    #[serde(default)]
+    pub transition_version_state: TransitionVersionState,
     pub expire_restored: bool,
     pub data_dir: Option<Uuid>,
     pub mod_time: Option<OffsetDateTime>,
@@ -544,6 +556,20 @@ impl FileInfo {
         {
             return Err(Error::FileCorrupt);
         }
+        let transition_state_valid = match self.transition_version_state {
+            TransitionVersionState::Unknown => true,
+            TransitionVersionState::KnownDisabled => self.transition_version.is_none() && self.transition_version_id.is_none(),
+            TransitionVersionState::SuspendedNull => {
+                self.transition_version.as_deref() == Some("null") && self.transition_version_id.is_none()
+            }
+            TransitionVersionState::Exact => self
+                .transition_version
+                .as_deref()
+                .is_some_and(|version| version != "null" && !version.is_empty()),
+        };
+        if !transition_state_valid {
+            return Err(Error::FileCorrupt);
+        }
 
         let erasure_layout = match mode {
             ValidationMode::RequireErasure => Some(self.validate_erasure_geometry()?),
@@ -837,6 +863,7 @@ impl FileInfo {
             && self.transitioned_objname == other.transitioned_objname
             && self.transition_version_id == other.transition_version_id
             && self.transition_version == other.transition_version
+            && self.transition_version_state == other.transition_version_state
     }
 
     /// Check if metadata maps are equal
@@ -1722,6 +1749,11 @@ mod tests {
                     transition_tier,
                     transition_version_id,
                     transition_version: transition_version_id.map(|version_id| version_id.to_string()),
+                    transition_version_state: if transition_version_id.is_some() {
+                        TransitionVersionState::Exact
+                    } else {
+                        TransitionVersionState::Unknown
+                    },
                     expire_restored,
                     data_dir,
                     mod_time,

crates/filemeta/src/filemeta/version.rs

@@ -26,13 +26,14 @@ use super::msgp_decode::{
     PrependByteReader, prealloc_hint, read_exact_vec, read_nil_or_array_len, read_nil_or_map_len, skip_msgp_value,
 };
 use super::*;
-use crate::ChecksumInfo;
+use crate::{ChecksumInfo, TransitionVersionState};
 use rustfs_utils::HashAlgorithm;
 use rustfs_utils::http::{
     RUSTFS_INTERNAL_PREFIX, SUFFIX_CRC, SUFFIX_FREE_VERSION, SUFFIX_INLINE_DATA, SUFFIX_PURGESTATUS, SUFFIX_TIER_FV_ID,
     SUFFIX_TIER_FV_MARKER, SUFFIX_TRANSITION_STATUS, SUFFIX_TRANSITION_TIER, SUFFIX_TRANSITION_TIER_DESTINATION_ID,
-    SUFFIX_TRANSITIONED_OBJECTNAME, SUFFIX_TRANSITIONED_VERSION_ID, contains_key_bytes, get_bytes, get_consistent_bytes, get_str,
+    SUFFIX_TRANSITIONED_OBJECTNAME, SUFFIX_TRANSITIONED_VERSION_ID, SUFFIX_TRANSITIONED_VERSION_STATE, contains_key_bytes,
-    has_internal_suffix, insert_bytes, is_internal_key, remove_bytes, strip_internal_prefix,
+    get_bytes, get_consistent_bytes, get_str, has_internal_suffix, insert_bytes, is_internal_key, remove_bytes,
+    strip_internal_prefix,
 };
 
 const MSGPACK_EXT8: u8 = 0xc7;
@@ -255,28 +256,83 @@ fn parse_legacy_uuid_bytes(bytes: &[u8], field: &str) -> Result<Option<Uuid>> {
 /// Legacy RustFS writes used 16 raw UUID bytes. New writes and MinIO-migrated
 /// records use the provider's exact UTF-8 version text. Empty, nil UUID, and
 /// malformed bytes are not usable remote versions.
-fn transitioned_version_from_meta_sys(meta_sys: &HashMap<String, Vec<u8>>) -> Option<String> {
+fn transitioned_version_from_meta_sys(meta_sys: &HashMap<String, Vec<u8>>) -> Result<Option<String>> {
-    let value = get_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_ID)?;
+    if !contains_key_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_ID) {
+        return Ok(None);
+    }
+    let Some(value) = get_consistent_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_ID) else {
+        return Ok(None);
+    };
+    let value = value.to_vec();
     if value.is_empty() {
-        return None;
+        return Ok(None);
     }
     if let Ok(id) = Uuid::from_slice(&value) {
-        return (!id.is_nil()).then(|| id.to_string());
+        return Ok((!id.is_nil()).then(|| id.to_string()));
     }
-    let value = String::from_utf8(value).ok()?;
+    let Ok(value) = String::from_utf8(value) else {
+        return Ok(None);
+    };
     if value.is_empty()
         || value.len() > MAX_TRANSITION_VERSION_LEN
         || value.chars().any(char::is_control)
         || Uuid::parse_str(&value).is_ok_and(|id| id.is_nil())
     {
-        None
+        Ok(None)
     } else {
-        Some(value)
+        Ok(Some(value))
+    }
+}
+
+fn transition_version_state_from_meta_sys(
+    meta_sys: &HashMap<String, Vec<u8>>,
+    version: Option<&str>,
+) -> Result<TransitionVersionState> {
+    if !contains_key_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE) {
+        return Ok(TransitionVersionState::Unknown);
+    }
+    let value = get_consistent_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE).ok_or(Error::FileCorrupt)?;
+    let state = match value {
+        b"known-disabled" => TransitionVersionState::KnownDisabled,
+        b"suspended-null" => TransitionVersionState::SuspendedNull,
+        b"exact" => TransitionVersionState::Exact,
+        b"unknown" => TransitionVersionState::Unknown,
+        _ => return Err(Error::FileCorrupt),
+    };
+    let valid = match state {
+        TransitionVersionState::Unknown | TransitionVersionState::KnownDisabled => version.is_none(),
+        TransitionVersionState::SuspendedNull => version == Some("null"),
+        TransitionVersionState::Exact => version.is_some_and(|value| value != "null"),
+    };
+    valid.then_some(state).ok_or(Error::FileCorrupt)
+}
+
+fn transition_version_state_bytes(state: TransitionVersionState) -> &'static [u8] {
+    match state {
+        TransitionVersionState::Unknown => b"unknown",
+        TransitionVersionState::KnownDisabled => b"known-disabled",
+        TransitionVersionState::SuspendedNull => b"suspended-null",
+        TransitionVersionState::Exact => b"exact",
+    }
+}
+
+fn set_transition_version_state(meta_sys: &mut HashMap<String, Vec<u8>>, state: TransitionVersionState) {
+    if state == TransitionVersionState::Unknown {
+        remove_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE);
+    } else {
+        insert_bytes(
+            meta_sys,
+            SUFFIX_TRANSITIONED_VERSION_STATE,
+            transition_version_state_bytes(state).to_vec(),
+        );
     }
 }
 
 fn legacy_transitioned_version_id_from_meta_sys(meta_sys: &HashMap<String, Vec<u8>>) -> Option<Uuid> {
-    transitioned_version_from_meta_sys(meta_sys).and_then(|value| Uuid::parse_str(&value).ok())
+    transitioned_version_from_meta_sys(meta_sys)
+        .ok()
+        .flatten()
+        .and_then(|value| Uuid::parse_str(&value).ok())
 }
 
 fn transitioned_version_bytes(fi: &FileInfo) -> Option<Vec<u8>> {
@@ -2414,7 +2470,8 @@ impl MetaObject {
         let transitioned_objname = get_bytes(&self.meta_sys, SUFFIX_TRANSITIONED_OBJECTNAME)
             .map(|v| String::from_utf8_lossy(&v).to_string())
             .unwrap_or_default();
-        let transition_version = transitioned_version_from_meta_sys(&self.meta_sys);
+        let transition_version = transitioned_version_from_meta_sys(&self.meta_sys)?;
+        let transition_version_state = transition_version_state_from_meta_sys(&self.meta_sys, transition_version.as_deref())?;
         let transition_version_id = transition_version.as_deref().and_then(|value| Uuid::parse_str(value).ok());
         let transition_tier = get_bytes(&self.meta_sys, SUFFIX_TRANSITION_TIER)
             .map(|v| String::from_utf8_lossy(&v).to_string())
@@ -2437,6 +2494,7 @@ impl MetaObject {
             transitioned_objname,
             transition_version_id,
             transition_version,
+            transition_version_state,
             transition_tier,
             ..Default::default()
         })
@@ -2452,6 +2510,7 @@ impl MetaObject {
         if let Some(transition_version) = transitioned_version_bytes(fi) {
             insert_bytes(&mut self.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, transition_version);
         }
+        set_transition_version_state(&mut self.meta_sys, fi.transition_version_state);
         insert_bytes(&mut self.meta_sys, SUFFIX_TRANSITION_TIER, fi.transition_tier.as_bytes().to_vec());
         if let Some(destination_id) = get_str(&fi.metadata, SUFFIX_TRANSITION_TIER_DESTINATION_ID) {
             insert_bytes(&mut self.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID, destination_id.into_bytes());
@@ -2515,6 +2574,7 @@ impl MetaObject {
                 SUFFIX_TRANSITION_TIER,
                 SUFFIX_TRANSITIONED_OBJECTNAME,
                 SUFFIX_TRANSITIONED_VERSION_ID,
+                SUFFIX_TRANSITIONED_VERSION_STATE,
             ] {
                 if let Some(v) = get_bytes(&self.meta_sys, suffix) {
                     insert_bytes(&mut delete_marker.meta_sys, suffix, v);
@@ -2579,6 +2639,9 @@ impl From<FileInfo> for MetaObject {
         if let Some(transition_version) = transitioned_version_bytes(&value) {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, transition_version);
         }
+        if !value.transition_status.is_empty() {
+            set_transition_version_state(&mut meta_sys, value.transition_version_state);
+        }
 
         if !value.transition_tier.is_empty() {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITION_TIER, value.transition_tier.as_bytes().to_vec());
@@ -2720,8 +2783,11 @@ impl MetaDeleteMarker {
                 .map(|v| String::from_utf8_lossy(&v).to_string())
                 .unwrap_or_default();
 
-            fi.transition_version = transitioned_version_from_meta_sys(&self.meta_sys);
+            fi.transition_version = transitioned_version_from_meta_sys(&self.meta_sys).ok().flatten();
             fi.transition_version_id = legacy_transitioned_version_id_from_meta_sys(&self.meta_sys);
+            fi.transition_version_state =
+                transition_version_state_from_meta_sys(&self.meta_sys, fi.transition_version.as_deref())
+                    .unwrap_or(TransitionVersionState::Unknown);
         }
 
         fi
@@ -2877,6 +2943,9 @@ impl From<FileInfo> for MetaDeleteMarker {
         if let Some(transition_version) = transitioned_version_bytes(&value) {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, transition_version);
         }
+        if !value.transition_status.is_empty() || value.tier_free_version() {
+            set_transition_version_state(&mut meta_sys, value.transition_version_state);
+        }
         if !value.transition_tier.is_empty() {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITION_TIER, value.transition_tier.as_bytes().to_vec());
         }
@@ -4125,6 +4194,7 @@ mod tests {
             .expect("into_fileinfo");
         assert_eq!(fi.transition_version_id, Some(id));
         assert_eq!(fi.transition_version, Some(id.to_string()));
+        assert_eq!(fi.transition_version_state, TransitionVersionState::Unknown);
     }
 
     #[test]
@@ -4136,6 +4206,61 @@ mod tests {
             .expect("opaque transition version id must decode");
         assert_eq!(fi.transition_version_id, None);
         assert_eq!(fi.transition_version.as_deref(), Some("opaque-generation-42"));
+        assert_eq!(fi.transition_version_state, TransitionVersionState::Unknown);
+    }
+
+    #[test]
+    fn meta_object_transition_version_state_exact_round_trips_dual_keys() {
+        let id = sample_version_id();
+        let expected_version = id.to_string();
+        let fi = FileInfo {
+            transition_status: "complete".to_string(),
+            transition_version: Some(expected_version.clone()),
+            transition_version_state: TransitionVersionState::Exact,
+            ..Default::default()
+        };
+
+        let object = MetaObject::from(fi);
+        assert_eq!(
+            object
+                .meta_sys
+                .get(&format!("{RUSTFS_INTERNAL_PREFIX}{SUFFIX_TRANSITIONED_VERSION_STATE}"))
+                .map(Vec::as_slice),
+            Some(b"exact".as_slice())
+        );
+        assert_eq!(
+            object
+                .meta_sys
+                .get(&format!(
+                    "{}{SUFFIX_TRANSITIONED_VERSION_STATE}",
+                    rustfs_utils::http::MINIO_INTERNAL_PREFIX
+                ))
+                .map(Vec::as_slice),
+            Some(b"exact".as_slice())
+        );
+        assert_eq!(
+            legacy_transitioned_version_id_from_meta_sys(&object.meta_sys),
+            Some(id),
+            "UUID exact writes must remain readable by the legacy UUID consumer"
+        );
+        let decoded = object.into_fileinfo("b", "k", false).expect("exact state should round trip");
+        assert_eq!(decoded.transition_version_state, TransitionVersionState::Exact);
+        assert_eq!(decoded.transition_version.as_deref(), Some(expected_version.as_str()));
+    }
+
+    #[test]
+    fn meta_object_transition_version_state_conflict_fails_closed() {
+        let mut sys = HashMap::new();
+        insert_bytes(&mut sys, SUFFIX_TRANSITIONED_VERSION_ID, sample_version_id().as_bytes().to_vec());
+        sys.insert(format!("{RUSTFS_INTERNAL_PREFIX}{SUFFIX_TRANSITIONED_VERSION_STATE}"), b"exact".to_vec());
+        sys.insert(
+            format!("{}{SUFFIX_TRANSITIONED_VERSION_STATE}", rustfs_utils::http::MINIO_INTERNAL_PREFIX),
+            b"known-disabled".to_vec(),
+        );
+
+        make_meta_object_with_sys(sys)
+            .into_fileinfo("b", "k", false)
+            .expect_err("conflicting state keys must fail closed");
     }
 
     #[test]

crates/utils/src/http/metadata_compat.rs

@@ -37,6 +37,7 @@ pub const SUFFIX_CRC: &str = "crc";
 pub const SUFFIX_TRANSITION_STATUS: &str = "transition-status";
 pub const SUFFIX_TRANSITIONED_OBJECTNAME: &str = "transitioned-object";
 pub const SUFFIX_TRANSITIONED_VERSION_ID: &str = "transitioned-versionID";
+pub const SUFFIX_TRANSITIONED_VERSION_STATE: &str = "transitioned-version-state";
 pub const SUFFIX_TRANSITION_TIER: &str = "transition-tier";
 pub const SUFFIX_TRANSITION_TIER_DESTINATION_ID: &str = "transition-tier-destination-id";
 pub const SUFFIX_RESTORE_OPERATION_ID: &str = "restore-operation-id";