From 8a96fedfdd44d5f6193bda0cc49d769f439ee94b Mon Sep 17 00:00:00 2001 From: Hauser Date: Thu, 15 Oct 2026 12:02:41 +0800 Subject: [PATCH] fix(ecstore): retain pool metadata repair ownership (#8552) * fix(ecstore): retain system metadata repair ownership Return system metadata convergence failures to the fenced writer after committed cleanup instead of submitting an unexecutable MRF handoff. Drain system rename tails before acknowledging and preserve ordinary incarnation-bound object repair. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * refactor(ecstore): separate PUT commit and completion outcomes Keep committed metadata writes on their retained cleanup owner when the writer must handle repair. Propagate the completion error at the API boundary without a duplicate temporary-data deletion or worker wait. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(ecstore): scope repair completion to pool metadata owners Keep ordinary system metadata quorum-compatible and explicitly select owner-observed repair completion for fenced pool metadata CAS writes. Seed format-heal metadata before injecting the missing-format fault. Preserve degraded bucket recreation and test all completion modes. Co-Authored-By: heihutu Co-Authored-By: zhi22915 --------- Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/ecstore/src/core/pools.rs | 10 +- crates/ecstore/src/object_api/types.rs | 4 + crates/ecstore/src/set_disk/mod.rs | 25 +- crates/ecstore/src/set_disk/ops/object.rs | 401 +++++++++++++++++++--- crates/ecstore/src/store/heal.rs | 47 ++- docs/operations/pool-metadata-recovery.md | 2 + 6 files changed, 421 insertions(+), 68 deletions(-) diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index f807eb603..22e5b802f 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -6685,12 +6685,12 @@ fn pool_meta_cas_options(object: &str, token: &PoolMetaCasToken, fence: &PoolMet 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() }; fence.add_to_options(&mut opts, object); + opts.write_completion = crate::object_api::WriteCompletion::TailDrainedWithOwnerRepair; Ok(opts) } @@ -6735,7 +6735,11 @@ where "payload_sha256": rustfs_utils::crypto::hex(Sha256::digest(&data)), "if_match": opts.http_preconditions.as_ref().and_then(|p| p.if_match.as_deref()), "if_none_match": opts.http_preconditions.as_ref().and_then(|p| p.if_none_match.as_deref()), - "tail_drained": opts.write_completion == crate::object_api::WriteCompletion::TailDrained, + "tail_drained": matches!( + opts.write_completion, + crate::object_api::WriteCompletion::TailDrained + | crate::object_api::WriteCompletion::TailDrainedWithOwnerRepair + ), "no_lock": opts.no_lock, }) }); @@ -24692,7 +24696,7 @@ mod pools_tests { _ => panic!("phase fixture must receive valid pool metadata"), }; assert!(opts.max_parity && opts.no_lock); - assert_eq!(opts.write_completion, crate::object_api::WriteCompletion::TailDrained); + assert_eq!(opts.write_completion, crate::object_api::WriteCompletion::TailDrainedWithOwnerRepair); self.trace.calls.lock().expect("trace lock").push(WriteCall { pool: self.pool, phase, diff --git a/crates/ecstore/src/object_api/types.rs b/crates/ecstore/src/object_api/types.rs index 896191833..d35792297 100644 --- a/crates/ecstore/src/object_api/types.rs +++ b/crates/ecstore/src/object_api/types.rs @@ -898,6 +898,10 @@ pub enum WriteCompletion { /// 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, + /// Drain the rename fan-out and return system-metadata repair to the + /// caller's existing deployment/transaction owner. This does not grant + /// write or heal authority and does not change the quorum requirement. + TailDrainedWithOwnerRepair, } /// Storage-owned write mode inherited by physical rewrites. This selects the diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index cd736e932..8e129558b 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -4778,6 +4778,10 @@ impl DiskHealthEntry { } } +#[derive(Debug, thiserror::Error)] +#[error("committed system metadata write requires repair by its owning writer")] +pub(crate) struct SystemMetadataRepairRequired; + impl SetDisks { pub(in crate::set_disk) async fn persist_delete_marker_purge( &self, @@ -4888,7 +4892,7 @@ impl SetDisks { state = "mrf_durable_admission_failed", bucket, object, version_id, pool_index = self.pool_index, set_index = self.set_index, error = %error, - "Partial write repair falling back to the heal channel" + "Partial write repair responsibility could not be persisted" ); false } @@ -4898,7 +4902,17 @@ impl SetDisks { pub(in crate::set_disk) async fn submit_rename_tail_heal( &self, mut request: rustfs_heal_contracts::heal_channel::HealChannelRequest, - ) { + completion: crate::object_api::WriteCompletion, + ) -> Result<()> { + // System metadata repair remains with the writer's deployment and + // transaction fence; a user-bucket MRF authority cannot authorize it. + if crate::bucket::utils::is_meta_bucketname(&request.bucket) { + return if completion == crate::object_api::WriteCompletion::TailDrainedWithOwnerRepair { + Err(Error::other(SystemMetadataRepairRequired)) + } else { + Ok(()) + }; + } if let Some(object) = request.object_prefix.as_deref() && self .persist_partial_write( @@ -4909,10 +4923,10 @@ impl SetDisks { ) .await { - return; + return Ok(()); } if request.expected_bucket_incarnation_id.is_none() { - return; + return Ok(()); } request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Mrf; #[cfg(test)] @@ -4924,11 +4938,12 @@ impl SetDisks { .clone(); if let Some(capture) = capture { let _ = capture.send(request); - return; + return Ok(()); } } let _ = rustfs_heal_contracts::heal_channel::send_heal_request(request).await; + Ok(()) } #[cfg(test)] diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index 47e166d42..7f267aee0 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -3747,6 +3747,12 @@ pub(in crate::set_disk) fn merge_replication_metadata_lww( changed } +struct CommittedPutOutcome { + // Namespace publication succeeded even if its writer must retain repair + // responsibility. Completion errors must not select rollback cleanup. + completion: Result<(ObjectInfo, Option)>, +} + impl SetDisks { /// Failure cleanup stays awaited, but shares the same execution budget as /// successful PUT continuations when isolation is enabled. @@ -5093,7 +5099,8 @@ impl SetDisks { || commit_object_lock_guard.is_some() || commit_decommission_object_lock_guard.is_some() || commit_publication_guard.is_some(); - let commit_allows_early_ack = write_context.allows_early_ack(opts.write_completion) + let commit_write_completion = opts.write_completion; + let commit_allows_early_ack = write_context.allows_early_ack(commit_write_completion) && !(opts.data_movement && opts.has_decommission_capacity_reservation()) && commit_owns_namespace_guard && commit_scanner_publication_scope.is_none(); @@ -5570,7 +5577,9 @@ impl SetDisks { ); }) }, - |request| async move { heal_set.submit_rename_tail_heal(request).await }, + move |request| async move { + let _ = heal_set.submit_rename_tail_heal(request, commit_write_completion).await; + }, )); } } @@ -5671,7 +5680,7 @@ impl SetDisks { ); } - if needs_immediate_heal { + let repair_result = if needs_immediate_heal { rustfs_io_metrics::record_put_object_stage_duration( rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_HEAL_SUBMIT_TRIGGERED, 1.0, @@ -5689,17 +5698,19 @@ impl SetDisks { .or_else(|| commit_version_suspended.then(Uuid::nil)) .map(|version_id| version_id.to_string()); request.expected_bucket_incarnation_id = source_bucket_incarnation_id; - commit_set.submit_rename_tail_heal(request).await; + let result = commit_set.submit_rename_tail_heal(request, commit_write_completion).await; rustfs_io_metrics::record_put_object_stage_duration_from( rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_HEAL_SUBMIT, heal_submit_started, ); + result } else { rustfs_io_metrics::record_put_object_stage_duration( rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_HEAL_SUBMIT_SKIPPED, 1.0, ); - } + Ok(()) + }; let rename_stage_elapsed = rename_stage_start.elapsed(); let rename_stage_ms = rename_stage_elapsed.as_millis() as u64; @@ -5909,6 +5920,7 @@ impl SetDisks { event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY, component = LOG_COMPONENT_ECSTORE, subsystem = LOG_SUBSYSTEM_SET_DISK, + result = if repair_result.is_ok() { "success" } else { "repair_required" }, bucket = %commit_bucket, object = %commit_object, write_quorum, @@ -5919,7 +5931,6 @@ impl SetDisks { cleanup_ms = cleanup_stage_ms.unwrap_or_default(), cleanup_present = cleanup_stage_ms.is_some(), commit_tail_ms = total_commit_tail_ms as u64, - result = "success", "SetDisk put_object stage summary" ); } @@ -5940,26 +5951,29 @@ impl SetDisks { }); } - let object_info_build_started = rustfs_io_metrics::put_stage_timer(); - let object_info = ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned); - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_OBJECT_INFO_BUILD, - object_info_build_started, - ); - let post_object_info_to_return_started = rustfs_io_metrics::put_stage_timer(); - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_POST_OBJECT_INFO_TO_RETURN, - post_object_info_to_return_started, - ); - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_AFTER_RENAME_TO_RETURN, - commit_after_rename_to_return_started, - ); - rustfs_io_metrics::record_put_object_stage_duration_from( - rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_TOTAL, - commit_closure_total_started, - ); - Ok((object_info, old_current_size)) + let completion = repair_result.map(|()| { + let object_info_build_started = rustfs_io_metrics::put_stage_timer(); + let object_info = ObjectInfo::from_file_info(&fi, &commit_bucket, &commit_object, commit_is_versioned); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_OBJECT_INFO_BUILD, + object_info_build_started, + ); + let post_object_info_to_return_started = rustfs_io_metrics::put_stage_timer(); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_NAMESPACE_LOCK_HELD_POST_OBJECT_INFO_TO_RETURN, + post_object_info_to_return_started, + ); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_AFTER_RENAME_TO_RETURN, + commit_after_rename_to_return_started, + ); + rustfs_io_metrics::record_put_object_stage_duration_from( + rustfs_io_metrics::PUT_STAGE_PUT_OBJECT_COMMIT_CLOSURE_TOTAL, + commit_closure_total_started, + ); + (object_info, old_current_size) + }); + Ok(CommittedPutOutcome { completion }) }; if let Some(handoff) = commit_cancellation_handoff_tx.take() { @@ -6011,8 +6025,7 @@ impl SetDisks { } }; let operation_select_wait_started = rustfs_io_metrics::put_stage_timer(); - let result: Result<(ObjectInfo, Option)> = if let Some((cancellation, mut handoff_rx)) = cancellation_wait - { + let result: Result = if let Some((cancellation, mut handoff_rx)) = cancellation_wait { tokio::pin!(operation); tokio::select! { biased; @@ -6037,27 +6050,6 @@ impl SetDisks { operation_result_return_started, ); - if issue3031_diag_enabled() - && let Err(err) = &result - { - let stage_hint = if err.to_string().contains("not enough disks to write") { - "writer_setup_or_quorum" - } else { - "unknown" - }; - warn!( - event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY, - component = LOG_COMPONENT_ECSTORE, - subsystem = LOG_SUBSYSTEM_SET_DISK, - bucket = %bucket, - object = %object, - result = "error", - stage_hint, - error = %err, - "SetDisk put_object stage summary" - ); - } - if issue3031_diag_enabled() { warn!( target: "rustfs_ecstore::set_disk", @@ -6111,6 +6103,29 @@ impl SetDisks { } } + let result = result.and_then(|committed| committed.completion); + + if issue3031_diag_enabled() + && let Err(err) = &result + { + let stage_hint = if err.to_string().contains("not enough disks to write") { + "writer_setup_or_quorum" + } else { + "unknown" + }; + warn!( + event = EVENT_SET_DISK_PUT_OBJECT_STAGE_SUMMARY, + component = LOG_COMPONENT_ECSTORE, + subsystem = LOG_SUBSYSTEM_SET_DISK, + result = "error", + bucket = %bucket, + object = %object, + stage_hint, + error = %err, + "SetDisk put_object stage summary" + ); + } + result } } @@ -21784,6 +21799,294 @@ mod put_object_tmp_cleanup_tests { .await; } + #[tokio::test] + async fn system_metadata_repair_error_does_not_wait_for_committed_tmp_cleanup() { + const CHILD: &str = "RUSTFS_METADATA_REPAIR_CLEANUP_TEST_CHILD"; + if std::env::var_os(CHILD).is_none() { + let output = tokio::task::spawn_blocking(|| { + std::process::Command::new(std::env::current_exe().expect("test executable must be available")) + .args([ + "--exact", + "set_disk::ops::object::put_object_tmp_cleanup_tests::system_metadata_repair_error_does_not_wait_for_committed_tmp_cleanup", + "--nocapture", + ]) + .env(CHILD, "1") + .env("RUSTFS_CLEANUP_ISOLATE_ENABLE", "true") + .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_MAX_PENDING", "2") + .env("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_WORKERS", "1") + .env("RUSTFS_CLEANUP_CPUS", "none") + .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_COUNTERFACTUAL_SKIP") + .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_DEFER_HOLD_WORKER") + .env_remove("RUSTFS_PUT_RENAME_TAIL_CLEANUP_ZERO_TARGET_TMP_DELETE_SKIP") + .output() + .expect("the isolated cleanup regression child must run") + }) + .await + .expect("cleanup regression child launcher must join"); + assert!( + output.status.success(), + "cleanup regression failed:\n{}\n{}", + String::from_utf8_lossy(&output.stdout), + String::from_utf8_lossy(&output.stderr) + ); + assert!(String::from_utf8_lossy(&output.stdout).contains("1 passed")); + return; + } + + let (dirs, _disks, set) = hermetic_set_disks(4).await; + let occupied = cleanup_runtime::reserve() + .expect("the cleanup blocker must be admitted") + .expect("the child must enable cleanup isolation"); + let (ready, started) = tokio::sync::oneshot::channel(); + let (release, released) = tokio::sync::oneshot::channel(); + let blocker = cleanup_runtime::spawn(Some(occupied), async move { + ready.send(()).expect("the fixture must observe the cleanup worker"); + let _ = released.await; + }); + tokio::time::timeout(Duration::from_secs(30), started) + .await + .expect("the cleanup blocker must acquire the execution worker") + .expect("the blocker readiness sender must stay alive"); + + let object = "pool.bin.identity-cleanup-owner"; + let body = b"committed metadata with pending repair".to_vec(); + let fault = rename_fault_injection::fail_rename_on(object, &[0]); + let response = tokio::time::timeout( + Duration::from_secs(30), + crate::config::com::save_config_with_opts( + Arc::clone(&set), + object, + body.clone(), + &ObjectOptions { + max_parity: true, + write_completion: WriteCompletion::TailDrainedWithOwnerRepair, + ..Default::default() + }, + ), + ) + .await; + release.send(()).expect("the cleanup blocker must stay alive until released"); + tokio::time::timeout(Duration::from_secs(30), blocker) + .await + .expect("the released cleanup blocker must finish") + .expect("the cleanup blocker task must join"); + let error = response + .expect("repair-required response must not wait for the committed temporary cleanup owner") + .expect_err("the committed minority failure must reach the metadata writer"); + assert!( + matches!(error, Error::Io(ref inner) if inner.get_ref().is_some_and(|source| source.is::())) + ); + drop(fault); + wait_for_tmp_workspace_to_drain(&dirs, "the retained cleanup owner must eventually remove staged metadata").await; + let committed = crate::config::com::read_config(Arc::clone(&set), object) + .await + .expect("the repair-required metadata quorum must remain readable"); + assert_eq!(committed, body); + } + + #[tokio::test] + async fn rename_tail_heal_keeps_system_authority_separate_from_user_incarnation() { + let (_dirs, _disks, set) = hermetic_set_disks(4).await; + let mut heals = set.capture_test_rename_tail_heals(); + let incarnation = Uuid::new_v4(); + for bucket in [RUSTFS_META_BUCKET, ".minio.sys"] { + for expected in [None, Some(incarnation)] { + let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options( + bucket.to_string(), + Some("pool.bin.identity".to_string()), + false, + Some(HealChannelPriority::Normal), + Some(0), + Some(0), + ); + request.expected_bucket_incarnation_id = expected; + let error = set + .submit_rename_tail_heal(request, WriteCompletion::TailDrainedWithOwnerRepair) + .await + .expect_err("system repair must remain with its writer"); + assert!( + matches!(error, Error::Io(ref inner) if inner.get_ref().is_some_and(|source| source.is::())) + ); + assert!( + heals.try_recv().is_err(), + "even a supplied user incarnation cannot authorize system metadata heal" + ); + } + } + let mut request = rustfs_heal_contracts::heal_channel::create_heal_request_with_options( + "ordinary-bucket".to_string(), + Some("ordinary-object".to_string()), + false, + Some(HealChannelPriority::Normal), + Some(0), + Some(0), + ); + request.expected_bucket_incarnation_id = Some(incarnation); + set.submit_rename_tail_heal(request, WriteCompletion::Quorum) + .await + .expect("ordinary incarnation-bound fallback must remain available"); + let admitted = heals + .try_recv() + .expect("the ordinary fallback must actually reach the channel"); + assert_eq!(admitted.expected_bucket_incarnation_id, Some(incarnation)); + assert_eq!(admitted.source, rustfs_heal_contracts::heal_channel::HealRequestSource::Mrf); + assert_eq!(admitted.object_prefix.as_deref(), Some("ordinary-object")); + } + + #[tokio::test] + #[serial_test::serial(capacity_dirty_scope)] + async fn system_metadata_partial_write_respects_completion_contract() { + temp_env::async_with_vars([(ENV_RUSTFS_PUT_RENAME_EARLY_ACK_ENABLE, Some("true"))], async { + for completion in [ + WriteCompletion::Quorum, + WriteCompletion::TailDrained, + WriteCompletion::TailDrainedWithOwnerRepair, + ] { + let (dirs, disks, set) = hermetic_set_disks(4).await; + let object = format!("pool.bin.identity-{completion:?}"); + let original = b"original deployment identity".to_vec(); + let fixture_options = ObjectOptions { + max_parity: true, + write_completion: WriteCompletion::TailDrained, + ..Default::default() + }; + crate::config::com::save_config_with_opts(Arc::clone(&set), &object, original.clone(), &fixture_options) + .await + .expect("healthy system metadata write must succeed without a bucket incarnation"); + let original_info = set + .get_object_info(RUSTFS_META_BUCKET, &object, &ObjectOptions::default()) + .await + .expect("original metadata must be readable before the partial overwrite"); + let original_etag = original_info.etag.expect("original metadata must have an ETag"); + let updated = b"updated deployment identity".to_vec(); + 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 write_object = object.clone(); + let write_body = updated.clone(); + let write = tokio::spawn(async move { + crate::config::com::save_config_with_opts( + writer, + &write_object, + write_body, + &ObjectOptions { + max_parity: true, + write_completion: completion, + ..Default::default() + }, + ) + .await + }); + tokio::time::timeout(Duration::from_secs(30), barrier.wait_until_paused()) + .await + .expect("the metadata rename tail must reach the failure barrier"); + tokio::time::timeout(Duration::from_secs(30), async { + loop { + let mut updated_copies = 0; + for disk in &disks { + let info = disk + .read_version("", RUSTFS_META_BUCKET, &object, "", &ReadOptions::default()) + .await + .expect("each disk must retain either the original or updated metadata"); + if info.metadata.get("etag").is_some_and(|etag| etag != &original_etag) { + updated_copies += 1; + } + } + if updated_copies == 3 { + break; + } + tokio::task::yield_now().await; + } + }) + .await + .expect("the metadata overwrite must reach quorum while the last disk is paused"); + if completion != WriteCompletion::Quorum { + assert!( + !write.is_finished(), + "drained completion must wait for the rename tail before acknowledging" + ); + } + barrier.release(); + let result = tokio::time::timeout(Duration::from_secs(30), write) + .await + .expect("the metadata tail must drain") + .expect("the metadata writer task must join"); + if completion == WriteCompletion::TailDrainedWithOwnerRepair { + let error = result.expect_err("the pool metadata owner must receive repair responsibility"); + let Error::Io(error) = error else { + panic!("system metadata repair must retain its typed error"); + }; + assert!( + error + .get_ref() + .is_some_and(|source| source.is::()), + "the config writer must retain the exact repair-required cause: {error}" + ); + } else { + result.expect("ordinary system metadata must preserve quorum-success semantics"); + } + tokio::time::timeout(Duration::from_secs(30), async { + while tasks.running() != 0 { + tokio::task::yield_now().await; + } + }) + .await + .expect("the committed rename tail must eventually finish for every completion mode"); + assert_eq!(tasks.running(), 0, "repair failure must be returned after all rename tasks drain"); + assert!(heals.try_recv().is_err(), "system metadata must never enqueue an unfenced MRF heal"); + let mut stale_copies = 0; + let mut committed_etags = Vec::new(); + for disk in &disks { + let info = disk + .read_version("", RUSTFS_META_BUCKET, &object, "", &ReadOptions::default()) + .await + .expect("every physical disk must retain its metadata after the partial rename"); + let etag = info.metadata.get("etag").expect("each metadata copy must retain an ETag"); + if etag == &original_etag { + stale_copies += 1; + } else { + committed_etags.push(etag.clone()); + } + } + assert_eq!(stale_copies, 1, "exactly the failed rename must retain the old metadata"); + assert_eq!(committed_etags.len(), 3, "the new metadata must retain its write quorum"); + assert!(committed_etags.iter().all(|etag| etag == &committed_etags[0])); + let committed = + tokio::time::timeout(Duration::from_secs(30), crate::config::com::read_config(Arc::clone(&set), &object)) + .await + .expect("the committed metadata read must finish after the rename tail drains") + .expect("a repair-required result must not roll back the committed metadata quorum"); + assert_eq!(committed, updated, "the read must see the new bytes after cache invalidation"); + tokio::time::timeout(Duration::from_secs(30), async { + while !non_trash_tmp_entries(&dirs).await.is_empty() { + tokio::task::yield_now().await; + } + }) + .await + .expect("repair-required results must still schedule committed temporary-data cleanup"); + drop(fault); + drop(barrier); + tokio::time::timeout( + Duration::from_secs(30), + crate::config::com::save_config_with_opts(Arc::clone(&set), &object, updated.clone(), &fixture_options), + ) + .await + .expect("the reconciliation must finish after the fault and barrier are disarmed") + .expect("the owning writer must be able to reconcile after the failed disk recovers"); + for disk in &disks { + let info = disk + .read_version("", RUSTFS_META_BUCKET, &object, "", &ReadOptions::default()) + .await + .expect("the reconciled metadata must exist on every disk"); + assert_ne!(info.metadata.get("etag"), Some(&original_etag)); + } + } + }) + .await; + } + #[tokio::test] #[serial_test::serial(capacity_dirty_scope)] async fn tail_drained_put_preserves_quorum_success_without_unfenced_fallback_heal() { diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs index 28f33f48c..2a7dbad4d 100644 --- a/crates/ecstore/src/store/heal.rs +++ b/crates/ecstore/src/store/heal.rs @@ -3740,7 +3740,6 @@ mod tests { let mut recoverable_temp_dirs = Vec::new(); let mut recoverable_endpoints = Vec::new(); let mut recoverable_disks = Vec::new(); - let mut unformatted_disk = None; for disk_index in 0..3 { let temp_dir = tempfile::tempdir().expect("temporary disk root should be created"); let mut endpoint = Endpoint::try_from(temp_dir.path().to_str().expect("temporary path should be UTF-8")) @@ -3757,15 +3756,11 @@ mod tests { ) .await .expect("temporary disk should open"); - if disk_index < 2 { - let mut disk_format = recoverable_format.clone(); - disk_format.erasure.this = recoverable_format.erasure.sets[0][disk_index]; - save_format_file(&Some(disk.clone()), &Some(disk_format)) - .await - .expect("recoverable format should be written"); - } else { - unformatted_disk = Some(disk.clone()); - } + let mut disk_format = recoverable_format.clone(); + disk_format.erasure.this = recoverable_format.erasure.sets[0][disk_index]; + save_format_file(&Some(disk.clone()), &Some(disk_format)) + .await + .expect("recoverable format should be written before pool metadata"); recoverable_temp_dirs.push(temp_dir); recoverable_endpoints.push(endpoint); recoverable_disks.push(Some(disk)); @@ -3814,6 +3809,36 @@ mod tests { .expect("pool metadata should be persisted before format heal"); *store.pool_meta_save_gate.lock().await = PoolMetaWriteState::default(); + let replacement_endpoint = store.pools[1].endpoints.endpoints.as_ref()[2].clone(); + tokio::fs::remove_file( + recoverable_temp_dirs[2] + .path() + .join(crate::disk::RUSTFS_META_BUCKET) + .join(FORMAT_CONFIG_FILE), + ) + .await + .expect("the target format must be removed only after durable pool metadata exists"); + let replacement_disk = new_disk( + &replacement_endpoint, + &DiskOption { + cleanup: false, + health_check: false, + }, + ) + .await + .expect("the unformatted replacement disk must reopen"); + assert!(matches!( + load_format_erasure(&replacement_disk, true).await, + Err(crate::disk::error::DiskError::UnformattedDisk) + )); + let replacement_set = &store.pools[1].disk_set[0]; + let replacement_slot = replacement_set + .set_endpoints + .iter() + .position(|endpoint| endpoint == &replacement_endpoint) + .expect("the replacement must remain in its original set"); + replacement_set.disks.write().await[replacement_slot] = Some(replacement_disk.clone()); + let (result, err) = store .handle_heal_format(false) .await @@ -3823,7 +3848,7 @@ mod tests { "foreign format majority must not be downgraded to a successful heal: {err:?}" ); assert_eq!(result.disk_count, 3, "the recoverable pool should still be inspected"); - let healed = load_format_erasure(&unformatted_disk.expect("the unformatted disk handle should be retained"), true) + let healed = load_format_erasure(&replacement_disk, true) .await .expect("the later pool should be healed despite the first pool error"); assert_eq!(healed.erasure.this, recoverable_format.erasure.sets[0][2]); diff --git a/docs/operations/pool-metadata-recovery.md b/docs/operations/pool-metadata-recovery.md index d4281530a..fd0cf6112 100644 --- a/docs/operations/pool-metadata-recovery.md +++ b/docs/operations/pool-metadata-recovery.md @@ -51,6 +51,8 @@ An unreadable replica, lost fence, or conditional-write conflict leaves the orig - `rustfs_pool_metadata_blocks_total{reason}` and `rustfs_pool_metadata_recoveries_total` count block and recovery transitions. The original cause and phase remain attached to local typed errors; storage/RPC error numbers and on-disk formats are unchanged. - Node and cluster-write readiness include `pool_meta_write_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md). +Pool metadata CAS explicitly selects `TailDrainedWithOwnerRepair`: it waits for the complete rename result and returns repair responsibility to its existing deployment and transaction owner when write quorum commits but a replica fails or signatures diverge. Cache invalidation and the committed cleanup handoff precede that error. Ordinary system metadata retains quorum-success semantics under `Quorum` and `TailDrained`; draining a tail does not require every disk to succeed. A repair-required completion does not enter rollback cleanup or wait for another temporary-data deletion. The committed bytes are not rolled back. Generic MRF and heal-channel requests cannot supply the deployment and transaction authority needed to repair system metadata; pool CAS retains the uncertain outcome and its existing recovery fence. Ordinary object repair keeps its bucket-incarnation fencing. + If a block persists, inspect the first block and subsequent recovery phase, restore disk/peer readability, and verify every metadata and identity copy before restarting. Do not delete metadata to make readiness green. ## Disk replacement and metadata erasure