diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index d8b3e3b16..17feee1b5 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/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)> { + 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, Option, 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, diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index 74a2da980..7955a80a7 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/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 [ diff --git a/rustfs/src/storage/rpc/node_service.rs b/rustfs/src/storage/rpc/node_service.rs index 7278991c5..25b200fb2 100644 --- a/rustfs/src/storage/rpc/node_service.rs +++ b/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, + 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));