From 5cd58319ed6148ed7f09f2a4d0b4e46e429f043a Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 10 Sep 2026 21:09:02 +0800 Subject: [PATCH] fix(heal): preserve retryable batch failures during recovery (#7642) * fix(heal): preserve retryable batch failures during recovery * test(heal): pin prebuilt hooks binaries in ci --- .github/workflows/ci.yml | 1 + .github/workflows/e2e-replication-nightly.yml | 1 + crates/e2e_test/README.md | 7 ++ .../src/heal_erasure_disk_rebuild_test.rs | 19 +++++ crates/ecstore/src/disk/local/commit.rs | 41 +++++++++++ crates/heal/src/heal/manager.rs | 5 +- crates/heal/src/heal/manager/tests.rs | 73 ++++++++++++++----- 7 files changed, 127 insertions(+), 20 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 7e982054c..b7f8e849a 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -997,6 +997,7 @@ jobs: # debug binary; each test spawns its own rustfs server on a random port. - name: Run e2e full suite env: + CARGO_BIN_EXE_rustfs: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs RUSTFS_E2E_STARTUP_CAS_BINARY: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs RUSTFS_E2E_STARTUP_CAS_BUILD_MANIFEST: ${{ runner.temp }}/rustfs-startup-cas-input/rustfs.e2e-startup-cas-build.json RUSTFS_E2E_STARTUP_CAS_ARTIFACT_DIR: ${{ runner.temp }}/rustfs-startup-cas-evidence diff --git a/.github/workflows/e2e-replication-nightly.yml b/.github/workflows/e2e-replication-nightly.yml index e59d1155a..d82c53f17 100644 --- a/.github/workflows/e2e-replication-nightly.yml +++ b/.github/workflows/e2e-replication-nightly.yml @@ -156,6 +156,7 @@ jobs: - name: Run cluster fault e2e nightly suite env: + CARGO_BIN_EXE_rustfs: ${{ github.workspace }}/target/debug/rustfs RUSTFS_E2E_LOG_DIR: ${{ runner.temp }}/rustfs-e2e-nightly-logs run: cargo nextest run --profile e2e-nightly -p e2e_test diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md index 28da958bc..cc2398076 100644 --- a/crates/e2e_test/README.md +++ b/crates/e2e_test/README.md @@ -38,6 +38,13 @@ All commands assume repo root. `cargo test` triggers an on-demand build of the `rustfs` binary from [`src/common.rs`](src/common.rs) (`rustfs_binary_path`) on first use — the first invocation is slow, later ones reuse the binary. +Root-heal interruption scenarios use a test-only commit barrier. Prebuild with `e2e-test-hooks` and pin that binary so concurrent cases do not replace it through on-demand builds: + +```bash +cargo build -p rustfs --bin rustfs --features e2e-test-hooks +CARGO_BIN_EXE_rustfs="$PWD/target/debug/rustfs" cargo nextest run -p e2e_test -E 'test(heal_erasure_disk_rebuild_test)' +``` + ```bash # Whole crate (default = ignored tests skipped) cargo nextest run -p e2e_test diff --git a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs index 415673a17..a4df5f4b2 100644 --- a/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs +++ b/crates/e2e_test/src/heal_erasure_disk_rebuild_test.rs @@ -1530,6 +1530,16 @@ mod tests { } } + // Keep the partial-repair checkpoint stable across readiness and admin + // requests. Endpoint-blackhole tests must prove their own network stall. + let commit_barrier = if scenario != InterruptionScenario::TargetEndpointBlackhole { + let barrier = replaced_disk.join(".rustfs.sys/e2e-heal-commit-barrier"); + std::fs::create_dir_all(barrier.parent().ok_or("commit barrier has no parent")?)?; + std::fs::write(&barrier, format!("{bucket}/cluster/online/"))?; + Some(barrier) + } else { + None + }; cluster.start_node_from_binary(1, &server_binary).await?; let status_url = format!("{}/rustfs/admin/v3/background-heal/status", cluster.nodes[0].url); @@ -1663,6 +1673,12 @@ mod tests { sleep(Duration::from_millis(10)).await; }; + if let Some(barrier) = &commit_barrier { + assert!( + barrier.with_extension("admitted").is_file(), + "interruption tests require a server built with e2e-test-hooks" + ); + } let pre_interrupt_status_body = signed_admin_post(&status_url, None, &cluster.access_key, &cluster.secret_key).await?; let pre_interrupt_status: serde_json::Value = serde_json::from_str(&pre_interrupt_status_body) .map_err(|err| format!("pre-interrupt background heal status is not JSON ({err}): {pre_interrupt_status_body}"))?; @@ -1868,6 +1884,9 @@ mod tests { } } } + if let Some(barrier) = &commit_barrier { + std::fs::remove_file(barrier)?; + } cluster.start_node_from_binary(interruption_node, &server_binary).await?; if interruption_node == 0 { let target = cluster.nodes[1] diff --git a/crates/ecstore/src/disk/local/commit.rs b/crates/ecstore/src/disk/local/commit.rs index aa32f6fd7..903e861bc 100644 --- a/crates/ecstore/src/disk/local/commit.rs +++ b/crates/ecstore/src/disk/local/commit.rs @@ -47,6 +47,43 @@ use tokio::fs; use tracing::{info, warn}; use uuid::Uuid; +/// Hold later repair publications after admitting one baseline object. The +/// fixture arms this on one replacement disk before rejoining the cluster. +#[cfg(feature = "e2e-test-hooks")] +async fn wait_for_heal_commit_test_barrier(root: &Path, bucket: &str, object: &str) -> Result<()> { + use tokio::io::AsyncWriteExt; + + let barrier = root.join(".rustfs.sys/e2e-heal-commit-barrier"); + let prefix = match fs::read_to_string(&barrier).await { + Ok(prefix) => prefix, + Err(error) if error.kind() == ErrorKind::NotFound => return Ok(()), + Err(error) => return Err(error.into()), + }; + let key = format!("{bucket}/{object}"); + if prefix.is_empty() || !key.starts_with(&prefix) { + return Ok(()); + } + let admitted = barrier.with_extension("admitted"); + match fs::OpenOptions::new().write(true).create_new(true).open(&admitted).await { + Ok(mut file) => { + file.write_all(key.as_bytes()).await?; + return Ok(()); + } + Err(error) if error.kind() == ErrorKind::AlreadyExists => {} + Err(error) => return Err(error.into()), + } + let deadline = tokio::time::Instant::now() + std::time::Duration::from_secs(120); + loop { + if !fs::try_exists(&barrier).await? || fs::read_to_string(&admitted).await? == key { + return Ok(()); + } + if tokio::time::Instant::now() >= deadline { + return Err(std::io::Error::new(ErrorKind::TimedOut, "heal commit test barrier was not released").into()); + } + tokio::time::sleep(std::time::Duration::from_millis(10)).await; + } +} + fn rollback_committed_rename_std( dst_file_path: &Path, new_data_path: Option<&Path>, @@ -253,6 +290,10 @@ impl LocalDisk { state: &mut RenameDataState, ) -> Result { crate::hp_guard!("LocalDisk::rename_data"); + #[cfg(feature = "e2e-test-hooks")] + if fi.is_healing() { + wait_for_heal_commit_test_barrier(&self.root, dst_volume, dst_path).await?; + } let mut fi = fi; // A non-force DeleteBucket must not remove a directory while a local // object commit is publishing into it. The peer's empty scan remains diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index b269c446c..54ed5cf8e 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -548,7 +548,10 @@ fn retry_budget_for_result(task: &HealTask, result: &Result<()>, retryable_batch } let error = err.to_string(); - if !err.is_recoverable_heal() { + // Batch aggregation preserves the typed classification in its counters, + // while the returned task error retains only the first error's display text. + let retryable_batch_result = retryable_batch_failure && matches!(err, Error::TaskExecutionFailed { .. }); + if !retryable_batch_result && !err.is_recoverable_heal() { return None; } diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 6b04bad40..9f60ded32 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -2614,27 +2614,62 @@ fn test_retry_request_for_recoverable_error_stops_at_limit() { #[tokio::test] async fn test_retry_request_rescans_batch_when_all_exhausted_objects_are_retryable() { - let storage: Arc = Arc::new(MockStorage); - let task = HealTask::from_request(HealRequest::bucket("bucket".to_string()), storage); - let result = Err(task - .record_batch_failure(BatchHealFailure { - scope: "bucket:bucket".to_string(), - failed: 1, - retryable: 1, - permanent: 0, - first_object: "object".to_string(), - first_error: "Lock acquisition timeout".to_string(), - }) - .await); + for source_error in [ + Error::Disk(DiskError::FaultyDisk), + Error::Disk(DiskError::FaultyRemoteDisk), + Error::Storage(EcstoreError::SlowDown), + Error::TaskExecutionFailed { + message: "Lock acquisition timeout".to_string(), + }, + ] { + assert!(source_error.is_recoverable_heal()); + let first_error = source_error.to_string(); + let storage: Arc = Arc::new(MockStorage); + let task = HealTask::from_request(HealRequest::bucket("bucket".to_string()), storage); + let result = Err(task + .record_batch_failure(BatchHealFailure { + scope: "bucket:bucket".to_string(), + failed: 1, + retryable: 1, + permanent: 0, + first_object: "object".to_string(), + first_error: first_error.clone(), + }) + .await); - let (retry_request, retry_delay, error) = retry_request_for_result_with_budget(&task, &result) - .await - .expect("all-retryable batch failure should rescan within the manager retry budget"); + let (retry_request, retry_delay, error) = retry_request_for_result_with_budget(&task, &result) + .await + .expect("all-retryable batch failure should rescan within the manager retry budget"); - assert_eq!(retry_request.id, task.id); - assert_eq!(retry_request.retry_attempts, 1); - assert!(retry_delay > Duration::ZERO); - assert!(error.contains("Lock acquisition timeout")); + assert_eq!(retry_request.id, task.id); + assert_eq!(retry_request.retry_attempts, 1); + assert!(retry_delay > Duration::ZERO); + assert!(error.contains(&first_error)); + } +} + +#[tokio::test] +async fn test_retry_request_does_not_rescan_cancelled_or_timed_out_retryable_batch() { + for terminal_error in [Error::TaskCancelled, Error::TaskTimeout] { + let storage: Arc = Arc::new(MockStorage); + let task = HealTask::from_request(HealRequest::bucket("bucket".to_string()), storage); + let _ = task + .record_batch_failure(BatchHealFailure { + scope: "bucket:bucket".to_string(), + failed: 1, + retryable: 1, + permanent: 0, + first_object: "object".to_string(), + first_error: Error::Disk(DiskError::FaultyDisk).to_string(), + }) + .await; + + assert!( + retry_request_for_result_with_budget(&task, &Err(terminal_error)) + .await + .is_none() + ); + } } #[tokio::test]