fix(heal): supervise scheduler task panics

crates/e2e_test/src/kms/bucket_default_encryption_test.rs

@@ -37,7 +37,7 @@ async fn test_bucket_default_sse_s3_put_object() -> Result<(), Box<dyn std::erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -159,7 +159,7 @@ async fn test_bucket_default_sse_kms_put_object() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -278,7 +278,7 @@ async fn test_bucket_default_sse_kms_multipart_crc32() -> Result<(), Box<dyn std
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -475,7 +475,7 @@ async fn test_explicit_encryption_overrides_bucket_default() -> Result<(), Box<d
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -570,7 +570,7 @@ async fn test_sse_kms_without_key_id_populates_default() -> Result<(), Box<dyn s
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/common.rs

@@ -40,7 +40,7 @@ use std::time::Duration;
 use tokio::fs;
 use tokio::net::TcpStream;
 use tokio::time::sleep;
-use tracing::{debug, error, info, warn};
+use tracing::{debug, error, info};
 
 // KMS-specific constants
 pub const TEST_BUCKET: &str = "kms-test-bucket";
@@ -177,49 +177,6 @@ pub async fn get_kms_status(
     Ok(status)
 }
 
-/// Poll the KMS status endpoint until the backend reports ready or the timeout
-/// expires.  Replaces hard-coded `sleep(Duration::from_secs(3))` startup waits
-/// with an active readiness probe so tests start as soon as KMS is usable
-/// (typically < 1 s) instead of always waiting the full 3 s.
-///
-/// Uses exponential back-off starting at 200 ms (doubling each attempt, capped
-/// at 1 s) up to a total wall-clock budget of 5 s.
-pub async fn wait_for_kms_ready(
-    base_url: &str,
-    access_key: &str,
-    secret_key: &str,
-) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-    let total_deadline = Duration::from_secs(5);
-    let start = tokio::time::Instant::now();
-    let mut backoff = Duration::from_millis(200);
-    let max_backoff = Duration::from_secs(1);
-    let mut first_attempt = true;
-
-    loop {
-        if !first_attempt {
-            if start.elapsed() >= total_deadline {
-                return Err("KMS failed to become ready within 5 seconds".into());
-            }
-            sleep(backoff).await;
-            backoff = (backoff * 2).min(max_backoff);
-        }
-        first_attempt = false;
-
-        match get_kms_status(base_url, access_key, secret_key).await {
-            Ok(status) => {
-                info!("KMS is ready (status: {})", status);
-                return Ok(());
-            }
-            Err(e) => {
-                if start.elapsed() >= total_deadline {
-                    return Err(format!("KMS did not become ready within 5 s: last error: {e}").into());
-                }
-                warn!(error = %e, elapsed_ms = start.elapsed().as_millis() as u64, "KMS not ready yet, retrying…");
-            }
-        }
-    }
-}
-
 /// Create a default KMS key for testing and return the created key ID
 pub async fn create_default_key(
     base_url: &str,
@@ -904,13 +861,6 @@ impl LocalKMSTestEnvironment {
         Ok(default_key_id.to_string())
     }
 
-    /// Poll the KMS status endpoint until the backend reports ready.
-    ///
-    /// Prefer this over a fixed `sleep` after calling `start_rustfs_for_local_kms`.
-    pub async fn wait_for_kms_ready(&self) -> Result<(), Box<dyn std::error::Error + Send + Sync>> {
-        wait_for_kms_ready(&self.base_env.url, &self.base_env.access_key, &self.base_env.secret_key).await
-    }
-
     /// Configure Local KMS backend with a predefined default key
     pub async fn configure_local_kms(&self) -> Result<String, Box<dyn std::error::Error + Send + Sync>> {
         // Use a fixed, predictable default key ID

crates/e2e_test/src/kms/copy_object_self_copy_sse_test.rs

@@ -61,7 +61,7 @@ async fn test_metadata_replace_self_copy_of_sse_object_stays_decryptable() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let client = kms_env.base_env.create_s3_client();
     // Deliberately an UNVERSIONED bucket: that is the branch where the store layer can service
@@ -160,7 +160,7 @@ async fn test_metadata_replace_self_copy_dropping_sse_rewrites_plaintext() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let client = kms_env.base_env.create_s3_client();
     // Unversioned, and deliberately WITHOUT a bucket default-encryption rule, so the copy below
@@ -256,7 +256,7 @@ async fn test_metadata_replace_self_copy_under_bucket_default_sse_stays_decrypta
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let client = kms_env.base_env.create_s3_client();
     let bucket = "copy-object-self-copy-bucket-default-sse-test";

crates/e2e_test/src/kms/copy_object_version_restore_sse_test.rs

@@ -56,7 +56,7 @@ async fn test_self_copy_of_historical_sse_s3_version_is_readable() {
         )
         .await
         .expect("failed to start RustFS with local KMS");
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let client = kms_env.base_env.create_s3_client();
     let bucket = "copy-object-version-restore-sse-test";

crates/e2e_test/src/kms/encryption_metadata_test.rs

@@ -87,7 +87,7 @@ async fn test_head_reports_managed_metadata_for_sse_s3() -> Result<(), Box<dyn s
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -147,7 +147,7 @@ async fn test_head_reports_managed_metadata_for_sse_kms_and_copy() -> Result<(),
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -250,7 +250,7 @@ async fn test_multipart_upload_writes_encrypted_data() -> Result<(), Box<dyn std
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_comprehensive_test.rs

@@ -24,6 +24,7 @@ use super::common::{
     test_sse_kms_encryption, test_sse_s3_encryption,
 };
 use crate::common::{TEST_BUCKET, init_logging};
+use tokio::time::{Duration, sleep};
 use tracing::info;
 
 /// Comprehensive test: Full KMS workflow with all encryption types
@@ -34,7 +35,7 @@ async fn test_comprehensive_kms_full_workflow() -> Result<(), Box<dyn std::error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -102,7 +103,7 @@ async fn test_comprehensive_stress_test() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -136,7 +137,7 @@ async fn test_comprehensive_key_isolation() -> Result<(), Box<dyn std::error::Er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -207,7 +208,7 @@ async fn test_comprehensive_concurrent_operations() -> Result<(), Box<dyn std::e
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -252,7 +253,7 @@ async fn test_comprehensive_performance_benchmark() -> Result<(), Box<dyn std::e
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_edge_cases_test.rs

@@ -44,7 +44,7 @@ async fn test_kms_zero_byte_file_encryption() -> Result<(), Box<dyn std::error::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -117,7 +117,7 @@ async fn test_kms_single_byte_file_encryption() -> Result<(), Box<dyn std::error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -209,7 +209,7 @@ async fn test_kms_multipart_boundary_conditions() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -284,7 +284,7 @@ async fn test_kms_invalid_key_scenarios() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -371,7 +371,7 @@ async fn test_kms_concurrent_encryption() -> Result<(), Box<dyn std::error::Erro
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = Arc::new(kms_env.base_env.create_s3_client());
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -478,7 +478,7 @@ async fn test_kms_key_validation_security() -> Result<(), Box<dyn std::error::Er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_fault_recovery_test.rs

@@ -37,7 +37,7 @@ async fn test_kms_key_directory_unavailable() -> Result<(), Box<dyn std::error::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -127,7 +127,7 @@ async fn test_kms_corrupted_key_files() -> Result<(), Box<dyn std::error::Error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -218,7 +218,7 @@ async fn test_kms_multipart_upload_interruption() -> Result<(), Box<dyn std::err
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -401,7 +401,7 @@ async fn test_kms_resource_constraints() -> Result<(), Box<dyn std::error::Error
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/e2e_test/src/kms/kms_local_test.rs

@@ -46,7 +46,7 @@ async fn test_local_kms_end_to_end() -> Result<(), Box<dyn std::error::Error + S
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -127,7 +127,7 @@ async fn test_local_kms_key_isolation() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -227,7 +227,7 @@ async fn test_local_kms_large_file() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait a moment for RustFS to fully start up and initialize KMS
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 
@@ -309,7 +309,7 @@ async fn test_local_kms_multipart_upload() {
         .expect("Failed to start RustFS with Local KMS");
 
     // Wait for KMS initialization
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     info!("RustFS started with KMS auto-configuration, default_key_id: {}", default_key_id);
 

crates/e2e_test/src/kms/kms_vault_test.rs

@@ -20,6 +20,7 @@
 
 use crate::common::{TEST_BUCKET, init_logging};
 use serial_test::serial;
+use tokio::time::{Duration, sleep};
 use tracing::{error, info};
 
 use super::common::{
@@ -45,13 +46,8 @@ impl VaultKmsTestContext {
 
         start_kms(&env.base_env.url, &env.base_env.access_key, &env.base_env.secret_key).await?;
 
-        // Wait for KMS to finish initialising.
-        super::common::wait_for_kms_ready(
-            &env.base_env.url,
-            &env.base_env.access_key,
-            &env.base_env.secret_key,
-        )
-        .await?;
+        // Allow Vault to finish initialising token auth and transit engine.
+        sleep(Duration::from_secs(2)).await;
 
         Ok(Self { env })
     }

crates/e2e_test/src/kms/multipart_encryption_test.rs

@@ -33,7 +33,7 @@ async fn test_step1_basic_single_file_encryption() -> Result<(), Box<dyn std::er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -89,7 +89,7 @@ async fn test_step2_basic_multipart_upload_without_encryption() -> Result<(), Bo
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -187,7 +187,7 @@ async fn test_step3_multipart_upload_with_sse_s3() -> Result<(), Box<dyn std::er
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -310,7 +310,7 @@ async fn test_step4_large_multipart_upload_with_encryption() -> Result<(), Box<d
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;
@@ -435,7 +435,7 @@ async fn test_step5_all_encryption_types_multipart() -> Result<(), Box<dyn std::
 
     let mut kms_env = LocalKMSTestEnvironment::new().await?;
     let _default_key_id = kms_env.start_rustfs_for_local_kms().await?;
-    kms_env.wait_for_kms_ready().await?;
+    tokio::time::sleep(tokio::time::Duration::from_secs(3)).await;
 
     let s3_client = kms_env.base_env.create_s3_client();
     kms_env.base_env.create_test_bucket(TEST_BUCKET).await?;

crates/heal/src/heal/manager/tests.rs

@@ -110,7 +110,10 @@ impl HealStorageAPI for MockStorage {
         Ok(Vec::new())
     }
 
-    async fn get_bucket_info(&self, _bucket: &str) -> Result<Option<BucketInfo>> {
+    async fn get_bucket_info(&self, bucket: &str) -> Result<Option<BucketInfo>> {
+        if bucket == "panic" {
+            panic!("test-only panic payload must not escape the scheduler");
+        }
         Ok(None)
     }
 
@@ -1021,6 +1024,231 @@ async fn test_task_alias_is_removed_after_terminal_completion() {
     assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
 }
 
+#[tokio::test]
+#[serial_test::serial]
+async fn scheduler_panic_releases_active_slot_and_allows_same_target_readmission() {
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+    let request = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
+    let task_id = request.id.clone();
+
+    assert_eq!(
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("panic request should be admitted"),
+        HealAdmissionResult::Accepted
+    );
+    let duplicate = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
+    let duplicate_id = duplicate.id.clone();
+    assert_eq!(
+        manager
+            .submit_heal_request(duplicate)
+            .await
+            .expect("same target should merge while active is queued"),
+        HealAdmissionResult::Merged
+    );
+    assert_eq!(manager.canonical_task_id(&duplicate_id).await, task_id);
+    process_manager_queue_once(&manager).await;
+
+    let status = tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if let Ok(status) = manager.get_task_status(&task_id).await
+                && matches!(status, HealTaskStatus::Failed { .. })
+            {
+                break status;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("panic task should reach a terminal status");
+    assert_eq!(
+        status,
+        HealTaskStatus::Failed {
+            error: PANICKED_HEAL_TASK_ERROR.to_string()
+        }
+    );
+    assert_eq!(manager.get_active_task_count().await, 0);
+    assert_eq!(manager.get_queue_length().await, 0);
+    assert!(manager.retrying_heals.lock().await.is_empty());
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert!(manager.completed_heals.lock().await.contains_key(&task_id));
+    assert_eq!(manager.canonical_task_id(&duplicate_id).await, duplicate_id);
+
+    let readmitted = bucket_request("panic", HealPriority::Normal, HealRequestSource::Admin);
+    assert_eq!(
+        manager
+            .submit_heal_request(readmitted)
+            .await
+            .expect("same target should be re-admitted after a panic"),
+        HealAdmissionResult::Accepted
+    );
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn retry_child_panic_finishes_parent_once() {
+    clear_scheduler_panic();
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+    let task_id = request.id.clone();
+    assert_eq!(
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("retry request should be admitted"),
+        HealAdmissionResult::Accepted
+    );
+    arm_scheduler_panic(SchedulerPanicPoint::RetryChild, &task_id);
+    process_manager_queue_once(&manager).await;
+
+    let status = tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if let Ok(status) = manager.get_task_status(&task_id).await
+                && matches!(status, HealTaskStatus::Failed { .. })
+            {
+                break status;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("retry child panic should finish the parent");
+    clear_scheduler_panic();
+    assert_eq!(
+        status,
+        HealTaskStatus::Failed {
+            error: PANICKED_HEAL_TASK_ERROR.to_string()
+        }
+    );
+    assert_eq!(manager.get_active_task_count().await, 0);
+    assert_eq!(manager.get_queue_length().await, 0);
+    assert!(manager.retrying_heals.lock().await.is_empty());
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert_eq!(manager.completed_heals.lock().await.len(), 1);
+    assert_eq!(manager.get_statistics().await.failed_tasks, 1);
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cleanup_panic_is_supervised() {
+    clear_scheduler_panic();
+    let notice_bucket = "cleanup-panic-mrf";
+    let notice_object = "object";
+    let _ = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
+    let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);
+    let manager = HealManager::new(storage, None);
+    let mut request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
+    request.source = HealRequestSource::Admin;
+    let task_id = request.id.clone();
+    assert_eq!(
+        manager
+            .submit_heal_request(request)
+            .await
+            .expect("cleanup request should be admitted"),
+        HealAdmissionResult::Accepted
+    );
+    manager
+        .mrf_repair_notice_targets
+        .lock()
+        .expect("mrf repair notice registry poisoned")
+        .insert(
+            task_id.clone(),
+            vec![MrfRepairNoticeTarget {
+                bucket: Arc::from(notice_bucket),
+                object: Arc::from(notice_object),
+                version_id: None,
+            }],
+        );
+    arm_scheduler_panic(SchedulerPanicPoint::Cleanup, &task_id);
+    process_manager_queue_once(&manager).await;
+
+    let status = tokio::time::timeout(Duration::from_secs(1), async {
+        loop {
+            if let Ok(status) = manager.get_task_status(&task_id).await
+                && matches!(status, HealTaskStatus::Completed)
+            {
+                break status;
+            }
+            tokio::time::sleep(Duration::from_millis(10)).await;
+        }
+    })
+    .await
+    .expect("cleanup panic should leave a terminal status");
+    clear_scheduler_panic();
+    assert_eq!(status, HealTaskStatus::Completed);
+    assert_eq!(manager.get_active_task_count().await, 0);
+    assert!(manager.task_aliases.lock().await.is_empty());
+    assert_eq!(manager.completed_heals.lock().await.len(), 1);
+    assert_eq!(manager.get_statistics().await.successful_tasks, 1);
+    let events = rustfs_common::mrf_channel::take_mrf_repaired_events_for(notice_bucket);
+    assert_eq!(events.len(), 1, "cleanup panic must preserve successful MRF notice delivery");
+    assert_eq!(events[0].object.as_ref(), notice_object);
+}
+
+#[tokio::test]
+#[serial_test::serial]
+async fn cancelled_retry_child_panic_does_not_rearchive_failed_status() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+    let task_id = request.id.clone();
+    let retry_cancel_token = insert_retrying_request(&manager, request.clone()).await;
+
+    manager
+        .cancel_task(&task_id)
+        .await
+        .expect("retry cancellation should succeed");
+    assert!(retry_cancel_token.is_cancelled());
+
+    let state = PanicCleanupState {
+        active_heals: manager.active_heals.clone(),
+        heal_queue: manager.heal_queue.clone(),
+        completed_heals: manager.completed_heals.clone(),
+        task_aliases: manager.task_aliases.clone(),
+        retrying_heals: manager.retrying_heals.clone(),
+        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
+        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
+        statistics: manager.statistics.clone(),
+    };
+    finish_panicked_retry_child(task_id.clone(), request.heal_type, retry_cancel_token, state).await;
+
+    assert!(manager.retrying_heals.lock().await.is_empty());
+    assert!(manager.completed_heals.lock().await.is_empty());
+    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
+}
+
+#[tokio::test]
+async fn active_cancel_wins_parent_panic_cleanup_without_completed_status() {
+    let manager = HealManager::new(Arc::new(MockStorage), None);
+    let request = HealRequest::new(HealType::Cluster, HealOptions::default(), HealPriority::Normal);
+    let task_id = request.id.clone();
+    let task = Arc::new(HealTask::from_request(request, Arc::new(MockStorage)));
+    manager.active_heals.lock().await.insert(task_id.clone(), task.clone());
+
+    manager
+        .cancel_task(&task_id)
+        .await
+        .expect("active task cancellation should win");
+    assert_eq!(task.get_status().await, HealTaskStatus::Cancelled);
+
+    let state = PanicCleanupState {
+        active_heals: manager.active_heals.clone(),
+        heal_queue: manager.heal_queue.clone(),
+        completed_heals: manager.completed_heals.clone(),
+        task_aliases: manager.task_aliases.clone(),
+        retrying_heals: manager.retrying_heals.clone(),
+        mrf_repair_notice_targets: manager.mrf_repair_notice_targets.clone(),
+        replacement_recovery_anchors: manager.replacement_recovery_anchors.clone(),
+        statistics: manager.statistics.clone(),
+    };
+    finish_panicked_heal_task(task, task_id, state).await;
+
+    assert!(manager.completed_heals.lock().await.is_empty());
+    assert_eq!(manager.get_statistics().await.failed_tasks, 0);
+}
+
 #[tokio::test]
 async fn test_duplicate_admission_is_atomic_with_queue_to_active_transition() {
     let storage: Arc<dyn HealStorageAPI> = Arc::new(MockStorage);