test(ecstore): add shared MockWarmBackend test utility for lifecycle and tier tests (#4716)

* test(ecstore): extract shared MockWarmBackend into a test-util feature (backlog#1148 ilm-6)

The tier/lifecycle integration tests carried two byte-for-byte copies of an
in-memory WarmBackend mock — one in crates/scanner/tests and one in
rustfs/src/app — plus duplicated register_mock_tier and polling helpers. Both
implemented the same ecstore WarmBackend trait.

Consolidate them into ecstore behind a new `test-util` feature, exposed via the
`rustfs_ecstore::api::tier::test_util` facade:

- MockWarmBackend: in-memory WarmBackend with an operation log (for ordering
  assertions such as "local delete precedes remote remove") and fault injection
  (FaultConfig): unreachable, HTTP 5xx, credential rejection, injected latency,
  plus external_remove to simulate an out-of-band remote deletion.
- register_mock_tier / register_mock_tier_backend: register the mock into any
  TierConfigMgr handle (the global manager used by scanner tests or a
  per-instance one used by the app tests).
- xl.meta transition assertion helpers: read_transition_meta,
  assert_transition_meta_consistent (cross-shard consistency of the
  status/tier/remote-key/remote-version-id tuple plus free-version count), and
  free_version_count.
- polling helpers: wait_for_remote_absence, wait_for_object_count,
  wait_for_free_version_absence.

Both existing copies now consume this single definition; `rg 'struct
MockWarmBackend'` collapses to one. The feature is enabled only from
[dev-dependencies], so it never links into the production binary (resolver 3).

Designed for downstream ilm-8 (restore lifecycle) and ilm-11 (tier fault
injection matrix). Coordinates with #4706 (ilm-2), which adds op-logging to the
scanner mock — that op-logging is now part of this shared surface, so #4706
should rebase onto it.

Refs rustfs/backlog#1148 (ilm-6), rustfs/backlog#1155.

* test(ecstore): fix shared MockWarmBackend usage after main merge

- Access stored objects via MockWarmBackend::contains() instead of the now
  private inner objects map (fixes E0609 after the shared test-util refactor).
- Drop dead ReadCloser/ReaderImpl/DiskAPI imports and the unused
  transition_api test re-exports the mock extraction left behind.
- Reword the scanner/rustfs test-util dependency comments so they no longer
  embed the literal rustfs_ecstore:: path that trips the ECStore
  architecture-migration guard.

crates/ecstore/Cargo.toml

@@ -34,6 +34,10 @@ workspace = true
 default = []
 rio-v2 = ["dep:rustfs-rio-v2"]
 hotpath = ["dep:hotpath", "hotpath/hotpath", "rustfs-filemeta/hotpath", "rustfs-rio/hotpath"]
+# Exposes shared lifecycle/tier test utilities (MockWarmBackend, fault
+# injection, xl.meta transition assertions) via `api::tier::test_util`.
+# Enable only from `[dev-dependencies]` (rustfs/backlog#1148 ilm-6).
+test-util = []
 
 [dependencies]
 hotpath = { workspace = true, optional = true }

crates/ecstore/src/api/mod.rs

@@ -429,4 +429,13 @@ pub mod tier {
             WarmBackend, WarmBackendGetOpts, WarmBackendImpl, build_transition_put_options, check_warm_backend, new_warm_backend,
         };
     }
+
+    #[cfg(feature = "test-util")]
+    pub mod test_util {
+        pub use crate::services::tier::test_util::{
+            FaultConfig, MockStoredObject, MockWarmBackend, MockWarmOp, TransitionMeta, assert_transition_meta_consistent,
+            free_version_count, read_transition_meta, register_mock_tier, register_mock_tier_backend,
+            wait_for_free_version_absence,
+        };
+    }
 }

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

@@ -12,6 +12,8 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+#[cfg(feature = "test-util")]
+pub mod test_util;
 pub mod tier;
 pub mod tier_admin;
 pub mod tier_config;

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

@@ -0,0 +1,611 @@
+// Copyright 2024 RustFS Team
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
+//! Shared test utilities for tier / lifecycle integration tests.
+//!
+//! Gated behind the `test-util` feature (`rustfs/backlog#1148` ilm-6). This
+//! module used to be duplicated verbatim in two places — the scanner
+//! integration tests (`crates/scanner/tests/lifecycle_integration_test.rs`)
+//! and the app-layer transition tests
+//! (`rustfs/src/app/lifecycle_transition_api_test.rs`). Both now consume this
+//! single definition. Consuming it from an external test crate requires
+//! enabling the feature as a dev-dependency, for example:
+//!
+//! ```toml
+//! [dev-dependencies]
+//! rustfs-ecstore = { workspace = true, features = ["test-util"] }
+//! ```
+//!
+//! and importing through the crate facade
+//! (`rustfs_ecstore::api::tier::test_util`).
+//!
+//! What it provides:
+//!
+//! - [`MockWarmBackend`]: an in-memory [`WarmBackend`] implementation backed by
+//!   a `HashMap`. It records an operation log (for ordering assertions such as
+//!   "local delete precedes remote remove") and supports fault injection
+//!   ([`FaultConfig`]): unreachable backend, HTTP 5xx, credential rejection, and
+//!   injected latency. It also exposes [`MockWarmBackend::external_remove`] to
+//!   simulate a remote object being deleted out-of-band.
+//! - [`register_mock_tier`]: register a [`MockWarmBackend`] as a tier in a
+//!   [`TierConfigMgr`] handle (works with both the global manager and a
+//!   per-instance manager).
+//! - xl.meta transition-state assertion helpers ([`read_transition_meta`],
+//!   [`assert_transition_meta_consistent`], [`free_version_count`]): read the
+//!   on-disk `xl.meta` and assert the `(status, tier, remote object, remote
+//!   version id)` tuple plus free-version count, cross-checking every shard
+//!   disk for consistency.
+//! - Polling helpers ([`MockWarmBackend::wait_for_remote_absence`],
+//!   [`MockWarmBackend::wait_for_object_count`],
+//!   [`wait_for_free_version_absence`]) so tests never assert a single read
+//!   against asynchronously-applied state.
+//!
+//! Downstream consumers: ilm-8 (restore lifecycle) and ilm-11 (tier fault
+//! injection matrix) build directly on this surface.
+
+use std::collections::HashMap;
+use std::io::Cursor;
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use async_trait::async_trait;
+use tokio::io::AsyncReadExt;
+use tokio::sync::{Mutex, RwLock};
+use uuid::Uuid;
+
+use crate::client::transition_api::{ReadCloser, ReaderImpl};
+use crate::disk::endpoint::Endpoint;
+use crate::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, new_disk};
+use crate::services::tier::tier::TierConfigMgr;
+use crate::services::tier::tier_config::{TierConfig, TierMinIO, TierType};
+use crate::services::tier::warm_backend::{WarmBackend, WarmBackendGetOpts, build_transition_put_options};
+use rustfs_filemeta::FileMeta;
+use rustfs_utils::path::path_join_buf;
+
+/// Default polling cadence used by the `wait_for_*` helpers.
+const POLL_INTERVAL: Duration = Duration::from_millis(50);
+
+/// A fault to inject into [`MockWarmBackend`] operations.
+///
+/// Faults are checked at the start of every trait method. When set, the backend
+/// returns a representative [`std::io::Error`] instead of performing the
+/// operation, letting tests exercise transition/restore/sweep error paths
+/// without a real remote tier. Latency, if set, is applied before the fault
+/// check so slow-and-failing behaviour can be modelled together.
+#[derive(Clone, Debug, Default)]
+pub struct FaultConfig {
+    /// Simulate a backend that cannot be reached (connection refused).
+    pub unreachable: bool,
+    /// Simulate an HTTP 5xx server error on every operation.
+    pub server_error: bool,
+    /// Simulate rejected / rotated credentials (permission denied).
+    pub reject_credentials: bool,
+    /// Inject latency before each operation resolves.
+    pub latency: Option<Duration>,
+}
+
+impl FaultConfig {
+    fn error(&self) -> Option<std::io::Error> {
+        use std::io::{Error, ErrorKind};
+        if self.unreachable {
+            return Some(Error::new(ErrorKind::ConnectionRefused, "mock warm backend unreachable"));
+        }
+        if self.reject_credentials {
+            return Some(Error::new(ErrorKind::PermissionDenied, "mock warm backend rejected credentials"));
+        }
+        if self.server_error {
+            return Some(Error::other("mock warm backend returned HTTP 503"));
+        }
+        None
+    }
+}
+
+/// A single operation recorded by [`MockWarmBackend`].
+///
+/// Ordering assertions (for example, "the local delete lands before any remote
+/// remove" from the expire/GET race regression) read this log via
+/// [`MockWarmBackend::op_log`]. [`MockWarmOp::ExternalRemove`] is only produced
+/// by [`MockWarmBackend::external_remove`], never by the trait `remove`.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub enum MockWarmOp {
+    Put { object: String },
+    Get { object: String },
+    Remove { object: String },
+    ExternalRemove { object: String },
+    InUse,
+}
+
+/// A stored object inside [`MockWarmBackend`].
+#[derive(Clone, Default)]
+pub struct MockStoredObject {
+    pub bytes: Vec<u8>,
+    pub metadata: HashMap<String, String>,
+    pub remote_version_id: String,
+}
+
+#[derive(Default)]
+struct MockWarmBackendInner {
+    objects: Mutex<HashMap<String, MockStoredObject>>,
+    faults: Mutex<FaultConfig>,
+    op_log: Mutex<Vec<MockWarmOp>>,
+}
+
+/// In-memory [`WarmBackend`] for lifecycle / tiering integration tests.
+///
+/// Cloning shares the same underlying storage, fault configuration, and
+/// operation log (everything is behind an [`Arc`]), so a clone registered in a
+/// [`TierConfigMgr`] observes the same state as the handle held by the test.
+#[derive(Clone, Default)]
+pub struct MockWarmBackend {
+    inner: Arc<MockWarmBackendInner>,
+}
+
+impl MockWarmBackend {
+    /// Create an empty backend with no faults injected.
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    // ---- fault injection -------------------------------------------------
+
+    /// Replace the entire fault configuration.
+    pub async fn set_faults(&self, faults: FaultConfig) {
+        *self.inner.faults.lock().await = faults;
+    }
+
+    /// Toggle "backend unreachable" (connection refused on every operation).
+    pub async fn set_unreachable(&self, unreachable: bool) {
+        self.inner.faults.lock().await.unreachable = unreachable;
+    }
+
+    /// Toggle "HTTP 5xx" server errors on every operation.
+    pub async fn set_server_error(&self, server_error: bool) {
+        self.inner.faults.lock().await.server_error = server_error;
+    }
+
+    /// Toggle "credential rejected" (permission denied) on every operation.
+    pub async fn set_reject_credentials(&self, reject: bool) {
+        self.inner.faults.lock().await.reject_credentials = reject;
+    }
+
+    /// Set (or clear, with `None`) injected latency applied before each op.
+    pub async fn set_latency(&self, latency: Option<Duration>) {
+        self.inner.faults.lock().await.latency = latency;
+    }
+
+    /// Clear all injected faults, restoring healthy behaviour.
+    pub async fn clear_faults(&self) {
+        *self.inner.faults.lock().await = FaultConfig::default();
+    }
+
+    async fn precondition(&self) -> Result<(), std::io::Error> {
+        let (latency, error) = {
+            let faults = self.inner.faults.lock().await;
+            (faults.latency, faults.error())
+        };
+        if let Some(latency) = latency {
+            tokio::time::sleep(latency).await;
+        }
+        match error {
+            Some(err) => Err(err),
+            None => Ok(()),
+        }
+    }
+
+    // ---- operation log ---------------------------------------------------
+
+    async fn record(&self, op: MockWarmOp) {
+        self.inner.op_log.lock().await.push(op);
+    }
+
+    /// Snapshot of every operation the backend has performed, in order.
+    pub async fn op_log(&self) -> Vec<MockWarmOp> {
+        self.inner.op_log.lock().await.clone()
+    }
+
+    /// Clear the operation log without touching stored objects or faults.
+    pub async fn clear_op_log(&self) {
+        self.inner.op_log.lock().await.clear();
+    }
+
+    /// Number of trait `remove` calls recorded (excludes `external_remove`).
+    pub async fn remove_count(&self) -> usize {
+        self.inner
+            .op_log
+            .lock()
+            .await
+            .iter()
+            .filter(|op| matches!(op, MockWarmOp::Remove { .. }))
+            .count()
+    }
+
+    /// Number of `get` calls recorded — useful to assert restore reads hit the
+    /// local copy rather than the remote tier.
+    pub async fn get_count(&self) -> usize {
+        self.inner
+            .op_log
+            .lock()
+            .await
+            .iter()
+            .filter(|op| matches!(op, MockWarmOp::Get { .. }))
+            .count()
+    }
+
+    /// Number of `put` calls recorded.
+    pub async fn put_count(&self) -> usize {
+        self.inner
+            .op_log
+            .lock()
+            .await
+            .iter()
+            .filter(|op| matches!(op, MockWarmOp::Put { .. }))
+            .count()
+    }
+
+    // ---- storage inspection ---------------------------------------------
+
+    /// Whether the backend currently stores `object`.
+    pub async fn contains(&self, object: &str) -> bool {
+        self.inner.objects.lock().await.contains_key(object)
+    }
+
+    /// Number of objects currently stored.
+    pub async fn object_count(&self) -> usize {
+        self.inner.objects.lock().await.len()
+    }
+
+    /// A clone of the stored object, if present.
+    pub async fn stored(&self, object: &str) -> Option<MockStoredObject> {
+        self.inner.objects.lock().await.get(object).cloned()
+    }
+
+    /// A clone of the raw bytes stored for `object`, if present.
+    pub async fn bytes(&self, object: &str) -> Option<Vec<u8>> {
+        self.inner.objects.lock().await.get(object).map(|o| o.bytes.clone())
+    }
+
+    /// A clone of the metadata stored for `object`, if present.
+    pub async fn metadata(&self, object: &str) -> Option<HashMap<String, String>> {
+        self.inner.objects.lock().await.get(object).map(|o| o.metadata.clone())
+    }
+
+    /// Simulate the remote object being deleted out-of-band (for example by an
+    /// operator or lifecycle rule on the remote tier), bypassing the trait
+    /// `remove`. Records [`MockWarmOp::ExternalRemove`].
+    pub async fn external_remove(&self, object: &str) {
+        self.inner.objects.lock().await.remove(object);
+        self.record(MockWarmOp::ExternalRemove {
+            object: object.to_string(),
+        })
+        .await;
+    }
+
+    // ---- polling helpers -------------------------------------------------
+
+    /// Poll until `object` is absent from the backend, or `timeout` elapses.
+    /// Returns `true` if the object disappeared within the budget.
+    pub async fn wait_for_remote_absence(&self, object: &str, timeout: Duration) -> bool {
+        let deadline = tokio::time::Instant::now() + timeout;
+        loop {
+            if !self.contains(object).await {
+                return true;
+            }
+            if tokio::time::Instant::now() >= deadline {
+                return false;
+            }
+            tokio::time::sleep(POLL_INTERVAL).await;
+        }
+    }
+
+    /// Poll until the backend holds exactly `expected` objects, or `timeout`
+    /// elapses. Returns `true` if the count was reached within the budget.
+    pub async fn wait_for_object_count(&self, expected: usize, timeout: Duration) -> bool {
+        let deadline = tokio::time::Instant::now() + timeout;
+        loop {
+            if self.object_count().await == expected {
+                return true;
+            }
+            if tokio::time::Instant::now() >= deadline {
+                return false;
+            }
+            tokio::time::sleep(POLL_INTERVAL).await;
+        }
+    }
+
+    // ---- internal helpers -----------------------------------------------
+
+    async fn put_bytes(&self, object: &str, bytes: Vec<u8>, metadata: HashMap<String, String>) -> String {
+        let remote_version_id = Uuid::new_v4().to_string();
+        self.inner.objects.lock().await.insert(
+            object.to_string(),
+            MockStoredObject {
+                bytes,
+                metadata,
+                remote_version_id: remote_version_id.clone(),
+            },
+        );
+        remote_version_id
+    }
+
+    async fn read_bytes(&self, reader: ReaderImpl) -> Result<Vec<u8>, std::io::Error> {
+        match reader {
+            ReaderImpl::Body(bytes) => Ok(bytes.to_vec()),
+            ReaderImpl::ObjectBody(mut reader) => {
+                let mut buf = Vec::new();
+                reader.stream.read_to_end(&mut buf).await?;
+                Ok(buf)
+            }
+        }
+    }
+}
+
+#[async_trait]
+impl WarmBackend for MockWarmBackend {
+    async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
+        self.precondition().await?;
+        let bytes = self.read_bytes(r).await?;
+        let version = self.put_bytes(object, bytes, HashMap::new()).await;
+        self.record(MockWarmOp::Put {
+            object: object.to_string(),
+        })
+        .await;
+        Ok(version)
+    }
+
+    async fn put_with_meta(
+        &self,
+        object: &str,
+        r: ReaderImpl,
+        _length: i64,
+        meta: HashMap<String, String>,
+    ) -> Result<String, std::io::Error> {
+        self.precondition().await?;
+        let bytes = self.read_bytes(r).await?;
+        // Mirror the real transition put: promote content headers and drop the
+        // internal replication / object-lock defaults so tests can assert that
+        // transitioned objects don't inherit them.
+        let opts = build_transition_put_options(String::new(), meta);
+        let mut metadata = opts.user_metadata.clone();
+        if !opts.content_type.is_empty() {
+            metadata.insert("content-type".to_string(), opts.content_type.clone());
+        }
+        if !opts.content_encoding.is_empty() {
+            metadata.insert("content-encoding".to_string(), opts.content_encoding.clone());
+        }
+        if !opts.cache_control.is_empty() {
+            metadata.insert("cache-control".to_string(), opts.cache_control.clone());
+        }
+        if !opts.internal.replication_status.as_str().is_empty() {
+            metadata.insert(
+                "x-amz-replication-status".to_string(),
+                opts.internal.replication_status.as_str().to_string(),
+            );
+        }
+        if !opts.legalhold.as_str().is_empty() {
+            metadata.insert("x-amz-object-lock-legal-hold".to_string(), opts.legalhold.as_str().to_string());
+        }
+        let version = self.put_bytes(object, bytes, metadata).await;
+        self.record(MockWarmOp::Put {
+            object: object.to_string(),
+        })
+        .await;
+        Ok(version)
+    }
+
+    async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
+        self.precondition().await?;
+        self.record(MockWarmOp::Get {
+            object: object.to_string(),
+        })
+        .await;
+        let objects = self.inner.objects.lock().await;
+        let Some(stored) = objects.get(object) else {
+            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found"));
+        };
+        let bytes = &stored.bytes;
+
+        let start = opts.start_offset.max(0) as usize;
+        let end = if opts.length > 0 {
+            start.saturating_add(opts.length as usize).min(bytes.len())
+        } else {
+            bytes.len()
+        };
+
+        Ok(tokio::io::BufReader::new(Cursor::new(bytes[start.min(bytes.len())..end].to_vec())))
+    }
+
+    async fn remove(&self, object: &str, _rv: &str) -> Result<(), std::io::Error> {
+        self.precondition().await?;
+        self.inner.objects.lock().await.remove(object);
+        self.record(MockWarmOp::Remove {
+            object: object.to_string(),
+        })
+        .await;
+        Ok(())
+    }
+
+    async fn in_use(&self) -> Result<bool, std::io::Error> {
+        self.precondition().await?;
+        self.record(MockWarmOp::InUse).await;
+        Ok(false)
+    }
+}
+
+/// Register `backend` as a MinIO-typed tier named `tier_name` in the tier
+/// config manager reachable through `handle`, returning a clone of the backend
+/// that shares its state.
+///
+/// Works with any [`TierConfigMgr`] handle — the global one
+/// (`rustfs_ecstore::api::runtime::global_tier_config_mgr`) used by the scanner
+/// integration tests, or a per-instance one used by the app-layer tests.
+pub async fn register_mock_tier(handle: &Arc<RwLock<TierConfigMgr>>, tier_name: &str) -> MockWarmBackend {
+    let backend = MockWarmBackend::new();
+    register_mock_tier_backend(handle, tier_name, backend.clone()).await;
+    backend
+}
+
+/// Like [`register_mock_tier`] but registers an existing `backend` (for example
+/// one that already had faults injected before wiring it up).
+pub async fn register_mock_tier_backend(handle: &Arc<RwLock<TierConfigMgr>>, tier_name: &str, backend: MockWarmBackend) {
+    let mut tier_config_mgr = handle.write().await;
+    tier_config_mgr.tiers.insert(
+        tier_name.to_string(),
+        TierConfig {
+            version: "v1".to_string(),
+            tier_type: TierType::MinIO,
+            name: tier_name.to_string(),
+            minio: Some(TierMinIO {
+                access_key: "minioadmin".to_string(),
+                secret_key: "minioadmin".to_string(),
+                bucket: "mock-tier".to_string(),
+                endpoint: "http://127.0.0.1:0".to_string(),
+                prefix: format!("mock/{}/", Uuid::new_v4()),
+                region: String::new(),
+                ..Default::default()
+            }),
+            ..Default::default()
+        },
+    );
+    tier_config_mgr.driver_cache.insert(tier_name.to_string(), Box::new(backend));
+}
+
+/// The transition-state tuple read from an on-disk `xl.meta`, plus the object's
+/// free-version count.
+#[derive(Clone, Debug, PartialEq, Eq)]
+pub struct TransitionMeta {
+    /// `transition_status` (e.g. `"complete"`), empty when not transitioned.
+    pub status: String,
+    /// The tier name the object was transitioned to.
+    pub tier: String,
+    /// The remote object key (`transitioned_objname`).
+    pub remote_object: String,
+    /// The remote version id (`transition_version_id`), if any.
+    pub remote_version_id: Option<String>,
+    /// Number of free versions retained for the object.
+    pub free_version_count: usize,
+}
+
+async fn open_disk(disk_path: &Path) -> Option<crate::disk::DiskStore> {
+    let mut endpoint = Endpoint::try_from(disk_path.to_str()?).ok()?;
+    endpoint.set_pool_index(0);
+    endpoint.set_set_index(0);
+    endpoint.set_disk_index(0);
+    new_disk(
+        &endpoint,
+        &DiskOption {
+            cleanup: false,
+            health_check: false,
+        },
+    )
+    .await
+    .ok()
+}
+
+/// Count the free versions retained for `object` on a single disk's `xl.meta`.
+///
+/// The free-version metadata removal lands asynchronously after the remote
+/// object disappears, so callers typically poll via
+/// [`wait_for_free_version_absence`] instead of asserting a single read.
+pub async fn free_version_count(disk_path: &Path, bucket: &str, object: &str) -> usize {
+    let Some(disk) = open_disk(disk_path).await else {
+        return 0;
+    };
+    let Ok(data) = disk
+        .read_metadata(bucket, &path_join_buf(&[object, STORAGE_FORMAT_FILE]))
+        .await
+    else {
+        return 0;
+    };
+    let Ok(meta) = FileMeta::load(&data) else {
+        return 0;
+    };
+    meta.get_file_info_versions(bucket, object, false)
+        .map(|v| v.free_versions.len())
+        .unwrap_or(0)
+}
+
+/// Read the transition-state tuple for `object` from a single disk's `xl.meta`.
+///
+/// Returns `None` if the metadata cannot be read or decoded. The transition
+/// fields are taken from the newest version that carries a transition record;
+/// if no version is transitioned, they are taken from the current version (and
+/// will be empty).
+pub async fn read_transition_meta(disk_path: &Path, bucket: &str, object: &str) -> Option<TransitionMeta> {
+    let disk = open_disk(disk_path).await?;
+    let data = disk
+        .read_metadata(bucket, &path_join_buf(&[object, STORAGE_FORMAT_FILE]))
+        .await
+        .ok()?;
+    let meta = FileMeta::load(&data).ok()?;
+    let versions = meta.get_file_info_versions(bucket, object, false).ok()?;
+    let free_version_count = versions.free_versions.len();
+
+    let fi = versions
+        .versions
+        .iter()
+        .find(|v| !v.transition_status.is_empty() || !v.transitioned_objname.is_empty())
+        .or_else(|| versions.versions.first())?;
+
+    Some(TransitionMeta {
+        status: fi.transition_status.clone(),
+        tier: fi.transition_tier.clone(),
+        remote_object: fi.transitioned_objname.clone(),
+        remote_version_id: fi.transition_version_id.map(|id| id.to_string()),
+        free_version_count,
+    })
+}
+
+/// Assert that every disk in `disk_paths` reports the same transition tuple for
+/// `object`, returning that tuple. Panics with a descriptive message if any
+/// disk is missing the object or disagrees — this is the shard-consistency
+/// check required by ilm-6 (the `(status, tier, remote key, remote version id)`
+/// four-tuple plus free-version count must match across all erasure shards).
+pub async fn assert_transition_meta_consistent<P: AsRef<Path>>(disk_paths: &[P], bucket: &str, object: &str) -> TransitionMeta {
+    assert!(!disk_paths.is_empty(), "assert_transition_meta_consistent needs at least one disk");
+
+    let mut expected: Option<TransitionMeta> = None;
+    for disk_path in disk_paths {
+        let disk_path = disk_path.as_ref();
+        let meta = read_transition_meta(disk_path, bucket, object)
+            .await
+            .unwrap_or_else(|| panic!("missing xl.meta for {bucket}/{object} on disk {}", disk_path.display()));
+        match &expected {
+            None => expected = Some(meta),
+            Some(expected) => assert_eq!(
+                *expected,
+                meta,
+                "transition metadata mismatch for {bucket}/{object} on disk {} (expected {expected:?}, got {meta:?})",
+                disk_path.display()
+            ),
+        }
+    }
+
+    expected.expect("at least one disk was inspected")
+}
+
+/// Poll until `object` retains no free versions on `disk_path`, or `timeout`
+/// elapses. Returns `true` if the free versions drained within the budget.
+pub async fn wait_for_free_version_absence(disk_path: &Path, bucket: &str, object: &str, timeout: Duration) -> bool {
+    let deadline = tokio::time::Instant::now() + timeout;
+    loop {
+        if free_version_count(disk_path, bucket, object).await == 0 {
+            return true;
+        }
+        if tokio::time::Instant::now() >= deadline {
+            return false;
+        }
+        tokio::time::sleep(POLL_INTERVAL).await;
+    }
+}

crates/scanner/Cargo.toml

@@ -59,6 +59,9 @@ serial_test = { workspace = true }
 temp-env = { workspace = true }
 uuid = { workspace = true, features = ["v4", "serde"] }
 tokio = { workspace = true, features = ["test-util"] }
+# Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
+# the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
+rustfs-ecstore = { workspace = true, features = ["test-util"] }
 
 [lib]
 doctest = false

crates/scanner/tests/lifecycle_integration_test.rs

@@ -14,27 +14,23 @@
 
 use futures::FutureExt;
 use rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT;
-use rustfs_filemeta::FileMeta;
 use rustfs_scanner::scanner_folder::ScannerItem;
 use rustfs_scanner::scanner_io::ScannerIODisk;
 use rustfs_scanner::{
     ScannerObjectInfo as ObjectInfo, ScannerObjectOptions as ObjectOptions, ScannerPutObjReader as PutObjReader,
     scanner::init_data_scanner,
 };
-use rustfs_utils::path::path_join_buf;
 use s3s::dto::RestoreRequest;
 use serial_test::serial;
 use std::{
     collections::HashMap,
     env,
-    io::Cursor,
     path::{Path, PathBuf},
     sync::{Arc, Once, OnceLock},
     time::Duration,
 };
 use tokio::fs;
 use tokio::io::AsyncReadExt;
-use tokio::sync::Mutex;
 use tokio_util::sync::CancellationToken;
 use tracing::info;
 use uuid::Uuid;
@@ -42,13 +38,13 @@ use uuid::Uuid;
 mod storage_api;
 
 use storage_api::lifecycle::{
-    BUCKET_LIFECYCLE_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, DiskAPI as _, DiskOption,
-    ECStore, EcstoreError, Endpoint, EndpointServerPools, Endpoints, IlmAction, LcEvent, LcEventSrc, ListOperations as _,
-    MakeBucketOptions, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, PoolEndpoints, ReadCloser, ReaderImpl,
-    STORAGE_FORMAT_FILE, ScannerWarmBackend, TierConfig, TierMinIO, TierType, TransitionOptions, WarmBackendGetOpts,
-    build_transition_put_options, enqueue_transition_for_existing_objects, expire_transitioned_object, get_bucket_metadata,
-    get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found,
-    is_err_version_not_found, new_disk, path2_bucket_object_with_base_path, update_bucket_metadata,
+    BUCKET_LIFECYCLE_CONFIG, BucketOperations, BucketOptions, BucketVersioningSys, CompletePart, DiskOption, ECStore,
+    EcstoreError, Endpoint, EndpointServerPools, Endpoints, IlmAction, LcEvent, LcEventSrc, ListOperations as _,
+    MakeBucketOptions, MockWarmBackend, MultipartOperations as _, ObjectIO as _, ObjectOperations as _, PoolEndpoints,
+    STORAGE_FORMAT_FILE, TransitionOptions, assert_transition_meta_consistent, enqueue_transition_for_existing_objects,
+    expire_transitioned_object, free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry,
+    init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk,
+    path2_bucket_object_with_base_path, register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence,
 };
 
 static GLOBAL_ENV: OnceLock<(Vec<PathBuf>, Arc<ECStore>)> = OnceLock::new();
@@ -397,67 +393,6 @@ async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &
     }
 }
 
-async fn wait_for_remote_absence(backend: &MockWarmBackend, object: &str, timeout: Duration) -> bool {
-    let deadline = tokio::time::Instant::now() + timeout;
-
-    loop {
-        if !backend.objects.lock().await.contains_key(object) {
-            return true;
-        }
-
-        if tokio::time::Instant::now() >= deadline {
-            return false;
-        }
-
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-}
-
-async fn free_version_count(disk_path: &Path, bucket: &str, object: &str) -> usize {
-    let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
-    endpoint.set_pool_index(0);
-    endpoint.set_set_index(0);
-    endpoint.set_disk_index(0);
-    let disk = new_disk(
-        &endpoint,
-        &DiskOption {
-            cleanup: false,
-            health_check: false,
-        },
-    )
-    .await
-    .expect("failed to open local disk");
-    let data = disk
-        .read_metadata(bucket, &path_join_buf(&[object, STORAGE_FORMAT_FILE]))
-        .await;
-    let Ok(data) = data else {
-        return 0;
-    };
-    let meta = FileMeta::load(&data).expect("failed to load file metadata");
-    meta.get_file_info_versions(bucket, object, false)
-        .expect("failed to decode file info versions")
-        .free_versions
-        .len()
-}
-
-/// The free-version metadata removal lands asynchronously after the remote
-/// object disappears, so poll instead of asserting a single read.
-async fn wait_for_free_version_absence(disk_path: &Path, bucket: &str, object: &str, timeout: Duration) -> bool {
-    let deadline = tokio::time::Instant::now() + timeout;
-
-    loop {
-        if free_version_count(disk_path, bucket, object).await == 0 {
-            return true;
-        }
-
-        if tokio::time::Instant::now() >= deadline {
-            return false;
-        }
-
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-}
-
 async fn object_version_count(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> usize {
     let mut marker = None;
     let mut version_marker = None;
@@ -499,22 +434,6 @@ async fn wait_for_version_count(ecstore: &Arc<ECStore>, bucket: &str, object: &s
     }
 }
 
-async fn wait_for_remote_object_count(backend: &MockWarmBackend, expected: usize, timeout: Duration) -> bool {
-    let deadline = tokio::time::Instant::now() + timeout;
-
-    loop {
-        if backend.objects.lock().await.len() == expected {
-            return true;
-        }
-
-        if tokio::time::Instant::now() >= deadline {
-            return false;
-        }
-
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-}
-
 async fn scan_object_with_lifecycle(disk_path: &Path, bucket: &str, object: &str) {
     let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap();
     endpoint.set_pool_index(0);
@@ -595,147 +514,11 @@ async fn scan_object_metadata(disk_path: &Path, bucket: &str, object: &str) {
     disk.get_size(item).await.expect("scanner get_size should succeed");
 }
 
-#[derive(Clone, Default)]
-struct MockStoredObject {
-    bytes: Vec<u8>,
-    metadata: HashMap<String, String>,
-}
-
-#[derive(Clone, Default)]
-struct MockWarmBackend {
-    objects: Arc<Mutex<HashMap<String, MockStoredObject>>>,
-    // Ordered log of remote-tier mutating operations observed by this backend
-    // (e.g. `remove:<obj>`). Used by the expire/GET race regression test to
-    // assert that `expire_transitioned_object` never issues a synchronous
-    // remote-tier removal (the #3491 local-first ordering contract).
-    ops_log: Arc<Mutex<Vec<String>>>,
-}
-
-#[allow(dead_code)]
-impl MockWarmBackend {
-    async fn remote_remove_count(&self) -> usize {
-        self.ops_log
-            .lock()
-            .await
-            .iter()
-            .filter(|op| op.starts_with("remove:"))
-            .count()
-    }
-}
-
-impl MockWarmBackend {
-    async fn put_bytes(&self, object: &str, bytes: Vec<u8>, metadata: HashMap<String, String>) -> String {
-        self.objects
-            .lock()
-            .await
-            .insert(object.to_string(), MockStoredObject { bytes, metadata });
-        Uuid::new_v4().to_string()
-    }
-
-    async fn read_bytes(&self, reader: ReaderImpl) -> Result<Vec<u8>, std::io::Error> {
-        match reader {
-            ReaderImpl::Body(bytes) => Ok(bytes.to_vec()),
-            ReaderImpl::ObjectBody(mut reader) => {
-                let mut buf = Vec::new();
-                reader.stream.read_to_end(&mut buf).await?;
-                Ok(buf)
-            }
-        }
-    }
-}
-
-#[async_trait::async_trait]
-impl ScannerWarmBackend for MockWarmBackend {
-    async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
-        let bytes = self.read_bytes(r).await?;
-        Ok(self.put_bytes(object, bytes, HashMap::new()).await)
-    }
-
-    async fn put_with_meta(
-        &self,
-        object: &str,
-        r: ReaderImpl,
-        _length: i64,
-        meta: HashMap<String, String>,
-    ) -> Result<String, std::io::Error> {
-        let bytes = self.read_bytes(r).await?;
-        let opts = build_transition_put_options(String::new(), meta);
-        let mut metadata = opts.user_metadata.clone();
-        if !opts.content_type.is_empty() {
-            metadata.insert("content-type".to_string(), opts.content_type.clone());
-        }
-        if !opts.content_encoding.is_empty() {
-            metadata.insert("content-encoding".to_string(), opts.content_encoding.clone());
-        }
-        if !opts.cache_control.is_empty() {
-            metadata.insert("cache-control".to_string(), opts.cache_control.clone());
-        }
-        if !opts.internal.replication_status.as_str().is_empty() {
-            metadata.insert(
-                "x-amz-replication-status".to_string(),
-                opts.internal.replication_status.as_str().to_string(),
-            );
-        }
-        if !opts.legalhold.as_str().is_empty() {
-            metadata.insert("x-amz-object-lock-legal-hold".to_string(), opts.legalhold.as_str().to_string());
-        }
-        Ok(self.put_bytes(object, bytes, metadata).await)
-    }
-
-    async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-        let objects = self.objects.lock().await;
-        let Some(stored) = objects.get(object) else {
-            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found"));
-        };
-        let bytes = &stored.bytes;
-
-        let start = opts.start_offset.max(0) as usize;
-        let end = if opts.length > 0 {
-            start.saturating_add(opts.length as usize).min(bytes.len())
-        } else {
-            bytes.len()
-        };
-
-        Ok(tokio::io::BufReader::new(Cursor::new(bytes[start.min(bytes.len())..end].to_vec())))
-    }
-
-    async fn remove(&self, object: &str, _rv: &str) -> Result<(), std::io::Error> {
-        self.ops_log.lock().await.push(format!("remove:{object}"));
-        self.objects.lock().await.remove(object);
-        Ok(())
-    }
-
-    async fn in_use(&self) -> Result<bool, std::io::Error> {
-        Ok(false)
-    }
-}
-
+/// Register the shared [`MockWarmBackend`] into the global tier config manager
+/// used by the scanner integration tests. Thin wrapper over the shared
+/// `register_mock_tier` helper (rustfs/backlog#1148 ilm-6).
 async fn register_mock_tier(tier_name: &str) -> MockWarmBackend {
-    let backend = MockWarmBackend::default();
-    let tier_config_mgr_handle = get_global_tier_config_mgr();
-    let mut tier_config_mgr = tier_config_mgr_handle.write().await;
-    tier_config_mgr.tiers.insert(
-        tier_name.to_string(),
-        TierConfig {
-            version: "v1".to_string(),
-            tier_type: TierType::MinIO,
-            name: tier_name.to_string(),
-            minio: Some(TierMinIO {
-                access_key: "minioadmin".to_string(),
-                secret_key: "minioadmin".to_string(),
-                bucket: "mock-tier".to_string(),
-                endpoint: "http://127.0.0.1:0".to_string(),
-                prefix: format!("mock/{}/", Uuid::new_v4()),
-                region: String::new(),
-                ..Default::default()
-            }),
-            ..Default::default()
-        },
-    );
-    tier_config_mgr
-        .driver_cache
-        .insert(tier_name.to_string(), Box::new(backend.clone()));
-    backend
+    register_mock_tier_util(&get_global_tier_config_mgr(), tier_name).await
 }
 
 async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> Option<ObjectInfo> {
@@ -842,11 +625,11 @@ mod serial_tests {
             .expect("object should transition before expiry");
         let remote_object = transitioned.transitioned_object.name.clone();
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "transitioned remote object should exist in the mock tier before expiry"
         );
         assert_eq!(
-            backend.remote_remove_count().await,
+            backend.remove_count().await,
             0,
             "no remote-tier removal should have happened before expiry"
         );
@@ -924,13 +707,13 @@ mod serial_tests {
         // remote `remove` was issued. Reverting to remote-first ordering makes
         // both assertions fail.
         assert_eq!(
-            backend.remote_remove_count().await,
+            backend.remove_count().await,
             0,
             "expire_transitioned_object must NOT issue a synchronous remote-tier removal (local-first \
              ordering, #3491); remote cleanup is deferred to free-version recovery"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "remote tier object must still exist immediately after expiry (deferred cleanup, #3491)"
         );
 
@@ -953,7 +736,7 @@ mod serial_tests {
     #[serial]
     #[ignore = "FAILING on main: excluded from the serial ILM lane pending a fix, see rustfs/backlog#1148 (ilm-1 partial)"]
     async fn test_transition_and_restore_flows() {
-        let (_disk_paths, ecstore) = setup_test_env().await;
+        let (disk_paths, ecstore) = setup_test_env().await;
 
         let tier_name = format!("COLDTIER{}", &Uuid::new_v4().simple().to_string()[..8]).to_uppercase();
         let backend = register_mock_tier(&tier_name).await;
@@ -993,11 +776,11 @@ mod serial_tests {
 
         assert_eq!(put_info.transitioned_object.status, "complete");
         assert_eq!(put_info.transitioned_object.tier, tier_name);
-        assert!(backend.objects.lock().await.contains_key(&put_info.transitioned_object.name));
+        assert!(backend.contains(&put_info.transitioned_object.name).await);
         {
-            let stored = backend.objects.lock().await;
-            let transitioned = stored
-                .get(&put_info.transitioned_object.name)
+            let transitioned = backend
+                .stored(&put_info.transitioned_object.name)
+                .await
                 .expect("transitioned object should be present in mock backend");
             assert_eq!(transitioned.metadata.get("content-type"), Some(&"text/plain".to_string()));
             assert!(
@@ -1010,6 +793,12 @@ mod serial_tests {
             );
         }
 
+        // Cross-shard xl.meta transition assertion helper (rustfs/backlog#1148 ilm-6):
+        // every disk must agree on the transition tuple for the object.
+        let put_meta = assert_transition_meta_consistent(&disk_paths, put_bucket.as_str(), put_object).await;
+        assert_eq!(put_meta.status, "complete");
+        assert_eq!(put_meta.tier, tier_name);
+
         let multipart_bucket = format!("test-immediate-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
         let multipart_object = "test/multipart.txt";
 
@@ -1063,13 +852,7 @@ mod serial_tests {
 
         assert_eq!(multipart_info.transitioned_object.status, "complete");
         assert_eq!(multipart_info.transitioned_object.tier, tier_name);
-        assert!(
-            backend
-                .objects
-                .lock()
-                .await
-                .contains_key(&multipart_info.transitioned_object.name)
-        );
+        assert!(backend.contains(&multipart_info.transitioned_object.name).await);
 
         let src_bucket = format!("test-immediate-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
         let dst_bucket = format!("test-immediate-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
@@ -1114,7 +897,7 @@ mod serial_tests {
 
         assert_eq!(copy_info.transitioned_object.status, "complete");
         assert_eq!(copy_info.transitioned_object.tier, tier_name);
-        assert!(backend.objects.lock().await.contains_key(&copy_info.transitioned_object.name));
+        assert!(backend.contains(&copy_info.transitioned_object.name).await);
 
         let bucket_name = format!("test-lifecycle-update-{}", &Uuid::new_v4().simple().to_string()[..8]);
         let object_name = "test/existing.txt";
@@ -1137,7 +920,7 @@ mod serial_tests {
 
         assert_eq!(info.transitioned_object.status, "complete");
         assert_eq!(info.transitioned_object.tier, tier_name);
-        assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+        assert!(backend.contains(&info.transitioned_object.name).await);
 
         let bucket_name = format!("test-restore-mpu-{}", &Uuid::new_v4().simple().to_string()[..8]);
         let object_name = "test/restore.txt";
@@ -1284,7 +1067,7 @@ mod serial_tests {
             .await
             .expect("object should transition before overwrite");
         let stale_remote_object = transitioned.transitioned_object.name.clone();
-        assert!(backend.objects.lock().await.contains_key(&stale_remote_object));
+        assert!(backend.contains(&stale_remote_object).await);
 
         ecstore
             .delete_object(bucket_name.as_str(), object_name, ObjectOptions::default())
@@ -1296,7 +1079,7 @@ mod serial_tests {
             "deleting a transitioned null version should leave a free version for async cleanup"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&stale_remote_object),
+            backend.contains(&stale_remote_object).await,
             "stale transitioned remote object should still exist before scanner fallback runs"
         );
 
@@ -1304,7 +1087,9 @@ mod serial_tests {
         scan_object_metadata(&disk_paths[0], bucket_name.as_str(), object_name).await;
 
         assert!(
-            wait_for_remote_absence(&backend, &stale_remote_object, TRANSITION_WAIT_TIMEOUT).await,
+            backend
+                .wait_for_remote_absence(&stale_remote_object, TRANSITION_WAIT_TIMEOUT)
+                .await,
             "scanner should enqueue stale free-version cleanup for the transitioned remote object"
         );
         assert!(
@@ -1344,7 +1129,7 @@ mod serial_tests {
             .await
             .expect("object should transition after compensation backfill");
         let stale_remote_object = transitioned.transitioned_object.name.clone();
-        assert!(backend.objects.lock().await.contains_key(&stale_remote_object));
+        assert!(backend.contains(&stale_remote_object).await);
 
         ecstore
             .delete_object(bucket_name.as_str(), object_name, ObjectOptions::default())
@@ -1356,7 +1141,7 @@ mod serial_tests {
             "deleting a compensation-transitioned null version should leave a free version for async cleanup"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&stale_remote_object),
+            backend.contains(&stale_remote_object).await,
             "stale transitioned remote object should still exist before scanner cleanup runs"
         );
 
@@ -1364,7 +1149,9 @@ mod serial_tests {
         scan_object_metadata(&disk_paths[0], bucket_name.as_str(), object_name).await;
 
         assert!(
-            wait_for_remote_absence(&backend, &stale_remote_object, TRANSITION_WAIT_TIMEOUT).await,
+            backend
+                .wait_for_remote_absence(&stale_remote_object, TRANSITION_WAIT_TIMEOUT)
+                .await,
             "scanner should clean stale remote object even after immediate compensation transitioned it"
         );
         assert!(
@@ -1400,7 +1187,7 @@ mod serial_tests {
             .await
             .expect("object should transition after immediate compensation backfill");
         let remote_object = transitioned.transitioned_object.name.clone();
-        assert!(backend.objects.lock().await.contains_key(&remote_object));
+        assert!(backend.contains(&remote_object).await);
 
         enqueue_transition_for_existing_objects(ecstore.clone(), bucket_name.as_str())
             .await
@@ -1413,7 +1200,7 @@ mod serial_tests {
         assert_eq!(info.transitioned_object.status, "complete");
         assert_eq!(info.transitioned_object.tier, tier_name);
         assert_eq!(info.transitioned_object.name, remote_object);
-        assert!(backend.objects.lock().await.contains_key(&remote_object));
+        assert!(backend.contains(&remote_object).await);
     }
 
     #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -1490,7 +1277,7 @@ mod serial_tests {
 
         assert_eq!(info.transitioned_object.status, "complete");
         assert_eq!(info.transitioned_object.tier, tier_name);
-        assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+        assert!(backend.contains(&info.transitioned_object.name).await);
 
         scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await;
 
@@ -1578,7 +1365,7 @@ mod serial_tests {
         scan_object_with_lifecycle(&disk_paths[0], bucket_name.as_str(), object_name).await;
 
         assert!(
-            wait_for_remote_object_count(&backend, 2, TRANSITION_WAIT_TIMEOUT).await,
+            backend.wait_for_object_count(2, TRANSITION_WAIT_TIMEOUT).await,
             "noncurrent transition should still move the previous version into the remote tier"
         );
     }
@@ -1610,7 +1397,7 @@ mod serial_tests {
             .await
             .expect("current version should transition after compensation backfill");
         let remote_object = transitioned.transitioned_object.name.clone();
-        assert!(backend.objects.lock().await.contains_key(&remote_object));
+        assert!(backend.contains(&remote_object).await);
 
         ecstore
             .delete_object(
@@ -1626,7 +1413,7 @@ mod serial_tests {
             "versioned delete modeled with versioned flags should create a delete marker"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "creating a delete marker should not remove the transitioned remote object version"
         );
     }
@@ -1658,7 +1445,7 @@ mod serial_tests {
             .await
             .expect("current version should transition after compensation backfill");
         let remote_object = transitioned.transitioned_object.name.clone();
-        assert!(backend.objects.lock().await.contains_key(&remote_object));
+        assert!(backend.contains(&remote_object).await);
 
         ecstore
             .delete_object(
@@ -1674,7 +1461,7 @@ mod serial_tests {
             "modeled versioned delete should create delete marker before cleanup"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "delete marker creation should not remove transitioned remote object"
         );
 
@@ -1689,7 +1476,7 @@ mod serial_tests {
             "delete marker should remain before DelMarkerExpiration due time"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "pre-due delete marker lifecycle scan should not remove transitioned remote object"
         );
 
@@ -1703,7 +1490,7 @@ mod serial_tests {
             "expired object delete marker lifecycle should eventually clean up the delete marker"
         );
         assert!(
-            backend.objects.lock().await.contains_key(&remote_object),
+            backend.contains(&remote_object).await,
             "delete marker lifecycle cleanup should not remove transitioned remote object"
         );
     }

crates/scanner/tests/storage_api/mod.rs

@@ -23,15 +23,17 @@ pub(crate) use rustfs_ecstore::api::bucket::metadata_sys::{
 };
 pub(crate) use rustfs_ecstore::api::bucket::versioning_sys::BucketVersioningSys;
 pub(crate) use rustfs_ecstore::api::capacity::path2_bucket_object_with_base_path;
-pub(crate) use rustfs_ecstore::api::client::transition_api::{ReadCloser, ReaderImpl};
-pub(crate) use rustfs_ecstore::api::disk::{DiskAPI, DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, new_disk};
+pub(crate) use rustfs_ecstore::api::disk::{DiskOption, STORAGE_FORMAT_FILE, endpoint::Endpoint, new_disk};
 pub(crate) use rustfs_ecstore::api::error::{Error as EcstoreError, is_err_object_not_found, is_err_version_not_found};
 pub(crate) use rustfs_ecstore::api::layout::{EndpointServerPools, Endpoints, PoolEndpoints};
 pub(crate) use rustfs_ecstore::api::runtime::global_tier_config_mgr as get_global_tier_config_mgr;
 pub(crate) use rustfs_ecstore::api::storage::{ECStore, init_local_disks};
-pub(crate) use rustfs_ecstore::api::tier::tier_config::{TierConfig, TierMinIO, TierType};
-pub(crate) use rustfs_ecstore::api::tier::warm_backend::{
-    WarmBackend as ScannerWarmBackend, WarmBackendGetOpts, build_transition_put_options,
+// Shared lifecycle/tier test utilities (rustfs/backlog#1148 ilm-6). The mock
+// backend and xl.meta assertion helpers now live in ecstore behind the
+// `test-util` feature instead of being copied into this crate.
+pub(crate) use rustfs_ecstore::api::tier::test_util::{
+    MockWarmBackend, assert_transition_meta_consistent, free_version_count, register_mock_tier as register_mock_tier_util,
+    wait_for_free_version_absence,
 };
 use rustfs_storage_api as storage_contracts;
 
@@ -42,11 +44,11 @@ pub(crate) mod lifecycle {
     };
 
     pub(crate) use super::{
-        BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DiskAPI, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools,
-        Endpoints, IlmAction, LcEvent, LcEventSrc, PoolEndpoints, ReadCloser, ReaderImpl, STORAGE_FORMAT_FILE,
-        ScannerWarmBackend, TierConfig, TierMinIO, TierType, TransitionOptions, WarmBackendGetOpts, build_transition_put_options,
-        enqueue_transition_for_existing_objects, expire_transitioned_object, get_bucket_metadata, get_global_tier_config_mgr,
-        init_background_expiry, init_bucket_metadata_sys, init_local_disks, is_err_object_not_found, is_err_version_not_found,
-        new_disk, path2_bucket_object_with_base_path, update_bucket_metadata,
+        BUCKET_LIFECYCLE_CONFIG, BucketVersioningSys, DiskOption, ECStore, EcstoreError, Endpoint, EndpointServerPools,
+        Endpoints, IlmAction, LcEvent, LcEventSrc, MockWarmBackend, PoolEndpoints, STORAGE_FORMAT_FILE, TransitionOptions,
+        assert_transition_meta_consistent, enqueue_transition_for_existing_objects, expire_transitioned_object,
+        free_version_count, get_bucket_metadata, get_global_tier_config_mgr, init_background_expiry, init_bucket_metadata_sys,
+        init_local_disks, is_err_object_not_found, is_err_version_not_found, new_disk, path2_bucket_object_with_base_path,
+        register_mock_tier_util, update_bucket_metadata, wait_for_free_version_absence,
     };
 }

rustfs/Cargo.toml

@@ -215,6 +215,9 @@ opentelemetry_sdk = { workspace = true }
 rsa = { workspace = true }
 rcgen = { workspace = true }
 criterion = { workspace = true, features = ["html_reports"] }
+# Enables the shared MockWarmBackend / xl.meta assertion helpers exposed via
+# the ecstore `api::tier::test_util` facade module (rustfs/backlog#1148 ilm-6).
+rustfs-ecstore = { workspace = true, features = ["test-util"] }
 
 [build-dependencies]
 http.workspace = true

rustfs/src/app/lifecycle_transition_api_test.rs

@@ -16,7 +16,6 @@ use super::storage_api::test::bucket::{
     lifecycle,
     metadata::{BUCKET_LIFECYCLE_CONFIG, OBJECT_LOCK_CONFIG},
     metadata_sys,
-    transition_api::{ReadCloser, ReaderImpl},
 };
 use super::storage_api::test::contract::{
     bucket::{BucketOperations, BucketOptions, MakeBucketOptions},
@@ -26,7 +25,7 @@ use super::storage_api::test::contract::{
 };
 use super::storage_api::test::ecfs::FS;
 use super::storage_api::test::object_utils::to_s3s_etag;
-use super::storage_api::test::runtime::{AppWarmBackend, TierConfig, TierType, WarmBackendGetOpts};
+use super::storage_api::test::runtime::{MockWarmBackend, register_mock_tier as register_mock_tier_util};
 use super::storage_api::test::{
     ECStore, Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, StorageObjectInfo as ObjectInfo,
     StorageObjectOptions as ObjectOptions, StoragePutObjReader as PutObjReader,
@@ -44,17 +43,15 @@ use rustfs_utils::http::{SUFFIX_FORCE_DELETE, insert_header};
 use s3s::{S3Request, dto::*};
 use serial_test::serial;
 use std::{
-    collections::HashMap,
     convert::Infallible,
     env, fs as stdfs,
-    io::Cursor,
     path::PathBuf,
     sync::{Arc, Once, OnceLock},
     time::Duration,
 };
 use tokio::fs;
 use tokio::io::AsyncReadExt;
-use tokio::sync::{Barrier, Mutex};
+use tokio::sync::Barrier;
 use tokio_util::sync::CancellationToken;
 use uuid::Uuid;
 
@@ -285,90 +282,11 @@ fn del_marker_expiration_lifecycle_configuration(days: i32) -> BucketLifecycleCo
     }
 }
 
-#[derive(Clone, Default)]
-struct MockWarmBackend {
-    objects: Arc<Mutex<HashMap<String, Vec<u8>>>>,
-}
-
-impl MockWarmBackend {
-    async fn put_bytes(&self, object: &str, bytes: Vec<u8>) -> String {
-        self.objects.lock().await.insert(object.to_string(), bytes);
-        Uuid::new_v4().to_string()
-    }
-
-    async fn read_bytes(&self, reader: ReaderImpl) -> Result<Vec<u8>, std::io::Error> {
-        match reader {
-            ReaderImpl::Body(bytes) => Ok(bytes.to_vec()),
-            ReaderImpl::ObjectBody(mut reader) => {
-                let mut buf = Vec::new();
-                reader.stream.read_to_end(&mut buf).await?;
-                Ok(buf)
-            }
-        }
-    }
-}
-
-#[async_trait::async_trait]
-impl AppWarmBackend for MockWarmBackend {
-    async fn put(&self, object: &str, r: ReaderImpl, _length: i64) -> Result<String, std::io::Error> {
-        let bytes = self.read_bytes(r).await?;
-        Ok(self.put_bytes(object, bytes).await)
-    }
-
-    async fn put_with_meta(
-        &self,
-        object: &str,
-        r: ReaderImpl,
-        _length: i64,
-        _meta: HashMap<String, String>,
-    ) -> Result<String, std::io::Error> {
-        let bytes = self.read_bytes(r).await?;
-        Ok(self.put_bytes(object, bytes).await)
-    }
-
-    async fn get(&self, object: &str, _rv: &str, opts: WarmBackendGetOpts) -> Result<ReadCloser, std::io::Error> {
-        let objects = self.objects.lock().await;
-        let Some(bytes) = objects.get(object) else {
-            return Err(std::io::Error::new(std::io::ErrorKind::NotFound, "mock object not found"));
-        };
-
-        let start = opts.start_offset.max(0) as usize;
-        let end = if opts.length > 0 {
-            start.saturating_add(opts.length as usize).min(bytes.len())
-        } else {
-            bytes.len()
-        };
-
-        Ok(tokio::io::BufReader::new(Cursor::new(bytes[start.min(bytes.len())..end].to_vec())))
-    }
-
-    async fn remove(&self, object: &str, _rv: &str) -> Result<(), std::io::Error> {
-        self.objects.lock().await.remove(object);
-        Ok(())
-    }
-
-    async fn in_use(&self) -> Result<bool, std::io::Error> {
-        Ok(false)
-    }
-}
-
+/// Register the shared [`MockWarmBackend`] into this instance's tier config
+/// manager. Thin wrapper over the shared `register_mock_tier` helper
+/// (rustfs/backlog#1148 ilm-6).
 async fn register_mock_tier(tier_name: &str) -> MockWarmBackend {
-    let backend = MockWarmBackend::default();
-    let tier_config_mgr_handle = current_tier_config_handle();
-    let mut tier_config_mgr = tier_config_mgr_handle.write().await;
-    tier_config_mgr.tiers.insert(
-        tier_name.to_string(),
-        TierConfig {
-            version: "v1".to_string(),
-            tier_type: TierType::MinIO,
-            name: tier_name.to_string(),
-            ..Default::default()
-        },
-    );
-    tier_config_mgr
-        .driver_cache
-        .insert(tier_name.to_string(), Box::new(backend.clone()));
-    backend
+    register_mock_tier_util(&current_tier_config_handle(), tier_name).await
 }
 
 async fn wait_for_transition(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> Option<ObjectInfo> {
@@ -441,22 +359,6 @@ where
     }
 }
 
-async fn wait_for_remote_absence(backend: &MockWarmBackend, object: &str, timeout: Duration) -> bool {
-    let deadline = tokio::time::Instant::now() + timeout;
-
-    loop {
-        if !backend.objects.lock().await.contains_key(object) {
-            return true;
-        }
-
-        if tokio::time::Instant::now() >= deadline {
-            return false;
-        }
-
-        tokio::time::sleep(Duration::from_millis(50)).await;
-    }
-}
-
 async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
     let deadline = tokio::time::Instant::now() + timeout;
 
@@ -851,7 +753,7 @@ async fn put_and_copy_object_transition_immediately_via_usecases() {
 
     assert_eq!(put_info.transitioned_object.status, "complete");
     assert_eq!(put_info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&put_info.transitioned_object.name));
+    assert!(backend.contains(&put_info.transitioned_object.name).await);
 
     let src_bucket = format!("test-api-copy-src-{}", &Uuid::new_v4().simple().to_string()[..8]);
     let dst_bucket = format!("test-api-copy-dst-{}", &Uuid::new_v4().simple().to_string()[..8]);
@@ -887,7 +789,7 @@ async fn put_and_copy_object_transition_immediately_via_usecases() {
 
     assert_eq!(copy_info.transitioned_object.status, "complete");
     assert_eq!(copy_info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&copy_info.transitioned_object.name));
+    assert!(backend.contains(&copy_info.transitioned_object.name).await);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -944,7 +846,7 @@ async fn complete_multipart_upload_transitions_immediately_via_usecase() {
 
     assert_eq!(info.transitioned_object.status, "complete");
     assert_eq!(info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+    assert!(backend.contains(&info.transitioned_object.name).await);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -993,13 +895,7 @@ async fn get_transitioned_object_uses_remote_codec_fallback_path() {
         assert_eq!(transitioned.transitioned_object.status, "complete");
         assert_eq!(transitioned.transitioned_object.tier, tier_name);
         assert!(!transitioned.transitioned_object.name.is_empty());
-        assert!(
-            backend
-                .objects
-                .lock()
-                .await
-                .contains_key(&transitioned.transitioned_object.name)
-        );
+        assert!(backend.contains(&transitioned.transitioned_object.name).await);
 
         let actual = read_object_bytes(&ecstore, bucket.as_str(), object).await;
         assert_eq!(actual, payload);
@@ -1036,7 +932,7 @@ async fn delete_transitioned_object_removes_remote_tier_copy_via_usecase() {
         .expect("object should transition before delete usecase runs");
     let remote_object = transitioned.transitioned_object.name.clone();
 
-    assert!(backend.objects.lock().await.contains_key(&remote_object));
+    assert!(backend.contains(&remote_object).await);
 
     let mut req = build_request(
         DeleteObjectInput::builder()
@@ -1058,7 +954,7 @@ async fn delete_transitioned_object_removes_remote_tier_copy_via_usecase() {
     );
 
     assert!(
-        wait_for_remote_absence(&backend, &remote_object, TRANSITION_WAIT_TIMEOUT).await,
+        backend.wait_for_remote_absence(&remote_object, TRANSITION_WAIT_TIMEOUT).await,
         "transitioned object should be removed from remote tier after delete usecase"
     );
 }
@@ -1134,7 +1030,7 @@ async fn immediate_transition_timeout_eventually_completes_via_compensation() {
 
     assert_eq!(info.transitioned_object.status, "complete");
     assert_eq!(info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+    assert!(backend.contains(&info.transitioned_object.name).await);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -1184,7 +1080,7 @@ async fn compensation_driven_copy_still_completes_transition() {
 
     assert_eq!(info.transitioned_object.status, "complete");
     assert_eq!(info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+    assert!(backend.contains(&info.transitioned_object.name).await);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -1244,7 +1140,7 @@ async fn compensation_driven_complete_multipart_upload_still_transitions() {
 
     assert_eq!(info.transitioned_object.status, "complete");
     assert_eq!(info.transitioned_object.tier, tier_name);
-    assert!(backend.objects.lock().await.contains_key(&info.transitioned_object.name));
+    assert!(backend.contains(&info.transitioned_object.name).await);
 }
 
 #[tokio::test(flavor = "multi_thread", worker_threads = 1)]
@@ -1276,7 +1172,7 @@ async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() {
         .expect("object should eventually transition after compensation backfill");
     let remote_object = transitioned.transitioned_object.name.clone();
 
-    assert!(backend.objects.lock().await.contains_key(&remote_object));
+    assert!(backend.contains(&remote_object).await);
 
     let mut req = build_request(
         DeleteObjectInput::builder()
@@ -1298,7 +1194,7 @@ async fn compensation_driven_transition_still_cleans_remote_tier_on_delete() {
     );
 
     assert!(
-        wait_for_remote_absence(&backend, &remote_object, TRANSITION_WAIT_TIMEOUT).await,
+        backend.wait_for_remote_absence(&remote_object, TRANSITION_WAIT_TIMEOUT).await,
         "transitioned object should be removed from remote tier after delete usecase"
     );
 }
@@ -1332,7 +1228,7 @@ async fn compensation_driven_versioned_delete_still_creates_delete_marker() {
         .expect("object should eventually transition after compensation backfill");
     let remote_object = transitioned.transitioned_object.name.clone();
 
-    assert!(backend.objects.lock().await.contains_key(&remote_object));
+    assert!(backend.contains(&remote_object).await);
 
     let req = build_request(
         DeleteObjectInput::builder()
@@ -1352,7 +1248,7 @@ async fn compensation_driven_versioned_delete_still_creates_delete_marker() {
         "versioned delete should create a delete marker after compensation-driven transition"
     );
     assert!(
-        backend.objects.lock().await.contains_key(&remote_object),
+        backend.contains(&remote_object).await,
         "creating a delete marker should not remove the transitioned remote object version"
     );
 }
@@ -1387,7 +1283,7 @@ async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() {
         .expect("object should eventually transition after compensation backfill");
     let remote_object = transitioned.transitioned_object.name.clone();
 
-    assert!(backend.objects.lock().await.contains_key(&remote_object));
+    assert!(backend.contains(&remote_object).await);
 
     let req = build_request(
         DeleteObjectInput::builder()
@@ -1407,7 +1303,7 @@ async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() {
         "versioned delete should create a delete marker before lifecycle cleanup"
     );
     assert!(
-        backend.objects.lock().await.contains_key(&remote_object),
+        backend.contains(&remote_object).await,
         "delete marker creation should keep the transitioned remote object version"
     );
 
@@ -1429,7 +1325,7 @@ async fn compensation_driven_delete_marker_still_honors_lifecycle_cleanup() {
         "delete marker should remain visible after lifecycle update until cleanup completes"
     );
     assert!(
-        backend.objects.lock().await.contains_key(&remote_object),
+        backend.contains(&remote_object).await,
         "delete marker lifecycle cleanup should not remove the transitioned remote object version"
     );
 }

rustfs/src/app/storage_api.rs

@@ -145,14 +145,11 @@ pub(crate) mod runtime {
     pub(crate) type TierConfigMgr = crate::storage::storage_api::TierConfigMgr;
     pub(crate) type TransitionState = crate::storage::storage_api::TransitionState;
 
+    // Shared MockWarmBackend + register_mock_tier helper (rustfs/backlog#1148 ilm-6).
+    // The mock (and the tier types it used to need) now live in ecstore behind
+    // the `test-util` feature.
     #[cfg(test)]
-    pub(crate) type TierConfig = crate::storage::storage_api::ecstore_tier::tier_config::TierConfig;
-    #[cfg(test)]
-    pub(crate) type TierType = crate::storage::storage_api::ecstore_tier::tier_config::TierType;
-    #[cfg(test)]
-    pub(crate) use crate::storage::storage_api::ecstore_tier::warm_backend::WarmBackend as AppWarmBackend;
-    #[cfg(test)]
-    pub(crate) type WarmBackendGetOpts = crate::storage::storage_api::ecstore_tier::warm_backend::WarmBackendGetOpts;
+    pub(crate) use crate::storage::storage_api::ecstore_tier::test_util::{MockWarmBackend, register_mock_tier};
 
     pub(crate) fn set_global_storage_class(cfg: StorageClassConfig) {
         crate::storage::storage_api::ecstore_config::set_global_storage_class(cfg);
@@ -807,12 +804,6 @@ pub(crate) mod bucket {
         }
     }
 
-    #[cfg(test)]
-    pub(crate) mod transition_api {
-        pub(crate) type ReadCloser = crate::storage::storage_api::ecstore_client::transition_api::ReadCloser;
-        pub(crate) type ReaderImpl = crate::storage::storage_api::ecstore_client::transition_api::ReaderImpl;
-    }
-
     pub(crate) mod versioning_sys {
         pub(crate) type BucketVersioningSys = crate::storage::storage_api::ecstore_bucket::versioning_sys::BucketVersioningSys;
     }

rustfs/src/storage/storage_api.rs

@@ -335,8 +335,6 @@ pub(crate) mod ecstore_capacity {
 }
 
 pub(crate) mod ecstore_client {
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::client::transition_api;
     pub(crate) use rustfs_ecstore::api::client::{admin_handler_utils, object_api_utils};
 }
 
@@ -464,9 +462,11 @@ pub(crate) mod ecstore_storage {
 
 pub(crate) mod ecstore_tier {
     pub(crate) use rustfs_ecstore::api::tier::tier::TierConfigMgr;
-    #[cfg(test)]
-    pub(crate) use rustfs_ecstore::api::tier::warm_backend;
     pub(crate) use rustfs_ecstore::api::tier::{tier, tier_admin, tier_config, tier_handlers};
+    // Shared lifecycle/tier test utilities behind ecstore's `test-util` feature
+    // (rustfs/backlog#1148 ilm-6). Only linked into test builds.
+    #[cfg(test)]
+    pub(crate) use rustfs_ecstore::api::tier::test_util;
 }
 
 pub(crate) const BUCKET_ACCELERATE_CONFIG: &str = ecstore_bucket::metadata::BUCKET_ACCELERATE_CONFIG;