diff --git a/crates/ecstore/src/set_disk/ops/heal.rs b/crates/ecstore/src/set_disk/ops/heal.rs index 11df66bc2..3d4624b4e 100644 --- a/crates/ecstore/src/set_disk/ops/heal.rs +++ b/crates/ecstore/src/set_disk/ops/heal.rs @@ -2812,9 +2812,20 @@ mod heal_result_report_tests { } let mut reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]); - set.put_object(&bucket, object, &mut reader, &ObjectOptions::default()) - .await - .expect("source object should be written"); + // This fixture reads and removes physical shards immediately after + // PUT. A lock-owning PUT may quorum-ack before its rename tail + // drains, so keep the setup on the full-fanout commit path. + set.put_object( + &bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("source object should be written"); let source = disks[2] .read_version("", &bucket, object, "", &ReadOptions::default()) .await @@ -3230,9 +3241,20 @@ mod heal_result_report_tests { } let mut reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]); - set.put_object(bucket, object, &mut reader, &ObjectOptions::default()) - .await - .expect("source object should be written"); + // The target-evidence readback below asserts per-disk state right + // after PUT. A lock-owning PUT may quorum-ack before its rename tail + // drains, so keep the setup on the full-fanout commit path. + set.put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("source object should be written"); let source = disks[2] .read_version("", bucket, object, "", &ReadOptions::default()) .await @@ -3279,6 +3301,9 @@ mod heal_result_report_tests { .await .expect("versioned bucket should be created"); + // The per-version target-evidence readback below asserts per-disk + // state right after PUT. A lock-owning PUT may quorum-ack before its + // rename tail drains, so keep the setup on the full-fanout commit path. let mut old_reader = PutObjReader::from_vec(vec![0x5a; 1024 * 1024]); let old_info = set .put_object( @@ -3287,6 +3312,7 @@ mod heal_result_report_tests { &mut old_reader, &ObjectOptions { versioned: true, + no_lock: true, ..Default::default() }, ) @@ -3304,6 +3330,7 @@ mod heal_result_report_tests { &mut latest_reader, &ObjectOptions { versioned: true, + no_lock: true, ..Default::default() }, ) @@ -4325,9 +4352,20 @@ mod heal_result_report_tests { const PAYLOAD_SIZE: usize = 1024 * 1024; let mut initial_reader = PutObjReader::from_vec(vec![0x11; PAYLOAD_SIZE]); - set.put_object(bucket, object, &mut initial_reader, &ObjectOptions::default()) - .await - .expect("initial object should be written"); + // This fixture reads and removes physical shards immediately after + // PUT. A lock-owning PUT may quorum-ack before its rename tail drains, + // so keep the setup on the full-fanout commit path. + set.put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + .expect("initial object should be written"); let current = disks[2] .read_version("", bucket, object, "", &ReadOptions::default()) @@ -4415,11 +4453,12 @@ mod heal_result_report_tests { // Give the heal something to rebuild on alternating rounds: remove a // shard of the current data dir right before the race. if round % 2 == 1 { - let current = disks[2] - .read_version("", bucket, object, "", &ReadOptions::default()) - .await - .expect("current metadata should be readable"); - if let Some(data_dir) = current.data_dir { + // The previous round's lock-owning PUT may still be + // draining its rename tail on this disk; shard damage is + // best-effort here, so skip injection when it lags. + if let Ok(current) = disks[2].read_version("", bucket, object, "", &ReadOptions::default()).await + && let Some(data_dir) = current.data_dir + { let shard = temp_dirs[3] .path() .join(bucket) diff --git a/crates/ecstore/src/set_disk/ops/multipart.rs b/crates/ecstore/src/set_disk/ops/multipart.rs index 5ff71ad11..96883d31d 100644 --- a/crates/ecstore/src/set_disk/ops/multipart.rs +++ b/crates/ecstore/src/set_disk/ops/multipart.rs @@ -5459,19 +5459,42 @@ mod tests { disk.make_volume(bucket).await.expect("bucket volume should be created"); } let mut initial_reader = PutObjReader::from_vec(b"old multipart body".to_vec()); + // A lock-owning PUT may quorum-ack before its rename tail drains, and + // cache priming refuses to publish while a straggler disk still reads + // as an error; keep the setup on the full-fanout commit path. set_disks - .put_object(bucket, object, &mut initial_reader, &ObjectOptions::default()) + .put_object( + bucket, + object, + &mut initial_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("initial object should be written"); - set_disks - .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) - .await - .expect("initial metadata should resolve"); - let generation = set_disks - .get_object_metadata_cache_generation(bucket, object) - .expect("metadata cache generation should be active"); - let retired_key = GetObjectMetadataCacheKey::new(bucket, object, generation); - assert!(set_disks.get_object_metadata_cache.get(&retired_key).await.is_some()); + // The publish is also bounded by the cache TTL, so re-prime until the + // current generation is observably cached instead of asserting on a + // single read that a loaded host can stall past expiry. + let retired_key = tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + set_disks + .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) + .await + .expect("initial metadata should resolve"); + let generation = set_disks + .get_object_metadata_cache_generation(bucket, object) + .expect("metadata cache generation should be active"); + let key = GetObjectMetadataCacheKey::new(bucket, object, generation); + if set_disks.get_object_metadata_cache.get(&key).await.is_some() { + return key; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("metadata priming should publish the current generation"); let upload = set_disks .new_multipart_upload(bucket, object, &ObjectOptions::default()) diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index ffa18783e..6392e54a5 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -10223,8 +10223,19 @@ mod inline_put_commit_path_tests { make_bucket(&disk_stores, bucket).await; let mut reader = PutObjReader::from_vec(payload.clone()); + // This test asserts the committed inline shard on every individual + // disk. A lock-owning PUT may quorum-ack before its rename tail + // drains, so keep the setup on the full-fanout commit path. set_disks - .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("inline PUT should commit"); @@ -10724,8 +10735,19 @@ mod inline_put_commit_path_tests { make_bucket(&disk_stores, bucket).await; let mut reader = PutObjReader::from_vec(Vec::new()); + // This test asserts the committed layout on every individual disk. A + // lock-owning PUT may quorum-ack before its rename tail drains, so + // keep the setup on the full-fanout commit path. set_disks - .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("zero-length PUT should commit through the existing pipeline"); @@ -11060,22 +11082,42 @@ mod metadata_mutation_generation_tests { payload: &[u8], ) -> (ObjectInfo, GetObjectMetadataCacheKey) { let mut reader = PutObjReader::from_vec(payload.to_vec()); + // A lock-owning PUT may quorum-ack before its rename tail drains, and + // cache priming refuses to publish while a straggler disk still reads + // as an error; keep the setup on the full-fanout commit path. let info = set_disks - .put_object(bucket, object, &mut reader, &ObjectOptions::default()) + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("test object should be written"); - set_disks - .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) - .await - .expect("test object metadata should resolve"); - let generation = set_disks - .get_object_metadata_cache_generation(bucket, object) - .expect("metadata cache generation should be active"); - let key = GetObjectMetadataCacheKey::new(bucket, object, generation); - assert!( - set_disks.get_object_metadata_cache.get(&key).await.is_some(), - "metadata priming should publish the current generation" - ); + // The publish is also bounded by the cache TTL, so re-prime until the + // current generation is observably cached instead of asserting on a + // single read that a loaded host can stall past expiry. + let key = tokio::time::timeout(std::time::Duration::from_secs(30), async { + loop { + set_disks + .get_object_fileinfo(bucket, object, &ObjectOptions::default(), true, false) + .await + .expect("test object metadata should resolve"); + let generation = set_disks + .get_object_metadata_cache_generation(bucket, object) + .expect("metadata cache generation should be active"); + let key = GetObjectMetadataCacheKey::new(bucket, object, generation); + if set_disks.get_object_metadata_cache.get(&key).await.is_some() { + return key; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("metadata priming should publish the current generation"); (info, key) } @@ -15240,8 +15282,20 @@ mod heterogeneous_pool_put_tests { make_bucket(&disk_stores, bucket).await; let mut default_reader = PutObjReader::from_vec(b"default gate stays epoch-free".to_vec()); + // Both epoch readbacks below inspect every disk right after PUT. A + // lock-owning PUT may quorum-ack before its rename tail drains, so keep + // these commits on the full-fanout path; the fencing gate itself is + // driven by the fleet proof and env vars, never by the lock option. set_disks - .put_object(bucket, "default.bin", &mut default_reader, &ObjectOptions::default()) + .put_object( + bucket, + "default.bin", + &mut default_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("default PUT should commit"); assert_eq!( @@ -15258,7 +15312,15 @@ mod heterogeneous_pool_put_tests { async { let mut fenced_reader = PutObjReader::from_vec(b"fenced epoch commit".to_vec()); set_disks - .put_object(bucket, "fenced.bin", &mut fenced_reader, &ObjectOptions::default()) + .put_object( + bucket, + "fenced.bin", + &mut fenced_reader, + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) .await .expect("fenced PUT should commit with a live proof"); },