From 763f246f8ae4778846cfbe50a9f51adcf6ba7848 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sun, 12 Jul 2026 00:20:45 +0800 Subject: [PATCH] test(ecstore): add shared MockWarmBackend test utility for lifecycle and tier tests (#4716) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit * 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 | 4 + crates/ecstore/src/api/mod.rs | 9 + crates/ecstore/src/services/tier/mod.rs | 2 + crates/ecstore/src/services/tier/test_util.rs | 611 ++++++++++++++++++ crates/scanner/Cargo.toml | 3 + .../tests/lifecycle_integration_test.rs | 311 ++------- crates/scanner/tests/storage_api/mod.rs | 24 +- rustfs/Cargo.toml | 3 + .../src/app/lifecycle_transition_api_test.rs | 148 +---- rustfs/src/app/storage_api.rs | 17 +- rustfs/src/storage/storage_api.rs | 8 +- 11 files changed, 724 insertions(+), 416 deletions(-) create mode 100644 crates/ecstore/src/services/tier/test_util.rs diff --git a/crates/ecstore/Cargo.toml b/crates/ecstore/Cargo.toml index 1a119731a..0a6536ee7 100644 --- a/crates/ecstore/Cargo.toml +++ b/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 } diff --git a/crates/ecstore/src/api/mod.rs b/crates/ecstore/src/api/mod.rs index c9c0ca89f..25dc523e4 100644 --- a/crates/ecstore/src/api/mod.rs +++ b/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, + }; + } } diff --git a/crates/ecstore/src/services/tier/mod.rs b/crates/ecstore/src/services/tier/mod.rs index 4209d4da9..15f975f99 100644 --- a/crates/ecstore/src/services/tier/mod.rs +++ b/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; diff --git a/crates/ecstore/src/services/tier/test_util.rs b/crates/ecstore/src/services/tier/test_util.rs new file mode 100644 index 000000000..183514661 --- /dev/null +++ b/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, +} + +impl FaultConfig { + fn error(&self) -> Option { + 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, + pub metadata: HashMap, + pub remote_version_id: String, +} + +#[derive(Default)] +struct MockWarmBackendInner { + objects: Mutex>, + faults: Mutex, + op_log: Mutex>, +} + +/// 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, +} + +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) { + 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 { + 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 { + 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> { + 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> { + 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, metadata: HashMap) -> 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, 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 { + 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, + ) -> Result { + 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 { + 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 { + 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>, 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>, 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, + /// Number of free versions retained for the object. + pub free_version_count: usize, +} + +async fn open_disk(disk_path: &Path) -> Option { + 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 { + 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>(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 = 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; + } +} diff --git a/crates/scanner/Cargo.toml b/crates/scanner/Cargo.toml index e7f4ed9ad..db5a510f4 100644 --- a/crates/scanner/Cargo.toml +++ b/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 diff --git a/crates/scanner/tests/lifecycle_integration_test.rs b/crates/scanner/tests/lifecycle_integration_test.rs index fb73d12c7..faa7a72e1 100644 --- a/crates/scanner/tests/lifecycle_integration_test.rs +++ b/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, Arc)> = OnceLock::new(); @@ -397,67 +393,6 @@ async fn wait_for_object_absence(ecstore: &Arc, 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, 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, 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, - metadata: HashMap, -} - -#[derive(Clone, Default)] -struct MockWarmBackend { - objects: Arc>>, - // Ordered log of remote-tier mutating operations observed by this backend - // (e.g. `remove:`). 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>>, -} - -#[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, metadata: HashMap) -> 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, 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 { - 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, - ) -> Result { - 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 { - 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 { - 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, bucket: &str, object: &str, timeout: Duration) -> Option { @@ -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(©_info.transitioned_object.name)); + assert!(backend.contains(©_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" ); } diff --git a/crates/scanner/tests/storage_api/mod.rs b/crates/scanner/tests/storage_api/mod.rs index 2a59dffb9..cc653eeae 100644 --- a/crates/scanner/tests/storage_api/mod.rs +++ b/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, }; } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index fbe1c748e..0d710d10e 100644 --- a/rustfs/Cargo.toml +++ b/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 diff --git a/rustfs/src/app/lifecycle_transition_api_test.rs b/rustfs/src/app/lifecycle_transition_api_test.rs index 7c0c06e8d..6e2ac47fa 100644 --- a/rustfs/src/app/lifecycle_transition_api_test.rs +++ b/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>>>, -} - -impl MockWarmBackend { - async fn put_bytes(&self, object: &str, bytes: Vec) -> String { - self.objects.lock().await.insert(object.to_string(), bytes); - Uuid::new_v4().to_string() - } - - async fn read_bytes(&self, reader: ReaderImpl) -> Result, 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 { - 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, - ) -> Result { - 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 { - 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 { - 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(¤t_tier_config_handle(), tier_name).await } async fn wait_for_transition(ecstore: &Arc, bucket: &str, object: &str, timeout: Duration) -> Option { @@ -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, 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(©_info.transitioned_object.name)); + assert!(backend.contains(©_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" ); } diff --git a/rustfs/src/app/storage_api.rs b/rustfs/src/app/storage_api.rs index 7a2fdc376..f6148478f 100644 --- a/rustfs/src/app/storage_api.rs +++ b/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; } diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 788bf8615..efeef4b8c 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/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;