feat(ilm): inspect legacy transition version state (#7581)

crates/ecstore/src/api/mod.rs

@@ -38,6 +38,16 @@ pub mod bucket {
     }
 
     pub mod lifecycle {
+        pub mod legacy_transition_state_reconcile {
+            pub use crate::bucket::lifecycle::legacy_transition_state_reconcile::{
+                LegacyTransitionStateCopyRepresentation, LegacyTransitionStateMetadataAlias, LegacyTransitionStateReconcileError,
+                LegacyTransitionStateReconcileOutcome, LegacyTransitionStateReconcileReadiness,
+                LegacyTransitionStateReconcileRequest, LegacyTransitionStateReconcileResponse,
+                LegacyTransitionStateReconcileSelector, LegacyTransitionStateSetRepresentation, LegacyTransitionStateSource,
+                LegacyTransitionStateTarget,
+            };
+        }
+
         pub mod bucket_lifecycle_audit {
             pub use crate::bucket::lifecycle::bucket_lifecycle_audit::LcEventSrc;
         }

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

@@ -18,6 +18,7 @@ mod config_boundary;
 pub mod core;
 mod durable_namespace;
 pub mod evaluator;
+pub mod legacy_transition_state_reconcile;
 pub mod manual_transition_job;
 mod metadata_boundary;
 pub(crate) use metadata_boundary::{LifecycleExpiryConfigs, get_expiry_configs, get_lifecycle_config};

crates/ecstore/src/services/notification_sys.rs

@@ -568,6 +568,10 @@ pub(crate) fn tier_delete_journal_topology_generation(proof: &TierDeleteJournalF
     stable_tier_delete_journal_topology_generation(&proof.token.topology_fingerprint)
 }
 
+pub(crate) fn cross_pool_fence_topology_generation(proof: &CrossPoolFenceFleetProofToken) -> String {
+    stable_tier_delete_journal_topology_generation(&proof.0.topology_fingerprint)
+}
+
 /// Acquire one non-cloneable authority that must span the complete reconcile
 /// effect window, including its final strong readback.
 pub async fn acquire_legacy_transition_state_reconcile_fleet_proof() -> Option<LegacyTransitionStateReconcileFleetProofToken> {
@@ -1120,6 +1124,14 @@ pub(crate) fn install_remote_version_state_fleet_proof_for_test(topology_fingerp
     RemoteVersionStateFleetProofGuard
 }
 
+#[cfg(all(test, feature = "test-util"))]
+pub(crate) fn install_current_remote_version_state_fleet_proof_for_test() -> RemoteVersionStateFleetProofGuard {
+    let topology = REMOTE_VERSION_STATE_PROBE_TOPOLOGY
+        .get()
+        .expect("the test store must bind its fleet topology before installing a writer proof");
+    install_remote_version_state_fleet_proof_for_test(topology)
+}
+
 #[cfg(all(test, feature = "test-util"))]
 pub(crate) struct TransitionTransactionCompactionFleetProofGuard;
 

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

@@ -653,6 +653,26 @@ impl MockWarmBackend {
 
 #[async_trait]
 impl WarmBackend for MockWarmBackend {
+    async fn probe_legacy_metadata(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<super::warm_backend::LegacyTransitionStateProbe, std::io::Error> {
+        use super::warm_backend::LegacyTransitionStateProbe as Probe;
+        let candidate = match remote_version {
+            Some(version) if !version.is_empty() => self.probe_transition_version(object, version).await?,
+            _ => self.probe_transition_candidate(object).await?,
+        };
+        Ok(match candidate {
+            TransitionCandidateProbe::Missing => Probe::Missing,
+            TransitionCandidateProbe::UnversionedPresent => Probe::UnversionedPresent,
+            TransitionCandidateProbe::VersionedPresent(version) if version == "null" => Probe::SuspendedNullPresent,
+            TransitionCandidateProbe::VersionedPresent(version) => Probe::VersionedPresent(version),
+            TransitionCandidateProbe::Ambiguous => Probe::Ambiguous,
+            TransitionCandidateProbe::Unsupported => Probe::Unsupported,
+        })
+    }
+
     fn validate_remote_version_id(&self, remote_version_id: &str) -> Result<(), std::io::Error> {
         if remote_version_id.is_empty() {
             return Ok(());
@@ -874,8 +894,9 @@ pub async fn register_mock_tier_backend(handle: &Arc<RwLock<TierConfigMgr>>, tie
             ..Default::default()
         },
     );
-    tier_config_mgr
-        .install_test_driver(tier_name, Box::new(backend))
+    drop(tier_config_mgr);
+    TierConfigMgr::install_test_driver_in(handle, tier_name, Box::new(backend))
+        .await
         .expect("mock tier driver should install");
 }
 

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

@@ -2322,6 +2322,14 @@ struct SharedWarmBackendProxy(SharedWarmBackend);
 
 #[async_trait::async_trait]
 impl WarmBackend for SharedWarmBackendProxy {
+    async fn probe_legacy_metadata(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> io::Result<crate::services::tier::warm_backend::LegacyTransitionStateProbe> {
+        self.0.probe_legacy_metadata(object, remote_version).await
+    }
+
     async fn validate(&self) -> io::Result<()> {
         self.0.validate().await
     }
@@ -2490,6 +2498,27 @@ impl TierOperationLease {
         self.inner.driver.probe_transition_version(object, remote_version_id).await
     }
 
+    pub(crate) async fn probe_legacy_transition_state(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> io::Result<crate::services::tier::warm_backend::LegacyTransitionStateProbe> {
+        let Some(reconciler) = self
+            .inner
+            .reconciler
+            .get_or_try_init(|| async {
+                crate::services::tier::warm_backend::new_transition_candidate_reconciler(&self.inner.tier_config)
+                    .await
+                    .map(|reconciler| reconciler.map(Arc::from))
+            })
+            .await
+            .map_err(|err| io::Error::other(err.message))?
+        else {
+            return self.inner.driver.probe_legacy_metadata(object, remote_version).await;
+        };
+        reconciler.probe_legacy_transition_state(object, remote_version).await
+    }
+
     pub(crate) fn is_current_generation(&self) -> bool {
         lock_unpoisoned(&self.runtime)
             .generations
@@ -6013,6 +6042,19 @@ impl TierConfigMgr {
         Ok(())
     }
 
+    #[cfg(any(test, feature = "test-util"))]
+    pub(crate) async fn install_test_driver_in(
+        handle: &Arc<RwLock<Self>>,
+        tier_name: &str,
+        driver: WarmBackendImpl,
+    ) -> std::result::Result<(), AdminError> {
+        let mut manager = handle.write().await;
+        // Register the generation runtime before installing the mock so its
+        // explicit lack of a network reconciler survives the first lease.
+        tier_driver_runtime(handle, &manager);
+        manager.install_test_driver(tier_name, driver)
+    }
+
     #[cfg(any(test, feature = "test-util"))]
     pub(crate) fn install_test_driver(
         &mut self,

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

@@ -89,6 +89,18 @@ pub enum TransitionCandidateProbe {
     Unsupported,
 }
 
+/// Live evidence for repairing legacy metadata. Ordinary candidate GETs do not
+/// establish the bucket's versioning model and cannot supply this authority.
+#[derive(Clone, Debug, Eq, PartialEq)]
+pub enum LegacyTransitionStateProbe {
+    Missing,
+    UnversionedPresent,
+    SuspendedNullPresent,
+    VersionedPresent(String),
+    Ambiguous,
+    Unsupported,
+}
+
 #[derive(Clone, Copy)]
 pub(crate) struct TransitionCandidateIdentity {
     pub transaction_id: uuid::Uuid,
@@ -97,6 +109,14 @@ pub(crate) struct TransitionCandidateIdentity {
 
 #[async_trait::async_trait]
 pub(crate) trait TransitionCandidateReconciler {
+    async fn probe_legacy_transition_state(
+        &self,
+        _object: &str,
+        _remote_version: Option<&str>,
+    ) -> Result<LegacyTransitionStateProbe, std::io::Error> {
+        Ok(LegacyTransitionStateProbe::Unsupported)
+    }
+
     async fn probe_transition_candidate_for(
         &self,
         object: &str,
@@ -106,6 +126,14 @@ pub(crate) trait TransitionCandidateReconciler {
 
 #[async_trait::async_trait]
 pub trait WarmBackend {
+    async fn probe_legacy_metadata(
+        &self,
+        _object: &str,
+        _remote_version: Option<&str>,
+    ) -> Result<LegacyTransitionStateProbe, std::io::Error> {
+        Ok(LegacyTransitionStateProbe::Unsupported)
+    }
+
     async fn validate(&self) -> Result<(), std::io::Error> {
         Ok(())
     }
@@ -448,6 +476,18 @@ impl MeteredWarmBackend {
 
 #[async_trait::async_trait]
 impl WarmBackend for MeteredWarmBackend {
+    async fn probe_legacy_metadata(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<LegacyTransitionStateProbe, std::io::Error> {
+        let result = self.inner.probe_legacy_metadata(object, remote_version).await;
+        if matches!(result, Ok(LegacyTransitionStateProbe::Unsupported)) {
+            return result;
+        }
+        Self::record(TierRequestOperation::Probe, result)
+    }
+
     /// Delegated without a counter: only one backend issues a remote request
     /// here, and every other one takes the trait default, so a `validate`
     /// counter would mostly record requests that never happened.
@@ -524,6 +564,18 @@ struct MeteredTransitionCandidateReconciler {
 
 #[async_trait::async_trait]
 impl TransitionCandidateReconciler for MeteredTransitionCandidateReconciler {
+    async fn probe_legacy_transition_state(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<LegacyTransitionStateProbe, std::io::Error> {
+        let result = self.inner.probe_legacy_transition_state(object, remote_version).await;
+        if matches!(result, Ok(LegacyTransitionStateProbe::Unsupported)) {
+            return result;
+        }
+        MeteredWarmBackend::record(TierRequestOperation::Probe, result)
+    }
+
     async fn probe_transition_candidate_for(
         &self,
         object: &str,

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

@@ -101,6 +101,19 @@ impl WarmBackend for WarmBackendMinIO {
 
 #[async_trait::async_trait]
 impl crate::services::tier::warm_backend::TransitionCandidateReconciler for WarmBackendMinIO {
+    async fn probe_legacy_transition_state(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<super::warm_backend::LegacyTransitionStateProbe, std::io::Error> {
+        crate::services::tier::warm_backend::TransitionCandidateReconciler::probe_legacy_transition_state(
+            &self.0,
+            object,
+            remote_version,
+        )
+        .await
+    }
+
     async fn probe_transition_candidate_for(
         &self,
         object: &str,

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

@@ -146,6 +146,19 @@ impl WarmBackend for WarmBackendRustFS {
 
 #[async_trait::async_trait]
 impl crate::services::tier::warm_backend::TransitionCandidateReconciler for WarmBackendRustFS {
+    async fn probe_legacy_transition_state(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<super::warm_backend::LegacyTransitionStateProbe, std::io::Error> {
+        crate::services::tier::warm_backend::TransitionCandidateReconciler::probe_legacy_transition_state(
+            &self.0,
+            object,
+            remote_version,
+        )
+        .await
+    }
+
     async fn probe_transition_candidate_for(
         &self,
         object: &str,

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

@@ -376,7 +376,6 @@ struct TransitionCandidateVersions {
 }
 
 impl TransitionCandidateVersions {
-    #[cfg(test)]
     fn extend(&mut self, remote_object: &str, versions: &ListVersionsResult) {
         for version in versions.versions.iter().filter(|version| version.key == remote_object) {
             if self.version_id.is_some() {
@@ -512,6 +511,20 @@ mod tests {
     }
 
     async fn candidate_probe_fixture() -> Option<(WarmBackendS3, tokio::task::JoinHandle<Vec<String>>)> {
+        scripted_probe_fixture([
+                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nx-amz-version-id: opaque-version\r\nConnection: close\r\n\r\nx",
+                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nConnection: close\r\n\r\nx",
+                "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>",
+            ].into_iter().map(str::to_owned).collect()).await
+    }
+
+    async fn scripted_probe_fixture(responses: Vec<String>) -> Option<(WarmBackendS3, tokio::task::JoinHandle<Vec<String>>)> {
         let listener = match tokio::net::TcpListener::bind("127.0.0.1:0").await {
             Ok(listener) => listener,
             Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => return None,
@@ -522,17 +535,6 @@ mod tests {
             .expect("listener local address should be available")
             .to_string();
         let fixture = tokio::spawn(async move {
-            let responses = [
-                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nx-amz-version-id: opaque-version\r\nConnection: close\r\n\r\nx",
-                "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nConnection: close\r\n\r\nx",
-                "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 {
                 let (mut stream, _) = listener.accept().await.expect("fixture should accept candidate GET");
@@ -673,6 +675,117 @@ mod tests {
         assert!(requests[8].to_ascii_lowercase().contains("?versionid=historical-version"));
     }
 
+    fn legacy_probe_xml_response(body: &str) -> String {
+        format!(
+            "HTTP/1.1 200 OK\r\nContent-Type: application/xml\r\nContent-Length: {}\r\nConnection: close\r\n\r\n{body}",
+            body.len()
+        )
+    }
+
+    fn legacy_probe_versioning_response(status: &str) -> String {
+        let state = if status.is_empty() {
+            String::new()
+        } else {
+            format!("<Status>{status}</Status>")
+        };
+        legacy_probe_xml_response(&format!(
+            "<VersioningConfiguration xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\">{state}</VersioningConfiguration>"
+        ))
+    }
+
+    fn legacy_probe_versions_response(versions: &[&str]) -> String {
+        let versions = versions.iter().map(|version| format!(
+            "<Version><Key>archive/object</Key><VersionId>{version}</VersionId><IsLatest>true</IsLatest><LastModified>2026-09-01T00:00:00Z</LastModified><ETag>\"legacy-etag\"</ETag><Size>7</Size><StorageClass>STANDARD</StorageClass></Version>"
+        )).collect::<String>();
+        legacy_probe_xml_response(&format!(
+            "<ListVersionsResult xmlns=\"http://s3.amazonaws.com/doc/2006-03-01/\"><Name>bucket</Name><Prefix>archive/object</Prefix><KeyMarker/><VersionIdMarker/><MaxKeys>1000</MaxKeys><IsTruncated>false</IsTruncated>{versions}</ListVersionsResult>"
+        ))
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_probe_verifies_disabled_suspended_and_enabled_responses() {
+        use super::super::warm_backend::LegacyTransitionStateProbe as Probe;
+        for (initial, confirmed, version, expected) in [
+            ("", "", "null", Probe::UnversionedPresent),
+            ("Suspended", "Suspended", "null", Probe::SuspendedNullPresent),
+            ("Enabled", "Enabled", "version-a", Probe::VersionedPresent("version-a".to_string())),
+            ("Enabled", "Enabled", "null", Probe::SuspendedNullPresent),
+            ("", "", "unexpected-version", Probe::Ambiguous),
+            ("Suspended", "Enabled", "null", Probe::Ambiguous),
+        ] {
+            let responses = vec![
+                legacy_probe_versioning_response(initial),
+                legacy_probe_versions_response(&[version]),
+                legacy_probe_versioning_response(confirmed),
+            ];
+            let (backend, fixture) = scripted_probe_fixture(responses)
+                .await
+                .expect("legacy probe loopback fixture");
+            let result =
+                tokio::time::timeout(Duration::from_secs(10), backend.probe_legacy_transition_state("archive/object", None))
+                    .await
+                    .expect("legacy probe must finish")
+                    .expect("legacy probe should decode provider XML");
+            assert_eq!(result, expected, "initial={initial} confirmed={confirmed} version={version}");
+            let requests = fixture.await.expect("legacy probe fixture should finish");
+            assert_eq!(requests.len(), 3);
+            assert!(requests.iter().all(|request| request.starts_with("GET ")));
+            assert!(requests[0].lines().next().expect("request line").contains("versioning"));
+            assert!(requests[1].lines().next().expect("request line").contains("versions"));
+            assert!(requests[2].lines().next().expect("request line").contains("versioning"));
+        }
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_probe_preserves_historical_exact_version() {
+        use super::super::warm_backend::LegacyTransitionStateProbe as Probe;
+        let responses = vec![
+            legacy_probe_versioning_response("Enabled"),
+            "HTTP/1.1 206 Partial Content\r\nContent-Length: 1\r\nx-amz-version-id: historical-version\r\nConnection: close\r\n\r\nx".to_string(),
+            legacy_probe_versioning_response("Enabled"),
+        ];
+        let (backend, fixture) = scripted_probe_fixture(responses).await.expect("exact legacy probe fixture");
+        assert_eq!(
+            backend
+                .probe_legacy_transition_state("archive/object", Some("historical-version"))
+                .await
+                .expect("exact version proof"),
+            Probe::VersionedPresent("historical-version".to_string())
+        );
+        let requests = fixture.await.expect("exact probe fixture should finish");
+        assert!(
+            requests[1]
+                .lines()
+                .next()
+                .expect("request line")
+                .contains("versionId=historical-version")
+        );
+        assert!(requests[1].to_ascii_lowercase().contains("range: bytes=0-0"));
+        assert!(requests.iter().all(|request| request.starts_with("GET ")));
+    }
+
+    #[tokio::test]
+    async fn legacy_transition_state_probe_retains_multiple_candidates() {
+        use super::super::warm_backend::LegacyTransitionStateProbe as Probe;
+        let responses = vec![
+            legacy_probe_versioning_response("Enabled"),
+            legacy_probe_versions_response(&["version-a", "version-b"]),
+        ];
+        let (backend, fixture) = scripted_probe_fixture(responses)
+            .await
+            .expect("ambiguous legacy probe fixture");
+        assert_eq!(
+            backend
+                .probe_legacy_transition_state("archive/object", None)
+                .await
+                .expect("ambiguous proof"),
+            Probe::Ambiguous
+        );
+        let requests = fixture.await.expect("ambiguous fixture should finish");
+        assert_eq!(requests.len(), 2);
+        assert!(requests.iter().all(|request| request.starts_with("GET ")));
+    }
+
     fn list_versions(versions: &[(&str, &str)], delete_markers: &[(&str, &str)], is_truncated: bool) -> ListVersionsResult {
         ListVersionsResult {
             versions: versions
@@ -921,6 +1034,56 @@ impl WarmBackend for WarmBackendS3 {
 
 #[async_trait::async_trait]
 impl TransitionCandidateReconciler for WarmBackendS3 {
+    async fn probe_legacy_transition_state(
+        &self,
+        object: &str,
+        remote_version: Option<&str>,
+    ) -> Result<super::warm_backend::LegacyTransitionStateProbe, std::io::Error> {
+        use super::warm_backend::LegacyTransitionStateProbe as Probe;
+
+        let initial_versioning = self.remote_bucket_versioning().await?;
+        let candidate = if let Some(version) = remote_version.filter(|version| !version.is_empty()) {
+            validate_remote_version_id(version)?;
+            match self.probe_transition_version(object, version).await? {
+                TransitionCandidateProbe::VersionedPresent(actual) if actual == version => Some(actual),
+                TransitionCandidateProbe::Missing => return Ok(Probe::Missing),
+                _ => return Ok(Probe::Ambiguous),
+            }
+        } else {
+            let remote_object = self.get_dest(object);
+            let mut opts = ListObjectsOptions::default();
+            opts.set("prefix", &remote_object);
+            opts.set("max-keys", "1000");
+            let mut key_marker = String::new();
+            let mut version_marker = String::new();
+            let mut candidates = TransitionCandidateVersions::default();
+            let mut complete = false;
+            // This is one synchronous record inspection, not an unbounded
+            // remote history scan. The caller also bounds the whole probe.
+            for _ in 0..128 {
+                let page = self
+                    .client
+                    .list_object_versions_query(&self.bucket, &opts, &key_marker, &version_marker, "")
+                    .await?;
+                candidates.extend(&remote_object, &page);
+                if candidates.ambiguous {
+                    return Ok(Probe::Ambiguous);
+                }
+                if !page.is_truncated {
+                    complete = true;
+                    break;
+                }
+                advance_version_markers(&mut key_marker, &mut version_marker, &page)?;
+            }
+            if !complete {
+                return Ok(Probe::Ambiguous);
+            }
+            candidates.version_id
+        };
+        let confirmed_versioning = self.remote_bucket_versioning().await?;
+        classify_legacy_transition_state(candidate.as_deref(), initial_versioning, confirmed_versioning)
+    }
+
     async fn probe_transition_candidate_for(
         &self,
         object: &str,
@@ -931,3 +1094,32 @@ impl TransitionCandidateReconciler for WarmBackendS3 {
             .await
     }
 }
+
+fn classify_legacy_transition_state(
+    candidate: Option<&str>,
+    initial: RemoteBucketVersioning,
+    confirmed: RemoteBucketVersioning,
+) -> Result<super::warm_backend::LegacyTransitionStateProbe, std::io::Error> {
+    use super::warm_backend::LegacyTransitionStateProbe as Probe;
+    if initial != confirmed {
+        return Ok(Probe::Ambiguous);
+    }
+    let Some(version) = candidate else {
+        return Ok(Probe::Missing);
+    };
+    if !version.is_empty() {
+        validate_remote_version_id(version)?;
+        if uuid::Uuid::parse_str(version).is_ok_and(|id| id.is_nil()) {
+            return Err(std::io::Error::new(
+                std::io::ErrorKind::InvalidData,
+                "legacy tier probe returned a nil version identifier",
+            ));
+        }
+    }
+    Ok(match (confirmed, version) {
+        (RemoteBucketVersioning::Disabled, "" | "null") => Probe::UnversionedPresent,
+        (RemoteBucketVersioning::Disabled, _) | (_, "") => Probe::Ambiguous,
+        (_, "null") => Probe::SuspendedNullPresent,
+        (_, version) => Probe::VersionedPresent(version.to_string()),
+    })
+}

crates/ecstore/src/set_disk/mod.rs

@@ -3888,6 +3888,31 @@ pub struct SetDisks {
     >,
 }
 
+/// Read every physical copy before selecting a version quorum. A minority
+/// legacy record is still evidence and must not disappear behind a majority
+/// not-found result. Only an explicit file/volume absence produces `None`;
+/// an unreadable disk cannot prove that no conflicting copy exists.
+pub(crate) async fn read_legacy_transition_state_metadata_copies(
+    set: &SetDisks,
+    bucket: &str,
+    object: &str,
+) -> std::result::Result<Vec<Option<Vec<u8>>>, DiskError> {
+    let disk_object = rustfs_utils::path::encode_dir_object(object);
+    let disks = set.get_disks_internal().await;
+    if disks.is_empty() {
+        return Err(DiskError::DiskNotFound);
+    }
+
+    let (copies, errs) = SetDisks::read_all_raw_file_info(&disks, bucket, disk_object.as_str(), false).await;
+    for err in errs.into_iter().flatten() {
+        if !matches!(err, DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) {
+            return Err(err);
+        }
+    }
+
+    Ok(copies.into_iter().map(|copy| copy.map(|copy| copy.buf)).collect())
+}
+
 // DistributedLock sends the raw ObjectKey to its clients; LockRegistry clones
 // each endpoint's canonical Arc, so an exact Arc set identifies the lock domain.
 pub(crate) fn same_distributed_lock_domain(left: &[Arc<dyn LockClient>], right: &[Arc<dyn LockClient>]) -> bool {

crates/ecstore/src/store/init.rs

@@ -12803,6 +12803,227 @@ mod tests {
         body
     }
 
+    #[cfg(feature = "test-util")]
+    #[test]
+    #[serial_test::serial(storage_class_env)]
+    fn legacy_transition_state_inspection_and_apply_keep_all_disk_copies_unchanged() {
+        run_large_stack_async_test("legacy-state-reconcile-inspection", legacy_transition_state_inspection_and_apply_case);
+    }
+
+    #[cfg(feature = "test-util")]
+    async fn legacy_transition_state_inspection_and_apply_case() {
+        use crate::bucket::lifecycle::legacy_transition_state_reconcile::{
+            LegacyTransitionStateReconcileOutcome as Outcome, LegacyTransitionStateReconcileRequest,
+            LegacyTransitionStateReconcileSelector,
+        };
+        for (remote_version, expected_state) in [
+            ("", rustfs_filemeta::TransitionVersionState::KnownDisabled),
+            ("null", rustfs_filemeta::TransitionVersionState::SuspendedNull),
+            ("opaque-version", rustfs_filemeta::TransitionVersionState::Exact),
+        ] {
+            let temp_dir = tempfile::tempdir().expect("legacy reconcile store directory");
+            let (ctx, store, _shutdown) =
+                without_storage_class_env(build_isolated_test_store(temp_dir.path(), "legacy-state-reconcile-inspect", &[4]))
+                    .await;
+            crate::bucket::metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await;
+            let tier_name = "LEGACY-RECONCILE";
+            let backend = register_mock_tier(&ctx.tier_config_mgr(), tier_name).await;
+            backend.set_put_remote_version(Some(remote_version.to_string())).await;
+            let bucket = "legacy-state-reconcile-bucket";
+            let object = "archive.bin";
+            store
+                .make_bucket(bucket, &MakeBucketOptions::default())
+                .await
+                .expect("create legacy fixture bucket");
+            let mut reader = PutObjReader::from_vec(b"legacy reconcile body".repeat(1024));
+            let source = store
+                .put_object(bucket, object, &mut reader, &ObjectOptions::default())
+                .await
+                .expect("write source");
+            {
+                // Create the fixture under the existing remote-version writer
+                // gate. This does not authorize legacy metadata reconciliation.
+                let _proof = crate::services::notification_sys::install_current_remote_version_state_fleet_proof_for_test();
+                temp_env::async_with_vars(
+                    [
+                        (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_WRITE, Some("true")),
+                        (rustfs_config::ENV_TIER_REMOTE_VERSION_STATE_FLEET_CONFIRMED, Some("true")),
+                    ],
+                    store.transition_object(
+                        bucket,
+                        object,
+                        &ObjectOptions {
+                            transition: TransitionOptions {
+                                status: TRANSITION_PENDING.to_string(),
+                                tier: tier_name.to_string(),
+                                etag: source.etag.clone().expect("source ETag"),
+                                ..Default::default()
+                            },
+                            mod_time: source.mod_time,
+                            ..Default::default()
+                        },
+                    ),
+                )
+                .await
+                .expect("transition source");
+            }
+            assert!(
+                crate::services::notification_sys::acquire_legacy_transition_state_reconcile_fleet_proof()
+                    .await
+                    .is_none(),
+                "fixture setup must not grant the missing reconciliation write capability"
+            );
+            let selector = LegacyTransitionStateReconcileSelector {
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                version_id: "null".to_string(),
+            };
+            if expected_state == rustfs_filemeta::TransitionVersionState::Exact {
+                backend
+                    .set_transition_candidate_probe_override(Some(
+                        crate::services::tier::warm_backend::TransitionCandidateProbe::Ambiguous,
+                    ))
+                    .await;
+            }
+            let converged = store
+                .inspect_legacy_transition_state(selector.clone())
+                .await
+                .expect("inspect an already explicit transition");
+            assert_eq!(converged.outcome, Outcome::Migrated, "{converged:?}");
+            assert!(!converged.changed);
+            backend.set_transition_candidate_probe_override(None).await;
+            rewrite_transitioned_xlmeta_as_legacy_unknown(temp_dir.path(), 0, bucket, object, remote_version.is_empty()).await;
+            let paths = (0..4)
+                .map(|disk| {
+                    temp_dir
+                        .path()
+                        .join(format!("pool0/set0/disk{disk}/{bucket}/{object}/{STORAGE_FORMAT_FILE}"))
+                })
+                .collect::<Vec<_>>();
+            let mut original = Vec::new();
+            for path in &paths {
+                original.push(tokio::fs::read(path).await.expect("original xl.meta"));
+            }
+            backend.clear_op_log().await;
+            let inspection = store.inspect_legacy_transition_state(selector.clone());
+            assert!(
+                std::mem::size_of_val(&inspection) <= 4 * 1024,
+                "admin inspection future must remain stack-bounded"
+            );
+            let inspected = inspection.await.expect("inspect legacy state");
+            assert_eq!(inspected.outcome, Outcome::ReadyToMigrate, "{inspected:?}");
+            assert!(!inspected.readiness.post_ready, "current fleet cannot authorize conditional writes");
+            let target = inspected.target.expect("live probe should establish one model");
+            assert_eq!(target.state, expected_state);
+            let request = LegacyTransitionStateReconcileRequest {
+                confirm: true,
+                selector,
+                source: inspected.source.expect("immutable source"),
+                original_sets: inspected.original_sets,
+                target,
+                reconciliation_digest: inspected.reconciliation_digest.expect("expected tuple digest"),
+            };
+            let mut tampered = request.clone();
+            tampered.source.remote_object.push_str("-other");
+            let probes_before = backend.op_log().await.len();
+            let rejected = store
+                .reconcile_legacy_transition_state(tampered)
+                .await
+                .expect("reject tampered tuple");
+            assert_eq!(rejected.outcome, Outcome::Corrupt);
+            assert_eq!(backend.op_log().await.len(), probes_before, "invalid digest must not probe the backend");
+            let applied = store
+                .reconcile_legacy_transition_state(request)
+                .await
+                .expect("apply must report unavailable write authority");
+            assert_eq!(applied.outcome, Outcome::BackendUnavailable, "{applied:?}");
+            assert_eq!(applied.reason_code, "write_fence_unavailable");
+            assert!(!applied.changed);
+            for (path, expected) in paths.iter().zip(&original) {
+                assert_eq!(tokio::fs::read(path).await.expect("xl.meta after inspection"), *expected);
+            }
+            assert_eq!(backend.remove_count().await, 0);
+            assert!(
+                backend
+                    .op_log()
+                    .await
+                    .iter()
+                    .all(|operation| matches!(operation, MockWarmOp::Probe { .. }))
+            );
+
+            backend.set_unreachable(true).await;
+            let unavailable = store
+                .inspect_legacy_transition_state(LegacyTransitionStateReconcileSelector {
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    version_id: "null".to_string(),
+                })
+                .await
+                .expect("unreachable tier is a diagnostic outcome");
+            assert_eq!(unavailable.outcome, Outcome::BackendUnavailable);
+            assert!(!unavailable.changed);
+            backend.set_unreachable(false).await;
+            for candidate in ["", "00000000-0000-0000-0000-000000000000", "bad\nversion"] {
+                backend
+                    .set_transition_candidate_probe_override(Some(
+                        crate::services::tier::warm_backend::TransitionCandidateProbe::VersionedPresent(candidate.to_string()),
+                    ))
+                    .await;
+                let invalid_proof = store
+                    .inspect_legacy_transition_state(LegacyTransitionStateReconcileSelector {
+                        bucket: bucket.to_string(),
+                        object: object.to_string(),
+                        version_id: "null".to_string(),
+                    })
+                    .await
+                    .expect("invalid backend proof is a diagnostic outcome");
+                assert_eq!(invalid_proof.outcome, Outcome::BackendUnavailable, "{invalid_proof:?}");
+                assert!(invalid_proof.target.is_none());
+            }
+            backend.set_transition_candidate_probe_override(None).await;
+
+            backend.clear_op_log().await;
+            for path in &paths[1..] {
+                tokio::fs::remove_file(path)
+                    .await
+                    .expect("hide majority metadata copies in fixture");
+            }
+            let minority = store
+                .inspect_legacy_transition_state(LegacyTransitionStateReconcileSelector {
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    version_id: "null".to_string(),
+                })
+                .await
+                .expect("inspect minority legacy record");
+            assert_eq!(
+                minority.outcome,
+                Outcome::BackendUnavailable,
+                "a minority owner must remain visible: {minority:?}"
+            );
+            assert!(
+                backend.op_log().await.is_empty(),
+                "unproven metadata quorum cannot initiate a remote probe"
+            );
+            for (path, bytes) in paths.iter().zip(&original) {
+                tokio::fs::write(path, bytes).await.expect("restore fixture copies");
+            }
+            tokio::fs::write(&paths[0], b"corrupt-xl-meta")
+                .await
+                .expect("inject corrupt copy");
+            let corrupt = store
+                .inspect_legacy_transition_state(LegacyTransitionStateReconcileSelector {
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    version_id: "null".to_string(),
+                })
+                .await
+                .expect("inspect corrupt legacy record");
+            assert_eq!(corrupt.outcome, Outcome::Corrupt, "{corrupt:?}");
+            assert!(backend.op_log().await.is_empty(), "corruption must fail before backend I/O");
+        }
+    }
+
     #[cfg(feature = "test-util")]
     #[tokio::test]
     #[serial_test::serial(storage_class_env)]

docs/architecture/ilm-tiering-persistence-contracts.md

@@ -522,7 +522,7 @@ The following single-record protocol approves how historical objects without Rus
 
 ### Current
 
-An absent `transitioned-version-state` key decodes as `TransitionVersionState::Unknown`. The current GET and free-version cleanup paths reject that state rather than interpreting an empty remote version as unversioned. There is no admin route that repairs this field in `xl.meta`. The existing transition-transaction reconcile route operates on expired `UploadOutcomeUnknown` transaction records and can exact-delete their canonical candidates; it is a separate protocol and must not be reused for metadata reconciliation.
+An absent `transitioned-version-state` key decodes as `TransitionVersionState::Unknown`. Non-destructive compatibility reads distinguish legacy absence from explicit `Unknown`; empty-version reads require a bounded probe. Legacy free-version cleanup requires a persisted exact remote version and does not infer unversioned deletion from an empty field. The single-record Admin routes below now inspect physical copies and live remote-version evidence. They do not yet persist repaired state: POST reports `write_fence_unavailable` until conditional per-generation metadata writes and the dedicated fleet capability are available. The existing transition-transaction reconcile route operates on expired `UploadOutcomeUnknown` transaction records and can exact-delete their canonical candidates; it is a separate protocol and must not be reused for metadata reconciliation.
 
 An explicitly persisted `unknown`, a malformed state, conflicting RustFS/MinIO compatibility keys, an invalid or nil version identifier, and a partial transition tuple are not legacy absence. They remain invalid or ambiguous and fail closed.
 

docs/operations/tier-ilm-debugging.md

@@ -262,9 +262,11 @@ The full schema, lease, mixed-version, retry, privacy, and metric requirements a
 
 ## Reconcile legacy transition-version metadata
 
-This section describes an **approved target that is not implemented yet**. The current server has no admin route that backfills a missing `transitioned-version-state` in `xl.meta`. Do not use the transaction reconcile route above for this purpose: that route owns an upload transaction candidate and may delete it, while legacy metadata reconciliation is non-destructive and may update only the exact local metadata version.
+The single-record inspection routes below are implemented. GET audits every physical metadata copy, verifies the legacy tuple, and performs a bounded live version-model probe. POST validates confirmation and the complete expected tuple, repeats inspection, and returns `backend-unavailable` with `reason_code: write_fence_unavailable`, `changed: false`, and `post_ready: false` when migration is needed. Persistent backfill remains disabled until conditional per-generation `xl.meta` writes and the dedicated fleet capability are available. An already explicit, converged record can return `migrated` without changing bytes.
 
-The approved interface is synchronous and accepts exactly one bucket/object/local-version tuple:
+Do not use the transaction reconcile route above for this purpose: that route owns an upload transaction candidate and may delete it, while legacy metadata inspection never mutates the remote tier or local metadata.
+
+The interface is synchronous and accepts exactly one bucket/object/local-version tuple:
 
 ```text
 GET  /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<object>&versionId=<local-version-id>
@@ -273,7 +275,7 @@ POST /rustfs/admin/v3/ilm/transition/state/reconcile?bucket=<bucket>&object=<obj
 
 `versionId` is required; use the literal `null` for a locally unversioned object. An omitted or empty selector is invalid. GET requires `admin:ListTier`. It reports the authoritative all-pool tuple, destination identity, fleet/topology readiness, live probe classification, opaque expected-tuple digest, and a machine-readable diagnosis. It returns `ready-to-migrate`, not `migrated`, when a missing state is provable because GET is read-only.
 
-POST requires `admin:SetTier`, `confirm: true`, and the complete immutable source tuple, original per-set missing-state representations, proposed target, and reconciliation digest returned by GET. The server rereads every authoritative copy and repeats the bounded live backend probe; provider console output or an operator-supplied state is diagnostic evidence only, never write authority. A retry accepts only copies that still match their digest-bound original representation or already equal the exact proven target; any other divergence is stale or corrupt. The server may persist only one of these exact state/version pairs, together with the bound destination identity:
+POST requires `admin:SetTier`, `confirm: true`, and the complete immutable source tuple, original per-set missing-state representations, proposed target, and reconciliation digest returned by GET. The current implementation rejects any changed expected tuple and does not write. The approved future writer must reread every authoritative copy and repeat the bounded live backend probe; provider console output or an operator-supplied state is diagnostic evidence only, never write authority. Its retry may accept only copies that still match their digest-bound original representation or already equal the exact proven target; any other divergence is stale or corrupt. That writer may persist only one of these exact state/version pairs, together with the bound destination identity:
 
 | Proven remote model | State | Version value |
 |---|---|---|
@@ -283,9 +285,9 @@ POST requires `admin:SetTier`, `confirm: true`, and the complete immutable sourc
 
 The response outcome is `migrated`, `retained-ambiguous`, `corrupt`, or `backend-unavailable`. `migrated` means strong all-pool readback proved the same state and destination identity on every authoritative copy; it can be idempotent with `changed=false`. Ambiguous/missing/multiple probe results are retained, and explicit `Unknown`, malformed or conflicting dual keys, nil identifiers, partial tuples, or copies outside the exact `{original missing representation, proven target}` retry subset fail closed. An unavailable backend, tier generation, metadata quorum, or required strong readback reports `backend-unavailable`; a monotonic partial write is retained for retry and never rolled back.
 
-The POST does not issue remote DELETE or PUT, remove local data, create a free-version, clean a transaction/journal, or change tier configuration. It holds the approved fleet/topology, bucket-lifecycle, exact tier-generation/destination, and stable all-pool object-version fences across authoritative reread and the bounded probe; it rechecks them before quorum writes and after strong readback. A fleet containing a node that cannot preserve the explicit state/destination binding is inspect-only, and a cross-pool first match is never enough.
+The POST does not issue remote DELETE or PUT, remove local data, create a free-version, clean a transaction/journal, or change tier configuration. The future conditional writer must hold the approved fleet/topology, bucket-lifecycle, exact tier-generation/destination, and stable all-pool object-version fences across authoritative reread and the bounded probe; it must recheck them before quorum writes and after strong readback. A fleet containing a node that cannot preserve the explicit state/destination binding is inspect-only, and a cross-pool first match is never enough.
 
-There is intentionally no bucket, prefix, or fleet selector. Batch repair requires a separate durable, resumable job protocol and remains future work. Until the single-record route is implemented, retain affected metadata, use external inspection only for diagnosis, and never hand-edit `xl.meta` or enable remote cleanup by assuming that an empty version field means an unversioned tier.
+There is intentionally no bucket, prefix, or fleet selector. Batch repair requires a separate durable, resumable job protocol and remains future work. Until the conditional writer is implemented, use GET to distinguish a provable candidate, ambiguity, corrupt metadata, and an unavailable backend. MinIO, RustFS, and direct-S3 probes compare bucket versioning before and after examining an exact stored version or listing a unique candidate. Unsupported providers retain the record as ambiguous. Keep affected metadata and never hand-edit `xl.meta` or enable remote cleanup by assuming that an empty version field means an unversioned tier.
 
 The full approved fence, quorum, cross-set retry, destination-binding, and mixed-version contract is specified in [../architecture/ilm-tiering-persistence-contracts.md](../architecture/ilm-tiering-persistence-contracts.md#legacy-transitioned-version-state-reconciliation).
 

rustfs/src/admin/handlers/ilm_transition.rs

@@ -15,21 +15,22 @@
 use crate::admin::auth::authorize_admin_request;
 use crate::admin::router::{AdminOperation, Operation, S3Router};
 use crate::admin::runtime_sources::{current_action_credentials, object_store_from_extensions};
-use crate::admin::storage_api::bucket::is_reserved_or_invalid_bucket;
+use crate::admin::storage_api::bucket::{is_reserved_or_invalid_bucket, utils::is_valid_object_prefix};
 use crate::admin::storage_api::error::StorageError;
 use crate::admin::storage_api::lifecycle::{
     IlmRecoveryClassification, IlmRecoveryControlView, IlmRecoveryDispositionExecutionOutcome, IlmRecoveryDispositionReasonCode,
-    IlmRecoveryDispositionState, IlmRecoveryExportObservation, IlmRecoveryProtocol, ManualTransitionCancelCheck,
-    ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink, ManualTransitionQueueSnapshot,
-    ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission, ManualTransitionScopeAdmissionClaim,
-    TransitionOperatorDeleteResult, TransitionOperatorError, TransitionRecoveryRetryResult, TransitionRecoveryRetryStatus,
-    claim_manual_transition_scope_admission, create_recovery_export, delete_manual_transition_scope_admission_if_current,
-    delete_transition_candidate_for_operator, dry_run_recovery_disposition, enqueue_transition_for_existing_objects_scoped,
-    execute_recovery_disposition, finalize_missing_transition_transaction_for_operator, inspect_recovery_control,
-    inspect_recovery_export_observation, inspect_transition_recovery_retry_for_operator,
-    inspect_transition_transaction_for_operator, list_recovery_controls, load_manual_transition_job_record,
-    load_manual_transition_scope_admission, load_recovery_export, manual_transition_job_lease_expired,
-    manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired,
+    IlmRecoveryDispositionState, IlmRecoveryExportObservation, IlmRecoveryProtocol, LegacyTransitionStateReconcileError,
+    LegacyTransitionStateReconcileRequest, LegacyTransitionStateReconcileResponse, LegacyTransitionStateReconcileSelector,
+    ManualTransitionCancelCheck, ManualTransitionJobRecord, ManualTransitionJobState, ManualTransitionProgressSink,
+    ManualTransitionQueueSnapshot, ManualTransitionRunOptions, ManualTransitionRunReport, ManualTransitionScopeAdmission,
+    ManualTransitionScopeAdmissionClaim, TransitionOperatorDeleteResult, TransitionOperatorError, TransitionRecoveryRetryResult,
+    TransitionRecoveryRetryStatus, claim_manual_transition_scope_admission, create_recovery_export,
+    delete_manual_transition_scope_admission_if_current, delete_transition_candidate_for_operator, dry_run_recovery_disposition,
+    enqueue_transition_for_existing_objects_scoped, execute_recovery_disposition,
+    finalize_missing_transition_transaction_for_operator, inspect_recovery_control, inspect_recovery_export_observation,
+    inspect_transition_recovery_retry_for_operator, inspect_transition_transaction_for_operator, list_recovery_controls,
+    load_manual_transition_job_record, load_manual_transition_scope_admission, load_recovery_export,
+    manual_transition_job_lease_expired, manual_transition_queue_snapshot, manual_transition_scope_admission_lease_expired,
     persist_manual_transition_job_progress_if_owned, renew_manual_transition_job_lease_if_owned,
     request_manual_transition_job_cancel, retry_transition_recovery_for_operator, save_manual_transition_job_record,
     update_manual_transition_job_record,
@@ -267,6 +268,16 @@ pub fn register_ilm_transition_route(r: &mut S3Router<AdminOperation>) -> std::i
         format!("{ADMIN_PREFIX}/v3/ilm/recovery/exports/{{export_id}}").as_str(),
         AdminOperation(&IlmRecoveryExportDownloadHandler {}),
     )?;
+    r.insert(
+        Method::GET,
+        format!("{ADMIN_PREFIX}/v3/ilm/transition/state/reconcile").as_str(),
+        AdminOperation(&LegacyTransitionStateReconcileInspectHandler {}),
+    )?;
+    r.insert(
+        Method::POST,
+        format!("{ADMIN_PREFIX}/v3/ilm/transition/state/reconcile").as_str(),
+        AdminOperation(&LegacyTransitionStateReconcileApplyHandler {}),
+    )?;
     Ok(())
 }
 
@@ -1001,6 +1012,92 @@ fn validate_recovery_observation_receipt(
     Ok(receipt.observation)
 }
 
+#[derive(Debug, Deserialize)]
+#[serde(deny_unknown_fields)]
+struct LegacyTransitionStateReconcileQuery {
+    bucket: Option<String>,
+    object: Option<String>,
+    #[serde(rename = "versionId")]
+    version_id: Option<String>,
+}
+
+fn parse_legacy_transition_state_reconcile_query(query: Option<&str>) -> S3Result<LegacyTransitionStateReconcileSelector> {
+    let query: LegacyTransitionStateReconcileQuery = serde_urlencoded::from_bytes(query.unwrap_or_default().as_bytes())
+        .map_err(|_| s3_error!(InvalidArgument, "invalid legacy transition-state reconcile query"))?;
+    let bucket = query
+        .bucket
+        .filter(|bucket| !bucket.is_empty())
+        .ok_or_else(|| s3_error!(InvalidRequest, "bucket is required"))?;
+    if is_reserved_or_invalid_bucket(&bucket, false) {
+        return Err(s3_error!(InvalidBucketName, "invalid bucket name"));
+    }
+
+    let object = query
+        .object
+        .filter(|object| !object.is_empty())
+        .ok_or_else(|| s3_error!(InvalidRequest, "object is required"))?;
+    if !is_valid_object_prefix(&object) || object.contains('\n') || object.contains('\r') {
+        return Err(s3_error!(InvalidArgument, "invalid object name"));
+    }
+
+    let version_id = query
+        .version_id
+        .filter(|version_id| !version_id.is_empty())
+        .ok_or_else(|| s3_error!(InvalidRequest, "versionId is required"))?;
+    let version_id = if version_id == "null" {
+        version_id
+    } else {
+        let parsed = Uuid::parse_str(&version_id).map_err(|_| s3_error!(InvalidArgument, "invalid local versionId"))?;
+        if parsed.is_nil() {
+            return Err(s3_error!(InvalidArgument, "invalid local versionId"));
+        }
+        parsed.to_string()
+    };
+
+    Ok(LegacyTransitionStateReconcileSelector {
+        bucket,
+        object,
+        version_id,
+    })
+}
+
+fn validate_legacy_transition_state_reconcile_request(
+    query_selector: &LegacyTransitionStateReconcileSelector,
+    confirm: bool,
+    request_selector: &LegacyTransitionStateReconcileSelector,
+) -> S3Result<()> {
+    if !confirm {
+        return Err(s3_error!(
+            InvalidRequest,
+            "legacy transition-state reconciliation requires confirm=true; use GET to inspect without changes"
+        ));
+    }
+    if request_selector != query_selector {
+        return Err(s3_error!(InvalidRequest, "request selector must exactly match the query selector"));
+    }
+    Ok(())
+}
+
+fn map_legacy_transition_state_reconcile_error(err: LegacyTransitionStateReconcileError) -> S3Error {
+    match err {
+        LegacyTransitionStateReconcileError::InvalidSelector(_) | LegacyTransitionStateReconcileError::InvalidRequest(_) => {
+            s3_error!(InvalidRequest, "invalid legacy transition-state reconciliation request")
+        }
+        LegacyTransitionStateReconcileError::StaleExpectedTuple(_) | LegacyTransitionStateReconcileError::Corrupt(_) => {
+            s3_error!(OperationAborted, "legacy transition-state reconciliation metadata is stale or corrupt")
+        }
+        LegacyTransitionStateReconcileError::WriteFenceUnavailable(_) => {
+            s3_error!(
+                OperationAborted,
+                "legacy transition-state reconciliation could not acquire safe write authority"
+            )
+        }
+        LegacyTransitionStateReconcileError::BackendUnavailable(_) => {
+            s3_error!(InternalError, "legacy transition-state reconciliation backend is unavailable")
+        }
+    }
+}
+
 fn map_transition_operator_error(err: TransitionOperatorError) -> S3Error {
     match err {
         TransitionOperatorError::NotFound => s3_error!(NoSuchKey, "transition transaction not found"),
@@ -1961,6 +2058,52 @@ impl Operation for TransitionReconcileApplyHandler {
     }
 }
 
+pub struct LegacyTransitionStateReconcileInspectHandler {}
+
+#[async_trait::async_trait]
+impl Operation for LegacyTransitionStateReconcileInspectHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::ListTierAction).await?;
+        let selector = parse_legacy_transition_state_reconcile_query(req.uri.query())?;
+        let Some(store) = object_store_from_extensions(&req.extensions) else {
+            return Err(s3_error!(InternalError, "object store is not initialized"));
+        };
+        let response: LegacyTransitionStateReconcileResponse = store
+            .inspect_legacy_transition_state(selector)
+            .await
+            .map_err(map_legacy_transition_state_reconcile_error)?;
+        json_response(StatusCode::OK, &response)
+    }
+}
+
+pub struct LegacyTransitionStateReconcileApplyHandler {}
+
+#[async_trait::async_trait]
+impl Operation for LegacyTransitionStateReconcileApplyHandler {
+    async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        authorize_transition_admin_request(&req, AdminAction::SetTierAction).await?;
+        let selector = parse_legacy_transition_state_reconcile_query(req.uri.query())?;
+        let store = object_store_from_extensions(&req.extensions);
+        let mut input = req.input;
+        let body = input
+            .store_all_limited(MAX_ADMIN_REQUEST_BODY_SIZE)
+            .await
+            .map_err(|_| s3_error!(InvalidRequest, "legacy transition-state reconciliation body is too large or unreadable"))?;
+        let request: LegacyTransitionStateReconcileRequest = serde_json::from_slice(&body)
+            .map_err(|_| s3_error!(InvalidRequest, "legacy transition-state reconciliation request must be valid JSON"))?;
+        validate_legacy_transition_state_reconcile_request(&selector, request.confirm, &request.selector)?;
+        let Some(store) = store else {
+            return Err(s3_error!(InternalError, "object store is not initialized"));
+        };
+
+        let response: LegacyTransitionStateReconcileResponse = store
+            .reconcile_legacy_transition_state(request)
+            .await
+            .map_err(map_legacy_transition_state_reconcile_error)?;
+        json_response(StatusCode::OK, &response)
+    }
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -2592,12 +2735,87 @@ mod tests {
         let apply = src
             .split("impl Operation for TransitionReconcileApplyHandler")
             .nth(1)
-            .and_then(|block| block.split("#[cfg(test)]").next())
+            .and_then(|block| block.split("pub struct LegacyTransitionStateReconcileInspectHandler").next())
             .expect("apply handler block");
         assert!(apply.contains("AdminAction::SetTierAction"));
         assert!(!apply.contains("AdminAction::ListTierAction"));
     }
 
+    #[test]
+    fn legacy_transition_state_query_requires_one_exact_selector() {
+        let version_id = Uuid::new_v4();
+        let query = format!("bucket=test-bucket&object=logs%2F2026%20report&versionId={version_id}");
+        let selector = parse_legacy_transition_state_reconcile_query(Some(&query)).expect("exact version selector should parse");
+        assert_eq!(selector.bucket, "test-bucket");
+        assert_eq!(selector.object, "logs/2026 report");
+        assert_eq!(selector.version_id, version_id.to_string());
+
+        let unversioned =
+            parse_legacy_transition_state_reconcile_query(Some("bucket=test-bucket&object=logs%2Fcurrent&versionId=null"))
+                .expect("explicit null selector should parse");
+        assert_eq!(unversioned.version_id, "null");
+
+        for query in [
+            None,
+            Some("bucket=test-bucket&object=key"),
+            Some("bucket=test-bucket&object=&versionId=null"),
+            Some("bucket=test-bucket&object=key&versionId="),
+            Some("bucket=test-bucket&object=key&versionId=not-a-uuid"),
+            Some("bucket=test-bucket&object=key&versionId=00000000-0000-0000-0000-000000000000"),
+            Some("bucket=test-bucket&object=bad%0Akey&versionId=null"),
+            Some("bucket=test-bucket&object=key&versionId=null&prefix=wide"),
+            Some("bucket=test-bucket&bucket=other-bucket&object=key&versionId=null"),
+        ] {
+            assert!(
+                parse_legacy_transition_state_reconcile_query(query).is_err(),
+                "query should fail closed: {query:?}"
+            );
+        }
+    }
+
+    #[test]
+    fn legacy_transition_state_apply_requires_confirmation_and_matching_selector() {
+        let selector = LegacyTransitionStateReconcileSelector {
+            bucket: "test-bucket".to_string(),
+            object: "key".to_string(),
+            version_id: "null".to_string(),
+        };
+        let different = LegacyTransitionStateReconcileSelector {
+            object: "other-key".to_string(),
+            ..selector.clone()
+        };
+
+        assert!(validate_legacy_transition_state_reconcile_request(&selector, false, &selector).is_err());
+        assert!(validate_legacy_transition_state_reconcile_request(&selector, true, &different).is_err());
+        validate_legacy_transition_state_reconcile_request(&selector, true, &selector)
+            .expect("confirmed exact selector should pass handler validation");
+    }
+
+    #[test]
+    fn legacy_transition_state_routes_use_read_and_write_tier_actions() {
+        let src = include_str!("ilm_transition.rs");
+        let inspect = src
+            .split("impl Operation for LegacyTransitionStateReconcileInspectHandler")
+            .nth(1)
+            .and_then(|block| {
+                block
+                    .split("impl Operation for LegacyTransitionStateReconcileApplyHandler")
+                    .next()
+            })
+            .expect("legacy inspect handler block");
+        assert!(inspect.contains("AdminAction::ListTierAction"));
+        assert!(!inspect.contains("AdminAction::SetTierAction"));
+
+        let apply = src
+            .split("impl Operation for LegacyTransitionStateReconcileApplyHandler")
+            .nth(1)
+            .and_then(|block| block.split("#[cfg(test)]").next())
+            .expect("legacy apply handler block");
+        assert!(apply.contains("AdminAction::SetTierAction"));
+        assert!(!apply.contains("AdminAction::ListTierAction"));
+        assert!(apply.contains("validate_legacy_transition_state_reconcile_request"));
+    }
+
     #[test]
     fn manual_transition_query_defaults_to_bounded_run() {
         let (bucket, options, run_mode) =
@@ -3066,6 +3284,24 @@ mod tests {
         assert_eq!(cancel_err.message(), Some("authentication required"));
     }
 
+    #[tokio::test]
+    async fn legacy_transition_state_handlers_reject_missing_credentials_before_selector_or_body() {
+        let path = "/rustfs/admin/v3/ilm/transition/state/reconcile?bucket=test-bucket&object=key&versionId=null";
+        let inspect_err = LegacyTransitionStateReconcileInspectHandler {}
+            .call(credential_less_admin_request(Method::GET, path), Params::new())
+            .await
+            .expect_err("inspect handler must reject unsigned requests");
+        assert_eq!(inspect_err.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(inspect_err.message(), Some("authentication required"));
+
+        let apply_err = LegacyTransitionStateReconcileApplyHandler {}
+            .call(credential_less_admin_request(Method::POST, path), Params::new())
+            .await
+            .expect_err("apply handler must reject unsigned requests");
+        assert_eq!(apply_err.code(), &S3ErrorCode::InvalidRequest);
+        assert_eq!(apply_err.message(), Some("authentication required"));
+    }
+
     #[test]
     fn manual_transition_job_handlers_authorize_validate_and_load_store() {
         let src = include_str!("ilm_transition.rs");

rustfs/src/admin/route_policy.rs

@@ -550,6 +550,18 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[
         SET_TIER,
         RouteRiskLevel::High,
     ),
+    admin(
+        HttpMethod::Get,
+        "/rustfs/admin/v3/ilm/transition/state/reconcile",
+        LIST_TIER,
+        RouteRiskLevel::High,
+    ),
+    admin(
+        HttpMethod::Post,
+        "/rustfs/admin/v3/ilm/transition/state/reconcile",
+        SET_TIER,
+        RouteRiskLevel::High,
+    ),
     admin(
         HttpMethod::Get,
         "/rustfs/admin/v3/audit/target/list",
@@ -2212,12 +2224,16 @@ mod tests {
         assert_action(HttpMethod::Delete, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SET_TIER);
         assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", LIST_TIER);
         assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", SET_TIER);
+        assert_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/state/reconcile", LIST_TIER);
+        assert_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/state/reconcile", SET_TIER);
         assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/run", SERVER_INFO);
         assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/recovery/records", SERVER_INFO);
         assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SERVER_INFO);
         assert_not_action(HttpMethod::Delete, "/rustfs/admin/v3/ilm/transition/jobs/{job_id}", SERVER_INFO);
         assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", SET_TIER);
         assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/reconcile/{transaction_id}", LIST_TIER);
+        assert_not_action(HttpMethod::Get, "/rustfs/admin/v3/ilm/transition/state/reconcile", SET_TIER);
+        assert_not_action(HttpMethod::Post, "/rustfs/admin/v3/ilm/transition/state/reconcile", LIST_TIER);
     }
 
     #[test]

rustfs/src/admin/route_registration_test.rs

@@ -241,6 +241,8 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
             "/v3/ilm/transition/reconcile/{transaction_id}",
             "/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111",
         ),
+        admin_route(Method::GET, "/v3/ilm/transition/state/reconcile"),
+        admin_route(Method::POST, "/v3/ilm/transition/state/reconcile"),
         admin_route_sample(
             Method::DELETE,
             "/v3/ilm/transition/jobs/{job_id}",
@@ -993,6 +995,8 @@ fn test_register_routes_cover_representative_admin_paths() {
         Method::POST,
         &admin_path("/v3/ilm/transition/reconcile/11111111-1111-4111-8111-111111111111"),
     );
+    assert_route(&router, Method::GET, &admin_path("/v3/ilm/transition/state/reconcile"));
+    assert_route(&router, Method::POST, &admin_path("/v3/ilm/transition/state/reconcile"));
 
     assert_route(&router, Method::GET, &table_catalog_path("/config"));
     assert_route(&router, Method::PUT, &table_catalog_path("/buckets/analytics"));

rustfs/src/admin/storage_api.rs

@@ -231,6 +231,10 @@ pub(crate) mod lifecycle {
         renew_manual_transition_job_lease_if_owned, request_manual_transition_job_cancel, save_manual_transition_job_record,
         update_manual_transition_job_record,
     };
+    pub(crate) use crate::storage::storage_api::ecstore_bucket::lifecycle::legacy_transition_state_reconcile::{
+        LegacyTransitionStateReconcileError, LegacyTransitionStateReconcileRequest, LegacyTransitionStateReconcileResponse,
+        LegacyTransitionStateReconcileSelector,
+    };
     pub(crate) type ManualTransitionCancelCheck =
         super::ecstore_bucket::lifecycle::bucket_lifecycle_ops::ManualTransitionCancelCheck;
     pub(crate) type ManualTransitionQueueSnapshot =