From 394394cdfce6f17162f323e9338887085005288b Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E5=94=90=E5=B0=8F=E9=B8=AD?= Date: Tue, 1 Sep 2026 20:12:43 +0800 Subject: [PATCH] test(ecstore): deflake early-ack PUT fixtures in set_disk ops (#7009) Six set_disk::ops tests failed non-deterministically only under concurrent full-suite load, rotating between runs while each passed in isolation. All six share one root cause: a lock-owning put_object quorum-acks once the rename fanout reaches write quorum and lets a detached tail task finish the lagging disks, so a fixture that inspects per-disk state immediately after PUT can observe a disk the tail has not reached yet. The two heal report fixtures, the inline-commit fixture, and the transaction-fencing fixture read or delete physical shards right after PUT, and hit FileNotFound on a lagging disk. The two metadata-cache fixtures prime the cache after PUT, and the read fanout refuses to publish a cache entry while any disk still reports an error, so the priming read observably published nothing. Keep every affected setup PUT on the full-fanout commit path with no_lock: true, following the existing precedent in this module, so PUT returns only after every disk has committed. The option only governs lock acquisition, so it does not weaken what any of these fixtures assert; the transaction-fencing gate in particular is driven by the fleet proof and env vars, never by the lock option. Where a fixture also depends on cache publication, re-prime until the current generation is observably cached instead of asserting on a single read that a loaded host can stall past the cache TTL. The heal race fixture's shard damage injection is best-effort by construction, so it now skips injection when the previous round's tail still lags rather than unwrapping a read that may legitimately race. No production code changes, and no retries or sleeps added. --- crates/ecstore/src/set_disk/ops/heal.rs | 67 +++++++++++--- crates/ecstore/src/set_disk/ops/multipart.rs | 43 +++++++-- crates/ecstore/src/set_disk/ops/object.rs | 96 ++++++++++++++++---- 3 files changed, 165 insertions(+), 41 deletions(-) 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"); },