chore: sync heal lock fix with main

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

@@ -584,33 +584,173 @@ impl ExpiryOp for FreeVersionTask {
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum TransitionDeleteVersionPlan {
+    Direct { version_id_exact: bool },
+    ProbeLegacyUnknown,
+}
+
+fn legacy_transition_version_state_missing(oi: &ObjectInfo) -> Result<bool, std::io::Error> {
+    use rustfs_utils::http::metadata_compat::{
+        SUFFIX_TRANSITIONED_VERSION_ID, SUFFIX_TRANSITIONED_VERSION_STATE, contains_key_str, get_consistent_str,
+    };
+
+    if !contains_key_str(&oi.user_defined, SUFFIX_TRANSITIONED_VERSION_STATE) {
+        let version_key_present = contains_key_str(&oi.user_defined, SUFFIX_TRANSITIONED_VERSION_ID);
+        if version_key_present {
+            if oi.transitioned_object.version_id.is_empty() {
+                let has_non_empty_version = oi.user_defined.iter().any(|(key, value)| {
+                    rustfs_utils::http::metadata_compat::strip_internal_prefix_preserving_case(key)
+                        .is_some_and(|suffix| suffix.eq_ignore_ascii_case(SUFFIX_TRANSITIONED_VERSION_ID))
+                        && !value.is_empty()
+                });
+                if !has_non_empty_version {
+                    // MinIO writes the transitioned-versionID key with an empty value
+                    // for unversioned tier objects. The backend probe remains the proof.
+                    return Ok(true);
+                }
+            } else if get_consistent_str(&oi.user_defined, SUFFIX_TRANSITIONED_VERSION_ID)
+                == Some(oi.transitioned_object.version_id.as_str())
+            {
+                return Ok(true);
+            }
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "legacy remote tier version metadata is conflicting or malformed",
+            ));
+        }
+        if !oi.transitioned_object.version_id.is_empty() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "legacy remote tier version metadata is missing or inconsistent",
+            ));
+        }
+        return Ok(true);
+    }
+    let persisted = get_consistent_str(&oi.user_defined, SUFFIX_TRANSITIONED_VERSION_STATE).ok_or_else(|| {
+        std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier object has conflicting transition version state metadata",
+        )
+    })?;
+    if persisted != oi.transition_version_state.as_str() {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier object transition version state metadata changed during decoding",
+        ));
+    }
+    Ok(false)
+}
+
+fn transition_remote_version_delete_plan(oi: &ObjectInfo) -> Result<TransitionDeleteVersionPlan, std::io::Error> {
+    match oi.transition_version_state {
+        rustfs_filemeta::TransitionVersionState::Unknown => {
+            if legacy_transition_version_state_missing(oi)? {
+                Ok(TransitionDeleteVersionPlan::ProbeLegacyUnknown)
+            } else {
+                validate_transition_remote_version(oi)
+                    .map(|version_id_exact| TransitionDeleteVersionPlan::Direct { version_id_exact })
+            }
+        }
+        _ => validate_transition_remote_version(oi)
+            .map(|version_id_exact| TransitionDeleteVersionPlan::Direct { version_id_exact }),
+    }
+}
+
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+struct ResolvedTransitionDeleteVersion {
+    version_id_exact: bool,
+    remote_already_missing: bool,
+}
+
 async fn acquire_free_version_tier_lease(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
-) -> Result<(TierOperationLease, bool), std::io::Error> {
-    let version_id_exact = validate_transition_remote_version(oi)?;
+) -> Result<(TierOperationLease, TransitionDeleteVersionPlan), std::io::Error> {
+    let delete_plan = transition_remote_version_delete_plan(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"))?;
     let lease =
         TierConfigMgr::acquire_operation_lease_for_backend_identity(tier_config_mgr, &oi.transitioned_object.tier, identity)
             .await
             .map_err(std::io::Error::other)?;
-    Ok((lease, version_id_exact))
+    Ok((lease, delete_plan))
+}
+
+async fn resolve_transition_delete_version_plan(
+    oi: &ObjectInfo,
+    lease: &TierOperationLease,
+    delete_plan: TransitionDeleteVersionPlan,
+) -> Result<ResolvedTransitionDeleteVersion, std::io::Error> {
+    match delete_plan {
+        TransitionDeleteVersionPlan::Direct { version_id_exact } => Ok(ResolvedTransitionDeleteVersion {
+            version_id_exact,
+            remote_already_missing: false,
+        }),
+        TransitionDeleteVersionPlan::ProbeLegacyUnknown => {
+            let expected_version = oi.transitioned_object.version_id.as_str();
+            if expected_version.is_empty() {
+                return Err(std::io::Error::new(
+                    std::io::ErrorKind::WouldBlock,
+                    "remote tier cannot safely delete a legacy object without an exact version ID",
+                ));
+            }
+            let probe = lease
+                .probe_transition_version(&oi.transitioned_object.name, expected_version)
+                .await?;
+            match (expected_version, probe) {
+                (expected, crate::services::tier::warm_backend::TransitionCandidateProbe::VersionedPresent(actual))
+                    if expected == actual =>
+                {
+                    lease.validate_remote_version_id(expected)?;
+                    Ok(ResolvedTransitionDeleteVersion {
+                        version_id_exact: true,
+                        remote_already_missing: false,
+                    })
+                }
+                (_, crate::services::tier::warm_backend::TransitionCandidateProbe::Missing) => {
+                    Ok(ResolvedTransitionDeleteVersion {
+                        version_id_exact: false,
+                        remote_already_missing: true,
+                    })
+                }
+                (_, crate::services::tier::warm_backend::TransitionCandidateProbe::Unsupported) => Err(std::io::Error::new(
+                    std::io::ErrorKind::Unsupported,
+                    "remote tier cannot prove legacy transition delete state",
+                )),
+                _ => Err(std::io::Error::new(
+                    std::io::ErrorKind::WouldBlock,
+                    "remote tier object version state is unknown",
+                )),
+            }
+        }
+    }
+}
+
+async fn execute_resolved_transition_delete(
+    oi: &ObjectInfo,
+    lease: &TierOperationLease,
+    resolved: ResolvedTransitionDeleteVersion,
+) -> Result<(), std::io::Error> {
+    if !resolved.remote_already_missing {
+        delete_object_from_remote_tier_with_lease_idempotent(
+            &oi.transitioned_object.name,
+            &oi.transitioned_object.version_id,
+            lease,
+            resolved.version_id_exact,
+        )
+        .await?;
+    }
+    Ok(())
 }
 
 async fn delete_free_version_remote_object_with_lease(
     oi: &ObjectInfo,
     lease: &TierOperationLease,
-    version_id_exact: bool,
+    delete_plan: TransitionDeleteVersionPlan,
 ) -> Result<(), std::io::Error> {
-    delete_object_from_remote_tier_with_lease_idempotent(
-        &oi.transitioned_object.name,
-        &oi.transitioned_object.version_id,
-        lease,
-        version_id_exact,
-    )
-    .await?;
-    Ok(())
+    let resolved = resolve_transition_delete_version_plan(oi, lease, delete_plan).await?;
+    execute_resolved_transition_delete(oi, lease, resolved).await
 }
 
 fn free_version_physical_topology_generation(api: &ECStore) -> String {
@@ -641,6 +781,16 @@ fn free_version_remote_tuple_matches(candidate: &ObjectInfo, expected: &ObjectIn
     if candidate.transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
         || expected.transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
     {
+        let candidate_legacy_missing = legacy_transition_version_state_missing(candidate)?;
+        let expected_legacy_missing = legacy_transition_version_state_missing(expected)?;
+        if candidate.transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
+            && expected.transition_version_state == rustfs_filemeta::TransitionVersionState::Unknown
+            && candidate_legacy_missing
+            && expected_legacy_missing
+            && candidate.transitioned_object.version_id == expected.transitioned_object.version_id
+        {
+            return Ok(true);
+        }
         return Err(std::io::Error::new(
             std::io::ErrorKind::WouldBlock,
             "tier free-version remote version state is unknown",
@@ -716,7 +866,7 @@ async fn cleanup_free_version_exact(api: Arc<ECStore>, oi: &ObjectInfo, cancel:
         .acquire_bucket_lifecycle_read_lock(&oi.bucket)
         .await
         .map_err(std::io::Error::other)?;
-    let (lease, version_id_exact) = acquire_free_version_tier_lease(oi, &api.tier_config_mgr()).await?;
+    let (lease, delete_plan) = acquire_free_version_tier_lease(oi, &api.tier_config_mgr()).await?;
     let local_object = encode_dir_object(&oi.name);
     let object_guards = api
         .acquire_all_physical_object_write_locks("tier_free_version_cleanup", &oi.bucket, &local_object)
@@ -734,16 +884,30 @@ async fn cleanup_free_version_exact(api: Arc<ECStore>, oi: &ObjectInfo, cancel:
             "tier free-version cleanup fence is invalid before remote delete",
         ));
     }
+    let resolved = tokio::select! {
+        _ = cancel.cancelled() => {
+            return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "tier free-version cleanup was cancelled"));
+        }
+        result = tokio::time::timeout_at(deadline, resolve_transition_delete_version_plan(oi, &lease, delete_plan)) => {
+            result.map_err(|_| {
+                std::io::Error::new(std::io::ErrorKind::TimedOut, "tier free-version remote probe timed out")
+            })??
+        }
+    };
+    if !free_version_cleanup_fences_current(&topology_generation, &api, &bucket_guard, &object_guards, &lease, cancel, deadline) {
+        return Err(std::io::Error::new(
+            std::io::ErrorKind::WouldBlock,
+            "tier free-version cleanup fence changed after remote probe",
+        ));
+    }
     tokio::select! {
         _ = cancel.cancelled() => {
             return Err(std::io::Error::new(std::io::ErrorKind::Interrupted, "tier free-version cleanup was cancelled"));
         }
-        result = tokio::time::timeout_at(
-            deadline,
-            delete_free_version_remote_object_with_lease(oi, &lease, version_id_exact),
-        ) => {
-            result
-                .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "tier free-version remote delete timed out"))??;
+        result = tokio::time::timeout_at(deadline, execute_resolved_transition_delete(oi, &lease, resolved)) => {
+            result.map_err(|_| {
+                std::io::Error::new(std::io::ErrorKind::TimedOut, "tier free-version remote delete timed out")
+            })??;
         }
     }
     if !free_version_cleanup_fences_current(&topology_generation, &api, &bucket_guard, &object_guards, &lease, cancel, deadline) {
@@ -791,8 +955,8 @@ async fn delete_free_version_remote_object(
     oi: &ObjectInfo,
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
 ) -> Result<(), std::io::Error> {
-    let (lease, version_id_exact) = acquire_free_version_tier_lease(oi, tier_config_mgr).await?;
-    delete_free_version_remote_object_with_lease(oi, &lease, version_id_exact).await
+    let (lease, delete_plan) = acquire_free_version_tier_lease(oi, tier_config_mgr).await?;
+    delete_free_version_remote_object_with_lease(oi, &lease, delete_plan).await
 }
 
 #[allow(
@@ -808,8 +972,8 @@ where
     F: FnOnce() -> Fut,
     Fut: std::future::Future<Output = T>,
 {
-    let (lease, version_id_exact) = acquire_free_version_tier_lease(oi, tier_config_mgr).await?;
-    delete_free_version_remote_object_with_lease(oi, &lease, version_id_exact).await?;
+    let (lease, delete_plan) = acquire_free_version_tier_lease(oi, tier_config_mgr).await?;
+    delete_free_version_remote_object_with_lease(oi, &lease, delete_plan).await?;
     let result = delete_local().await;
     drop(lease);
     Ok(result)
@@ -4688,6 +4852,39 @@ fn validate_transition_remote_version(oi: &ObjectInfo) -> Result<bool, std::io::
     }
 }
 
+#[derive(Clone, Copy, Debug, Eq, PartialEq)]
+enum TransitionReadVersionPlan {
+    Direct,
+    ProbeLegacyUnversioned,
+}
+
+const LEGACY_TRANSITION_READ_PROBE_TIMEOUT: StdDuration = StdDuration::from_secs(30);
+
+fn transition_remote_version_read_plan(oi: &ObjectInfo) -> Result<TransitionReadVersionPlan, std::io::Error> {
+    let version = oi.transitioned_object.version_id.as_str();
+    match oi.transition_version_state {
+        rustfs_filemeta::TransitionVersionState::Unknown => {
+            if !legacy_transition_version_state_missing(oi)? {
+                return validate_transition_remote_version(oi).map(|_| TransitionReadVersionPlan::Direct);
+            }
+            if version.is_empty() {
+                Ok(TransitionReadVersionPlan::ProbeLegacyUnversioned)
+            } else {
+                Ok(TransitionReadVersionPlan::Direct)
+            }
+        }
+        rustfs_filemeta::TransitionVersionState::KnownDisabled if version.is_empty() => Ok(TransitionReadVersionPlan::Direct),
+        rustfs_filemeta::TransitionVersionState::SuspendedNull if version == "null" => Ok(TransitionReadVersionPlan::Direct),
+        rustfs_filemeta::TransitionVersionState::Exact if !version.is_empty() && version != "null" => {
+            Ok(TransitionReadVersionPlan::Direct)
+        }
+        _ => Err(std::io::Error::new(
+            std::io::ErrorKind::InvalidData,
+            "remote tier object version state conflicts with its version ID",
+        )),
+    }
+}
+
 // The resolver joins the tier manager as the second injected port this read
 // needs; grouping the request half into a struct would churn every call site of
 // a bug fix.
@@ -4702,7 +4899,12 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     tier_config_mgr: &Arc<RwLock<TierConfigMgr>>,
     resolver: Option<&dyn ObjectEncryptionResolver>,
 ) -> Result<GetObjectReader, std::io::Error> {
-    validate_transition_remote_version(oi)?;
+    let read_plan = transition_remote_version_read_plan(oi)?;
+    // Reject invalid ranges and encryption requests before a compatibility
+    // probe can amplify them into remote listing work.
+    let plan = ReadPlan::build_for_request(rs.clone(), oi, opts, h, resolver)
+        .await
+        .map_err(|err| std::io::Error::other(format!("building the read plan for {bucket}/{object} failed: {err}")))?;
     let expected_identity = tier_destination_id_from_metadata(&oi.user_defined)?;
     let lease = match expected_identity {
         Some(identity) => {
@@ -4716,7 +4918,36 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
         Err(err) => return Err(std::io::Error::other(err)),
     };
 
-    tgt_client.validate_remote_version_id(&oi.transitioned_object.version_id)?;
+    match read_plan {
+        TransitionReadVersionPlan::Direct => {
+            tgt_client.validate_remote_version_id(&oi.transitioned_object.version_id)?;
+        }
+        TransitionReadVersionPlan::ProbeLegacyUnversioned => {
+            // RUSTFS_COMPAT_TODO(backlog#2203): remove operation-time probing
+            // after an admin reconcile can persist every proven legacy state.
+            let probe = tokio::time::timeout(
+                LEGACY_TRANSITION_READ_PROBE_TIMEOUT,
+                tgt_client.probe_transition_candidate(&oi.transitioned_object.name),
+            )
+            .await
+            .map_err(|_| std::io::Error::new(std::io::ErrorKind::TimedOut, "legacy remote tier version probe timed out"))??;
+            match probe {
+                crate::services::tier::warm_backend::TransitionCandidateProbe::UnversionedPresent => {}
+                crate::services::tier::warm_backend::TransitionCandidateProbe::Unsupported => {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::Unsupported,
+                        "remote tier cannot prove legacy unversioned transition state",
+                    ));
+                }
+                _ => {
+                    return Err(std::io::Error::new(
+                        std::io::ErrorKind::InvalidData,
+                        "remote tier object version state is unknown",
+                    ));
+                }
+            }
+        }
+    }
 
     // The same read plan the local path uses, so the tier fetch is positioned in
     // the object's *stored* coordinate system and the stream is handed the same
@@ -4724,9 +4955,6 @@ pub(crate) async fn get_transitioned_object_reader_with_tier_manager(
     // through a plaintext-coordinate range and skipping the transform is how a
     // transitioned SSE object used to come back as silently corrupt bytes of the
     // right length (rustfs/rustfs#6025).
-    let plan = ReadPlan::build_for_request(rs.clone(), oi, opts, h, resolver)
-        .await
-        .map_err(|err| std::io::Error::other(format!("building the read plan for {bucket}/{object} failed: {err}")))?;
     let (off, length) = (plan.storage_offset() as i64, plan.storage_length());
     let mut gopts = WarmBackendGetOpts::default();
 
@@ -5599,11 +5827,13 @@ mod tests {
     use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
     use crate::object_api::{ObjectInfo, ObjectOptions, PutObjReader};
     #[cfg(feature = "test-util")]
+    use crate::services::tier::test_util::MockWarmOp;
+    #[cfg(feature = "test-util")]
     use crate::services::tier::test_util::register_mock_tier;
     #[cfg(feature = "test-util")]
     use crate::services::tier::tier::TierConfigMgr;
     #[cfg(feature = "test-util")]
-    use crate::services::tier::warm_backend::WarmBackend as _;
+    use crate::services::tier::warm_backend::{TransitionCandidateProbe, WarmBackend as _};
     use crate::set_disk::{MultipartCommitBarrier, MultipartCommitPause};
     use crate::set_disk::{RUSTFS_MULTIPART_BUCKET_KEY, RUSTFS_MULTIPART_OBJECT_KEY};
     use crate::storage_api_contracts::namespace::NamespaceLocking as _;
@@ -6299,7 +6529,75 @@ mod tests {
 
     #[cfg(feature = "test-util")]
     #[tokio::test]
-    async fn transitioned_get_rejects_unknown_version_state_before_backend_io() {
+    async fn transitioned_get_allows_legacy_unknown_exact_version_for_non_destructive_read() {
+        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 remote_object = format!("remote/{}", Uuid::new_v4());
+        let body = Bytes::from_static(b"legacy transitioned object body");
+        let remote_version = backend
+            .put(
+                &remote_object,
+                ReaderImpl::Body(body.clone()),
+                i64::try_from(body.len()).expect("body length should fit"),
+            )
+            .await
+            .expect("mock remote object should be stored");
+        let mut user_defined = HashMap::new();
+        insert_legacy_transition_version_id(&mut user_defined, &remote_version);
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: i64::try_from(body.len()).expect("body length should fit"),
+            transitioned_object: TransitionedObject {
+                name: remote_object,
+                version_id: remote_version,
+                status: crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(),
+                tier: tier.clone(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined.into(),
+            ..Default::default()
+        };
+
+        let range = Some(crate::storage_api_contracts::range::HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 7,
+            end: 18,
+        });
+        let mut reader = get_transitioned_object_reader_with_tier_manager(
+            &object_info.bucket,
+            &object_info.name,
+            &range,
+            &HeaderMap::new(),
+            &object_info,
+            &ObjectOptions::default(),
+            &manager,
+            None,
+        )
+        .await
+        .expect("legacy unknown state should still allow a non-destructive read");
+        let mut got = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut got)
+            .await
+            .expect("transitioned reader should drain");
+
+        assert_eq!(got, &body.as_ref()[7..=18]);
+        assert_eq!(backend.get_count().await, 1);
+        assert_eq!(backend.remove_count().await, 0);
+        assert_eq!(
+            TierConfigMgr::active_operation_lease_count(&manager, &tier).await,
+            0,
+            "tier generation lease should release after EOF"
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn transitioned_get_rejects_explicit_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;
@@ -6315,6 +6613,181 @@ mod tests {
                 ..Default::default()
             },
             transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined_with_transition_version_state(rustfs_filemeta::TransitionVersionState::Unknown).into(),
+            ..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,
+            None,
+        )
+        .await
+        {
+            Ok(_) => panic!("explicit unknown remote version state must fail before backend IO"),
+            Err(err) => err,
+        };
+
+        assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(backend.op_log().await, Vec::<MockWarmOp>::new());
+        assert_eq!(backend.get_count().await, 0);
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn transitioned_get_rejects_present_but_invalid_legacy_version_metadata() {
+        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;
+
+        for persisted_version in [
+            Uuid::nil().to_string(),
+            "\u{fffd}".to_string(),
+            "bad\u{0001}version".to_string(),
+        ] {
+            let mut user_defined = HashMap::new();
+            insert_legacy_transition_version_id(&mut user_defined, &persisted_version);
+            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: tier.clone(),
+                    ..Default::default()
+                },
+                transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+                user_defined: user_defined.into(),
+                ..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,
+                None,
+            )
+            .await
+            {
+                Ok(_) => panic!("present but invalid legacy version metadata must fail before backend IO"),
+                Err(err) => err,
+            };
+
+            assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        }
+
+        assert_eq!(backend.op_log().await, Vec::<MockWarmOp>::new());
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn transitioned_get_probes_legacy_empty_unknown_state_before_unversioned_read() {
+        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;
+        backend.set_put_remote_version(Some(String::new())).await;
+        let remote_object = format!("remote/{}", Uuid::new_v4());
+        let body = Bytes::from_static(b"legacy unversioned transitioned object body");
+        let remote_version = backend
+            .put(
+                &remote_object,
+                ReaderImpl::Body(body.clone()),
+                i64::try_from(body.len()).expect("body length should fit"),
+            )
+            .await
+            .expect("mock remote object should be stored");
+        assert!(remote_version.is_empty());
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: i64::try_from(body.len()).expect("body length should fit"),
+            transitioned_object: TransitionedObject {
+                name: remote_object.clone(),
+                version_id: String::new(),
+                status: crate::bucket::lifecycle::lifecycle::TRANSITION_COMPLETE.to_string(),
+                tier: tier.clone(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: HashMap::from([("x-minio-internal-transitioned-versionID".to_string(), String::new())]).into(),
+            ..Default::default()
+        };
+
+        let mut reader = get_transitioned_object_reader_with_tier_manager(
+            &object_info.bucket,
+            &object_info.name,
+            &None,
+            &HeaderMap::new(),
+            &object_info,
+            &ObjectOptions::default(),
+            &manager,
+            None,
+        )
+        .await
+        .expect("probe-proven legacy unversioned state should allow a non-destructive read");
+        let mut got = Vec::new();
+        reader
+            .stream
+            .read_to_end(&mut got)
+            .await
+            .expect("transitioned reader should drain");
+
+        assert_eq!(got, body.as_ref());
+        assert_eq!(backend.remove_count().await, 0);
+        assert_eq!(
+            backend.op_log().await,
+            vec![
+                MockWarmOp::Put {
+                    object: remote_object.clone()
+                },
+                MockWarmOp::Probe {
+                    object: remote_object.clone()
+                },
+                MockWarmOp::Get { object: remote_object },
+            ]
+        );
+        assert_eq!(
+            TierConfigMgr::active_operation_lease_count(&manager, &tier).await,
+            0,
+            "tier generation lease should release after EOF"
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn transitioned_get_rejects_ambiguous_empty_unknown_state_without_backend_get() {
+        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 remote_object = format!("remote/{}", Uuid::new_v4());
+        backend
+            .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::VersionedPresent(
+                "versioned-candidate".to_string(),
+            )))
+            .await;
+        let object_info = ObjectInfo {
+            bucket: "bucket".to_string(),
+            name: "object".to_string(),
+            size: 1,
+            transitioned_object: TransitionedObject {
+                name: remote_object.clone(),
+                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()
         };
 
@@ -6330,19 +6803,28 @@ mod tests {
         )
         .await
         {
-            Ok(_) => panic!("unknown remote version state must fail before backend IO"),
+            Ok(_) => panic!("versioned legacy unknown state without stored version must fail before backend GET"),
             Err(err) => err,
         };
 
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert_eq!(backend.op_log().await, vec![MockWarmOp::Probe { object: remote_object }]);
         assert_eq!(backend.get_count().await, 0);
+        assert_eq!(backend.remove_count().await, 0);
     }
 
     #[cfg(feature = "test-util")]
     #[tokio::test]
-    async fn free_version_delete_rejects_unknown_version_state_before_backend_io() {
+    async fn free_version_delete_rejects_explicit_unknown_before_backend_io() {
         let manager = TierConfigMgr::new();
         let backend = register_mock_tier(&manager, "WARM").await;
+        let identity = test_tier_destination_identity(&manager, "WARM").await;
+        let mut user_defined = user_defined_with_tier_destination_identity(identity);
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut user_defined,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+            rustfs_filemeta::TransitionVersionState::Unknown.as_str().to_string(),
+        );
         let object_info = ObjectInfo {
             transitioned_object: TransitionedObject {
                 name: "remote/object".to_string(),
@@ -6351,17 +6833,251 @@ mod tests {
                 ..Default::default()
             },
             transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined.into(),
             ..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");
+            .expect_err("explicit unknown cleanup must fail before backend IO");
 
         assert_eq!(err.kind(), std::io::ErrorKind::InvalidData);
+        assert!(err.to_string().contains("version state is unknown"));
+        assert_eq!(backend.op_log().await, Vec::<MockWarmOp>::new());
         assert_eq!(backend.remove_count().await, 0);
     }
 
+    #[cfg(feature = "test-util")]
+    async fn test_tier_destination_identity(
+        manager: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
+        tier: &str,
+    ) -> crate::services::tier::tier::TierDestinationId {
+        TierConfigMgr::acquire_operation_lease(manager, tier)
+            .await
+            .expect("test tier lease should be available")
+            .backend_identity()
+    }
+
+    #[cfg(feature = "test-util")]
+    fn user_defined_with_tier_destination_identity(
+        identity: crate::services::tier::tier::TierDestinationId,
+    ) -> HashMap<String, String> {
+        let mut user_defined = HashMap::new();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut user_defined,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITION_TIER_DESTINATION_ID,
+            rustfs_utils::crypto::hex(identity),
+        );
+        user_defined
+    }
+
+    #[cfg(feature = "test-util")]
+    fn user_defined_with_transition_version_state(state: rustfs_filemeta::TransitionVersionState) -> HashMap<String, String> {
+        let mut user_defined = HashMap::new();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut user_defined,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+            state.as_str().to_string(),
+        );
+        user_defined
+    }
+
+    #[cfg(feature = "test-util")]
+    fn insert_legacy_transition_version_id(user_defined: &mut HashMap<String, String>, version_id: &str) {
+        rustfs_utils::http::metadata_compat::insert_str(
+            user_defined,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_ID,
+            version_id.to_string(),
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn free_version_tuple_rejects_mixed_legacy_missing_and_explicit_unknown() {
+        let manager = TierConfigMgr::new();
+        register_mock_tier(&manager, "WARM").await;
+        let identity = test_tier_destination_identity(&manager, "WARM").await;
+        let mut legacy_metadata = user_defined_with_tier_destination_identity(identity);
+        insert_legacy_transition_version_id(&mut legacy_metadata, "legacy-version");
+        let mut explicit_metadata = legacy_metadata.clone();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut explicit_metadata,
+            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+            rustfs_filemeta::TransitionVersionState::Unknown.as_str().to_string(),
+        );
+        let make_info = |user_defined: HashMap<String, String>| 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,
+            user_defined: user_defined.into(),
+            ..Default::default()
+        };
+
+        let err = super::free_version_remote_tuple_matches(&make_info(legacy_metadata), &make_info(explicit_metadata))
+            .expect_err("mixed legacy-missing and explicit unknown provenance must fail closed");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock);
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn free_version_delete_probes_exact_version_hidden_by_current_delete_marker() {
+        let manager = TierConfigMgr::new();
+        let tier = "WARM";
+        let backend = register_mock_tier(&manager, tier).await;
+        let identity = test_tier_destination_identity(&manager, tier).await;
+        let remote_object = format!("remote/{}", Uuid::new_v4());
+        let body = Bytes::from_static(b"legacy exact cleanup body");
+        let remote_version = backend
+            .put(
+                &remote_object,
+                ReaderImpl::Body(body),
+                i64::try_from(b"legacy exact cleanup body".len()).expect("body length should fit"),
+            )
+            .await
+            .expect("mock remote object should be stored");
+        let mut user_defined = user_defined_with_tier_destination_identity(identity);
+        insert_legacy_transition_version_id(&mut user_defined, &remote_version);
+        backend
+            .set_transition_candidate_probe_override(Some(TransitionCandidateProbe::Missing))
+            .await;
+        assert_eq!(
+            backend
+                .probe_transition_candidate_state(&remote_object)
+                .await
+                .expect("current remote view should be readable"),
+            TransitionCandidateProbe::Missing,
+            "a current delete marker must hide the historical data version from an unversioned probe"
+        );
+        backend.clear_op_log().await;
+        let object_info = ObjectInfo {
+            transitioned_object: TransitionedObject {
+                name: remote_object.clone(),
+                version_id: remote_version,
+                tier: tier.to_string(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined.into(),
+            ..Default::default()
+        };
+
+        super::delete_free_version_remote_object(&object_info, &manager)
+            .await
+            .expect("probe-proven legacy exact cleanup should delete the remote version");
+        super::delete_free_version_remote_object(&object_info, &manager)
+            .await
+            .expect("a retry after the exact remote version is already missing should be idempotent");
+
+        assert_eq!(
+            backend.op_log().await,
+            vec![
+                MockWarmOp::Get {
+                    object: remote_object.clone()
+                },
+                MockWarmOp::Remove {
+                    object: remote_object.clone()
+                },
+                MockWarmOp::Get {
+                    object: remote_object.clone()
+                },
+            ]
+        );
+        assert_eq!(
+            backend.remove_versions().await,
+            vec![(remote_object, object_info.transitioned_object.version_id)]
+        );
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn free_version_delete_retains_legacy_unknown_unversioned_object() {
+        let manager = TierConfigMgr::new();
+        let tier = "WARM";
+        let backend = register_mock_tier(&manager, tier).await;
+        backend.set_put_remote_version(Some(String::new())).await;
+        let identity = test_tier_destination_identity(&manager, tier).await;
+        let remote_object = format!("remote/{}", Uuid::new_v4());
+        let body = Bytes::from_static(b"legacy unversioned cleanup body");
+        let remote_version = backend
+            .put(
+                &remote_object,
+                ReaderImpl::Body(body),
+                i64::try_from(b"legacy unversioned cleanup body".len()).expect("body length should fit"),
+            )
+            .await
+            .expect("mock remote object should be stored");
+        assert!(remote_version.is_empty());
+        backend.clear_op_log().await;
+        let mut user_defined = user_defined_with_tier_destination_identity(identity);
+        user_defined.insert("x-minio-internal-transitioned-versionID".to_string(), String::new());
+        let object_info = ObjectInfo {
+            transitioned_object: TransitionedObject {
+                name: remote_object.clone(),
+                version_id: String::new(),
+                tier: tier.to_string(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined.into(),
+            ..Default::default()
+        };
+
+        let err = super::delete_free_version_remote_object(&object_info, &manager)
+            .await
+            .expect_err("legacy unversioned cleanup cannot exclude a versioning-state race");
+
+        assert_eq!(err.kind(), std::io::ErrorKind::WouldBlock);
+        assert!(backend.op_log().await.is_empty());
+        assert_eq!(backend.remove_count().await, 0);
+        assert!(backend.remove_versions().await.is_empty());
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    async fn free_version_delete_does_not_remove_a_different_remote_version() {
+        let manager = TierConfigMgr::new();
+        let tier = "WARM";
+        let backend = register_mock_tier(&manager, tier).await;
+        let identity = test_tier_destination_identity(&manager, tier).await;
+        let remote_object = format!("remote/{}", Uuid::new_v4());
+        backend.set_put_remote_version(Some("different-version".to_string())).await;
+        backend
+            .put(
+                &remote_object,
+                ReaderImpl::Body(Bytes::from_static(b"different remote version")),
+                i64::try_from(b"different remote version".len()).expect("body length should fit"),
+            )
+            .await
+            .expect("different remote version should be stored");
+        backend.clear_op_log().await;
+        let mut user_defined = user_defined_with_tier_destination_identity(identity);
+        insert_legacy_transition_version_id(&mut user_defined, "legacy-version");
+        let object_info = ObjectInfo {
+            transitioned_object: TransitionedObject {
+                name: remote_object.clone(),
+                version_id: "legacy-version".to_string(),
+                tier: tier.to_string(),
+                ..Default::default()
+            },
+            transition_version_state: rustfs_filemeta::TransitionVersionState::Unknown,
+            user_defined: user_defined.into(),
+            ..Default::default()
+        };
+
+        super::delete_free_version_remote_object(&object_info, &manager)
+            .await
+            .expect("a missing exact legacy version should be an idempotent cleanup success");
+
+        assert_eq!(backend.op_log().await, vec![MockWarmOp::Get { object: remote_object }]);
+        assert_eq!(backend.remove_count().await, 0);
+        assert!(backend.remove_versions().await.is_empty());
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     async fn free_version_remote_delete_requires_persisted_destination_identity() {

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

@@ -1170,6 +1170,7 @@ pub async fn save_manual_transition_job_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()
@@ -1242,6 +1243,7 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1270,6 +1272,7 @@ pub(crate) async fn save_manual_transition_task_if_absent(
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1621,6 +1624,7 @@ pub async fn save_manual_transition_scope_admission_if_absent(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1672,6 +1676,7 @@ pub async fn save_manual_transition_scope_admission_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

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

@@ -1733,6 +1733,7 @@ async fn save_config_if_none_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -1832,6 +1833,7 @@ async fn save_decommission_manifest_checkpoint_if_match(
 
         let mut opts = ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             no_lock: true,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(observed_etag),
@@ -1960,6 +1962,7 @@ async fn save_config_if_match_fenced(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag.to_string()),
                 ..Default::default()
@@ -3780,6 +3783,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -3869,6 +3873,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()
@@ -3893,6 +3898,7 @@ where
                 data.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use super::runtime_boundary as runtime_sources;
 use crate::bucket::lifecycle::bucket_lifecycle_ops::ExpiryOp;
@@ -72,9 +70,11 @@ static REMOTE_DELETE_BREAKER: LazyLock<Mutex<RemoteDeleteBreaker>> = LazyLock::n
 });
 
 #[cfg(test)]
-static REMOTE_TIER_DELETE_TEST_HOOK: std::sync::LazyLock<
-    std::sync::Mutex<Option<Box<dyn Fn(&str, &str, &str) -> std::io::Result<()> + Send + Sync>>>,
-> = std::sync::LazyLock::new(|| std::sync::Mutex::new(None));
+type RemoteTierDeleteTestHook = Box<dyn Fn(&str, &str, &str) -> std::io::Result<()> + Send + Sync>;
+
+#[cfg(test)]
+static REMOTE_TIER_DELETE_TEST_HOOK: std::sync::LazyLock<std::sync::Mutex<Option<RemoteTierDeleteTestHook>>> =
+    std::sync::LazyLock::new(|| std::sync::Mutex::new(None));
 
 #[derive(Debug)]
 struct RemoteDeleteBreaker {
@@ -107,7 +107,7 @@ impl RemoteDeleteBreaker {
     fn prune(&mut self, now: Instant) {
         while let Some(ts) = self.failures.front().copied() {
             if now.duration_since(ts) > self.window {
-                self.failures.pop_front();
+                let _ = self.failures.pop_front();
             } else {
                 break;
             }
@@ -137,10 +137,10 @@ fn is_signer_header_error(err: &std::io::Error) -> bool {
         return false;
     }
 
-    if let Some(source) = err.get_ref() {
-        if error_chain_contains_signer_header_marker(source) {
-            return true;
-        }
+    if let Some(source) = err.get_ref()
+        && error_chain_contains_signer_header_marker(source)
+    {
+        return true;
     }
 
     let message = err.to_string().to_ascii_lowercase();
@@ -205,7 +205,7 @@ impl ObjSweeper {
 
     #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn with_version(&mut self, vid: Option<Uuid>) -> &Self {
-        self.version_id = vid.clone();
+        self.version_id = vid;
         self
     }
 
@@ -219,7 +219,7 @@ impl ObjSweeper {
     #[allow(dead_code, reason = "MinIO-parity surface with no caller in this port (backlog#1823)")]
     pub fn get_opts(&self) -> lifecycle::ObjectOpts {
         let mut opts = ObjectOpts {
-            version_id: self.version_id.clone(),
+            version_id: self.version_id,
             versioned: self.versioned,
             version_suspended: self.suspended,
             ..Default::default()
@@ -388,8 +388,8 @@ impl Jentry {
 impl ExpiryOp for Jentry {
     fn op_hash(&self) -> u64 {
         let mut hasher = Sha256::new();
-        hasher.update(format!("{}", self.tier_name).as_bytes());
-        hasher.update(format!("{}", self.obj_name).as_bytes());
+        hasher.update(self.tier_name.as_bytes());
+        hasher.update(self.obj_name.as_bytes());
         xxh64::xxh64(hasher.finalize().as_slice(), XXHASH_SEED)
     }
 
@@ -436,7 +436,7 @@ async fn delete_object_from_remote_tier_raw_with_manager(
     tier_name: &str,
     tier_config_mgr: &Arc<tokio::sync::RwLock<TierConfigMgr>>,
 ) -> Result<(), std::io::Error> {
-    let lease = TierConfigMgr::acquire_operation_lease(&tier_config_mgr, tier_name)
+    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, true).await

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

@@ -612,6 +612,7 @@ pub(crate) async fn save_transition_transaction_record(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -658,6 +659,7 @@ pub(crate) async fn save_transition_transaction_record_if_current(
         data.clone(),
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(etag),
                 ..Default::default()

crates/ecstore/src/bucket/on_demand_migration/backfill.rs

@@ -684,6 +684,7 @@ async fn write_checkpoint(
     };
     let opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         http_preconditions: Some(preconditions),
         ..Default::default()
     };

crates/ecstore/src/bucket/on_demand_migration/breaker.rs

@@ -86,7 +86,12 @@ impl BreakerVerdict {
             Some(SourceError::Throttled | SourceError::Timeout | SourceError::Connect(_) | SourceError::ServerError(_)) => {
                 BreakerVerdict::Failure
             }
-            Some(SourceError::AccessDenied | SourceError::Unsupported(_) | SourceError::Other(_)) => BreakerVerdict::Neutral,
+            Some(
+                SourceError::AccessDenied
+                | SourceError::Unsupported(_)
+                | SourceError::InvalidPagination(_)
+                | SourceError::Other(_),
+            ) => BreakerVerdict::Neutral,
         }
     }
 }

crates/ecstore/src/bucket/on_demand_migration/list_through.rs

@@ -189,8 +189,8 @@ pub enum SourceListPlan {
     /// delimiter — the source's own roll-up boundary matches the request's.
     Page { prefix: String },
     /// `filter.prefix` reaches past a delimiter, so every key the source could
-    /// contribute rolls into this one common prefix. One bounded probe listing
-    /// decides whether it exists; there is nothing to paginate.
+    /// contribute rolls into this one common prefix. Bounded probes follow
+    /// empty progressing pages until a key proves existence or the source ends.
     Folded { probe_prefix: String, common_prefix: String },
 }
 
@@ -279,6 +279,29 @@ pub struct FetchRequest {
     pub token: Option<String>,
 }
 
+/// Invalid pagination metadata. Opaque cursor values are never included in errors.
+#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
+pub enum ListPageError {
+    #[error("truncated listing has no continuation token")]
+    Missing,
+    #[error("truncated listing has an empty continuation token")]
+    Empty,
+    #[error("truncated listing repeats a continuation token")]
+    Repeated,
+}
+
+pub(crate) fn validate_list_page(is_truncated: bool, token: Option<&str>, next_token: Option<&str>) -> Result<(), ListPageError> {
+    if is_truncated {
+        match next_token {
+            None => return Err(ListPageError::Missing),
+            Some("") => return Err(ListPageError::Empty),
+            Some(next) if Some(next) == token => return Err(ListPageError::Repeated),
+            Some(_) => {}
+        }
+    }
+    Ok(())
+}
+
 #[derive(Debug, Default)]
 struct SideState {
     start: SideCursor,
@@ -364,6 +387,11 @@ impl ListThroughMerger {
     /// or `filter.prefix` excludes it.
     pub fn disable_source(&mut self) {
         self.source.disabled = true;
+        // A refill can fail after a valid first page. A local-only response
+        // must discard both that source payload and its ordering horizon.
+        self.source.entries.clear();
+        self.source.pages.clear();
+        self.source.more = false;
     }
 
     pub fn next_fetch(&self) -> Option<FetchRequest> {
@@ -378,7 +406,13 @@ impl ListThroughMerger {
     /// Records one fetched page. `entries` must be sorted by `name` and already
     /// filtered with [`Self::accepts`]; the caller keeps the matching payloads
     /// in the same order.
-    pub fn push_page(&mut self, side: MergeSide, entries: Vec<ListEntryKey>, is_truncated: bool, next_token: Option<String>) {
+    pub fn push_page(
+        &mut self,
+        side: MergeSide,
+        entries: Vec<ListEntryKey>,
+        is_truncated: bool,
+        next_token: Option<String>,
+    ) -> Result<(), ListPageError> {
         let state = match side {
             MergeSide::Local => &mut self.local,
             MergeSide::Source => &mut self.source,
@@ -387,15 +421,19 @@ impl ListThroughMerger {
             Some(last) => last.next_token.clone(),
             None => state.start.token.clone(),
         };
-        // A truncated page without a cursor cannot be continued; treating the
-        // side as finished is the only alternative to looping on it forever.
-        state.more = is_truncated && next_token.is_some();
+        validate_list_page(is_truncated, token.as_deref(), next_token.as_deref())?;
+        // Also reject a cycle through an earlier page in this bounded fetch.
+        if is_truncated && state.pages.iter().any(|page| page.token == next_token) {
+            return Err(ListPageError::Repeated);
+        }
+        state.more = is_truncated;
         state.pages.push(FetchedPage {
             token,
             count: entries.len(),
             next_token: is_truncated.then_some(next_token).flatten(),
         });
         state.entries.extend(entries);
+        Ok(())
     }
 
     pub fn finish(self) -> MergeOutcome {
@@ -599,9 +637,15 @@ mod tests {
                 let (entries, truncated, next) = reference_page(keys, prefix, delimiter, fetch.token.as_deref(), max_keys);
                 let kept: Vec<ListEntryKey> = entries.into_iter().filter(|entry| merger.accepts(&entry.name)).collect();
                 buffers[usize::from(fetch.side == MergeSide::Source)].extend(kept.iter().cloned());
-                merger.push_page(fetch.side, kept, truncated, next);
+                merger
+                    .push_page(fetch.side, kept, truncated, next)
+                    .expect("reference provider pages must advance");
             }
             let outcome = merger.finish();
+            assert_eq!(outcome.is_truncated, outcome.next_token.is_some());
+            if outcome.is_truncated {
+                assert_ne!(outcome.next_token, token, "every truncated merged page must make progress");
+            }
             page_sizes.push(outcome.picks.len());
             for pick in &outcome.picks {
                 let entry = buffers[usize::from(pick.side == MergeSide::Source)][pick.index].clone();
@@ -616,11 +660,25 @@ mod tests {
     }
 
     fn expected(local: &[String], source: &[String], prefix: &str, delimiter: Option<&str>) -> Vec<ListEntryKey> {
-        let mut all: Vec<String> = local.iter().chain(source.iter()).cloned().collect();
-        all.sort();
-        all.dedup();
-        let (entries, _, _) = reference_page(&all, prefix, delimiter, None, usize::MAX);
-        entries
+        // This oracle builds the complete namespace independently of the
+        // provider's page/marker helper and the production merger.
+        let mut namespace = std::collections::BTreeMap::new();
+        for key in local.iter().chain(source) {
+            let Some(suffix) = key.strip_prefix(prefix) else {
+                continue;
+            };
+            if let Some(delimiter) = delimiter.filter(|delimiter| !delimiter.is_empty())
+                && let Some((directory, _)) = suffix.split_once(delimiter)
+            {
+                namespace.insert(format!("{prefix}{directory}{delimiter}"), true);
+                continue;
+            }
+            namespace.insert(key.clone(), false);
+        }
+        namespace
+            .into_iter()
+            .map(|(name, is_prefix)| ListEntryKey { name, is_prefix })
+            .collect()
     }
 
     #[test]
@@ -662,7 +720,9 @@ mod tests {
                 token: None
             })
         );
-        merger.push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("a")], false, None)
+            .expect("local EOF is valid");
         assert_eq!(merger.next_fetch(), None);
         let outcome = merger.finish();
         assert_eq!(outcome.picks.len(), 1);
@@ -683,12 +743,14 @@ mod tests {
         };
         let mut merger = ListThroughMerger::new(1, Some(&resume));
         merger.disable_source();
-        merger.push_page(
-            MergeSide::Local,
-            vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
-            true,
-            Some("local-2".to_string()),
-        );
+        merger
+            .push_page(
+                MergeSide::Local,
+                vec![ListEntryKey::object("b"), ListEntryKey::object("c")],
+                true,
+                Some("local-2".to_string()),
+            )
+            .expect("local cursor advances");
         let outcome = merger.finish();
         assert!(outcome.is_truncated);
         let token = outcome.next_token.expect("truncated page carries a token");
@@ -698,6 +760,212 @@ mod tests {
         assert_eq!(token.local.as_deref(), Some("local-1"), "a partly read page is re-listed");
     }
 
+    #[test]
+    fn truncated_pages_require_a_nonempty_advancing_cursor() {
+        for side in [MergeSide::Local, MergeSide::Source] {
+            for entries in [vec![], vec![ListEntryKey::object("a")]] {
+                for (next, expected) in [
+                    (None, Err(ListPageError::Missing)),
+                    (Some(""), Err(ListPageError::Empty)),
+                    (Some("stuck"), Err(ListPageError::Repeated)),
+                    (Some("advances"), Ok(())),
+                ] {
+                    let resume = ListThroughToken::new(
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        SideCursor {
+                            token: Some("stuck".into()),
+                            done: false,
+                        },
+                        None,
+                    );
+                    let mut merger = ListThroughMerger::new(2, Some(&resume));
+                    let result = merger.push_page(side, entries.clone(), true, next.map(str::to_string));
+                    assert_eq!(result, expected, "{side:?}, {entries:?}, {next:?}");
+                    let state = if side == MergeSide::Local {
+                        &merger.local
+                    } else {
+                        &merger.source
+                    };
+                    assert_eq!(state.pages.len(), usize::from(result.is_ok()), "invalid page must not be accepted");
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn repeated_empty_cursor_is_rejected_before_an_identical_page_can_escape() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("stuck".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        assert_eq!(
+            merger.next_fetch(),
+            Some(FetchRequest {
+                side: MergeSide::Source,
+                token: Some("stuck".into())
+            })
+        );
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn empty_pages_may_advance_within_the_fetch_budget_until_eof() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger.push_page(MergeSide::Local, vec![], false, None).expect("local EOF");
+        for next in ["opaque-z", "opaque-a"] {
+            assert_eq!(merger.next_fetch().expect("bounded source fetch").side, MergeSide::Source);
+            merger
+                .push_page(MergeSide::Source, vec![], true, Some(next.into()))
+                .expect("opaque cursor advances regardless of sort order");
+        }
+        assert!(merger.next_fetch().is_none(), "two source fetches exhaust the request budget");
+        let outcome = merger.finish();
+        assert!(outcome.picks.is_empty());
+        assert!(outcome.is_truncated);
+        let token = outcome.next_token.expect("empty progressing page has a cursor");
+        assert_eq!(token.source.as_deref(), Some("opaque-a"));
+        let mut merger = ListThroughMerger::new(2, Some(&token));
+        assert_eq!(merger.next_fetch().expect("source resumes").token.as_deref(), Some("opaque-a"));
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("result")], false, None)
+            .expect("source EOF");
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Source,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn a_cursor_cycle_inside_the_fetch_budget_is_rejected() {
+        let resume = ListThroughToken::new(
+            SideCursor { token: None, done: true },
+            SideCursor {
+                token: Some("first".into()),
+                done: false,
+            },
+            None,
+        );
+        let mut merger = ListThroughMerger::new(2, Some(&resume));
+        merger
+            .push_page(MergeSide::Source, vec![], true, Some("second".into()))
+            .expect("first page advances");
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("first".into())),
+            Err(ListPageError::Repeated)
+        );
+    }
+
+    #[test]
+    fn source_refill_failure_discards_buffered_source_entries_and_horizon() {
+        let mut merger = ListThroughMerger::new(2, None);
+        merger
+            .push_page(MergeSide::Local, vec![ListEntryKey::object("z")], false, None)
+            .expect("local EOF");
+        merger
+            .push_page(MergeSide::Source, vec![ListEntryKey::object("a")], true, Some("stuck".into()))
+            .expect("first source page advances");
+        assert_eq!(merger.next_fetch().expect("source refill is required").token.as_deref(), Some("stuck"));
+        assert_eq!(
+            merger.push_page(MergeSide::Source, vec![], true, Some("stuck".into())),
+            Err(ListPageError::Repeated)
+        );
+        merger.disable_source();
+        let outcome = merger.finish();
+        assert_eq!(
+            outcome.picks,
+            vec![MergePick {
+                side: MergeSide::Local,
+                index: 0
+            }]
+        );
+        assert!(!outcome.is_truncated);
+        assert!(outcome.next_token.is_none());
+    }
+
+    #[test]
+    fn list_through_static_namespace_boundary_matrix() {
+        let corpus = [
+            "a",
+            "a/",
+            "a/b",
+            "a/b/child",
+            "a0",
+            "b",
+            "b/leaf",
+            "quote\"&<",
+            "space key",
+            "z",
+            "é",
+            "中/文",
+        ];
+        for count in [0, 1, 3, 4, corpus.len()] {
+            let keys: Vec<String> = corpus[..count].iter().map(|key| (*key).to_string()).collect();
+            for placement in 0..3 {
+                let (local, source): (Vec<_>, Vec<_>) =
+                    keys.iter()
+                        .enumerate()
+                        .fold((vec![], vec![]), |(mut local, mut source), (index, key)| {
+                            if placement != 1 || index % 2 == 0 {
+                                local.push(key.clone());
+                            }
+                            if placement != 0 || index % 2 == 0 {
+                                source.push(key.clone());
+                            }
+                            (local, source)
+                        });
+                for prefix in ["", "a", "a/", "中/"] {
+                    for delimiter in [None, Some("/")] {
+                        for max_keys in [1, 3, 4] {
+                            let oracle = expected(&local, &source, prefix, delimiter);
+                            let (emitted, sizes) = walk(&local, &source, prefix, delimiter, max_keys);
+                            assert_eq!(
+                                emitted.iter().map(|(entry, _)| entry.clone()).collect::<Vec<_>>(),
+                                oracle,
+                                "count={count}, placement={placement}, prefix={prefix}, delimiter={delimiter:?}, max={max_keys}"
+                            );
+                            let expected_sizes: Vec<_> = if oracle.is_empty() {
+                                vec![0]
+                            } else {
+                                oracle.chunks(max_keys).map(<[ListEntryKey]>::len).collect()
+                            };
+                            assert_eq!(sizes, expected_sizes, "exact max and max+1 boundaries must agree");
+                        }
+                    }
+                }
+            }
+        }
+    }
+
+    #[test]
+    fn list_through_large_overlap_walk_keeps_all_5300_keys() {
+        let source: Vec<_> = (0..5000).map(|index| format!("k{index:05}")).collect();
+        let local: Vec<_> = (4800..5300).map(|index| format!("k{index:05}")).collect();
+        let (emitted, sizes) = walk(&local, &source, "", None, 333);
+        assert_eq!(emitted.len(), 5300);
+        for (index, (entry, side)) in emitted.iter().enumerate() {
+            assert_eq!(entry.name, format!("k{index:05}"));
+            assert_eq!(*side, if index >= 4800 { MergeSide::Local } else { MergeSide::Source });
+        }
+        assert_eq!(sizes, [vec![333; 15], vec![305]].concat());
+    }
+
     #[test]
     fn token_round_trips_and_rejects_tampering() {
         let token = ListThroughToken::new(
@@ -796,7 +1064,10 @@ mod tests {
     }
 
     proptest! {
-        #![proptest_config(ProptestConfig::with_cases(256))]
+        #![proptest_config(ProptestConfig {
+            rng_seed: proptest::test_runner::RngSeed::Fixed(0xec5706),
+            ..ProptestConfig::with_cases(256)
+        })]
 
         /// Full pagination of a merged listing equals the sorted, deduplicated
         /// union of both sides, with every shared key served by local, and no

crates/ecstore/src/bucket/on_demand_migration/source_client.rs

@@ -25,6 +25,7 @@
 //! Client-supplied `If-*`, `Authorization`, `Host` and SSE-C headers are never
 //! forwarded: v1 rejects SSE-C source objects outright.
 
+use super::list_through::{ListPageError, validate_list_page};
 use crate::bucket::remote_s3_client::{
     PathStyle, RemoteCredentials, RemoteS3ClientError, RemoteS3EndpointSpec, RemoteS3RetryPolicy, build_remote_s3_config,
 };
@@ -223,6 +224,8 @@ pub enum SourceError {
     ServerError(u16),
     #[error("unsupported source object: {0}")]
     Unsupported(String),
+    #[error("invalid source listing: {0}")]
+    InvalidPagination(#[from] ListPageError),
     #[error("source request failed: {0}")]
     Other(String),
 }
@@ -245,6 +248,7 @@ impl SourceError {
             SourceError::Connect(_) => "connect",
             SourceError::ServerError(_) => "server_error",
             SourceError::Unsupported(_) => "unsupported",
+            SourceError::InvalidPagination(_) => "invalid_pagination",
             SourceError::Other(_) => "other",
         }
     }
@@ -714,6 +718,7 @@ impl SourceClient {
                 ..*request
             })
             .await?;
+        validate_list_page(page.is_truncated, request.continuation_token, page.next_continuation_token.as_deref())?;
         page.objects = page
             .objects
             .into_iter()
@@ -800,11 +805,6 @@ impl SourceBackend for S3SourceBackend {
 
         let is_truncated = output.is_truncated.unwrap_or(false);
         let next_continuation_token = output.next_continuation_token;
-        if is_truncated && next_continuation_token.is_none() {
-            return Err(SourceError::Other(
-                "source reported a truncated listing without a continuation token".to_string(),
-            ));
-        }
         let objects = output
             .contents
             .unwrap_or_default()
@@ -1274,7 +1274,9 @@ mod tests {
 <CommonPrefixes><Prefix>data/photos/</Prefix></CommonPrefixes>
 <CommonPrefixes><Prefix>outside/</Prefix></CommonPrefixes>
 </ListBucketResult>"#;
-        let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), body)]).await;
+        let next_body = body.replace("data/opaque", "data/next");
+        let (client, requests) =
+            scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), body), ok(Vec::new(), &next_body)]).await;
         let first = client
             .list_page(&SourceListRequest {
                 prefix: Some("photos/"),
@@ -1336,7 +1338,104 @@ mod tests {
             .list_objects_v2(None, None, 10)
             .await
             .expect_err("truncated page without token is corrupt");
-        assert!(matches!(err, SourceError::Other(_)), "{err:?}");
+        assert!(matches!(err, SourceError::InvalidPagination(ListPageError::Missing)), "{err:?}");
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_s3_cursor_progress_before_mapping_entries() {
+        for contents in ["", "<Contents><Key>data/a</Key><Size>1</Size></Contents>"] {
+            for (truncated, next, expected) in [
+                (true, None, Some(ListPageError::Missing)),
+                (true, Some(""), Some(ListPageError::Empty)),
+                (true, Some("stuck"), Some(ListPageError::Repeated)),
+                (true, Some("opaque-next"), None),
+                (false, None, None),
+                (false, Some("stuck"), None),
+            ] {
+                let next_xml = next
+                    .map(|next| format!("<NextContinuationToken>{next}</NextContinuationToken>"))
+                    .unwrap_or_default();
+                let body = format!(
+                    "<ListBucketResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><IsTruncated>{truncated}</IsTruncated>{next_xml}{contents}</ListBucketResult>"
+                );
+                let (client, requests) = scripted_client(&spec(Some("data/")), vec![ok(Vec::new(), &body)]).await;
+                let result = client
+                    .list_page(&SourceListRequest {
+                        continuation_token: Some("stuck"),
+                        max_keys: 2,
+                        ..Default::default()
+                    })
+                    .await;
+                match expected {
+                    Some(expected) => {
+                        let error = result.expect_err("malformed pagination must fail at the provider boundary");
+                        assert!(
+                            matches!(&error, SourceError::InvalidPagination(actual) if *actual == expected),
+                            "{error:?}"
+                        );
+                        assert_eq!(error.class_label(), "invalid_pagination");
+                        assert!(!error.is_retryable());
+                        assert!(!error.to_string().contains("stuck"), "errors must not echo opaque tokens");
+                    }
+                    None => {
+                        let page = result.expect("progressing empty/nonempty pages and EOF are valid");
+                        assert_eq!(page.is_truncated, truncated);
+                        assert_eq!(page.next_continuation_token.as_deref(), next);
+                        assert_eq!(page.objects.len(), usize::from(!contents.is_empty()));
+                        if let Some(object) = page.objects.first() {
+                            assert_eq!(object.key, "a");
+                        }
+                    }
+                }
+                let requests = recorded(&requests);
+                assert_eq!(requests.len(), 1, "invalid pagination must not be retried");
+                assert!(requests[0].uri.contains("continuation-token=stuck"));
+            }
+        }
+    }
+
+    struct ListOnlyBackend(SourcePage);
+
+    #[async_trait::async_trait]
+    impl SourceBackend for ListOnlyBackend {
+        async fn list(&self, request: &SourceListRequest<'_>) -> Result<SourcePage, SourceError> {
+            assert_eq!(request.continuation_token, Some("stuck"), "opaque cursors reach every provider unchanged");
+            Ok(self.0.clone())
+        }
+
+        async fn head(&self, _key: &str) -> Result<SourceHead, SourceError> {
+            panic!("unexpected HEAD in list test")
+        }
+        async fn get(&self, _key: &str, _range: Option<&HTTPRangeSpec>) -> Result<SourceGet, SourceError> {
+            panic!("unexpected GET in list test")
+        }
+        async fn tagging(&self, _key: &str) -> Result<HashMap<String, String>, SourceError> {
+            panic!("unexpected tagging in list test")
+        }
+        async fn probe(&self) -> Result<(), SourceError> {
+            panic!("unexpected probe in list test")
+        }
+    }
+
+    #[tokio::test]
+    async fn list_page_validates_non_s3_provider_cursors_at_the_common_boundary() {
+        for (next, expected) in [
+            (None, ListPageError::Missing),
+            (Some(""), ListPageError::Empty),
+            (Some("stuck"), ListPageError::Repeated),
+        ] {
+            let mut client = prefix_client(Some("data/".into()));
+            client.backend = Box::new(ListOnlyBackend(SourcePage {
+                is_truncated: true,
+                next_continuation_token: next.map(str::to_string),
+                ..Default::default()
+            }));
+            let error = client
+                .list_objects_v2(None, Some("stuck"), 2)
+                .await
+                .expect_err("all providers must advance pagination");
+            assert!(matches!(error, SourceError::InvalidPagination(actual) if actual == expected));
+        }
     }
 
     const TAGGING_BODY: &str = r#"<?xml version="1.0" encoding="UTF-8"?>

crates/ecstore/src/bucket/on_demand_migration/stats.rs

@@ -177,7 +177,7 @@ impl From<&SourceError> for PullFailureReason {
             SourceError::Connect(_) => PullFailureReason::SourceConnect,
             SourceError::ServerError(_) => PullFailureReason::SourceServerError,
             SourceError::Unsupported(_) => PullFailureReason::SourceUnsupported,
-            SourceError::Other(_) => PullFailureReason::SourceOther,
+            SourceError::InvalidPagination(_) | SourceError::Other(_) => PullFailureReason::SourceOther,
         }
     }
 }

crates/ecstore/src/bucket/remote_s3_client.rs

@@ -652,9 +652,10 @@ async fn build_aws_s3_http_client_from_tls_path() -> Option<SharedHttpClient> {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use aws_smithy_async::time::TimeSource;
     use aws_smithy_runtime_api::http::StatusCode as SmithyStatusCode;
     use std::sync::Mutex;
-    use std::sync::atomic::{AtomicUsize, Ordering};
+    use std::sync::atomic::{AtomicU64, AtomicUsize, Ordering};
 
     fn spec(endpoint: &str, secure: bool) -> RemoteS3EndpointSpec {
         RemoteS3EndpointSpec {
@@ -824,6 +825,174 @@ mod tests {
         );
     }
 
+    #[derive(Clone, Debug)]
+    struct ClockSkewTimeSource(Arc<AtomicU64>);
+
+    impl TimeSource for ClockSkewTimeSource {
+        fn now(&self) -> SystemTime {
+            SystemTime::UNIX_EPOCH + Duration::from_secs(self.0.load(Ordering::SeqCst))
+        }
+    }
+
+    #[derive(Clone, Debug)]
+    struct ClockSkewConnector {
+        request_headers: RecordedHeaders,
+        error_code: &'static str,
+        skew_seconds: i64,
+        clock: ClockSkewTimeSource,
+    }
+
+    fn recorded_header<'a>(headers: &'a [(String, String)], name: &str) -> &'a str {
+        headers
+            .iter()
+            .find(|(key, _)| key.eq_ignore_ascii_case(name))
+            .map(|(_, value)| value.as_str())
+            .unwrap_or_else(|| panic!("signed request must contain {name}"))
+    }
+
+    fn signing_time(headers: &[(String, String)]) -> chrono::NaiveDateTime {
+        chrono::NaiveDateTime::parse_from_str(recorded_header(headers, "x-amz-date"), "%Y%m%dT%H%M%SZ")
+            .expect("SDK signing timestamp must use the SigV4 format")
+    }
+
+    impl SmithyHttpConnector for ClockSkewConnector {
+        fn call(&self, request: HttpRequest) -> HttpConnectorFuture {
+            let mut headers = self.request_headers.lock().expect("clock skew request capture lock");
+            assert!(headers.len() < 3, "clock skew fixture must not exceed two GET attempts and one HEAD");
+            headers.push(
+                request
+                    .headers()
+                    .iter()
+                    .map(|(key, value)| (key.to_string(), value.to_string()))
+                    .collect(),
+            );
+            let server_time = chrono::DateTime::<chrono::Utc>::from(self.clock.now()).naive_utc()
+                + chrono::Duration::seconds(self.skew_seconds);
+            let (status, body) = if headers.len() == 1 {
+                (
+                    403,
+                    format!("<Error><Code>{}</Code><Message>Clock skew fixture</Message></Error>", self.error_code),
+                )
+            } else {
+                (200, String::new())
+            };
+            let response = http::Response::builder()
+                .status(status)
+                .header("date", server_time.format("%a, %d %b %Y %H:%M:%S GMT").to_string())
+                .header("content-type", "application/xml")
+                .header("content-length", body.len())
+                .body(SdkBody::from(body))
+                .expect("clock skew fixture response");
+            HttpConnectorFuture::ready(Ok(HttpResponse::try_from(response).expect("Smithy fixture response")))
+        }
+    }
+
+    async fn clock_skew_client(
+        error_code: &'static str,
+        skew_seconds: i64,
+        retry: RemoteS3RetryPolicy,
+    ) -> (S3Client, RecordedHeaders, ClockSkewTimeSource) {
+        let headers: RecordedHeaders = Arc::new(Mutex::new(Vec::new()));
+        let clock = ClockSkewTimeSource(Arc::new(AtomicU64::new(1_700_000_000)));
+        let connector = SharedHttpConnector::new(ClockSkewConnector {
+            request_headers: Arc::clone(&headers),
+            error_code,
+            skew_seconds,
+            clock: clock.clone(),
+        });
+        let mut spec = spec("s3.example.com", true);
+        spec.retry = retry;
+        let config = build_remote_s3_config(&spec)
+            .await
+            .expect("clock skew fixture uses the production outbound configuration")
+            .http_client(http_client_fn(move |_settings, _components| connector.clone()))
+            .time_source(clock.clone())
+            .build();
+        (S3Client::from_conf(config), headers, clock)
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn remote_s3_clock_skew_retries_resign_and_seed_next_operation() {
+        for error_code in ["RequestTimeTooSkewed", "SignatureDoesNotMatch"] {
+            for skew_seconds in [-600, 600] {
+                let (client, headers, clock) = clock_skew_client(error_code, skew_seconds, REPLICATION_TARGET_RETRY_POLICY).await;
+                let initial = chrono::DateTime::<chrono::Utc>::from(clock.now()).naive_utc();
+                client
+                    .get_object()
+                    .bucket("bucket")
+                    .key("object")
+                    .send()
+                    .await
+                    .expect("clock skew GET must retry successfully");
+                assert_eq!(
+                    headers.lock().expect("captured requests").len(),
+                    2,
+                    "{error_code}: GET needs exactly one retry"
+                );
+                clock.0.fetch_add(17, Ordering::SeqCst);
+                // SDK signing time is independent of Tokio's retry/scheduler clock.
+                tokio::time::advance(Duration::from_secs(61)).await;
+                client
+                    .head_bucket()
+                    .bucket("bucket")
+                    .send()
+                    .await
+                    .expect("subsequent HEAD must use the client's cached skew");
+                let headers = headers.lock().expect("captured signed requests");
+                assert_eq!(headers.len(), 3, "subsequent operation must succeed on its first attempt");
+                assert_eq!(signing_time(&headers[0]), initial, "the first attempt must use the injected clock");
+                assert_eq!(
+                    signing_time(&headers[1]),
+                    initial + chrono::Duration::seconds(skew_seconds),
+                    "{error_code}: retry must apply the measured offset exactly"
+                );
+                assert_eq!(
+                    signing_time(&headers[2]),
+                    initial + chrono::Duration::seconds(skew_seconds + 17),
+                    "{error_code}: the next operation must apply cached skew to the advanced signing clock"
+                );
+                let signature = |index: usize| {
+                    recorded_header(&headers[index], "authorization")
+                        .rsplit_once("Signature=")
+                        .expect("SigV4 authorization contains a signature")
+                        .1
+                };
+                assert_ne!(
+                    signature(0),
+                    signature(1),
+                    "{error_code}: retry must be signed again after adjusting its date"
+                );
+            }
+        }
+    }
+
+    #[tokio::test(start_paused = true)]
+    async fn remote_s3_clock_skew_respects_one_attempt_policy() {
+        use aws_smithy_types::error::metadata::ProvideErrorMetadata;
+
+        for error_code in ["RequestTimeTooSkewed", "SignatureDoesNotMatch"] {
+            for retry in [
+                RemoteS3RetryPolicy::Disabled,
+                RemoteS3RetryPolicy::Standard { max_attempts: 1 },
+            ] {
+                let (client, headers, _clock) = clock_skew_client(error_code, 600, retry).await;
+                let error = client
+                    .get_object()
+                    .bucket("bucket")
+                    .key("object")
+                    .send()
+                    .await
+                    .expect_err("clock skew must not override the caller's one-attempt budget");
+                assert_eq!(error.as_service_error().and_then(ProvideErrorMetadata::code), Some(error_code));
+                assert_eq!(
+                    headers.lock().expect("captured requests").len(),
+                    1,
+                    "{error_code}: {retry:?} must send exactly one request"
+                );
+            }
+        }
+    }
+
     #[test]
     fn path_style_auto_and_path_force_path_style() {
         assert!(PathStyle::Auto.force_path_style());

crates/ecstore/src/core/pools.rs

@@ -5493,6 +5493,7 @@ where
     fence.ensure_held()?;
     let mut opts = ObjectOptions {
         max_parity: true,
+        write_completion: crate::object_api::WriteCompletion::TailDrained,
         no_lock: true,
         http_preconditions: Some(pool_meta_cas_preconditions(token, object)?),
         ..Default::default()
@@ -14412,6 +14413,7 @@ impl ECStore {
                 encoded.clone(),
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_none_match: Some("*".to_string()),
                         ..Default::default()
@@ -14566,6 +14568,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(http_preconditions),
                     ..Default::default()
                 },
@@ -14957,6 +14960,7 @@ impl ECStore {
                 encoded,
                 &ObjectOptions {
                     max_parity: true,
+                    write_completion: crate::object_api::WriteCompletion::TailDrained,
                     http_preconditions: Some(HTTPPreconditions {
                         if_match: Some(etag),
                         ..Default::default()

crates/ecstore/src/disk/disk_store.rs

@@ -317,6 +317,22 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         dst_path: &str,
         external_guard: Option<Arc<dyn Send + Sync>>,
     ) -> Result<RenameDataResp> {
+        self.rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, external_guard)
+            .await
+            .result
+    }
+}
+
+impl LocalDiskWrapper {
+    pub(in crate::disk) async fn rename_data_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        external_guard: Option<Arc<dyn Send + Sync>>,
+    ) -> super::RenameDataObservation {
         let operation = self.clone();
         let src_volume = src_volume.to_owned();
         let src_path = src_path.to_owned();
@@ -333,22 +349,35 @@ impl DiskStoreRenameDataExt for LocalDiskWrapper {
         } else {
             get_max_timeout_duration()
         };
-        run_owned_mutation(external_guard, move || async move {
-            operation
+        let observed = run_owned_mutation(external_guard, move || async move {
+            let mut preflight_rejection = None;
+            let result = operation
                 .track_disk_health_mutation(
                     "rename_data",
                     DiskMetricMutation::Write,
                     || async {
-                        operation
-                            .disk
-                            .rename_data_borrowed(&src_volume, &src_path, &fi, &dst_volume, &dst_path)
-                            .await
+                        // Preserve the former DiskAPI future's single boxing boundary.
+                        let observed =
+                            Box::pin(
+                                operation
+                                    .disk
+                                    .rename_data_observed(&src_volume, &src_path, &fi, &dst_volume, &dst_path),
+                            )
+                            .await;
+                        preflight_rejection = observed.preflight_rejection;
+                        observed.result
                     },
                     timeout_duration,
                 )
-                .await
+                .await;
+            // Health tracking must observe the real disk error, not an Ok tuple.
+            Ok(super::RenameDataObservation {
+                result,
+                preflight_rejection,
+            })
         })
-        .await
+        .await;
+        observed.unwrap_or_else(|error| super::RenameDataObservation::unknown(Err(error)))
     }
 }
 
@@ -2588,6 +2617,46 @@ mod tests {
         assert_eq!(wrapper.metrics_snapshot().api_calls.get("unknown"), Some(&1));
     }
 
+    #[tokio::test]
+    async fn rename_preflight_evidence_preserves_health_errors_and_owned_reply() {
+        for source_exists in [false, true] {
+            for guarded in [false, true] {
+                let dir = tempfile::tempdir().expect("temp dir should be created");
+                let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be valid UTF-8"))
+                    .expect("endpoint should parse");
+                let disk = Arc::new(LocalDisk::new(&endpoint, false).await.expect("local disk should be created"));
+                if source_exists {
+                    disk.make_volume("source").await.expect("source volume should exist");
+                }
+                let wrapper = LocalDiskWrapper::new(disk, false);
+                let drops = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+                let external_guard = guarded.then(|| Arc::new(DropProbe(Arc::clone(&drops))) as Arc<dyn Send + Sync>);
+                let mut file_info = FileInfo::new("object", 1, 0);
+                file_info.mod_time = Some(::time::OffsetDateTime::now_utc());
+                file_info.erasure.index = 1;
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", external_guard)
+                    .await;
+                assert!(observed.rejected_before_publication(), "normal access rejection must carry proof");
+                assert!(matches!(observed.result, Err(DiskError::VolumeNotFound)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.api_calls.get("rename_data"), Some(&1));
+                assert_eq!(snapshot.total_writes, 0, "health tracking must not observe the rejection as Ok");
+                assert_eq!(drops.load(Ordering::SeqCst), usize::from(guarded));
+
+                wrapper.health.force_runtime_state_for_test(RuntimeDriveHealthState::Offline);
+                let observed = wrapper
+                    .rename_data_observed("source", "object", &file_info, "missing-destination", "object", None)
+                    .await;
+                assert!(!observed.rejected_before_publication(), "wrapper errors carry no local preflight proof");
+                assert!(matches!(observed.result, Err(DiskError::FaultyDisk)));
+                let snapshot = wrapper.metrics_snapshot();
+                assert_eq!(snapshot.total_errors_availability, 1);
+                assert_eq!(snapshot.total_writes, 0);
+            }
+        }
+    }
+
     #[tokio::test]
     async fn local_disk_health_wrapper_counts_returned_availability_errors() {
         let dir = tempfile::tempdir().expect("temp dir should be created");

crates/ecstore/src/disk/mod.rs

@@ -75,6 +75,25 @@ use time::OffsetDateTime;
 use tokio::io::{AsyncRead, AsyncWrite};
 use uuid::Uuid;
 
+/// Local preflight evidence stays outside DiskAPI and the RPC response format.
+pub(crate) struct RenameDataObservation {
+    pub(crate) result: Result<RenameDataResp>,
+    preflight_rejection: Option<local::LocalRenamePreflightRejection>,
+}
+
+impl RenameDataObservation {
+    fn unknown(result: Result<RenameDataResp>) -> Self {
+        Self {
+            result,
+            preflight_rejection: None,
+        }
+    }
+
+    pub(crate) fn rejected_before_publication(&self) -> bool {
+        self.result.is_err() && self.preflight_rejection.is_some()
+    }
+}
+
 const QUOTA_MUTATION_FENCE_PREFIX: &str = "tmp/quota-mutation-fences/";
 pub(crate) const QUOTA_MUTATION_FENCE_METADATA_SUFFIX: &str = "quota-mutation-fence-token";
 
@@ -711,6 +730,36 @@ impl Disk {
             .await
     }
 
+    pub(crate) async fn rename_data_borrowed_with_fence_observed(
+        &self,
+        src_volume: &str,
+        src_path: &str,
+        fi: &FileInfo,
+        dst_volume: &str,
+        dst_path: &str,
+        scanner_publication_lease_token: Option<Uuid>,
+    ) -> RenameDataObservation {
+        match self {
+            Disk::Local(local_disk) => {
+                local_disk
+                    .rename_data_observed(src_volume, src_path, fi, dst_volume, dst_path, None)
+                    .await
+            }
+            Disk::Remote(remote_disk) => RenameDataObservation::unknown(
+                remote_disk
+                    .rename_data_borrowed_with_fence(
+                        src_volume,
+                        src_path,
+                        fi,
+                        dst_volume,
+                        dst_path,
+                        scanner_publication_lease_token,
+                    )
+                    .await,
+            ),
+        }
+    }
+
     pub(crate) async fn rename_data_borrowed_with_fence(
         &self,
         src_volume: &str,

crates/ecstore/src/object_api/types.rs

@@ -870,6 +870,18 @@ impl TierFreeVersionReceiptSink {
     }
 }
 
+/// Internal PUT completion boundary; this does not change fsync or write quorum.
+#[doc(hidden)]
+#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
+pub enum WriteCompletion {
+    /// Return at write quorum when the commit owner can retain its guards.
+    #[default]
+    Quorum,
+    /// Drain the rename fan-out before returning. Minority failures still heal
+    /// after a successful quorum commit; this does not require every disk to succeed.
+    TailDrained,
+}
+
 #[derive(Default, Clone)]
 pub struct ObjectOptions {
     // Use the maximum parity (N/2), used when saving server configuration files
@@ -896,6 +908,10 @@ pub struct ObjectOptions {
     /// Persisted bucket incarnation observed before authorization.
     pub expected_bucket_incarnation_id: Option<Uuid>,
     pub no_lock: bool,
+    /// Control-plane writers that immediately read or CAS the same namespace
+    /// key use TailDrained without changing namespace lock ownership.
+    #[doc(hidden)]
+    pub write_completion: WriteCompletion,
     /// True when an upper layer already holds the object read lock before
     /// forwarding a no_lock read to the set layer.
     pub metadata_cache_safe: bool,

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

@@ -701,7 +701,7 @@ impl WarmBackend for MockWarmBackend {
         Ok(version)
     }
 
-    async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
+    async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
         self.precondition().await?;
         let barrier = self.inner.get_barrier.lock().await.take();
         if let Some(barrier) = barrier {
@@ -719,6 +719,9 @@ impl WarmBackend for MockWarmBackend {
         let Some(stored) = objects.get(object) else {
             return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found"));
         };
+        if !rv.is_empty() && stored.remote_version_id != rv {
+            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "NoSuchVersion"));
+        }
         let bytes = &stored.bytes;
 
         let start = opts.start_offset.max(0) as usize;

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

@@ -2346,6 +2346,10 @@ impl WarmBackend for SharedWarmBackendProxy {
         self.0.probe_transition_candidate(object).await
     }
 
+    async fn probe_transition_version(&self, object: &str, remote_version_id: &str) -> io::Result<TransitionCandidateProbe> {
+        self.0.probe_transition_version(object, remote_version_id).await
+    }
+
     async fn in_use(&self) -> io::Result<bool> {
         self.0.in_use().await
     }
@@ -2458,6 +2462,15 @@ impl TierOperationLease {
         Ok(())
     }
 
+    pub(crate) async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> io::Result<TransitionCandidateProbe> {
+        self.validate_remote_version_id(remote_version_id)?;
+        self.inner.driver.probe_transition_version(object, remote_version_id).await
+    }
+
     pub(crate) fn is_current_generation(&self) -> bool {
         lock_unpoisoned(&self.runtime)
             .generations

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de};
 
@@ -145,7 +143,7 @@ mod tests {
 
         assert_eq!(creds.access_key, "access");
         assert_eq!(creds.secret_key, "secret");
-        assert_eq!(creds.creds_json.as_slice(), &service_account[..]);
+        assert_eq!(creds.creds_json.as_slice(), service_account);
 
         let wire = serde_json::to_value(&creds).expect("madmin tier credentials should encode");
         assert_eq!(wire["access"], "access");
@@ -162,7 +160,7 @@ mod tests {
         .expect("the former RustFS field names and byte-array encoding should remain readable");
         assert_eq!(legacy.access_key, "legacy-access");
         assert_eq!(legacy.secret_key, "legacy-secret");
-        assert_eq!(legacy.creds_json.as_slice(), &service_account[..]);
+        assert_eq!(legacy.creds_json.as_slice(), service_account);
     }
 
     #[test]

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

@@ -460,6 +460,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -556,6 +557,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current_etag.to_string()),
                 ..Default::default()

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

@@ -494,6 +494,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_none_match: Some("*".to_string()),
                 ..Default::default()
@@ -549,6 +550,7 @@ where
         data,
         &ObjectOptions {
             max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
             http_preconditions: Some(HTTPPreconditions {
                 if_match: Some(current.record_etag.clone()),
                 ..Default::default()

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

@@ -40,6 +40,7 @@ use rustfs_s3_client::credentials::{Credentials, SignatureType, Static, Value};
 use rustfs_s3_client::transition_api::{BucketLookupType, Options, TransitionClient, TransitionCore};
 use rustfs_s3_client::{
     admin_handler_utils::AdminError,
+    api_error_response::to_error_response,
     api_put_object::{AdvancedPutOptions, PutObjectOptions},
     transition_api::{ReadCloser, ReaderImpl},
 };
@@ -48,11 +49,14 @@ use rustfs_utils::egress::validate_outbound_url;
 use rustfs_utils::http::headers::{
     CACHE_CONTROL, CONTENT_DISPOSITION, CONTENT_ENCODING, CONTENT_LANGUAGE, CONTENT_TYPE, EXPIRES, HeaderExt as _,
 };
-use s3s::dto::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ReplicationStatus};
 use s3s::header::{
     X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OBJECT_LOCK_RETAIN_UNTIL_DATE, X_AMZ_REPLICATION_STATUS,
     X_AMZ_STORAGE_CLASS,
 };
+use s3s::{
+    S3ErrorCode,
+    dto::{ObjectLockLegalHoldStatus, ObjectLockRetentionMode, ReplicationStatus},
+};
 use std::collections::HashMap;
 use std::sync::Arc;
 use std::time::Duration;
@@ -141,6 +145,42 @@ pub trait WarmBackend {
     async fn probe_transition_candidate(&self, _object: &str) -> Result<TransitionCandidateProbe, std::io::Error> {
         Ok(TransitionCandidateProbe::Unsupported)
     }
+    async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        if remote_version_id.is_empty() {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidInput,
+                "an exact tier probe requires a remote version ID",
+            ));
+        }
+        self.validate_remote_version_id(remote_version_id)?;
+        match self
+            .get(
+                object,
+                remote_version_id,
+                WarmBackendGetOpts {
+                    start_offset: 0,
+                    length: 1,
+                },
+            )
+            .await
+        {
+            Ok(_) => Ok(TransitionCandidateProbe::VersionedPresent(remote_version_id.to_string())),
+            Err(err) if matches!(to_error_response(&err).code, S3ErrorCode::InvalidRange) => {
+                Ok(TransitionCandidateProbe::VersionedPresent(remote_version_id.to_string()))
+            }
+            Err(err)
+                if err.kind() == std::io::ErrorKind::NotFound
+                    || matches!(to_error_response(&err).code, S3ErrorCode::NoSuchKey | S3ErrorCode::NoSuchVersion) =>
+            {
+                Ok(TransitionCandidateProbe::Missing)
+            }
+            Err(err) => Err(err),
+        }
+    }
     async fn in_use(&self) -> Result<bool, std::io::Error>;
 }
 
@@ -437,6 +477,17 @@ impl WarmBackend for MeteredWarmBackend {
         Self::record(TierRequestOperation::Probe, result)
     }
 
+    async fn probe_transition_version(
+        &self,
+        object: &str,
+        remote_version_id: &str,
+    ) -> Result<TransitionCandidateProbe, std::io::Error> {
+        Self::record(
+            TierRequestOperation::Probe,
+            self.inner.probe_transition_version(object, remote_version_id).await,
+        )
+    }
+
     async fn in_use(&self) -> Result<bool, std::io::Error> {
         Self::record(TierRequestOperation::InUse, self.inner.in_use().await)
     }

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::{HashMap, HashSet};
 use std::future::Future;
@@ -146,11 +144,11 @@ pub struct WarmBackendGCS {
 
 impl WarmBackendGCS {
     pub async fn new(conf: &TierGCS, tier: &str) -> Result<Self, std::io::Error> {
-        if conf.creds == "" {
+        if conf.creds.is_empty() {
             return Err(std::io::Error::other("both access and secret keys are required"));
         }
 
-        if conf.bucket == "" {
+        if conf.bucket.is_empty() {
             return Err(std::io::Error::other("no bucket name was provided"));
         }
 
@@ -195,11 +193,11 @@ impl WarmBackendGCS {
     }
 
     pub fn get_dest(&self, object: &str) -> String {
-        let mut dest_obj = object.to_string();
-        if self.prefix != "" {
-            dest_obj = format!("{}/{}", &self.prefix, object);
+        if self.prefix.is_empty() {
+            object.to_string()
+        } else {
+            format!("{}/{}", self.prefix, object)
         }
-        return dest_obj;
     }
 }
 
@@ -223,7 +221,7 @@ impl WarmBackend for WarmBackendGCS {
         let bucket = gcs_bucket_resource_name(&self.bucket);
         let Ok(res) = Box::pin(
             self.client
-                .write_object(&bucket, &self.get_dest(object), Bytes::from(d))
+                .write_object(&bucket, self.get_dest(object), Bytes::from(d))
                 .send_buffered(),
         )
         .await
@@ -240,7 +238,7 @@ impl WarmBackend for WarmBackendGCS {
 
     async fn get(&self, object: &str, rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
         let bucket = gcs_bucket_resource_name(&self.bucket);
-        let mut req = self.client.read_object(&bucket, &self.get_dest(object));
+        let mut req = self.client.read_object(&bucket, self.get_dest(object));
         let mut max_response_bytes = None;
         if let Some(generation) = parse_generation(rv)? {
             req = req.set_generation(generation);

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

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

@@ -529,6 +529,10 @@ mod tests {
                 "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
                 "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 66\r\nConnection: close\r\n\r\n<Error><Code>NoSuchObject</Code><Message>missing</Message></Error>",
                 "HTTP/1.1 403 Forbidden\r\nContent-Type: application/xml\r\nContent-Length: 65\r\nConnection: close\r\n\r\n<Error><Code>AccessDenied</Code><Message>denied</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
+                "HTTP/1.1 416 Range Not Satisfiable\r\nContent-Type: application/xml\r\nContent-Length: 72\r\nConnection: close\r\n\r\n<Error><Code>InvalidRange</Code><Message>empty version</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 67\r\nConnection: close\r\n\r\n<Error><Code>NoSuchVersion</Code><Message>missing</Message></Error>",
+                "HTTP/1.1 404 Not Found\r\nContent-Type: application/xml\r\nContent-Length: 63\r\nConnection: close\r\n\r\n<Error><Code>NoSuchKey</Code><Message>missing</Message></Error>",
             ];
             let mut requests = Vec::new();
             for response in responses {
@@ -622,15 +626,52 @@ mod tests {
             .await
             .expect_err("an authorization failure must not be mistaken for a missing key");
         assert_eq!(to_error_response(&err).code, S3ErrorCode::AccessDenied);
+        assert_eq!(
+            backend
+                .probe_transition_candidate("delete-marker-hidden")
+                .await
+                .expect("a current delete marker should hide the data version"),
+            TransitionCandidateProbe::Missing
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("delete-marker-hidden", "historical-version")
+                .await
+                .expect("the stored historical version should be probed exactly"),
+            TransitionCandidateProbe::VersionedPresent("historical-version".to_string())
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("delete-marker-hidden", "missing-version")
+                .await
+                .expect("a missing exact version should be classified"),
+            TransitionCandidateProbe::Missing
+        );
+        assert_eq!(
+            backend
+                .probe_transition_version("missing-object", "historical-version")
+                .await
+                .expect("a missing key for an exact version probe should be classified"),
+            TransitionCandidateProbe::Missing
+        );
 
         let requests = fixture.await.expect("candidate fixture should join");
-        for request in requests {
+        for request in &requests[..6] {
             let request = request.to_ascii_lowercase();
             assert!(request.starts_with("get /bucket/"), "candidate discovery must use object GET");
             assert!(request.contains("\r\nrange: bytes=0-0\r\n"));
             assert!(!request.contains("?versioning"));
             assert!(!request.contains("?versions"));
         }
+        for request in &requests[6..] {
+            let request = request.to_ascii_lowercase();
+            assert!(request.starts_with("get /bucket/"), "exact discovery must use object GET");
+            assert!(request.contains("\r\nrange: bytes=0-0\r\n"));
+        }
+        assert!(!requests[5].to_ascii_lowercase().contains("versionid="));
+        assert!(requests[6].to_ascii_lowercase().contains("?versionid=historical-version"));
+        assert!(requests[7].to_ascii_lowercase().contains("?versionid=missing-version"));
+        assert!(requests[8].to_ascii_lowercase().contains("?versionid=historical-version"));
     }
 
     fn list_versions(versions: &[(&str, &str)], delete_markers: &[(&str, &str)], is_truncated: bool) -> ListVersionsResult {

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

@@ -15,8 +15,6 @@
 #![allow(unused_variables)]
 #![allow(unused_mut)]
 #![allow(unused_assignments)]
-#![allow(unused_must_use)]
-#![allow(clippy::all)]
 
 use std::collections::HashMap;
 

crates/ecstore/src/set_disk/mod.rs

@@ -876,7 +876,7 @@ pub use ops::multipart::{MultipartCommitBarrier, MultipartCommitPause};
 pub(crate) use ops::object::DeleteObjectCommitBarrier;
 #[cfg(any(test, feature = "test-util"))]
 pub(crate) use ops::object::TransitionCleanupStoreBarrier as SetDiskTransitionCleanupStoreBarrier;
-#[cfg(test)]
+#[cfg(all(test, feature = "test-util"))]
 pub(crate) use ops::object::TransitionUploadedCommitBarrier as SetDiskTransitionUploadedCommitBarrier;
 pub(crate) use ops::object::body_cache_plaintext_len;
 #[cfg(all(test, feature = "test-util"))]

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

@@ -299,11 +299,11 @@ use crate::error::is_err_invalid_upload_id;
 use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed};
 use crate::object_api::{
     NamespaceLockFence, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode,
-    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY,
+    SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, WriteCompletion,
 };
 use crate::services::notification_sys::RemoteVersionStateFleetProofToken;
 use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata};
-use crate::set_disk::core::io_primitives::{RenameTailCleanup, finish_rename_tail_heal};
+use crate::set_disk::core::io_primitives::{RenameRollbackReceipt, RenameTailCleanup, finish_rename_tail_heal};
 #[cfg(test)]
 use crate::storage_api_contracts::namespace::NamespaceLocking;
 #[cfg(test)]
@@ -3548,6 +3548,7 @@ impl SetDisks {
                 (None, None, None)
             };
         let mut tmp_cleanup_owned = false;
+        let rollback_receipt = RenameRollbackReceipt::default();
         let operation = async {
             let erasure = Arc::new(erasure_from_file_info(&fi, false)?);
 
@@ -4256,6 +4257,7 @@ impl SetDisks {
             let commit_bucket = bucket.to_owned();
             let commit_object = object.to_owned();
             let commit_tmp_dir = tmp_dir.clone();
+            let commit_rollback_receipt = rollback_receipt.clone();
             let commit_object_lock_guard = object_lock_guard.take();
             let commit_decommission_object_lock_guard = decommission_object_lock_guard.take();
             let commit_publication_guard = publication_commit_guard.take();
@@ -4266,13 +4268,17 @@ impl SetDisks {
             // complete rename fan-out drains. Keep this path synchronous so
             // its terminal state is known before the coordinator releases
             // remote leases.
-            let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation())
-                && (commit_object_lock_guard.is_some()
-                    || commit_decommission_object_lock_guard.is_some()
-                    || commit_publication_guard.is_some())
+            let commit_owns_namespace_guard = commit_object_lock_guard.is_some()
+                || commit_decommission_object_lock_guard.is_some()
+                || commit_publication_guard.is_some();
+            let commit_allows_early_ack = opts.write_completion == WriteCompletion::Quorum
+                && !(opts.data_movement && opts.has_decommission_capacity_reservation())
+                && commit_owns_namespace_guard
                 && commit_scanner_publication_scope.is_none();
+            // Full-tail callers also transfer owned guards to the coordinator:
+            // cancelling their ACK waiter must not cancel an in-flight rename.
             let detach_commit_owner = commit_scanner_publication_scope.is_some()
-                || commit_allows_early_ack
+                || commit_owns_namespace_guard
                 || commit_bucket_lifecycle_guard.is_some()
                 || quota_mutation_fence;
             let commit_write_path_label = write_path.metric_label();
@@ -4452,7 +4458,8 @@ impl SetDisks {
                         write_quorum,
                         commit_scanner_publication_lease_tokens.as_ref(),
                     )
-                    .with_publication_scope(commit_scanner_publication_scope.clone()),
+                    .with_publication_scope(commit_scanner_publication_scope.clone())
+                    .with_rollback_receipt(commit_rollback_receipt.clone()),
                 )
                 .await;
                 if let Some(scope) = commit_scanner_publication_scope.as_ref() {
@@ -4585,6 +4592,11 @@ impl SetDisks {
                 let rename_commit = match rename_result {
                     Ok(commit) => commit,
                     Err(err) => {
+                        if commit_rollback_receipt.is_incomplete() {
+                            // Incomplete undo retains the staging source and
+                            // rollback backup for recovery; cleanup is unsafe.
+                            return Err(err.into());
+                        }
                         if let Err(cleanup_err) = commit_set.delete_all(RUSTFS_META_TMP_BUCKET, &commit_tmp_dir).await {
                             warn!(tmp_dir = %commit_tmp_dir, error = ?cleanup_err, "failed to cleanup put_object temporary data");
                         } else if issue3031_diag_enabled() {
@@ -4617,9 +4629,8 @@ impl SetDisks {
                     request.object_version_id = committed_version_id
                         .or_else(|| commit_version_suspended.then(Uuid::nil))
                         .map(|version_id| version_id.to_string());
-                    tokio::spawn(async move {
-                        let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await;
-                    });
+                    let heal_set = commit_set.clone();
+                    tokio::spawn(async move { heal_set.submit_rename_tail_heal(request).await });
                 }
 
                 let rename_stage_elapsed = rename_stage_start.elapsed();
@@ -4885,7 +4896,7 @@ impl SetDisks {
                     );
                 }
             });
-        } else {
+        } else if !rollback_receipt.is_incomplete() {
             // Failure path (quorum loss / rollback): keep the cleanup inline so
             // a failed PUT never returns while its tmp shards are still on disk
             // (state-residue hardening tracked by backlog#864 / backlog#898).
@@ -17494,27 +17505,69 @@ mod put_object_tmp_cleanup_tests {
     }
 
     #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
     async fn put_object_failure_cleans_tmp_workspace_inline() {
-        let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                let (temp_dirs, _disk_stores, set_disks) = hermetic_set_disks(4).await;
+                let bucket = "tmp-clean-missing-bucket";
+                let object = "orphan-object";
+                let barrier = PutObjectCommitBarrier::install(bucket, object, PutObjectCommitPause::BeforeNamespace);
+                let writer = Arc::clone(&set_disks);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("missing-bucket PUT must stage before rename");
+                let staged = non_trash_tmp_entries(&temp_dirs).await;
+                assert_eq!(staged.len(), 4, "every disk must have a staged workspace before rejection");
+                for workspace in staged {
+                    let mut entries = tokio::fs::read_dir(&workspace)
+                        .await
+                        .expect("staged workspace should be readable");
+                    let mut shards = 0;
+                    while let Some(entry) = entries.next_entry().await.expect("staged data directory should be readable") {
+                        if entry.file_type().await.expect("staged entry type").is_dir() {
+                            let part = tokio::fs::metadata(entry.path().join("part.1"))
+                                .await
+                                .expect("staging must contain an actual erasure shard");
+                            assert!(part.len() > 0, "the shard must be written before the missing-bucket failure");
+                            shards += 1;
+                        }
+                    }
+                    assert_eq!(shards, 1);
+                }
+                assert!(temp_dirs.iter().all(|dir| !dir.path().join(bucket).exists()));
+                barrier.release();
+                let err = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("missing-bucket PUT must finish")
+                    .expect("PUT task should join")
+                    .expect_err("put_object into a missing bucket volume must fail");
+                assert!(matches!(err, StorageError::VolumeNotFound), "original disk error expected: {err}");
 
-        // The bucket volume is never created, so the shards are written into
-        // the tmp workspace and the commit fails at rename_data with a quorum
-        // error — exercising the failure-path cleanup.
-        let mut reader = PutObjReader::from_vec(vec![9u8; TEST_OBJECT_SIZE]);
-        let err = set_disks
-            .put_object("tmp-clean-missing-bucket", "orphan-object", &mut reader, &ObjectOptions::default())
-            .await
-            .expect_err("put_object into a missing bucket volume must fail");
-
-        // No polling: the failure path must clean the tmp workspace inline,
-        // before put_object returns (backlog#864 / backlog#898 hardening).
-        let leftovers = non_trash_tmp_entries(&temp_dirs).await;
-        assert!(
-            leftovers.is_empty(),
-            "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
-        );
-
-        drop(temp_dirs);
+                // No polling: known pre-publication rejection must clean staging
+                // inline, before PUT returns (backlog#864 / backlog#898).
+                let leftovers = non_trash_tmp_entries(&temp_dirs).await;
+                assert!(
+                    leftovers.is_empty(),
+                    "failed PUT must not leave tmp shards behind, leftovers: {leftovers:?}, err: {err}"
+                );
+            }
+        })
+        .await;
     }
 
     #[tokio::test]
@@ -18157,6 +18210,354 @@ mod put_object_tmp_cleanup_tests {
         .await;
     }
 
+    async fn make_completion_test_bucket(disks: &[DiskStore], bucket: &str) {
+        for disk in disks {
+            disk.make_volume(bucket)
+                .await
+                .expect("completion test bucket should be created");
+        }
+    }
+
+    /// Observe the actual metadata quorum while the remaining rename is parked.
+    /// A completed task count alone can race tasks that have not started yet.
+    async fn wait_for_paused_tail_metadata_quorum(disks: &[DiskStore], bucket: &str, object: &str) {
+        tokio::time::timeout(Duration::from_secs(30), async {
+            loop {
+                let mut committed = 0;
+                for disk in disks {
+                    match disk.read_version("", bucket, object, "", &ReadOptions::default()).await {
+                        Ok(_) => committed += 1,
+                        Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {}
+                        Err(err) => panic!("unexpected metadata error while observing {bucket}/{object}: {err}"),
+                    }
+                }
+                if committed == 3 {
+                    break;
+                }
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("three disks must publish metadata while the fourth rename remains paused");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_waits_for_tail_and_allows_immediate_cas() {
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for size in [4096, 1024 * 1024] {
+                let (_dirs, disks, set) = hermetic_set_disks(4).await;
+                let bucket = "put-full-tail-cas";
+                let object = "full-tail-cas-object";
+                make_completion_test_bucket(&disks, bucket).await;
+                let tasks = rename_fanout_barrier::observe_tasks(object);
+                let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                let writer = Arc::clone(&set);
+                let put = tokio::spawn(async move {
+                    let mut reader = PutObjReader::from_vec(vec![b'1'; size]);
+                    writer
+                        .put_object(
+                            bucket,
+                            object,
+                            &mut reader,
+                            &ObjectOptions {
+                                write_completion: WriteCompletion::TailDrained,
+                                ..Default::default()
+                            },
+                        )
+                        .await
+                });
+                tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                    .await
+                    .expect("full-tail PUT must reach the rename barrier");
+                wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+                assert!(!put.is_finished(), "full-tail PUT must remain pending after metadata quorum");
+                let mut lock_probe = Box::pin(set.acquire_write_lock_diag("full_tail_probe", bucket, object));
+                assert!(
+                    futures::poll!(lock_probe.as_mut()).is_pending(),
+                    "the owned namespace guard must remain held"
+                );
+                barrier.release();
+                let written = tokio::time::timeout(Duration::from_secs(30), put)
+                    .await
+                    .expect("full-tail PUT should finish after release")
+                    .expect("full-tail PUT task should join")
+                    .expect("full-tail PUT must commit");
+                assert_eq!(tasks.running(), 0, "full-tail response must follow every rename task");
+                drop(
+                    tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                        .await
+                        .expect("same-key lock should be available on return")
+                        .expect("same-key lock probe should succeed"),
+                );
+                for disk in &disks {
+                    disk.read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("successful full-tail PUT must publish on every healthy disk");
+                }
+                drop(barrier);
+                let mut replacement = PutObjReader::from_vec(b"cas successor".to_vec());
+                set.put_object(
+                    bucket,
+                    object,
+                    &mut replacement,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        http_preconditions: Some(HTTPPreconditions {
+                            if_match: written.etag,
+                            ..Default::default()
+                        }),
+                        ..Default::default()
+                    },
+                )
+                .await
+                .expect("immediate same-key CAS must acquire the namespace guard");
+                let mut read = set
+                    .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                    .await
+                    .expect("CAS successor must be immediately readable");
+                let mut body = Vec::new();
+                read.stream.read_to_end(&mut body).await.expect("successor body must drain");
+                assert_eq!(body, b"cas successor");
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_preserves_quorum_success_and_heals_failed_tail() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-heal";
+        let object = "full-tail-heal-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let mut heals = set.capture_test_rename_tail_heals();
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0]);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("failed tail must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        assert!(!put.is_finished(), "committed quorum must still wait for the failing tail");
+        barrier.release();
+        tokio::time::timeout(Duration::from_secs(30), put)
+            .await
+            .expect("failed tail should drain")
+            .expect("PUT task should join")
+            .expect("a minority tail error must not negate committed quorum");
+        assert_eq!(tasks.running(), 0);
+        let heal = tokio::time::timeout(Duration::from_secs(30), heals.recv())
+            .await
+            .expect("failed tail must schedule heal")
+            .expect("heal capture must remain connected");
+        assert_eq!(heal.bucket, bucket);
+        assert_eq!(heal.object_prefix.as_deref(), Some(object));
+        let info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("committed object must remain readable despite the failed tail");
+        assert_eq!(info.size, TEST_OBJECT_SIZE as i64);
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_rejects_quorum_minus_one() {
+        let (_dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = "put-full-tail-no-quorum";
+        let object = "full-tail-no-quorum-object";
+        make_completion_test_bucket(&disks, bucket).await;
+        let _fault = rename_fault_injection::fail_rename_on(object, &[0, 1]);
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+        let err = set
+            .put_object(
+                bucket,
+                object,
+                &mut reader,
+                &ObjectOptions {
+                    write_completion: WriteCompletion::TailDrained,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect_err("draining two successful disks cannot satisfy write quorum three");
+        assert!(
+            matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+            "original quorum error expected: {err}"
+        );
+        assert_eq!(tasks.running(), 0, "failed fan-out and rollback must complete before return");
+        assert!(
+            set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(),
+            "failed fresh write must not become visible"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn put_incomplete_rollback_preserves_staging_and_old_version_backup() {
+        use crate::set_disk::core::io_primitives::rollback_fault_injection;
+
+        temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async {
+            for write_completion in [WriteCompletion::Quorum, WriteCompletion::TailDrained] {
+                for fault in [
+                    rollback_fault_injection::Fault::Io,
+                    rollback_fault_injection::Fault::VolumeNotFoundAfterRename,
+                ] {
+                    let (dirs, disks, set) = hermetic_set_disks(4).await;
+                    let bucket = "put-incomplete-undo";
+                    let object = "incomplete-undo-object";
+                    make_completion_test_bucket(&disks, bucket).await;
+                    let mut old_reader = PutObjReader::from_vec(vec![b'0'; TEST_OBJECT_SIZE]);
+                    set.put_object(
+                        bucket,
+                        object,
+                        &mut old_reader,
+                        &ObjectOptions {
+                            write_completion: WriteCompletion::TailDrained,
+                            ..Default::default()
+                        },
+                    )
+                    .await
+                    .expect("old generation should be completely committed");
+                    wait_for_tmp_workspace_to_drain(&dirs, "old PUT must leave no unrelated staging").await;
+                    let old = disks[0]
+                        .read_version("", bucket, object, "", &ReadOptions::default())
+                        .await
+                        .expect("old metadata must be readable");
+                    let old_data_dir = old.data_dir.expect("non-inline old version needs a data directory");
+                    let tasks = rename_fanout_barrier::observe_tasks(object);
+                    let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+                    let _rename_fault = rename_fault_injection::fail_rename_on(object, &[2, 3]);
+                    let _undo_fault = rollback_fault_injection::arm(object, 0, fault);
+                    let writer = Arc::clone(&set);
+                    let put = tokio::spawn(async move {
+                        let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+                        writer
+                            .put_object(
+                                bucket,
+                                object,
+                                &mut reader,
+                                &ObjectOptions {
+                                    write_completion,
+                                    ..Default::default()
+                                },
+                            )
+                            .await
+                    });
+                    tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+                        .await
+                        .expect("overwrite must enter the actual rename fan-out before failure injection");
+                    barrier.release();
+                    let err = tokio::time::timeout(Duration::from_secs(30), put)
+                        .await
+                        .expect("incomplete undo must return without hanging")
+                        .expect("PUT task should join")
+                        .expect_err("two renamed disks cannot satisfy write quorum three");
+                    assert!(
+                        matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)),
+                        "original quorum error expected: {err}"
+                    );
+                    assert_eq!(tasks.running(), 0, "every rename and undo task must be reaped before return");
+                    let leftovers = non_trash_tmp_entries(&dirs).await;
+                    assert!(!leftovers.is_empty(), "incomplete undo must retain the new staging source for recovery");
+                    let backups = dirs
+                        .iter()
+                        .filter(|dir| {
+                            dir.path()
+                                .join(bucket)
+                                .join(object)
+                                .join(old_data_dir.to_string())
+                                .join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
+                                .exists()
+                        })
+                        .count();
+                    assert_eq!(backups, 1, "exactly the failed undo disk must retain its old-version backup");
+                    // The remaining three disks still serve the old generation;
+                    // the failed minority must never become an acknowledged write.
+                    let mut read = set
+                        .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default())
+                        .await
+                        .expect("old generation must remain readable after incomplete rollback");
+                    let mut body = Vec::new();
+                    read.stream
+                        .read_to_end(&mut body)
+                        .await
+                        .expect("old generation should stream");
+                    assert_eq!(body, vec![b'0'; TEST_OBJECT_SIZE]);
+                }
+            }
+        })
+        .await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial(capacity_dirty_scope)]
+    async fn tail_drained_put_owned_commit_survives_waiter_cancellation() {
+        let (dirs, disks, set) = hermetic_set_disks(4).await;
+        let bucket = RUSTFS_META_BUCKET;
+        let object = "full-tail-cancelled-receipt";
+        // Internal config writes do not own a bucket lifecycle guard. The object
+        // guard alone must keep the full-tail coordinator alive after cancellation.
+        let tasks = rename_fanout_barrier::observe_tasks(object);
+        let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME);
+        let writer = Arc::clone(&set);
+        let put = tokio::spawn(async move {
+            let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]);
+            writer
+                .put_object(
+                    bucket,
+                    object,
+                    &mut reader,
+                    &ObjectOptions {
+                        write_completion: WriteCompletion::TailDrained,
+                        ..Default::default()
+                    },
+                )
+                .await
+        });
+        tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused())
+            .await
+            .expect("cancelled receipt must first reach the rename barrier");
+        wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await;
+        put.abort();
+        assert!(put.await.expect_err("ACK waiter should cancel").is_cancelled());
+        let mut lock_probe = Box::pin(set.acquire_write_lock_diag("cancelled_full_tail_probe", bucket, object));
+        assert!(
+            futures::poll!(lock_probe.as_mut()).is_pending(),
+            "owned coordinator must retain the namespace guard after waiter cancellation"
+        );
+        barrier.release();
+        drop(
+            tokio::time::timeout(Duration::from_secs(30), lock_probe)
+                .await
+                .expect("cancelled coordinator must eventually release its guard")
+                .expect("post-commit lock probe should succeed"),
+        );
+        assert_eq!(tasks.running(), 0, "cancelled coordinator must reap every rename task");
+        for disk in &disks {
+            disk.read_version("", bucket, object, "", &ReadOptions::default())
+                .await
+                .expect("caller cancellation must not interrupt committed receipt materialization");
+        }
+        wait_for_tmp_workspace_to_drain(&dirs, "cancelled full-tail commit should release staging ownership").await;
+    }
+
     #[tokio::test]
     #[serial_test::serial(capacity_dirty_scope)]
     async fn no_lock_put_waits_for_rename_tail_under_outer_guard() {
@@ -18184,6 +18585,7 @@ mod put_object_tmp_cleanup_tests {
                         &mut reader,
                         &ObjectOptions {
                             no_lock: true,
+                            write_completion: WriteCompletion::TailDrained,
                             ..Default::default()
                         },
                     )
@@ -18209,7 +18611,18 @@ mod put_object_tmp_cleanup_tests {
             put.await
                 .expect("no-lock PUT task should join")
                 .expect("no-lock PUT should commit after the rename tail releases");
+            let mut lock_probe = Box::pin(set_disks.acquire_write_lock_diag("borrowed_full_tail_probe", bucket, object));
+            assert!(
+                futures::poll!(lock_probe.as_mut()).is_pending(),
+                "full-tail PUT must not release the caller's outer guard"
+            );
             drop(outer_guard);
+            drop(
+                tokio::time::timeout(Duration::from_secs(5), lock_probe)
+                    .await
+                    .expect("outer owner releasing its guard should unblock the probe")
+                    .expect("post-outer-guard probe should succeed"),
+            );
         })
         .await;
     }

crates/ecstore/src/set_disk/transition_matrix_tests.rs

@@ -18,6 +18,7 @@ use super::{
 };
 use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time};
 use crate::ecstore_validation_blackbox::make_local_set_disks;
+use crate::object_api::WriteCompletion;
 use crate::services::tier::test_util::register_mock_tier;
 use crate::storage_api_contracts::bucket::BucketOperations;
 use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _};
@@ -72,7 +73,7 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() {
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -185,7 +186,7 @@ async fn prepared_snapshot_transition_duplicate_and_late_get_use_committed_remot
             object,
             &mut reader,
             &ObjectOptions {
-                no_lock: true,
+                write_completion: WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )

crates/ecstore/src/store/init.rs

@@ -2979,6 +2979,33 @@ mod tests {
     #[cfg(feature = "test-util")]
     const DECOMMISSION_TEST_FAULT_STAGE_TIERED: &str = "decommission_tiered_object";
 
+    fn decommission_retry_fault_hook(
+        bucket: &str,
+        object: &str,
+        faults: Arc<AtomicUsize>,
+    ) -> crate::core::pools::DecommissionTestFaultDecision {
+        let target_bucket = bucket.to_string();
+        let target_object = object.to_string();
+        Arc::new(move |stage, bucket, object, _attempt, succeeded| {
+            if !succeeded
+                || stage != DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
+                || bucket != target_bucket
+                || object != target_object
+            {
+                return false;
+            }
+
+            // Entry retries reset the local attempt; real copy errors can skip
+            // successful attempts. Only injected faults spend this global budget.
+            faults
+                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
+                    (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1))
+                        .then_some(faults.saturating_add(1))
+                })
+                .is_ok()
+        })
+    }
+
     async fn seed_decommission_source(
         store: &Arc<crate::store::ECStore>,
         bucket: &str,
@@ -5120,6 +5147,33 @@ mod tests {
         shutdown.cancel();
     }
 
+    #[test]
+    fn decommission_retry_fault_budget_counts_successes_across_attempt_changes() {
+        for attempts in [[1, 2, 3], [1, 1, 2], [1, 3, 3]] {
+            let faults = Arc::new(AtomicUsize::new(0));
+            let hook = decommission_retry_fault_hook("bucket", "object", Arc::clone(&faults));
+
+            for (stage, bucket, object, succeeded) in [
+                ("other-stage", "bucket", "object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "other-bucket", "object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "other-object", true),
+                (DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", false),
+            ] {
+                assert!(!hook(stage, bucket, object, 1, succeeded));
+            }
+            assert_eq!(faults.load(Ordering::SeqCst), 0, "unrelated or failed copies must not consume faults");
+
+            for (index, attempt) in attempts.into_iter().enumerate() {
+                assert_eq!(
+                    hook(DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT, "bucket", "object", attempt, true),
+                    index < 2,
+                    "attempts={attempts:?}, index={index}"
+                );
+            }
+            assert_eq!(faults.load(Ordering::SeqCst), 2, "attempts={attempts:?}");
+        }
+    }
+
     #[test]
     #[serial_test::serial(storage_class_env)]
     fn decommission_entry_retries_source_changed_without_canceling_other_bucket() {
@@ -5214,31 +5268,8 @@ mod tests {
                     ));
 
                     let ordinary_faults = Arc::new(AtomicUsize::new(0));
-                    let ordinary_faults_for_hook = Arc::clone(&ordinary_faults);
-                    let fault_bucket = other_bucket.clone();
-                    let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(Arc::new(
-                        move |stage, bucket, object, attempt, succeeded| {
-                            let candidate = succeeded
-                                && stage == DECOMMISSION_TEST_FAULT_STAGE_MIGRATE_OBJECT
-                                && bucket == fault_bucket.as_str()
-                                && object == other_object;
-                            if !candidate {
-                                return false;
-                            }
-
-                            // Keep the fault budget global across any
-                            // entry-level re-list; its inner attempt counter
-                            // restarts after SourceChanged.
-                            ordinary_faults_for_hook
-                                .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |faults| {
-                                    let next_fault = faults.saturating_add(1);
-                                    (faults < crate::core::pools::DECOMMISSION_VERSION_COPY_ATTEMPTS.saturating_sub(1)
-                                        && attempt == next_fault)
-                                        .then_some(next_fault)
-                                })
-                                .is_ok()
-                        },
-                    ));
+                    let fault_hook = decommission_retry_fault_hook(&other_bucket, other_object, Arc::clone(&ordinary_faults));
+                    let _fault_guard = crate::core::pools::DecommissionTestFaultGuard::install(fault_hook);
 
                     let rx = CancellationToken::new();
                     let source_changed_exhaustions = Arc::new(AtomicUsize::new(0));
@@ -8045,10 +8076,15 @@ mod tests {
         );
         assert!(com::read_config(store.pools[0].clone(), &second_page_path).await.is_ok());
 
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone())
+        let full_tail = ObjectOptions {
+            max_parity: true,
+            write_completion: crate::object_api::WriteCompletion::TailDrained,
+            ..Default::default()
+        };
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone(), &full_tail)
             .await
             .expect("second page receipt should restore");
-        com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec())
+        com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec(), &full_tail)
             .await
             .expect("second page receipt should corrupt deterministically");
         let corrupt = store
@@ -11575,6 +11611,7 @@ mod tests {
         pool_index: usize,
         bucket: &str,
         object: &str,
+        minio_unversioned: bool,
     ) {
         for disk_index in 0..4 {
             let metadata_path =
@@ -11608,6 +11645,11 @@ mod tests {
                 ] {
                     rustfs_utils::http::metadata_compat::remove_bytes(&mut object_meta.meta_sys, suffix);
                 }
+                if minio_unversioned {
+                    object_meta
+                        .meta_sys
+                        .insert("x-minio-internal-transitioned-versionID".to_string(), Vec::new());
+                }
                 *shallow = rustfs_filemeta::FileMetaShallowVersion::try_from(version)
                     .expect("legacy transitioned version should re-encode");
             }
@@ -11618,6 +11660,152 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "test-util")]
+    async fn read_store_body(
+        store: &Arc<crate::store::ECStore>,
+        bucket: &str,
+        object: &str,
+        range: Option<HTTPRangeSpec>,
+        opts: &ObjectOptions,
+    ) -> Vec<u8> {
+        let mut reader = store
+            .get_object_reader(bucket, object, range, HeaderMap::new(), opts)
+            .await
+            .expect("object reader should open");
+        let mut body = Vec::new();
+        reader.stream.read_to_end(&mut body).await.expect("object body should drain");
+        body
+    }
+
+    #[cfg(feature = "test-util")]
+    #[tokio::test]
+    #[serial_test::serial(storage_class_env)]
+    async fn legacy_unknown_unversioned_transition_supports_head_get_and_range_without_backfill() {
+        let temp_dir = tempfile::tempdir().expect("create legacy unknown unversioned store dir");
+        let (ctx, store, _shutdown) =
+            without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-unknown-unversioned-read", &[4])).await;
+        crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+        let tier_name = "LEGACY-UNKNOWN-UNVERSIONED-READ";
+        let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
+        backend.set_put_remote_version(Some(String::new())).await;
+        let bucket = "legacy-unknown-unversioned-read-bucket";
+        let object = "object.bin";
+        let payload = b"legacy unversioned remote tier object remains readable".repeat(1024);
+        store
+            .make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("legacy source bucket should be created");
+        let mut reader = PutObjReader::from_vec(payload.clone());
+        let source = store
+            .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("legacy source should be written");
+        store
+            .transition_object(
+                bucket,
+                object,
+                &ObjectOptions {
+                    transition: TransitionOptions {
+                        status: TRANSITION_PENDING.to_string(),
+                        tier: tier_name.to_string(),
+                        etag: source.etag.clone().expect("legacy source should have an etag"),
+                        ..Default::default()
+                    },
+                    mod_time: source.mod_time,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("legacy source should transition");
+        rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 0, bucket, object, true).await;
+        backend.clear_op_log().await;
+
+        let opts = ObjectOptions {
+            metadata_cache_safe: false,
+            ..Default::default()
+        };
+        let head = store
+            .get_object_info(bucket, object, &opts)
+            .await
+            .expect("legacy transitioned HEAD should use local metadata");
+        assert_eq!(head.transition_version_state, rustfs_filemeta::TransitionVersionState::Unknown);
+        assert!(head.transitioned_object.version_id.is_empty());
+        assert_eq!(
+            head.user_defined
+                .get("x-minio-internal-transitioned-versionID")
+                .map(String::as_str),
+            Some(""),
+            "the MinIO empty version-key provenance must survive xl.meta decoding"
+        );
+        assert!(
+            !rustfs_utils::http::metadata_compat::contains_key_str(
+                &head.user_defined,
+                rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+            ),
+            "the compatibility read must not synthesize version-state metadata"
+        );
+
+        let full_body = read_store_body(&store, bucket, object, None, &opts).await;
+        assert_eq!(full_body, payload);
+
+        let range = HTTPRangeSpec {
+            is_suffix_length: false,
+            start: 7,
+            end: 38,
+        };
+        let ranged_body = read_store_body(&store, bucket, object, Some(range), &opts).await;
+        assert_eq!(ranged_body, &payload[7..=38]);
+
+        let after_read = store.pools[0]
+            .get_disks_by_key(object)
+            .load_file_info_versions_exact(bucket, object)
+            .await
+            .expect("legacy metadata should remain readable after GET")
+            .expect("legacy object metadata should remain on disk")
+            .versions
+            .into_iter()
+            .find(|version| version.transition_status == rustfs_filemeta::TRANSITION_COMPLETE)
+            .expect("legacy transitioned source should remain visible after GET");
+        assert_eq!(after_read.transition_version_state, rustfs_filemeta::TransitionVersionState::Unknown);
+        assert!(after_read.transition_version.is_none());
+        assert!(after_read.transition_version_id.is_none());
+        assert_eq!(
+            after_read
+                .metadata
+                .get("x-minio-internal-transitioned-versionID")
+                .map(String::as_str),
+            Some(""),
+            "the MinIO empty version-key provenance must remain after GET and Range GET"
+        );
+        assert!(
+            !rustfs_utils::http::metadata_compat::contains_key_str(
+                &after_read.metadata,
+                rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+            ),
+            "the compatibility read must remain side-effect free"
+        );
+
+        assert_eq!(
+            backend.op_log().await,
+            vec![
+                MockWarmOp::Probe {
+                    object: after_read.transitioned_objname.clone(),
+                },
+                MockWarmOp::Get {
+                    object: after_read.transitioned_objname.clone(),
+                },
+                MockWarmOp::Probe {
+                    object: after_read.transitioned_objname.clone(),
+                },
+                MockWarmOp::Get {
+                    object: after_read.transitioned_objname,
+                },
+            ],
+            "legacy reads should probe before each unversioned GET and never mutate local metadata"
+        );
+        assert_eq!(backend.remove_count().await, 0);
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]
@@ -11658,7 +11846,7 @@ mod tests {
                 )
                 .await
                 .expect("legacy source should transition");
-            rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 0, bucket, object).await;
+            rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 0, bucket, object, false).await;
             let legacy = store.pools[0]
                 .get_disks_by_key(object)
                 .load_file_info_versions_exact(bucket, object)
@@ -12799,7 +12987,7 @@ mod tests {
                 .expect("merge-loser source should transition");
             copy_test_xlmeta_between_pools(temp_dir.path(), 0, 1, bucket, object).await;
         }
-        rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 1, bucket, "legacy/item.bin").await;
+        rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 1, bucket, "legacy/item.bin", false).await;
         backend.set_remove_failure(true);
         store.pools[1]
             .delete_object(bucket, "hidden/item.bin", ObjectOptions::default())
@@ -16866,6 +17054,10 @@ mod tests {
                             .find(|version| version.version_id == history.version_id)
                             .expect("transitioned history should exist");
                         transitioned.transition_version_state = rustfs_filemeta::TransitionVersionState::Unknown;
+                        rustfs_utils::http::metadata_compat::remove_str(
+                            &mut transitioned.metadata,
+                            rustfs_utils::http::metadata_compat::SUFFIX_TRANSITIONED_VERSION_STATE,
+                        );
                         metadata
                             .add_version(transitioned)
                             .expect("unknown state should replace the transitioned version");

crates/ecstore/src/store/mod.rs

@@ -425,7 +425,7 @@ pub(crate) mod init_format;
 pub(crate) mod list_objects;
 mod multipart;
 mod object;
-#[cfg(any(test, feature = "test-util"))]
+#[cfg(feature = "test-util")]
 pub use object::DeleteAfterObjectLockSnapshotBarrier;
 pub(crate) use object::{
     DecommissionFixedReadAnchor, ObjectLockDiagGuard, RemoteTuplePublicationCommitGuard, RemoteTuplePublicationFence,