test(heal): cover start retry and deadline outcome contracts (#7242)

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

@@ -14,6 +14,9 @@
 
 use super::*;
 use crate::heal::EcstoreError;
+use crate::heal::outcome::{
+    HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealObjectDisposition, HealTraversalCoverage,
+};
 use crate::heal::resume::{CheckpointManager, ReplacementTargetIdentity};
 use crate::heal::storage::{HealObjectInfo, HealStorageAPI};
 use crate::heal::task::{BatchHealFailure, HealOptions, HealPriority, HealRequest, HealTask, HealType};
@@ -515,10 +518,15 @@ impl HealStorageAPI for MockStorage {
     async fn heal_object(
         &self,
         bucket: &str,
-        _object: &str,
+        object: &str,
         _version_id: Option<&str>,
         _opts: &HealOpts,
     ) -> Result<(HealResultItem, Option<Error>)> {
+        if bucket.starts_with("heal-start-retry-deadline-object-") && object == "blocked" {
+            let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone();
+            hook.started.notify_one();
+            std::future::pending::<()>().await;
+        }
         if bucket == "completed-retention-failed" {
             return Err(Error::TaskExecutionFailed {
                 message: "retention fixture failure".to_string(),
@@ -580,11 +588,35 @@ impl HealStorageAPI for MockStorage {
 
     async fn list_objects_for_heal_page(
         &self,
-        _bucket: &str,
+        bucket: &str,
         _prefix: &str,
-        _continuation_token: Option<&str>,
+        continuation_token: Option<&str>,
         _include_lifecycle_object_info: bool,
     ) -> Result<(Vec<crate::heal::storage::HealListItem>, Option<String>, bool)> {
+        if bucket.starts_with("heal-start-retry-deadline-") {
+            if continuation_token.is_some() {
+                let hook = COMPLETED_RETENTION_HOOKS.lock().await[bucket].clone();
+                hook.started.notify_one();
+                std::future::pending::<()>().await;
+            }
+            let listing_timeout = bucket.starts_with("heal-start-retry-deadline-listing-");
+            let names = if listing_timeout {
+                vec!["completed"]
+            } else {
+                vec!["completed", "blocked"]
+            };
+            let objects = names
+                .into_iter()
+                .map(|name| crate::heal::storage::HealListItem {
+                    name: name.to_string(),
+                    version_id: None,
+                    mod_time_unix_nanos: None,
+                    lifecycle_object_info: None,
+                    is_delete_marker: false,
+                })
+                .collect();
+            return Ok((objects, listing_timeout.then(|| "next".to_string()), listing_timeout));
+        }
         Ok((Vec::new(), None, false))
     }
 
@@ -607,6 +639,161 @@ impl HealStorageAPI for MockStorage {
     }
 }
 
+async fn assert_heal_start_retry_control_preserves_real_executor_progress(cancel: bool) {
+    for phase in ["listing", "object"] {
+        let bucket = format!("heal-start-retry-deadline-{phase}-{cancel}");
+        let manager = HealManager::new(Arc::new(MockStorage), None);
+        let mut request = HealRequest::new(
+            HealType::Prefix {
+                bucket: bucket.clone(),
+                prefix: String::new(),
+            },
+            HealOptions {
+                timeout: Some(if cancel {
+                    Duration::from_secs(60)
+                } else {
+                    Duration::from_millis(200)
+                }),
+                ..Default::default()
+            },
+            HealPriority::High,
+        );
+        request.source = HealRequestSource::Admin;
+        let task_id = request.id.clone();
+        let hook = Arc::new(CompletedRetentionHook::default());
+        {
+            let mut hooks = COMPLETED_RETENTION_HOOKS.lock().await;
+            hooks.insert(bucket.clone(), Arc::clone(&hook));
+            hooks.insert(task_id.clone(), Arc::clone(&hook));
+        }
+        manager.submit_heal_request(request).await.expect("admit deadline task");
+        process_manager_queue_once(&manager).await;
+        tokio::time::timeout(Duration::from_secs(5), hook.started.notified())
+            .await
+            .expect("executor reaches blocked storage");
+        let active = manager.get_task_report(&task_id).await.expect("active report");
+        assert_eq!(active.progress.expect("real completed object progress").objects_healed, 1);
+        if cancel {
+            manager.active_heals.lock().await[&task_id].cancel_token.cancel();
+        }
+        tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
+            .await
+            .expect("deadline archives task");
+        let report = manager.get_task_report(&task_id).await.expect("terminal report");
+        assert_eq!(
+            report.status,
+            if cancel {
+                HealTaskStatus::Cancelled
+            } else {
+                HealTaskStatus::Timeout
+            },
+            "blocked {phase}"
+        );
+        let progress = report.progress.expect("terminal progress retained");
+        assert_eq!(progress.objects_healed, 1);
+        assert_eq!(progress.objects_failed, 0, "interrupted object has no terminal storage result");
+        assert_eq!(report.result_items.len(), 1, "completed result retained");
+        let outcome = report.outcome.expect("canonical terminal outcome retained");
+        assert_eq!(
+            outcome.execution,
+            HealExecutionOutcome::Aborted(if cancel {
+                HealAbortReason::Cancelled
+            } else {
+                HealAbortReason::Deadline
+            })
+        );
+        assert_eq!(outcome.coverage, HealTraversalCoverage::Partial);
+        assert_eq!(outcome.counters.healed, 0, "legacy success supplies no authoritative repair proof");
+        let completed = outcome
+            .objects
+            .iter()
+            .find(|item| item.identity.object == "completed")
+            .expect("completed object diagnostic retained");
+        assert_eq!(completed.disposition, HealObjectDisposition::Unknown);
+        if phase == "object" {
+            let interrupted = outcome
+                .objects
+                .iter()
+                .find(|item| item.identity.object == "blocked")
+                .expect("interrupted object diagnostic retained");
+            assert_eq!(
+                interrupted.disposition,
+                if cancel {
+                    HealObjectDisposition::Cancelled
+                } else {
+                    HealObjectDisposition::Deferred {
+                        reason: HealDeferredReason::Deadline,
+                        retry_not_before: None,
+                    }
+                }
+            );
+        } else {
+            assert_eq!(outcome.objects.len(), 1, "an unread page cannot supply object identities");
+        }
+        assert!(!manager.active_heals.lock().await.contains_key(&task_id));
+        assert!(!manager.retrying_heals.lock().await.contains_key(&task_id));
+        assert!(!manager.heal_queue.lock().await.contains_request_id(&task_id));
+        hook.finish.notify_one();
+        COMPLETED_RETENTION_HOOKS
+            .lock()
+            .await
+            .retain(|key, _| key != &bucket && key != &task_id);
+    }
+}
+
+#[tokio::test]
+async fn heal_start_retry_deadline_preserves_real_executor_progress() {
+    assert_heal_start_retry_control_preserves_real_executor_progress(false).await;
+}
+
+#[tokio::test]
+async fn heal_start_retry_cancellation_preserves_real_executor_progress() {
+    assert_heal_start_retry_control_preserves_real_executor_progress(true).await;
+}
+
+#[tokio::test]
+async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() {
+    let manager = HealManager::new(
+        Arc::new(MockStorage),
+        Some(HealConfig {
+            task_timeout: Duration::ZERO,
+            ..Default::default()
+        }),
+    );
+    let mut request = HealRequest::object("retry-transition".to_string(), "object".to_string(), None);
+    request.source = HealRequestSource::Admin;
+    request.options.timeout = Some(Duration::from_secs(60));
+    let task_id = request.id.clone();
+    let created_at = request.created_at;
+    let hook = Arc::new(CompletedRetentionHook::default());
+    COMPLETED_RETENTION_HOOKS
+        .lock()
+        .await
+        .insert(task_id.clone(), Arc::clone(&hook));
+    manager
+        .submit_heal_request(request)
+        .await
+        .expect("admit explicit-budget task");
+    process_manager_queue_once(&manager).await;
+    tokio::time::timeout(Duration::from_secs(5), hook.handoff.notified())
+        .await
+        .expect("real read-quorum failure prepares retry");
+    let retry = manager.retrying_heals.lock().await[&task_id].request.clone();
+    assert_eq!(retry.id, task_id);
+    assert_eq!(retry.created_at, created_at);
+    assert_eq!(retry.source, HealRequestSource::Admin);
+    assert_eq!(retry.retry_attempts, 1);
+    let remaining = retry.options.timeout.expect("retry retains explicit budget");
+    assert!(remaining > Duration::ZERO && remaining < Duration::from_secs(60));
+    assert!(matches!(
+        manager.get_task_status(&task_id).await.expect("retry remains queryable"),
+        HealTaskStatus::Retrying { retry_attempt: 1, .. }
+    ));
+    manager.cancel_task(&task_id).await.expect("cancel held retry");
+    hook.finish.notify_one();
+    COMPLETED_RETENTION_HOOKS.lock().await.remove(&task_id);
+}
+
 struct ManagerRecoveryTestHook {
     replacement_resume_disk: DiskStore,
     listed: StdMutex<bool>,

rustfs/src/admin/handlers/heal.rs

@@ -1636,6 +1636,44 @@ mod tests {
         assert!(executed.load(Ordering::SeqCst));
     }
 
+    #[tokio::test]
+    async fn heal_start_retry_preflight_failures_do_not_create_request_identities() {
+        let hip = HealInitParams {
+            bucket: "bucket".to_string(),
+            ..Default::default()
+        };
+        let mut request_ids = Vec::new();
+        for attempt in 0..3 {
+            let executed_ids = &mut request_ids;
+            let request_params = &hip;
+            let result = execute_after_heal_control_capability(
+                || async {
+                    if attempt < 2 {
+                        Err(super::cluster_heal_control_unavailable("test_capability_failure"))
+                    } else {
+                        Ok(())
+                    }
+                },
+                || async move {
+                    let request = build_heal_channel_request(request_params);
+                    executed_ids.push(request.id);
+                    Ok(())
+                },
+            )
+            .await;
+            if attempt < 2 {
+                assert!(result.is_err(), "failed capability checks must not start a heal");
+                assert!(
+                    request_ids.is_empty(),
+                    "preflight failure must precede request construction and admission"
+                );
+            } else {
+                result.expect("restored capabilities allow the first execution");
+                assert_eq!(request_ids.len(), 1);
+            }
+        }
+    }
+
     #[test]
     fn replacement_recovery_status_response_reports_cluster_proof() {
         let local = replacement_snapshot("11111111-1111-4111-8111-111111111111");
@@ -1743,6 +1781,21 @@ mod tests {
         assert!(decoded.is_none());
     }
 
+    #[test]
+    fn heal_start_retry_conflicts_keep_actionable_public_reasons() {
+        for (reason, label) in [
+            (HealAdmissionDropReason::AlreadyRunning, "already_running"),
+            (HealAdmissionDropReason::OverlappingPaths, "overlapping_paths"),
+        ] {
+            let error = reject_heal_admission(HealAdmissionResult::Dropped(reason));
+            assert_eq!(error.code(), &S3ErrorCode::OperationAborted);
+            assert!(
+                error.to_string().contains(label),
+                "the caller must distinguish conflicts from transient coordination failure"
+            );
+        }
+    }
+
     #[test]
     fn test_reject_heal_admission_preserves_retry_semantics() {
         for admission in [

rustfs/src/storage/rpc/node_service.rs

@@ -2930,6 +2930,136 @@ mod tests {
         assert_eq!(err.code(), tonic::Code::InvalidArgument);
     }
 
+    fn heal_start_retry_fixture() -> (
+        Arc<HealManager>,
+        rustfs_heal_contracts::heal_channel::HealChannelRequest,
+        rustfs_protos::heal_control::RequestMetadata,
+    ) {
+        let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));
+        let mut request = rustfs_heal_contracts::heal_channel::create_heal_request(
+            "bucket".to_string(),
+            Some("prefix".to_string()),
+            true,
+            None,
+        );
+        request.source = rustfs_heal_contracts::heal_channel::HealRequestSource::Admin;
+        request.recursive = Some(true);
+        let now = i64::try_from(OffsetDateTime::now_utc().unix_timestamp_nanos() / 1_000_000).expect("fixture clock fits in i64");
+        let metadata = rustfs_protos::heal_control::RequestMetadata::new(*Uuid::new_v4().as_bytes(), now, now + 30_000, 7);
+        (manager, request, metadata)
+    }
+
+    #[tokio::test]
+    async fn heal_start_retry_exact_forced_envelope_returns_cached_admission() {
+        let (manager, request, metadata) = heal_start_retry_fixture();
+        let request_id = request.id.clone();
+        let envelope = rustfs_protos::heal_control::Envelope::start(request, metadata).expect("valid forced start");
+        let lost_response =
+            execute_heal_control_envelope_with_manager(envelope.clone(), metadata.coordinator_epoch, Some(manager.clone()))
+                .await
+                .expect("first request is admitted before its response is lost");
+        assert_eq!(manager.operations_snapshot().await.queue_length, 1);
+
+        // The caller sees no first response, but retries the original envelope.
+        let replayed = execute_heal_control_envelope_with_manager(envelope, metadata.coordinator_epoch, Some(manager.clone()))
+            .await
+            .expect("an exact envelope replay must recover its receipt");
+        assert_eq!(replayed, lost_response);
+        assert_eq!(
+            manager.operations_snapshot().await.queue_length,
+            1,
+            "forceStart must not be executed twice"
+        );
+        let outcome = rustfs_protos::heal_control::decode_result(&replayed)
+            .and_then(|result| result.into_outcome(&request_id, metadata.coordinator_epoch))
+            .expect("matching canonical receipt");
+        assert!(matches!(outcome, rustfs_protos::heal_control::Outcome::Start {
+            task_id, admission: rustfs_protos::heal_control::Admission::Accepted,
+        } if task_id == request_id));
+    }
+
+    #[tokio::test]
+    async fn heal_start_retry_new_forced_request_is_a_distinct_start() {
+        let (manager, request, metadata) = heal_start_retry_fixture();
+        let first_id = request.id.clone();
+        let first = rustfs_protos::heal_control::Envelope::start(request.clone(), metadata).expect("first start");
+        let _lost_response = execute_heal_control_envelope_with_manager(first, metadata.coordinator_epoch, Some(manager.clone()))
+            .await
+            .expect("first admission");
+
+        // A fresh HTTP forceStart request intentionally requests another start.
+        let mut next_request = request;
+        next_request.id = Uuid::new_v4().to_string();
+        let next_id = next_request.id.clone();
+        let next_metadata = rustfs_protos::heal_control::RequestMetadata {
+            nonce: *Uuid::new_v4().as_bytes(),
+            ..metadata
+        };
+        let next = rustfs_protos::heal_control::Envelope::start(next_request, next_metadata).expect("new forced start");
+        let response = execute_heal_control_envelope_with_manager(next, metadata.coordinator_epoch, Some(manager.clone()))
+            .await
+            .expect("forceStart preserves its explicit admission semantics");
+        let outcome = rustfs_protos::heal_control::decode_result(&response)
+            .and_then(|result| result.into_outcome(&next_id, metadata.coordinator_epoch))
+            .expect("new receipt");
+        assert!(matches!(outcome, rustfs_protos::heal_control::Outcome::Start {
+            task_id, admission: rustfs_protos::heal_control::Admission::Accepted,
+        } if task_id == next_id && task_id != first_id));
+        assert_eq!(
+            manager.operations_snapshot().await.queue_length,
+            2,
+            "a caller must not treat a new forced request as an idempotent transport retry"
+        );
+    }
+
+    #[tokio::test]
+    async fn heal_start_retry_same_id_with_changed_envelope_conflicts_before_admission() {
+        let (manager, request, metadata) = heal_start_retry_fixture();
+        let original = rustfs_protos::heal_control::Envelope::start(request.clone(), metadata).expect("original start");
+        let receipt =
+            execute_heal_control_envelope_with_manager(original.clone(), metadata.coordinator_epoch, Some(manager.clone()))
+                .await
+                .expect("original admission");
+        let mut changed_options = request.clone();
+        changed_options.remove_corrupted = Some(true);
+        let changed_metadata = rustfs_protos::heal_control::RequestMetadata {
+            nonce: *Uuid::new_v4().as_bytes(),
+            ..metadata
+        };
+        for changed in [
+            rustfs_protos::heal_control::Envelope::start(changed_options, metadata).expect("changed options"),
+            rustfs_protos::heal_control::Envelope::start(request, changed_metadata).expect("changed nonce"),
+        ] {
+            let error = execute_heal_control_envelope_with_manager(changed, metadata.coordinator_epoch, Some(manager.clone()))
+                .await
+                .expect_err("one request ID cannot identify different envelope bytes");
+            assert_eq!(error.code(), tonic::Code::AlreadyExists);
+            assert_eq!(manager.operations_snapshot().await.queue_length, 1);
+        }
+        assert_eq!(
+            execute_heal_control_envelope_with_manager(original, metadata.coordinator_epoch, Some(manager))
+                .await
+                .expect("conflicts must preserve the original receipt"),
+            receipt
+        );
+    }
+
+    #[tokio::test]
+    async fn heal_start_retry_wrong_coordinator_epoch_cannot_admit_locally() {
+        let (manager, request, metadata) = heal_start_retry_fixture();
+        let request_id = request.id.clone();
+        let envelope = rustfs_protos::heal_control::Envelope::start(request, metadata).expect("start envelope");
+        let error = execute_heal_control_envelope_with_manager(envelope, metadata.coordinator_epoch + 1, Some(manager.clone()))
+            .await
+            .expect_err("a different coordinator epoch cannot accept the request");
+        assert_eq!(error.code(), tonic::Code::FailedPrecondition);
+        assert_eq!(manager.operations_snapshot().await.queue_length, 0);
+        assert!(matches!(
+            manager.get_task_status(&request_id).await,
+            Err(rustfs_heal::Error::TaskNotFound { .. })
+        ));
+    }
+
     #[tokio::test]
     async fn heal_control_executor_preserves_canonical_token_and_drops_query_results() {
         let manager = Arc::new(HealManager::new(Arc::new(HealControlMockStorage), None));