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