From 76f4342ae29e000ec452ed71c548d0815bf0379f Mon Sep 17 00:00:00 2001 From: Chris Date: Sat, 24 Oct 2026 08:50:13 +0900 Subject: [PATCH] feat(scanner): observe canonical heal task terminals (#8851) --- crates/heal-contracts/src/completion.rs | 105 +++++ crates/heal-contracts/src/lib.rs | 2 + crates/heal-contracts/src/outcome.rs | 110 ++++++ crates/heal/src/heal/erasure_healer/admin.rs | 1 + crates/heal/src/heal/manager.rs | 21 + crates/heal/src/heal/manager/tests.rs | 1 + .../heal/src/heal/manager/tests/completion.rs | 88 +++++ crates/heal/src/heal/outcome.rs | 120 +----- crates/heal/src/heal/task.rs | 2 +- crates/scanner/src/heal_completion.rs | 320 +++++++++++++++ crates/scanner/src/heal_completion/tests.rs | 364 ++++++++++++++++++ crates/scanner/src/lib.rs | 1 + crates/scanner/src/scanner_folder.rs | 2 +- .../src/scanner_folder/item_actions.rs | 8 +- crates/scanner/src/scanner_folder/tests.rs | 13 +- docs/operations/scanner-runtime-controls.md | 32 ++ 16 files changed, 1080 insertions(+), 110 deletions(-) create mode 100644 crates/heal-contracts/src/completion.rs create mode 100644 crates/heal-contracts/src/outcome.rs create mode 100644 crates/heal/src/heal/manager/tests/completion.rs create mode 100644 crates/scanner/src/heal_completion.rs create mode 100644 crates/scanner/src/heal_completion/tests.rs diff --git a/crates/heal-contracts/src/completion.rs b/crates/heal-contracts/src/completion.rs new file mode 100644 index 000000000..fe1b4bb67 --- /dev/null +++ b/crates/heal-contracts/src/completion.rs @@ -0,0 +1,105 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Ephemeral, bounded terminal observations. Delivery is never a heal commit gate. + +use crate::{ + heal_channel::HealScanMode, + outcome::{HealExecutionOutcome, HealObjectReceipt}, +}; +use std::sync::{Arc, OnceLock}; +use tokio::sync::broadcast; + +pub const MAX_COMPLETION_IDENTITY_BYTES: usize = 4096; +const COMPLETION_CHANNEL_CAPACITY: usize = 1024; + +#[derive(Debug, Clone)] +pub struct HealCompletion { + /// The accepted owner, never a discarded merged request ID. + pub task_id: String, + pub scan_mode: HealScanMode, + pub dry_run: bool, + pub execution: Option, + /// Only an untruncated single-object result is retained. A missing result + /// does not mean absence, success, or payload integrity. + pub object: Option, +} + +fn sender() -> &'static broadcast::Sender> { + static SENDER: OnceLock>> = OnceLock::new(); + SENDER.get_or_init(|| broadcast::channel(COMPLETION_CHANNEL_CAPACITY).0) +} + +pub fn has_heal_completion_observers() -> bool { + sender().receiver_count() > 0 +} + +/// Subscribe before admission so a fast worker cannot precede subscription. +pub fn subscribe_heal_completions() -> broadcast::Receiver> { + sender().subscribe() +} + +/// Oversized diagnostics and lagging consumers lose observations, never repairs. +pub fn publish_heal_completion(completion: HealCompletion) { + let identity_bytes = completion.object.as_ref().map_or(0, |object| { + object + .identity + .bucket + .capacity() + .saturating_add(object.identity.object.capacity()) + .saturating_add(object.identity.version_id.as_ref().map_or(0, String::capacity)) + }); + if completion.task_id.capacity().saturating_add(identity_bytes) <= MAX_COMPLETION_IDENTITY_BYTES { + let _ = sender().send(Arc::new(completion)); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn completion_delivery_is_bounded_and_does_not_require_a_consumer() { + let mut receiver = subscribe_heal_completions(); + for index in 0..=COMPLETION_CHANNEL_CAPACITY { + publish_heal_completion(HealCompletion { + task_id: format!("task-{index}"), + scan_mode: HealScanMode::Deep, + dry_run: false, + execution: Some(HealExecutionOutcome::Completed), + object: None, + }); + } + assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Lagged(1)))); + while receiver.try_recv().is_ok() {} + let mut oversized = String::with_capacity(MAX_COMPLETION_IDENTITY_BYTES + 1); + oversized.push_str("short-but-large-allocation"); + publish_heal_completion(HealCompletion { + task_id: oversized, + scan_mode: HealScanMode::Deep, + dry_run: false, + execution: None, + object: None, + }); + assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Empty))); + drop(receiver); + publish_heal_completion(HealCompletion { + task_id: "no-consumer".into(), + scan_mode: HealScanMode::Deep, + dry_run: false, + execution: Some(HealExecutionOutcome::Completed), + object: None, + }); + } +} diff --git a/crates/heal-contracts/src/lib.rs b/crates/heal-contracts/src/lib.rs index 8da3d1360..838fd2570 100644 --- a/crates/heal-contracts/src/lib.rs +++ b/crates/heal-contracts/src/lib.rs @@ -12,4 +12,6 @@ // See the License for the specific language governing permissions and // limitations under the License. +pub mod completion; pub mod heal_channel; +pub mod outcome; diff --git a/crates/heal-contracts/src/outcome.rs b/crates/heal-contracts/src/outcome.rs new file mode 100644 index 000000000..5b4402ae8 --- /dev/null +++ b/crates/heal-contracts/src/outcome.rs @@ -0,0 +1,110 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Typed object results shared by producers and the heal executor. + +use serde::{Deserialize, Serialize}; +use std::time::SystemTime; +use uuid::Uuid; + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HealObjectKind { + Object, + Metadata, + Decode, + DeleteMarkerPurge, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct HealObjectIdentity { + pub kind: HealObjectKind, + pub bucket: String, + pub object: String, + /// The requested version; None remains unresolved, never an absence proof. + pub version_id: Option, + pub bucket_incarnation_id: Option, + pub pool_index: Option, + pub set_index: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HealDeferredReason { + DanglingDeleteGrace, + RetiredMarkerProof, + TransientUsageCache, + TransientExistenceCheck, + Deadline, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HealFailureClass { + Recoverable, + RetryExhausted, + Permanent, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde( + tag = "state", + content = "details", + rename_all = "snake_case", + rename_all_fields = "camelCase", + deny_unknown_fields +)] +pub enum HealObjectDisposition { + /// The legacy storage response does not prove the requested check or commit. + Unknown, + Repaired, + VerifiedHealthy, + /// Authoritative metadata/presence proof only. This does not certify + /// payload integrity. + MetadataHealthy, + AuthoritativelyAbsent, + Deferred { + reason: HealDeferredReason, + retry_not_before: Option, + }, + Failed(HealFailureClass), + Cancelled, + DryRunObserved, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct HealObjectReceipt { + pub identity: HealObjectIdentity, + pub disposition: HealObjectDisposition, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum HealAbortReason { + Cancelled, + Deadline, + Untraversable, +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] +#[serde(tag = "state", content = "reason", rename_all = "snake_case")] +pub enum HealExecutionOutcome { + #[default] + Pending, + Running, + Completed, + CompletedWithErrors, + Aborted(HealAbortReason), +} diff --git a/crates/heal/src/heal/erasure_healer/admin.rs b/crates/heal/src/heal/erasure_healer/admin.rs index 18f818e5b..27908a2b5 100644 --- a/crates/heal/src/heal/erasure_healer/admin.rs +++ b/crates/heal/src/heal/erasure_healer/admin.rs @@ -13,6 +13,7 @@ // limitations under the License. use super::*; +use crate::heal::outcome::HealObjectReceiptExt; use crate::heal::outcome::{ HealDeferredReason, HealFailureClass, HealObjectDisposition, HealObjectIdentity, HealObjectKind, HealObjectOutcome as CanonicalObjectOutcome, HealTaskOutcome, diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index cbe5f4e1f..bf1f5fbdf 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -315,6 +315,27 @@ async fn publish_completed_heal( aliases.retain(|alias_id, alias| alias_id != task_id && alias.task_id != task_id); } prune_completed_heal_statuses(&mut retained); + drop(retained); + drop(aliases); + if terminal && rustfs_heal_contracts::completion::has_heal_completion_observers() { + // Publish the canonical terminal only after ownership settles. Retrying + // attempts and command success are not object completion observations. + let outcome = completed.outcome.as_deref(); + let object = outcome + .filter(|outcome| completed.heal_type.is_per_object() && !outcome.objects_truncated && outcome.objects.len() == 1) + .and_then(|outcome| outcome.objects.front()) + .map(|object| crate::heal::outcome::HealObjectReceipt { + identity: object.identity.clone(), + disposition: object.disposition.clone(), + }); + rustfs_heal_contracts::completion::publish_heal_completion(rustfs_heal_contracts::completion::HealCompletion { + task_id: task_id.to_owned(), + scan_mode: completed.options.scan_mode, + dry_run: completed.options.dry_run, + execution: outcome.map(|outcome| outcome.execution), + object, + }); + } } #[derive(Debug, Clone)] diff --git a/crates/heal/src/heal/manager/tests.rs b/crates/heal/src/heal/manager/tests.rs index 23de19d64..288284ccd 100644 --- a/crates/heal/src/heal/manager/tests.rs +++ b/crates/heal/src/heal/manager/tests.rs @@ -30,6 +30,7 @@ pub(super) mod admin_overlap; mod root_recovery; use uuid::Uuid; mod auto_scan; +mod completion; mod disposition; mod legacy_ambiguity_logging; mod running_mainline; diff --git a/crates/heal/src/heal/manager/tests/completion.rs b/crates/heal/src/heal/manager/tests/completion.rs new file mode 100644 index 000000000..c0cd0d67b --- /dev/null +++ b/crates/heal/src/heal/manager/tests/completion.rs @@ -0,0 +1,88 @@ +use super::*; +use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome}; +use rustfs_heal_contracts::{completion::subscribe_heal_completions, heal_channel::HealScanMode}; + +#[tokio::test] +async fn canonical_completion_publishes_typed_objects_only_for_terminal_untruncated_results() { + let mut receiver = subscribe_heal_completions(); + let completed_heals = Mutex::new(HashMap::new()); + let aliases = Mutex::new(HashMap::new()); + let id = Uuid::new_v4().to_string(); + let alias = Uuid::new_v4().to_string(); + aliases + .lock() + .await + .insert(alias.clone(), HealTaskAlias { task_id: id.clone() }); + let mut completed = completed_retention_fixture(SystemTime::now()); + completed.heal_type = HealType::Object { + bucket: "bucket".into(), + object: "object".into(), + version_id: None, + }; + completed.options.scan_mode = HealScanMode::Normal; + let mut outcome = HealTaskOutcome::default(); + outcome.record(HealObjectOutcome { + identity: HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: "bucket".into(), + object: "object".into(), + version_id: None, + bucket_incarnation_id: Some(Uuid::from_u128(44)), + pool_index: Some(0), + set_index: Some(1), + }, + disposition: HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + }, + detail: None, + }); + outcome.finish(None); + completed.outcome = Some(Arc::new(outcome.clone())); + completed.status = HealTaskStatus::Retrying { + error: "retry".into(), + retry_attempt: 1, + }; + publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false).await; + while let Ok(observed) = receiver.try_recv() { + assert_ne!(observed.task_id, id, "a retry attempt must not publish a terminal"); + } + completed.status = HealTaskStatus::Completed; + publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), true).await; + let observed = tokio::time::timeout(Duration::from_secs(1), async { + loop { + let event = receiver.recv().await.expect("terminal broadcaster stays open"); + if event.task_id == id { + break event; + } + } + }) + .await + .expect("canonical terminal must be published"); + assert_eq!(observed.scan_mode, HealScanMode::Normal); + assert_eq!(observed.execution, Some(HealExecutionOutcome::Completed)); + assert!(matches!( + observed.object.as_ref().expect("single typed outcome").disposition, + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + .. + } + )); + assert!(completed_heals.lock().await.contains_key(&alias), "existing alias retention is preserved"); + + outcome.objects_truncated = true; + completed.outcome = Some(Arc::new(outcome)); + let truncated_id = Uuid::new_v4().to_string(); + publish_completed_heal(&completed_heals, &aliases, &truncated_id, completed, true).await; + let truncated = tokio::time::timeout(Duration::from_secs(1), async { + loop { + let event = receiver.recv().await.expect("terminal broadcaster stays open"); + if event.task_id == truncated_id { + break event; + } + } + }) + .await + .expect("truncated terminal must still be observable"); + assert!(truncated.object.is_none(), "a retained tail cannot certify a whole result"); +} diff --git a/crates/heal/src/heal/outcome.rs b/crates/heal/src/heal/outcome.rs index 86550f5a2..572f5d032 100644 --- a/crates/heal/src/heal/outcome.rs +++ b/crates/heal/src/heal/outcome.rs @@ -17,95 +17,26 @@ use rustfs_heal_contracts::heal_channel::LEGACY_OBJECT_IDENTITY_UNVERIFIED_DETAIL; use serde::{Deserialize, Serialize}; -use std::{collections::VecDeque, time::SystemTime}; +use std::collections::VecDeque; +#[cfg(test)] use uuid::Uuid; const MAX_OUTCOME_ITEMS: usize = 128; const MAX_OUTCOME_BYTES: usize = 64 * 1024; const MAX_OUTCOME_DETAIL_BYTES: usize = 1024; -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum HealObjectKind { - Object, - Metadata, - Decode, - DeleteMarkerPurge, +pub use rustfs_heal_contracts::outcome::{ + HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealFailureClass, HealObjectDisposition, HealObjectIdentity, + HealObjectKind, HealObjectReceipt, +}; + +/// Repair responsibility policy stays private to the heal executor. +pub(crate) trait HealObjectReceiptExt { + fn verified_for(&self, expected: &HealObjectIdentity) -> bool; } -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct HealObjectIdentity { - pub kind: HealObjectKind, - pub bucket: String, - pub object: String, - /// The requested version; None remains unresolved, never an absence proof. - pub version_id: Option, - pub bucket_incarnation_id: Option, - pub pool_index: Option, - pub set_index: Option, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum HealDeferredReason { - DanglingDeleteGrace, - RetiredMarkerProof, - TransientUsageCache, - TransientExistenceCheck, - Deadline, -} - -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum HealFailureClass { - Recoverable, - RetryExhausted, - Permanent, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde( - tag = "state", - content = "details", - rename_all = "snake_case", - rename_all_fields = "camelCase", - deny_unknown_fields -)] -pub enum HealObjectDisposition { - /// The legacy storage response does not prove the requested check or commit. - Unknown, - Repaired, - VerifiedHealthy, - /// Authoritative metadata/presence proof only. This does not certify - /// payload integrity. - MetadataHealthy, - AuthoritativelyAbsent, - Deferred { - reason: HealDeferredReason, - retry_not_before: Option, - }, - Failed(HealFailureClass), - Cancelled, - DryRunObserved, -} - -#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "camelCase", deny_unknown_fields)] -pub struct HealObjectOutcome { - pub identity: HealObjectIdentity, - pub disposition: HealObjectDisposition, - pub detail: Option, -} - -#[derive(Debug, Clone, PartialEq, Eq)] -pub struct HealObjectReceipt { - pub identity: HealObjectIdentity, - pub disposition: HealObjectDisposition, -} - -impl HealObjectReceipt { - pub(crate) fn verified_for(&self, expected: &HealObjectIdentity) -> bool { +impl HealObjectReceiptExt for HealObjectReceipt { + fn verified_for(&self, expected: &HealObjectIdentity) -> bool { let disposition_verifies = match self.disposition { HealObjectDisposition::Repaired | HealObjectDisposition::VerifiedHealthy @@ -127,6 +58,14 @@ impl HealObjectReceipt { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "camelCase", deny_unknown_fields)] +pub struct HealObjectOutcome { + pub identity: HealObjectIdentity, + pub disposition: HealObjectDisposition, + pub detail: Option, +} + impl HealObjectOutcome { fn retained_bytes(&self) -> usize { size_of::() @@ -146,25 +85,6 @@ pub enum HealTraversalCoverage { Complete, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] -#[serde(rename_all = "snake_case")] -pub enum HealAbortReason { - Cancelled, - Deadline, - Untraversable, -} - -#[derive(Debug, Clone, Copy, Default, PartialEq, Eq, Serialize, Deserialize)] -#[serde(tag = "state", content = "reason", rename_all = "snake_case")] -pub enum HealExecutionOutcome { - #[default] - Pending, - Running, - Completed, - CompletedWithErrors, - Aborted(HealAbortReason), -} - #[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "camelCase")] pub struct HealOutcomeCounters { diff --git a/crates/heal/src/heal/task.rs b/crates/heal/src/heal/task.rs index 03f60ce53..bd083732a 100644 --- a/crates/heal/src/heal/task.rs +++ b/crates/heal/src/heal/task.rs @@ -18,7 +18,7 @@ use crate::heal::{ erasure_healer::target_outcomes_complete, outcome::{ HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealFailureClass, HealObjectDisposition, HealObjectIdentity, - HealObjectKind, HealObjectOutcome, HealObjectReceipt, HealTaskOutcome, + HealObjectKind, HealObjectOutcome, HealObjectReceipt, HealObjectReceiptExt, HealTaskOutcome, }, progress::HealProgress, resume::{ diff --git a/crates/scanner/src/heal_completion.rs b/crates/scanner/src/heal_completion.rs new file mode 100644 index 000000000..deb7bf753 --- /dev/null +++ b/crates/scanner/src/heal_completion.rs @@ -0,0 +1,320 @@ +// Copyright 2026 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Live candidate diagnostics, deliberately separate from persisted retry hints +//! and traversal coverage. Missing content/leader identities remain unverified. + +use metrics::{counter, gauge}; +use rustfs_heal_contracts::{ + completion::{HealCompletion, MAX_COMPLETION_IDENTITY_BYTES, subscribe_heal_completions}, + heal_channel::{HealAdmissionReceipt, HealChannelRequest, HealScanMode}, + outcome::{HealExecutionOutcome, HealObjectDisposition, HealObjectIdentity, HealObjectKind}, +}; +use std::{ + collections::{HashMap, VecDeque}, + sync::{Arc, Mutex}, + time::{Duration, Instant}, +}; +use tokio::sync::{broadcast, mpsc}; + +const MAX_PENDING: usize = 1024; +const MAX_RECENT: usize = 1024; +const PENDING_TTL: Duration = Duration::from_secs(60 * 60); +const RECENT_TTL: Duration = Duration::from_secs(10 * 60); + +#[derive(Debug)] +pub(crate) struct Candidate { + identity: Option, + mode: HealScanMode, + dry_run: bool, + first_seen: Instant, + count: u64, +} + +impl Candidate { + fn from_request(request: &HealChannelRequest) -> Option { + let bytes = request + .bucket + .len() + .saturating_add(request.object_prefix.as_ref().map_or(0, String::len)) + .saturating_add(request.object_version_id.as_ref().map_or(0, String::len)); + if bytes > MAX_COMPLETION_IDENTITY_BYTES { + return None; + } + Some(Self { + identity: request.object_prefix.as_ref().map(|object| HealObjectIdentity { + kind: HealObjectKind::Object, + bucket: request.bucket.clone(), + object: object.clone(), + version_id: request.object_version_id.clone(), + bucket_incarnation_id: request.expected_bucket_incarnation_id, + pool_index: request.pool_index, + set_index: request.set_index, + }), + mode: request.scan_mode.unwrap_or_default(), + dry_run: request.dry_run.unwrap_or(false), + first_seen: Instant::now(), + count: 1, + }) + } +} + +struct Recent { + completion: Arc, + counted: bool, +} + +#[derive(Default)] +struct Observations { + pending: HashMap, + recent: HashMap, + recent_order: VecDeque<(Instant, String)>, +} + +#[derive(Debug)] +struct Settlement { + candidates: u64, + first_task_observation: bool, + disposition: &'static str, + scope: &'static str, + mode: HealScanMode, +} + +impl Settlement { + fn new(candidate: Candidate, recent: &mut Recent) -> Self { + let completion = &recent.completion; + let disposition = match completion.object.as_ref().map(|object| &object.disposition) { + Some(HealObjectDisposition::VerifiedHealthy) => "verified_healthy", + Some(HealObjectDisposition::MetadataHealthy) => "metadata_healthy", + Some(HealObjectDisposition::Repaired) => "repaired", + Some(HealObjectDisposition::AuthoritativelyAbsent) => "absent", + Some(HealObjectDisposition::Deferred { .. }) => "deferred", + Some(HealObjectDisposition::Failed(_)) => "failed", + Some(HealObjectDisposition::Cancelled) => "cancelled", + Some(HealObjectDisposition::DryRunObserved) => "dry_run", + Some(HealObjectDisposition::Unknown) | None => "unknown", + }; + let scope = if candidate.dry_run || completion.dry_run { + "dry_run" + } else if completion.execution != Some(HealExecutionOutcome::Completed) { + "execution_incomplete" + } else if candidate.mode == HealScanMode::Deep && completion.scan_mode != HealScanMode::Deep { + "insufficient_scan" + } else { + match (candidate.identity.as_ref(), completion.object.as_ref()) { + (Some(expected), Some(actual)) if expected != &actual.identity => "identity_mismatch", + (Some(expected), Some(_)) + if expected.bucket_incarnation_id.is_some_and(|id| !id.is_nil()) + && expected.pool_index.is_some() + && expected.set_index.is_some() + && expected.version_id.as_ref().is_some_and(|version| !version.is_empty()) => + { + // Even an exact version and VerifiedHealthy do not bind the + // candidate's content and leader epoch with today's contract. + "content_and_leader_unknown" + } + _ => "identity_unknown", + } + }; + let first_task_observation = !recent.counted; + recent.counted = true; + Self { + candidates: candidate.count, + first_task_observation, + disposition, + scope, + mode: candidate.mode, + } + } + + fn record(self) { + counter!("rustfs_scanner_heal_candidate_terminal_total", + "disposition" => self.disposition, "scope" => self.scope, "scan_mode" => self.mode.as_str()) + .increment(self.candidates); + if self.first_task_observation { + counter!("rustfs_scanner_heal_task_terminal_observations_total").increment(1); + } + // No terminal here can certify payload coverage. In particular repaired, + // metadata-healthy and authoritative absence never imply verified bytes. + record_unknown(self.scope, self.candidates); + } +} + +fn record_unknown(reason: &'static str, count: u64) { + counter!("rustfs_scanner_heal_candidate_unknown_total", "reason" => reason).increment(count); +} + +impl Observations { + fn admit(&mut self, task_id: String, candidate: Candidate) -> Option { + if let Some(recent) = self.recent.get_mut(&task_id) { + return Some(Settlement::new(candidate, recent)); + } + if let Some(pending) = self.pending.get_mut(&task_id) { + pending.count = pending.count.saturating_add(candidate.count); + if pending.identity != candidate.identity { + pending.identity = None; + } + if candidate.mode == HealScanMode::Deep { + pending.mode = HealScanMode::Deep; + } + pending.dry_run |= candidate.dry_run; + } else if self.pending.len() < MAX_PENDING { + self.pending.insert(task_id, candidate); + } else { + record_unknown("pending_full", candidate.count); + } + None + } + + fn complete(&mut self, completion: Arc, now: Instant) -> Option { + if self.recent.contains_key(&completion.task_id) { + return None; + } + let task_id = completion.task_id.clone(); + let mut recent = Recent { + completion, + counted: false, + }; + let settled = self + .pending + .remove(&task_id) + .map(|candidate| Settlement::new(candidate, &mut recent)); + if self.recent.len() == MAX_RECENT { + self.evict_recent(); + } + self.recent_order.push_back((now, task_id.clone())); + self.recent.insert(task_id, recent); + settled + } + + fn evict_recent(&mut self) { + if let Some((_, id)) = self.recent_order.pop_front() { + self.recent.remove(&id); + } + } + + fn lost(&mut self) { + for (_, candidate) in self.pending.drain() { + record_unknown("completion_lost", candidate.count); + } + self.recent.clear(); + self.recent_order.clear(); + } + + fn maintain(&mut self, now: Instant) { + self.pending.retain(|_, candidate| { + if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL { + record_unknown("expired", candidate.count); + false + } else { + true + } + }); + while self + .recent_order + .front() + .is_some_and(|(at, _)| now.saturating_duration_since(*at) >= RECENT_TTL) + { + self.evict_recent(); + } + let oldest = self + .pending + .values() + .map(|candidate| now.saturating_duration_since(candidate.first_seen)) + .max(); + gauge!("rustfs_scanner_heal_pending_tasks").set(self.pending.len() as f64); + gauge!("rustfs_scanner_heal_pending_oldest_seconds").set(oldest.unwrap_or_default().as_secs_f64()); + } +} + +type AdmissionSender = mpsc::Sender<(String, Candidate)>; + +fn observer_from(registry: &Mutex>) -> AdmissionSender { + // This lock protects installation only; no I/O or await occurs while held. + let mut installed = registry.lock().unwrap_or_else(|poisoned| poisoned.into_inner()); + if let Some(sender) = installed.as_ref().filter(|sender| !sender.is_closed()) { + return sender.clone(); + } + let sender = { + let (sender, mut admissions) = mpsc::channel(MAX_PENDING); + let mut completions = subscribe_heal_completions(); + tokio::spawn(async move { + let mut observations = Observations::default(); + let mut tick = tokio::time::interval(Duration::from_secs(5)); + tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); + loop { + let settlement = tokio::select! { + candidate = admissions.recv() => match candidate { + Some((id, candidate)) => observations.admit(id, candidate), + None => break, + }, + completion = completions.recv() => match completion { + Ok(completion) => observations.complete(completion, Instant::now()), + Err(broadcast::error::RecvError::Lagged(skipped)) => { + counter!("rustfs_scanner_heal_completion_lost_total").increment(skipped); + observations.lost(); + None + }, + Err(broadcast::error::RecvError::Closed) => { observations.lost(); break }, + }, + _ = tick.tick() => { observations.maintain(Instant::now()); None }, + }; + if let Some(settlement) = settlement { + settlement.record(); + } + } + }); + sender + }; + *installed = Some(sender.clone()); + sender +} + +fn observer() -> AdmissionSender { + static OBSERVER: Mutex> = Mutex::new(None); + observer_from(&OBSERVER) +} + +/// Capture only the identity actually available at discovery, before dispatch. +pub(crate) fn prepare(request: &HealChannelRequest) -> Option { + let _ = observer(); + Candidate::from_request(request) +} + +pub(crate) fn observe(candidate: Option, receipt: Result<&HealAdmissionReceipt, ()>) { + observe_on(&observer(), candidate, receipt); +} + +fn observe_on(sender: &AdmissionSender, candidate: Option, receipt: Result<&HealAdmissionReceipt, ()>) { + match receipt { + Ok(receipt) if receipt.result.is_admitted() => { + let Some(candidate) = candidate else { + record_unknown("identity_too_large", 1); + return; + }; + let task_id = &receipt.task_id; + if task_id.is_empty() || task_id.len() > 128 { + record_unknown("missing_task_id", 1); + return; + } + if sender.try_send((task_id.clone(), candidate)).is_err() { + record_unknown("observer_full_or_closed", 1); + } + } + _ => record_unknown("not_admitted", 1), + } +} + +#[cfg(test)] +mod tests; diff --git a/crates/scanner/src/heal_completion/tests.rs b/crates/scanner/src/heal_completion/tests.rs new file mode 100644 index 000000000..ff1ff60cb --- /dev/null +++ b/crates/scanner/src/heal_completion/tests.rs @@ -0,0 +1,364 @@ +use super::*; +use rustfs_heal_contracts::outcome::{HealAbortReason, HealDeferredReason, HealFailureClass, HealObjectReceipt}; +use uuid::Uuid; + +fn request() -> HealChannelRequest { + HealChannelRequest { + bucket: "bucket".into(), + object_prefix: Some("object".into()), + object_version_id: Some(Uuid::from_u128(11).to_string()), + expected_bucket_incarnation_id: Some(Uuid::from_u128(22)), + pool_index: Some(0), + set_index: Some(1), + scan_mode: Some(HealScanMode::Deep), + ..HealChannelRequest::new() + } +} + +fn candidate() -> Candidate { + Candidate::from_request(&request()).expect("bounded candidate") +} + +fn completion(task: &str, disposition: HealObjectDisposition) -> Arc { + Arc::new(HealCompletion { + task_id: task.into(), + scan_mode: HealScanMode::Deep, + dry_run: false, + execution: Some(HealExecutionOutcome::Completed), + object: Some(HealObjectReceipt { + identity: candidate().identity.expect("exact object identity"), + disposition, + }), + }) +} + +#[test] +fn merged_candidates_close_only_on_the_canonical_task_once() { + let mut observations = Observations::default(); + assert!(observations.admit("owner".into(), candidate()).is_none()); + assert!(observations.admit("owner".into(), candidate()).is_none()); + assert_eq!(observations.pending.len(), 1); + assert!( + observations + .complete(completion("discarded-id", HealObjectDisposition::VerifiedHealthy), Instant::now()) + .is_none() + ); + assert_eq!(observations.pending.len(), 1); + let terminal = completion("owner", HealObjectDisposition::VerifiedHealthy); + let settled = observations + .complete(terminal.clone(), Instant::now()) + .expect("canonical completion settles both candidates"); + assert_eq!(settled.candidates, 2); + assert!(settled.first_task_observation); + assert_eq!(settled.scope, "content_and_leader_unknown"); + assert!(observations.pending.is_empty()); + assert!( + observations.complete(terminal, Instant::now()).is_none(), + "duplicate completion is ignored" + ); + let merged_late = observations + .admit("owner".into(), candidate()) + .expect("terminal before merged admission is retained"); + assert!(!merged_late.first_task_observation); + assert_eq!(merged_late.candidates, 1); +} + +#[test] +fn terminal_before_admission_is_observed_without_claiming_coverage() { + let mut observations = Observations::default(); + assert!( + observations + .complete(completion("fast", HealObjectDisposition::Repaired), Instant::now()) + .is_none() + ); + let settled = observations + .admit("fast".into(), candidate()) + .expect("fast terminal retained"); + assert_eq!(settled.disposition, "repaired"); + assert_eq!(settled.scope, "content_and_leader_unknown"); + assert!(settled.first_task_observation); +} + +#[test] +fn replacement_bucket_version_or_location_cannot_match_an_old_result() { + for change in 0..4 { + let mut observed = candidate(); + let identity = observed.identity.as_mut().expect("candidate identity"); + match change { + 0 => identity.bucket_incarnation_id = Some(Uuid::from_u128(23)), + 1 => identity.version_id = Some(Uuid::from_u128(12).to_string()), + 2 => identity.pool_index = Some(1), + _ => identity.set_index = Some(2), + } + let mut recent = Recent { + completion: completion("old", HealObjectDisposition::VerifiedHealthy), + counted: false, + }; + assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_mismatch"); + } +} + +#[test] +fn unversioned_overwrite_and_unknown_leader_never_certify_payload() { + let mut observed = candidate(); + observed.identity.as_mut().expect("identity").version_id = None; + let mut terminal = (*completion("same-name", HealObjectDisposition::VerifiedHealthy)).clone(); + terminal.object.as_mut().expect("receipt").identity.version_id = None; + let mut recent = Recent { + completion: Arc::new(terminal), + counted: false, + }; + assert_eq!(Settlement::new(observed, &mut recent).scope, "identity_unknown"); + let mut recent = Recent { + completion: completion("old-leader", HealObjectDisposition::VerifiedHealthy), + counted: false, + }; + assert_eq!(Settlement::new(candidate(), &mut recent).scope, "content_and_leader_unknown"); +} + +#[test] +fn normal_metadata_and_unproven_repairs_do_not_satisfy_deep_candidates() { + for disposition in [HealObjectDisposition::MetadataHealthy, HealObjectDisposition::Repaired] { + let mut terminal = (*completion("normal", disposition)).clone(); + terminal.scan_mode = HealScanMode::Normal; + let mut recent = Recent { + completion: Arc::new(terminal), + counted: false, + }; + assert_eq!(Settlement::new(candidate(), &mut recent).scope, "insufficient_scan"); + } +} + +#[test] +fn command_success_does_not_hide_deferred_failed_cancelled_or_dry_run_objects() { + for (disposition, label) in [ + ( + HealObjectDisposition::Deferred { + reason: HealDeferredReason::Deadline, + retry_not_before: None, + }, + "deferred", + ), + (HealObjectDisposition::Failed(HealFailureClass::Permanent), "failed"), + (HealObjectDisposition::Cancelled, "cancelled"), + (HealObjectDisposition::DryRunObserved, "dry_run"), + (HealObjectDisposition::AuthoritativelyAbsent, "absent"), + ] { + let mut recent = Recent { + completion: completion("completed-command", disposition), + counted: false, + }; + let settled = Settlement::new(candidate(), &mut recent); + assert_eq!(settled.disposition, label); + assert_eq!(settled.scope, "content_and_leader_unknown"); + } + let mut terminal = (*completion("cancelled", HealObjectDisposition::VerifiedHealthy)).clone(); + terminal.execution = Some(HealExecutionOutcome::Aborted(HealAbortReason::Cancelled)); + let mut recent = Recent { + completion: Arc::new(terminal), + counted: false, + }; + assert_eq!(Settlement::new(candidate(), &mut recent).scope, "execution_incomplete"); +} + +#[test] +fn missing_or_truncated_object_result_remains_unknown() { + let mut terminal = (*completion("truncated", HealObjectDisposition::VerifiedHealthy)).clone(); + terminal.object = None; + let mut recent = Recent { + completion: Arc::new(terminal), + counted: false, + }; + assert_eq!(Settlement::new(candidate(), &mut recent).scope, "identity_unknown"); +} + +#[test] +fn overflow_loss_expiry_and_restart_are_bounded_and_do_not_restore_pending() { + let mut observations = Observations::default(); + for index in 0..=MAX_PENDING { + observations.admit(format!("pending-{index}"), candidate()); + } + assert_eq!(observations.pending.len(), MAX_PENDING); + assert!(!observations.pending.contains_key(&format!("pending-{MAX_PENDING}"))); + observations.lost(); + assert!(observations.pending.is_empty()); + assert!( + observations + .complete(completion("pending-0", HealObjectDisposition::VerifiedHealthy), Instant::now()) + .is_none() + ); + observations.admit("expires".into(), candidate()); + observations.maintain(Instant::now() + PENDING_TTL); + assert!(observations.pending.is_empty()); + assert!(observations.recent.is_empty()); + let mut restarted = Observations::default(); + assert!( + restarted + .complete(completion("expires", HealObjectDisposition::VerifiedHealthy), Instant::now()) + .is_none() + ); +} + +#[test] +fn terminal_history_eviction_loses_evidence_without_inventing_a_completion() { + let mut observations = Observations::default(); + for index in 0..=MAX_RECENT { + observations.complete( + completion(&format!("terminal-{index}"), HealObjectDisposition::VerifiedHealthy), + Instant::now(), + ); + } + assert_eq!(observations.recent.len(), MAX_RECENT); + assert_eq!(observations.recent_order.len(), MAX_RECENT); + assert!(observations.admit("terminal-0".into(), candidate()).is_none()); + assert!(observations.pending.contains_key("terminal-0")); +} + +#[test] +fn identity_size_is_bounded_before_cloning() { + let mut large = request(); + large.object_prefix = Some("x".repeat(MAX_COMPLETION_IDENTITY_BYTES)); + assert!(Candidate::from_request(&large).is_none()); +} + +#[test] +fn observer_reinstalls_after_runtime_teardown() { + let registry = Mutex::new(None); + let first = { + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("first runtime"); + runtime.block_on(async { observer_from(®istry) }) + }; + assert!(first.is_closed(), "runtime teardown closes the old observer"); + let runtime = tokio::runtime::Builder::new_current_thread() + .enable_all() + .build() + .expect("replacement runtime"); + runtime.block_on(async { + let replacement = observer_from(®istry); + assert!(!replacement.is_closed(), "a stale global sender must not suppress a new observer"); + replacement + .try_send(("new-runtime".into(), candidate())) + .expect("replacement observer admits work"); + }); +} + +#[derive(Default)] +struct CompletionRecorder(Mutex>>); + +impl metrics::Recorder for CompletionRecorder { + fn describe_counter(&self, _: metrics::KeyName, _: Option, _: metrics::SharedString) {} + fn describe_gauge(&self, _: metrics::KeyName, _: Option, _: metrics::SharedString) {} + fn describe_histogram(&self, _: metrics::KeyName, _: Option, _: metrics::SharedString) {} + fn register_counter(&self, key: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Counter { + metrics::Counter::from_arc( + self.0 + .lock() + .expect("completion recorder") + .entry(key.clone()) + .or_default() + .clone(), + ) + } + fn register_gauge(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Gauge { + metrics::Gauge::noop() + } + fn register_histogram(&self, _: &metrics::Key, _: &metrics::Metadata<'_>) -> metrics::Histogram { + metrics::Histogram::noop() + } +} + +impl CompletionRecorder { + fn total(&self, metric: &str) -> u64 { + self.0 + .lock() + .expect("completion recorder") + .iter() + .filter(|(key, _)| key.name() == metric) + .map(|(_, value)| value.load(std::sync::atomic::Ordering::Relaxed)) + .sum() + } + + async fn wait_for_terminals(&self, expected: u64) { + tokio::time::timeout(Duration::from_secs(2), async { + while self.total("rustfs_scanner_heal_candidate_terminal_total") < expected { + tokio::task::yield_now().await; + } + }) + .await + .expect("the live observer must settle admitted candidates"); + } +} + +// A local installation avoids sharing another test runtime's observer. The +// current-thread runtime keeps its spawned observer under this local recorder. +#[tokio::test] +async fn live_observer_associates_early_completion_with_canonical_merged_receipts() { + use rustfs_heal_contracts::{completion::publish_heal_completion, heal_channel::HealAdmissionResult}; + + let recorder = CompletionRecorder::default(); + let _recorder = metrics::set_default_local_recorder(&recorder); + let registry = Mutex::new(None); + let sender = observer_from(®istry); + let canonical_id = Uuid::new_v4().to_string(); + let sentinel_id = Uuid::new_v4().to_string(); + let source_request = request(); + assert_ne!(source_request.id, canonical_id, "merged requests retain their owner's ID"); + + observe_on( + &sender, + Some(candidate()), + Ok(&HealAdmissionReceipt { + result: HealAdmissionResult::Accepted, + task_id: sentinel_id.clone(), + }), + ); + publish_heal_completion((*completion(&canonical_id, HealObjectDisposition::VerifiedHealthy)).clone()); + // FIFO on the real broadcast channel makes this terminal an acknowledgement + // that the earlier unmatched completion was consumed before its admission. + publish_heal_completion((*completion(&sentinel_id, HealObjectDisposition::MetadataHealthy)).clone()); + recorder.wait_for_terminals(1).await; + + let merged = HealAdmissionReceipt { + result: HealAdmissionResult::Merged, + task_id: canonical_id, + }; + observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged)); + recorder.wait_for_terminals(2).await; + observe_on(&sender, Candidate::from_request(&source_request), Ok(&merged)); + recorder.wait_for_terminals(3).await; + + assert_eq!(recorder.total("rustfs_scanner_heal_candidate_terminal_total"), 3); + assert_eq!( + recorder.total("rustfs_scanner_heal_task_terminal_observations_total"), + 2, + "two merged candidates must add only one canonical task observation" + ); + assert_eq!( + recorder.total("rustfs_scanner_heal_candidate_unknown_total"), + 3, + "real typed terminals still lack content and leader proof" + ); + let metrics = recorder.0.lock().expect("completion recorder"); + let (_, verified_healthy) = metrics + .iter() + .find(|(key, _)| { + key.name() == "rustfs_scanner_heal_candidate_terminal_total" + && key + .labels() + .any(|label| label.key() == "disposition" && label.value() == "verified_healthy") + }) + .expect("typed storage result must reach the real terminal metric"); + assert_eq!(verified_healthy.load(std::sync::atomic::Ordering::Relaxed), 2); + for (key, _) in metrics + .iter() + .filter(|(key, _)| key.name() == "rustfs_scanner_heal_candidate_terminal_total") + { + assert!( + key.labels() + .any(|label| label.key() == "scope" && label.value() == "content_and_leader_unknown") + ); + } +} diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index aa126f168..11ba75b64 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -59,6 +59,7 @@ use uuid::Uuid; pub mod data_usage_define; pub mod error; +mod heal_completion; pub mod prefix_usage; pub mod raw_page_index; mod remote_scanner; diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 9580c0ca1..0bb265220 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -50,7 +50,7 @@ use rustfs_filemeta::{ }; use rustfs_heal_contracts::heal_channel::{ HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest, - HealRequestSource, HealScanMode, send_heal_request_with_admission, + HealRequestSource, HealScanMode, send_heal_request_with_receipt, }; use rustfs_scanner_metrics::metrics::{ CloseDiskGuard, IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource, diff --git a/crates/scanner/src/scanner_folder/item_actions.rs b/crates/scanner/src/scanner_folder/item_actions.rs index bb866bd8d..bcc9bd003 100644 --- a/crates/scanner/src/scanner_folder/item_actions.rs +++ b/crates/scanner/src/scanner_folder/item_actions.rs @@ -480,8 +480,11 @@ pub(super) async fn send_scanner_heal_request( ) -> Result { let priority = request.priority; let trace_context = scanner_heal_candidate_trace_context(&request); - match send_heal_request_with_admission(request).await { - Ok(result) => { + let candidate = crate::heal_completion::prepare(&request); + match send_heal_request_with_receipt(request).await { + Ok(receipt) => { + crate::heal_completion::observe(candidate, Ok(&receipt)); + let result = receipt.result; record_heal_candidate_admission(candidate_type, priority, result); if let Some(trace_context) = trace_context.as_ref() { emit_scanner_heal_candidate_trace(ScannerHealCandidateTrace { @@ -498,6 +501,7 @@ pub(super) async fn send_scanner_heal_request( Ok(result) } Err(err) => { + crate::heal_completion::observe(candidate, Err(())); counter!( "rustfs_heal_candidate_enqueue_total", "type" => candidate_type.to_string(), diff --git a/crates/scanner/src/scanner_folder/tests.rs b/crates/scanner/src/scanner_folder/tests.rs index f45888efd..d9a9180d5 100644 --- a/crates/scanner/src/scanner_folder/tests.rs +++ b/crates/scanner/src/scanner_folder/tests.rs @@ -2589,20 +2589,21 @@ async fn test_scan_folder_exits_when_abandoned_child_listing_finishes() { let heal_starts_clone = heal_starts.clone(); let healed_versions = Arc::new(Mutex::new(Vec::>::new())); let healed_versions_clone = healed_versions.clone(); - let mut heal_rx = - rustfs_heal_contracts::heal_channel::init_heal_channel().expect("heal channel should initialize once for scanner tests"); + let (_commands, mut heal_rx) = rustfs_heal_contracts::heal_channel::init_heal_channels() + .expect("heal channels should initialize once for scanner tests"); let _heal_responder = tokio::spawn(async move { while let Some(command) = heal_rx.recv().await { - if let rustfs_heal_contracts::heal_channel::HealChannelCommand::Start { - request, response_tx, .. - } = command + let rustfs_heal_contracts::heal_channel::HealReceiptCommand { request, response_tx } = command; { heal_starts_clone.fetch_add(1, Ordering::Relaxed); healed_versions_clone .lock() .expect("heal version capture lock should not be poisoned") .push(request.object_version_id); - let _ = response_tx.send(Ok(HealAdmissionResult::Accepted)); + let _ = response_tx.send(Ok(rustfs_heal_contracts::heal_channel::HealAdmissionReceipt { + result: HealAdmissionResult::Accepted, + task_id: request.id, + })); } } }); diff --git a/docs/operations/scanner-runtime-controls.md b/docs/operations/scanner-runtime-controls.md index a4db173ec..ba5fe642f 100644 --- a/docs/operations/scanner-runtime-controls.md +++ b/docs/operations/scanner-runtime-controls.md @@ -502,6 +502,38 @@ The pacing gate holds neither namespace locks nor I/O/page permits while sleepin A missing provider, zero pause, or both class thresholds set to zero preserves unpaced execution. Missing counts follow the existing shared pressure interpreter; they are observations, not health, quorum or resource-ownership proof. The current provider exposes node-level workload classes, so this does not claim independent per-set foreground measurements or a hard global resource budget. Runtime waits increment `rustfs_heal_mainline_throttle_total` with `source=admin`, `result=delayed`, and a foreground-pressure or `recovery_window` reason. Real p99/throughput protection requires the separate W20 fixed-load ABBA measurements. +## Live Scanner Heal Completions + +Scanner admissions now retain the canonical heal task ID, including merged +requests. Node-local metrics associate these live candidates with the executor's +typed terminal result: + +| Metric | Meaning | +|---|---| +| `rustfs_scanner_heal_pending_tasks` | Canonical tasks awaiting a terminal observation. | +| `rustfs_scanner_heal_pending_oldest_seconds` | Age of the oldest retained pending candidate; refreshed every five seconds. | +| `rustfs_scanner_heal_candidate_terminal_total` | Associated candidates, labeled by object `disposition`, expected `scan_mode`, and unresolved `scope`. | +| `rustfs_scanner_heal_task_terminal_observations_total` | Associated task observations, deduplicated within the bounded recent history. This is not an execution count. | +| `rustfs_scanner_heal_candidate_unknown_total` | Candidates whose integrity remains unknown, labeled by reason. | +| `rustfs_scanner_heal_completion_lost_total` | Terminal notifications skipped by a lagging observer. | + +These metrics use the configured metrics exporter. A task's `Completed` state +can include a deferred object; the `disposition` label preserves that distinction. +`MetadataHealthy`, `Repaired`, and authoritative absence never imply verified +payload bytes. Even `VerifiedHealthy` cannot certify scanner coverage: the current +contract does not bind the candidate's content identity and leader epoch. No +verified-byte or full-bucket coverage counter is advanced here. + +The admission buffer, pending map, terminal channel, and recent history each hold +at most 1,024 entries. Retained identities are capped at 4 KiB; terminal history +expires after ten minutes and pending observations after one hour. Overflow, +notification loss, missing identity, and expiry remain unknown. Queue cancellation +or displacement paths without a terminal snapshot also age out as unknown. +Process restart clears this diagnostic state; a restarted runtime reinstalls its +observer. Counts and pending age are not a durable integrity ledger or a cold-range +completion deadline. Existing periodic traversal and pending retry hints provide +compensation independently. + ## Pending Heal Hints The scanner's persisted pending-heal cache is a best-effort retry backstop, bounded to 10,000 hints per bucket and a 24-hour age limit. These limits do not authorize garbage collection of committed durable repair obligations. A durable owner must retain its independent replay record until verified object completion or an equivalent durable successor permits removal.