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

@@ -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

.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
 

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

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]

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<RenameDataResp> {
         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

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;
     }
 

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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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]