chore: sync main for ODM regression validation

.config/e2e-full-selection.txt

@@ -1,2 +1,2 @@
 sha256-darwin=a881fd7d3f5cb94654221ca85b8b30cce1b95e608824a55a15339cbc294e6d34
-sha256-linux=e9a8d64e73f627c4d26c236dbbba690c9ee03a9e26d42a4244515b4439365535
+sha256-linux=a2933d83dfe74ffa03410a0959333a1c48288b8469ca9f17273d449d7510c24b

.config/ecstore-required-tests.json

@@ -0,0 +1,72 @@
+{
+  "lane": "ci/test-and-lint",
+  "tests": [
+    {
+      "invariant": "write-quorum",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::inline_put_commit_path_tests::inline_put_direct_commit_accepts_exact_quorum_and_rejects_quorum_minus_one"
+    },
+    {
+      "invariant": "metadata-rollback",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::core::io_primitives::tests::write_unique_file_info_reverts_metadata_when_write_quorum_fails"
+    },
+    {
+      "invariant": "stale-writer",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::put_object_tmp_cleanup_tests::put_object_no_lock_aborts_after_outer_namespace_lock_loss"
+    },
+    {
+      "invariant": "range-body",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::object::transition_upload_integrity_tests::transitioned_compressed_object_range_get_returns_plaintext_slice"
+    },
+    {
+      "invariant": "multipart-cancellation",
+      "suite": "rustfs-ecstore",
+      "name": "set_disk::ops::multipart::tests::cancelled_complete_keeps_upload_lock_through_tail_cleanup"
+    },
+    {
+      "invariant": "list-uncommitted-version",
+      "suite": "rustfs-filemeta",
+      "name": "metacache::tests::resolve_with_write_quorum_slack_keeps_partial_latest_hidden_during_merge"
+    },
+    {
+      "invariant": "minio-object-fixture",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::parses_real_minio_object_xlmeta"
+    },
+    {
+      "invariant": "corrupt-part-arrays",
+      "suite": "rustfs-filemeta",
+      "name": "filemeta::test::crc_valid_but_part_arrays_corrupt_into_fileinfo_errors_not_panics"
+    }
+  ],
+  "fixtures": [
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_large_bin.xlmeta.hex",
+      "sha256": "e8093767806d701e639b48d023190e858fbc4cde69bcfd83c22af8cba8452ce5",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_small_txt.xlmeta.hex",
+      "sha256": "2a415ad3a3be5a9440035d4026ff880e0e8c1ec1701be9f4e077734e8dce03da",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/filemeta/tests/fixtures/minio/object_versioned_txt.xlmeta.hex",
+      "sha256": "7f21f50c326dd8b0228deb6dbdb7052b3d0a3f8ee6c85d43486f0e6bb7a97261",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/ecstore/tests/fixtures/minio/bucket_metadata.blob.hex",
+      "sha256": "f2b6e260aff106adf6039feb1c645686e84e75404ff725491fb18668be5db203",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    },
+    {
+      "path": "crates/ecstore/tests/fixtures/minio/bucket_metadata_full.xlmeta.hex",
+      "sha256": "3b6de589519c08a1614c8bd409bb8199c17d42043861b07bce513075e6fbfc12",
+      "source": "MinIO RELEASE.2025-07-23T15-54-02Z; crates/ecstore/tests/fixtures/minio/README.md"
+    }
+  ]
+}

.github/workflows/ci.yml

@@ -269,6 +269,7 @@ jobs:
           CARGO_BUILD_JOBS: ${{ (github.event_name == 'push' || github.event_name == 'workflow_dispatch') && '3' || '2' }}
         run: |
           mkdir -p artifacts/test-and-lint
+          rm -f target/nextest/ci/junit.xml
           ./scripts/ci/resource_sampler.sh start nextest
           trap './scripts/ci/resource_sampler.sh stop' EXIT
           set +e
@@ -277,6 +278,12 @@ jobs:
             --status-level all --final-status-level all \
             2>&1 | tee artifacts/test-and-lint/nextest.log
           status=${PIPESTATUS[0]}
+          if [[ "${status}" -eq 0 ]]; then
+            cargo nextest list --profile ci --all --exclude e2e_test --message-format json \
+              > artifacts/test-and-lint/core-test-listing.json \
+              && python3 scripts/check_test_wiring.py --check-core artifacts/test-and-lint/core-test-listing.json \
+              && test -s target/nextest/ci/junit.xml || status=$?
+          fi
           {
             echo "command=cargo nextest run --profile ci --all --exclude e2e_test"
             echo "exit_status=${status}"

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/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/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_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/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/store/init.rs

@@ -12033,6 +12033,7 @@ mod tests {
         pool_index: usize,
         bucket: &str,
         object: &str,
+        minio_unversioned: bool,
     ) {
         for disk_index in 0..4 {
             let metadata_path =
@@ -12066,6 +12067,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");
             }
@@ -12076,6 +12082,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)]
@@ -12116,7 +12268,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)
@@ -13257,7 +13409,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())
@@ -17326,6 +17478,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,

crates/filemeta/src/filemeta/version.rs

@@ -297,6 +297,20 @@ fn transitioned_version_from_bytes(value: Option<&[u8]>, state: TransitionVersio
     }
 }
 
+fn transition_version_metadata_value(raw: &[u8], decoded: Option<&str>) -> String {
+    decoded.map(str::to_owned).unwrap_or_else(|| {
+        if raw.is_empty() {
+            String::new()
+        } else {
+            String::from_utf8_lossy(raw).into_owned()
+        }
+    })
+}
+
+fn is_transition_version_metadata_key(key: &str) -> bool {
+    strip_internal_prefix_preserving_case(key).is_some_and(|suffix| suffix.eq_ignore_ascii_case(SUFFIX_TRANSITIONED_VERSION_ID))
+}
+
 fn validate_transition_version_state(state: TransitionVersionState, version: Option<&str>) -> Result<()> {
     let valid = match state {
         TransitionVersionState::Unknown | TransitionVersionState::KnownDisabled => version.is_none(),
@@ -366,14 +380,26 @@ impl<'a> DerivedInternalMetadata<'a> {
             }
             *slot = Some(value.as_slice());
         }
+        fn merge_consistent<'a>(canonical: Option<&'a [u8]>, legacy: Option<&'a [u8]>) -> Result<Option<&'a [u8]>> {
+            if let (Some(canonical), Some(legacy)) = (canonical, legacy)
+                && canonical != legacy
+            {
+                return Err(Error::FileCorrupt);
+            }
+            Ok(canonical.or(legacy))
+        }
+
         Ok(Self {
             checksum: canonical.checksum.or(legacy.checksum),
             part_checksums: canonical.part_checksums.or(legacy.part_checksums),
-            transition_status: canonical.transition_status.or(legacy.transition_status),
-            transitioned_object: canonical.transitioned_object.or(legacy.transitioned_object),
-            transitioned_version: canonical.transitioned_version.or(legacy.transitioned_version),
-            transitioned_version_state: canonical.transitioned_version_state.or(legacy.transitioned_version_state),
-            transition_tier: canonical.transition_tier.or(legacy.transition_tier),
+            transition_status: merge_consistent(canonical.transition_status, legacy.transition_status)?,
+            transitioned_object: merge_consistent(canonical.transitioned_object, legacy.transitioned_object)?,
+            transitioned_version: merge_consistent(canonical.transitioned_version, legacy.transitioned_version)?,
+            transitioned_version_state: merge_consistent(
+                canonical.transitioned_version_state,
+                legacy.transitioned_version_state,
+            )?,
+            transition_tier: merge_consistent(canonical.transition_tier, legacy.transition_tier)?,
         })
     }
 }
@@ -438,8 +464,14 @@ impl FileInfo {
     }
 }
 
-fn set_transition_version_state(meta_sys: &mut HashMap<String, Vec<u8>>, state: TransitionVersionState) {
-    if state == TransitionVersionState::Unknown {
+fn set_transition_version_state(
+    meta_sys: &mut HashMap<String, Vec<u8>>,
+    state: TransitionVersionState,
+    source_metadata: &HashMap<String, String>,
+) {
+    if state == TransitionVersionState::Unknown
+        && !rustfs_utils::http::metadata_compat::contains_key_str(source_metadata, SUFFIX_TRANSITIONED_VERSION_STATE)
+    {
         remove_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE);
     } else {
         insert_bytes(meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE, state.as_str().as_bytes().to_vec());
@@ -2643,6 +2675,11 @@ impl MetaObject {
         if derived_metadata.transitioned_version_state.is_some() {
             validate_transition_version_state(transition_version_state, transition_version.as_deref())?;
         }
+        for (key, value) in &self.meta_sys {
+            if is_transition_version_metadata_key(key) {
+                metadata.insert(key.to_owned(), transition_version_metadata_value(value, transition_version.as_deref()));
+            }
+        }
         let transition_version_id = transition_version.as_deref().and_then(|value| Uuid::parse_str(value).ok());
         let transition_tier = derived_metadata
             .transition_tier
@@ -2689,7 +2726,7 @@ impl MetaObject {
         } else {
             remove_bytes(&mut self.meta_sys, SUFFIX_TRANSITIONED_VERSION_ID);
         }
-        set_transition_version_state(&mut self.meta_sys, fi.transition_version_state);
+        set_transition_version_state(&mut self.meta_sys, fi.transition_version_state, &fi.metadata);
         insert_bytes(&mut self.meta_sys, SUFFIX_TRANSITION_TIER, fi.transition_tier.as_bytes().to_vec());
         if let Some(destination_id) = get_str(&fi.metadata, SUFFIX_TRANSITION_TIER_DESTINATION_ID) {
             insert_bytes(&mut self.meta_sys, SUFFIX_TRANSITION_TIER_DESTINATION_ID, destination_id.into_bytes());
@@ -2830,7 +2867,7 @@ impl From<FileInfo> for MetaObject {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, transition_version);
         }
         if !value.transition_status.is_empty() {
-            set_transition_version_state(&mut meta_sys, value.transition_version_state);
+            set_transition_version_state(&mut meta_sys, value.transition_version_state, &value.metadata);
         }
 
         if !value.transition_tier.is_empty() {
@@ -2985,6 +3022,12 @@ impl MetaDeleteMarker {
             fi.transition_version_state = transition_version_state_from_bytes(derived_metadata.transitioned_version_state)?;
             fi.transition_version =
                 transitioned_version_from_bytes(derived_metadata.transitioned_version, fi.transition_version_state);
+            for (key, value) in &self.meta_sys {
+                if is_transition_version_metadata_key(key) {
+                    fi.metadata
+                        .insert(key.to_owned(), transition_version_metadata_value(value, fi.transition_version.as_deref()));
+                }
+            }
             fi.transition_version_id = fi.transition_version.as_deref().and_then(|value| Uuid::parse_str(value).ok());
             if derived_metadata.transitioned_version_state.is_some() {
                 validate_transition_version_state(fi.transition_version_state, fi.transition_version.as_deref())?;
@@ -3152,7 +3195,7 @@ impl From<FileInfo> for MetaDeleteMarker {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITIONED_VERSION_ID, transition_version);
         }
         if !value.transition_status.is_empty() || value.tier_free_version() {
-            set_transition_version_state(&mut meta_sys, value.transition_version_state);
+            set_transition_version_state(&mut meta_sys, value.transition_version_state, &value.metadata);
         }
         if !value.transition_tier.is_empty() {
             insert_bytes(&mut meta_sys, SUFFIX_TRANSITION_TIER, value.transition_tier.as_bytes().to_vec());
@@ -4574,6 +4617,7 @@ mod tests {
             .into_fileinfo("b", "k", false)
             .expect("into_fileinfo");
         assert_eq!(fi.transition_version_id, None);
+        assert_eq!(get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID), Some(String::new()));
     }
 
     #[test]
@@ -4585,6 +4629,10 @@ mod tests {
             .into_fileinfo("b", "k", false)
             .expect("into_fileinfo");
         assert_eq!(fi.transition_version_id, None);
+        assert!(
+            get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID).is_some_and(|value| !value.is_empty()),
+            "nil UUID bytes must remain distinguishable from an empty MinIO version"
+        );
     }
 
     #[test]
@@ -4598,6 +4646,7 @@ mod tests {
         assert_eq!(fi.transition_version_id, Some(id));
         assert_eq!(fi.transition_version, Some(id.to_string()));
         assert_eq!(fi.transition_version_state, TransitionVersionState::Unknown);
+        assert_eq!(get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID), Some(id.to_string()));
     }
 
     #[test]
@@ -4637,6 +4686,36 @@ mod tests {
         assert_eq!(fi.transition_version_state, TransitionVersionState::Unknown);
     }
 
+    #[test]
+    fn meta_object_transition_version_state_explicit_unknown_is_not_legacy_missing() {
+        let mut metadata = HashMap::new();
+        rustfs_utils::http::metadata_compat::insert_str(
+            &mut metadata,
+            SUFFIX_TRANSITIONED_VERSION_STATE,
+            TransitionVersionState::Unknown.as_str().to_string(),
+        );
+        let fi = FileInfo {
+            transition_status: "complete".to_string(),
+            transition_version_state: TransitionVersionState::Unknown,
+            metadata,
+            ..Default::default()
+        };
+
+        let object = MetaObject::from(fi);
+        assert_eq!(
+            get_consistent_bytes(&object.meta_sys, SUFFIX_TRANSITIONED_VERSION_STATE),
+            Some(b"unknown".as_slice())
+        );
+        let decoded = object
+            .into_fileinfo("b", "k", false)
+            .expect("explicit unknown state should decode");
+        assert_eq!(decoded.transition_version_state, TransitionVersionState::Unknown);
+        assert_eq!(
+            rustfs_utils::http::metadata_compat::get_consistent_str(&decoded.metadata, SUFFIX_TRANSITIONED_VERSION_STATE,),
+            Some("unknown")
+        );
+    }
+
     #[test]
     fn meta_object_transition_version_state_exact_round_trips_dual_keys() {
         let id = sample_version_id();
@@ -4753,6 +4832,10 @@ mod tests {
             .expect("invalid transition version bytes must not fail the object read");
         assert_eq!(fi.transition_version_id, None);
         assert_eq!(fi.transition_version, None);
+        assert!(
+            get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID).is_some_and(|value| !value.is_empty()),
+            "invalid raw bytes must remain distinguishable from an empty MinIO version"
+        );
     }
 
     #[test]
@@ -4795,6 +4878,10 @@ mod tests {
         .into_fileinfo("b", "k", false)
         .expect("nil tier version should remain an absent remote version");
         assert_eq!(fi.transition_version_id, None);
+        assert!(
+            get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID).is_some_and(|value| !value.is_empty()),
+            "nil UUID bytes must remain distinguishable from an empty MinIO version"
+        );
     }
 
     #[test]
@@ -4812,6 +4899,7 @@ mod tests {
         .expect("legacy binary UUID tier version should decode");
         assert_eq!(fi.transition_version_id, Some(id));
         assert_eq!(fi.transition_version, Some(id.to_string()));
+        assert_eq!(get_str(&fi.metadata, SUFFIX_TRANSITIONED_VERSION_ID), Some(id.to_string()));
     }
 
     #[test]
@@ -4910,6 +4998,23 @@ mod tests {
         assert_eq!(err, Error::FileCorrupt);
     }
 
+    #[test]
+    fn meta_object_transition_version_state_mixed_case_alias_conflict_fails_closed() {
+        let sys = HashMap::from([
+            (
+                format!("{RUSTFS_INTERNAL_PREFIX}{SUFFIX_TRANSITIONED_VERSION_STATE}"),
+                b"unknown".to_vec(),
+            ),
+            ("X-Minio-Internal-transitioned-version-state".to_string(), b"exact".to_vec()),
+        ]);
+
+        let err = make_meta_object_with_sys(sys)
+            .into_fileinfo("b", "k", false)
+            .expect_err("mixed-case transition state aliases must agree");
+
+        assert_eq!(err, Error::FileCorrupt);
+    }
+
     #[test]
     fn version_header_sorts_before_prefers_object_over_delete_marker_on_equal_mod_time() {
         let object = FileMetaVersionHeader {

docs/testing/ci-gates.md

@@ -109,3 +109,13 @@ Use an exact preview tag for an end-to-end release rehearsal. Manual dispatches
 ## Change checklist
 
 Update this file in the same PR when a job or check name changes, a workflow gains or loses a `pull_request` or `schedule` trigger, required contexts or strict/merge-queue policy change, report-only vs gating semantics change, or `.github/scheduled-validations.json` membership changes. Do not copy timeouts, crons, or test counts here.
+
+## ECStore invariant selection
+
+The existing `ci.yml` test-and-lint job runs the ordinary ECStore and filemeta tests. After that run, `scripts/check_test_wiring.py --check-core` checks the same nextest profile and package selection against `.config/ecstore-required-tests.json`. Every named test must exist, match the filter, and be non-ignored; the job also requires a nonempty JUnit report. This checks membership without running the tests twice. `core-test-listing.json`, JUnit, and the run log are retained in the existing test-and-lint artifact.
+
+The manifest records a minimum set of invariants: write quorum, metadata rollback, stale-writer lock loss, plaintext Range content, multipart cancellation, hiding uncommitted LIST versions, real MinIO metadata, and corrupt part arrays. Renaming or moving a required test must update the manifest in the same change after checking the compiled listing. Extend this list as new deterministic regressions land; it is not a claim that all storage invariants are covered.
+
+The checked-in MinIO corpus is pinned by file SHA256 and its documented source release. The static wiring guard and the CI selection check both reject missing or changed fixtures. These are metadata fixtures, not a legacy shard-body corpus or proof of crash durability. Optional `legacy_bitrot_read_test` runs may still skip when their external corpus is absent; they do not satisfy a required compatibility lane. Real encrypted fixture reads remain in `minio-interop.yml`, and multi-node fault schedules remain in the existing nightly cluster lane. In-process reopen tests do not establish power-loss durability.
+
+Run `python3 scripts/check_test_wiring.py --self-test` to exercise the negative cases: removed/ignored/filtered tests, malformed listing, absent fixtures, and wrong fixture hashes. Do not update hashes merely to silence the guard; a fixture change needs source/provenance and compatibility review.

rustfs/src/app/object/copy.rs

@@ -394,7 +394,7 @@ impl DefaultObjectUsecase {
         // Bucket metadata uses the bucket name as its namespace-lock key. Load
         // every copy-time bucket snapshot before a same-object key can collide
         // with that key (for example, copying `bucket/bucket` onto itself).
-        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
+        let bucket_sse_config = load_bucket_default_sse_config(&bucket).await?;
         let object_lock_config_state = load_bucket_object_lock_config_state(&bucket).await?;
         if cp_src_dst_same && key == bucket {
             dst_opts.object_lock_config_snapshot =
@@ -1388,4 +1388,48 @@ mod tests {
             .unwrap_err();
         assert_eq!(err.code(), &S3ErrorCode::InvalidRequest);
     }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_copy_object_refuses_a_bucket_whose_encryption_config_is_unreadable() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("copy-sse-unreadable-{}", Uuid::new_v4());
+        let source = "source.bin";
+        let destination = "destination.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("unreadable-encryption copy bucket must be created");
+        let mut reader = PutObjReader::from_vec(b"copied while the bucket still had a readable configuration".to_vec());
+        store
+            .put_object(&bucket, source, &mut reader, &ObjectOptions::default())
+            .await
+            .expect("copy source object must be written");
+        install_unreadable_bucket_sse_config(&bucket).await;
+
+        let input = CopyObjectInput::builder()
+            .copy_source(CopySource::Bucket {
+                bucket: bucket.clone().into(),
+                key: source.into(),
+                version_id: None,
+            })
+            .bucket(bucket.clone())
+            .key(destination.to_string())
+            .build()
+            .expect("copy input must build");
+        let usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+
+        let err = Box::pin(usecase.execute_copy_object(build_request(input, Method::PUT)))
+            .await
+            .expect_err("an unreadable bucket encryption configuration must refuse the copy");
+
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        let lookup_err = store
+            .get_object_info(&bucket, destination, &ObjectOptions::default())
+            .await
+            .expect_err("a refused copy must not leave a destination object behind");
+        assert!(is_err_object_not_found(&lookup_err), "{lookup_err}");
+    }
 }

rustfs/src/app/object/extract.rs

@@ -2037,7 +2037,7 @@ impl DefaultObjectUsecase {
         let sse_customer_key_md5 = sse_customer_key_md5.or(h_md5);
 
         let original_sse = server_side_encryption.or(extract_server_side_encryption_from_headers(&req.headers)?);
-        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
+        let bucket_sse_config = load_bucket_default_sse_config(&bucket).await?;
         let (mut effective_sse, mut effective_kms_key_id) = resolve_bucket_default_sse(
             bucket_sse_config.as_ref().map(|(config, _timestamp)| config),
             original_sse,

rustfs/src/app/object/put.rs

@@ -1489,8 +1489,9 @@ impl DefaultObjectUsecase {
         };
 
         let sse_config_stage_start = put_stage_metrics_enabled.then(Instant::now);
-        let bucket_sse_config = metadata_sys::get_sse_config(&bucket).await.ok();
+        let bucket_sse_config = load_bucket_default_sse_config(&bucket).await;
         rustfs_io_metrics::record_put_object_stage_duration_from("app_sse_config_lookup", sse_config_stage_start);
+        let bucket_sse_config = bucket_sse_config?;
         debug!(
             target: "rustfs::app::object_usecase",
             component = "app",
@@ -3923,4 +3924,121 @@ mod tests {
             .expect_err("writes after the zero-byte quota update must be denied");
         assert!(matches!(err, StorageError::QuotaExceeded { current: 4096, limit: 0 }));
     }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_put_object_refuses_a_bucket_whose_encryption_config_is_unreadable() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("put-sse-unreadable-{}", Uuid::new_v4());
+        let object = "object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("unreadable-encryption PUT bucket must be created");
+        install_unreadable_bucket_sse_config(&bucket).await;
+
+        let payload = Bytes::from_static(b"an operator mandated encryption for this bucket");
+        let input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
+            .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
+            .build()
+            .expect("PUT input must build");
+        let usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+
+        let err = Box::pin(usecase.execute_put_object(&FS::new(), build_request(input, Method::PUT)))
+            .await
+            .expect_err("an unreadable bucket encryption configuration must refuse the write");
+
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        let lookup_err = store
+            .get_object_info(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect_err("a refused PUT must not leave an object behind");
+        assert!(is_err_object_not_found(&lookup_err), "{lookup_err}");
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_put_object_still_writes_plaintext_without_bucket_encryption() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("put-sse-absent-{}", Uuid::new_v4());
+        let object = "object.bin";
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("plaintext PUT bucket must be created");
+
+        let payload = Bytes::from_static(b"no default encryption is configured for this bucket");
+        let input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key(object.to_string())
+            .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
+            .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
+            .build()
+            .expect("PUT input must build");
+        let usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+
+        Box::pin(usecase.execute_put_object(&FS::new(), build_request(input, Method::PUT)))
+            .await
+            .expect("a bucket without default encryption must still accept a plaintext write");
+
+        let stored = store
+            .get_object_info(&bucket, object, &ObjectOptions::default())
+            .await
+            .expect("the plaintext object must be readable");
+        assert_eq!(stored.size, i64::try_from(payload.len()).expect("test payload length must fit i64"));
+        assert!(
+            !stored
+                .user_defined
+                .keys()
+                .any(|key| key.eq_ignore_ascii_case(AMZ_SERVER_SIDE_ENCRYPTION)
+                    || key.starts_with("x-rustfs-encryption-")
+                    || key.starts_with("x-minio-encryption-")),
+            "the object must carry no encryption metadata: {:?}",
+            stored.user_defined
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn execute_put_object_extract_refuses_a_bucket_whose_encryption_config_is_unreadable() {
+        use crate::app::storage_api::test::contract::bucket::{BucketOperations as _, MakeBucketOptions};
+
+        let (store, context) = real_store_test_context().await;
+        let bucket = format!("extract-sse-unreadable-{}", Uuid::new_v4());
+        store
+            .make_bucket(&bucket, &MakeBucketOptions::default())
+            .await
+            .expect("unreadable-encryption extract bucket must be created");
+        install_unreadable_bucket_sse_config(&bucket).await;
+
+        let payload = Bytes::from_static(b"archive bytes that must never be unpacked in plaintext");
+        let input = PutObjectInput::builder()
+            .bucket(bucket.clone())
+            .key("archive.tar".to_string())
+            .body(Some(StreamingBlob::from(s3s::Body::from(payload.clone()))))
+            .content_length(Some(i64::try_from(payload.len()).expect("test payload length must fit i64")))
+            .build()
+            .expect("extract PUT input must build");
+        let mut req = build_request(input, Method::PUT);
+        req.headers.insert(AMZ_SNOWBALL_EXTRACT, HeaderValue::from_static("true"));
+        let usecase = DefaultObjectUsecase::with_context(Some(Arc::clone(&context)));
+
+        let err = Box::pin(usecase.execute_put_object(&FS::new(), req))
+            .await
+            .expect_err("an unreadable bucket encryption configuration must refuse the extract upload");
+
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+        let lookup_err = store
+            .get_object_info(&bucket, "archive.tar", &ObjectOptions::default())
+            .await
+            .expect_err("a refused extract upload must not leave an object behind");
+        assert!(is_err_object_not_found(&lookup_err), "{lookup_err}");
+    }
 }

rustfs/src/app/object/shared.rs

@@ -269,6 +269,129 @@ pub(super) fn resolve_bucket_default_sse(
     (effective_sse, effective_kms_key_id)
 }
 
+/// The bucket's default encryption configuration for a write path.
+///
+/// `Ok(None)` carries one meaning only — this bucket has no default encryption
+/// — and the write proceeds in plaintext exactly as before. Every other
+/// outcome refuses the write rather than collapsing onto that same value: an
+/// encryption blob that exists but cannot be read fails closed in
+/// `get_sse_config` since rustfs/rustfs#7172, and swallowing that error here
+/// stores plaintext into a bucket whose operator mandated encryption, with
+/// nothing returned to the client and nothing in the object to tell it apart
+/// afterwards (rustfs/backlog#2287).
+///
+/// The states the lookup can report, and what each one does:
+///
+/// * configured and readable — apply the bucket default;
+/// * no encryption blob at all, including a bucket that does not exist and a
+///   bucket whose metadata document is absent — `ConfigNotFound`, so a cold
+///   cache and a missing bucket are never turned into a refusal, and the write
+///   still fails later with its own `NoSuchBucket`;
+/// * blob present but unparseable — deterministic, so retrying cannot help;
+///   surfaces as `InternalError` until an operator repairs or removes it;
+/// * the metadata read itself failed (namespace lock, quorum, disk, an
+///   uninitialized metadata system) — transient, and the typed error maps to
+///   the retryable `ServiceUnavailable`.
+///
+/// The last two are distinguished by the typed error the accessor returns, not
+/// re-derived here: [`ApiError`] already separates them. This mirrors
+/// `prepare_sse_configuration` in `storage::sse`, the resolver the multipart
+/// writer uses, which has always failed closed on the same lookup.
+pub(super) async fn load_bucket_default_sse_config(
+    bucket: &str,
+) -> S3Result<Option<(ServerSideEncryptionConfiguration, OffsetDateTime)>> {
+    classify_bucket_default_sse_lookup(bucket, metadata_sys::get_sse_config(bucket).await)
+}
+
+fn classify_bucket_default_sse_lookup(
+    bucket: &str,
+    lookup: Result<(ServerSideEncryptionConfiguration, OffsetDateTime), StorageError>,
+) -> S3Result<Option<(ServerSideEncryptionConfiguration, OffsetDateTime)>> {
+    match lookup {
+        Ok(config) => Ok(Some(config)),
+        Err(err) if err == StorageError::ConfigNotFound => Ok(None),
+        Err(err) => {
+            let api_error = ApiError::from(err);
+            error!(
+                event = "bucket_sse_config_lookup_failed",
+                component = LOG_COMPONENT_APP,
+                subsystem = LOG_SUBSYSTEM_OBJECT,
+                result = "write_refused",
+                bucket = %bucket,
+                code = %api_error.code.as_str(),
+                error = %api_error,
+                "Bucket default encryption is unreadable; refusing the write instead of storing plaintext"
+            );
+            Err(api_error.into())
+        }
+    }
+}
+
+#[cfg(test)]
+mod bucket_default_sse_lookup_tests {
+    use super::*;
+    use s3s::dto::{ServerSideEncryptionByDefault, ServerSideEncryptionRule};
+    use time::OffsetDateTime;
+
+    fn sse_config() -> ServerSideEncryptionConfiguration {
+        ServerSideEncryptionConfiguration {
+            rules: vec![ServerSideEncryptionRule {
+                apply_server_side_encryption_by_default: Some(ServerSideEncryptionByDefault {
+                    sse_algorithm: ServerSideEncryption::from_static(ServerSideEncryption::AES256),
+                    kms_master_key_id: None,
+                }),
+                blocked_encryption_types: None,
+                bucket_key_enabled: None,
+            }],
+        }
+    }
+
+    #[test]
+    fn an_absent_configuration_still_writes_plaintext() {
+        let resolved = classify_bucket_default_sse_lookup("bucket", Err(StorageError::ConfigNotFound))
+            .expect("a bucket without default encryption must keep writing plaintext");
+
+        assert!(resolved.is_none());
+        assert_eq!(resolve_bucket_default_sse(None, None, None, false), (None, None));
+    }
+
+    #[test]
+    fn a_readable_configuration_is_returned() {
+        let resolved = classify_bucket_default_sse_lookup("bucket", Ok((sse_config(), OffsetDateTime::UNIX_EPOCH)))
+            .expect("a readable configuration must not refuse the write")
+            .expect("a readable configuration must be applied");
+
+        assert_eq!(resolved.0.rules.len(), 1);
+    }
+
+    #[test]
+    fn an_unreadable_configuration_refuses_the_write() {
+        let err = classify_bucket_default_sse_lookup(
+            "bucket",
+            Err(StorageError::other("persisted bucket encryption configuration is invalid")),
+        )
+        .expect_err("a corrupt encryption blob must never degrade to plaintext");
+
+        assert_eq!(err.code(), &S3ErrorCode::InternalError);
+    }
+
+    #[test]
+    fn an_unavailable_metadata_read_refuses_the_write_as_retryable() {
+        let err = classify_bucket_default_sse_lookup("bucket", Err(StorageError::ErasureReadQuorum))
+            .expect_err("an unreadable metadata subsystem must never degrade to plaintext");
+
+        assert_eq!(err.code(), &S3ErrorCode::ServiceUnavailable);
+    }
+
+    #[test]
+    fn a_missing_bucket_keeps_its_own_error() {
+        let err = classify_bucket_default_sse_lookup("bucket", Err(StorageError::BucketNotFound("bucket".to_string())))
+            .expect_err("a bucket-not-found lookup must not be reported as an encryption failure");
+
+        assert_eq!(err.code(), &S3ErrorCode::NoSuchBucket);
+    }
+}
+
 #[cfg(test)]
 mod deadlock_request_guard_tests {
     use super::DeadlockRequestGuard;

rustfs/src/app/object/test_support.rs

@@ -96,3 +96,36 @@ pub(super) fn real_cold_fill_plan(
     };
     plan
 }
+
+/// A store with an ambient `AppContext`, for tests that drive a handler end to
+/// end without the object-data-cache overrides of
+/// [`real_cold_fill_test_context`].
+pub(super) async fn real_store_test_context() -> (Arc<ECStore>, Arc<AppContext>) {
+    let store = crate::app::gating_test_env::shared_gating_ecstore().await;
+    if current_app_context().is_none() {
+        crate::app::runtime_sources::install_test_app_context(Arc::clone(&store)).await;
+    }
+    let ambient = current_app_context().expect("real-store tests require an ambient AppContext");
+    let context = Arc::new(AppContext::new(Arc::clone(&store), ambient.iam(), ambient.kms()));
+    (store, context)
+}
+
+/// Leave the bucket in the state a damaged encryption blob produces: the raw
+/// document is retained and the typed configuration stays `None`, which is the
+/// durable "exists but cannot be read" signal `get_sse_config` fails closed on
+/// (rustfs/rustfs#7172).
+pub(super) async fn install_unreadable_bucket_sse_config(bucket: &str) {
+    use crate::app::storage_api::test::{get_global_bucket_metadata_sys, set_bucket_metadata};
+
+    let sys = get_global_bucket_metadata_sys().expect("bucket metadata system must be initialized");
+    let metadata = {
+        let sys = sys.read().await;
+        sys.get(bucket).await.expect("bucket metadata must be cached")
+    };
+    let mut metadata = (*metadata).clone();
+    metadata.encryption_config_xml = b"<ServerSideEncryptionConfiguration>truncated".to_vec();
+    metadata.sse_config = None;
+    set_bucket_metadata(bucket.to_string(), metadata)
+        .await
+        .expect("unreadable bucket encryption configuration must be installed");
+}

scripts/check_test_wiring.py

@@ -764,8 +764,53 @@ def check_profile_listing(root: Path, profile: str, listing: Path) -> list[str]:
     return []
 
 
+def core_requirements(root: Path) -> dict:
+    data = json.loads((root / ".config/ecstore-required-tests.json").read_text())
+    if not data["tests"] or not data["fixtures"]:
+        raise ValueError("core test and fixture requirements must not be empty")
+    identities = [(test["suite"], test["name"]) for test in data["tests"]]
+    if len(set(identities)) != len(identities):
+        raise ValueError("duplicate core test requirement")
+    return data
+
+
+def check_core_fixtures(root: Path) -> list[str]:
+    try:
+        fixtures = core_requirements(root)["fixtures"]
+        errors = []
+        for fixture in fixtures:
+            path = (root / fixture["path"]).resolve()
+            if not path.is_relative_to(root.resolve()):
+                raise ValueError("core fixture path escapes repository")
+            if not path.is_file():
+                errors.append(f"{fixture['path']}: required core fixture missing")
+            elif hashlib.sha256(path.read_bytes()).hexdigest() != fixture["sha256"]:
+                errors.append(f"{fixture['path']}: core fixture sha256 mismatch")
+        return errors
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        return [f"cannot validate core fixtures: {error}"]
+
+
+def check_core_listing(root: Path, listing: Path) -> list[str]:
+    """Check the existing CI run's selection, not a second filtered test run."""
+    try:
+        required = core_requirements(root)["tests"]
+        suites = json.loads(listing.read_text())["rust-suites"]
+        if not isinstance(suites, dict):
+            raise ValueError("rust-suites must be an object")
+        errors = check_core_fixtures(root)
+        for test in required:
+            testcase = suites.get(test["suite"], {}).get("testcases", {}).get(test["name"], {})
+            if testcase.get("ignored") is not False or testcase.get("filter-match", {}).get("status") != "matches":
+                errors.append(f"{test['invariant']}: required test not selected: {test['suite']}::{test['name']}")
+        return errors
+    except (OSError, KeyError, TypeError, ValueError) as error:
+        return [f"cannot read core nextest listing: {error}"]
+
+
 def validate(root: Path) -> list[str]:
     errors: list[str] = []
+    errors.extend(check_core_fixtures(root))
     errors.extend(check_e2e_modules(root))
     errors.extend(check_vault_test_groups(root))
     errors.extend(check_ilm_build_budget(root))
@@ -779,6 +824,32 @@ def validate(root: Path) -> list[str]:
 
 
 class SelfTests(unittest.TestCase):
+    def test_core_gate_rejects_missing_ignored_filtered_and_corrupt_inputs(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            root = Path(tmp)
+            (root / ".config").mkdir()
+            fixture = root / "fixture.hex"
+            fixture.write_text("4142")
+            requirements = {
+                "tests": [{"invariant": "commit", "suite": "store", "name": "commit_test"}],
+                "fixtures": [{"path": "fixture.hex", "sha256": hashlib.sha256(fixture.read_bytes()).hexdigest()}],
+            }
+            (root / ".config/ecstore-required-tests.json").write_text(json.dumps(requirements))
+            listing = root / "listing.json"
+            good = {"ignored": False, "filter-match": {"status": "matches"}}
+            for case, testcase in (("selected", good), ("missing", {}), ("ignored", dict(good, ignored=True)),
+                                   ("filtered", dict(good, **{"filter-match": {"status": "mismatch"}}))):
+                with self.subTest(case=case):
+                    listing.write_text(json.dumps({"rust-suites": {"store": {"testcases": {"commit_test": testcase}}}}))
+                    self.assertEqual(bool(check_core_listing(root, listing)), case != "selected")
+            listing.write_text(json.dumps({"rust-suites": {"store": {"testcases": {"commit_test": good}}}}))
+            fixture.write_text("4143")
+            self.assertIn("sha256 mismatch", check_core_listing(root, listing)[0])
+            fixture.unlink()
+            self.assertIn("fixture missing", check_core_listing(root, listing)[0])
+            listing.write_text("not json")
+            self.assertIn("cannot read", check_core_listing(root, listing)[0])
+
     def test_ilm_lane_keeps_the_measured_cargo_build_budget(self) -> None:
         with tempfile.TemporaryDirectory() as tmp:
             root = Path(tmp)
@@ -981,6 +1052,7 @@ class SelfTests(unittest.TestCase):
                 mock.patch(__name__ + ".check_e2e_modules", return_value=[]),
                 mock.patch(__name__ + ".check_vault_test_groups", return_value=[]),
                 mock.patch(__name__ + ".check_fuzz_targets", return_value=[]),
+                mock.patch(__name__ + ".check_core_fixtures", return_value=[]),
                 mock.patch(__name__ + ".check_runner_selection", return_value=[]),
                 mock.patch(__name__ + ".check_workflow_readiness", return_value=[]),
                 mock.patch(__name__ + ".check_profile_definitions", return_value=[]),
@@ -1393,6 +1465,11 @@ def main() -> int:
     if sys.argv[1:] == ["--self-test"]:
         suite = unittest.defaultTestLoader.loadTestsFromTestCase(SelfTests)
         return 0 if unittest.TextTestRunner(verbosity=2).run(suite).wasSuccessful() else 1
+    if len(sys.argv) == 3 and sys.argv[1] == "--check-core":
+        errors = check_core_listing(ROOT, Path(sys.argv[2]))
+        for error in errors:
+            print(f"ERROR: {error}", file=sys.stderr)
+        return 1 if errors else 0
     if len(sys.argv) == 4 and sys.argv[1] == "--check-profile":
         errors = check_profile_listing(ROOT, sys.argv[2], Path(sys.argv[3]))
         if errors:
@@ -1410,7 +1487,7 @@ def main() -> int:
         return 0
     if sys.argv[1:]:
         print(
-            "usage: check_test_wiring.py [--self-test | --check-profile PROFILE LISTING | "
+            "usage: check_test_wiring.py [--self-test | --check-core LISTING | --check-profile PROFILE LISTING | "
             "--update-profile PROFILE LISTING PLATFORM]",
             file=sys.stderr,
         )