From 413b880b238ad242d2872f9cd47208c2195f8ebf Mon Sep 17 00:00:00 2001 From: overtrue Date: Sat, 5 Sep 2026 09:24:29 +0800 Subject: [PATCH] fix(ecstore): drain durable control-plane write tails --- .../bucket/lifecycle/manual_transition_job.rs | 5 + .../bucket/lifecycle/tier_delete_journal.rs | 6 + .../lifecycle/transition_transaction.rs | 2 + crates/ecstore/src/core/pools.rs | 4 + crates/ecstore/src/object_api/types.rs | 16 + .../src/services/tier/tier_mutation_intent.rs | 2 + .../src/services/tier/tier_probe_intent.rs | 2 + crates/ecstore/src/set_disk/ops/object.rs | 283 +++++++++++++++++- .../src/set_disk/transition_matrix_tests.rs | 5 +- crates/ecstore/src/store/init.rs | 9 +- .../ecstore-validation-suite-design.md | 16 + 11 files changed, 337 insertions(+), 13 deletions(-) diff --git a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs index b480ba468..050228b25 100644 --- a/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs +++ b/crates/ecstore/src/bucket/lifecycle/manual_transition_job.rs @@ -1170,6 +1170,7 @@ pub async fn save_manual_transition_job_record_if_current( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(current_etag.to_string()), ..Default::default() @@ -1242,6 +1243,7 @@ pub(crate) async fn save_manual_transition_worker_result_if_absent( data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -1270,6 +1272,7 @@ pub(crate) async fn save_manual_transition_task_if_absent( data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -1621,6 +1624,7 @@ pub async fn save_manual_transition_scope_admission_if_absent( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -1672,6 +1676,7 @@ pub async fn save_manual_transition_scope_admission_if_current( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(current_etag.to_string()), ..Default::default() diff --git a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs index 3a6a7e451..2270bce3b 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_delete_journal.rs @@ -1733,6 +1733,7 @@ async fn save_config_if_none_fenced( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -1832,6 +1833,7 @@ async fn save_decommission_manifest_checkpoint_if_match( let mut opts = ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, no_lock: true, http_preconditions: Some(HTTPPreconditions { if_match: Some(observed_etag), @@ -1960,6 +1962,7 @@ async fn save_config_if_match_fenced( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(etag.to_string()), ..Default::default() @@ -3780,6 +3783,7 @@ where data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -3869,6 +3873,7 @@ where data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(etag), ..Default::default() @@ -3893,6 +3898,7 @@ where data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() diff --git a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs index 87df9fe0d..82e32f598 100644 --- a/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs +++ b/crates/ecstore/src/bucket/lifecycle/transition_transaction.rs @@ -612,6 +612,7 @@ pub(crate) async fn save_transition_transaction_record( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -658,6 +659,7 @@ pub(crate) async fn save_transition_transaction_record_if_current( data.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(etag), ..Default::default() diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 4973f6831..94f3a6344 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -5493,6 +5493,7 @@ where fence.ensure_held()?; let mut opts = ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, no_lock: true, http_preconditions: Some(pool_meta_cas_preconditions(token, object)?), ..Default::default() @@ -14412,6 +14413,7 @@ impl ECStore { encoded.clone(), &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -14566,6 +14568,7 @@ impl ECStore { encoded, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(http_preconditions), ..Default::default() }, @@ -14957,6 +14960,7 @@ impl ECStore { encoded, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(etag), ..Default::default() diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 85e270679..701596cbf 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -870,6 +870,18 @@ impl TierFreeVersionReceiptSink { } } +/// Internal PUT completion boundary; this does not change fsync or write quorum. +#[doc(hidden)] +#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)] +pub enum WriteCompletion { + /// Return at write quorum when the commit owner can retain its guards. + #[default] + Quorum, + /// Drain the rename fan-out before returning. Minority failures still heal + /// after a successful quorum commit; this does not require every disk to succeed. + TailDrained, +} + #[derive(Default, Clone)] pub struct ObjectOptions { // Use the maximum parity (N/2), used when saving server configuration files @@ -896,6 +908,10 @@ pub struct ObjectOptions { /// Persisted bucket incarnation observed before authorization. pub expected_bucket_incarnation_id: Option, pub no_lock: bool, + /// Control-plane writers that immediately read or CAS the same namespace + /// key use TailDrained without changing namespace lock ownership. + #[doc(hidden)] + pub write_completion: WriteCompletion, /// True when an upper layer already holds the object read lock before /// forwarding a no_lock read to the set layer. pub metadata_cache_safe: bool, diff --git a/crates/ecstore/src/services/tier/tier_mutation_intent.rs b/crates/ecstore/src/services/tier/tier_mutation_intent.rs index 288b02969..eef2dcf7a 100644 --- a/crates/ecstore/src/services/tier/tier_mutation_intent.rs +++ b/crates/ecstore/src/services/tier/tier_mutation_intent.rs @@ -460,6 +460,7 @@ where data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -556,6 +557,7 @@ where data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(current_etag.to_string()), ..Default::default() diff --git a/crates/ecstore/src/services/tier/tier_probe_intent.rs b/crates/ecstore/src/services/tier/tier_probe_intent.rs index 3d11402a5..b3d96dc05 100644 --- a/crates/ecstore/src/services/tier/tier_probe_intent.rs +++ b/crates/ecstore/src/services/tier/tier_probe_intent.rs @@ -494,6 +494,7 @@ where data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_none_match: Some("*".to_string()), ..Default::default() @@ -549,6 +550,7 @@ where data, &ObjectOptions { max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, http_preconditions: Some(HTTPPreconditions { if_match: Some(current.record_etag.clone()), ..Default::default() diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 5a220e432..8db359d3f 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -299,7 +299,7 @@ use crate::error::is_err_invalid_upload_id; use crate::object_api::{GetObjectBodySource, get_object_body_cache_hook_suppressed}; use crate::object_api::{ NamespaceLockFence, ReplicationStatusWritebackCondition, ReplicationStatusWritebackMode, - SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, + SCANNER_PUBLICATION_LEASE_FENCE_METADATA_KEY, WriteCompletion, }; use crate::services::notification_sys::RemoteVersionStateFleetProofToken; use crate::services::tier::tier::{TierConfigMgr, TierDestinationId, TierOperationLease, tier_destination_id_from_metadata}; @@ -4266,13 +4266,17 @@ impl SetDisks { // complete rename fan-out drains. Keep this path synchronous so // its terminal state is known before the coordinator releases // remote leases. - let commit_allows_early_ack = !(opts.data_movement && opts.has_decommission_capacity_reservation()) - && (commit_object_lock_guard.is_some() - || commit_decommission_object_lock_guard.is_some() - || commit_publication_guard.is_some()) + let commit_owns_namespace_guard = commit_object_lock_guard.is_some() + || commit_decommission_object_lock_guard.is_some() + || commit_publication_guard.is_some(); + let commit_allows_early_ack = opts.write_completion == WriteCompletion::Quorum + && !(opts.data_movement && opts.has_decommission_capacity_reservation()) + && commit_owns_namespace_guard && commit_scanner_publication_scope.is_none(); + // Full-tail callers also transfer owned guards to the coordinator: + // cancelling their ACK waiter must not cancel an in-flight rename. let detach_commit_owner = commit_scanner_publication_scope.is_some() - || commit_allows_early_ack + || commit_owns_namespace_guard || commit_bucket_lifecycle_guard.is_some() || quota_mutation_fence; let commit_write_path_label = write_path.metric_label(); @@ -4617,9 +4621,8 @@ impl SetDisks { request.object_version_id = committed_version_id .or_else(|| commit_version_suspended.then(Uuid::nil)) .map(|version_id| version_id.to_string()); - tokio::spawn(async move { - let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await; - }); + let heal_set = commit_set.clone(); + tokio::spawn(async move { heal_set.submit_rename_tail_heal(request).await }); } let rename_stage_elapsed = rename_stage_start.elapsed(); @@ -18157,6 +18160,256 @@ mod put_object_tmp_cleanup_tests { .await; } + async fn make_completion_test_bucket(disks: &[DiskStore], bucket: &str) { + for disk in disks { + disk.make_volume(bucket) + .await + .expect("completion test bucket should be created"); + } + } + + /// Observe the actual metadata quorum while the remaining rename is parked. + /// A completed task count alone can race tasks that have not started yet. + async fn wait_for_paused_tail_metadata_quorum(disks: &[DiskStore], bucket: &str, object: &str) { + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let mut committed = 0; + for disk in disks { + match disk.read_version("", bucket, object, "", &ReadOptions::default()).await { + Ok(_) => committed += 1, + Err(DiskError::FileNotFound | DiskError::FileVersionNotFound) => {} + Err(err) => panic!("unexpected metadata error while observing {bucket}/{object}: {err}"), + } + } + if committed == 3 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("three disks must publish metadata while the fourth rename remains paused"); + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn tail_drained_put_waits_for_tail_and_allows_immediate_cas() { + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + for size in [4096, 1024 * 1024] { + let (_dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = "put-full-tail-cas"; + let object = "full-tail-cas-object"; + make_completion_test_bucket(&disks, bucket).await; + let tasks = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let writer = Arc::clone(&set); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; size]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("full-tail PUT must reach the rename barrier"); + wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await; + assert!(!put.is_finished(), "full-tail PUT must remain pending after metadata quorum"); + let mut lock_probe = Box::pin(set.acquire_write_lock_diag("full_tail_probe", bucket, object)); + assert!( + futures::poll!(lock_probe.as_mut()).is_pending(), + "the owned namespace guard must remain held" + ); + barrier.release(); + let written = tokio::time::timeout(Duration::from_secs(30), put) + .await + .expect("full-tail PUT should finish after release") + .expect("full-tail PUT task should join") + .expect("full-tail PUT must commit"); + assert_eq!(tasks.running(), 0, "full-tail response must follow every rename task"); + drop( + tokio::time::timeout(Duration::from_secs(5), lock_probe) + .await + .expect("same-key lock should be available on return") + .expect("same-key lock probe should succeed"), + ); + for disk in &disks { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("successful full-tail PUT must publish on every healthy disk"); + } + drop(barrier); + let mut replacement = PutObjReader::from_vec(b"cas successor".to_vec()); + set.put_object( + bucket, + object, + &mut replacement, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + http_preconditions: Some(HTTPPreconditions { + if_match: written.etag, + ..Default::default() + }), + ..Default::default() + }, + ) + .await + .expect("immediate same-key CAS must acquire the namespace guard"); + let mut read = set + .get_object_reader(bucket, object, None, HeaderMap::new(), &ObjectOptions::default()) + .await + .expect("CAS successor must be immediately readable"); + let mut body = Vec::new(); + read.stream.read_to_end(&mut body).await.expect("successor body must drain"); + assert_eq!(body, b"cas successor"); + } + }) + .await; + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn tail_drained_put_preserves_quorum_success_and_heals_failed_tail() { + let (_dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = "put-full-tail-heal"; + let object = "full-tail-heal-object"; + make_completion_test_bucket(&disks, bucket).await; + let mut heals = set.capture_test_rename_tail_heals(); + let tasks = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let _fault = rename_fault_injection::fail_rename_on(object, &[0]); + let writer = Arc::clone(&set); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("failed tail must first reach the rename barrier"); + wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await; + assert!(!put.is_finished(), "committed quorum must still wait for the failing tail"); + barrier.release(); + tokio::time::timeout(Duration::from_secs(30), put) + .await + .expect("failed tail should drain") + .expect("PUT task should join") + .expect("a minority tail error must not negate committed quorum"); + assert_eq!(tasks.running(), 0); + let heal = tokio::time::timeout(Duration::from_secs(30), heals.recv()) + .await + .expect("failed tail must schedule heal") + .expect("heal capture must remain connected"); + assert_eq!(heal.bucket, bucket); + assert_eq!(heal.object_prefix.as_deref(), Some(object)); + let info = set + .get_object_info(bucket, object, &ObjectOptions::default()) + .await + .expect("committed object must remain readable despite the failed tail"); + assert_eq!(info.size, TEST_OBJECT_SIZE as i64); + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn tail_drained_put_rejects_quorum_minus_one() { + let (_dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = "put-full-tail-no-quorum"; + let object = "full-tail-no-quorum-object"; + make_completion_test_bucket(&disks, bucket).await; + let _fault = rename_fault_injection::fail_rename_on(object, &[0, 1]); + let tasks = rename_fanout_barrier::observe_tasks(object); + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + let err = set + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + .expect_err("draining two successful disks cannot satisfy write quorum three"); + assert!( + matches!(err, Error::ErasureWriteQuorum | Error::InsufficientWriteQuorum(_, _)), + "original quorum error expected: {err}" + ); + assert_eq!(tasks.running(), 0, "failed fan-out and rollback must complete before return"); + assert!( + set.get_object_info(bucket, object, &ObjectOptions::default()).await.is_err(), + "failed fresh write must not become visible" + ); + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn tail_drained_put_owned_commit_survives_waiter_cancellation() { + let (dirs, disks, set) = hermetic_set_disks(4).await; + let bucket = RUSTFS_META_BUCKET; + let object = "full-tail-cancelled-receipt"; + // Internal config writes do not own a bucket lifecycle guard. The object + // guard alone must keep the full-tail coordinator alive after cancellation. + let tasks = rename_fanout_barrier::observe_tasks(object); + let barrier = rename_fanout_barrier::arm(object, 0, rename_fanout_barrier::PHASE_RENAME); + let writer = Arc::clone(&set); + let put = tokio::spawn(async move { + let mut reader = PutObjReader::from_vec(vec![b'1'; TEST_OBJECT_SIZE]); + writer + .put_object( + bucket, + object, + &mut reader, + &ObjectOptions { + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("cancelled receipt must first reach the rename barrier"); + wait_for_paused_tail_metadata_quorum(&disks, bucket, object).await; + put.abort(); + assert!(put.await.expect_err("ACK waiter should cancel").is_cancelled()); + let mut lock_probe = Box::pin(set.acquire_write_lock_diag("cancelled_full_tail_probe", bucket, object)); + assert!( + futures::poll!(lock_probe.as_mut()).is_pending(), + "owned coordinator must retain the namespace guard after waiter cancellation" + ); + barrier.release(); + drop( + tokio::time::timeout(Duration::from_secs(30), lock_probe) + .await + .expect("cancelled coordinator must eventually release its guard") + .expect("post-commit lock probe should succeed"), + ); + assert_eq!(tasks.running(), 0, "cancelled coordinator must reap every rename task"); + for disk in &disks { + disk.read_version("", bucket, object, "", &ReadOptions::default()) + .await + .expect("caller cancellation must not interrupt committed receipt materialization"); + } + wait_for_tmp_workspace_to_drain(&dirs, "cancelled full-tail commit should release staging ownership").await; + } + #[tokio::test] #[serial_test::serial(capacity_dirty_scope)] async fn no_lock_put_waits_for_rename_tail_under_outer_guard() { @@ -18184,6 +18437,7 @@ mod put_object_tmp_cleanup_tests { &mut reader, &ObjectOptions { no_lock: true, + write_completion: WriteCompletion::TailDrained, ..Default::default() }, ) @@ -18209,7 +18463,18 @@ mod put_object_tmp_cleanup_tests { put.await .expect("no-lock PUT task should join") .expect("no-lock PUT should commit after the rename tail releases"); + let mut lock_probe = Box::pin(set_disks.acquire_write_lock_diag("borrowed_full_tail_probe", bucket, object)); + assert!( + futures::poll!(lock_probe.as_mut()).is_pending(), + "full-tail PUT must not release the caller's outer guard" + ); drop(outer_guard); + drop( + tokio::time::timeout(Duration::from_secs(5), lock_probe) + .await + .expect("outer owner releasing its guard should unblock the probe") + .expect("post-outer-guard probe should succeed"), + ); }) .await; } diff --git a/crates/ecstore/src/set_disk/transition_matrix_tests.rs b/crates/ecstore/src/set_disk/transition_matrix_tests.rs index 8924b2a4a..9b5912ce7 100644 --- a/crates/ecstore/src/set_disk/transition_matrix_tests.rs +++ b/crates/ecstore/src/set_disk/transition_matrix_tests.rs @@ -18,6 +18,7 @@ use super::{ }; use crate::bucket::lifecycle::lifecycle::{TRANSITION_COMPLETE, TRANSITION_PENDING, TransitionOptions, expected_expiry_time}; use crate::ecstore_validation_blackbox::make_local_set_disks; +use crate::object_api::WriteCompletion; use crate::services::tier::test_util::register_mock_tier; use crate::storage_api_contracts::bucket::BucketOperations; use crate::storage_api_contracts::object::{ObjectIO as _, ObjectOperations as _}; @@ -72,7 +73,7 @@ async fn transition_and_restore_reclaim_prior_metadata_generations() { object, &mut reader, &ObjectOptions { - no_lock: true, + write_completion: WriteCompletion::TailDrained, ..Default::default() }, ) @@ -185,7 +186,7 @@ async fn prepared_snapshot_transition_duplicate_and_late_get_use_committed_remot object, &mut reader, &ObjectOptions { - no_lock: true, + write_completion: WriteCompletion::TailDrained, ..Default::default() }, ) diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 22d7e37c3..73905389f 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -8045,10 +8045,15 @@ mod tests { ); assert!(com::read_config(store.pools[0].clone(), &second_page_path).await.is_ok()); - com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone()) + let full_tail = ObjectOptions { + max_parity: true, + write_completion: crate::object_api::WriteCompletion::TailDrained, + ..Default::default() + }; + com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, receipt_bytes.clone(), &full_tail) .await .expect("second page receipt should restore"); - com::save_config(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec()) + com::save_config_with_opts(store.pools[target_pool_idx].clone(), &second_page_path, b"{corrupt".to_vec(), &full_tail) .await .expect("second page receipt should corrupt deterministically"); let corrupt = store diff --git a/docs/testing/ecstore-validation-suite-design.md b/docs/testing/ecstore-validation-suite-design.md index e9bec1fe4..de717d1b9 100644 --- a/docs/testing/ecstore-validation-suite-design.md +++ b/docs/testing/ecstore-validation-suite-design.md @@ -54,6 +54,22 @@ Fail-closed invariants every row enforces: Fault injection is explicit and deterministic: local disk mocks for unit tests, process-level disk manipulation (`crates/e2e_test/src/chaos.rs`) for e2e tests. Property tests replay a fixed seed for payload, range, and missing-shard selection. +### PUT completion fixtures + +`ObjectOptions::default()` uses `WriteCompletion::Quorum`: a namespace-lock-owning PUT may acknowledge write quorum while its rename tail retains the lock. A fixture that immediately inspects every disk or primes a metadata generation must set `write_completion: WriteCompletion::TailDrained` and keep normal locking. TailDrained waits for the existing rename fan-out; it does not require every disk to succeed or change fsync policy. Codec-only `no_lock` fixtures do not cover namespace locking. + +The object tests reuse `rename_fanout_barrier::arm(object, disk_slot, phase)` and `observe_tasks(object)`. Wait for the barrier with a deadline, observe actual metadata quorum with `wait_for_paused_tail_metadata_quorum`, then release or cancel. The metadata check distinguishes a real quorum from disk tasks that have not started. Assert zero remaining rename tasks after the owned coordinator releases its lock; cancellation tests also wait for staging cleanup. + +| Fixture | Completion boundary | +|---|---| +| `early_ack_tail_drain_retains_namespace_lock_until_background_rename_finishes` | Default PUT returns before the parked tail; a second writer remains blocked. | +| `tail_drained_put_*` | Explicit full-tail PUT retains its guard, preserves quorum success with a failed minority, rejects quorum-minus-one, and survives ACK waiter cancellation. | +| `transition_and_restore_reclaim_prior_metadata_generations` | Both source fixtures use TailDrained before cache priming, with normal namespace locks. | +| `object_transaction_fencing_persists_epoch_on_multipart_commit` | Multipart completion already always drains rename before inspecting all per-disk transaction UUIDs. | +| `decommission_durable_ilm_receipt_pagination_fails_closed_on_second_page`, `dispatch_completion_cas_is_bounded_and_reaches_the_tail` | Durable receipt, journal, and manifest writers choose TailDrained; the pagination fixture also drains deliberate receipt replacement writes. | + +Select these checks with `cargo nextest list -p rustfs-ecstore --features test-util -E 'test(tail_drained_put) | test(early_ack_tail_drain) | test(no_lock_put_waits_for_rename_tail) | test(object_transaction_fencing_persists_epoch_on_multipart_commit) | test(transition_and_restore_reclaim) | test(decommission_durable_ilm_receipt_pagination) | test(dispatch_completion_cas)'`, then run the same expression under the default and CI profiles without retries. Remaining crash, reopen, rollback, and lock-loss schedules use the existing domain tests; this completion fixture is not a replacement for those checks. + ### Coverage gate `full` and `destructive` run `cargo llvm-cov -p rustfs-ecstore --lib` and fail when line coverage of the gate scope is below `--unit-coverage-min`. The default minimum and the 100% target for EC read, write, decode, heal, metadata-quorum, and rollback paths are the `UNIT_COVERAGE_*` constants at the top of the runner. `cargo-llvm-cov` must be installed unless `--skip-coverage` is passed explicitly. The default scope `ec-critical` is: