mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 06:18:43 +00:00
fix(lifecycle): read exact object versions after writes (#8601)
* fix(lifecycle): read exact object versions after writes * test(lifecycle): respect ECStore dependency boundaries
This commit is contained in:
cxymds
@@ -3966,44 +3966,28 @@ pub(crate) async fn enqueue_immediate_expiry(api: Arc<ECStore>, 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() {
|
||||
|
||||
@@ -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<String>,
|
||||
}
|
||||
|
||||
fn observers() -> &'static Mutex<HashMap<String, Arc<Mutex<Counts>>>> {
|
||||
static OBSERVERS: OnceLock<Mutex<HashMap<String, Arc<Mutex<Counts>>>>> = OnceLock::new();
|
||||
OBSERVERS.get_or_init(Mutex::default)
|
||||
}
|
||||
|
||||
pub(crate) struct Observer {
|
||||
bucket: String,
|
||||
counts: Arc<Mutex<Counts>>,
|
||||
}
|
||||
|
||||
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,
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -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<W: AsyncWrite + Unpin + Send>(&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<RawFileInfo> {
|
||||
#[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)?;
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<Option<rustfs_filemeta::MetaCacheEntry>> {
|
||||
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,
|
||||
|
||||
@@ -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<Vec<ObjectInfo>> {
|
||||
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<Self>,
|
||||
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");
|
||||
|
||||
@@ -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<TempDir>, Arc<ECStore>, 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<Uuid>, 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!("<VersioningConfiguration><Status>{state}</Status></VersioningConfiguration>").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::<Vec<_>>();
|
||||
// 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#"<LifecycleConfiguration><Rule><ID>history</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>30</Days></Expiration><NoncurrentVersionExpiration><NoncurrentDays>1</NoncurrentDays><NewerNoncurrentVersions>1</NewerNoncurrentVersions></NoncurrentVersionExpiration></Rule></LifecycleConfiguration>"#,
|
||||
).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::<Vec<_>>()
|
||||
};
|
||||
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::<Vec<_>>();
|
||||
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#"<LifecycleConfiguration><Rule><ID>expire</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>30</Days></Expiration></Rule></LifecycleConfiguration>"#.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#"<LifecycleConfiguration><Rule><ID>expire</ID><Status>Enabled</Status><Filter><Prefix></Prefix></Filter><Expiration><Days>1</Days></Expiration></Rule></LifecycleConfiguration>"#,
|
||||
).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::<Vec<_>>();
|
||||
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]
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -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,
|
||||
|
||||
Reference in New Issue
Block a user