From e76cb67e93b2ae2d4aa9ed9184fbe11c573fe222 Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 16 Oct 2026 11:09:55 +0800 Subject: [PATCH] fix(lifecycle): read exact object versions after writes (#8601) * fix(lifecycle): read exact object versions after writes * test(lifecycle): respect ECStore dependency boundaries --- .../bucket/lifecycle/bucket_lifecycle_ops.rs | 131 +++-- crates/ecstore/src/disk/lifecycle_test_io.rs | 72 +++ crates/ecstore/src/disk/local.rs | 4 + crates/ecstore/src/disk/mod.rs | 2 + crates/ecstore/src/set_disk/metadata.rs | 52 ++ crates/ecstore/src/store/list_objects.rs | 78 ++- .../list_objects/lifecycle_exact_tests.rs | 498 ++++++++++++++++++ crates/ecstore/src/store/object.rs | 2 +- 8 files changed, 801 insertions(+), 38 deletions(-) create mode 100644 crates/ecstore/src/disk/lifecycle_test_io.rs create mode 100644 crates/ecstore/src/store/list_objects/lifecycle_exact_tests.rs diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 3f1f699e3..be12180d5 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -3966,44 +3966,28 @@ pub(crate) async fn enqueue_immediate_expiry(api: Arc, oi: &ObjectInfo, return; }; - let mut marker = None; - let mut version_marker = None; - let mut object_infos = Vec::new(); - - loop { - let page = match api - .clone() - .list_object_versions_for_lifecycle(&oi.bucket, &oi.name, marker.clone(), version_marker.clone(), None, 1000) - .await - { - Ok(page) => page, - Err(err) => { - warn!( - event = EVENT_LIFECYCLE_EVALUATION_FAILED, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_LIFECYCLE, - bucket = %oi.bucket, - object = %oi.name, - error = %err, - reason = "list_versions_failed", - "Failed to load lifecycle version group" - ); - return; - } - }; - - object_infos.extend(page.objects.into_iter().filter(|object| object.name == oi.name)); - - if !page.is_truncated { - break; + let object_infos = match api + .get_object_versions_for_lifecycle(&oi.bucket, &oi.name, opts, configs.bucket_incarnation_id) + .await + { + Ok(objects) => objects, + Err(err) => { + warn!( + event = EVENT_LIFECYCLE_EVALUATION_FAILED, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_LIFECYCLE, + result = "failed", + bucket = %oi.bucket, + object = %oi.name, + error = %err, + reason = "read_versions_failed", + "Failed to load lifecycle version group" + ); + return; } - - marker = page.next_marker; - version_marker = page.next_version_idmarker; - } - + }; if object_infos.is_empty() { - object_infos.push(oi.clone()); + return; } let object_opts = object_infos @@ -13221,6 +13205,81 @@ mod tests { assert_eq!(event.action, IlmAction::NoneAction); } + #[tokio::test] + #[serial] + async fn lifecycle_exact_immediate_expiry_preserves_guards_and_skips_stale_writes() { + let (paths, ecstore) = setup_test_env().await; + let bucket = format!("lifecycle-exact-expiry-{}", Uuid::new_v4().simple()); + let object = "object"; + create_test_bucket(&ecstore, &bucket).await; + let oi = ecstore + .put_object( + &bucket, + object, + &mut PutObjReader::from_vec(b"payload".to_vec()), + &ObjectOptions::default(), + ) + .await + .expect("create object before configuring expiry"); + + // Exercise persisted legacy immediate rules without relaxing the S3 + // validator, which correctly rejects newly submitted Expiration.Days=0. + let sys = metadata_sys::bucket_metadata_sys_of(&ecstore.ctx).expect("metadata system"); + let sys = sys.read().await.clone(); + let mut metadata = (*sys.get(&bucket).await.expect("bucket metadata")).clone(); + let mut lifecycle = latest_expiration_lifecycle(); + lifecycle.rules[0].expiration.as_mut().expect("expiration rule").days = Some(0); + metadata.lifecycle_config_xml = crate::bucket::utils::serialize(&lifecycle).expect("encode legacy policy"); + metadata.lifecycle_config = Some(lifecycle); + sys.persist_and_set(metadata).await.expect("persist legacy policy"); + + let state = install_unconsumed_runtime_expiry_worker(&ecstore, 4).await; + super::enqueue_immediate_expiry(ecstore.clone(), &oi, LcEventSrc::S3PutObject, &ObjectOptions::default()).await; + assert_eq!(state.read().await.stats.pending_tasks(), 1, "exact reads must preserve immediate expiry"); + + let state = install_unconsumed_runtime_expiry_worker(&ecstore, 4).await; + for path in &paths { + let metadata_path = path.join(&bucket).join(object).join(crate::disk::STORAGE_FORMAT_FILE); + let mut meta = rustfs_filemeta::FileMeta::load(&fs::read(&metadata_path).await.expect("read object metadata")) + .expect("decode object metadata"); + let mut version = meta + .get_all_file_info_versions(&bucket, object, true) + .expect("version group") + .versions + .remove(0); + version.metadata.insert("x-amz-object-lock-legal-hold".into(), "ON".into()); + meta.add_version(version).expect("persist legal hold"); + fs::write(metadata_path, meta.marshal_msg().expect("encode held metadata")) + .await + .expect("write legal hold"); + } + super::enqueue_immediate_expiry(ecstore.clone(), &oi, LcEventSrc::S3PutObject, &ObjectOptions::default()).await; + assert_eq!( + state.read().await.stats.pending_tasks(), + 0, + "use current lock metadata, not the stale write result" + ); + + for path in &paths { + fs::write(path.join(&bucket).join(object).join(crate::disk::STORAGE_FORMAT_FILE), b"corrupt") + .await + .expect("inject corrupt metadata"); + } + super::enqueue_immediate_expiry(ecstore.clone(), &oi, LcEventSrc::S3PutObject, &ObjectOptions::default()).await; + assert_eq!(state.read().await.stats.pending_tasks(), 0, "read failure must never enqueue expiry"); + for path in &paths { + fs::remove_dir_all(path.join(&bucket).join(object)) + .await + .expect("simulate concurrent object removal"); + } + super::enqueue_immediate_expiry(ecstore.clone(), &oi, LcEventSrc::S3PutObject, &ObjectOptions::default()).await; + assert_eq!( + state.read().await.stats.pending_tasks(), + 0, + "confirmed absence must not fall back to the old ObjectInfo" + ); + } + #[tokio::test] #[serial] async fn lifecycle_expiry_fails_closed_on_corrupt_object_lock_metadata() { diff --git a/crates/ecstore/src/disk/lifecycle_test_io.rs b/crates/ecstore/src/disk/lifecycle_test_io.rs new file mode 100644 index 000000000..e331b8869 --- /dev/null +++ b/crates/ecstore/src/disk/lifecycle_test_io.rs @@ -0,0 +1,72 @@ +// 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. + +//! Bucket-scoped observations distinguish exact metadata reads from walks. +use std::collections::HashMap; +use std::sync::{Arc, Mutex, OnceLock}; + +#[derive(Default, Clone)] +pub(crate) struct Counts { + pub walks: usize, + pub reads: Vec, +} + +fn observers() -> &'static Mutex>>> { + static OBSERVERS: OnceLock>>>> = OnceLock::new(); + OBSERVERS.get_or_init(Mutex::default) +} + +pub(crate) struct Observer { + bucket: String, + counts: Arc>, +} + +impl Observer { + pub fn new(bucket: &str) -> Self { + let counts = Arc::new(Mutex::new(Counts::default())); + assert!( + observers() + .lock() + .expect("I/O observers lock") + .insert(bucket.to_owned(), counts.clone()) + .is_none(), + "each test must use its own bucket" + ); + Self { + bucket: bucket.to_owned(), + counts, + } + } + + pub fn snapshot(&self) -> Counts { + self.counts.lock().expect("I/O counts lock").clone() + } +} + +impl Drop for Observer { + fn drop(&mut self) { + observers().lock().expect("I/O observers lock").remove(&self.bucket); + } +} + +pub(crate) fn observe(bucket: &str, object: Option<&str>) { + let observer = observers().lock().expect("I/O observers lock").get(bucket).cloned(); + if let Some(observer) = observer { + let mut counts = observer.lock().expect("I/O counts lock"); + match object { + Some(object) => counts.reads.push(object.to_owned()), + None => counts.walks += 1, + } + } +} diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index 953eab121..b7c7ff210 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -11656,6 +11656,8 @@ impl DiskAPI for LocalDisk { // TODO(backlog): support io.writer cancellation and early termination in walk_dir #[tracing::instrument(level = "trace", skip_all)] async fn walk_dir(&self, opts: WalkDirOptions, wr: &mut W) -> Result<()> { + #[cfg(test)] + super::lifecycle_test_io::observe(&opts.bucket, None); self.wait_for_startup_cleanup().await; // Callers that do not pin a stall budget inherit the configured default, so @@ -12225,6 +12227,8 @@ impl DiskAPI for LocalDisk { #[tracing::instrument(level = "trace", skip_all)] async fn read_xl(&self, volume: &str, path: &str, read_data: bool) -> Result { + #[cfg(test)] + super::lifecycle_test_io::observe(volume, Some(path)); crate::hp_guard!("LocalDisk::read_xl"); let file_path = self.io_get_object_path(volume, path)?; let file_dir = self.io_get_bucket_path(volume)?; diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index e800af80a..5bf770e44 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -24,6 +24,8 @@ pub mod format; pub mod fs; pub mod health_state; mod io_admission; +#[cfg(test)] +pub(crate) mod lifecycle_test_io; pub mod local; pub mod os; diff --git a/crates/ecstore/src/set_disk/metadata.rs b/crates/ecstore/src/set_disk/metadata.rs index 36e3bf6d6..6be934391 100644 --- a/crates/ecstore/src/set_disk/metadata.rs +++ b/crates/ecstore/src/set_disk/metadata.rs @@ -35,6 +35,58 @@ struct FileInfoIdentityGroup { } impl SetDisks { + /// Read one key's complete version stack without enumerating its parent. + pub(crate) async fn read_object_versions_for_lifecycle( + &self, + bucket: &str, + object: &str, + ) -> disk::error::Result> { + let disks = self.get_disks_internal().await; + if disks.len() != self.set_drive_count || disks.is_empty() { + return Err(DiskError::ErasureReadQuorum); + } + let disk_object = rustfs_utils::path::encode_dir_object(object); + let (mut raw, mut errors) = Self::read_all_raw_file_info(&disks, bucket, &disk_object, false).await; + for (metadata, error) in raw.iter_mut().zip(&mut errors) { + if metadata.as_ref().is_some_and(|metadata| metadata.buf.is_empty()) { + *metadata = None; + *error = Some(DiskError::FileCorrupt); + } + } + // An unavailable disk is not evidence that a version group is absent. + // Unlike a walk, an exact read must also decide the all-missing case. + if raw.iter().all(Option::is_none) { + return if errors.iter().all(|error| { + matches!( + error, + Some(DiskError::FileNotFound | DiskError::FileVersionNotFound | DiskError::VolumeNotFound) + ) + }) { + Ok(None) + } else { + Err(DiskError::ErasureReadQuorum) + }; + } + let entries = rustfs_filemeta::MetaCacheEntries( + raw.into_iter() + .map(|raw| { + raw.map(|raw| rustfs_filemeta::MetaCacheEntry { + name: object.to_owned(), + metadata: raw.buf, + cached: None, + reusable: false, + }) + }) + .collect(), + ); + // Reuse the listing's per-version authority, including the object's + // actual erasure layout. Never turn unreadable/stale metadata into a + // singleton made from the preceding write's ObjectInfo. + Self::resolve_listed_versions(bucket, entries, &errors, self.set_drive_count, self.default_parity_count)? + .map(Some) + .ok_or(DiskError::ErasureReadQuorum) + } + /// Resolve each listed version using the same metadata authority as an exact read. pub(crate) fn resolve_listed_versions( bucket: &str, diff --git a/crates/ecstore/src/store/list_objects.rs b/crates/ecstore/src/store/list_objects.rs index 73ab5c1da..868d7c62b 100644 --- a/crates/ecstore/src/store/list_objects.rs +++ b/crates/ecstore/src/store/list_objects.rs @@ -4264,6 +4264,79 @@ impl ECStore { .await } + /// Capture one logical key's version group for write-triggered expiry. + pub(crate) async fn get_object_versions_for_lifecycle( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + bucket_incarnation_id: Uuid, + ) -> Result> { + crate::bucket::utils::check_object_args(bucket, object)?; + let mut opts = opts.clone(); + // Lock order matches writes: bucket lifecycle, metadata validation, + // then object namespace. Borrow fences supplied by an outer operation. + let bucket_guard = if opts.bucket_lifecycle_lock_fence.is_none() { + let guard = self.acquire_bucket_lifecycle_read_lock(bucket).await?; + opts.add_bucket_lifecycle_lock_guard(&guard); + Some(guard) + } else { + None + }; + let incarnation = crate::bucket::metadata_sys::get_bucket_incarnation_id_for_options_in(&self.ctx, bucket, &opts).await?; + if incarnation != bucket_incarnation_id { + return Err(Error::BucketNotFound(bucket.to_owned())); + } + let metadata_sys = crate::bucket::metadata_sys::bucket_metadata_sys_of(&self.ctx)?; + let metadata_sys = metadata_sys.read().await.clone(); + let (versioning, _) = metadata_sys.get_versioning_config(bucket).await?; + let disk_object = path::encode_dir_object(object); + let object_guard = self + .acquire_object_read_lock_if_needed("lifecycle_versions", bucket, &disk_object, &mut opts) + .await?; + + // As with version listings, include every physical set: migration can + // leave a key's history in more than its current hash destination. + let reads = self + .pools + .iter() + .flat_map(|pool| &pool.disk_set) + .map(|set| set.read_object_versions_for_lifecycle(bucket, object)); + let mut entries = Vec::new(); + for result in join_all(reads).await { + if let Some(entry) = result.map_err(|error| to_object_err(error.into(), vec![bucket, object]))? { + entries.push(entry); + } + } + if self.pools.is_empty() + || bucket_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) + || opts + .bucket_lifecycle_lock_fence + .as_ref() + .is_some_and(|fence| fence.is_lock_lost()) + || opts.namespace_lock_fence.as_ref().is_some_and(|fence| fence.is_lock_lost()) + || object_guard.as_ref().is_some_and(|guard| guard.is_lock_lost()) + { + return Err(Error::PreconditionFailed); + } + let mut entries = entries.into_iter(); + let Some(mut entry) = entries.next() else { + return Ok(Vec::new()); + }; + if entries.len() > 0 { + merge_object_entry_versions(&mut entry, entries)?; + } + // This projection excludes tier free-version bookkeeping but retains + // pending version purges for the evaluator's replication safeguards. + let versions = entry.file_info_versions(bucket)?; + let versioned = versioning.versioned(object); + Ok(versions + .versions + .iter() + .map(|version| ObjectInfo::from_file_info(version, bucket, object, versioned)) + .collect()) + } + async fn inner_list_object_versions_with_projection( self: Arc, input: ListObjectVersionsInput<'_>, @@ -7293,6 +7366,9 @@ fn calc_common_counter(infos: &[DiskInfo], read_quorum: usize) -> u64 { #[cfg(test)] mod differential_tests; +#[cfg(test)] +mod lifecycle_exact_tests; + #[cfg(test)] mod test { use super::{ @@ -7522,7 +7598,7 @@ mod test { } } - fn test_pending_version_purge_meta_entry(bucket: &str, name: &str) -> (MetaCacheEntry, Uuid) { + pub(super) fn test_pending_version_purge_meta_entry(bucket: &str, name: &str) -> (MetaCacheEntry, Uuid) { let version_id = Uuid::from_u128(0x1234_5678_9abc_def0_1234_5678_9abc_def0); let data_dir = Uuid::from_u128(0xabcd_ef12_3456_7890_abcd_ef12_3456_7890); let mod_time = time::OffsetDateTime::from_unix_timestamp(1_705_312_300).expect("valid timestamp"); diff --git a/crates/ecstore/src/store/list_objects/lifecycle_exact_tests.rs b/crates/ecstore/src/store/list_objects/lifecycle_exact_tests.rs new file mode 100644 index 000000000..91cf0c0a5 --- /dev/null +++ b/crates/ecstore/src/store/list_objects/lifecycle_exact_tests.rs @@ -0,0 +1,498 @@ +// 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. + +use super::*; +use crate::bucket::lifecycle::core::IlmAction; +use crate::bucket::lifecycle::evaluator::Evaluator; +use crate::bucket::metadata::{BUCKET_LIFECYCLE_CONFIG, BUCKET_VERSIONING_CONFIG}; +use crate::bucket::metadata_sys; +use crate::disk::lifecycle_test_io::Observer; +use crate::layout::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; +use crate::object_api::PutObjReader; +use crate::runtime::instance::InstanceContext; +use crate::storage_api_contracts::bucket::{BucketOperations, MakeBucketOptions}; +use crate::storage_api_contracts::multipart::{CompletePart, MultipartOperations}; +use crate::storage_api_contracts::object::ObjectIO; +use rustfs_filemeta::{FileInfo, ObjectPartInfo}; +use tempfile::TempDir; +use time::OffsetDateTime; + +async fn fixture(pool_count: usize) -> (Vec, Arc, String, Uuid) { + let mut dirs = Vec::new(); + let mut pools = Vec::new(); + for pool_idx in 0..pool_count { + let mut endpoints = Vec::new(); + for disk_idx in 0..4 { + let dir = tempfile::tempdir().expect("disk directory"); + let mut endpoint = + crate::disk::endpoint::Endpoint::try_from(dir.path().to_str().expect("disk path")).expect("disk endpoint"); + endpoint.set_pool_index(pool_idx); + endpoint.set_set_index(0); + endpoint.set_disk_index(disk_idx); + dirs.push(dir); + endpoints.push(endpoint); + } + pools.push(PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: format!("lifecycle-exact-pool-{pool_idx}"), + platform: "test".into(), + }); + } + let endpoints = EndpointServerPools(pools); + let ctx = Arc::new(InstanceContext::new()); + crate::store::init_local_disks_with_instance_ctx(&ctx, endpoints.clone()) + .await + .expect("initialize disks"); + let store = + ECStore::new_with_instance_ctx("127.0.0.1:0".parse().expect("test address"), endpoints, CancellationToken::new(), ctx) + .await + .expect("initialize store"); + metadata_sys::init_bucket_metadata_sys(store.clone(), Vec::new()).await; + let bucket = format!("lifecycle-exact-{}", Uuid::new_v4().simple()); + store + .make_bucket(&bucket, &MakeBucketOptions::default()) + .await + .expect("create bucket"); + let incarnation = metadata_sys::get_bucket_incarnation_id_in(&store.ctx, &bucket) + .await + .expect("bucket incarnation"); + (dirs, store, bucket, incarnation) +} + +fn version(object: &str, id: Option, timestamp: i64) -> FileInfo { + let mut fi = FileInfo::new(object, 2, 2); + fi.name = object.to_owned(); + fi.version_id = id; + fi.versioned = id.is_some(); + fi.data_dir = Some(Uuid::from_u128(0x1234)); + fi.mod_time = Some(OffsetDateTime::from_unix_timestamp(timestamp).expect("version timestamp")); + fi.size = 1; + fi.parts = vec![ObjectPartInfo { + number: 1, + size: 1, + actual_size: 1, + ..Default::default() + }]; + fi.metadata.insert("etag".into(), timestamp.to_string()); + fi +} + +async fn seed(dirs: &[TempDir], bucket: &str, object: &str, versions: &[FileInfo]) { + for (index, dir) in dirs.iter().enumerate() { + let mut meta = FileMeta::new(); + for version in versions { + let mut fi = version.clone(); + fi.volume = bucket.to_owned(); + if !fi.is_canonical_delete_marker() { + fi.erasure.index = index % 4 + 1; + } + fi.validate_for_metadata_read().expect("valid fixture version"); + meta.add_version(fi).expect("encode version"); + } + let path = dir.path().join(bucket).join(path::encode_dir_object(object)); + tokio::fs::create_dir_all(&path).await.expect("object directory"); + tokio::fs::write(path.join(crate::disk::STORAGE_FORMAT_FILE), meta.marshal_msg().expect("encode metadata")) + .await + .expect("persist fixture metadata"); + } +} + +async fn configure_versioning(store: &ECStore, bucket: &str, state: &str) { + if state == "Unversioned" { + return; + } + metadata_sys::update_in( + &store.ctx, + bucket, + BUCKET_VERSIONING_CONFIG, + format!("{state}").into_bytes(), + ) + .await + .expect("persist versioning configuration"); +} + +#[tokio::test] +async fn lifecycle_exact_matches_listing_for_complete_version_groups() { + let (dirs, store, bucket, incarnation) = fixture(1).await; + for state in ["Unversioned", "Enabled", "Suspended"] { + configure_versioning(&store, &bucket, state).await; + for object in ["abc", "nested/abc", "directory/"] { + let mut versions = vec![version(object, None, 1_700_000_000)]; + if state != "Unversioned" { + versions.push(version(object, Some(Uuid::from_u128(1)), 1_700_000_001)); + versions.push(FileInfo { + name: object.into(), + version_id: Some(Uuid::from_u128(2)), + deleted: true, + mod_time: Some(OffsetDateTime::from_unix_timestamp(1_700_000_002).expect("marker timestamp")), + ..Default::default() + }); + } + seed(&dirs, &bucket, object, &versions).await; + let sibling = format!("{object}-sibling"); + seed(&dirs, &bucket, &sibling, &[version(&sibling, None, 1_700_000_003)]).await; + let observed = Observer::new(&bucket); + let exact = store + .get_object_versions_for_lifecycle(&bucket, object, &ObjectOptions::default(), incarnation) + .await + .unwrap_or_else(|error| panic!("read {state} versions for {object}: {error}")); + let counts = observed.snapshot(); + assert_eq!(counts.walks, 0, "exact metadata reads must never enumerate a directory"); + assert_eq!(counts.reads.len(), 4, "one exact metadata read per physical drive"); + assert!(counts.reads.iter().all(|key| key == &path::encode_dir_object(object))); + let mut listed = store + .clone() + .list_object_versions_for_lifecycle(&bucket, object, None, None, None, 1000) + .await + .expect("reference lifecycle listing") + .objects + .into_iter() + .filter(|info| info.name == object) + .collect::>(); + // The legacy list projection uses ambient bucket versioning. This + // fixture has an isolated store; restore its known null-version + // representation before comparing the version-group semantics. + if state != "Unversioned" { + for info in &mut listed { + if info.version_id.is_none() { + info.version_id = Some(Uuid::nil()); + } + } + assert!(exact.iter().any(|info| info.version_id == Some(Uuid::nil()))); + } + assert!(observed.snapshot().walks > 0, "the old prefix-listing path must exercise the observer"); + assert_eq!(exact.len(), versions.len(), "{state}: {object}"); + assert_eq!(exact.len(), listed.len()); + for (exact, listed) in exact.iter().zip(&listed) { + assert_eq!( + ( + exact.version_id, + exact.is_latest, + exact.delete_marker, + exact.num_versions, + exact.successor_mod_time + ), + ( + listed.version_id, + listed.is_latest, + listed.delete_marker, + listed.num_versions, + listed.successor_mod_time + ) + ); + assert_eq!(exact.user_defined, listed.user_defined); + assert_eq!(exact.version_purge_status, listed.version_purge_status); + } + let policy = crate::bucket::utils::deserialize( + br#"historyEnabled3011"#, + ).expect("reference policy"); + let evaluator = Evaluator::new(Arc::new(policy)); + let to_opts = |objects: &[ObjectInfo]| { + objects + .iter() + .map(crate::bucket::lifecycle::core::object_opts_from_object_info) + .collect::>() + }; + let exact_events = evaluator.eval(&to_opts(&exact)).await.expect("evaluate exact group"); + let listed_events = evaluator.eval(&to_opts(&listed)).await.expect("evaluate listed group"); + for (exact, listed) in exact_events.iter().zip(listed_events) { + assert_eq!((exact.action, exact.due, &exact.rule_id), (listed.action, listed.due, &listed.rule_id)); + } + } + } +} + +#[tokio::test] +async fn lifecycle_exact_merges_pools_and_more_than_one_page_of_versions() { + let (dirs, store, bucket, incarnation) = fixture(2).await; + configure_versioning(&store, &bucket, "Enabled").await; + let object = "history"; + let versions = (0..1002) + .map(|index| { + version( + object, + Some(Uuid::from_u128(index + 1)), + 1_700_000_000 + i64::try_from(index).expect("timestamp index"), + ) + }) + .collect::>(); + seed(&dirs[..4], &bucket, object, &versions[..601]).await; + seed(&dirs[4..], &bucket, object, &versions[600..]).await; + let observed = Observer::new(&bucket); + let exact = store + .get_object_versions_for_lifecycle(&bucket, object, &ObjectOptions::default(), incarnation) + .await + .expect("merge independently authoritative pools"); + assert_eq!(exact.len(), 1002, "overlapping migration copies must be deduplicated without pagination"); + assert_eq!(exact[0].version_id, Some(Uuid::from_u128(1002))); + assert!(exact[0].is_latest); + assert!(exact.iter().all(|info| info.num_versions == 1002)); + for pair in exact.windows(2) { + assert!(!pair[1].is_latest); + assert_eq!(pair[1].successor_mod_time, pair[0].mod_time); + } + assert_eq!(observed.snapshot().walks, 0); + assert_eq!(observed.snapshot().reads.len(), 8, "read count depends on drives, not version pages"); + let mut conflict = versions[600].clone(); + conflict.metadata.insert("etag".into(), "conflicting-content".into()); + seed(&dirs[4..], &bucket, object, &[conflict]).await; + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &ObjectOptions::default(), incarnation) + .await + .is_err(), + "conflicting cross-pool copies cannot authorize expiry" + ); +} + +#[tokio::test] +async fn lifecycle_exact_rejects_uncertain_metadata_and_stale_bucket_identity() { + let (dirs, store, bucket, incarnation) = fixture(1).await; + let object = "probe"; + let opts = ObjectOptions::default(); + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &opts, incarnation) + .await + .expect("confirmed absence") + .is_empty() + ); + seed(&dirs, &bucket, object, &[version(object, None, 1_700_000_000)]).await; + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &opts, Uuid::new_v4()) + .await + .is_err() + ); + for dir in &dirs { + tokio::fs::write( + dir.path() + .join(bucket.as_str()) + .join(object) + .join(crate::disk::STORAGE_FORMAT_FILE), + b"corrupt", + ) + .await + .expect("corrupt fixture metadata"); + } + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &opts, incarnation) + .await + .is_err(), + "corruption is not absence" + ); + seed(&dirs, &bucket, object, &[version(object, None, 1_700_000_000)]).await; + let set = &store.pools[0].disk_set[0]; + let saved = set.disks.read().await.clone(); + *set.disks.write().await = vec![saved[0].clone(), saved[1].clone(), None, None]; + assert_eq!( + store + .get_object_versions_for_lifecycle(&bucket, object, &opts, incarnation) + .await + .expect("exactly the object's read quorum") + .len(), + 1 + ); + *set.disks.write().await = vec![saved[0].clone(), None, None, None]; + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &opts, incarnation) + .await + .is_err(), + "a minority cannot authorize expiry" + ); + assert!( + store + .get_object_versions_for_lifecycle(&bucket, "missing", &opts, incarnation) + .await + .is_err(), + "missing on one drive does not prove absence" + ); + *set.disks.write().await = saved; + let lost = ObjectOptions { + no_lock: true, + namespace_lock_fence: Some(crate::object_api::NamespaceLockFence::lost_for_test()), + ..Default::default() + }; + assert!( + store + .get_object_versions_for_lifecycle(&bucket, object, &lost, incarnation) + .await + .is_err(), + "an inherited lost object fence must reject the snapshot" + ); +} + +#[tokio::test] +async fn lifecycle_exact_write_hooks_never_walk_parent_directories() { + let (dirs, store, bucket, _) = fixture(1).await; + metadata_sys::update_in(&store.ctx, &bucket, BUCKET_LIFECYCLE_CONFIG, + br#"expireEnabled30"#.to_vec()) + .await.expect("configure expiration"); + for index in 0..64 { + let key = format!("probe-{index}"); + seed(&dirs, &bucket, &key, &[version(&key, None, 1_700_000_000)]).await; + } + for state in ["Unversioned", "Enabled", "Suspended"] { + configure_versioning(&store, &bucket, state).await; + let opts = ObjectOptions { + versioned: state == "Enabled", + version_suspended: state == "Suspended", + ..Default::default() + }; + let observer = Observer::new(&bucket); + let mut source = store + .put_object(&bucket, "probe", &mut PutObjReader::from_vec(b"payload".to_vec()), &opts) + .await + .expect("PUT with lifecycle"); + source.put_object_reader = Some(PutObjReader::from_vec(b"payload".to_vec())); + let copied = store + .copy_object(&bucket, "probe", &bucket, "copy", &mut source, &ObjectOptions::default(), &opts) + .await + .expect("COPY with lifecycle"); + assert_eq!(copied.size, source.size); + let upload = store + .new_multipart_upload(&bucket, "multipart", &opts) + .await + .expect("init multipart"); + let part = store + .put_object_part( + &bucket, + "multipart", + &upload.upload_id, + 1, + &mut PutObjReader::from_vec(b"part".to_vec()), + &opts, + ) + .await + .expect("upload part"); + let complete = store + .clone() + .complete_multipart_upload( + &bucket, + "multipart", + &upload.upload_id, + vec![CompletePart { + part_num: 1, + etag: part.etag, + checksum_crc32: None, + checksum_crc32c: None, + checksum_sha1: None, + checksum_sha256: None, + checksum_crc64nvme: None, + }], + &opts, + ) + .await + .expect("complete multipart with lifecycle"); + assert_eq!(complete.size, 4); + let counts = observer.snapshot(); + assert_eq!(counts.walks, 0, "{state}: successful writes must not run the version walker"); + for key in ["probe", "copy", "multipart"] { + assert!( + counts.reads.iter().filter(|read| read.as_str() == key).count() >= 4, + "{state}: {key} must run exact expiry evaluation" + ); + } + assert!( + !counts.reads.iter().any(|key| key.starts_with("probe-")), + "prefix siblings must never enter evaluation" + ); + } +} + +#[tokio::test] +async fn lifecycle_exact_keeps_pending_purges_visible_to_expiry_guards() { + let (dirs, store, bucket, incarnation) = fixture(1).await; + configure_versioning(&store, &bucket, "Enabled").await; + let object = "pending"; + let (entry, _) = super::test::test_pending_version_purge_meta_entry(&bucket, object); + for dir in &dirs { + let path = dir.path().join(&bucket).join(object); + tokio::fs::create_dir_all(&path).await.expect("object directory"); + tokio::fs::write(path.join(crate::disk::STORAGE_FORMAT_FILE), &entry.metadata) + .await + .expect("persist pending purge"); + } + let exact = store + .get_object_versions_for_lifecycle(&bucket, object, &ObjectOptions::default(), incarnation) + .await + .expect("read pending purge"); + assert_eq!(exact.len(), 1); + assert_eq!(exact[0].num_versions, 1); + assert_eq!(exact[0].version_purge_status, crate::bucket::replication::VersionPurgeStatusType::Pending); + let policy = crate::bucket::utils::deserialize( + br#"expireEnabled1"#, + ).expect("expiration policy"); + let opts = crate::bucket::lifecycle::core::object_opts_from_object_info(&exact[0]); + let events = Evaluator::new(Arc::new(policy)) + .eval(&[opts]) + .await + .expect("evaluate pending purge"); + assert_eq!(events[0].action, IlmAction::NoneAction); +} + +#[tokio::test] +async fn lifecycle_exact_io_is_independent_of_parent_directory_width() { + let (dirs, store, bucket, incarnation) = fixture(1).await; + let object = "probe"; + seed(&dirs, &bucket, object, &[version(object, None, 1_700_000_000)]).await; + let mut previous = 0; + for width in [64, 1024, 8192] { + let roots = dirs.iter().map(|dir| dir.path().join(&bucket)).collect::>(); + tokio::task::spawn_blocking(move || { + for root in roots { + for index in previous..width { + std::fs::create_dir(root.join(format!("sibling-{index:05}"))).expect("parent-directory entry"); + } + } + }) + .await + .expect("create wide directory"); + previous = width; + let mut exact_times = Vec::new(); + let mut listing_times = Vec::new(); + for _ in 0..5 { + let observer = Observer::new(&bucket); + let started = std::time::Instant::now(); + let exact = store + .get_object_versions_for_lifecycle(&bucket, object, &ObjectOptions::default(), incarnation) + .await + .expect("exact lifecycle read"); + exact_times.push(started.elapsed().as_micros()); + assert_eq!(exact.len(), 1); + assert_eq!(observer.snapshot().walks, 0); + assert_eq!(observer.snapshot().reads.len(), 4, "directory width {width}"); + let started = std::time::Instant::now(); + let listed = store + .clone() + .list_object_versions_for_lifecycle(&bucket, object, None, None, None, 1000) + .await + .expect("reference prefix scan"); + listing_times.push(started.elapsed().as_micros()); + assert_eq!(listed.objects.len(), 1); + assert!(observer.snapshot().walks > 0); + } + exact_times.sort_unstable(); + listing_times.sort_unstable(); + eprintln!( + "lifecycle directory width={width}: exact median={}us max={}us; prefix median={}us max={}us", + exact_times[2], exact_times[4], listing_times[2], listing_times[4] + ); + } +} diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs index 8a4413a0c..9c2127bfc 100644 --- a/crates/ecstore/src/store/object.rs +++ b/crates/ecstore/src/store/object.rs @@ -3457,7 +3457,7 @@ impl ECStore { Ok(guards) } - async fn acquire_object_read_lock_if_needed( + pub(super) async fn acquire_object_read_lock_if_needed( &self, op: &'static str, bucket: &str,