diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index d212a5cf5..b9802655b 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -196,6 +196,20 @@ impl HealChannelProcessor { .map_err(Error::other) } + /// Query a local automatic task; other tokens require coordinator routing. + pub async fn execute_local_automatic_query( + &self, + heal_path: &str, + client_token: &str, + since_seq: Option, + ) -> Result> { + self.heal_manager + .get_local_automatic_task_report(heal_path, client_token, since_seq) + .await? + .map(|report| Self::query_report_response(client_token.to_owned(), Ok(report))) + .transpose() + } + /// Execute cancellation directly against the manager. pub async fn execute_cancel_request(&self, heal_path: String, client_token: String) -> Result { let (response_tx, response_rx) = oneshot::channel(); @@ -463,6 +477,13 @@ impl HealChannelProcessor { .await }; + let response = Self::query_report_response(client_token, report)?; + let _ = response_tx.send(Ok(response.clone())); + self.publish_response(response); + Ok(()) + } + + fn query_report_response(client_token: String, report: Result) -> Result { let outcome = report.as_ref().ok().and_then(|report| report.outcome.clone()); let settings = report .as_ref() @@ -581,27 +602,21 @@ impl HealChannelProcessor { 0, ), Err(crate::Error::InvalidClientToken) => { - let response = HealChannelResponse { + return Ok(HealChannelResponse { request_id: client_token, success: false, data: None, error: Some("invalid heal client token".to_string()), - }; - let _ = response_tx.send(Ok(response.clone())); - self.publish_response(response); - return Ok(()); + }); } Err(err) => { let error_text = err.to_string(); - let response = HealChannelResponse { + return Ok(HealChannelResponse { request_id: client_token, success: false, data: None, error: Some(error_text), - }; - let _ = response_tx.send(Ok(response.clone())); - self.publish_response(response); - return Ok(()); + }); } }; @@ -618,17 +633,12 @@ impl HealChannelProcessor { }, )?; - let response = HealChannelResponse { + Ok(HealChannelResponse { request_id: client_token, success: true, data: Some(data), error: detail, - }; - - let _ = response_tx.send(Ok(response.clone())); - self.publish_response(response); - - Ok(()) + }) } /// Process cancel request diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index dd17e7c87..7899ae3d9 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -234,6 +234,8 @@ fn record_displaced_terminal( request: &HealRequest, ) -> Arc { let terminal = Arc::new(CompletedHealStatus { + task_id: request.id.clone(), + source: request.source, outcome: None, progress: None, retained_bytes: std::sync::OnceLock::new(), @@ -1018,9 +1020,9 @@ pub struct HealManager { /// cascade without re-locking. enum TaskStateLookup { Active(Arc), - Retrying(HealTaskStatus, HealOptions), + Retrying(HealTaskStatus, HealOptions, HealRequestSource), Completed(Arc), - Queued(HealOptions), + Queued(HealOptions, HealRequestSource), /// A durable root heal record that cannot be admitted or quarantined; /// it answers status queries with the reason until an operator repairs it. Unrunnable(String), @@ -2433,6 +2435,23 @@ impl HealManager { /// completed entry. `heal_path` additionally constrains the map matches /// the way the `*_for_path` variants always have. async fn lookup_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> Result { + let state = self.lookup_memory_task_state(canonical_task_id, heal_path).await; + if !matches!(state, TaskStateLookup::NotFound) { + return Ok(state); + } + if let Some(completed) = self.root_recovery.completed(canonical_task_id).await? + && heal_path.is_none_or(|path| heal_type_matches_path(&completed.heal_type, path)) + { + return Ok(TaskStateLookup::Completed(Arc::new(completed))); + } + // A durable record that cannot run still answers with its reason. + if let Some(reason) = self.root_recovery.record_issue(canonical_task_id).await { + return Ok(TaskStateLookup::Unrunnable(reason)); + } + Ok(TaskStateLookup::NotFound) + } + + async fn lookup_memory_task_state(&self, canonical_task_id: &str, heal_path: Option<&str>) -> TaskStateLookup { let matches_path = |heal_type: &HealType| heal_path.is_none_or(|path| heal_type_matches_path(heal_type, path)); { @@ -2441,7 +2460,7 @@ impl HealManager { .get(canonical_task_id) .filter(|task| matches_path(&task.heal_type)) { - return Ok(TaskStateLookup::Active(Arc::clone(task))); + return TaskStateLookup::Active(Arc::clone(task)); } } @@ -2451,7 +2470,7 @@ impl HealManager { .get(canonical_task_id) .filter(|retrying| matches_path(&retrying.request.heal_type)) { - return Ok(TaskStateLookup::Retrying(retrying.status(), retrying.request.options.clone())); + return TaskStateLookup::Retrying(retrying.status(), retrying.request.options.clone(), retrying.request.source); } } @@ -2464,7 +2483,7 @@ impl HealManager { prune_completed_heal_statuses(&mut completed_heals); if let Some(completed) = completed_heals.get(canonical_task_id).filter(|c| matches_path(&c.heal_type)) { if completed_status_is_retrying(&completed.status) { - return Ok(TaskStateLookup::Completed(Arc::clone(completed))); + return TaskStateLookup::Completed(Arc::clone(completed)); } terminal_completed = Some(Arc::clone(completed)); } @@ -2473,7 +2492,7 @@ impl HealManager { { let queue = self.heal_queue.lock().await; if let Some(request) = queue.request_matching_id_and_path(canonical_task_id, heal_path) { - return Ok(TaskStateLookup::Queued(request.options.clone())); + return TaskStateLookup::Queued(request.options.clone(), request.source); } } @@ -2486,22 +2505,10 @@ impl HealManager { .cloned(); } - if terminal_completed.is_none() - && let Some(completed) = self.root_recovery.completed(canonical_task_id).await? - && matches_path(&completed.heal_type) - { - terminal_completed = Some(Arc::new(completed)); - } - if let Some(completed) = terminal_completed { - return Ok(TaskStateLookup::Completed(completed)); + return TaskStateLookup::Completed(completed); } - // A durable root record that can neither run nor be quarantined still - // answers status queries with its reason. - if let Some(reason) = self.root_recovery.record_issue(canonical_task_id).await { - return Ok(TaskStateLookup::Unrunnable(reason)); - } - Ok(TaskStateLookup::NotFound) + TaskStateLookup::NotFound } async fn publish_admin_terminal( @@ -2525,6 +2532,8 @@ impl HealManager { ) -> Result { let previous = self.completed_heals.lock().await.get(task_id).cloned(); let mut completed = CompletedHealStatus { + task_id: task_id.to_owned(), + source, outcome: None, progress: None, retained_bytes: std::sync::OnceLock::new(), @@ -2560,9 +2569,9 @@ impl HealManager { let canonical_task_id = self.canonical_task_id(task_id).await; match self.lookup_task_state(&canonical_task_id, None).await? { TaskStateLookup::Active(task) => Ok(task.get_status().await), - TaskStateLookup::Retrying(status, _) => Ok(status), + TaskStateLookup::Retrying(status, _, _) => Ok(status), TaskStateLookup::Completed(completed) => Ok(completed.status.clone()), - TaskStateLookup::Queued(_) => Ok(HealTaskStatus::Pending), + TaskStateLookup::Queued(_, _) => Ok(HealTaskStatus::Pending), TaskStateLookup::Unrunnable(error) => Ok(HealTaskStatus::Failed { error }), TaskStateLookup::NotFound => Err(Error::TaskNotFound { task_id: task_id.to_string(), @@ -2581,9 +2590,9 @@ impl HealManager { let canonical_task_id = self.canonical_task_id(task_id).await; match self.lookup_task_state(&canonical_task_id, None).await? { TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await), - TaskStateLookup::Retrying(status, options) => Ok(empty_task_report_with_options(status, options)), + TaskStateLookup::Retrying(status, options, _) => Ok(empty_task_report_with_options(status, options)), TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)), - TaskStateLookup::Queued(options) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)), + TaskStateLookup::Queued(options, _) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)), TaskStateLookup::Unrunnable(error) => { Ok(empty_task_report_with_options(HealTaskStatus::Failed { error }, HealOptions::default())) } @@ -2597,6 +2606,42 @@ impl HealManager { self.get_task_report_for_path_since(heal_path, task_id, None).await } + /// Resolve a node-owned automatic task without granting local control over + /// Admin tasks. The source and report come from the same state snapshot. + pub async fn get_local_automatic_task_report( + &self, + heal_path: &str, + task_id: &str, + since: Option, + ) -> Result> { + let canonical_task_id = self.canonical_task_id(task_id).await; + if canonical_task_id != task_id { + return Ok(None); + } + let state = self.lookup_memory_task_state(&canonical_task_id, Some(heal_path)).await; + let source = match &state { + TaskStateLookup::Active(task) => task.source, + TaskStateLookup::Retrying(_, _, source) | TaskStateLookup::Queued(_, source) => *source, + TaskStateLookup::Completed(completed) if completed.task_id == task_id => completed.source, + TaskStateLookup::Completed(_) => return Ok(None), + TaskStateLookup::Unrunnable(_) | TaskStateLookup::NotFound => return Ok(None), + }; + if !matches!( + source, + HealRequestSource::Scanner | HealRequestSource::Mrf | HealRequestSource::AutoHeal | HealRequestSource::ReadRepair + ) { + return Ok(None); + } + let report = match state { + TaskStateLookup::Active(task) => active_task_report(&task, since).await, + TaskStateLookup::Retrying(status, options, _) => empty_task_report_with_options(status, options), + TaskStateLookup::Completed(completed) => completed_task_report(&completed, since), + TaskStateLookup::Queued(options, _) => empty_task_report_with_options(HealTaskStatus::Pending, options), + TaskStateLookup::Unrunnable(_) | TaskStateLookup::NotFound => return Ok(None), + }; + Ok(Some(report)) + } + /// Incremental variant of [`Self::get_task_report_for_path`] (HS-06). pub async fn get_task_report_for_path_since( &self, @@ -2607,9 +2652,9 @@ impl HealManager { let canonical_task_id = self.canonical_task_id(task_id).await; match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? { TaskStateLookup::Active(task) => Ok(active_task_report(&task, since).await), - TaskStateLookup::Retrying(status, options) => Ok(empty_task_report_with_options(status, options)), + TaskStateLookup::Retrying(status, options, _) => Ok(empty_task_report_with_options(status, options)), TaskStateLookup::Completed(completed) => Ok(completed_task_report(&completed, since)), - TaskStateLookup::Queued(options) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)), + TaskStateLookup::Queued(options, _) => Ok(empty_task_report_with_options(HealTaskStatus::Pending, options)), TaskStateLookup::Unrunnable(error) => { Ok(empty_task_report_with_options(HealTaskStatus::Failed { error }, HealOptions::default())) } @@ -2633,9 +2678,9 @@ impl HealManager { let canonical_task_id = self.canonical_task_id(task_id).await; match self.lookup_task_state(&canonical_task_id, Some(heal_path)).await? { TaskStateLookup::Active(task) => Ok(task.get_status().await), - TaskStateLookup::Retrying(status, _) => Ok(status), + TaskStateLookup::Retrying(status, _, _) => Ok(status), TaskStateLookup::Completed(completed) => Ok(completed.status.clone()), - TaskStateLookup::Queued(_) => Ok(HealTaskStatus::Pending), + TaskStateLookup::Queued(_, _) => Ok(HealTaskStatus::Pending), TaskStateLookup::Unrunnable(error) => Ok(HealTaskStatus::Failed { error }), TaskStateLookup::NotFound => { if self.path_has_task(heal_path).await { diff --git a/crates/heal/src/heal/manager/queue.rs b/crates/heal/src/heal/manager/queue.rs index fa0da6cd1..c3c0d6adc 100644 --- a/crates/heal/src/heal/manager/queue.rs +++ b/crates/heal/src/heal/manager/queue.rs @@ -107,7 +107,9 @@ pub(super) enum QueuePushOutcome { #[derive(Debug, Clone)] pub(super) struct CompletedHealStatus { + pub(super) task_id: String, pub(super) heal_type: HealType, + pub(super) source: HealRequestSource, /// Options used to execute the task, retained for token-scoped status. pub(super) options: HealOptions, pub(super) status: HealTaskStatus, @@ -135,6 +137,7 @@ impl CompletedHealStatus { fn measure_retained_bytes(&self) -> usize { let mut bytes = size_of::(); let mut add = |amount: usize| bytes = bytes.saturating_add(amount); + add(self.task_id.capacity()); add(self.outcome.as_ref().map_or(0, |outcome| outcome.retained_bytes())); match &self.heal_type { HealType::Cluster => {} @@ -256,7 +259,9 @@ impl CompletedHealStatus { outcome.finish(Some(crate::heal::outcome::HealAbortReason::Cancelled)); } let mut snapshot = Self { + task_id: task.id.clone(), heal_type: task.heal_type.clone(), + source: task.source, options: task.options.clone(), status, progress: Some(task.get_progress().await), diff --git a/crates/heal/src/heal/manager/root_recovery.rs b/crates/heal/src/heal/manager/root_recovery.rs index 158064dd4..8c12a513d 100644 --- a/crates/heal/src/heal/manager/root_recovery.rs +++ b/crates/heal/src/heal/manager/root_recovery.rs @@ -459,6 +459,8 @@ impl RootHealTerminal { fn into_completed(self) -> CompletedHealStatus { CompletedHealStatus { + task_id: self.task_id, + source: HealRequestSource::Admin, outcome: None, progress: self.progress, retained_bytes: std::sync::OnceLock::new(), diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index c97340b90..a54f7f7d8 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -115,6 +115,8 @@ struct MockStorage; fn completed_retention_fixture(completed_at: SystemTime) -> CompletedHealStatus { CompletedHealStatus { + task_id: "fixture-token".to_string(), + source: HealRequestSource::Internal, outcome: None, heal_type: HealType::Cluster, options: HealOptions::default(), @@ -2636,6 +2638,181 @@ async fn test_active_duplicate_token_can_query_and_cancel_original_task() { ); } +#[tokio::test] +async fn local_automatic_query_preserves_source_path_token_and_state() { + use crate::heal::channel::HealChannelProcessor; + use crate::heal::outcome::{ + HealExecutionOutcome, HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealObjectOutcome, HealTaskOutcome, + HealTraversalCoverage, + }; + + for source in [ + HealRequestSource::Scanner, + HealRequestSource::Mrf, + HealRequestSource::AutoHeal, + HealRequestSource::ReadRepair, + HealRequestSource::Admin, + HealRequestSource::Internal, + ] { + for state in ["queued", "active", "retrying", "completed", "displaced"] { + let storage: Arc = Arc::new(MockStorage); + let manager = Arc::new(HealManager::new_without_root_recovery_for_test(storage.clone(), None)); + let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None); + request.source = source; + let token = request.id.clone(); + let task = Arc::new(HealTask::from_request(request.clone(), storage)); + let mut snapshot = CompletedHealStatus::snapshot(&task, HealTaskStatus::Completed).await; + assert_eq!(snapshot.source, source, "completion must retain actual task source"); + let mut outcome = HealTaskOutcome::default(); + outcome.execution = HealExecutionOutcome::Completed; + outcome.coverage = HealTraversalCoverage::Complete; + outcome.record(HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: "bucket".to_string(), + object: "object".to_string(), + version_id: None, + bucket_incarnation_id: Some(uuid::Uuid::nil()), + pool_index: Some(0), + set_index: Some(0), + }, + disposition: HealObjectDisposition::VerifiedHealthy, + detail: None, + }); + snapshot.outcome = Some(Arc::new(outcome)); + snapshot.seqed_items = vec![( + 3, + HealResultItem { + bucket: "bucket".to_string(), + object: "object".to_string(), + ..Default::default() + }, + )]; + snapshot.min_seq = 3; + snapshot.next_seq = 4; + let expected_status = match state { + "queued" => { + manager.heal_queue.lock().await.push(request.clone()); + HealTaskStatus::Pending + } + "active" => { + manager.active_heals.lock().await.insert(token.clone(), task); + HealTaskStatus::Pending + } + "retrying" => { + manager.retrying_heals.lock().await.insert( + token.clone(), + RetryingHeal { + request: request.clone(), + error: "transient".to_string(), + cancel_token: CancellationToken::new(), + }, + ); + HealTaskStatus::Retrying { + error: "transient".to_string(), + retry_attempt: request.retry_attempts, + } + } + "completed" => { + manager.completed_heals.lock().await.insert(token.clone(), Arc::new(snapshot)); + HealTaskStatus::Completed + } + "displaced" => { + let terminal = record_displaced_terminal(&manager.displaced_terminals, &request); + terminal.status.clone() + } + _ => unreachable!("fixed test states"), + }; + let automatic = !matches!(source, HealRequestSource::Admin | HealRequestSource::Internal); + let report = manager + .get_local_automatic_task_report("bucket/object", &token, Some(2)) + .await + .expect("local source lookup"); + assert_eq!(report.is_some(), automatic, "source={source:?} state={state}"); + if let Some(report) = report { + assert_eq!(report.status, expected_status); + if state == "completed" { + let outcome = report.outcome.as_ref().expect("retained structured outcome"); + assert_eq!(outcome.objects.len(), 1); + assert_eq!(outcome.objects[0].identity.object, "object"); + assert_eq!(outcome.objects[0].identity.pool_index, Some(0)); + assert_eq!(outcome.objects[0].identity.set_index, Some(0)); + assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::VerifiedHealthy); + assert_eq!(report.min_seq, 3); + assert_eq!(report.next_seq, 4); + assert_eq!(report.result_items.len(), 1); + assert_eq!(report.result_items[0].object, "object"); + } + let processor = HealChannelProcessor::new(manager.clone()); + let local = processor + .execute_local_automatic_query("bucket/object", &token, Some(2)) + .await + .expect("local response") + .expect("automatic task response"); + let direct = processor + .execute_query_request_since("bucket/object".to_string(), token.clone(), Some(2)) + .await + .expect("original query response"); + assert!(local.success); + assert_eq!(local.request_id, direct.request_id); + assert_eq!(local.data, direct.data, "same report encoding and cursor contract"); + assert_eq!(local.error, direct.error); + let coordinator = Arc::new(HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None)); + let missing = HealChannelProcessor::new(coordinator) + .execute_query_request("bucket/object".to_string(), token.clone()) + .await + .expect("coordinator query"); + let body: serde_json::Value = + serde_json::from_slice(missing.data.as_deref().expect("coordinator response")).expect("coordinator JSON"); + assert_eq!(body["summary"], "notFound", "a remote manager does not own the automatic task"); + if state == "completed" { + let completed = manager + .completed_heals + .lock() + .await + .get(&token) + .cloned() + .expect("original completion"); + manager + .completed_heals + .lock() + .await + .insert("retired-alias".to_string(), completed); + assert!( + manager + .get_local_automatic_task_report("bucket/object", "retired-alias", None) + .await + .expect("retired alias lookup") + .is_none() + ); + } + } + assert!( + manager + .get_local_automatic_task_report("bucket/other", &token, None) + .await + .expect("wrong path lookup") + .is_none() + ); + assert!( + manager + .get_local_automatic_task_report("bucket/object", "unknown", None) + .await + .expect("unknown token lookup") + .is_none() + ); + manager.insert_task_alias("admin-alias", &token).await; + assert!( + manager + .get_local_automatic_task_report("bucket/object", "admin-alias", None) + .await + .expect("alias lookup") + .is_none() + ); + } + } +} + #[tokio::test] async fn test_queued_duplicate_token_can_query_and_cancel_original_request() { let storage: Arc = Arc::new(MockStorage); @@ -3565,6 +3742,8 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) -> Arc::new(CompletedHealStatus { progress: None, outcome: None, + source: request.source, + task_id: request.id.clone(), retained_bytes: std::sync::OnceLock::new(), heal_type: request.heal_type, options: request.options.clone(), @@ -4413,6 +4592,8 @@ async fn test_retrying_completion_outranks_the_queue_for_the_same_id() { Arc::new(CompletedHealStatus { progress: None, outcome: None, + source: request.source, + task_id: request.id.clone(), retained_bytes: std::sync::OnceLock::new(), heal_type: request.heal_type.clone(), options: request.options.clone(), @@ -4452,6 +4633,8 @@ async fn test_get_task_status_reads_recent_completed_status() { Arc::new(CompletedHealStatus { progress: None, outcome: None, + source: HealRequestSource::Internal, + task_id: "completed-token".to_string(), retained_bytes: std::sync::OnceLock::new(), heal_type: HealType::Bucket { bucket: "bucket".to_string(), @@ -4485,6 +4668,8 @@ async fn test_get_task_report_for_path_reads_completed_items() { Arc::new(CompletedHealStatus { progress: None, outcome: None, + source: HealRequestSource::Internal, + task_id: "completed-token".to_string(), retained_bytes: std::sync::OnceLock::new(), heal_type: HealType::Object { bucket: "bucket".to_string(), diff --git a/crates/heal/src/heal/manager/tests/root_recovery.rs b/crates/heal/src/heal/manager/tests/root_recovery.rs index 837260b58..5aa75bb15 100644 --- a/crates/heal/src/heal/manager/tests/root_recovery.rs +++ b/crates/heal/src/heal/manager/tests/root_recovery.rs @@ -115,6 +115,8 @@ fn admin_request(heal_type: HealType) -> HealRequest { fn completed_admin_status(heal_type: &HealType, completed_at: SystemTime) -> CompletedHealStatus { CompletedHealStatus { + task_id: "admin-fixture".to_string(), + source: HealRequestSource::Admin, outcome: None, heal_type: heal_type.clone(), options: HealOptions::default(), @@ -1237,6 +1239,8 @@ async fn root_recovery_completed_non_root_admin_is_queryable_after_restart() { .expect("durable bucket responsibility"); let completed = CompletedHealStatus { outcome: None, + source: request.source, + task_id: request.id.clone(), heal_type: request.heal_type.clone(), options: request.options.clone(), status: HealTaskStatus::Completed, diff --git a/docs/architecture/heal-terminal-reports.md b/docs/architecture/heal-terminal-reports.md index 988f2809a..ce327cf7e 100644 --- a/docs/architecture/heal-terminal-reports.md +++ b/docs/architecture/heal-terminal-reports.md @@ -4,6 +4,10 @@ Admin heal tokens retain their terminal status for ten minutes from the original ## Publication and Recovery +Authenticated token queries first inspect the receiving node's in-memory automatic tasks. An original Scanner, MRF, automatic-Heal, or read-repair token can return its path-bound snapshot on that node, including the retained outcome and incremental cursor. The task source and original ID remain attached to the completion snapshot; aliases cannot enter this local query path. The local lookup never reads administrator recovery records. + +Administrator tokens and unknown local identities continue through the cluster coordinator. Starting and cancelling tasks retain their coordinator route. Querying another node does not discover or aggregate a remote automatic task, and this path does not extend automatic-task retention across process restart. + The original schema-1 `terminal-root-heal-.json` remains the commit marker. Its separate `heal-terminal-report-.json` report has its own schema version and embeds the exact terminal identity. Both files belong to the same coordinator disk as the pending intent. Publication writes the report through the storage owner's conditional file update, writes the terminal marker, and then conditionally removes the pending intent. Recovery never treats a report without its marker as committed. A committed terminal continues to suppress stale pending work even when its report is unreadable. An initial publication failure leaves the existing responsibility available for recovery. diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index f0ef8c491..dc1fec710 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -1051,6 +1051,21 @@ async fn route_cluster_heal_control( }) } +async fn query_heal_status_on_owner( + local: L, + coordinator: C, +) -> S3Result +where + L: Future>>, + C: FnOnce() -> CF, + CF: Future>, +{ + match local.await? { + Some(response) => Ok(response), + None => coordinator().await, + } +} + async fn execute_after_heal_start_preflight( endpoints: &EndpointServerPools, options: &HealOpts, @@ -1511,20 +1526,41 @@ impl Operation for HealHandler { let heal_path = encode_heal_control_path(&hip.bucket, &hip.obj_prefix); if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop { let client_token = hip.client_token.clone(); - let request_id = uuid::Uuid::new_v4().to_string(); - let context = app_context - .clone() - .ok_or_else(|| cluster_heal_control_unavailable("app_context_unavailable"))?; - let route = prepare_heal_control_route(&context)?; - let envelope = rustfs_protos::heal_control::Envelope::query( - request_id.clone(), - new_heal_control_metadata(&route)?, - heal_path, - client_token.clone(), - hip.since_seq, + // Automatic workers run on their owning node. Only an exact + // path/token/source snapshot can answer locally; Admin tokens and + // unknown identities retain the coordinator authority boundary. + let response = query_heal_status_on_owner( + async { + let Some(manager) = rustfs_heal::get_heal_manager() else { + return Ok(None); + }; + let processor = rustfs_heal::HealChannelProcessor::new(manager.clone()); + timeout( + PEER_HEAL_STATUS_TIMEOUT, + processor.execute_local_automatic_query(&heal_path, &client_token, hip.since_seq), + ) + .await + .map_err(|_| cluster_heal_control_unavailable("local_task_query_expired"))? + .map_err(|_| cluster_heal_control_unavailable("local_task_query_failed")) + }, + || async { + let request_id = uuid::Uuid::new_v4().to_string(); + let context = app_context + .clone() + .ok_or_else(|| cluster_heal_control_unavailable("app_context_unavailable"))?; + let route = prepare_heal_control_route(&context)?; + let envelope = rustfs_protos::heal_control::Envelope::query( + request_id.clone(), + new_heal_control_metadata(&route)?, + heal_path.clone(), + client_token.clone(), + hip.since_seq, + ) + .map_err(|err| s3_error!(InternalError, "encode heal control query failed: {err}"))?; + submit_cluster_heal_channel_command(context, route, envelope, &request_id, client_token.clone()).await + }, ) - .map_err(|err| s3_error!(InternalError, "encode heal control query failed: {err}"))?; - let response = submit_cluster_heal_channel_command(context, route, envelope, &request_id, client_token).await?; + .await?; if !response.success { return Err(s3_error!( InternalError, @@ -1972,6 +2008,52 @@ mod tests { use std::sync::atomic::{AtomicBool, Ordering}; use uuid::Uuid; + #[tokio::test] + async fn local_automatic_heal_query_preserves_owner_and_coordinator_errors() { + let response = || rustfs_heal_contracts::heal_channel::HealChannelResponse { + request_id: "original-token".to_string(), + success: true, + data: Some(b"retained-owner-report".to_vec()), + error: None, + }; + let called = AtomicBool::new(false); + let local = super::query_heal_status_on_owner(async { Ok(Some(response())) }, || async { + called.store(true, Ordering::SeqCst); + Ok(response()) + }) + .await + .expect("local automatic owner response"); + assert!(!called.load(Ordering::SeqCst)); + assert_eq!(local.request_id, "original-token"); + assert_eq!(local.data.as_deref(), Some(b"retained-owner-report".as_slice())); + + let fallback = super::query_heal_status_on_owner(async { Ok(None) }, || async { + called.store(true, Ordering::SeqCst); + Ok(response()) + }) + .await + .expect("unknown local identity uses coordinator"); + assert!(called.swap(false, Ordering::SeqCst)); + assert_eq!(fallback.data, local.data); + + let error = + super::query_heal_status_on_owner(async { Err(s3s::s3_error!(InternalError, "local lookup failed")) }, || async { + called.store(true, Ordering::SeqCst); + Ok(response()) + }) + .await + .expect_err("local failure must not become coordinator success"); + assert_eq!(error.code(), &S3ErrorCode::InternalError); + assert!(!called.load(Ordering::SeqCst)); + + let error = super::query_heal_status_on_owner(async { Ok(None) }, || async { + Err(s3s::s3_error!(ServiceUnavailable, "coordinator unavailable")) + }) + .await + .expect_err("coordinator error must remain visible"); + assert_eq!(error.code(), &S3ErrorCode::ServiceUnavailable); + } + #[test] fn mrf_legacy_risk_action_requires_explicit_acknowledgment_and_audit_fields() { let base = json!({