diff --git a/crates/e2e_test/src/upgrade_compatibility_test.rs b/crates/e2e_test/src/upgrade_compatibility_test.rs index 51ff93d6f..35e0c8eaf 100644 --- a/crates/e2e_test/src/upgrade_compatibility_test.rs +++ b/crates/e2e_test/src/upgrade_compatibility_test.rs @@ -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. diff --git a/crates/ecstore/src/disk/local.rs b/crates/ecstore/src/disk/local.rs index cbd2a0227..ea3a39ff8 100644 --- a/crates/ecstore/src/disk/local.rs +++ b/crates/ecstore/src/disk/local.rs @@ -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] diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 733576508..fad556f55 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -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, } diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 87f713040..a13252d8e 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -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::>(); + 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() { diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index 115f1ead8..ae017204f 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -472,14 +472,14 @@ impl SetDisks { entry: Arc, 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)] diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 64656bcb4..86cf16be7 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -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");