mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-13 09:17:41 +08:00
chore: integrate concurrent improve/base updates
Keep the published QuorumSelected body-GET cache fix and its mutation-boundary assertions alongside the RPC reset classification and exact storage-test stack selectors. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
@@ -2345,7 +2345,7 @@ fn layout_reported_parts(attributes: &aws_sdk_s3::operation::get_object_attribut
|
||||
attributes.object_parts().map(|parts| parts.parts().len())
|
||||
}
|
||||
|
||||
async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
|
||||
async fn assert_layout_full_body(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
|
||||
let label = case.label();
|
||||
let head = layout_head(client, case).await?;
|
||||
assert_eq!(
|
||||
@@ -2362,7 +2362,12 @@ async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &st
|
||||
let full = layout_get(client, case, None, None).await?.body.collect().await?.into_bytes();
|
||||
assert_eq!(full.len(), case.body.len(), "{context}: {label}: full GET length");
|
||||
assert!(full == case.body, "{context}: {label}: full GET body must equal the rc.5 upload");
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn assert_layout_readable(client: &Client, case: &LayoutCase, context: &str) -> TestResult {
|
||||
assert_layout_full_body(client, case, context).await?;
|
||||
let label = case.label();
|
||||
if case.is_multipart_layout() {
|
||||
let first = case.part_sizes[0];
|
||||
let range = format!("bytes={}-{}", first - 32, first + 31);
|
||||
@@ -2641,10 +2646,11 @@ async fn direct_upgrade_from_rc5_preserves_multipart_layouts() -> TestResult {
|
||||
"rc.5 wrote a legacy layout"
|
||||
);
|
||||
}
|
||||
// The rc.5 writer must itself still read what it wrote, so a later
|
||||
// failure is attributable to the upgrade rather than to the fixture.
|
||||
// Prove the rc.5 writer persisted every original byte before upgrading.
|
||||
// Its compressed cross-part range reader has a legacy limitation; the
|
||||
// upgraded reader must still satisfy every range and partNumber assertion.
|
||||
for case in &cases {
|
||||
assert_layout_readable(&old_client, case, "rc.5").await?;
|
||||
assert_layout_full_body(&old_client, case, "rc.5").await?;
|
||||
}
|
||||
|
||||
// Upgrade in place.
|
||||
|
||||
@@ -19169,6 +19169,326 @@ mod test {
|
||||
}]
|
||||
}
|
||||
|
||||
// This is a real process interruption of LocalDisk, not a server or
|
||||
// power-loss test. The child stops at the rollback-backup unlink.
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn quorum_delete_cleanup_process_crash_preserves_retry_evidence() {
|
||||
use crate::disk::os::prepared_publication_test_hooks as hooks;
|
||||
use std::os::unix::process::ExitStatusExt;
|
||||
|
||||
const CHILD_ROOT: &str = "RUSTFS_TEST_QUORUM_CLEANUP_ROOT";
|
||||
const CHILD_TRANSACTION: &str = "RUSTFS_TEST_QUORUM_CLEANUP_TRANSACTION";
|
||||
const CHILD_READY: &str = "RUSTFS_TEST_QUORUM_CLEANUP_READY";
|
||||
if let Some(root) = std::env::var_os(CHILD_ROOT) {
|
||||
let endpoint =
|
||||
Endpoint::try_from(Path::new(&root).to_str().expect("UTF-8 child disk path")).expect("parse child endpoint");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("open child disk");
|
||||
let transaction =
|
||||
Uuid::parse_str(&std::env::var(CHILD_TRANSACTION).expect("child transaction")).expect("parse child transaction");
|
||||
let ready = std::env::var_os(CHILD_READY).expect("child readiness path");
|
||||
let rollback_path = format!("object/{transaction}");
|
||||
let backup = disk
|
||||
.io_get_object_path("bucket", &rollback_path)
|
||||
.expect("child transaction path")
|
||||
.join(STORAGE_FORMAT_FILE_BACKUP);
|
||||
let _hook = hooks::install_at(hooks::Stage::Remove, &backup, move || {
|
||||
std::fs::write(ready, b"before-backup-unlink").expect("publish reached boundary");
|
||||
loop {
|
||||
std::thread::park();
|
||||
}
|
||||
});
|
||||
disk.delete(
|
||||
"bucket",
|
||||
&rollback_path,
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
old_data_dir: Some(transaction),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("child cleanup should reach its boundary");
|
||||
panic!("child unexpectedly returned without being killed");
|
||||
}
|
||||
|
||||
let dir = tempfile::tempdir().expect("create test disk");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
|
||||
ensure_test_volume(&disk, "bucket").await;
|
||||
let residue = write_committed_delete_residue(&disk, "bucket", "object").await;
|
||||
let original_backup = fs::read(residue.backup()).await.expect("read original backup");
|
||||
let original_part = fs::read(residue.part()).await.expect("read original part");
|
||||
|
||||
for attempt in 0..2 {
|
||||
let ready = dir.path().join(format!("cleanup-ready-{attempt}"));
|
||||
let mut child = std::process::Command::new(std::env::current_exe().expect("current test binary"))
|
||||
.args([
|
||||
"--exact",
|
||||
"disk::local::test::quorum_delete_cleanup_process_crash_preserves_retry_evidence",
|
||||
"--nocapture",
|
||||
])
|
||||
.env(CHILD_ROOT, dir.path())
|
||||
.env(CHILD_TRANSACTION, residue.transaction.to_string())
|
||||
.env(CHILD_READY, &ready)
|
||||
.spawn()
|
||||
.expect("spawn this test's cleanup process");
|
||||
let reached = tokio::time::timeout(Duration::from_secs(30), async {
|
||||
while !ready.exists() {
|
||||
if child.try_wait().expect("check child status").is_some() {
|
||||
return false;
|
||||
}
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
}
|
||||
true
|
||||
})
|
||||
.await;
|
||||
// Child is an owned process handle; no process-name matching or
|
||||
// external PID is accepted by this test.
|
||||
if child.try_wait().expect("check owned child").is_none() {
|
||||
child.kill().expect("SIGKILL the owned cleanup child");
|
||||
}
|
||||
let status = child.wait().expect("reap cleanup child");
|
||||
assert!(matches!(reached, Ok(true)), "child must reach the exact cleanup boundary");
|
||||
assert_eq!(status.signal(), Some(9), "child must have been interrupted by SIGKILL");
|
||||
println!(
|
||||
"cleanup crash {attempt}: backup={} marker={} part={}",
|
||||
residue.backup().exists(),
|
||||
residue.marker().exists(),
|
||||
residue.part().exists()
|
||||
);
|
||||
assert_eq!(fs::read(residue.backup()).await.expect("backup at paused unlink"), original_backup);
|
||||
assert_eq!(fs::read(residue.part()).await.expect("payload survives each crash"), original_part);
|
||||
assert!(residue.marker().is_file(), "transaction marker survives each crash");
|
||||
assert!(
|
||||
residue
|
||||
.object_dir
|
||||
.join(residue.transaction.to_string())
|
||||
.join(DELETE_QUORUM_COMMITTED_FILE)
|
||||
.is_file(),
|
||||
"quorum receipt survives each crash"
|
||||
);
|
||||
}
|
||||
|
||||
let reopened = LocalDisk::new(&endpoint, false)
|
||||
.await
|
||||
.expect("reopen disk after process interruption");
|
||||
reopened
|
||||
.delete(
|
||||
"bucket",
|
||||
&format!("object/{}", residue.transaction),
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
old_data_dir: Some(residue.transaction),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("retry the already committed coordinator transaction");
|
||||
assert_eq!(original_part, b"stale");
|
||||
let bucket_result = reopened.delete_volume("bucket", false).await;
|
||||
println!("non-force bucket deletion after retry: {bucket_result:?}");
|
||||
assert!(
|
||||
!residue.object_dir.exists(),
|
||||
"repeated delete after two crashes must reclaim this known-successful transaction"
|
||||
);
|
||||
bucket_result.expect("non-force bucket deletion after retry");
|
||||
}
|
||||
|
||||
// Normal post-quorum cleanup must preserve its data-dir marker until the
|
||||
// rollback backup is gone, including across an interrupted retry.
|
||||
#[cfg(not(windows))]
|
||||
#[tokio::test]
|
||||
async fn quorum_delete_cleanup_removes_backup_before_data_and_can_retry() {
|
||||
use crate::disk::os::prepared_publication_test_hooks as hooks;
|
||||
|
||||
let dir = tempfile::tempdir().expect("create test disk");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
|
||||
let bucket = "bucket";
|
||||
let object = "object";
|
||||
ensure_test_volume(&disk, bucket).await;
|
||||
let residue = write_committed_delete_residue(&disk, bucket, object).await;
|
||||
let data_path = residue.part();
|
||||
let rollback_path = format!("{object}/{}", residue.transaction);
|
||||
let opts = DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
old_data_dir: Some(residue.transaction),
|
||||
..Default::default()
|
||||
};
|
||||
let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
|
||||
let (release, release_rx) = std::sync::mpsc::channel::<()>();
|
||||
let _hook = hooks::install_at(hooks::Stage::Remove, &data_path, move || {
|
||||
let _ = entered_tx.send(());
|
||||
let _ = release_rx.recv();
|
||||
});
|
||||
let mut cleanup = Box::pin(disk.delete(bucket, &rollback_path, opts.clone()));
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
tokio::select! {
|
||||
entered = entered_rx => entered.expect("cleanup must reach data removal"),
|
||||
result = cleanup.as_mut() => panic!("cleanup ended before data removal: {result:?}"),
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("cleanup reaches physical payload unlink");
|
||||
|
||||
let backup_gone = !residue.backup().exists();
|
||||
let marker_present = residue.marker().exists();
|
||||
let original_part = fs::read(residue.part()).await.expect("paused data must remain readable");
|
||||
drop(cleanup);
|
||||
|
||||
// The in-flight unlink retains metadata ownership after cancellation.
|
||||
// Retry must wait for that physical operation before removing proof.
|
||||
let mut retry = Box::pin(disk.delete(bucket, &rollback_path, opts));
|
||||
assert!(
|
||||
tokio::time::timeout(Duration::from_millis(100), retry.as_mut())
|
||||
.await
|
||||
.is_err(),
|
||||
"retry cannot bypass the parked cleanup's namespace ownership"
|
||||
);
|
||||
assert!(residue.marker().is_file());
|
||||
assert!(
|
||||
residue
|
||||
.object_dir
|
||||
.join(residue.transaction.to_string())
|
||||
.join(DELETE_QUORUM_COMMITTED_FILE)
|
||||
.is_file()
|
||||
);
|
||||
drop(release);
|
||||
tokio::time::timeout(Duration::from_secs(10), retry)
|
||||
.await
|
||||
.expect("retry completes after the abandoned unlink releases ownership")
|
||||
.expect("retry cleanup");
|
||||
|
||||
assert!(
|
||||
backup_gone,
|
||||
"post-quorum cleanup must remove the backup before touching its data-dir evidence"
|
||||
);
|
||||
assert!(marker_present, "data removal must still have its transaction marker");
|
||||
assert_eq!(original_part, b"stale");
|
||||
assert!(!residue.object_dir.exists(), "interrupted cleanup and retry must leave no object residue");
|
||||
}
|
||||
|
||||
#[cfg(not(windows))]
|
||||
#[tokio::test]
|
||||
async fn quorum_delete_cleanup_backup_error_preserves_data_and_marker() {
|
||||
use crate::disk::os::prepared_publication_test_hooks as hooks;
|
||||
|
||||
let dir = tempfile::tempdir().expect("create test disk");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
|
||||
let bucket = "bucket";
|
||||
let object = "object";
|
||||
ensure_test_volume(&disk, bucket).await;
|
||||
let residue = write_committed_delete_residue(&disk, bucket, object).await;
|
||||
let backup_bytes = fs::read(residue.backup()).await.expect("read rollback evidence");
|
||||
let backup = residue.backup();
|
||||
let held_backup = residue.object_dir.join("held-backup");
|
||||
let blocked_backup = backup.clone();
|
||||
let retained_backup = held_backup.clone();
|
||||
// Replace only the classified backup inode with a directory so the
|
||||
// real unlink fails, while retaining its exact bytes for retry.
|
||||
let _hook = hooks::install_at(hooks::Stage::Remove, &backup, move || {
|
||||
std::fs::rename(&blocked_backup, retained_backup).expect("retain rollback bytes");
|
||||
std::fs::create_dir(blocked_backup).expect("block the actual backup unlink");
|
||||
});
|
||||
let rollback_path = format!("{object}/{}", residue.transaction);
|
||||
let opts = DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
old_data_dir: Some(residue.transaction),
|
||||
..Default::default()
|
||||
};
|
||||
disk.delete(bucket, &rollback_path, opts.clone())
|
||||
.await
|
||||
.expect_err("backup removal failure must be returned");
|
||||
|
||||
assert!(backup.is_dir(), "the native backup unlink must have encountered the blocker");
|
||||
assert_eq!(fs::read(&held_backup).await.expect("backup bytes must survive"), backup_bytes);
|
||||
assert!(residue.marker().exists(), "backup failure must retain the data-dir transaction marker");
|
||||
assert_eq!(fs::read(residue.part()).await.expect("payload must survive backup failure"), b"stale");
|
||||
|
||||
fs::remove_dir(&backup).await.expect("remove the one-shot unlink blocker");
|
||||
fs::rename(&held_backup, &backup)
|
||||
.await
|
||||
.expect("restore the unchanged backup inode");
|
||||
assert_eq!(fs::read(&backup).await.expect("canonical rollback backup"), backup_bytes);
|
||||
disk.delete(bucket, &rollback_path, opts)
|
||||
.await
|
||||
.expect("retry cleanup after one-shot unlink failure");
|
||||
assert!(!residue.object_dir.exists(), "successful retry must reclaim the transaction");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn quorum_delete_cleanup_preserves_other_generations_transactions_and_children() {
|
||||
let dir = tempfile::tempdir().expect("create test disk");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("UTF-8 disk path")).expect("parse endpoint");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("create local disk");
|
||||
let bucket = "bucket";
|
||||
let object = "object";
|
||||
ensure_test_volume(&disk, bucket).await;
|
||||
let residue = write_committed_delete_residue(&disk, bucket, object).await;
|
||||
let newer_data_dir = Uuid::new_v4();
|
||||
let newer_data = residue.object_dir.join(newer_data_dir.to_string());
|
||||
fs::create_dir_all(&newer_data).await.expect("create newer data directory");
|
||||
fs::write(newer_data.join("part.1"), b"newer payload")
|
||||
.await
|
||||
.expect("write newer payload");
|
||||
let metadata = test_meta(test_file_info(object, Uuid::new_v4(), Some(newer_data_dir), None));
|
||||
fs::write(residue.object_dir.join(STORAGE_FORMAT_FILE), &metadata)
|
||||
.await
|
||||
.expect("publish newer metadata before old transaction cleanup");
|
||||
let unrelated = Uuid::new_v4();
|
||||
let unrelated_backup = residue
|
||||
.object_dir
|
||||
.join(unrelated.to_string())
|
||||
.join(STORAGE_FORMAT_FILE_BACKUP);
|
||||
fs::create_dir_all(unrelated_backup.parent().expect("backup parent"))
|
||||
.await
|
||||
.expect("create unrelated transaction");
|
||||
fs::write(&unrelated_backup, b"corrupt unproven backup")
|
||||
.await
|
||||
.expect("write ambiguous backup");
|
||||
let mismatched_marker = newer_data.join(format!("{RESERVED_DELETE_DATA_DIR_MARKER_PREFIX}{unrelated}"));
|
||||
fs::write(&mismatched_marker, []).await.expect("write unrelated marker");
|
||||
let child = residue.object_dir.join("child").join(STORAGE_FORMAT_FILE);
|
||||
fs::create_dir_all(child.parent().expect("child parent"))
|
||||
.await
|
||||
.expect("create descendant");
|
||||
fs::write(&child, b"child metadata").await.expect("write descendant metadata");
|
||||
|
||||
disk.delete(
|
||||
bucket,
|
||||
&format!("{object}/{}", residue.transaction),
|
||||
DeleteOptions {
|
||||
recursive: true,
|
||||
immediate: true,
|
||||
old_data_dir: Some(residue.transaction),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await
|
||||
.expect("clean only the successful transaction");
|
||||
|
||||
assert!(!residue.backup().exists() && !residue.part().exists() && !residue.marker().exists());
|
||||
assert_eq!(
|
||||
fs::read(residue.object_dir.join(STORAGE_FORMAT_FILE))
|
||||
.await
|
||||
.expect("read newer metadata"),
|
||||
metadata
|
||||
);
|
||||
assert_eq!(fs::read(newer_data.join("part.1")).await.expect("read newer payload"), b"newer payload");
|
||||
assert_eq!(
|
||||
fs::read(&unrelated_backup).await.expect("read unproven backup"),
|
||||
b"corrupt unproven backup"
|
||||
);
|
||||
assert!(mismatched_marker.exists());
|
||||
assert_eq!(fs::read(child).await.expect("read descendant"), b"child metadata");
|
||||
}
|
||||
|
||||
// Internal maintenance retains its legacy cleanup behavior while user
|
||||
// buckets require the separate set-wide authorization boundary.
|
||||
#[tokio::test]
|
||||
|
||||
@@ -4734,7 +4734,9 @@ impl Hash for GetObjectMetadataCacheKey {
|
||||
#[derive(Clone, Copy, Debug, PartialEq, Eq)]
|
||||
enum GetObjectMetadataCacheCapability {
|
||||
Full,
|
||||
// Quorum-validated metadata without a complete body or write-target layout.
|
||||
// A quorum-selected layout can serve body reads, but is not a mutation layout.
|
||||
QuorumSelected,
|
||||
// Quorum-validated metadata used only to evaluate a terminal read condition.
|
||||
ConditionOnly,
|
||||
}
|
||||
|
||||
|
||||
@@ -23931,7 +23931,7 @@ mod delete_objects_lock_gating_tests {
|
||||
}
|
||||
put_plain_object(&set_disks, bucket, "obj-a").await;
|
||||
|
||||
let _outer_guard = set_disks
|
||||
let outer_guard = set_disks
|
||||
.new_ns_lock(bucket, "obj-a")
|
||||
.await
|
||||
.expect("namespace lock should be created")
|
||||
@@ -23984,6 +23984,15 @@ mod delete_objects_lock_gating_tests {
|
||||
|
||||
assert!(errs[0].is_none(), "no_lock batch delete should not fail with a lock error: {:?}", errs[0]);
|
||||
assert!(deleted[0].found, "existing object must still be deleted");
|
||||
for disk in &disk_stores {
|
||||
assert_eq!(
|
||||
disk.read_version("", bucket, "obj-a", "", &crate::disk::ReadOptions::default())
|
||||
.await
|
||||
.expect_err("every drive must confirm the deleted version is absent"),
|
||||
DiskError::FileNotFound
|
||||
);
|
||||
}
|
||||
drop(outer_guard);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -24596,6 +24605,88 @@ mod single_delete_namespace_owner_tests {
|
||||
.expect("seed a complete real object version");
|
||||
}
|
||||
|
||||
#[cfg(not(windows))]
|
||||
#[tokio::test]
|
||||
async fn quorum_delete_cleanup_coordinator_binds_single_and_batch() {
|
||||
use crate::disk::os::prepared_publication_test_hooks as hooks;
|
||||
|
||||
for batch in [false, true] {
|
||||
let (_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
|
||||
let bucket = "quorum-cleanup-order";
|
||||
let object = "object";
|
||||
let version = Uuid::new_v4();
|
||||
for disk in &disks {
|
||||
disk.make_volume(bucket).await.expect("create fixture bucket");
|
||||
}
|
||||
seed_version(&set, bucket, object, version, &vec![0x51; 256 * 1024]).await;
|
||||
let (observed, observations) = std::sync::mpsc::channel();
|
||||
let mut hooks = Vec::new();
|
||||
for disk in &disks {
|
||||
let fi = disk
|
||||
.read_version("", bucket, object, &version.to_string(), &ReadOptions::default())
|
||||
.await
|
||||
.expect("read seeded version");
|
||||
assert!(!fi.inline_data(), "fixture must reach physical data cleanup");
|
||||
let crate::disk::Disk::Local(local) = disk.as_ref() else {
|
||||
panic!("local fixture required");
|
||||
};
|
||||
let object_dir = local.get_disk().get_object_path_for_io(bucket, object).expect("object path");
|
||||
let data_dir = object_dir.join(fi.data_dir.expect("seeded data directory").to_string());
|
||||
let observed = observed.clone();
|
||||
hooks.push(hooks::install_at(hooks::Stage::Remove, &data_dir.join("part.1"), move || {
|
||||
let backup_present = std::fs::read_dir(&object_dir)
|
||||
.expect("object exists before payload unlink")
|
||||
.any(|entry| {
|
||||
entry
|
||||
.expect("read transaction entry")
|
||||
.path()
|
||||
.join(crate::disk::STORAGE_FORMAT_FILE_BACKUP)
|
||||
.exists()
|
||||
});
|
||||
observed.send(backup_present).expect("record actual cleanup order");
|
||||
}));
|
||||
}
|
||||
if batch {
|
||||
let (_, errors) = set
|
||||
.delete_objects(
|
||||
bucket,
|
||||
vec![ObjectToDelete {
|
||||
object_name: object.to_string(),
|
||||
version_id: Some(version),
|
||||
..Default::default()
|
||||
}],
|
||||
ObjectOptions {
|
||||
object_lock_config_snapshot: Some(Arc::new(ObjectLockConfigSnapshot::new(
|
||||
ObjectLockConfigState::ConfirmedAbsent,
|
||||
))),
|
||||
..Default::default()
|
||||
},
|
||||
)
|
||||
.await;
|
||||
assert!(errors.iter().all(Option::is_none), "batch delete must commit: {errors:?}");
|
||||
} else {
|
||||
set.delete_object_version(
|
||||
bucket,
|
||||
object,
|
||||
&FileInfo {
|
||||
name: object.to_string(),
|
||||
version_id: Some(version),
|
||||
..Default::default()
|
||||
},
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.expect("single delete must commit");
|
||||
}
|
||||
let observations = observations.try_iter().collect::<Vec<_>>();
|
||||
assert_eq!(observations.len(), 4, "all four disks must reach cleanup: batch={batch}");
|
||||
assert!(
|
||||
observations.iter().all(|backup_present| !backup_present),
|
||||
"the coordinator must bind post-quorum cleanup before any payload unlink: batch={batch}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial_test::serial]
|
||||
async fn internal_metadata_mutations_do_not_reenter_pool_metadata() {
|
||||
|
||||
@@ -472,14 +472,14 @@ impl SetDisks {
|
||||
entry: Arc<GetObjectMetadataCacheEntry>,
|
||||
after_insert: impl FnOnce(),
|
||||
) {
|
||||
// Every publisher shares Moka's per-key compute lock. A conditional
|
||||
// snapshot cannot downgrade a full layout or extend its original TTL.
|
||||
// Every publisher shares Moka's per-key compute lock. A partial snapshot
|
||||
// cannot downgrade a full layout or extend its original TTL.
|
||||
self.get_object_metadata_cache
|
||||
.entry(key.clone())
|
||||
.and_compute_with(|current| {
|
||||
std::future::ready(
|
||||
if !self.is_get_object_metadata_cache_generation_current(generation)
|
||||
|| (entry.capability == GetObjectMetadataCacheCapability::ConditionOnly
|
||||
|| (entry.capability != GetObjectMetadataCacheCapability::Full
|
||||
&& current
|
||||
.as_ref()
|
||||
.is_some_and(|existing| existing.value().capability == GetObjectMetadataCacheCapability::Full))
|
||||
@@ -632,7 +632,11 @@ impl SetDisks {
|
||||
match self.lookup_cached_get_object_fileinfo(bucket, object).await {
|
||||
MetadataCacheLookup::Hit(cached)
|
||||
if cached.capability == GetObjectMetadataCacheCapability::Full
|
||||
|| (cached.capability == GetObjectMetadataCacheCapability::QuorumSelected
|
||||
&& read_data
|
||||
&& allow_early_stop)
|
||||
|| (purpose == MetadataReadPurpose::GetObjectReader
|
||||
&& cached.capability == GetObjectMetadataCacheCapability::ConditionOnly
|
||||
&& allow_early_stop
|
||||
&& crate::object_api::get_object_read_condition_is_terminal(
|
||||
crate::object_api::GetObjectReadMetadata::from_file_info(bucket, object, &cached.fi),
|
||||
@@ -828,8 +832,10 @@ impl SetDisks {
|
||||
&fi,
|
||||
if metadata_fanout_complete {
|
||||
GetObjectMetadataCacheCapability::Full
|
||||
} else {
|
||||
} else if condition_stopped {
|
||||
GetObjectMetadataCacheCapability::ConditionOnly
|
||||
} else {
|
||||
GetObjectMetadataCacheCapability::QuorumSelected
|
||||
},
|
||||
),
|
||||
&parts_metadata,
|
||||
@@ -4130,6 +4136,7 @@ mod metadata_cache_tests {
|
||||
.await
|
||||
.expect("test object should be written before enabling the slow-tail fault");
|
||||
|
||||
let calls = disk_call_counters::observe(object);
|
||||
temp_env::async_with_vars(
|
||||
[
|
||||
("RUSTFS_GET_METADATA_EARLY_STOP_ENABLE", Some("true")),
|
||||
@@ -4142,7 +4149,6 @@ mod metadata_cache_tests {
|
||||
("RUSTFS_GET_METADATA_SLOWTAIL_FAULT_OBJECT_PREFIX", Some("objects/")),
|
||||
],
|
||||
async {
|
||||
let calls = disk_call_counters::observe(object);
|
||||
let first = tokio::time::timeout(
|
||||
Duration::from_millis(400),
|
||||
set.get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, true),
|
||||
@@ -4154,49 +4160,32 @@ mod metadata_cache_tests {
|
||||
assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 4);
|
||||
assert_eq!(calls.total(disk_call_counters::KIND_METADATA_SLOWTAIL_FAULT), 1);
|
||||
|
||||
let cached = set
|
||||
.cached_get_object_fileinfo(bucket, object)
|
||||
let second = set
|
||||
.get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, true)
|
||||
.await
|
||||
.expect("early-stop quorum must publish a conditional cache entry");
|
||||
assert_eq!(cached.capability, GetObjectMetadataCacheCapability::ConditionOnly);
|
||||
assert_eq!(cached.fi.name, object);
|
||||
assert!(cached.online_disks.iter().filter(|disk| disk.is_some()).count() >= cached.read_quorum);
|
||||
|
||||
let condition = Arc::new(CacheBoundaryCondition {
|
||||
bucket,
|
||||
object,
|
||||
etag: cached.fi.metadata["etag"].clone(),
|
||||
calls: AtomicUsize::new(0),
|
||||
terminal_calls: usize::MAX,
|
||||
});
|
||||
let second = crate::object_api::with_get_object_read_condition(
|
||||
condition,
|
||||
set.get_object_fileinfo_for_get_object_reader(bucket, object, &ObjectOptions::default(), true, true),
|
||||
)
|
||||
.await
|
||||
.expect("terminal conditional GET should use the quorum-published cache entry");
|
||||
assert!(second.shared_entry().is_some(), "terminal conditional GET should hit the published entry");
|
||||
.expect("second lookup should use the quorum-published metadata cache entry");
|
||||
assert!(second.shared_entry().is_some(), "the repeated GET should hit the published cache entry");
|
||||
assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 4);
|
||||
|
||||
let full = set
|
||||
.get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false)
|
||||
.await
|
||||
.expect("body-layout lookup must complete its own metadata fanout");
|
||||
assert!(full.owned.is_some(), "a conditional snapshot cannot serve a full body layout");
|
||||
assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 8);
|
||||
|
||||
let repeated = set
|
||||
.get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false)
|
||||
.await
|
||||
.expect("complete metadata should publish a reusable body layout");
|
||||
let cached = repeated
|
||||
.shared_entry()
|
||||
.expect("the repeated body GET must use the complete cached layout");
|
||||
assert_eq!(cached.capability, GetObjectMetadataCacheCapability::Full);
|
||||
assert_eq!(calls.total(disk_call_counters::KIND_READ_VERSION), 8);
|
||||
let cached = second.shared_entry().expect("second lookup should share cached metadata");
|
||||
assert_eq!(cached.fi.name, object);
|
||||
assert!(cached.online_disks.iter().filter(|disk| disk.is_some()).count() >= cached.read_quorum);
|
||||
},
|
||||
)
|
||||
.await;
|
||||
|
||||
let full_layout = set
|
||||
.get_object_fileinfo_gated(bucket, object, &ObjectOptions::default(), false, false)
|
||||
.await
|
||||
.expect("read-before-write consumers must fetch a complete layout");
|
||||
assert!(
|
||||
full_layout.shared_entry().is_none(),
|
||||
"a quorum-selected partial layout cannot satisfy mutation consumers"
|
||||
);
|
||||
assert!(
|
||||
calls.total(disk_call_counters::KIND_READ_VERSION) > 4,
|
||||
"mutation consumers must independently fan out after a quorum-only cache hit"
|
||||
);
|
||||
}
|
||||
|
||||
#[tokio::test(flavor = "multi_thread", worker_threads = 2)]
|
||||
|
||||
@@ -9594,8 +9594,18 @@ mod tests {
|
||||
#[cfg(feature = "test-util")]
|
||||
async fn decommission_migrates_and_verifies_registered_durable_ilm_records_scenario() {
|
||||
let temp_dir = tempfile::tempdir().expect("create temp store dir");
|
||||
let (ctx, store, _shutdown) =
|
||||
without_storage_class_env(build_isolated_test_store(temp_dir.path(), "durable-ilm-decommission", &[4, 4])).await;
|
||||
// The explicit recovery below must be the only journal consumer,
|
||||
// including when receipt validation outlasts the background interval.
|
||||
let mut instance_ctx = crate::runtime::instance::InstanceContext::new();
|
||||
instance_ctx.suppress_tier_delete_journal_recovery_for_test();
|
||||
let (ctx, store, _shutdown) = without_storage_class_env(build_isolated_test_store_with_layout(
|
||||
temp_dir.path(),
|
||||
"durable-ilm-decommission",
|
||||
&[(1, 4), (1, 4)],
|
||||
CancellationToken::new(),
|
||||
Some(Arc::new(instance_ctx)),
|
||||
))
|
||||
.await;
|
||||
|
||||
let tier_name = "DECOMMISSION-ILM";
|
||||
let backend = register_transition_reconcile_test_tier(&ctx.tier_config_mgr(), tier_name).await;
|
||||
@@ -10067,6 +10077,16 @@ mod tests {
|
||||
.await
|
||||
.expect("manual task receipt should restore after the corruption check");
|
||||
|
||||
assert!(matches!(
|
||||
com::read_config(store.pools[0].clone(), &tier_path).await,
|
||||
Err(Error::ConfigNotFound)
|
||||
));
|
||||
assert_eq!(
|
||||
com::read_config(store.pools[1].clone(), &tier_path)
|
||||
.await
|
||||
.expect("manual recovery must still observe the exact migrated tier journal"),
|
||||
tier_bytes
|
||||
);
|
||||
let tier_stats = recover_tier_delete_journal_entries(store.clone(), 100, None)
|
||||
.await
|
||||
.expect("tier journal recovery should consume the migrated record before completion");
|
||||
|
||||
Reference in New Issue
Block a user