fix(tier): gate opaque version recovery cleanup (#8006)

* fix(tier): gate opaque version recovery cleanup

* test(tier): authorize opaque recovery fixtures

---------

Co-authored-by: Chris <anzhengchao@gmail.com>

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

@@ -35,6 +35,7 @@ use crate::bucket::lifecycle::tier_sweeper::{
 use crate::disk::RUSTFS_META_BUCKET;
 use crate::error::{Error, Result as EcstoreResult};
 use crate::object_api::{ObjectInfo, ObjectOptions};
+use crate::services::notification_sys::acquire_remote_version_state_writer_fleet_proof;
 use crate::services::tier::{tier::TierConfigMgr, warm_backend::TransitionCandidateProbe};
 use crate::storage_api_contracts::{
     list::ListOperations as _,
@@ -1237,6 +1238,12 @@ pub async fn delete_transition_candidate_for_operator(
     lease
         .validate_remote_version_id(remote_version_id)
         .map_err(TransitionOperatorError::Remote)?;
+    if remote_version_requires_fleet_proof(remote_version_id) && acquire_remote_version_state_writer_fleet_proof().is_none() {
+        return Err(TransitionOperatorError::Remote(std::io::Error::new(
+            std::io::ErrorKind::Unsupported,
+            "opaque remote tier version cleanup requires the operator-attested live fleet capability gate",
+        )));
+    }
     let before_delete_probe = lease
         .probe_transition_candidate_for(&transaction.remote_object, transaction.transaction_id)
         .await
@@ -1845,6 +1852,11 @@ async fn recover_unknown_upload_outcome(
             ))
         }
         TransitionCandidateProbe::VersionedPresent(version_id) => {
+            if remote_version_requires_fleet_proof(&version_id) && acquire_remote_version_state_writer_fleet_proof().is_none() {
+                return Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
+                    IlmRecoveryErrorCode::RemoteVersionUnknown,
+                ));
+            }
             cleanup_recovered_unknown_upload_candidate(api, transaction, TransitionRemoteVersion::versioned(version_id)).await
         }
         TransitionCandidateProbe::Ambiguous => Ok(TransitionTransactionRecoveryOutcome::RetainedAmbiguous(
@@ -1940,6 +1952,14 @@ fn transition_source_lookup_options(transaction: &TransitionTransaction) -> Obje
 async fn delete_transition_remote_candidate(api: Arc<ECStore>, transaction: &TransitionTransaction) -> EcstoreResult<()> {
     let version_id = transaction.remote_version.tier_delete_version_id().unwrap_or_default();
     let version_id_exact = transaction.remote_version.kind == TransitionRemoteVersionKind::Versioned;
+    if version_id_exact
+        && remote_version_requires_fleet_proof(version_id)
+        && acquire_remote_version_state_writer_fleet_proof().is_none()
+    {
+        return Err(Error::other(
+            "opaque remote tier version cleanup requires the operator-attested live fleet capability gate",
+        ));
+    }
     delete_object_from_remote_tier_idempotent_with_manager_and_identity(
         &transaction.remote_object,
         version_id,
@@ -1953,6 +1973,10 @@ async fn delete_transition_remote_candidate(api: Arc<ECStore>, transaction: &Tra
     .map_err(Error::other)
 }
 
+fn remote_version_requires_fleet_proof(version_id: &str) -> bool {
+    !version_id.is_empty() && Uuid::parse_str(version_id).is_err()
+}
+
 pub async fn recover_transition_transaction_records(
     api: Arc<ECStore>,
     limit: usize,
@@ -2439,6 +2463,14 @@ mod tests {
         }
     }
 
+    #[test]
+    fn opaque_remote_versions_require_the_fleet_proof_boundary() {
+        assert!(!remote_version_requires_fleet_proof(""));
+        assert!(!remote_version_requires_fleet_proof(&Uuid::new_v4().to_string()));
+        assert!(remote_version_requires_fleet_proof("null"));
+        assert!(remote_version_requires_fleet_proof("b2-opaque-version"));
+    }
+
     #[test]
     fn remote_version_distinguishes_unknown_unversioned_and_versioned() {
         assert_eq!(TransitionRemoteVersion::known_from_put_response("").tier_delete_version_id(), None);

crates/ecstore/src/services/notification_sys.rs

@@ -453,6 +453,20 @@ pub(crate) fn acquire_remote_version_state_fleet_proof() -> Option<RemoteVersion
     acquire_fleet_capability_proof_from(&state, expected_topology, Instant::now()).map(RemoteVersionStateFleetProofToken)
 }
 
+pub(crate) fn acquire_remote_version_state_writer_fleet_proof() -> Option<RemoteVersionStateFleetProofToken> {
+    let requested = rustfs_utils::get_env_bool(
+        rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE,
+        rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_WRITE,
+    );
+    let fleet_confirmed = rustfs_utils::get_env_bool(
+        rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+        rustfs_config::DEFAULT_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED,
+    );
+    (requested && fleet_confirmed)
+        .then(acquire_remote_version_state_fleet_proof)
+        .flatten()
+}
+
 fn acquire_fleet_capability_proof_from(
     state: &FleetCapabilityProofState,
     expected_topology: &str,

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

@@ -6715,7 +6715,7 @@ fn remote_version_state_writer_enabled() -> bool {
 }
 
 fn remote_version_state_writer_fleet_proof() -> Option<RemoteVersionStateFleetProofToken> {
-    transaction_fencing_fleet_proof(remote_version_state_writer_requested())
+    crate::services::notification_sys::acquire_remote_version_state_writer_fleet_proof()
 }
 
 fn remote_version_state_writer_requested() -> bool {

crates/ecstore/src/store/init.rs

@@ -10052,9 +10052,18 @@ mod tests {
                 recovered_transition_version.clone(),
             )))
             .await;
-        let transaction_stats = recover_transition_transaction_records(store.clone(), 100, None)
+        let transaction_stats = {
+            let _proof = crate::services::notification_sys::install_current_remote_version_state_fleet_proof_for_test();
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE, Some("true")),
+                    (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED, Some("true")),
+                ],
+                recover_transition_transaction_records(store.clone(), 100, None),
+            )
             .await
-            .expect("transition recovery should advance and consume the migrated transaction before completion");
+            .expect("transition recovery should advance and consume the migrated transaction before completion")
+        };
         backend.set_transition_candidate_probe_override(None).await;
         assert_eq!(
             (
@@ -21938,9 +21947,18 @@ mod tests {
             expected_removes.push((transaction.remote_object, put_version));
         }
 
-        let stats = recover_transition_transaction_records(store.clone(), 100, None)
+        let stats = {
+            let _proof = crate::services::notification_sys::install_current_remote_version_state_fleet_proof_for_test();
+            temp_env::async_with_vars(
+                [
+                    (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE, Some("true")),
+                    (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED, Some("true")),
+                ],
+                recover_transition_transaction_records(store.clone(), 100, None),
+            )
             .await
-            .expect("transition transaction recovery should run");
+            .expect("transition transaction recovery should run")
+        };
 
         assert_eq!((stats.scanned, stats.recovered, stats.retained, stats.failed), (3, 3, 0, 0));
         assert_eq!(transition_transaction_record_count(store.clone()).await, 0);

docs/operations/tier-ilm-debugging.md

@@ -65,6 +65,8 @@ get_bytes(&self.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID)
 
 `transition_version_id == None` means only that no usable legacy UUID projection exists; it does not prove the remote bucket's versioning model. Only an explicit `KnownDisabled` state authorizes ordinary GET/DELETE to omit `versionId`. A missing state with an absent or empty version key remains `Unknown` and requires the bounded compatibility probe or the approved reconcile workflow; it never directly authorizes cleanup. A nil UUID (`00000000-...`) sent as `?versionId=` causes `NoSuchVersion`. Do not use `Uuid::from_slice(..).unwrap_or_default()` here: it converts an empty metadata value into `Uuid::nil()`, which is exactly that failure.
 
+Opaque remote version IDs, such as those returned by Backblaze B2, are treated as exact remote identifiers only after the operator-attested live fleet capability gate is active. The gate requires both `RUSTFS_TIER_REMOTE_VERSION_STATE_WRITE=true` and `RUSTFS_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED=true`, plus a current fleet proof. Recovery and operator cleanup retain an ambiguous candidate while that proof is absent or stale; they never issue a guessed opaque-version delete. Existing persisted exact opaque IDs remain routable, but new persistence and recovery cleanup fail closed until the gate is current.
+
 ## Inspect xl.meta directly
 
 ```bash