Merge remote-tracking branch 'origin/main' into cxymds/fix-transition-identity-flake

# Conflicts:
#	crates/ecstore/src/set_disk/ops/object.rs

crates/ecstore/src/api/mod.rs

@@ -128,7 +128,7 @@ pub mod bucket {
             get_object_lock_config, get_public_access_block_config, get_quota_config, get_replication_config,
             get_request_payment_config, get_sse_config, get_tagging_config, get_versioning_config, get_website_config,
             init_bucket_metadata_sys, list_bucket_targets, remove_bucket_metadata, set_bucket_metadata, update,
-            update_bucket_targets_under_transaction_lock,
+            update_bucket_targets_under_transaction_lock, update_config_with,
         };
     }
 

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(())
@@ -4201,6 +4202,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,
@@ -4210,6 +4228,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) => {
@@ -5506,6 +5525,7 @@ mod tests {
                 tier: tier.clone(),
                 ..Default::default()
             },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
             ..Default::default()
         };
 
@@ -5569,6 +5589,7 @@ mod tests {
                 tier,
                 ..Default::default()
             },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Exact,
             ..Default::default()
         };
 
@@ -5591,6 +5612,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() {
@@ -5640,6 +5725,7 @@ mod tests {
         oi.transitioned_object.tier = "WARM".to_string();
         oi.transitioned_object.name = "remote/object".to_string();
         oi.transitioned_object.version_id = "remote-version".to_string();
+        oi.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact;
         let local_delete_calls = Arc::new(std::sync::atomic::AtomicUsize::new(0));
 
         let legacy_err = delete_free_version_remote_object_then(&oi, &manager, {
@@ -5650,7 +5736,8 @@ mod tests {
         })
         .await
         .expect_err("legacy free-version without identity must be retained");
-        assert!(legacy_err.to_string().contains("no durable backend identity"));
+        assert_eq!(legacy_err.kind(), std::io::ErrorKind::Other);
+        assert_eq!(old_backend.remove_count().await, 0);
         assert_eq!(local_delete_calls.load(Ordering::Relaxed), 0);
 
         let mut invalid_metadata = HashMap::new();
@@ -5781,6 +5868,7 @@ mod tests {
         oi.transitioned_object.tier = "WARM".to_string();
         oi.transitioned_object.name = "remote/object".to_string();
         oi.transitioned_object.version_id = "remote-version".to_string();
+        oi.transition_version_state = rustfs_filemeta::TransitionVersionState::Exact;
 
         let err = match get_transitioned_object_reader_with_tier_manager(
             "bucket",
@@ -5796,7 +5884,12 @@ mod tests {
             Ok(_) => panic!("identity-bound GET must reject a same-name tier rebind"),
             Err(err) => err,
         };
-        assert!(err.to_string().contains("identity no longer matches"));
+        assert_eq!(err.kind(), std::io::ErrorKind::Other);
+        let admin_err = err
+            .get_ref()
+            .and_then(|source| source.downcast_ref::<crate::client::admin_handler_utils::AdminError>())
+            .expect("identity mismatch should retain the typed tier error");
+        assert_eq!(admin_err.code, crate::services::tier::tier::ERR_TIER_INVALID_CONFIG.code);
         assert_eq!(new_backend.get_count().await, 0);
 
         oi.user_defined = Arc::new(HashMap::new());
@@ -5902,7 +5995,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
@@ -6013,7 +6107,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
@@ -10081,6 +10176,87 @@ 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);
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn journal_replay_deletes_confirmed_exact_provider_token() {
+        let (_disk_paths, ecstore) = setup_test_env().await;
+        let (backend, _) = register_recovery_mock_tier(&ecstore).await;
+        let lease = TierConfigMgr::acquire_operation_lease(&ecstore.tier_config_mgr(), "WARM")
+            .await
+            .expect("mock tier lease should be available");
+        let identity = lease.backend_identity();
+        backend
+            .set_put_remote_version(Some("provider-version-token".to_string()))
+            .await;
+        lease
+            .put(
+                "remote/object",
+                crate::client::transition_api::ReaderImpl::Body(bytes::Bytes::from_static(b"candidate")),
+                9,
+            )
+            .await
+            .expect("confirmed remote candidate should be seeded");
+        backend.set_remove_failure(true);
+        backend.set_reject_non_empty_remote_versions(true);
+        let je = Jentry {
+            obj_name: "remote/object".to_string(),
+            version_id: "provider-version-token".to_string(),
+            tier_name: "WARM".to_string(),
+            backend_identity: Some(identity),
+            version_id_exact: true,
+            version_state: rustfs_filemeta::TransitionVersionState::Exact,
+        };
+
+        crate::set_disk::cleanup_rejected_transition_upload_durably(
+            &lease,
+            &je.obj_name,
+            &je.version_id,
+            true,
+            Some(ecstore.clone()),
+        )
+        .await
+        .expect("failed immediate cleanup should remain durable in the journal");
+        assert!(backend.contains(&je.obj_name).await);
+
+        backend.set_remove_failure(false);
+        crate::bucket::lifecycle::tier_delete_journal::process_tier_delete_journal_entry(ecstore, &je)
+            .await
+            .expect("identity-bound exact journal must retry confirmed candidate cleanup");
+
+        assert!(!backend.contains(&je.obj_name).await);
+        assert_eq!(backend.exact_remove_count(), 2);
+        assert_eq!(
+            backend.remove_versions().await,
+            vec![("remote/object".to_string(), "provider-version-token".to_string())]
+        );
+    }
+
     async fn seed_recoverable_free_version(
         disk_paths: &[PathBuf],
         bucket: &str,
@@ -10100,6 +10276,7 @@ mod tests {
                 identity,
             );
         }
+        let transition_version_id = Uuid::new_v4();
         let mut metadata = FileMeta::new();
         metadata
             .add_version(FileInfo {
@@ -10108,7 +10285,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

@@ -20,7 +20,10 @@ use tokio_util::sync::CancellationToken;
 use tracing::{debug, warn};
 
 use crate::bucket::lifecycle::config_boundary;
-use crate::bucket::lifecycle::tier_sweeper::{Jentry, delete_object_from_remote_tier_idempotent_with_manager_and_identity};
+use crate::bucket::lifecycle::tier_sweeper::{
+    Jentry, delete_confirmed_transition_candidate_exact_with_manager_and_identity,
+    delete_object_from_remote_tier_idempotent_with_manager_and_identity,
+};
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result};
 use crate::object_api::{GetObjectReader, ObjectInfo, ObjectOptions, PutObjReader};
@@ -42,6 +45,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 +59,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 {
-        Self {
-            version: if je.version_id_exact {
-                TIER_DELETE_JOURNAL_EXACT_VERSION
-            } else if je.backend_identity.is_some() {
+    fn from_jentry(je: &Jentry) -> Result<Self> {
+        validate_version_state(je.version_state, &je.version_id, je.version_id_exact)?;
+        let legacy_unknown = je.version_state == rustfs_filemeta::TransitionVersionState::Unknown;
+        let version = if legacy_unknown {
+            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 +99,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 {
-            1 => (None, false),
+        let (backend_identity, version_id_exact, version_state) = match self.version {
+            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 +127,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 +153,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 +214,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,18 +264,35 @@ 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"))?;
-    delete_object_from_remote_tier_idempotent_with_manager_and_identity(
-        &je.obj_name,
-        &je.version_id,
-        &je.tier_name,
-        backend_identity,
-        &api.tier_config_mgr(),
-        je.version_id_exact,
-    )
-    .await?;
+    if je.version_id_exact {
+        delete_confirmed_transition_candidate_exact_with_manager_and_identity(
+            &je.obj_name,
+            &je.version_id,
+            &je.tier_name,
+            backend_identity,
+            &api.tier_config_mgr(),
+        )
+        .await?;
+    } else {
+        delete_object_from_remote_tier_idempotent_with_manager_and_identity(
+            &je.obj_name,
+            &je.version_id,
+            &je.tier_name,
+            backend_identity,
+            &api.tier_config_mgr(),
+            false,
+        )
+        .await?;
+    }
     remove_tier_delete_journal_entry(api, je).await
 }
 
@@ -406,8 +478,9 @@ where
 #[cfg(test)]
 mod tests {
     use super::{
-        TIER_DELETE_JOURNAL_EXACT_VERSION, await_tier_delete_journal_recovery, decode_tier_delete_journal_entry,
-        encode_tier_delete_journal_entry, record_tier_delete_journal_backend_identity, tier_delete_journal_object_name,
+        TIER_DELETE_JOURNAL_EXACT_VERSION, TIER_DELETE_JOURNAL_STATE_VERSION, await_tier_delete_journal_recovery,
+        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 +493,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 +510,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 +525,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 +588,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 +645,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 +692,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;
         }
 
@@ -249,7 +252,11 @@ impl ObjSweeper {
                 version_id: self.transition_version_id.clone(),
                 tier_name: self.transition_tier.clone(),
                 backend_identity: None,
-                version_id_exact: false,
+                version_id_exact: matches!(
+                    self.transition_version_state,
+                    rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact
+                ),
+                version_state: self.transition_version_state,
             });
         }
         None
@@ -286,6 +293,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 {
@@ -330,7 +338,7 @@ async fn delete_object_from_remote_tier_raw_with_manager(
     let lease = TierConfigMgr::acquire_operation_lease(&tier_config_mgr, tier_name)
         .await
         .map_err(std::io::Error::other)?;
-    delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, &lease, false).await
+    delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, &lease, false, true).await
 }
 
 async fn delete_object_from_remote_tier_raw_with_lease(
@@ -338,8 +346,11 @@ async fn delete_object_from_remote_tier_raw_with_lease(
     rv_id: &str,
     lease: &TierOperationLease,
     version_id_exact: bool,
+    validate_remote_version_id: bool,
 ) -> Result<(), std::io::Error> {
-    lease.validate_remote_version_id(rv_id)?;
+    if validate_remote_version_id {
+        lease.validate_remote_version_id(rv_id)?;
+    }
 
     if remote_delete_breaker_is_open(Instant::now()).await {
         metrics::counter!(METRIC_DELETE_REMOTE_BREAKER_TOTAL).increment(1);
@@ -435,7 +446,53 @@ pub(crate) async fn delete_object_from_remote_tier_with_lease_idempotent(
     lease: &TierOperationLease,
     version_id_exact: bool,
 ) -> Result<RemoteTierDeleteOutcome, std::io::Error> {
-    match delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, lease, version_id_exact).await {
+    delete_object_from_remote_tier_with_lease_idempotent_inner(obj_name, rv_id, lease, version_id_exact, true).await
+}
+
+pub(crate) async fn delete_confirmed_transition_candidate_exact_with_lease_idempotent(
+    obj_name: &str,
+    rv_id: &str,
+    lease: &TierOperationLease,
+) -> Result<RemoteTierDeleteOutcome, std::io::Error> {
+    if rv_id.is_empty() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidInput,
+            "confirmed versioned transition candidate requires a non-empty remote version",
+        ));
+    }
+    #[cfg(test)]
+    if obj_name == "remote/empty-guard-probe" {
+        CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES.fetch_add(1, std::sync::atomic::Ordering::Relaxed);
+    }
+    delete_object_from_remote_tier_with_lease_idempotent_inner(obj_name, rv_id, lease, true, false).await
+}
+
+#[cfg(test)]
+static CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES: std::sync::atomic::AtomicUsize = std::sync::atomic::AtomicUsize::new(0);
+
+pub(crate) async fn delete_confirmed_transition_candidate_exact_with_manager_and_identity(
+    obj_name: &str,
+    rv_id: &str,
+    tier_name: &str,
+    backend_identity: TierDestinationId,
+    tier_config_mgr: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
+) -> Result<RemoteTierDeleteOutcome, std::io::Error> {
+    let lease = TierConfigMgr::acquire_operation_lease_for_backend_identity(tier_config_mgr, tier_name, backend_identity)
+        .await
+        .map_err(std::io::Error::other)?;
+    delete_confirmed_transition_candidate_exact_with_lease_idempotent(obj_name, rv_id, &lease).await
+}
+
+async fn delete_object_from_remote_tier_with_lease_idempotent_inner(
+    obj_name: &str,
+    rv_id: &str,
+    lease: &TierOperationLease,
+    version_id_exact: bool,
+    validate_remote_version_id: bool,
+) -> Result<RemoteTierDeleteOutcome, std::io::Error> {
+    match delete_object_from_remote_tier_raw_with_lease(obj_name, rv_id, lease, version_id_exact, validate_remote_version_id)
+        .await
+    {
         Ok(()) => Ok(RemoteTierDeleteOutcome::Deleted),
         Err(err) if is_remote_tier_not_found_error(&err) => Ok(RemoteTierDeleteOutcome::AlreadyRemoved),
         Err(err) => {
@@ -460,6 +517,7 @@ pub fn transitioned_delete_journal_entry(
     versioned: bool,
     suspended: bool,
     transitioned: &TransitionedObject,
+    transition_version_state: rustfs_filemeta::TransitionVersionState,
 ) -> Option<Jentry> {
     let sweeper = ObjSweeper {
         version_id,
@@ -468,6 +526,7 @@ pub fn transitioned_delete_journal_entry(
         transition_status: transitioned.status.clone(),
         transition_tier: transitioned.tier.clone(),
         transition_version_id: transitioned.version_id.clone(),
+        transition_version_state,
         remote_object: transitioned.name.clone(),
         ..Default::default()
     };
@@ -475,8 +534,13 @@ pub fn transitioned_delete_journal_entry(
     sweeper.should_remove_remote_object()
 }
 
-pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject) -> Option<Jentry> {
-    if transitioned.status != lifecycle::TRANSITION_COMPLETE {
+pub fn transitioned_force_delete_journal_entry(
+    transitioned: &TransitionedObject,
+    transition_version_state: rustfs_filemeta::TransitionVersionState,
+) -> Option<Jentry> {
+    if transitioned.status != lifecycle::TRANSITION_COMPLETE
+        || transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
+    {
         return None;
     }
 
@@ -485,7 +549,11 @@ pub fn transitioned_force_delete_journal_entry(transitioned: &TransitionedObject
         version_id: transitioned.version_id.clone(),
         tier_name: transitioned.tier.clone(),
         backend_identity: None,
-        version_id_exact: false,
+        version_id_exact: matches!(
+            transition_version_state,
+            rustfs_filemeta::TransitionVersionState::SuspendedNull | rustfs_filemeta::TransitionVersionState::Exact
+        ),
+        version_state: transition_version_state,
     })
 }
 
@@ -494,11 +562,14 @@ mod test {
     use crate::client::signer_error::invalid_utf8_header_error;
 
     use super::{
-        ERR_REMOTE_DELETE_BREAKER_OPEN, ERR_REMOTE_DELETE_LIMITER_CLOSED, RemoteDeleteBreaker, RemoteTierDeleteOutcome,
+        CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES, ERR_REMOTE_DELETE_BREAKER_OPEN, ERR_REMOTE_DELETE_LIMITER_CLOSED,
+        RemoteDeleteBreaker, RemoteTierDeleteOutcome, delete_confirmed_transition_candidate_exact_with_manager_and_identity,
         delete_object_from_remote_tier_idempotent, delete_object_from_remote_tier_idempotent_with_manager_and_identity,
-        is_remote_tier_not_found_error, is_signer_header_error, set_remote_tier_delete_test_hook,
-        should_record_remote_delete_failure,
+        is_remote_tier_not_found_error, is_signer_header_error, lifecycle, set_remote_tier_delete_test_hook,
+        should_record_remote_delete_failure, transitioned_delete_journal_entry, transitioned_force_delete_journal_entry,
     };
+    use crate::storage_api_contracts::lifecycle::TransitionedObject;
+    use rustfs_filemeta::TransitionVersionState;
     use std::io::{Error, ErrorKind};
     use std::time::{Duration, Instant};
 
@@ -542,6 +613,43 @@ mod test {
         assert!(should_record_remote_delete_failure(&Error::other("NoSuchVersion")));
     }
 
+    #[test]
+    fn transitioned_delete_journal_preserves_remote_version_state() {
+        let cases = [
+            (TransitionVersionState::Unknown, "legacy-version", None),
+            (TransitionVersionState::KnownDisabled, "", Some(false)),
+            (TransitionVersionState::SuspendedNull, "null", Some(true)),
+            (TransitionVersionState::Exact, "opaque-version", Some(true)),
+        ];
+
+        for (state, version_id, expected_exact) in cases {
+            let transitioned = TransitionedObject {
+                name: "remote/object".to_string(),
+                version_id: version_id.to_string(),
+                tier: "WARM".to_string(),
+                status: lifecycle::TRANSITION_COMPLETE.to_string(),
+                ..Default::default()
+            };
+            let regular = transitioned_delete_journal_entry(None, false, false, &transitioned, state);
+            let forced = transitioned_force_delete_journal_entry(&transitioned, state);
+
+            match expected_exact {
+                Some(expected_exact) => {
+                    let regular = regular.expect("known version state should produce a regular delete journal entry");
+                    assert_eq!(regular.version_state, state);
+                    assert_eq!(regular.version_id_exact, expected_exact);
+                    let forced = forced.expect("known version state should produce a forced delete journal entry");
+                    assert_eq!(forced.version_state, state);
+                    assert_eq!(forced.version_id_exact, expected_exact);
+                }
+                None => {
+                    assert!(regular.is_none());
+                    assert!(forced.is_none());
+                }
+            }
+        }
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn idempotent_remote_delete_treats_hooked_nosuchversion_as_already_removed() {
@@ -664,6 +772,55 @@ mod test {
         assert_eq!(backend.remove_versions().await, vec![("remote/object".to_string(), String::new())]);
     }
 
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn confirmed_transition_cleanup_deletes_exact_provider_token() {
+        CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES.store(0, std::sync::atomic::Ordering::Relaxed);
+        let manager = crate::services::tier::tier::TierConfigMgr::new();
+        let backend = crate::services::tier::test_util::register_mock_tier(&manager, "WARM").await;
+        let lease = crate::services::tier::tier::TierConfigMgr::acquire_operation_lease(&manager, "WARM")
+            .await
+            .expect("test tier lease should be available");
+        let identity = lease.backend_identity();
+        drop(lease);
+        backend.set_reject_non_empty_remote_versions(true);
+
+        let outcome = delete_confirmed_transition_candidate_exact_with_manager_and_identity(
+            "remote/object",
+            "provider-version-token",
+            "WARM",
+            identity,
+            &manager,
+        )
+        .await
+        .expect("confirmed upload compensation should delete the exact provider token");
+
+        assert_eq!(outcome, RemoteTierDeleteOutcome::Deleted);
+        assert_eq!(backend.exact_remove_count(), 1);
+        assert_eq!(
+            backend.remove_versions().await,
+            vec![("remote/object".to_string(), "provider-version-token".to_string())]
+        );
+
+        let err = delete_confirmed_transition_candidate_exact_with_manager_and_identity(
+            "remote/empty-guard-probe",
+            "",
+            "WARM",
+            identity,
+            &manager,
+        )
+        .await
+        .expect_err("confirmed versioned cleanup must reject an empty token");
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidInput);
+        assert_eq!(backend.remove_count().await, 1);
+        assert_eq!(
+            CONFIRMED_TRANSITION_EMPTY_GUARD_DISPATCHES.load(std::sync::atomic::Ordering::Relaxed),
+            0,
+            "empty remote versions must be rejected before exact cleanup dispatch"
+        );
+    }
+
     #[test]
     fn breaker_opens_at_threshold_and_recovers_after_window() {
         let mut breaker = RemoteDeleteBreaker::new(3, Duration::from_secs(30));

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

@@ -22,7 +22,10 @@ use uuid::Uuid;
 
 use crate::bucket::lifecycle::config_boundary;
 use crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE;
-use crate::bucket::lifecycle::tier_sweeper::delete_object_from_remote_tier_idempotent_with_manager_and_identity;
+use crate::bucket::lifecycle::tier_sweeper::{
+    delete_confirmed_transition_candidate_exact_with_manager_and_identity,
+    delete_object_from_remote_tier_idempotent_with_manager_and_identity,
+};
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result as EcstoreResult};
 use crate::object_api::ObjectOptions;
@@ -708,6 +711,21 @@ async fn recover_unknown_upload_outcome(
         TransitionCandidateProbe::UnversionedPresent => {
             cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::unversioned()).await
         }
+        TransitionCandidateProbe::VersionedPresent(version_id)
+            if Uuid::parse_str(&version_id).is_ok_and(|version_id| version_id.is_nil()) =>
+        {
+            delete_confirmed_transition_candidate_exact_with_manager_and_identity(
+                &transaction.remote_object,
+                &version_id,
+                &transaction.tier_name,
+                transaction.backend_fingerprint,
+                &api.tier_config_mgr(),
+            )
+            .await
+            .map_err(Error::other)?;
+            delete_transition_transaction_record(api, transaction.transaction_id).await?;
+            Ok(TransitionTransactionRecoveryOutcome::RemoteCandidateDeleted)
+        }
         TransitionCandidateProbe::VersionedPresent(version_id) => {
             cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await
         }

crates/ecstore/src/bucket/metadata.rs

@@ -737,6 +737,9 @@ impl BucketMetadata {
             }
             BUCKET_TAGGING_CONFIG => {
                 self.tagging_config_xml = data;
+                // Drop the parsed form (like lifecycle above) so clearing the
+                // payload can't leave stale parsed tags to be cached.
+                self.tagging_config = None;
                 self.tagging_config_updated_at = updated;
             }
             BUCKET_QUOTA_CONFIG_FILE => {
@@ -1318,6 +1321,30 @@ mod test {
         assert!(bm.lifecycle_config.is_none());
     }
 
+    /// Companion to the lifecycle case above. `parse_all_configs` skips empty
+    /// XML rather than clearing, so without the explicit reset a cleared
+    /// tagging config would keep serving the previously parsed tags.
+    #[test]
+    fn tagging_update_config_clears_parsed_config_on_delete() {
+        let mut bm = BucketMetadata::new("test-bucket");
+        let tagging_xml = br#"<Tagging><TagSet><Tag><Key>env</Key><Value>prod</Value></Tag></TagSet></Tagging>"#;
+
+        bm.update_config(BUCKET_TAGGING_CONFIG, tagging_xml.to_vec())
+            .expect("tagging config should update");
+        bm.parse_all_configs().expect("tagging config should parse");
+        assert!(bm.tagging_config.is_some());
+
+        bm.update_config(BUCKET_TAGGING_CONFIG, Vec::new())
+            .expect("tagging config delete should update metadata");
+
+        assert!(bm.tagging_config_xml.is_empty());
+        assert!(bm.tagging_config.is_none());
+
+        // A re-parse must not resurrect them either.
+        bm.parse_all_configs().expect("cleared tagging should parse");
+        assert!(bm.tagging_config.is_none());
+    }
+
     #[tokio::test]
     async fn marshal_msg_complete_example() {
         // Create a complete BucketMetadata with various configurations

crates/ecstore/src/bucket/metadata_sys.rs

@@ -239,6 +239,35 @@ pub async fn update_bucket_targets_under_transaction_lock(bucket: &str, data: Ve
     bucket_meta_sys.update(bucket, BUCKET_TARGETS_FILE, data).await
 }
 
+/// Read-modify-write one bucket config file under the metadata system's
+/// outer write guard.
+///
+/// `mutate` sees the freshly loaded on-disk metadata and returns the
+/// replacement payload for `config_file` (empty clears it, like
+/// [`delete`]). Both the read and the persisted write happen inside the
+/// same guard that [`update`] uses, so within this process the rewrite can
+/// neither clobber a concurrent update to another config file nor lose a
+/// concurrent write to the same one — unlike caching a mutated clone of
+/// previously read metadata.
+///
+/// This guard is process-local. Writers on other nodes still race, exactly
+/// as they do for [`update`]: each rewrites the whole metadata file, so the
+/// later save wins. What this narrows is the window — from "as stale as the
+/// local cache" down to a single metadata read plus write.
+pub async fn update_config_with<F>(bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
+where
+    F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
+{
+    let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
+    let _targets_guard = if config_file == BUCKET_TARGETS_FILE {
+        Some(acquire_bucket_targets_transaction_lock(bucket).await?)
+    } else {
+        None
+    };
+    let mut bucket_meta_sys = bucket_meta_sys_lock.write().await;
+    bucket_meta_sys.update_config_with(bucket, config_file, mutate).await
+}
+
 pub async fn acquire_bucket_targets_transaction_lock(bucket: &str) -> Result<rustfs_lock::NamespaceLockGuard> {
     let bucket_meta_sys_lock = get_bucket_metadata_sys()?;
     let api = bucket_meta_sys_lock.read().await.object_store();
@@ -603,24 +632,7 @@ impl BucketMetadataSys {
             return Err(Error::other("errServerNotInitialized"));
         };
 
-        if is_meta_bucketname(bucket) {
-            return Err(Error::other("errInvalidArgument"));
-        }
-
-        let mut bm = match load_bucket_metadata_parse(store, bucket, parse).await {
-            Ok(res) => res,
-            Err(err) => {
-                if !runtime_sources::setup_is_erasure().await
-                    && !runtime_sources::setup_is_dist_erasure().await
-                    && is_err_bucket_not_found(&err)
-                {
-                    BucketMetadata::new(bucket)
-                } else {
-                    error!("load bucket metadata failed: {}", err);
-                    return Err(err);
-                }
-            }
-        };
+        let mut bm = Self::load_bucket_metadata_for_update(store, bucket, parse).await?;
 
         let updated = bm.update_config(config_file, data)?;
 
@@ -629,6 +641,49 @@ impl BucketMetadataSys {
         Ok(updated)
     }
 
+    /// See the free [`update_config_with`]: same load-mutate-persist cycle as
+    /// [`Self::update`], with the payload computed from the loaded metadata
+    /// instead of supplied up front. Loads through this system's own store so
+    /// the read and the persisted write target the same instance.
+    async fn update_config_with<F>(&mut self, bucket: &str, config_file: &str, mutate: F) -> Result<OffsetDateTime>
+    where
+        F: FnOnce(&BucketMetadata) -> Result<Vec<u8>> + Send,
+    {
+        let mut bm = Self::load_bucket_metadata_for_update(self.api.clone(), bucket, true).await?;
+
+        let data = mutate(&bm)?;
+        let updated = bm.update_config(config_file, data)?;
+
+        self.save(bm).await?;
+
+        Ok(updated)
+    }
+
+    /// Load a bucket's on-disk metadata as the base of a config rewrite.
+    /// Outside erasure setups a missing metadata file degrades to a fresh
+    /// default (legacy buckets without one); erasure setups fail instead of
+    /// fabricating state that a quorum may still hold.
+    async fn load_bucket_metadata_for_update(store: Arc<ECStore>, bucket: &str, parse: bool) -> Result<BucketMetadata> {
+        if is_meta_bucketname(bucket) {
+            return Err(Error::other("errInvalidArgument"));
+        }
+
+        match load_bucket_metadata_parse(store, bucket, parse).await {
+            Ok(res) => Ok(res),
+            Err(err) => {
+                if !runtime_sources::setup_is_erasure().await
+                    && !runtime_sources::setup_is_dist_erasure().await
+                    && is_err_bucket_not_found(&err)
+                {
+                    Ok(BucketMetadata::new(bucket))
+                } else {
+                    error!("load bucket metadata failed: {}", err);
+                    Err(err)
+                }
+            }
+        }
+    }
+
     async fn save(&self, bm: BucketMetadata) -> Result<()> {
         if is_meta_bucketname(&bm.name) {
             return Err(Error::other("errInvalidArgument"));
@@ -1068,6 +1123,122 @@ mod tests {
 
         assert!(matches!(err, Error::Io(_)), "malformed persisted policy must surface its parse failure");
     }
+    /// A tagging rewrite through `update_config_with` (the Swift metadata
+    /// POST path) is persisted: it survives a metadata reload from disk, and
+    /// an emptied rewrite clears the config in the cached copy too instead of
+    /// leaving stale parsed tags behind.
+    #[tokio::test]
+    async fn update_config_with_persists_tagging_rewrite_across_disk_reload() {
+        use crate::bucket::metadata::BUCKET_TAGGING_CONFIG;
+        use s3s::dto::Tag;
+
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let mut sys = BucketMetadataSys::new(ecstore);
+
+        let bucket = "swift-tagging-bucket";
+        sys.persist_and_set(BucketMetadata::new(bucket))
+            .await
+            .expect("initial metadata should persist");
+
+        let tagging = Tagging {
+            tag_set: vec![Tag {
+                key: Some("swift-meta-color".to_string()),
+                value: Some("blue".to_string()),
+            }],
+        };
+        let xml = crate::bucket::utils::serialize::<Tagging>(&tagging).expect("tagging should serialize");
+        sys.update_config_with(bucket, BUCKET_TAGGING_CONFIG, move |bm| {
+            assert!(bm.tagging_config.is_none(), "rewrite must see the on-disk state");
+            Ok(xml)
+        })
+        .await
+        .expect("tagging rewrite should persist");
+
+        // Simulate the disk-truth reload that used to lose Swift writes: drop
+        // the cached entry and lazily re-load from the metadata file.
+        sys.metadata_map.write().await.clear();
+        let (tags, _) = sys
+            .get_tagging_config(bucket)
+            .await
+            .expect("tagging must survive a reload from disk");
+        assert_eq!(tags.tag_set.len(), 1);
+        assert_eq!(tags.tag_set[0].key.as_deref(), Some("swift-meta-color"));
+        assert_eq!(tags.tag_set[0].value.as_deref(), Some("blue"));
+
+        // An emptied rewrite clears the config everywhere.
+        sys.update_config_with(bucket, BUCKET_TAGGING_CONFIG, |bm| {
+            assert!(bm.tagging_config.is_some(), "rewrite must see the persisted tags");
+            Ok(Vec::new())
+        })
+        .await
+        .expect("clearing rewrite should persist");
+        assert_eq!(
+            sys.get_tagging_config(bucket).await.unwrap_err(),
+            Error::ConfigNotFound,
+            "cleared tagging must not be served from the cache"
+        );
+        sys.metadata_map.write().await.clear();
+        assert_eq!(
+            sys.get_tagging_config(bucket).await.unwrap_err(),
+            Error::ConfigNotFound,
+            "cleared tagging must not reappear after a reload from disk"
+        );
+    }
+
+    /// The load and the persisted write share one write guard, so concurrent
+    /// rewrites of the same config compose instead of clobbering each other.
+    /// Moving the load outside that guard loses all but the last tag.
+    #[tokio::test]
+    async fn concurrent_update_config_with_calls_do_not_lose_writes() {
+        use crate::bucket::metadata::BUCKET_TAGGING_CONFIG;
+        use s3s::dto::Tag;
+
+        let (_dirs, ecstore) = isolated_store_over_temp_disks().await;
+        let sys = Arc::new(RwLock::new(BucketMetadataSys::new(ecstore)));
+
+        let bucket = "swift-tagging-concurrent";
+        sys.read()
+            .await
+            .persist_and_set(BucketMetadata::new(bucket))
+            .await
+            .expect("initial metadata should persist");
+
+        const WRITERS: usize = 8;
+        let mut handles = Vec::with_capacity(WRITERS);
+        for idx in 0..WRITERS {
+            let sys = sys.clone();
+            handles.push(tokio::spawn(async move {
+                sys.write()
+                    .await
+                    .update_config_with(bucket, BUCKET_TAGGING_CONFIG, move |bm| {
+                        // Each writer merges its own tag onto whatever is
+                        // currently persisted — the Swift rewrite shape.
+                        let mut tagging = bm.tagging_config.clone().unwrap_or_else(|| Tagging { tag_set: vec![] });
+                        tagging.tag_set.push(Tag {
+                            key: Some(format!("swift-meta-key{idx}")),
+                            value: Some(idx.to_string()),
+                        });
+                        crate::bucket::utils::serialize::<Tagging>(&tagging).map_err(|e| Error::other(e.to_string()))
+                    })
+                    .await
+            }));
+        }
+
+        for handle in handles {
+            handle
+                .await
+                .expect("writer task should join")
+                .expect("rewrite should persist");
+        }
+
+        let (tags, _) = sys
+            .read()
+            .await
+            .get_tagging_config(bucket)
+            .await
+            .expect("tagging should be readable");
+        assert_eq!(tags.tag_set.len(), WRITERS, "every concurrent rewrite must survive: {tags:?}");
+    }
 
     fn target(bucket: &str, id: &str) -> BucketTarget {
         BucketTarget {

crates/ecstore/src/cluster/rpc/client.rs

@@ -14,7 +14,7 @@
 
 use crate::cluster::rpc::http_auth::RPC_CONTENT_SHA256_HEADER;
 use crate::cluster::rpc::{gen_tonic_signature_headers, normalize_tonic_rpc_audience};
-use crate::disk::error::{DiskError, Error as DiskErrorType};
+use crate::disk::error::{DiskError, Error as DiskErrorType, RpcStatusError};
 use crate::runtime::sources as runtime_sources;
 use http::Uri;
 use rustfs_protos::{
@@ -107,10 +107,79 @@ pub async fn node_service_time_out_client_no_auth(
     node_service_time_out_client(addr, TonicInterceptor::NoOp(NoOpInterceptor)).await
 }
 
+/// The typed `tonic::Status` an internode RPC failure was converted from, if
+/// this error carries one.
+pub(crate) fn embedded_tonic_status(io_err: &std::io::Error) -> Option<&tonic::Status> {
+    io_err.get_ref()?.downcast_ref::<RpcStatusError>().map(RpcStatusError::status)
+}
+
+/// Decide whether a gRPC status reports a peer we cannot currently reach,
+/// rather than an application outcome from a live peer.
+///
+/// `Unavailable` is the one code that means "no service behind this channel":
+/// the client transport raises it when the connection is broken, and the
+/// server's own not-ready gates use it deliberately.
+///
+/// `Unknown` is the client transport's escape hatch for a cause it could not
+/// map to a code — tower's "Service was not ready: <cause>", an h2 error with
+/// no gRPC mapping. Our handlers never return it, so there its message is the
+/// only evidence available and the anchored needles decide.
+///
+/// Every other code is an answer from a live peer and is never a transport
+/// failure, whatever its message says. That distinction is the point of
+/// classifying by code: a peer relaying its own downstream trouble as
+/// `Internal("connection refused ...")`, or a handler interpolating a local
+/// `io::Error` into `Status::internal`, answered us perfectly well. Marking it
+/// offline over that text is the bug this classification replaces. Likewise a
+/// `Cancelled` "Timeout expired" from the per-RPC channel deadline means the
+/// peer is slow, not gone; gating it would turn load into a partition.
+pub(crate) fn is_network_like_status(status: &tonic::Status) -> bool {
+    match status.code() {
+        tonic::Code::Unavailable => true,
+        tonic::Code::Unknown => message_has_network_needle(&status.to_string()),
+        _ => false,
+    }
+}
+
+/// Substring fallback for failures that only exist as text: dial errors
+/// wrapped by `get_client`, remote `error_info` payloads, and statuses
+/// flattened through `format!`. Needles must stay anchored to transport
+/// context — a bare word like "unavailable" also matches application text
+/// (e.g. a bucket named "unavailable-logs") and would take a healthy peer
+/// offline.
+pub(crate) fn message_has_network_needle(message: &str) -> bool {
+    let message = message.to_ascii_lowercase();
+    [
+        "temporarily offline",
+        "transport error",
+        // tonic >= 0.14 renders Code::Unavailable as
+        // `code: 'The service is currently unavailable'`.
+        "code: 'the service is currently unavailable'",
+        // RUSTFS_COMPAT_TODO(tonic-013-status-render): releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered the same status as `status: Unavailable`, and peers relay that text in error_info. Remove after the minimum supported RustFS peer version ships tonic >= 0.14.
+        "status: unavailable",
+        "error trying to connect",
+        "connection refused",
+        "connection reset",
+        "broken pipe",
+        "not connected",
+        "unexpected eof",
+        "timed out",
+        "deadline has elapsed",
+        "connection closed",
+        "connection aborted",
+        "tcp connect error",
+    ]
+    .iter()
+    .any(|needle| message.contains(needle))
+}
+
 pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
     match err {
         DiskError::Timeout => true,
         DiskError::Io(io_err) => {
+            if let Some(status) = embedded_tonic_status(io_err) {
+                return is_network_like_status(status);
+            }
             if matches!(
                 io_err.kind(),
                 ErrorKind::TimedOut
@@ -124,24 +193,7 @@ pub(crate) fn is_network_like_disk_error(err: &DiskErrorType) -> bool {
                 return true;
             }
 
-            let message = io_err.to_string().to_ascii_lowercase();
-            [
-                "transport error",
-                "unavailable",
-                "error trying to connect",
-                "connection refused",
-                "connection reset",
-                "broken pipe",
-                "not connected",
-                "unexpected eof",
-                "timed out",
-                "deadline has elapsed",
-                "connection closed",
-                "connection aborted",
-                "tcp connect error",
-            ]
-            .iter()
-            .any(|needle| message.contains(needle))
+            message_has_network_needle(&io_err.to_string())
         }
         _ => false,
     }
@@ -269,6 +321,70 @@ mod tests {
         let _ = provider.shutdown();
     }
 
+    #[test]
+    fn network_like_disk_error_uses_typed_status_code() {
+        // Transport-level Unavailable statuses justify retry/eviction.
+        assert!(is_network_like_disk_error(&DiskError::from(tonic::Status::unavailable(
+            "storage layer is not initialized"
+        ))));
+        // Application statuses from a live peer must not look network-like,
+        // even when their message contains transport-sounding words.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
+            "failed to heal bucket \"unavailable-logs\""
+        ))));
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
+            "No valid auth token"
+        ))));
+        // A slow peer that blew the per-RPC deadline is still answering.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::cancelled("Timeout expired"))));
+    }
+
+    #[test]
+    fn embedded_tonic_status_is_recovered_across_error_conversions() {
+        // DiskError and StorageError share one wrapper, so a status keeps its
+        // typed classification whichever error it was converted into first.
+        let from_storage: DiskErrorType = crate::error::Error::from(tonic::Status::unavailable("peer gone")).into();
+        let DiskError::Io(io_err) = &from_storage else {
+            panic!("status-derived disk error should stay an Io error");
+        };
+        assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
+
+        let from_disk = crate::error::Error::from(DiskError::from(tonic::Status::unavailable("peer gone")));
+        let crate::error::Error::Io(io_err) = &from_disk else {
+            panic!("status-derived storage error should stay an Io error");
+        };
+        assert_eq!(embedded_tonic_status(io_err).map(|status| status.code()), Some(tonic::Code::Unavailable));
+    }
+
+    #[test]
+    fn network_like_disk_error_ignores_transport_words_in_application_statuses() {
+        // Same contract as the peer client: a status the peer answered with
+        // is not a transport failure, so it must not drive a reconnect even
+        // when its message describes one.
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::internal(
+            "connection refused while dialing downstream backend"
+        ))));
+        assert!(!is_network_like_disk_error(&DiskError::from(tonic::Status::unauthenticated(
+            "connection reset while validating token"
+        ))));
+    }
+
+    #[test]
+    fn network_like_disk_error_requires_anchored_unavailable_needle() {
+        // Regression: a bare "unavailable" needle used to match application
+        // text such as a bucket name.
+        assert!(!is_network_like_disk_error(&DiskError::other("bucket \"unavailable-logs\" not found")));
+        // Anchored renderings of a flattened Unavailable status still match.
+        assert!(is_network_like_disk_error(&DiskError::other(
+            "code: 'The service is currently unavailable', message: \"peer gone\""
+        )));
+        assert!(is_network_like_disk_error(&DiskError::other(
+            "status: Unavailable, message: \"peer gone\""
+        )));
+        assert!(is_network_like_disk_error(&DiskError::other("connection refused")));
+        assert!(!is_network_like_disk_error(&DiskError::FileNotFound));
+    }
+
     #[test]
     fn test_signature_interceptor_keeps_auth_headers() {
         ensure_test_rpc_secret();

crates/ecstore/src/cluster/rpc/peer_rest_client.rs

@@ -13,8 +13,8 @@
 // limitations under the License.
 
 use crate::cluster::rpc::client::{
-    TonicInterceptor, gen_tonic_signature_interceptor, heal_control_time_out_client, node_service_time_out_client,
-    tier_mutation_control_time_out_client,
+    TonicInterceptor, embedded_tonic_status, gen_tonic_signature_interceptor, heal_control_time_out_client,
+    is_network_like_status, message_has_network_needle, node_service_time_out_client, tier_mutation_control_time_out_client,
 };
 use crate::cluster::rpc::{set_tonic_canonical_body_digest, verify_tonic_rpc_response_proof};
 use crate::error::{Error, Result};
@@ -471,26 +471,21 @@ impl PeerRestClient {
         self.offline.store(false, Ordering::Release);
     }
 
+    /// Whether this failure means the peer is unreachable, so it should be
+    /// gated offline and its connection evicted.
+    ///
+    /// RPC failures are classified by their typed gRPC code first
+    /// (`is_network_like_status`); an application error from a live peer must
+    /// never take it offline no matter what its message says. The substring
+    /// fallback only covers failures that exist purely as text, such as the
+    /// dial errors `get_client` wraps.
     fn is_network_like_error(err: &Error) -> bool {
-        let message = err.to_string().to_ascii_lowercase();
-        [
-            "temporarily offline",
-            "transport error",
-            "unavailable",
-            "error trying to connect",
-            "connection refused",
-            "connection reset",
-            "broken pipe",
-            "not connected",
-            "unexpected eof",
-            "timed out",
-            "deadline has elapsed",
-            "connection closed",
-            "connection aborted",
-            "tcp connect error",
-        ]
-        .iter()
-        .any(|needle| message.contains(needle))
+        if let Error::Io(io_err) = err
+            && let Some(status) = embedded_tonic_status(io_err)
+        {
+            return is_network_like_status(status);
+        }
+        message_has_network_needle(&err.to_string())
     }
 
     fn mark_offline_and_spawn_recovery(&self) {
@@ -1702,30 +1697,17 @@ fn tier_config_reload_connection_outcome(err: Error) -> TierConfigReloadOutcome
 }
 
 fn is_tier_config_reload_connection_failure(err: &Error) -> bool {
-    let message = err.to_string().to_ascii_lowercase();
+    let message = err.to_string();
+    // A bare "unavailable" is only trusted inside the local dial-failure
+    // wrapper from `get_client`, never in application text.
     if message
+        .to_ascii_lowercase()
         .split_once("can not get client, err:")
         .is_some_and(|(_, local_error)| local_error.contains("unavailable"))
     {
         return true;
     }
-    [
-        "temporarily offline",
-        "transport error",
-        "error trying to connect",
-        "connection refused",
-        "connection reset",
-        "connection closed",
-        "connection aborted",
-        "broken pipe",
-        "not connected",
-        "unexpected eof",
-        "timed out",
-        "deadline has elapsed",
-        "tcp connect error",
-    ]
-    .iter()
-    .any(|needle| message.contains(needle))
+    message_has_network_needle(&message)
 }
 
 fn tier_config_reload_remote_failure(error_info: Option<String>) -> TierConfigReloadOutcome {
@@ -2151,6 +2133,126 @@ mod tests {
         assert!(!PeerRestClient::is_network_like_error(&Error::NotImplemented));
     }
 
+    #[test]
+    fn peer_rest_client_network_classifier_uses_typed_status_code() {
+        // The one code that means "nothing is answering on this channel".
+        assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unavailable(
+            "storage layer is not initialized"
+        ))));
+        // Application statuses from a live peer must not mark it offline,
+        // even when their message contains transport-sounding words.
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::internal(
+            "failed to reload metadata for bucket \"unavailable-logs\""
+        ))));
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unauthenticated(
+            "No valid auth token"
+        ))));
+        // A request-budget expiry answered by a live peer is an application
+        // outcome, not a transport failure.
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::deadline_exceeded(
+            "heal control request expired"
+        ))));
+        // Unknown is the transport's escape hatch for a cause it could not
+        // map, and our handlers never return it, so there the text decides.
+        assert!(PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
+            "Service was not ready: transport error"
+        ))));
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(tonic::Status::unknown(
+            "peer response unknown"
+        ))));
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_ignores_transport_words_in_application_statuses() {
+        // The reason classification reads the code rather than the text: a
+        // peer that answers is reachable, even when what it says describes a
+        // connection failure of its own. A handler interpolating a local
+        // io::Error into Status::internal, or relaying trouble with its own
+        // downstream, must not cost us the channel to a healthy peer.
+        for status in [
+            tonic::Status::internal("connection refused while dialing downstream backend"),
+            tonic::Status::internal("write failed: broken pipe"),
+            tonic::Status::unauthenticated("connection reset while validating token"),
+            tonic::Status::failed_precondition("scanner lease timed out"),
+            tonic::Status::deadline_exceeded("heal control request timed out"),
+        ] {
+            let rendered = status.to_string();
+            assert!(
+                !PeerRestClient::is_network_like_error(&Error::from(status)),
+                "an answered application status must not mark the peer offline: {rendered}"
+            );
+        }
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_keeps_slow_peers_online() {
+        // The per-RPC channel deadline (RUSTFS_INTERNODE_RPC_TIMEOUT, 30s)
+        // surfaces as Cancelled "Timeout expired" carrying the transport
+        // cause as its source. A peer that is merely slow must stay online:
+        // gating it would spend a full recovery cycle fast-failing every RPC
+        // to a host that is still answering, turning load into a partition.
+        let timeout_status = tonic::Status::cancelled("Timeout expired");
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(timeout_status)));
+
+        let sourced = tonic::Status::from_error(Box::new(std::io::Error::other("Timeout expired")));
+        assert!(
+            std::error::Error::source(&sourced).is_some(),
+            "the transport builds this status through Status::from_error, which attaches the cause"
+        );
+        assert!(!PeerRestClient::is_network_like_error(&Error::from(sourced)));
+    }
+
+    #[test]
+    fn rpc_status_errors_keep_their_rendering_and_hide_peer_metadata() {
+        let err = Error::from(tonic::Status::unavailable("peer gone"));
+        assert_eq!(
+            err.to_string(),
+            "Io error: code: 'The service is currently unavailable', message: \"peer gone\""
+        );
+
+        // tonic's own Debug prints the MetadataMap, i.e. every response header
+        // the peer sent; those must not reach a log through this error.
+        let mut status = tonic::Status::unavailable("peer gone");
+        status
+            .metadata_mut()
+            .insert("authorization", "Bearer secret".parse().expect("valid header value"));
+        let rendered = format!("{:?}", Error::from(status));
+        assert!(!rendered.contains("Bearer secret"), "{rendered}");
+        assert!(!rendered.contains("MetadataMap"), "{rendered}");
+    }
+
+    #[test]
+    fn peer_rest_client_network_classifier_ignores_application_text_containing_unavailable() {
+        // Regression: a bare "unavailable" needle used to match application
+        // strings like these and take a healthy peer offline.
+        assert!(!PeerRestClient::is_network_like_error(&Error::other(
+            "peer replication statistics provider is unavailable"
+        )));
+        assert!(!PeerRestClient::is_network_like_error(&Error::other(
+            "bucket \"unavailable-logs\" not found"
+        )));
+        // Anchored renderings of a flattened Unavailable status still match:
+        // tonic >= 0.14 form ...
+        assert!(PeerRestClient::is_network_like_error(&Error::other(
+            "peer tier mutation commit RPC failed: code: 'The service is currently unavailable', message: \"peer gone\""
+        )));
+        // ... which is only anchored as long as tonic renders Unavailable this
+        // way. A tonic bump that reworded it leaves the typed path correct but
+        // this needle stale, so pin the coupling rather than discover it in a
+        // partition.
+        assert!(
+            tonic::Status::unavailable("peer gone")
+                .to_string()
+                .to_ascii_lowercase()
+                .contains("code: 'the service is currently unavailable'"),
+            "tonic reworded Code::Unavailable; update the anchored needle"
+        );
+        // ... and the tonic <= 0.13 form peers may relay in error_info.
+        assert!(PeerRestClient::is_network_like_error(&Error::other(
+            "peer tier mutation commit RPC failed: status: Unavailable, message: \"peer gone\""
+        )));
+    }
+
     #[test]
     fn tier_config_reload_outcome_keeps_tonic_and_remote_errors_typed() {
         assert!(matches!(
@@ -2193,6 +2295,14 @@ mod tests {
             tier_config_reload_connection_outcome(Error::other("can not get client, err: connection unavailable")),
             TierConfigReloadOutcome::TransientReconnect(_)
         ));
+        // The bare word is trusted only to the right of the dial-failure
+        // prefix, not anywhere in the message.
+        assert!(matches!(
+            tier_config_reload_connection_outcome(Error::other(
+                "bucket unavailable-logs rejected it, then: can not get client, err: some other reason"
+            )),
+            TierConfigReloadOutcome::Terminal(_)
+        ));
     }
 
     #[tokio::test]
@@ -2495,6 +2605,39 @@ mod tests {
         assert!(!client.offline.load(Ordering::Acquire));
     }
 
+    #[tokio::test]
+    async fn peer_rest_client_finalize_result_keeps_online_for_app_errors_mentioning_unavailable() {
+        // Regression: application error text containing "unavailable" (a
+        // remote error_info payload, or a bucket named "unavailable-logs" in
+        // a typed application status) must not take a healthy peer offline.
+        let client = test_peer_client();
+        let err = client
+            .finalize_result::<()>(Err(Error::other("peer replication statistics provider is unavailable")))
+            .await
+            .expect_err("application error should still be returned");
+        assert!(err.to_string().contains("provider is unavailable"));
+        assert!(!client.offline.load(Ordering::Acquire));
+
+        let err = client
+            .finalize_result::<()>(Err(Error::from(tonic::Status::internal(
+                "failed to reload metadata for bucket \"unavailable-logs\"",
+            ))))
+            .await
+            .expect_err("application status should still be returned");
+        assert!(err.to_string().contains("unavailable-logs"));
+        assert!(!client.offline.load(Ordering::Acquire));
+    }
+
+    #[tokio::test]
+    async fn peer_rest_client_finalize_result_marks_offline_for_typed_unavailable_status() {
+        let client = test_peer_client();
+        client
+            .finalize_result::<()>(Err(Error::from(tonic::Status::unavailable("storage layer is not initialized"))))
+            .await
+            .expect_err("network error should still be returned");
+        assert!(client.offline.load(Ordering::Acquire));
+    }
+
     #[tokio::test(flavor = "current_thread")]
     async fn peer_rest_recovery_probe_logs_keep_request_id_span_context() {
         let logs = CapturedLogs::default();

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/disk/disk_store.rs

@@ -13,9 +13,9 @@
 // limitations under the License.
 
 use crate::disk::{
-    CheckPartsResp, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint, Error,
-    FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result,
-    UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskError, DiskInfo, DiskInfoOptions, DiskLocation, Endpoint,
+    Error, FileInfoVersions, MmapCopyStageMetrics, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, Result,
+    SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
     health_state::{
         RuntimeDriveHealthState, classify_drive_recovery, get_drive_returning_probe_interval,
         get_drive_returning_success_threshold, get_drive_suspect_failure_threshold, record_drive_offline_duration,
@@ -1349,6 +1349,30 @@ impl DiskAPI for LocalDiskWrapper {
             .await
     }
 
+    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
+        self.track_disk_health(
+            || async { self.disk.acquire_snapshot_lease(volume, path).await },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
+        self.track_disk_health(
+            || async { self.disk.release_snapshot_lease(volume, path, token).await },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
+    async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
+        self.track_disk_health(
+            || async { self.disk.delete_data_dir(volume, path, opts).await },
+            get_max_timeout_duration(),
+        )
+        .await
+    }
+
     async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
         self.track_disk_health(
             || async { self.disk.write_metadata(org_volume, volume, path, fi).await },

crates/ecstore/src/disk/error.rs

@@ -337,9 +337,54 @@ impl From<DiskError> for std::io::Error {
     }
 }
 
+/// The single in-band representation of a failed internode RPC: it carries the
+/// typed `tonic::Status` so failure classifiers can read the gRPC code instead
+/// of substring-matching the rendered message (see `is_network_like_status`).
+/// Both `DiskError` and `StorageError` wrap statuses in this type, so one
+/// downcast recovers the status regardless of which error the status was
+/// converted into first.
+pub(crate) struct RpcStatusError(tonic::Status);
+
+impl RpcStatusError {
+    pub(crate) fn status(&self) -> &tonic::Status {
+        &self.0
+    }
+}
+
+impl From<tonic::Status> for RpcStatusError {
+    fn from(status: tonic::Status) -> Self {
+        Self(status)
+    }
+}
+
+impl std::fmt::Display for RpcStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        std::fmt::Display::fmt(&self.0, f)
+    }
+}
+
+/// `tonic::Status`'s own `Debug` prints its `MetadataMap`, i.e. every response
+/// header and trailer the peer sent. Those are remote-controlled and can carry
+/// credentials injected by a proxy in front of the peer, so keep them out of
+/// anything that reaches a log.
+impl std::fmt::Debug for RpcStatusError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("RpcStatusError")
+            .field("code", &self.0.code())
+            .field("message", &self.0.message())
+            .finish()
+    }
+}
+
+impl StdError for RpcStatusError {
+    fn source(&self) -> Option<&(dyn StdError + 'static)> {
+        Some(&self.0)
+    }
+}
+
 impl From<tonic::Status> for DiskError {
     fn from(e: tonic::Status) -> Self {
-        DiskError::other(e.message().to_string())
+        DiskError::Io(io::Error::other(RpcStatusError(e)))
     }
 }
 

crates/ecstore/src/disk/local.rs

@@ -18,11 +18,12 @@ use crate::data_usage::local_snapshot::ensure_data_usage_layout;
 use crate::disk::disk_store::{get_drive_walkdir_stall_timeout, get_object_disk_read_timeout};
 use crate::disk::{
     BUCKET_META_PREFIX, CHECK_PART_FILE_CORRUPT, CHECK_PART_FILE_NOT_FOUND, CHECK_PART_SUCCESS, CHECK_PART_UNKNOWN,
-    CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskMetrics,
-    FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize, PART_TRANSACTION_NEW_META,
-    PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET, RUSTFS_META_TMP_BUCKET,
-    RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, STORAGE_FORMAT_FILE,
-    STORAGE_FORMAT_FILE_BACKUP, UpdateMetadataOpts, VolumeInfo, WalkDirOptions, conv_part_err_to_int,
+    CHECK_PART_VOLUME_NOT_FOUND, CheckPartsResp, DataDirDeleteStatus, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions,
+    DiskLocation, DiskMetrics, FileInfoVersions, FileReader, FileWriter, MmapCopyStageMetrics, OldCurrentSize,
+    PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction, RUSTFS_META_BUCKET,
+    RUSTFS_META_TMP_BUCKET, RUSTFS_META_TMP_DELETED_BUCKET, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp,
+    STORAGE_FORMAT_FILE, STORAGE_FORMAT_FILE_BACKUP, SnapshotLeaseToken, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
+    conv_part_err_to_int,
     endpoint::Endpoint,
     error::{DiskError, Error, FileAccessDeniedWithContext, Result},
     error_conv::{to_access_error, to_file_error, to_unformatted_disk_error, to_volume_error},
@@ -66,7 +67,7 @@ use tokio::fs::{self, File};
 #[cfg(not(unix))]
 use tokio::io::AsyncReadExt;
 use tokio::io::{AsyncRead, AsyncSeekExt, AsyncWrite, AsyncWriteExt, ErrorKind, ReadBuf};
-use tokio::sync::{Notify, RwLock, Semaphore};
+use tokio::sync::{Mutex, Notify, RwLock, Semaphore};
 use tokio::time::{Instant, Sleep, interval_at, timeout};
 use tracing::{debug, error, info, warn};
 use uuid::Uuid;
@@ -3856,6 +3857,25 @@ pub struct LocalDisk {
     exit_signal: Option<tokio::sync::broadcast::Sender<()>>,
     io_backend: Arc<dyn LocalIoBackend>,
     file_sync_permits: Arc<Semaphore>,
+    snapshot_leases: Arc<Mutex<SnapshotLeaseRegistry>>,
+}
+
+#[derive(Clone, Debug, Eq, Hash, PartialEq)]
+struct SnapshotLeaseKey {
+    volume: String,
+    path: String,
+}
+
+#[derive(Default)]
+struct SnapshotLeaseEntry {
+    tokens: HashSet<SnapshotLeaseToken>,
+    pending_delete: Option<DeleteOptions>,
+    deleting: bool,
+}
+
+#[derive(Default)]
+struct SnapshotLeaseRegistry {
+    entries: HashMap<SnapshotLeaseKey, SnapshotLeaseEntry>,
 }
 
 impl Drop for LocalDisk {
@@ -4092,6 +4112,7 @@ impl LocalDisk {
             exit_signal: None,
             io_backend: build_local_io_backend(root.clone()),
             file_sync_permits: os::disk_file_sync_limiter(&root),
+            snapshot_leases: Arc::new(Mutex::new(SnapshotLeaseRegistry::default())),
         };
         let (info, _root) = get_disk_info(root.clone()).await.inspect_err(|err| {
             log_startup_disk_error("get_disk_info", &root, err);
@@ -4576,6 +4597,23 @@ impl LocalDisk {
         Ok(())
     }
 
+    async fn delete_unleased(&self, volume: &str, path: &str, opt: &DeleteOptions) -> Result<()> {
+        let volume_dir = self.get_bucket_path(volume)?;
+        if !skip_access_checks(volume)
+            && let Err(e) = access(&volume_dir).await
+        {
+            return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
+        }
+
+        let file_path = self.get_object_path(volume, path)?;
+        check_path_length(file_path.to_string_lossy().as_ref())?;
+        self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate)
+            .await?;
+        // A deleted shard must not remain readable through the io_uring fd cache.
+        self.io_backend.invalidate_cached_fds_under(volume, path);
+        Ok(())
+    }
+
     #[tracing::instrument(level = "trace", skip_all)]
     #[async_recursion::async_recursion]
     async fn delete_file(
@@ -6216,27 +6254,7 @@ impl DiskAPI for LocalDisk {
     #[tracing::instrument(level = "trace", skip_all)]
     async fn delete(&self, volume: &str, path: &str, opt: DeleteOptions) -> Result<()> {
         crate::hp_guard!("LocalDisk::delete");
-        let volume_dir = self.get_bucket_path(volume)?;
-        if !skip_access_checks(volume)
-            && let Err(e) = access(&volume_dir).await
-        {
-            return Err(to_access_error(e, DiskError::VolumeAccessDenied).into());
-        }
-
-        let file_path = self.get_object_path(volume, path)?;
-
-        check_path_length(file_path.to_string_lossy().to_string().as_str())?;
-
-        self.delete_file(&volume_dir, &file_path, opt.recursive, opt.immediate)
-            .await?;
-
-        // The inode is unlinked, but a cached descriptor would keep it readable —
-        // a deleted shard must not keep answering reads (backlog#1145). The part
-        // numbers under `path` are not known here, so this is the one caller that
-        // needs the predicate form.
-        self.io_backend.invalidate_cached_fds_under(volume, path);
-
-        Ok(())
+        self.delete_unleased(volume, path, &opt).await
     }
 
     #[tracing::instrument(level = "trace", skip_all)]
@@ -7824,6 +7842,118 @@ impl DiskAPI for LocalDisk {
         Ok(())
     }
 
+    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
+        let file_path = self.get_object_path(volume, path)?;
+        let key = SnapshotLeaseKey {
+            volume: volume.to_string(),
+            path: path.to_string(),
+        };
+        let token = {
+            let mut registry = self.snapshot_leases.lock().await;
+            if registry.entries.get(&key).is_some_and(|entry| entry.deleting) {
+                return Err(DiskError::FileNotFound);
+            }
+            let token = SnapshotLeaseToken::new();
+            registry.entries.entry(key).or_default().tokens.insert(token);
+            token
+        };
+        match fs::metadata(file_path).await {
+            Ok(metadata) if metadata.is_dir() => Ok(token),
+            Ok(_) => {
+                self.release_snapshot_lease(volume, path, token).await?;
+                Err(DiskError::FileNotFound)
+            }
+            Err(err) => {
+                self.release_snapshot_lease(volume, path, token).await?;
+                Err(to_file_error(err).into())
+            }
+        }
+    }
+
+    async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
+        let key = SnapshotLeaseKey {
+            volume: volume.to_string(),
+            path: path.to_string(),
+        };
+        let opts = {
+            let mut registry = self.snapshot_leases.lock().await;
+            let Some(entry) = registry.entries.get_mut(&key) else {
+                return Ok(());
+            };
+            entry.tokens.remove(&token);
+            if !entry.tokens.is_empty() || entry.deleting {
+                return Ok(());
+            }
+
+            let Some(opts) = entry.pending_delete.clone() else {
+                registry.entries.remove(&key);
+                return Ok(());
+            };
+            entry.deleting = true;
+            opts
+        };
+        let result = self.delete_unleased(volume, path, &opts).await;
+        let mut registry = self.snapshot_leases.lock().await;
+        match result {
+            Ok(()) => {
+                registry.entries.remove(&key);
+                Ok(())
+            }
+            Err(err) => {
+                if let Some(entry) = registry.entries.get_mut(&key) {
+                    entry.deleting = false;
+                }
+                Err(err)
+            }
+        }
+    }
+
+    async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
+        let key = SnapshotLeaseKey {
+            volume: volume.to_string(),
+            path: path.to_string(),
+        };
+        {
+            let mut registry = self.snapshot_leases.lock().await;
+            if let Some(entry) = registry.entries.get_mut(&key) {
+                if !entry.tokens.is_empty() {
+                    entry.pending_delete.get_or_insert_with(|| opts.clone());
+                    return Ok(DataDirDeleteStatus::Deferred);
+                }
+                if entry.deleting {
+                    entry.pending_delete.get_or_insert_with(|| opts.clone());
+                    return Ok(DataDirDeleteStatus::Deferred);
+                }
+                entry.deleting = true;
+                entry.pending_delete.get_or_insert_with(|| opts.clone());
+            } else {
+                registry.entries.insert(
+                    key.clone(),
+                    SnapshotLeaseEntry {
+                        pending_delete: Some(opts.clone()),
+                        deleting: true,
+                        ..Default::default()
+                    },
+                );
+            }
+        }
+
+        let result = self.delete_unleased(volume, path, &opts).await;
+        let mut registry = self.snapshot_leases.lock().await;
+        match result {
+            Ok(()) => {
+                registry.entries.remove(&key);
+                Ok(DataDirDeleteStatus::Deleted)
+            }
+            Err(err) => {
+                if let Some(entry) = registry.entries.get_mut(&key) {
+                    entry.deleting = false;
+                }
+                Err(err)
+            }
+        }
+    }
+
     #[tracing::instrument(level = "trace", skip_all)]
     async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()> {
         if !fi.metadata.is_empty() {
@@ -14800,6 +14930,162 @@ mod test {
         assert!(construction_source.is::<ErasureConstructionError>());
     }
 
+    #[tokio::test]
+    async fn snapshot_leases_defer_data_dir_cleanup_until_last_release() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-lease-volume";
+        let data_dir = path_join_buf(&["object", &Uuid::new_v4().to_string()]);
+        let first_part = path_join_buf(&[&data_dir, "part.1"]);
+        let later_part = path_join_buf(&[&data_dir, "part.2"]);
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &first_part, Bytes::from_static(b"first"))
+            .await
+            .expect("first shard should be written");
+        disk.write_all(volume, &later_part, Bytes::from_static(b"later"))
+            .await
+            .expect("later shard should be written");
+
+        let first = disk
+            .acquire_snapshot_lease(volume, &data_dir)
+            .await
+            .expect("first lease should be acquired");
+        let second = disk
+            .acquire_snapshot_lease(volume, &data_dir)
+            .await
+            .expect("second lease should be acquired");
+        let status = disk
+            .delete_data_dir(
+                volume,
+                &data_dir,
+                DeleteOptions {
+                    recursive: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("cleanup should be deferred");
+        assert_eq!(status, DataDirDeleteStatus::Deferred);
+        assert_eq!(
+            disk.read_all(volume, &later_part)
+                .await
+                .expect("a later multipart shard must remain openable while leased"),
+            Bytes::from_static(b"later")
+        );
+
+        disk.release_snapshot_lease(volume, &data_dir, first)
+            .await
+            .expect("first lease release should succeed");
+        assert!(
+            disk.read_all(volume, &first_part).await.is_ok(),
+            "one remaining lease must keep the data directory"
+        );
+        disk.release_snapshot_lease(volume, &data_dir, second)
+            .await
+            .expect("last lease release should run deferred cleanup");
+        disk.release_snapshot_lease(volume, &data_dir, second)
+            .await
+            .expect("releasing an already released token should be idempotent");
+        assert!(matches!(disk.read_all(volume, &first_part).await, Err(DiskError::FileNotFound)));
+    }
+
+    #[tokio::test]
+    async fn data_dir_cleanup_without_a_lease_keeps_existing_behavior() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
+        let volume = "snapshot-no-lease-volume";
+        let data_dir = path_join_buf(&["object", &Uuid::new_v4().to_string()]);
+        let part = path_join_buf(&[&data_dir, "part.1"]);
+        ensure_test_volume(&disk, volume).await;
+        disk.write_all(volume, &part, Bytes::from_static(b"payload"))
+            .await
+            .expect("test shard should be written");
+
+        let status = disk
+            .delete_data_dir(
+                volume,
+                &data_dir,
+                DeleteOptions {
+                    recursive: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("unleased cleanup should retain the existing delete behavior");
+        assert_eq!(status, DataDirDeleteStatus::Deleted);
+        assert!(matches!(disk.read_all(volume, &part).await, Err(DiskError::FileNotFound)));
+    }
+
+    #[tokio::test]
+    async fn snapshot_lease_acquire_and_cleanup_are_atomic() {
+        use tempfile::tempdir;
+
+        let root_dir = tempdir().expect("temp dir should be created");
+        let endpoint = Endpoint::try_from(root_dir.path().to_string_lossy().as_ref()).expect("endpoint should parse");
+        let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+        let volume = "snapshot-race-volume";
+        ensure_test_volume(&disk, volume).await;
+
+        for iteration in 0..32 {
+            let data_dir = path_join_buf(&["object", &format!("{iteration:032x}")]);
+            let part = path_join_buf(&[&data_dir, "part.1"]);
+            disk.write_all(volume, &part, Bytes::from_static(b"payload"))
+                .await
+                .expect("test shard should be written");
+            let barrier = Arc::new(tokio::sync::Barrier::new(3));
+            let acquire_disk = Arc::clone(&disk);
+            let acquire_barrier = Arc::clone(&barrier);
+            let acquire_path = data_dir.clone();
+            let acquire = tokio::spawn(async move {
+                acquire_barrier.wait().await;
+                acquire_disk.acquire_snapshot_lease(volume, &acquire_path).await
+            });
+            let delete_disk = Arc::clone(&disk);
+            let delete_barrier = Arc::clone(&barrier);
+            let delete_path = data_dir.clone();
+            let delete = tokio::spawn(async move {
+                delete_barrier.wait().await;
+                delete_disk
+                    .delete_data_dir(
+                        volume,
+                        &delete_path,
+                        DeleteOptions {
+                            recursive: true,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+            });
+            barrier.wait().await;
+
+            let acquired = acquire.await.expect("acquire task should join");
+            let deleted = delete
+                .await
+                .expect("delete task should join")
+                .expect("delete should either run or defer");
+            match acquired {
+                Ok(token) => {
+                    assert_eq!(deleted, DataDirDeleteStatus::Deferred);
+                    assert!(disk.read_all(volume, &part).await.is_ok());
+                    disk.release_snapshot_lease(volume, &data_dir, token)
+                        .await
+                        .expect("release should finish deferred cleanup");
+                }
+                Err(DiskError::FileNotFound) => {
+                    assert_eq!(deleted, DataDirDeleteStatus::Deleted);
+                }
+                Err(err) => panic!("unexpected lease acquisition error: {err}"),
+            }
+            assert!(matches!(disk.read_all(volume, &part).await, Err(DiskError::FileNotFound)));
+        }
+    }
+
     #[tokio::test]
     async fn local_disk_check_parts_rejects_zero_data_geometry_before_shard_math() {
         use tempfile::tempdir;

crates/ecstore/src/disk/mod.rs

@@ -72,6 +72,27 @@ pub type DiskStore = Arc<Disk>;
 pub type FileReader = Box<dyn AsyncRead + Send + Sync + Unpin>;
 pub type FileWriter = Box<dyn AsyncWrite + Send + Sync + Unpin>;
 
+#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq, Serialize, Deserialize)]
+pub struct SnapshotLeaseToken(Uuid);
+
+impl SnapshotLeaseToken {
+    pub fn new() -> Self {
+        Self(Uuid::new_v4())
+    }
+}
+
+impl Default for SnapshotLeaseToken {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+pub enum DataDirDeleteStatus {
+    Deleted,
+    Deferred,
+}
+
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 pub enum PartTransactionAction {
     Commit,
@@ -249,6 +270,27 @@ impl DiskAPI for Disk {
         }
     }
 
+    async fn acquire_snapshot_lease(&self, volume: &str, path: &str) -> Result<SnapshotLeaseToken> {
+        match self {
+            Disk::Local(local_disk) => local_disk.acquire_snapshot_lease(volume, path).await,
+            Disk::Remote(remote_disk) => remote_disk.acquire_snapshot_lease(volume, path).await,
+        }
+    }
+
+    async fn release_snapshot_lease(&self, volume: &str, path: &str, token: SnapshotLeaseToken) -> Result<()> {
+        match self {
+            Disk::Local(local_disk) => local_disk.release_snapshot_lease(volume, path, token).await,
+            Disk::Remote(remote_disk) => remote_disk.release_snapshot_lease(volume, path, token).await,
+        }
+    }
+
+    async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
+        match self {
+            Disk::Local(local_disk) => local_disk.delete_data_dir(volume, path, opts).await,
+            Disk::Remote(remote_disk) => remote_disk.delete_data_dir(volume, path, opts).await,
+        }
+    }
+
     #[tracing::instrument(level = "trace", skip_all)]
     async fn write_metadata(&self, _org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()> {
         match self {
@@ -646,6 +688,16 @@ pub trait DiskAPI: Debug + Send + Sync + 'static {
     ) -> Result<()>;
     async fn delete_versions(&self, volume: &str, versions: Vec<FileInfoVersions>, opts: DeleteOptions) -> Vec<Option<Error>>;
     async fn delete_paths(&self, volume: &str, paths: &[String]) -> Result<()>;
+    async fn acquire_snapshot_lease(&self, _volume: &str, _path: &str) -> Result<SnapshotLeaseToken> {
+        Err(Error::other("snapshot leases are not supported by this disk"))
+    }
+    async fn release_snapshot_lease(&self, _volume: &str, _path: &str, _token: SnapshotLeaseToken) -> Result<()> {
+        Err(Error::other("snapshot leases are not supported by this disk"))
+    }
+    async fn delete_data_dir(&self, volume: &str, path: &str, opts: DeleteOptions) -> Result<DataDirDeleteStatus> {
+        self.delete(volume, path, opts).await?;
+        Ok(DataDirDeleteStatus::Deleted)
+    }
     async fn write_metadata(&self, org_volume: &str, volume: &str, path: &str, fi: FileInfo) -> Result<()>;
     async fn update_metadata(&self, volume: &str, path: &str, fi: FileInfo, opts: &UpdateMetadataOpts) -> Result<()>;
     async fn read_version(

crates/ecstore/src/error/mod.rs

@@ -818,7 +818,11 @@ impl From<s3s::xml::DeError> for Error {
 
 impl From<tonic::Status> for Error {
     fn from(e: tonic::Status) -> Self {
-        Error::other(e.to_string())
+        // Keep the typed status as the io::Error payload instead of a
+        // flattened string so RPC failure classifiers can read the gRPC code
+        // via downcast (see `PeerRestClient::is_network_like_error`).
+        // `RpcStatusError` renders exactly as `e.to_string()` did.
+        Error::other(crate::disk::error::RpcStatusError::from(e))
     }
 }
 

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(),
@@ -464,11 +466,11 @@ impl ObjectInfo {
 
         let transitioned_object = TransitionedObject {
             name: fi.transitioned_objname.clone(),
-            version_id: if let Some(transition_version_id) = fi.transition_version_id {
-                transition_version_id.to_string()
-            } else {
-                "".to_string()
-            },
+            version_id: fi
+                .transition_version
+                .clone()
+                .or_else(|| fi.transition_version_id.map(|version_id| version_id.to_string()))
+                .unwrap_or_default(),
             status: fi.transition_status.clone(),
             free_version: fi.tier_free_version(),
             tier: fi.transition_tier.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/test_util.rs

@@ -931,7 +931,10 @@ pub async fn read_transition_meta(disk_path: &Path, bucket: &str, object: &str)
         status: fi.transition_status.clone(),
         tier: fi.transition_tier.clone(),
         remote_object: fi.transitioned_objname.clone(),
-        remote_version_id: fi.transition_version_id.map(|id| id.to_string()),
+        remote_version_id: fi
+            .transition_version
+            .clone()
+            .or_else(|| fi.transition_version_id.map(|id| id.to_string())),
         free_version_count,
     })
 }

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

@@ -47,8 +47,8 @@ use crate::diagnostics::get::{
     record_get_object_pipeline_failure_for_path, record_get_stage_duration_if_enabled,
 };
 use crate::disk::{
-    OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK, PartTransactionAction,
-    part_transaction_path,
+    DataDirDeleteStatus, OldCurrentSize, PART_TRANSACTION_NEW_META, PART_TRANSACTION_OLD_META, PART_TRANSACTION_ROLLBACK,
+    PartTransactionAction, part_transaction_path,
 };
 use crate::erasure::coding::BitrotReader;
 use crate::io_support::bitrot::ShardReader;
@@ -547,6 +547,8 @@ pub(in crate::set_disk) fn metadata_early_stop_candidate_matches(left: &FileInfo
         && left.transitioned_objname == right.transitioned_objname
         && 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
@@ -2985,29 +2987,48 @@ impl SetDisks {
                 Self::rename_fanout_barrier(&object_for_fault, idx, rename_fanout_barrier_phase::CLEANUP).await;
 
                 if let Some(err) = Self::cleanup_injected_error(&object_for_fault, idx) {
-                    return Some(err);
+                    return (false, Some(err));
                 }
                 if let Some(disk) = disk {
-                    disk.delete(
-                        &bucket,
-                        &file_path,
-                        DeleteOptions {
-                            recursive: true,
-                            ..Default::default()
-                        },
-                    )
-                    .await
-                    .err()
+                    match disk
+                        .delete_data_dir(
+                            &bucket,
+                            &file_path,
+                            DeleteOptions {
+                                recursive: true,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                    {
+                        Ok(DataDirDeleteStatus::Deleted) => (false, None),
+                        Ok(DataDirDeleteStatus::Deferred) => (true, None),
+                        Err(err) => (false, Some(err)),
+                    }
                 } else {
                     // `None` slot: ignored placeholder. It is not `attempted`, so
                     // classification excludes it from residue regardless.
-                    Some(DiskError::DiskNotFound)
+                    (false, Some(DiskError::DiskNotFound))
                 }
             })
         });
-        let errs: Vec<Option<DiskError>> = join_all(futures).await.into_iter().map(map_cleanup_join_result).collect();
+        let mut deferred = 0usize;
+        let errs: Vec<Option<DiskError>> = join_all(futures)
+            .await
+            .into_iter()
+            .map(|result| match result {
+                Ok((was_deferred, err)) => {
+                    deferred += usize::from(was_deferred);
+                    err
+                }
+                Err(join_err) => Some(DiskError::other(format!("old data dir cleanup task failed: {join_err}"))),
+            })
+            .collect();
 
-        classify_old_data_dir_cleanup(&errs, &attempted, write_quorum)
+        let mut cleanup = classify_old_data_dir_cleanup(&errs, &attempted, write_quorum);
+        cleanup.deferred = deferred;
+        cleanup.reclaimed = cleanup.reclaimed.saturating_sub(deferred);
+        cleanup
     }
 
     /// Test-only fault-injection seam for the old-data-dir cleanup path
@@ -3098,6 +3119,20 @@ impl SetDisks {
 
         rustfs_io_metrics::record_old_data_dir_cleanup(c.attempted, c.reclaimed, c.unreclaimed_disks.len(), c.below_quorum);
 
+        if c.deferred > 0 {
+            debug!(
+                event = EVENT_SET_DISK_WRITE,
+                component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_SET_DISK,
+                bucket = %bucket,
+                object = %object,
+                old_data_dir = %old_dir,
+                deferred = c.deferred,
+                state = "old_data_cleanup_deferred",
+                "Old data directory cleanup deferred for active snapshot leases"
+            );
+        }
+
         if actions.warn {
             warn!(
                 component = LOG_COMPONENT_ECSTORE,
@@ -4129,6 +4164,9 @@ pub(in crate::set_disk) struct OldDataDirCleanup {
     /// Number of attempted disks that returned `Ok` or a not-found variant
     /// (a missing dir == already reclaimed).
     pub reclaimed: usize,
+    /// Number of attempted disks that retained the directory for an active
+    /// snapshot lease and registered it for deletion after the final release.
+    pub deferred: usize,
     /// Indices of attempted disks that failed with a non-ignored, non-not-found
     /// error (including task panic/cancel). This is the residue that actually
     /// leaks and drives the leak metric + heal enqueue.
@@ -4191,6 +4229,7 @@ fn classify_old_data_dir_cleanup(errs: &[Option<DiskError>], attempted: &[bool],
     OldDataDirCleanup {
         attempted: attempted_count,
         reclaimed,
+        deferred: 0,
         unreclaimed_disks,
         below_quorum,
     }
@@ -5131,6 +5170,40 @@ mod tests {
         drop((disk1, disk2));
     }
 
+    #[tokio::test]
+    async fn commit_cleanup_reports_and_releases_deferred_snapshot_data_dirs() {
+        let bucket = "cleanup-lease-bucket";
+        let object = "cleanup-lease-object";
+        let old_data_dir = "11111111-1111-1111-1111-111111111111";
+        let committed_data_dir = "22222222-2222-2222-2222-222222222222";
+        let data_dir_path = format!("{object}/{old_data_dir}");
+        let shard_path = format!("{data_dir_path}/part.1");
+        let (_dir1, disk1) = read_multiple_test_disk(bucket, &[(&shard_path, b"one".as_slice())]).await;
+        let set = io_primitives_test_set(vec![Some(disk1.clone())], 0).await;
+        let lease = disk1
+            .acquire_snapshot_lease(bucket, &data_dir_path)
+            .await
+            .expect("snapshot lease should be acquired before cleanup");
+
+        let cleanup = set
+            .commit_rename_data_dir(&[Some(disk1.clone())], bucket, object, old_data_dir, committed_data_dir, 1)
+            .await;
+        assert_eq!(cleanup.attempted, 1);
+        assert_eq!(cleanup.reclaimed, 0);
+        assert_eq!(cleanup.deferred, 1);
+        assert!(cleanup.unreclaimed_disks.is_empty());
+        disk1
+            .read_all(bucket, &shard_path)
+            .await
+            .expect("deferred cleanup must leave later shard opens available");
+
+        disk1
+            .release_snapshot_lease(bucket, &data_dir_path, lease)
+            .await
+            .expect("final lease release should reclaim the old data directory");
+        assert!(matches!(disk1.read_all(bucket, &shard_path).await, Err(DiskError::FileNotFound)));
+    }
+
     /// Isolation guard: an armed barrier / observed object only affects its own
     /// object. A fan-out for a different (unobserved, unarmed) object must not be
     /// paused and must not accrue any tracked task count — so concurrent tests

crates/ecstore/src/set_disk/metadata.rs

@@ -578,6 +578,13 @@ impl SetDisks {
         Self::update_hash_str(hasher, &meta.transition_tier);
         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/mod.rs

@@ -692,6 +692,8 @@ pub(crate) use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 #[cfg(feature = "test-util")]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
+#[cfg(test)]
+pub(crate) use ops::object::cleanup_rejected_transition_upload_durably;
 mod read;
 mod replication;
 pub(crate) mod shard_source;

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

@@ -25,7 +25,10 @@ use crate::set_disk::read::GetObjectDownstreamWriter;
 
 use crate::bucket::lifecycle::{
     tier_delete_journal::{persist_tier_delete_journal_entry, remove_tier_delete_journal_entry},
-    tier_sweeper::{Jentry, RemoteTierDeleteOutcome, delete_object_from_remote_tier_with_lease_idempotent},
+    tier_sweeper::{
+        Jentry, RemoteTierDeleteOutcome, delete_confirmed_transition_candidate_exact_with_lease_idempotent,
+        delete_object_from_remote_tier_with_lease_idempotent,
+    },
     transition_transaction::{
         TransitionRemoteVersion, TransitionSourceIdentity, TransitionSourceVersionMode, TransitionTransaction,
         TransitionTransactionInit, TransitionTransactionState, delete_transition_transaction_record,
@@ -1663,7 +1666,11 @@ pub(crate) async fn cleanup_uncommitted_transition_upload(
     cleanup_version: &str,
     version_id_exact: bool,
 ) -> std::io::Result<RemoteTierDeleteOutcome> {
-    delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, version_id_exact).await
+    if version_id_exact {
+        delete_confirmed_transition_candidate_exact_with_lease_idempotent(object, cleanup_version, lease).await
+    } else {
+        delete_object_from_remote_tier_with_lease_idempotent(object, cleanup_version, lease, false).await
+    }
 }
 
 fn log_transition_upload_cleanup_failure(lease: &TierOperationLease, object: &str, cleanup_version: &str, err: &std::io::Error) {
@@ -1800,7 +1807,7 @@ impl Drop for TransitionUploadCleanup {
     }
 }
 
-async fn cleanup_rejected_transition_upload_durably(
+pub(crate) async fn cleanup_rejected_transition_upload_durably(
     lease: &TierOperationLease,
     object: &str,
     cleanup_version: &str,
@@ -1813,6 +1820,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() {
@@ -1972,12 +1986,83 @@ async fn advance_and_save_transition_transaction(
     next: TransitionTransactionState,
     remote_version: Option<TransitionRemoteVersion>,
 ) -> Result<()> {
+    #[cfg(test)]
+    record_transition_uploaded_save_attempt(transaction, next);
     transaction
         .advance(transaction.fence(), next, remote_version)
         .map_err(Error::other)?;
     save_transition_transaction_if_available(api, transaction).await
 }
 
+#[cfg(test)]
+struct TransitionUploadedSaveProbeState {
+    bucket: String,
+    object: String,
+    attempts: std::sync::atomic::AtomicUsize,
+}
+
+#[cfg(test)]
+struct TransitionUploadedSaveProbe {
+    state: Arc<TransitionUploadedSaveProbeState>,
+}
+
+#[cfg(test)]
+static TRANSITION_UPLOADED_SAVE_PROBE: std::sync::OnceLock<std::sync::Mutex<Option<Arc<TransitionUploadedSaveProbeState>>>> =
+    std::sync::OnceLock::new();
+
+#[cfg(test)]
+impl TransitionUploadedSaveProbe {
+    fn install(bucket: &str, object: &str) -> Self {
+        let state = Arc::new(TransitionUploadedSaveProbeState {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            attempts: std::sync::atomic::AtomicUsize::new(0),
+        });
+        let mut slot = TRANSITION_UPLOADED_SAVE_PROBE
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition uploaded-save probe mutex should not poison");
+        assert!(slot.is_none(), "transition uploaded-save probe must be installed by one test at a time");
+        *slot = Some(Arc::clone(&state));
+        drop(slot);
+        Self { state }
+    }
+
+    fn attempts(&self) -> usize {
+        self.state.attempts.load(std::sync::atomic::Ordering::Acquire)
+    }
+}
+
+#[cfg(test)]
+impl Drop for TransitionUploadedSaveProbe {
+    fn drop(&mut self) {
+        let mut slot = TRANSITION_UPLOADED_SAVE_PROBE
+            .get_or_init(|| std::sync::Mutex::new(None))
+            .lock()
+            .expect("transition uploaded-save probe mutex should not poison");
+        if slot.as_ref().is_some_and(|state| Arc::ptr_eq(state, &self.state)) {
+            *slot = None;
+        }
+    }
+}
+
+#[cfg(test)]
+fn record_transition_uploaded_save_attempt(transaction: &TransitionTransaction, next: TransitionTransactionState) {
+    if next != TransitionTransactionState::Uploaded {
+        return;
+    }
+    let state = TRANSITION_UPLOADED_SAVE_PROBE
+        .get_or_init(|| std::sync::Mutex::new(None))
+        .lock()
+        .expect("transition uploaded-save probe mutex should not poison")
+        .as_ref()
+        .filter(|state| state.bucket == transaction.source.bucket && state.object == transaction.source.object)
+        .cloned();
+    if let Some(state) = state {
+        state.attempts.fetch_add(1, std::sync::atomic::Ordering::AcqRel);
+    }
+}
+
 async fn delete_transition_transaction_if_available(api: Option<&Arc<ECStore>>, transaction_id: Uuid) -> Result<()> {
     if let Some(api) = api {
         return delete_transition_transaction_record(api.clone(), transaction_id).await;
@@ -2228,11 +2313,25 @@ async fn pause_transition_commit(bucket: &str, object: &str, pause: TransitionCo
     }
 }
 
-fn parse_transition_version_id(remote_version: &str) -> std::result::Result<Option<Uuid>, uuid::Error> {
+fn persisted_transition_version(
+    remote_version: &str,
+) -> std::io::Result<(Option<String>, rustfs_filemeta::TransitionVersionState)> {
     if remote_version.is_empty() {
-        return Ok(None);
+        return Ok((None, rustfs_filemeta::TransitionVersionState::KnownDisabled));
     }
-    Uuid::parse_str(remote_version).map(|version_id| (!version_id.is_nil()).then_some(version_id))
+    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)]
@@ -2470,16 +2569,17 @@ 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("").expect("empty remote version should be valid"), None);
         assert_eq!(
-            parse_transition_version_id(&Uuid::nil().to_string()).expect("nil remote version should be valid"),
-            None
+            persisted_transition_version("").expect("empty remote version identifies an unversioned tier"),
+            (None, TransitionVersionState::KnownDisabled)
         );
+        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);
@@ -2491,12 +2591,14 @@ mod transition_version_id_tests {
     }
 
     #[test]
-    fn preserves_valid_remote_id_and_rejects_invalid_text() {
+    fn preserves_uuid_and_gates_opaque_remote_ids() {
         let version_id = Uuid::new_v4();
         assert_eq!(
-            parse_transition_version_id(&version_id.to_string()).expect("UUID remote version should be valid"),
-            Some(version_id)
+            persisted_transition_version(&version_id.to_string()).expect("UUID remote version"),
+            (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()
@@ -2505,7 +2607,6 @@ mod transition_version_id_tests {
             TransitionUploadCandidate::from_put_response("opaque-version-token".to_string()).cleanup_version(),
             "opaque-version-token"
         );
-        assert!(parse_transition_version_id("not-a-uuid").is_err());
     }
 }
 
@@ -3775,6 +3876,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.into());
         }
+        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());
+            }
+        };
         if let Err(err) = advance_and_save_transition_transaction(
             transaction_api.as_ref(),
             &mut transaction,
@@ -3792,16 +3907,6 @@ 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 = match parse_transition_version_id(candidate.remote_version()) {
-            Ok(version_id) => version_id,
-            Err(err) => {
-                if upload_cleanup.cleanup().await.is_ok() {
-                    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;
@@ -3859,7 +3964,11 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
         current_fi.transition_status = TRANSITION_COMPLETE.to_string();
         current_fi.transitioned_objname = dest_obj;
         current_fi.transition_tier = opts.transition.tier.clone();
-        current_fi.transition_version_id = transition_version_id;
+        current_fi.transition_version_id = transition_version_id
+            .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,
@@ -4076,6 +4185,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
             let mut p_reader = PutObjReader::new(hash_reader);
             return match self_.clone().put_object(bucket, object, &mut p_reader, &ropts).await {
                 Ok(restored_info) => {
+                    let restored_info = self_.finalize_restore_metadata(bucket, object, &restored_info, &opts).await?;
                     send_event(EventArgs {
                         event_name: EventName::ObjectRestoreCompleted.as_str().to_string(),
                         bucket_name: bucket.to_string(),
@@ -4210,6 +4320,7 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks {
                 return set_restore_header_fn(&mut oi, Some(err)).await;
             }
         };
+        let restored_info = self_.finalize_restore_metadata(bucket, object, &restored_info, opts).await?;
         send_event(EventArgs {
             event_name: EventName::ObjectRestoreCompleted.as_str().to_string(),
             bucket_name: bucket.to_string(),
@@ -4668,6 +4779,33 @@ mod transition_commit_failure_tests {
     }
 
     #[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())]
+            );
+        }
+    }
+
     #[serial_test::serial(restore_multipart_failure_point)]
     async fn multipart_restore_aborts_every_post_create_failure() {
         let (temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
@@ -6615,6 +6753,45 @@ mod transition_upload_integrity_tests {
         );
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn unversioned_remote_version_is_persisted_without_version_id() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-unversioned-tier-bucket";
+        let object = "object.bin";
+        let payload = b"unversioned remote tier must commit without a version id".repeat(1024);
+        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        backend.set_put_remote_version(Some(String::new())).await;
+        let save_probe = TransitionUploadedSaveProbe::install(bucket, object);
+
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect("an unversioned remote version must commit");
+        let (fi, _, _) = set_disks
+            .get_object_fileinfo(
+                bucket,
+                object,
+                &ObjectOptions {
+                    no_lock: true,
+                    metadata_cache_safe: false,
+                    ..Default::default()
+                },
+                true,
+                false,
+            )
+            .await
+            .expect("committed unversioned transition metadata should be readable");
+        assert_eq!(fi.transition_version_id, None);
+        assert_eq!(fi.transition_version, None);
+        assert_eq!(fi.transition_version_state, rustfs_filemeta::TransitionVersionState::KnownDisabled);
+        assert_eq!(save_probe.attempts(), 1);
+        assert_eq!(backend.remove_count().await, 0);
+        assert_eq!(backend.object_count().await, 1);
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn opaque_remote_version_is_cleaned_before_parse_failure() {
@@ -6630,7 +6807,7 @@ mod transition_upload_integrity_tests {
         set_disks
             .transition_object(bucket, object, &transition_options(&original, tier_name))
             .await
-            .expect_err("an unparseable remote version must fail closed");
+            .expect_err("an opaque remote version must fail closed until the capability gate is active");
         let removed_versions = backend.remove_versions().await;
         assert_eq!(removed_versions.len(), 1);
         assert_eq!(removed_versions[0].1, "opaque-version-token");
@@ -6638,6 +6815,38 @@ mod transition_upload_integrity_tests {
         assert_local_source_intact(&set_disks, bucket, object, &payload).await;
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn nil_remote_version_is_cleaned_exactly_before_transaction_persistence() {
+        let (_temp_dirs, disk_stores, set_disks) = hermetic_set_disks(4).await;
+        let bucket = "transition-nil-version-bucket";
+        let object = "object.bin";
+        let payload = b"nil remote version must retain local data".repeat(1024);
+        let original = write_source(&set_disks, &disk_stores, bucket, object, &payload).await;
+        let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
+        let remote_version = Uuid::nil().to_string();
+        let backend = register_mock_tier(&runtime_sources::global_tier_config_mgr(), &tier_name).await;
+        backend.set_put_remote_version(Some(remote_version.clone())).await;
+        let save_probe = TransitionUploadedSaveProbe::install(bucket, object);
+
+        set_disks
+            .transition_object(bucket, object, &transition_options(&original, tier_name))
+            .await
+            .expect_err("a nil remote version must fail closed before transaction persistence");
+        let put_versions = backend.put_versions().await;
+        let removed_versions = backend.remove_versions().await;
+        assert_eq!(removed_versions, put_versions);
+        assert_eq!(removed_versions.len(), 1);
+        assert_eq!(
+            removed_versions.first().map(|(_, version)| version.as_str()),
+            Some(remote_version.as_str())
+        );
+        assert_eq!(save_probe.attempts(), 0, "nil remote version must be rejected before saving Uploaded");
+        assert_eq!(backend.exact_remove_count(), 1);
+        assert_eq!(backend.object_count().await, 0);
+        assert_local_source_intact(&set_disks, bucket, object, &payload).await;
+    }
+
     #[tokio::test]
     #[serial_test::serial]
     async fn authoritative_read_failure_after_upload_cleans_exact_candidate_and_preserves_source() {
@@ -6983,23 +7192,36 @@ mod transition_source_identity_matrix_tests {
             let object = format!("identity-{index}.bin");
             let payload = vec![u8::try_from(index + 1).expect("matrix index should fit u8"); 1024 * 1024];
             let mut reader = PutObjReader::from_vec(payload);
+            let source_version_id = Uuid::new_v4();
+            let source_opts = ObjectOptions {
+                version_id: Some(source_version_id.to_string()),
+                versioned: true,
+                ..Default::default()
+            };
             let original = set_disks
-                .put_object(bucket, &object, &mut reader, &ObjectOptions::default())
+                .put_object(bucket, &object, &mut reader, &source_opts)
                 .await
                 .expect("source object should be written");
             let (source, _, _) = set_disks
-                .get_object_fileinfo(bucket, &object, &ObjectOptions::default(), true, false)
+                .get_object_fileinfo(bucket, &object, &source_opts, true, false)
                 .await
                 .expect("source metadata should resolve");
+            assert_eq!(source.version_id, Some(source_version_id));
+            assert_eq!(
+                transition_source_identity(bucket, &object, &source, &source_opts, &get_raw_etag(&source.metadata))
+                    .expect("persisted versioned source identity should build")
+                    .version_mode,
+                TransitionSourceVersionMode::Versioned
+            );
             let opts = ObjectOptions {
                 no_lock: true,
+                versioned: true,
                 transition: TransitionOptions {
                     status: TRANSITION_PENDING.to_string(),
                     tier: tier_name.clone(),
                     etag: original.etag.clone().unwrap_or_default(),
                     ..Default::default()
                 },
-                version_id: original.version_id.map(|version| version.to_string()),
                 mod_time: original.mod_time,
                 ..Default::default()
             };
@@ -7029,12 +7251,19 @@ mod transition_source_identity_matrix_tests {
                     .await
                     .expect("single-field metadata drift should be written");
             }
+            let persisted_opts = ObjectOptions {
+                version_id: changed.version_id.map(|version_id| version_id.to_string()),
+                versioned: true,
+                ..Default::default()
+            };
+            let (persisted, _, _) = set_disks
+                .get_object_fileinfo(bucket, &object, &persisted_opts, true, false)
+                .await
+                .expect("drifted source metadata should resolve");
             put_barrier.release();
 
-            transition
-                .await
-                .expect("transition task should not panic")
-                .expect_err("transition must reject a source whose identity changed after upload");
+            let result = transition.await.expect("transition task should not panic");
+            assert!(result.is_err(), "transition must reject {field:?} drift");
             let expected_attempts = index + 1;
             assert_eq!(backend.put_count().await, expected_attempts);
             assert_eq!(backend.remove_count().await, expected_attempts);
@@ -7045,23 +7274,23 @@ mod transition_source_identity_matrix_tests {
             );
 
             match field {
-                IdentityField::DataDir => assert_ne!(source.data_dir, changed.data_dir),
-                IdentityField::ModTime => assert_ne!(source.mod_time, changed.mod_time),
-                IdentityField::Size => assert_ne!(source.size, changed.size),
-                IdentityField::Etag => assert_ne!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata)),
+                IdentityField::DataDir => assert_ne!(source.data_dir, persisted.data_dir),
+                IdentityField::ModTime => assert_ne!(source.mod_time, persisted.mod_time),
+                IdentityField::Size => assert_ne!(source.size, persisted.size),
+                IdentityField::Etag => assert_ne!(get_raw_etag(&source.metadata), get_raw_etag(&persisted.metadata)),
             }
-            assert_eq!(source.version_id, changed.version_id);
+            assert_eq!(source.version_id, persisted.version_id);
             if !matches!(field, IdentityField::DataDir) {
-                assert_eq!(source.data_dir, changed.data_dir);
+                assert_eq!(source.data_dir, persisted.data_dir);
             }
             if !matches!(field, IdentityField::ModTime) {
-                assert_eq!(source.mod_time, changed.mod_time);
+                assert_eq!(source.mod_time, persisted.mod_time);
             }
             if !matches!(field, IdentityField::Size) {
-                assert_eq!(source.size, changed.size);
+                assert_eq!(source.size, persisted.size);
             }
             if !matches!(field, IdentityField::Etag) {
-                assert_eq!(get_raw_etag(&source.metadata), get_raw_etag(&changed.metadata));
+                assert_eq!(get_raw_etag(&source.metadata), get_raw_etag(&persisted.metadata));
             }
         }
     }

crates/ecstore/src/set_disk/replication.rs

@@ -13,7 +13,10 @@
 // limitations under the License.
 
 use super::*;
+use crate::bucket::lifecycle::lifecycle;
+use rustfs_filemeta::RestoreStatusOps;
 use rustfs_utils::http::headers::{AMZ_RESTORE_EXPIRY_DAYS, AMZ_RESTORE_REQUEST_DATE};
+use s3s::dto::{RestoreStatus, Timestamp};
 
 #[derive(Clone, Copy, Debug, Eq, PartialEq)]
 struct RestoreCleanupIdentity {
@@ -43,6 +46,69 @@ impl RestoreCleanupIdentity {
 }
 
 impl SetDisks {
+    pub(super) async fn finalize_restore_metadata(
+        &self,
+        bucket: &str,
+        object: &str,
+        obj_info: &ObjectInfo,
+        opts: &ObjectOptions,
+    ) -> Result<ObjectInfo> {
+        let expected = RestoreCleanupIdentity::from_object_info(obj_info);
+        let expected_operation_id = restore_operation_id_from_metadata(&opts.user_defined)?;
+        let expected_etag = obj_info
+            .etag
+            .clone()
+            .unwrap_or_else(|| get_raw_etag(obj_info.user_defined.as_ref()));
+        let version_id = expected.version_id.map(|v| v.to_string());
+        let _lock_guard = if !opts.no_lock {
+            Some(
+                self.acquire_write_lock_diag("restore_finalize_metadata", bucket, object)
+                    .await?,
+            )
+        } else {
+            None
+        };
+        let read_opts = ObjectOptions {
+            version_id,
+            versioned: opts.versioned,
+            version_suspended: opts.version_suspended,
+            ..Default::default()
+        };
+        let (mut fi, _, disks) = self
+            .get_object_fileinfo_gated(bucket, object, &read_opts, false, false)
+            .await?;
+        if let Some(expected_operation_id) = expected_operation_id {
+            require_restore_operation_id(&fi.metadata, expected_operation_id)?;
+        }
+        if !expected.matches_file_info(&fi, &expected_etag) {
+            return Err(Error::other("restored object changed before restore metadata finalization"));
+        }
+        let restore_expiry =
+            lifecycle::expected_expiry_time(OffsetDateTime::now_utc(), opts.transition.restore_request.days.unwrap_or(1));
+        fi.metadata.insert(
+            X_AMZ_RESTORE.as_str().to_string(),
+            RestoreStatus {
+                is_restore_in_progress: Some(false),
+                restore_expiry_date: Some(Timestamp::from(restore_expiry)),
+            }
+            .to_string(),
+        );
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        self.update_object_meta_with_opts(
+            bucket,
+            object,
+            fi.clone(),
+            disks.as_slice(),
+            &UpdateMetadataOpts {
+                replace_user_metadata: true,
+                ..Default::default()
+            },
+        )
+        .await?;
+        self.invalidate_get_object_metadata_cache(bucket, object).await;
+        Ok(ObjectInfo::from_file_info(&fi, bucket, object, opts.versioned || opts.version_suspended))
+    }
+
     pub async fn update_restore_metadata(
         &self,
         bucket: &str,

crates/ecstore/src/set_disk/transition_matrix_tests.rs

@@ -13,10 +13,11 @@
 // limitations under the License.
 
 use super::*;
-use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions};
+use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time};
 use crate::ecstore_validation_blackbox::make_local_set_disks;
 use crate::services::tier::test_util::register_mock_tier;
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
+use rustfs_filemeta::{RestoreStatusOps as _, parse_restore_obj_status};
 use tokio::io::AsyncReadExt;
 
 async fn prime_metadata_generation(set_disks: &SetDisks, bucket: &str, object: &str) -> GetObjectMetadataCacheKey {
@@ -83,13 +84,57 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() {
     let transitioned_generation = prime_metadata_generation(&set_disks, bucket, object).await;
     let mut restore_opts = ObjectOptions::default();
     restore_opts.transition.restore_request.days = Some(1);
-    Arc::clone(&set_disks)
-        .restore_transitioned_object(bucket, object, &restore_opts)
-        .await
-        .expect("restore should succeed");
+    let restore_started = OffsetDateTime::now_utc();
+    let expiry_from_restore_start = temp_env::async_with_vars(
+        [
+            ("RUSTFS_ILM_DEBUG_DAY_SECS", Some("1")),
+            ("RUSTFS_ILM_PROCESS_TIME", Some("1")),
+        ],
+        async {
+            let expiry_from_restore_start = expected_expiry_time(restore_started, 1);
+            let get_barrier = backend.arm_get_barrier().await;
+            let restore_set = Arc::clone(&set_disks);
+            let restore =
+                tokio::spawn(async move { restore_set.restore_transitioned_object(bucket, object, &restore_opts).await });
+            get_barrier.wait_until_paused().await;
+            tokio::time::timeout(Duration::from_secs(5), async {
+                loop {
+                    if expected_expiry_time(OffsetDateTime::now_utc(), 1) > expiry_from_restore_start {
+                        break;
+                    }
+                    tokio::time::sleep(Duration::from_millis(10)).await;
+                }
+            })
+            .await
+            .expect("test clock should cross the next accelerated lifecycle boundary");
+            get_barrier.release();
+            restore
+                .await
+                .expect("restore task should join")
+                .expect("restore should succeed");
+            expiry_from_restore_start
+        },
+    )
+    .await;
     assert_generation_reclaimed(&set_disks, &transitioned_generation).await;
     assert_eq!(backend.get_count().await, 1, "restore should read the remote candidate exactly once");
 
+    let restored_info = set_disks
+        .get_object_info(bucket, object, &ObjectOptions::default())
+        .await
+        .expect("restored object metadata should be readable");
+    let restore_status = parse_restore_obj_status(
+        restored_info
+            .user_defined
+            .get(s3s::header::X_AMZ_RESTORE.as_str())
+            .expect("completed restore header should be present"),
+    )
+    .expect("completed restore header should parse");
+    assert!(
+        restore_status.expiry().expect("completed restore should have an expiry") > expiry_from_restore_start,
+        "restore expiry must be based on completion, not the time the remote copy started"
+    );
+
     let mut restored = Vec::new();
     set_disks
         .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())

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)
@@ -2720,10 +2722,12 @@ mod tests {
     #[serial_test::serial(storage_class_env)]
     async fn transition_transaction_recovery_deletes_provider_recovered_unknown_upload() {
         let versioned_remote = uuid::Uuid::new_v4().to_string();
+        let nil_remote = uuid::Uuid::nil().to_string();
         for (case, tier_name, remote_version) in [
             ("missing", "TXPROBEMISSING", None),
             ("unversioned", "TXPROBEUNVERSIONED", Some(String::new())),
             ("versioned", "TXPROBEVERSIONED", Some(versioned_remote)),
+            ("nil-version", "TXPROBENILVERSION", Some(nil_remote)),
         ] {
             let temp_dir = tempfile::tempdir().expect("create temp store dir");
             let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store(