Compare commits

...

13 Commits

Author SHA1 Message Date
overtrue 1f94f5cef4 test(scanner): isolate nil identity regressions 2026-10-28 11:10:17 +08:00
overtrue e2c7624359 fix(scanner): classify unknown heal identity fields 2026-10-28 11:09:41 +08:00
overtrue 72e7e70f26 test(scanner): cover heal version identity formats 2026-10-28 11:09:03 +08:00
overtrue 07c8d81c02 test(scanner): cover unknown heal identity fields 2026-10-28 11:07:41 +08:00
overtrue b22b326432 fix(scanner): retain distinct heal candidate groups 2026-10-28 11:06:07 +08:00
overtrue 3a3cd5bbdc test(scanner): mark expected candidate regressions 2026-10-28 11:02:33 +08:00
overtrue 5608495e0e test(scanner): cover expiry for additional groups 2026-10-28 11:01:46 +08:00
overtrue 906689b674 test(scanner): cover expiry at terminal boundaries 2026-10-28 11:01:00 +08:00
overtrue 543334128d test(scanner): cover reordered candidate expiry 2026-10-28 10:59:59 +08:00
overtrue 8cd556c272 test(scanner): cover mixed heal observation candidates 2026-10-28 10:58:58 +08:00
overtrue 8605f3d3d1 fix(heal): enable UUID serialization for contracts 2026-10-28 10:43:08 +08:00
overtrue 6e6777c6fd test(heal): combine queue ownership and observations
Test-only combination of observed source 839467a0 with queue changes c96fc5cf and scoped tests 41d06b4d.
2026-10-28 10:25:06 +08:00
overtrue 839467a01b fix(heal): prototype owner observation identity 2026-10-28 10:06:10 +08:00
21 changed files with 3211 additions and 364 deletions
Generated
+1
View File
@@ -9858,6 +9858,7 @@ version = "1.0.2"
dependencies = [
"serde",
"serde_json",
"thiserror 2.0.21",
"tokio",
"uuid",
]
+2 -1
View File
@@ -29,8 +29,9 @@ workspace = true
[dependencies]
serde = { workspace = true, features = ["derive"] }
thiserror.workspace = true
tokio = { workspace = true, features = ["sync"] }
uuid = { workspace = true, features = ["v4", "fast-rng", "macro-diagnostics"] }
uuid = { workspace = true, features = ["v4", "serde", "fast-rng", "macro-diagnostics"] }
[dev-dependencies]
serde_json = { workspace = true }
+223 -40
View File
@@ -18,12 +18,25 @@ use crate::{
heal_channel::HealScanMode,
outcome::{HealExecutionOutcome, HealObjectReceipt},
};
use std::sync::{Arc, OnceLock};
use std::sync::{
Arc, OnceLock,
atomic::{AtomicU64, Ordering},
};
use tokio::sync::broadcast;
pub const MAX_COMPLETION_IDENTITY_BYTES: usize = 4096;
const COMPLETION_CHANNEL_CAPACITY: usize = 1024;
/// A process-local observation identity. It is never persisted or sent over RPC.
pub type HealObservationId = u64;
/// Allocate one identity per accepted owner. Exhaustion is reported as unknown.
pub fn new_heal_observation_id() -> Option<HealObservationId> {
static NEXT: AtomicU64 = AtomicU64::new(1);
NEXT.try_update(Ordering::Relaxed, Ordering::Relaxed, |value| value.checked_add(1))
.ok()
}
#[derive(Debug, Clone)]
pub struct HealCompletion {
/// The accepted owner, never a discarded merged request ID.
@@ -36,33 +49,84 @@ pub struct HealCompletion {
pub object: Option<HealObjectReceipt>,
}
fn sender() -> &'static broadcast::Sender<Arc<HealCompletion>> {
static SENDER: OnceLock<broadcast::Sender<Arc<HealCompletion>>> = OnceLock::new();
SENDER.get_or_init(|| broadcast::channel(COMPLETION_CHANNEL_CAPACITY).0)
/// A completion paired with an optional in-process owner identity.
#[derive(Debug, Clone)]
pub struct ObservedHealCompletion {
pub completion: Arc<HealCompletion>,
pub observation_id: Option<HealObservationId>,
}
impl std::ops::Deref for ObservedHealCompletion {
type Target = HealCompletion;
fn deref(&self) -> &Self::Target {
&self.completion
}
}
struct CompletionBroadcaster {
legacy: broadcast::Sender<Arc<HealCompletion>>,
observed: broadcast::Sender<Arc<ObservedHealCompletion>>,
}
impl CompletionBroadcaster {
fn new(capacity: usize) -> Self {
Self {
legacy: broadcast::channel(capacity).0,
observed: broadcast::channel(capacity).0,
}
}
fn has_observers(&self) -> bool {
self.legacy.receiver_count() > 0 || self.observed.receiver_count() > 0
}
fn publish(&self, completion: HealCompletion, observation_id: Option<HealObservationId>) {
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 completion = Arc::new(completion);
let _ = self.legacy.send(Arc::clone(&completion));
let _ = self.observed.send(Arc::new(ObservedHealCompletion {
completion,
observation_id,
}));
}
}
}
fn broadcaster() -> &'static CompletionBroadcaster {
static BROADCASTER: OnceLock<CompletionBroadcaster> = OnceLock::new();
BROADCASTER.get_or_init(|| CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY))
}
pub fn has_heal_completion_observers() -> bool {
sender().receiver_count() > 0
broadcaster().has_observers()
}
/// Subscribe before admission so a fast worker cannot precede subscription.
pub fn subscribe_heal_completions() -> broadcast::Receiver<Arc<HealCompletion>> {
sender().subscribe()
broadcaster().legacy.subscribe()
}
pub fn subscribe_observed_heal_completions() -> broadcast::Receiver<Arc<ObservedHealCompletion>> {
broadcaster().observed.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));
}
publish_observed_heal_completion(completion, None);
}
/// Each channel is FIFO independently; relative delivery across channels is not ordered.
pub fn publish_observed_heal_completion(completion: HealCompletion, observation_id: Option<HealObservationId>) {
broadcaster().publish(completion, observation_id);
}
#[cfg(test)]
@@ -70,36 +134,155 @@ mod tests {
use super::*;
#[test]
fn completion_delivery_is_bounded_and_does_not_require_a_consumer() {
let mut receiver = subscribe_heal_completions();
fn public_completion_facades_share_receipts_and_preserve_missing_tokens() {
// Other completion tests use private broadcasters, keeping this global
// facade test isolated from their capacity and lag checks.
let mut legacy = subscribe_heal_completions();
let mut observed = subscribe_observed_heal_completions();
let completion = HealCompletion {
task_id: uuid::Uuid::new_v4().to_string(),
scan_mode: HealScanMode::Deep,
dry_run: false,
execution: None,
object: None,
};
publish_heal_completion(completion.clone());
let legacy_event = legacy.try_recv().expect("legacy subscriber receives old API event");
let observed_event = observed.try_recv().expect("observer receives old API event");
assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
assert_eq!(legacy_event.task_id, completion.task_id);
assert_eq!(observed_event.observation_id, None);
publish_observed_heal_completion(completion, Some(51));
let legacy_event = legacy.try_recv().expect("legacy subscriber receives observed API event");
let observed_event = observed.try_recv().expect("observer receives observed API event");
assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
assert_eq!(observed_event.observation_id, Some(51));
}
#[test]
fn legacy_and_observed_subscribers_receive_the_same_completion_in_each_channel_order() {
let id = uuid::Uuid::new_v4().to_string();
let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
let mut legacy = bus.legacy.subscribe();
let mut observed = bus.observed.subscribe();
assert!(bus.has_observers());
for (suffix, token) in [("first", Some(41)), ("second", None)] {
bus.publish(
HealCompletion {
task_id: format!("{id}-{suffix}"),
scan_mode: HealScanMode::Normal,
dry_run: false,
execution: None,
object: None,
},
token,
);
}
for (suffix, token) in [("first", Some(41)), ("second", None)] {
let expected = format!("{id}-{suffix}");
let legacy_event = loop {
let event = legacy.try_recv().expect("legacy completion");
if event.task_id == expected {
break event;
}
};
let observed_event = loop {
let event = observed.try_recv().expect("observed completion");
if event.task_id == expected {
break event;
}
};
assert!(Arc::ptr_eq(&legacy_event, &observed_event.completion));
assert_eq!(observed_event.observation_id, token);
}
drop(legacy);
assert!(bus.has_observers(), "observed-only subscribers require completion construction");
let completion = HealCompletion {
task_id: id.clone(),
scan_mode: HealScanMode::Deep,
dry_run: false,
execution: None,
object: None,
};
bus.publish(completion, None);
let observed_event = loop {
let event = observed.try_recv().expect("legacy publication reaches observed channel");
if event.task_id == id {
break event;
}
};
assert_eq!(observed_event.observation_id, None);
}
#[test]
fn legacy_and_observed_channels_report_lag_independently() {
let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
let mut legacy = bus.legacy.subscribe();
let mut observed = bus.observed.subscribe();
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,
});
bus.publish(
HealCompletion {
task_id: format!("dual-lag-{index}"),
scan_mode: HealScanMode::Normal,
dry_run: false,
execution: None,
object: None,
},
Some(42),
);
}
assert!(matches!(legacy.try_recv(), Err(broadcast::error::TryRecvError::Lagged(_))));
assert!(matches!(observed.try_recv(), Err(broadcast::error::TryRecvError::Lagged(_))));
}
#[test]
fn completion_delivery_is_bounded_and_does_not_require_a_consumer() {
let bus = CompletionBroadcaster::new(COMPLETION_CHANNEL_CAPACITY);
assert!(!bus.has_observers());
let mut receiver = bus.legacy.subscribe();
assert!(bus.has_observers());
for index in 0..=COMPLETION_CHANNEL_CAPACITY {
bus.publish(
HealCompletion {
task_id: format!("task-{index}"),
scan_mode: HealScanMode::Deep,
dry_run: false,
execution: Some(HealExecutionOutcome::Completed),
object: None,
},
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,
});
let mut observed = bus.observed.subscribe();
bus.publish(
HealCompletion {
task_id: oversized,
scan_mode: HealScanMode::Deep,
dry_run: false,
execution: None,
object: None,
},
None,
);
assert!(matches!(receiver.try_recv(), Err(broadcast::error::TryRecvError::Empty)));
assert!(matches!(observed.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,
});
drop(observed);
assert!(!bus.has_observers());
bus.publish(
HealCompletion {
task_id: "no-consumer".into(),
scan_mode: HealScanMode::Deep,
dry_run: false,
execution: Some(HealExecutionOutcome::Completed),
object: None,
},
None,
);
}
}
+216 -4
View File
@@ -271,6 +271,39 @@ pub struct HealAdmissionReceipt {
pub task_id: String,
}
/// Optional process-local observation metadata for a canonical receipt.
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ObservedHealAdmissionReceipt {
pub receipt: HealAdmissionReceipt,
/// Process-local owner identity; absent for remote or unproven owners.
pub observation_id: Option<crate::completion::HealObservationId>,
}
/// Errors from process-local observed admission and its channel initialization.
#[derive(Debug, thiserror::Error)]
pub enum HealObservationChannelError {
#[error("Heal channel sender already initialized")]
AlreadyInitialized,
#[error("Heal channel not initialized")]
NotInitialized,
#[error("Heal receipt command channel closed")]
LegacyCommandClosed,
#[error("Failed to receive heal admission response: {0}")]
LegacyResponseClosed(#[source] oneshot::error::RecvError),
#[error("Failed to receive observed heal admission response: {0}")]
ObservedResponseClosed(#[source] oneshot::error::RecvError),
#[error("{0}")]
AdmissionFailed(String),
}
impl std::ops::Deref for ObservedHealAdmissionReceipt {
type Target = HealAdmissionReceipt;
fn deref(&self) -> &Self::Target {
&self.receipt
}
}
impl HealAdmissionResult {
pub fn result_label(self) -> &'static str {
match self {
@@ -447,9 +480,19 @@ pub struct HealReceiptCommand {
/// Canonical-receipt command receiver.
pub type HealReceiptReceiver = mpsc::UnboundedReceiver<HealReceiptCommand>;
/// Process-local admission command carrying an observed receipt.
#[derive(Debug)]
pub struct ObservedHealReceiptCommand {
pub request: HealChannelRequest,
pub response_tx: oneshot::Sender<Result<ObservedHealAdmissionReceipt, HealObservationChannelError>>,
}
pub type ObservedHealReceiptReceiver = mpsc::UnboundedReceiver<ObservedHealReceiptCommand>;
struct HealChannelSenders {
command: HealChannelSender,
receipt: mpsc::UnboundedSender<HealReceiptCommand>,
observed: Option<mpsc::UnboundedSender<ObservedHealReceiptCommand>>,
}
/// Global heal channel sender
@@ -469,12 +512,31 @@ pub fn init_heal_channel() -> Result<HealChannelReceiver, &'static str> {
/// Initialize the legacy command and canonical-receipt channels atomically.
pub fn init_heal_channels() -> Result<(HealChannelReceiver, HealReceiptReceiver), &'static str> {
let (command_receiver, receipt_receiver, _) =
init_heal_channels_inner(false).map_err(|_| "Heal channel sender already initialized")?;
Ok((command_receiver, receipt_receiver))
}
/// Initialize an additional local observation channel while preserving the legacy channels.
pub fn init_heal_channels_with_observations()
-> Result<(HealChannelReceiver, HealReceiptReceiver, ObservedHealReceiptReceiver), HealObservationChannelError> {
init_heal_channels_inner(true)
}
fn init_heal_channels_inner(
with_observations: bool,
) -> Result<(HealChannelReceiver, HealReceiptReceiver, ObservedHealReceiptReceiver), HealObservationChannelError> {
let (command, command_receiver) = mpsc::unbounded_channel();
let (receipt, receipt_receiver) = mpsc::unbounded_channel();
let (observed, observed_receiver) = mpsc::unbounded_channel();
GLOBAL_HEAL_CHANNEL_SENDERS
.set(HealChannelSenders { command, receipt })
.map_err(|_| "Heal channel sender already initialized")?;
Ok((command_receiver, receipt_receiver))
.set(HealChannelSenders {
command,
receipt,
observed: with_observations.then_some(observed),
})
.map_err(|_| HealObservationChannelError::AlreadyInitialized)?;
Ok((command_receiver, receipt_receiver, observed_receiver))
}
/// Get global heal channel sender
@@ -515,10 +577,17 @@ pub fn subscribe_heal_responses() -> broadcast::Receiver<HealChannelResponse> {
/// Send heal start request and wait for structured admission feedback.
pub async fn send_heal_request_with_receipt(request: HealChannelRequest) -> Result<HealAdmissionReceipt, String> {
let (response_tx, response_rx) = oneshot::channel();
let senders = GLOBAL_HEAL_CHANNEL_SENDERS
.get()
.ok_or_else(|| "Heal channel not initialized".to_string())?;
send_heal_request_with_receipt_on(senders, request).await
}
async fn send_heal_request_with_receipt_on(
senders: &HealChannelSenders,
request: HealChannelRequest,
) -> Result<HealAdmissionReceipt, String> {
let (response_tx, response_rx) = oneshot::channel();
senders
.receipt
.send(HealReceiptCommand { request, response_tx })
@@ -528,6 +597,46 @@ pub async fn send_heal_request_with_receipt(request: HealChannelRequest) -> Resu
.map_err(|e| format!("Failed to receive heal admission response: {e}"))?
}
/// A missing or closed observation channel falls back only before a command is accepted.
pub async fn send_heal_request_with_observation(
request: HealChannelRequest,
) -> Result<ObservedHealAdmissionReceipt, HealObservationChannelError> {
let senders = GLOBAL_HEAL_CHANNEL_SENDERS
.get()
.ok_or(HealObservationChannelError::NotInitialized)?;
send_heal_request_with_observation_on(senders, request).await
}
async fn send_heal_request_with_observation_on(
senders: &HealChannelSenders,
mut request: HealChannelRequest,
) -> Result<ObservedHealAdmissionReceipt, HealObservationChannelError> {
if let Some(observed) = &senders.observed {
let (response_tx, response_rx) = oneshot::channel();
match observed.send(ObservedHealReceiptCommand { request, response_tx }) {
Ok(()) => {
return response_rx
.await
.map_err(HealObservationChannelError::ObservedResponseClosed)?;
}
Err(error) => request = error.0.request,
}
}
let (response_tx, response_rx) = oneshot::channel();
senders
.receipt
.send(HealReceiptCommand { request, response_tx })
.map_err(|_| HealObservationChannelError::LegacyCommandClosed)?;
let receipt = response_rx
.await
.map_err(HealObservationChannelError::LegacyResponseClosed)?
.map_err(HealObservationChannelError::AdmissionFailed)?;
Ok(ObservedHealAdmissionReceipt {
receipt,
observation_id: None,
})
}
/// Send heal start request and wait for structured admission feedback.
pub async fn send_heal_request_with_admission(request: HealChannelRequest) -> Result<HealAdmissionResult, String> {
let (response_tx, response_rx) = oneshot::channel();
@@ -775,6 +884,109 @@ mod auto_heal_disk_request_tests {
mod tests {
use super::*;
#[tokio::test]
async fn observed_receipt_falls_back_only_when_no_observed_command_was_accepted() {
for closed in [false, true] {
let (command, _command_rx) = mpsc::unbounded_channel();
let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
let observed = if closed {
let (sender, receiver) = mpsc::unbounded_channel();
drop(receiver);
Some(sender)
} else {
None
};
let senders = HealChannelSenders {
command,
receipt,
observed,
};
let request = HealChannelRequest::new();
let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, request).await });
let command = receipt_rx.recv().await.expect("legacy receipt command");
command
.response_tx
.send(Ok(HealAdmissionReceipt {
result: HealAdmissionResult::Accepted,
task_id: "owner".into(),
}))
.expect("caller waiting");
let received = task.await.expect("sender task").expect("legacy admission");
assert_eq!(received.task_id, "owner");
assert_eq!(received.observation_id, None);
}
}
#[tokio::test]
async fn lost_observed_reply_never_resends_the_request() {
let (command, _command_rx) = mpsc::unbounded_channel();
let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
let keep_receipt_open = receipt.clone();
let (observed, mut observed_rx) = mpsc::unbounded_channel();
let senders = HealChannelSenders {
command,
receipt,
observed: Some(observed),
};
let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, HealChannelRequest::new()).await });
let command = observed_rx.recv().await.expect("accepted observed command");
drop(command.response_tx);
let error = task
.await
.expect("sender task")
.expect_err("lost response must remain an error");
assert!(matches!(&error, HealObservationChannelError::ObservedResponseClosed(_)));
assert!(std::error::Error::source(&error).is_some());
assert!(matches!(receipt_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)));
drop(keep_receipt_open);
}
#[tokio::test]
async fn fallback_reports_a_typed_error_if_legacy_channel_is_closed() {
let (command, _command_rx) = mpsc::unbounded_channel();
let (receipt, receipt_rx) = mpsc::unbounded_channel();
drop(receipt_rx);
let senders = HealChannelSenders {
command,
receipt,
observed: None,
};
let error = send_heal_request_with_observation_on(&senders, HealChannelRequest::new())
.await
.expect_err("closed legacy channel must not accept work");
assert!(matches!(error, HealObservationChannelError::LegacyCommandClosed));
}
#[tokio::test]
async fn observed_receipt_keeps_the_owner_token_from_the_processor() {
let (command, _command_rx) = mpsc::unbounded_channel();
let (receipt, mut receipt_rx) = mpsc::unbounded_channel();
let keep_receipt_open = receipt.clone();
let (observed, mut observed_rx) = mpsc::unbounded_channel();
let senders = HealChannelSenders {
command,
receipt,
observed: Some(observed),
};
let task = tokio::spawn(async move { send_heal_request_with_observation_on(&senders, HealChannelRequest::new()).await });
let command = observed_rx.recv().await.expect("observed command");
command
.response_tx
.send(Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: HealAdmissionResult::Merged,
task_id: "owner".into(),
},
observation_id: Some(37),
}))
.expect("caller waiting");
let received = task.await.expect("sender task").expect("observed admission");
assert_eq!(received.task_id, "owner");
assert_eq!(received.observation_id, Some(37));
assert!(matches!(receipt_rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)));
drop(keep_receipt_open);
}
#[test]
fn heal_request_source_labels_are_stable() {
assert_eq!(HealRequestSource::Scanner.as_str(), "scanner");
+145 -7
View File
@@ -21,8 +21,8 @@ use crate::heal::{
use crate::{Error, Result};
use rustfs_heal_contracts::heal_channel::{
HealAdmissionReceipt, HealAdmissionResult, HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest,
HealChannelResponse, HealOpts, HealReceiptCommand, HealReceiptReceiver, HealRequestSource, HealScanMode,
publish_heal_response,
HealChannelResponse, HealObservationChannelError, HealOpts, HealReceiptCommand, HealReceiptReceiver, HealRequestSource,
HealScanMode, ObservedHealAdmissionReceipt, ObservedHealReceiptCommand, ObservedHealReceiptReceiver, publish_heal_response,
};
use rustfs_madmin::heal_commands::HealResultItem;
use serde::Serialize;
@@ -30,6 +30,24 @@ use std::sync::Arc;
use tokio::sync::{mpsc, oneshot};
use tracing::{debug, error, info};
enum StartResponseSender {
Legacy(oneshot::Sender<std::result::Result<HealAdmissionReceipt, String>>),
Observed(oneshot::Sender<std::result::Result<ObservedHealAdmissionReceipt, HealObservationChannelError>>),
}
impl StartResponseSender {
fn send(self, result: std::result::Result<ObservedHealAdmissionReceipt, String>) {
match self {
Self::Legacy(sender) => {
let _ = sender.send(result.map(|observed| observed.receipt));
}
Self::Observed(sender) => {
let _ = sender.send(result.map_err(HealObservationChannelError::AdmissionFailed));
}
}
}
}
const LOG_COMPONENT_HEAL: &str = "heal";
const LOG_SUBSYSTEM_CHANNEL: &str = "channel";
const EVENT_HEAL_CHANNEL_STATE: &str = "heal_channel_state";
@@ -215,9 +233,22 @@ impl HealChannelProcessor {
/// Start processing legacy and canonical-receipt heal channel requests.
pub async fn start_with_receipts(
&mut self,
receiver: HealChannelReceiver,
receipt_receiver: HealReceiptReceiver,
) -> Result<()> {
let (sender, receiver_observed) = mpsc::unbounded_channel();
drop(sender);
self.start_with_observed_receipts(receiver, receipt_receiver, receiver_observed)
.await
}
/// The three command channels are FIFO independently, with no ordering across channels.
pub async fn start_with_observed_receipts(
&mut self,
mut receiver: HealChannelReceiver,
mut receipt_receiver: HealReceiptReceiver,
mut observed_receiver: ObservedHealReceiptReceiver,
) -> Result<()> {
info!(
target: "rustfs::heal::channel",
@@ -229,6 +260,7 @@ impl HealChannelProcessor {
);
let mut receipt_channel_open = true;
let mut observed_channel_open = true;
loop {
tokio::select! {
command = receiver.recv() => {
@@ -276,6 +308,23 @@ impl HealChannelProcessor {
);
}
}
command = observed_receiver.recv(), if observed_channel_open => {
let Some(ObservedHealReceiptCommand { request, response_tx }) = command else {
observed_channel_open = false;
continue;
};
if let Err(e) = self.process_start_request_inner(request, false, true, StartResponseSender::Observed(response_tx)).await {
error!(
target: "rustfs::heal::channel",
event = EVENT_HEAL_CHANNEL_REQUEST,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_CHANNEL,
state = "receipt_process_failed",
error = %e,
"Heal receipt request processing failed"
);
}
}
response = self.response_receiver.recv() => {
if let Some(response) = response {
// Handle response if needed
@@ -344,10 +393,21 @@ impl HealChannelProcessor {
/// Process start request
async fn process_start_request(
&self,
mut request: HealChannelRequest,
request: HealChannelRequest,
preserve_alias: bool,
publish_canonical_id: bool,
response_tx: oneshot::Sender<std::result::Result<HealAdmissionReceipt, String>>,
) -> Result<()> {
self.process_start_request_inner(request, preserve_alias, publish_canonical_id, StartResponseSender::Legacy(response_tx))
.await
}
async fn process_start_request_inner(
&self,
mut request: HealChannelRequest,
preserve_alias: bool,
publish_canonical_id: bool,
response_tx: StartResponseSender,
) -> Result<()> {
if request.id.is_empty() {
request.id = uuid::Uuid::new_v4().to_string();
@@ -369,7 +429,7 @@ impl HealChannelProcessor {
Ok(heal_request) => heal_request,
Err(err) => {
let error_text = err.to_string();
let _ = response_tx.send(Err(error_text.clone()));
response_tx.send(Err(error_text.clone()));
self.publish_response(HealChannelResponse {
request_id: request.id,
success: false,
@@ -383,7 +443,7 @@ impl HealChannelProcessor {
// Submit to heal manager
match self
.heal_manager
.submit_heal_request_with_receipt_and_alias(heal_request, preserve_alias)
.submit_heal_request_with_observation_and_alias(heal_request, preserve_alias)
.await
{
Ok(receipt) => {
@@ -405,7 +465,7 @@ impl HealChannelProcessor {
request.id.clone()
};
self.publish_response(admission_response(response_id, receipt.result));
let _ = response_tx.send(Ok(receipt));
response_tx.send(Ok(receipt));
}
Err(e) => {
let error_text = e.to_string();
@@ -419,7 +479,7 @@ impl HealChannelProcessor {
error = %error_text,
"Heal start submission failed"
);
let _ = response_tx.send(Err(error_text.clone()));
response_tx.send(Err(error_text.clone()));
// Send error response
let response = HealChannelResponse {
@@ -1748,6 +1808,84 @@ mod tests {
.expect("legacy alias should cancel the canonical task");
}
#[tokio::test]
async fn observed_and_legacy_channels_share_the_real_admission_owner() {
let manager = create_test_heal_manager();
let mut processor = HealChannelProcessor::new(manager);
let (command_tx, command_rx) = mpsc::unbounded_channel();
let (receipt_tx, receipt_rx) = mpsc::unbounded_channel();
let (observed_tx, observed_rx) = mpsc::unbounded_channel();
let task = tokio::spawn(async move {
processor
.start_with_observed_receipts(command_rx, receipt_rx, observed_rx)
.await
});
let request = HealChannelRequest {
id: "observed-owner".into(),
bucket: "observed-owner-bucket".into(),
object_prefix: Some("object".into()),
source: HealRequestSource::Internal,
..Default::default()
};
let (response_tx, response_rx) = oneshot::channel();
observed_tx
.send(ObservedHealReceiptCommand {
request: request.clone(),
response_tx,
})
.expect("observed command");
let first = tokio::time::timeout(Duration::from_secs(1), response_rx)
.await
.expect("observed response")
.expect("sender")
.expect("admitted");
assert_eq!(first.result, HealAdmissionResult::Accepted);
assert!(first.observation_id.is_some());
let mut duplicate = request.clone();
duplicate.id = "legacy-observed-duplicate".into();
let (response_tx, response_rx) = oneshot::channel();
receipt_tx
.send(HealReceiptCommand {
request: duplicate,
response_tx,
})
.expect("legacy command");
let legacy = tokio::time::timeout(Duration::from_secs(1), response_rx)
.await
.expect("legacy response")
.expect("sender")
.expect("merged");
assert_eq!(legacy.result, HealAdmissionResult::Merged);
assert_eq!(legacy.task_id, first.task_id);
let mut duplicate = request;
duplicate.id = "observed-duplicate".into();
let (response_tx, response_rx) = oneshot::channel();
observed_tx
.send(ObservedHealReceiptCommand {
request: duplicate,
response_tx,
})
.expect("observed duplicate");
let merged = tokio::time::timeout(Duration::from_secs(1), response_rx)
.await
.expect("observed response")
.expect("sender")
.expect("merged");
assert_eq!(merged.result, HealAdmissionResult::Merged);
assert_eq!(merged.task_id, first.task_id);
assert_eq!(merged.observation_id, first.observation_id);
drop(command_tx);
assert!(
tokio::time::timeout(Duration::from_secs(1), task)
.await
.expect("processor stops")
.expect("join")
.is_ok()
);
}
#[tokio::test]
async fn test_public_legacy_start_processes_commands_with_receipt_channel_closed() {
let manager = create_test_heal_manager();
+210 -113
View File
@@ -29,7 +29,7 @@ use rustfs_concurrency::workload::{ForegroundPressure, foreground_pressure};
#[cfg(test)]
use rustfs_concurrency::{AdmissionState, WorkloadClass};
use rustfs_heal_contracts::heal_channel::{
HealAdmissionDropReason, HealAdmissionReceipt, HealAdmissionResult, HealRequestSource,
HealAdmissionDropReason, HealAdmissionReceipt, HealAdmissionResult, HealRequestSource, ObservedHealAdmissionReceipt,
};
use rustfs_madmin::heal_commands::HealResultItem;
#[cfg(test)]
@@ -297,6 +297,7 @@ async fn publish_completed_heal(
task_id: &str,
completed: CompletedHealStatus,
terminal: bool,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
) {
let completed = Arc::new(completed);
completed.retained_bytes();
@@ -328,13 +329,16 @@ async fn publish_completed_heal(
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,
});
rustfs_heal_contracts::completion::publish_observed_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,
},
observation_id,
);
}
}
@@ -625,6 +629,38 @@ fn request_matches_request(request: &HealRequest, existing: &HealRequest) -> boo
&& request.heal_endpoints == existing.heal_endpoints
}
fn lookup_request_id_admission(
request: &HealRequest,
active_heals: &HashMap<String, Arc<HealTask>>,
queue: &PriorityHealQueue,
retrying_heals: &HashMap<String, RetryingHeal>,
durable_owners: &[HealRequest],
) -> Option<(bool, &'static str, Option<rustfs_heal_contracts::completion::HealObservationId>)> {
active_heals
.get(&request.id)
.map(|task| (request_matches_task(request, task), "active", task.observation_id))
.or_else(|| {
// A request ID can name multiple queued sequences; the token is unknown.
queue
.requests()
.find(|queued| queued.id == request.id)
.map(|queued| (request_matches_request(request, queued), "queued", None))
})
.or_else(|| {
retrying_heals
.get(&request.id)
.map(|retrying| (request_matches_request(request, &retrying.request), "retrying", retrying.observation_id))
})
// A durable-only owner must be resumed by recovery, not overwritten
// with a new execution budget by a replay before recovery completes.
.or_else(|| {
durable_owners
.iter()
.find(|owner| owner.id == request.id)
.map(|_| (false, "durable", None))
})
}
fn retrying_heal_for_dedup_key(retrying_heals: &HashMap<String, RetryingHeal>, key: &str) -> Option<(String, HealType)> {
retrying_heals
.iter()
@@ -1292,6 +1328,7 @@ impl HealManager {
config: &HealConfig,
context: &'static str,
replacement_targets_frozen: bool,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
) -> HealAdmissionDecision {
let replacement_targets_frozen = replacement_targets_frozen || request.retry_attempts != 0;
let queue_len = queue.len();
@@ -1328,7 +1365,7 @@ impl HealManager {
let priority = request.priority;
let source = request.source;
if let Some(displaced) =
queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced)
queue.push_displacing_lower_priority_where(request, Self::queued_request_can_be_displaced, observation_id)
{
publish_heal_queue_length(queue);
Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
@@ -1479,7 +1516,7 @@ impl HealManager {
let priority = request.priority;
let source = request.source;
match queue.push_with_replacement_scope(request, replacement_targets_frozen) {
match queue.push_with_replacement_scope(request, replacement_targets_frozen, observation_id) {
QueuePushOutcome::Accepted => {
publish_heal_queue_length(queue);
Self::record_admission_metric(source, HealAdmissionResult::Accepted, context);
@@ -1951,6 +1988,17 @@ impl HealManager {
request: HealRequest,
preserve_alias: bool,
) -> Result<HealAdmissionReceipt> {
Ok(self
.submit_heal_request_with_observation_and_alias(request, preserve_alias)
.await?
.receipt)
}
pub(crate) async fn submit_heal_request_with_observation_and_alias(
&self,
request: HealRequest,
preserve_alias: bool,
) -> Result<ObservedHealAdmissionReceipt> {
self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, None)
.await
}
@@ -1991,8 +2039,10 @@ impl HealManager {
mrf_notice_target: MrfRepairNoticeTarget,
) -> Result<HealAdmissionReceipt> {
let preserve_alias = !mrf_notice_target.kind.is_durable();
self.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target))
.await
Ok(self
.submit_heal_request_with_receipt_alias_and_mrf_notice(request, preserve_alias, true, Some(mrf_notice_target))
.await?
.receipt)
}
#[cfg(test)]
@@ -2058,7 +2108,7 @@ impl HealManager {
preserve_alias: bool,
accept_same_request_id_replay: bool,
mrf_notice_target: Option<MrfRepairNoticeTarget>,
) -> Result<HealAdmissionReceipt> {
) -> Result<ObservedHealAdmissionReceipt> {
let admission_start = Instant::now();
if request.id.is_empty() {
request.id = uuid::Uuid::new_v4().to_string();
@@ -2119,77 +2169,10 @@ impl HealManager {
let mut queue = self.heal_queue.lock().await;
let mut retrying_heals = self.retrying_heals.lock().await;
let request_id_admission = active_heals
.get(&request.id)
.map(|task| (request_matches_task(&request, task), "active"))
.or_else(|| {
queue
.requests()
.find(|queued| queued.id == request.id)
.map(|queued| (request_matches_request(&request, queued), "queued"))
})
.or_else(|| {
retrying_heals
.get(&request.id)
.map(|retrying| (request_matches_request(&request, &retrying.request), "retrying"))
})
// A durable-only owner must be resumed by recovery, not overwritten
// with a new execution budget by a replay before recovery completes.
.or_else(|| {
durable_owners
.iter()
.find(|owner| owner.id == request.id)
.map(|_| (false, "durable"))
});
if let Some((matches_existing, duplicate_state)) = request_id_admission {
let admission = if matches_existing {
if accept_same_request_id_replay {
HealAdmissionResult::Accepted
} else {
Self::duplicate_admission_for_request(&request, &config)
}
} else {
HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
};
if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
&& let Some(target) = mrf_notice_target
{
let mut targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets);
Self::insert_mrf_repair_notice_target(&mut targets, &request.id, target);
}
drop(retrying_heals);
drop(queue);
drop(active_heals);
let lock_timing =
HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
Self::record_admission_metric(request.source, admission, "duplicate");
self.record_admission_observation(HealAdmissionObservation {
source,
result: admission,
context: "duplicate",
force_start,
displaced: false,
start_duration: admission_start.elapsed(),
lock_timing,
});
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
duplicate_state,
result = admission.result_label(),
reason = admission.reason_label(),
"Heal queue admission reused an existing request id"
);
return Ok(HealAdmissionReceipt {
result: admission,
task_id: request.id,
});
}
let mut request_id_admission =
lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners);
if source == HealRequestSource::Admin && force_start {
if request_id_admission.is_none() && source == HealRequestSource::Admin && force_start {
let overlapping = active_heals
.values()
.map(|task| (&task.id, &task.heal_type, &task.options, task.source))
@@ -2235,6 +2218,59 @@ impl HealManager {
active_lock_wait += active_lock_acquired.duration_since(reacquire_start);
queue = self.heal_queue.lock().await;
retrying_heals = self.retrying_heals.lock().await;
// Non-admin admission can claim this id while cancellation drops the state locks.
request_id_admission = lookup_request_id_admission(&request, &active_heals, &queue, &retrying_heals, &durable_owners);
}
if let Some((matches_existing, duplicate_state, observation_id)) = request_id_admission {
let admission = if matches_existing {
if accept_same_request_id_replay {
HealAdmissionResult::Accepted
} else {
Self::duplicate_admission_for_request(&request, &config)
}
} else {
HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
};
if matches!(admission, HealAdmissionResult::Accepted | HealAdmissionResult::Merged)
&& let Some(target) = mrf_notice_target
{
let mut targets = lock_mrf_repair_notice_targets(&self.mrf_repair_notice_targets);
Self::insert_mrf_repair_notice_target(&mut targets, &request.id, target);
}
drop(retrying_heals);
drop(queue);
drop(active_heals);
let lock_timing =
HealAdmissionLockTiming::finish(lock_phase_start, active_lock_acquired, active_lock_wait, active_lock_hold);
Self::record_admission_metric(request.source, admission, "duplicate");
self.record_admission_observation(HealAdmissionObservation {
source,
result: admission,
context: "duplicate",
force_start,
displaced: false,
start_duration: admission_start.elapsed(),
lock_timing,
});
debug!(
target: "rustfs::heal::manager",
event = EVENT_HEAL_QUEUE_ADMISSION,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_MANAGER,
request_id = %request.id,
duplicate_state,
result = admission.result_label(),
reason = admission.reason_label(),
"Heal queue admission reused an existing request id"
);
return Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: admission,
task_id: request.id,
},
observation_id,
});
}
let rejection = if source == HealRequestSource::Admin {
@@ -2282,26 +2318,45 @@ impl HealManager {
None
};
let scanner_queued_owner_required = request.source == HealRequestSource::Scanner
&& mrf_notice_target.is_none()
&& request.expected_mrf_bucket_incarnation_id.is_none();
let duplicate = (!request.force_start && rejection.is_none()).then(|| {
active_heal_for_dedup_key(&active_heals, &dedup_key)
.map(|(task_id, _)| (task_id, "active"))
.or_else(|| {
queue
.queued_request_id_for_dedup_key(&dedup_key)
.map(|queued_id| (queued_id.to_string(), "queued"))
.map(|(task_id, _)| {
let observation_id = active_heals.get(&task_id).and_then(|task| task.observation_id);
(task_id, "active", observation_id)
})
.or_else(|| {
let owner = if scanner_queued_owner_required {
queue.queued_scanner_owner_for_dedup_key(&dedup_key)
} else {
queue.queued_legacy_owner_for_dedup_key(&dedup_key)
};
owner
.map(|(id, observation_id)| (id.to_string(), "queued", observation_id))
.or_else(|| {
(scanner_queued_owner_required && queue.dedup_keys.contains_key(&dedup_key))
.then(|| (request.id.clone(), "queued_owner_unknown", None))
})
})
.or_else(|| {
retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| {
let observation_id = retrying_heals.get(&task_id).and_then(|retry| retry.observation_id);
(task_id, "retrying", observation_id)
})
})
.or_else(|| retrying_heal_for_dedup_key(&retrying_heals, &dedup_key).map(|(task_id, _)| (task_id, "retrying")))
});
if let Some((merged_task_id, duplicate_state)) = duplicate.flatten() {
if let Some((merged_task_id, duplicate_state, observation_id)) = duplicate.flatten() {
// HS-06: under the minio_error overlap policy an exact duplicate
// admin start reports the typed AlreadyRunning rejection instead
// of the silent merge (MinIO's ErrHealAlreadyRunning).
let mut admission = if (request.source == HealRequestSource::Admin
&& config.overlap_policy == HealOverlapPolicy::MinioError)
let mut admission = if duplicate_state == "queued_owner_unknown"
|| (request.source == HealRequestSource::Admin && config.overlap_policy == HealOverlapPolicy::MinioError)
|| (duplicate_state != "queued" && mrf_notice_target.as_ref().is_some_and(|target| target.kind.is_durable()))
{
// A running/retrying task may have observed the object before
// this write committed. Its receipt cannot prove the new lease.
// An unknown queued owner cannot certify a receipt. A running or
// retrying task may have observed the object before this write.
HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning)
} else {
Self::duplicate_admission_for_request(&request, &config)
@@ -2400,9 +2455,12 @@ impl HealManager {
HealAdmissionResult::Accepted | HealAdmissionResult::Full => {}
}
return Ok(HealAdmissionReceipt {
result: admission,
task_id: merged_task_id,
return Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: admission,
task_id: merged_task_id,
},
observation_id,
});
}
@@ -2433,9 +2491,12 @@ impl HealManager {
result = "overlap_rejected",
"Admin heal start rejected by overlap policy"
);
return Ok(HealAdmissionReceipt {
result: HealAdmissionResult::Dropped(reason),
task_id: overlap_task_id,
return Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: HealAdmissionResult::Dropped(reason),
task_id: overlap_task_id,
},
observation_id: None,
});
}
@@ -2477,9 +2538,12 @@ impl HealManager {
result = "dropped_by_disposition",
"Heal queue admission answered from the disposition memory"
);
return Ok(HealAdmissionReceipt {
result: admission,
task_id: recorded_task_id,
return Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: admission,
task_id: recorded_task_id,
},
observation_id: None,
});
}
}
@@ -2507,17 +2571,31 @@ impl HealManager {
.lock()
.unwrap_or_else(|poisoned| poisoned.into_inner())
.contains_key(&request.id);
let admission_decision = Self::admit_request_to_queue(&mut queue, request, &config, "submit", recovered_replacement);
let mut observation_id = rustfs_heal_contracts::completion::new_heal_observation_id();
let admission_decision =
Self::admit_request_to_queue(&mut queue, request, &config, "submit", recovered_replacement, observation_id);
let admission = admission_decision.result;
if durable_handoff_required && !admission.is_admitted() {
self.root_recovery
.remove(&request_id, &request_heal_type, request_source)
.await?;
}
if admission == HealAdmissionResult::Merged
&& let Some(queued_id) = queue.queued_request_id_for_dedup_key(&dedup_key)
{
task_id = queued_id.to_owned();
if admission == HealAdmissionResult::Merged {
observation_id = None;
let owner = if scanner_queued_owner_required {
queue.queued_scanner_owner_for_dedup_key(&dedup_key)
} else {
queue.queued_legacy_owner_for_dedup_key(&dedup_key)
};
if let Some((id, owner_observation_id)) = owner {
task_id = id.to_owned();
observation_id = owner_observation_id;
} else if scanner_queued_owner_required {
return Err(Error::Other("Scanner queued owner is unavailable".to_owned()));
}
}
if !admission.is_admitted() {
observation_id = None;
}
let should_notify = matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
let displaced_task_id = admission_decision.displaced_task_id().map(ToOwned::to_owned);
@@ -2563,9 +2641,12 @@ impl HealManager {
lock_timing,
});
Ok(HealAdmissionReceipt {
result: admission,
task_id,
Ok(ObservedHealAdmissionReceipt {
receipt: HealAdmissionReceipt {
result: admission,
task_id,
},
observation_id,
})
}
@@ -2887,7 +2968,15 @@ impl HealManager {
.await?;
self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
task.cancel().await?;
publish_completed_heal(&self.completed_heals, &self.task_aliases, &canonical_task_id, completed, true).await;
publish_completed_heal(
&self.completed_heals,
&self.task_aliases,
&canonical_task_id,
completed,
true,
task.observation_id,
)
.await;
active_heals.remove(&canonical_task_id);
publish_active_heal_count(&active_heals);
info!(
@@ -2998,7 +3087,15 @@ impl HealManager {
.await?;
self.root_recovery.remove(&task.id, &task.heal_type, task.source).await?;
task.cancel().await?;
publish_completed_heal(&self.completed_heals, &self.task_aliases, task_id, completed, true).await;
publish_completed_heal(
&self.completed_heals,
&self.task_aliases,
task_id,
completed,
true,
task.observation_id,
)
.await;
}
active_heals.remove(task_id);
cancelled += 1;
+1 -1
View File
@@ -693,7 +693,7 @@ impl HealManager {
let config = config.read().await;
let mut queue = heal_queue.lock().await;
let admission_decision =
Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan", recovered_replacement);
Self::admit_request_to_queue(&mut queue, req, &config, "auto_scan", recovered_replacement, rustfs_heal_contracts::completion::new_heal_observation_id());
let admission = admission_decision.result;
let should_notify =
matches!(admission, HealAdmissionResult::Accepted) && config.event_driven_scheduler_enable;
+196 -57
View File
@@ -13,15 +13,31 @@
// limitations under the License.
/// The priority heal queue and its per-key dedup index.
use super::*;
use std::collections::BTreeMap;
/// Per-key bookkeeping for the queued-request dedup index: how many queued
/// requests hold the key, and the id of the first request that opened it —
/// the O(1) stand-in for the former heap scan when a merge receipt needs to
/// name a queued representative.
/// Per-key bookkeeping for the queued-request dedup index. The legacy
/// representative is retained for existing non-scanner callers; scanner
/// receipts use the separately tracked queued holders.
#[derive(Debug)]
pub(super) struct DedupKeyEntry {
pub(super) refcount: usize,
pub(super) representative_request_id: String,
representative_sequence: u64,
representative_observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
pub(super) scanner_queued_owners: ScannerQueuedOwners,
}
#[derive(Debug)]
pub(super) struct QueuedOwner {
request_id: String,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
}
#[derive(Debug)]
pub(super) enum ScannerQueuedOwners {
Single(u64),
// Once a key has forced duplicates, keep every live holder until it is released.
Indexed(BTreeMap<u64, QueuedOwner>),
}
/// Priority queue wrapper for heal requests
@@ -70,10 +86,25 @@ pub(super) struct PriorityQueueItem {
pub(super) sequence: u64,
pub(super) dedup_key: String,
pub(super) request: HealRequest,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
/// Execution or durable recovery fixed these targets; grace waits can have zero retries.
pub(super) replacement_targets_frozen: bool,
}
/// The in-process identity moves with the queue item, including forced duplicates.
pub(super) struct QueuedHeal {
pub(super) request: HealRequest,
pub(super) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
}
impl std::ops::Deref for QueuedHeal {
type Target = HealRequest;
fn deref(&self) -> &Self::Target {
&self.request
}
}
impl Eq for PriorityQueueItem {}
impl PartialEq for PriorityQueueItem {
@@ -283,6 +314,7 @@ pub(super) struct HealTaskAlias {
#[derive(Debug, Clone)]
pub(super) struct RetryingHeal {
pub(super) request: HealRequest,
pub(super) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
pub(super) error: String,
pub(super) cancel_token: CancellationToken,
}
@@ -306,11 +338,7 @@ impl PriorityHealQueue {
#[cfg(test)]
pub(super) fn pop_next(&mut self) -> Option<HealRequest> {
self.heap.pop().map(|item| {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
self.remove_best_effort_count(item.request.source);
item.request
})
self.heap.pop().map(|item| self.remove_queued_item(item).request)
}
pub(super) fn is_empty(&self) -> bool {
@@ -318,10 +346,15 @@ impl PriorityHealQueue {
}
pub(super) fn push(&mut self, request: HealRequest) -> QueuePushOutcome {
self.push_with_replacement_scope(request, false)
self.push_with_replacement_scope(request, false, rustfs_heal_contracts::completion::new_heal_observation_id())
}
pub(super) fn push_with_replacement_scope(&mut self, request: HealRequest, frozen: bool) -> QueuePushOutcome {
pub(super) fn push_with_replacement_scope(
&mut self,
request: HealRequest,
frozen: bool,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
) -> QueuePushOutcome {
let key = Self::make_dedup_key(&request);
// Check for duplicates unless the caller explicitly forces admission.
@@ -332,20 +365,60 @@ impl PriorityHealQueue {
// also reserves the dedup key for later non-forced duplicates. The first
// request that opens the key becomes the named representative for merge
// receipts (taken before `request` moves into the heap).
self.dedup_keys
.entry(key.clone())
.or_insert_with(|| DedupKeyEntry {
refcount: 0,
representative_request_id: request.id.clone(),
})
.refcount += 1;
self.sequence += 1;
let sequence = self.sequence;
if let Some(entry) = self.dedup_keys.get_mut(&key) {
match &mut entry.scanner_queued_owners {
ScannerQueuedOwners::Single(first_sequence) => {
let mut owners = BTreeMap::new();
if entry.refcount == 1 && entry.representative_sequence == *first_sequence {
owners.insert(
*first_sequence,
QueuedOwner {
request_id: entry.representative_request_id.clone(),
observation_id: entry.representative_observation_id,
},
);
}
owners.insert(
sequence,
QueuedOwner {
request_id: request.id.clone(),
observation_id,
},
);
entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(owners);
}
ScannerQueuedOwners::Indexed(owners) => {
owners.insert(
sequence,
QueuedOwner {
request_id: request.id.clone(),
observation_id,
},
);
}
}
entry.refcount += 1;
} else {
self.dedup_keys.insert(
key.clone(),
DedupKeyEntry {
refcount: 1,
representative_request_id: request.id.clone(),
representative_sequence: sequence,
representative_observation_id: observation_id,
scanner_queued_owners: ScannerQueuedOwners::Single(sequence),
},
);
}
let source = request.source;
self.heap.push(PriorityQueueItem {
priority: request.priority,
sequence: self.sequence,
dedup_key: key,
request,
observation_id,
replacement_targets_frozen: frozen,
});
if is_best_effort_source(source) {
@@ -371,7 +444,12 @@ impl PriorityHealQueue {
self.can_displace_lower_priority_where(priority, |_| true)
}
pub(super) fn push_displacing_lower_priority_where<F>(&mut self, request: HealRequest, can_displace: F) -> Option<HealRequest>
pub(super) fn push_displacing_lower_priority_where<F>(
&mut self,
request: HealRequest,
can_displace: F,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
) -> Option<HealRequest>
where
F: Fn(&HealRequest) -> bool,
{
@@ -402,7 +480,7 @@ impl PriorityHealQueue {
self.heap = retained;
let displaced = displaced.map(|item| {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
self.remove_best_effort_count(item.request.source);
self.refresh_dedup_representative(&item.dedup_key);
item.request
@@ -413,7 +491,7 @@ impl PriorityHealQueue {
// into `debug_assert_eq!` — in release builds (`debug_assertions` off) the whole
// macro, including its argument expression, is compiled out, which would silently
// drop the new high-priority request after having already evicted a queued item.
let outcome = self.push(request);
let outcome = self.push_with_replacement_scope(request, false, observation_id);
debug_assert_eq!(outcome, QueuePushOutcome::Accepted);
}
@@ -422,7 +500,7 @@ impl PriorityHealQueue {
#[cfg(test)]
pub(super) fn push_displacing_lower_priority(&mut self, request: HealRequest) -> Option<HealRequest> {
self.push_displacing_lower_priority_where(request, |_| true)
self.push_displacing_lower_priority_where(request, |_| true, rustfs_heal_contracts::completion::new_heal_observation_id())
}
/// Get statistics about queue contents by priority
@@ -446,14 +524,10 @@ impl PriorityHealQueue {
#[cfg(test)]
pub(super) fn pop(&mut self) -> Option<HealRequest> {
self.heap.pop().map(|item| {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
self.remove_best_effort_count(item.request.source);
item.request
})
self.heap.pop().map(|item| self.remove_queued_item(item).request)
}
pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
pub(super) fn pop_runnable_with_skips<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<QueuedHeal>, Vec<String>)
where
F: Fn(&HealRequest) -> bool,
G: Fn(&HealRequest) -> Option<String>,
@@ -475,17 +549,10 @@ impl PriorityHealQueue {
self.restore_deferred_items(deferred);
(
selected.map(|item| {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
self.remove_best_effort_count(item.request.source);
item.request
}),
skipped,
)
(selected.map(|item| self.remove_queued_item(item)), skipped)
}
pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<HealRequest>, Vec<String>)
pub(super) fn pop_runnable_with_fairness<F, G>(&mut self, can_run: F, skip_label: G) -> (Option<QueuedHeal>, Vec<String>)
where
F: Fn(&HealRequest) -> bool,
G: Fn(&HealRequest) -> Option<String>,
@@ -554,15 +621,22 @@ impl PriorityHealQueue {
self.heap = BinaryHeap::from(items);
let chosen = chosen.map(|item| {
let request = item.request;
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
self.remove_best_effort_count(request.source);
let request = self.remove_queued_item(item);
self.record_dispatch(&request);
request
});
(chosen, skipped)
}
fn remove_queued_item(&mut self, item: PriorityQueueItem) -> QueuedHeal {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
self.remove_best_effort_count(item.request.source);
QueuedHeal {
request: item.request,
observation_id: item.observation_id,
}
}
fn record_dispatch(&mut self, request: &HealRequest) {
if is_best_effort_source(request.source) {
self.mrf_normal_dispatches_since_best_effort = 0;
@@ -667,12 +741,34 @@ impl PriorityHealQueue {
}
}
pub(super) fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, key: &str) {
if let Some(entry) = dedup_keys.get_mut(key) {
fn decrement_or_remove_dedup_key(dedup_keys: &mut HashMap<String, DedupKeyEntry>, item: &PriorityQueueItem) {
if let Some(entry) = dedup_keys.get_mut(&item.dedup_key) {
match &mut entry.scanner_queued_owners {
ScannerQueuedOwners::Single(sequence) if entry.refcount != 1 || *sequence != item.sequence => {
entry.scanner_queued_owners = ScannerQueuedOwners::Indexed(BTreeMap::new());
entry.representative_observation_id = None;
}
ScannerQueuedOwners::Single(_) => {}
ScannerQueuedOwners::Indexed(owners) => {
let matches_holder = owners.len() == entry.refcount
&& owners.get(&item.sequence).is_some_and(|owner| {
owner.request_id == item.request.id && owner.observation_id == item.observation_id
});
if matches_holder {
owners.remove(&item.sequence);
} else {
owners.clear();
entry.representative_observation_id = None;
}
}
}
if entry.refcount <= 1 {
dedup_keys.remove(key);
dedup_keys.remove(&item.dedup_key);
} else {
entry.refcount -= 1;
if entry.representative_sequence == item.sequence {
entry.representative_observation_id = None;
}
}
}
}
@@ -744,18 +840,59 @@ impl PriorityHealQueue {
}
pub(super) fn queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
self.dedup_keys.get(key).map(|entry| entry.representative_request_id.as_str())
self.queued_legacy_owner_for_dedup_key(key).map(|(id, _)| id)
}
/// Re-elect the representative for `key` from the queue entries holding
/// it. Needed after a holder leaves the queue *without* becoming active
/// (canceled by id, or displaced): the former opener may be the request
/// that just left, and a merge receipt must never name an id that
/// resolves nowhere. The scheduler pop path does not need this — the
/// popped request surfaces in `active_heals` under the same id and the
/// duplicate pre-check consults active heals before the queue. No-op for
/// released keys; the survivor scan only runs when a key still has
/// holders, which under forced duplicates is the rare admin path.
pub(super) fn queued_legacy_owner_for_dedup_key(
&self,
key: &str,
) -> Option<(&str, Option<rustfs_heal_contracts::completion::HealObservationId>)> {
let entry = self.dedup_keys.get(key)?;
Some((entry.representative_request_id.as_str(), entry.representative_observation_id))
}
pub(super) fn queued_scanner_owner_for_dedup_key(
&self,
key: &str,
) -> Option<(&str, Option<rustfs_heal_contracts::completion::HealObservationId>)> {
let entry = self.dedup_keys.get(key)?;
match &entry.scanner_queued_owners {
ScannerQueuedOwners::Single(sequence) if entry.refcount == 1 && entry.representative_sequence == *sequence => {
Some((entry.representative_request_id.as_str(), entry.representative_observation_id))
}
ScannerQueuedOwners::Indexed(owners) if owners.len() == entry.refcount => owners
.first_key_value()
.map(|(_, owner)| (owner.request_id.as_str(), owner.observation_id)),
_ => None,
}
}
#[cfg(test)]
pub(super) fn scanner_queued_request_id_for_dedup_key(&self, key: &str) -> Option<&str> {
self.queued_scanner_owner_for_dedup_key(key).map(|(id, _)| id)
}
#[cfg(test)]
pub(super) fn queued_observation_for_dedup_key(
&self,
key: &str,
) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
self.queued_legacy_owner_for_dedup_key(key)
.and_then(|(_, observation_id)| observation_id)
}
#[cfg(test)]
pub(super) fn queued_observation_for_request_id(
&self,
id: &str,
) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
let mut matching = self.heap.iter().filter(|item| item.request.id == id);
let observation_id = matching.next()?.observation_id;
if matching.next().is_some() { None } else { observation_id }
}
/// Re-elect the legacy representative from the heap after cancellation or
/// displacement. Dequeue leaves it unchanged for existing callers.
pub(super) fn refresh_dedup_representative(&mut self, key: &str) {
if !self.dedup_keys.contains_key(key) {
return;
@@ -764,10 +901,12 @@ impl PriorityHealQueue {
.heap
.iter()
.find(|item| item.dedup_key == key)
.map(|item| item.request.id.clone())
.map(|item| (item.request.id.clone(), item.sequence, item.observation_id))
&& let Some(entry) = self.dedup_keys.get_mut(key)
{
entry.representative_request_id = id;
entry.representative_request_id = id.0;
entry.representative_sequence = id.1;
entry.representative_observation_id = id.2;
}
}
@@ -786,7 +925,7 @@ impl PriorityHealQueue {
while let Some(item) = self.heap.pop() {
if removed.is_none() && item.request.id == request_id {
let key = item.dedup_key.clone();
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &key);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
self.remove_best_effort_count(item.request.source);
affected_key = Some(key);
removed = Some(item.request);
@@ -812,7 +951,7 @@ impl PriorityHealQueue {
while let Some(item) = self.heap.pop() {
if should_remove(&item.request) {
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item.dedup_key);
Self::decrement_or_remove_dedup_key(&mut self.dedup_keys, &item);
self.remove_best_effort_count(item.request.source);
affected_keys.push(item.dedup_key);
removed.push(item.request);
+42 -3
View File
@@ -15,6 +15,27 @@ use super::disposition::record_task_disposition;
/// The heal scheduler: queue consumption loop and its skip/metric helpers.
use super::*;
pub(super) fn current_task_observation_id(
active: &HashMap<String, Arc<HealTask>>,
task: &Arc<HealTask>,
) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
active
.get(&task.id)
.filter(|owner| Arc::ptr_eq(owner, task))
.and_then(|_| task.observation_id)
}
pub(super) fn current_retry_observation_id(
retrying: &HashMap<String, RetryingHeal>,
task_id: &str,
cancel_token: &CancellationToken,
) -> Option<rustfs_heal_contracts::completion::HealObservationId> {
retrying
.get(task_id)
.filter(|owner| &owner.cancel_token == cancel_token)
.and_then(|owner| owner.observation_id)
}
pub(super) fn retry_admission_delay(
grace_deadline: Option<u64>,
ordinary_delay: Duration,
@@ -211,7 +232,9 @@ impl HealManager {
queue.pop_runnable_with_fairness(|_| true, |_| None).0
};
if let Some(mut request) = selected_request {
if let Some(queued) = selected_request {
let observation_id = queued.observation_id;
let mut request = queued.request;
request.options.timeout.get_or_insert(config.task_timeout);
let task_priority = request.priority;
let task_type_label = heal_request_type_label(&request).to_string();
@@ -229,6 +252,7 @@ impl HealManager {
let mainline_pacer = running_mainline_pacer(&request, &config, workload_provider);
let task = Arc::new(
HealTask::from_replacement_recovery_request(request, storage.clone(), replacement_resume_endpoint)
.with_observation_id(observation_id)
.with_mainline_pacer(mainline_pacer)
.with_admin_recovery(root_recovery.clone()),
);
@@ -336,6 +360,9 @@ impl HealManager {
tests::pause_completed_retention_before_publish(&task_id, &completed_status).await;
let mut active_heals_guard = active_heals_clone.lock().await;
let owns_completion = active_heals_guard.contains_key(&task_id);
// The legacy scheduler checks only the ID; observation needs the
// actual task object because IDs can be reused concurrently.
let observation_id = current_task_observation_id(&active_heals_guard, &task);
let cancelled_snapshot = if owns_completion {
None
} else {
@@ -428,6 +455,7 @@ impl HealManager {
request.id.clone(),
RetryingHeal {
request: request.clone(),
observation_id,
error: error.clone(),
cancel_token: cancel_token.clone(),
},
@@ -446,6 +474,7 @@ impl HealManager {
&task_id,
completed_status_entry,
terminal_completion,
observation_id,
)
.await;
}
@@ -593,8 +622,18 @@ impl HealManager {
if retry_cancel_token.is_cancelled() || !retrying.contains_key(&retry_request_id) {
return;
}
let admission_decision =
Self::admit_request_to_queue(&mut queue, retry_request.clone(), &retry_config, "retry", true);
// A later retry can reuse the ID. Only the same
// cancellation token proves this is its owner.
let retry_observation_id =
current_retry_observation_id(&retrying, &retry_request_id, &retry_cancel_token);
let admission_decision = Self::admit_request_to_queue(
&mut queue,
retry_request.clone(),
&retry_config,
"retry",
true,
retry_observation_id,
);
let admission = admission_decision.result;
let should_notify = matches!(admission, HealAdmissionResult::Accepted)
&& retry_config.event_driven_scheduler_enable;
+57 -5
View File
@@ -34,6 +34,7 @@ mod auto_scan;
mod completion;
mod disposition;
mod legacy_ambiguity_logging;
mod queued_representative;
mod running_mainline;
use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo};
@@ -935,6 +936,7 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_and_identity() {
#[tokio::test]
async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
let mut terminals = rustfs_heal_contracts::completion::subscribe_observed_heal_completions();
let manager = HealManager::new_without_root_recovery_for_test(
Arc::new(MockStorage),
Some(HealConfig {
@@ -948,10 +950,11 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
request.options.timeout = Some(Duration::from_secs(60));
let task_id = request.id.clone();
let created_at = request.created_at;
manager
.submit_heal_request(request)
let receipt = manager
.submit_heal_request_with_observation_and_alias(request, false)
.await
.expect("admit explicit-budget task");
let observation_id = receipt.observation_id.expect("accepted owner has an observation identity");
let mut previous_remaining = Duration::from_secs(60);
for expected_attempt in 1..=3 {
@@ -967,7 +970,9 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
.await
.expect("recoverable executor failure should enter retry backoff");
let retry = manager.retrying_heals.lock().await[&task_id].request.clone();
let retry_owner = manager.retrying_heals.lock().await[&task_id].clone();
assert_eq!(retry_owner.observation_id, Some(observation_id));
let retry = retry_owner.request;
assert_eq!(retry.id, task_id);
assert_eq!(retry.created_at, created_at);
assert_eq!(retry.source, HealRequestSource::Admin);
@@ -993,6 +998,10 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
.await
.expect("retry backoff should requeue the same task id");
assert_eq!(manager.operations_snapshot().await.queue_length, 1);
assert_eq!(
manager.heal_queue.lock().await.queued_observation_for_request_id(&task_id),
Some(observation_id)
);
}
process_manager_queue_once(&manager).await;
@@ -1013,6 +1022,17 @@ async fn heal_start_retry_scheduler_carries_explicit_budget_across_retries() {
);
assert_eq!(manager.operations_snapshot().await.queue_length, 0);
assert_eq!(manager.operations_snapshot().await.retrying_tasks, 0);
let terminal = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let terminal = terminals.recv().await.expect("terminal channel remains open");
if terminal.task_id == task_id {
break terminal;
}
}
})
.await
.expect("retry exhaustion emits one terminal");
assert_eq!(terminal.observation_id, Some(observation_id));
}
struct ManagerRecoveryTestHook {
@@ -1206,6 +1226,32 @@ fn test_push_displacing_lower_priority_actually_enqueues_new_request() {
assert_eq!(queue.len(), 0);
}
#[test]
fn queued_observation_tracks_the_holder_sequence_when_ids_repeat() {
let mut queue = PriorityHealQueue::new();
let first = bucket_request("same-owner-key", HealPriority::Normal, HealRequestSource::Internal);
let key = PriorityHealQueue::make_dedup_key(&first);
let mut second = first.clone();
second.force_start = true;
assert_eq!(queue.push_with_replacement_scope(first, false, Some(11)), QueuePushOutcome::Accepted);
assert_eq!(queue.push_with_replacement_scope(second, false, Some(12)), QueuePushOutcome::Accepted);
assert_eq!(queue.queued_observation_for_dedup_key(&key), Some(11));
assert_eq!(
queue.queued_observation_for_request_id(&queue.heap.peek().expect("holder").request.id),
None
);
let (first, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
assert_eq!(first.expect("first holder").observation_id, Some(11));
assert_eq!(
queue.queued_observation_for_dedup_key(&key),
None,
"a departed representative is unproven"
);
let (second, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
assert_eq!(second.expect("second holder").observation_id, Some(12));
}
#[test]
fn queued_request_id_for_dedup_key_tracks_the_representative() {
let mut queue = PriorityHealQueue::new();
@@ -1214,6 +1260,7 @@ fn queued_request_id_for_dedup_key_tracks_the_representative() {
let first_id = first.id.clone();
let first_key = PriorityHealQueue::make_dedup_key(&first);
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
assert!(queue.queued_observation_for_dedup_key(&first_key).is_some());
// A forced duplicate of the same target opens a second entry under
// the same key; the representative stays the request that opened it.
@@ -1341,6 +1388,7 @@ async fn durable_mrf_anchor_snapshot_follows_live_owners_and_ignores_stale_notic
manager.retrying_heals.lock().await.insert(
"retry-merged-active".into(),
RetryingHeal {
observation_id: None,
request: retry_request,
error: "retry pending".into(),
cancel_token: CancellationToken::new(),
@@ -2590,6 +2638,7 @@ async fn automatic_replacement_frozen_targets_reject_late_disks_but_cover_duplic
manager.retrying_heals.lock().await.insert(
owner_id.clone(),
RetryingHeal {
observation_id: None,
request: owner,
error: "target unavailable".to_string(),
cancel_token: CancellationToken::new(),
@@ -2601,7 +2650,7 @@ async fn automatic_replacement_frozen_targets_reject_late_disks_but_cover_duplic
let mut queue = manager.heal_queue.lock().await;
let context = if state == "grace" { "retry" } else { "auto_scan" };
assert_eq!(
HealManager::admit_request_to_queue(&mut queue, owner, &config, context, true).result,
HealManager::admit_request_to_queue(&mut queue, owner, &config, context, true, None).result,
HealAdmissionResult::Accepted
);
assert!(
@@ -2725,7 +2774,7 @@ fn automatic_replacement_direct_admission_checks_frozen_targets_before_capacity(
let owner_id = owner.id.clone();
assert_eq!(queue.push(owner), QueuePushOutcome::Accepted);
let admit = |queue: &mut PriorityHealQueue, request, frozen| {
HealManager::admit_request_to_queue(queue, request, &config, "auto_scan", frozen).result
HealManager::admit_request_to_queue(queue, request, &config, "auto_scan", frozen, None).result
};
assert_eq!(
admit(&mut queue, automatic_replacement_request(&["replacement-b"]), false),
@@ -3958,6 +4007,7 @@ async fn insert_retrying_request(manager: &HealManager, request: HealRequest) ->
manager.retrying_heals.lock().await.insert(
task_id.clone(),
RetryingHeal {
observation_id: None,
request: request.clone(),
error: "Lock acquisition timeout".to_string(),
cancel_token: cancel_token.clone(),
@@ -4510,6 +4560,7 @@ async fn test_operations_snapshot_counts_retry_backoff_as_owned_work() {
manager.retrying_heals.lock().await.insert(
task_id,
RetryingHeal {
observation_id: None,
request,
error: "transient".to_string(),
cancel_token: CancellationToken::new(),
@@ -4789,6 +4840,7 @@ async fn test_get_task_report_preserves_retrying_options() {
manager.retrying_heals.lock().await.insert(
task_id.clone(),
RetryingHeal {
observation_id: None,
request,
error: "transient".to_string(),
cancel_token: CancellationToken::new(),
@@ -37,6 +37,7 @@ async fn install_owner(manager: &HealManager, request: HealRequest, state: Owner
manager.retrying_heals.lock().await.insert(
id.clone(),
RetryingHeal {
observation_id: None,
request,
error: "recoverable fixture error".to_string(),
cancel_token: CancellationToken::new(),
@@ -450,6 +451,177 @@ async fn admin_overlap_normal_start_cannot_enter_force_start_cancellation_window
assert_eq!(manager.heal_queue.lock().await.len(), 1);
}
#[tokio::test]
async fn admin_force_start_rechecks_request_id_after_cancelling_overlaps() {
for source in [HealRequestSource::Scanner, HealRequestSource::Internal] {
let manager = manager_with_policy(HealOverlapPolicy::Merge);
let previous_id = install_owner(&manager, admin_prefix_request("bucket", "scope/child/"), OwnerState::Active).await;
let aliases = manager.task_aliases.lock().await;
let mut replacement = admin_prefix_request("bucket", "scope/");
replacement.force_start = true;
let request_id = replacement.id.clone();
let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement));
tokio::time::timeout(
Duration::from_secs(5),
std::future::poll_fn(|cx| {
assert!(
std::pin::pin!(tokio::task::unconstrained(forced.as_mut()))
.poll(cx)
.is_pending()
);
if manager.active_heals.try_lock().is_ok() {
Poll::Ready(())
} else {
cx.waker().wake_by_ref();
Poll::Pending
}
}),
)
.await
.expect("forceStart reaches cancellation with registry locks released");
let mut competing = HealRequest::object("other-bucket".to_string(), "object".to_string(), None);
competing.id = request_id.clone();
competing.source = source;
competing.options.recreate_missing = false;
let receipt = manager
.submit_heal_request_with_receipt(competing)
.await
.expect("non-admin admission during forceStart cancellation");
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
assert_eq!(receipt.task_id, request_id);
drop(aliases);
let receipt = forced.await.expect("forceStart returns a conflict");
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
assert_eq!(receipt.task_id, request_id);
let queue = manager.heal_queue.lock().await;
let same_id: Vec<_> = queue.requests().filter(|request| request.id == request_id).collect();
assert_eq!(same_id.len(), 1, "conflicting requests must not share an id");
assert_eq!(same_id[0].source, source);
drop(queue);
assert!(
matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)),
"overlapping admin cancellation precedes the conflict recheck"
);
}
}
#[tokio::test]
async fn forced_id_conflict_preserves_scanner_queued_owner() {
for competing_source in [HealRequestSource::Scanner, HealRequestSource::Internal] {
let manager = manager_with_policy(HealOverlapPolicy::Merge);
let previous_id = install_owner(&manager, admin_object_request("object"), OwnerState::Active).await;
let mut queued_ids = Vec::new();
for source in [
HealRequestSource::Scanner,
HealRequestSource::Internal,
HealRequestSource::Scanner,
] {
let mut request = HealRequest::object("bucket".to_string(), "object".to_string(), None);
request.source = source;
request.force_start = true;
let id = request.id.clone();
let receipt = manager
.submit_heal_request_with_receipt(request)
.await
.expect("queue forced work");
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
assert_eq!(receipt.task_id, id);
queued_ids.push(id);
}
// The first holder leaves the queue, leaving two forced tasks queued.
let selected = {
let mut queue = manager.heal_queue.lock().await;
let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None);
assert!(skipped.is_empty());
let selected = selected.expect("first queued task should run");
assert_eq!(selected.id, queued_ids[0]);
assert_eq!(queue.len(), 2);
selected
};
let QueuedHeal { request, observation_id } = selected;
let completed_task = HealTask::from_request(request, manager.storage.clone()).with_observation_id(observation_id);
let mut completed = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
Arc::make_mut(completed.outcome.as_mut().expect("completed task outcome")).finish(None);
manager
.completed_heals
.lock()
.await
.insert(queued_ids[0].clone(), Arc::new(completed));
let aliases = manager.task_aliases.lock().await;
let mut replacement = admin_object_request("object");
replacement.force_start = true;
let competing_id = replacement.id.clone();
let mut forced = Box::pin(manager.submit_heal_request_with_receipt(replacement));
tokio::time::timeout(
Duration::from_secs(5),
std::future::poll_fn(|cx| {
assert!(
std::pin::pin!(tokio::task::unconstrained(forced.as_mut()))
.poll(cx)
.is_pending()
);
if manager.active_heals.try_lock().is_ok() {
Poll::Ready(())
} else {
cx.waker().wake_by_ref();
Poll::Pending
}
}),
)
.await
.expect("forceStart reaches cancellation with registry locks released");
let mut competing = HealRequest::object("bucket".to_string(), "object".to_string(), None);
competing.id = competing_id.clone();
competing.source = competing_source;
competing.force_start = true;
let receipt = manager
.submit_heal_request_with_receipt(competing)
.await
.expect("competing same-key forced work");
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
assert_eq!(receipt.task_id, competing_id);
drop(aliases);
let receipt = forced.await.expect("forceStart returns an ID conflict");
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
assert_eq!(receipt.task_id, competing_id);
assert!(!manager.active_heals.lock().await.contains_key(&competing_id));
assert!(!manager.retrying_heals.lock().await.contains_key(&competing_id));
assert!(matches!(manager.get_task_status(&previous_id).await, Ok(HealTaskStatus::Cancelled)));
{
let queue = manager.heal_queue.lock().await;
let same_id: Vec<_> = queue.requests().filter(|request| request.id == competing_id).collect();
assert_eq!(same_id.len(), 1, "the competing ID must have one live owner");
assert_eq!(same_id[0].source, competing_source);
assert_eq!(queue.len(), 3);
}
let mut scanner = HealRequest::object("bucket".to_string(), "object".to_string(), None);
scanner.source = HealRequestSource::Scanner;
let receipt = manager
.submit_heal_request_with_receipt(scanner)
.await
.expect("scanner should merge into queued work");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_ne!(receipt.task_id, queued_ids[0], "the completed holder cannot own a queued receipt");
assert!(
manager
.heal_queue
.lock()
.await
.requests()
.any(|request| request.id == receipt.task_id),
"the canonical receipt must name a live queued task"
);
}
}
#[tokio::test]
async fn admin_overlap_cancel_cannot_miss_a_queued_to_active_transition() {
let manager = manager_with_policy(HealOverlapPolicy::Merge);
@@ -1,6 +1,208 @@
use super::*;
use crate::heal::outcome::{HealObjectIdentity, HealObjectKind, HealObjectOutcome};
use rustfs_heal_contracts::{completion::subscribe_heal_completions, heal_channel::HealScanMode};
use rustfs_heal_contracts::{
completion::{subscribe_heal_completions, subscribe_observed_heal_completions},
heal_channel::HealScanMode,
};
#[test]
fn observation_requires_the_current_task_or_retry_object() {
let request = bucket_request("observation-object-owner", HealPriority::Normal, HealRequestSource::Internal);
let id = request.id.clone();
let stale = Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(301)));
let current = Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(302)));
let active = HashMap::from([(id.clone(), Arc::clone(&current))]);
assert_eq!(current_task_observation_id(&active, &stale), None);
assert_eq!(current_task_observation_id(&active, &current), Some(302));
let current_token = CancellationToken::new();
let stale_token = CancellationToken::new();
let retrying = HashMap::from([(
id.clone(),
RetryingHeal {
request,
observation_id: Some(303),
error: "retrying".into(),
cancel_token: current_token.clone(),
},
)]);
assert_eq!(current_retry_observation_id(&retrying, &id, &stale_token), None);
assert_eq!(current_retry_observation_id(&retrying, &id, &current_token), Some(303));
}
#[tokio::test]
async fn queued_same_id_replay_is_unknown_but_canonical_merge_retains_owner_identity() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let request = bucket_request("queued-id-replay-observation", HealPriority::Normal, HealRequestSource::Internal);
let first = manager
.submit_heal_request_with_observation_and_alias(request.clone(), false)
.await
.expect("first owner");
assert!(first.observation_id.is_some());
let replay = manager
.submit_heal_request_with_observation_and_alias(request, false)
.await
.expect("same ID replay");
assert_eq!(replay.result, HealAdmissionResult::Accepted);
assert_eq!(replay.task_id, first.task_id);
assert_eq!(replay.observation_id, None);
let duplicate = bucket_request("queued-id-replay-observation", HealPriority::Normal, HealRequestSource::Scanner);
let merged = manager
.submit_heal_request_with_observation_and_alias(duplicate, false)
.await
.expect("canonical merge");
assert_eq!(merged.result, HealAdmissionResult::Merged);
assert_eq!(merged.task_id, first.task_id);
assert_eq!(merged.observation_id, first.observation_id);
}
#[tokio::test]
async fn merged_receipt_uses_live_active_or_retrying_owner_identity() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let request = bucket_request("owner-observation-states", HealPriority::Normal, HealRequestSource::Internal);
let id = request.id.clone();
manager.active_heals.lock().await.insert(
id.clone(),
Arc::new(HealTask::from_request(request.clone(), Arc::new(MockStorage)).with_observation_id(Some(201))),
);
let scanner = bucket_request("owner-observation-states", HealPriority::Normal, HealRequestSource::Scanner);
let active = manager
.submit_heal_request_with_observation_and_alias(scanner.clone(), false)
.await
.expect("active merge");
assert_eq!(active.result, HealAdmissionResult::Merged);
assert_eq!(active.task_id, id);
assert_eq!(active.observation_id, Some(201));
manager.active_heals.lock().await.remove(&id);
manager.retrying_heals.lock().await.insert(
id.clone(),
RetryingHeal {
request,
observation_id: Some(202),
error: "retrying".into(),
cancel_token: CancellationToken::new(),
},
);
let mut scanner_retry = scanner;
scanner_retry.id = Uuid::new_v4().to_string();
let retrying = manager
.submit_heal_request_with_observation_and_alias(scanner_retry, false)
.await
.expect("retry merge");
assert_eq!(retrying.result, HealAdmissionResult::Merged);
assert_eq!(retrying.task_id, id);
assert_eq!(retrying.observation_id, Some(202));
}
#[tokio::test]
async fn cancelled_or_displaced_queue_owner_does_not_reuse_its_observation_identity() {
for displace in [false, true] {
let manager = HealManager::new_without_root_recovery_for_test(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 1,
event_driven_scheduler_enable: false,
..Default::default()
}),
);
let request = bucket_request("removed-observation-owner", HealPriority::Low, HealRequestSource::Scanner);
let old = manager
.submit_heal_request_with_observation_and_alias(request.clone(), false)
.await
.expect("first owner");
assert_eq!(old.result, HealAdmissionResult::Accepted);
assert!(old.observation_id.is_some());
if displace {
let urgent = bucket_request("displacing-owner", HealPriority::Urgent, HealRequestSource::Internal);
let urgent_id = urgent.id.clone();
let admitted = manager
.submit_heal_request_with_observation_and_alias(urgent, false)
.await
.expect("displacing owner");
assert_eq!(admitted.result, HealAdmissionResult::Accepted);
assert!(
manager
.heal_queue
.lock()
.await
.requests()
.all(|queued| queued.id != old.task_id)
);
manager.cancel_task(&urgent_id).await.expect("free queue after displacement");
} else {
manager.cancel_task(&old.task_id).await.expect("cancel queued owner");
}
let next = manager
.submit_heal_request_with_observation_and_alias(request, false)
.await
.expect("readmit same id");
assert_eq!(next.result, HealAdmissionResult::Accepted);
assert_eq!(next.task_id, old.task_id);
assert!(next.observation_id.is_some());
assert_ne!(next.observation_id, old.observation_id);
}
}
#[tokio::test]
async fn reused_id_has_distinct_owner_generations_in_receipts_and_terminals() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let mut receiver = subscribe_observed_heal_completions();
let request = bucket_request("observation-id-reuse", HealPriority::Normal, HealRequestSource::Internal);
let first = manager
.submit_heal_request_with_observation_and_alias(request.clone(), false)
.await
.expect("first owner");
let mut scanner = bucket_request("observation-id-reuse", HealPriority::Normal, HealRequestSource::Scanner);
let merged = manager
.submit_heal_request_with_observation_and_alias(scanner.clone(), false)
.await
.expect("first merge");
assert_eq!(merged.result, HealAdmissionResult::Merged);
assert_eq!(merged.task_id, first.task_id);
assert!(first.observation_id.is_some());
assert_eq!(merged.observation_id, first.observation_id);
process_manager_queue_once(&manager).await;
let first_terminal = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let terminal = receiver.recv().await.expect("terminal channel");
if terminal.task_id == first.task_id {
break terminal;
}
}
})
.await
.expect("first terminal");
assert_eq!(first_terminal.observation_id, first.observation_id);
let second = manager
.submit_heal_request_with_observation_and_alias(request, false)
.await
.expect("reused id");
assert_eq!(second.result, HealAdmissionResult::Accepted);
assert_eq!(second.task_id, first.task_id);
assert_ne!(second.observation_id, first.observation_id);
scanner.id = Uuid::new_v4().to_string();
let merged_again = manager
.submit_heal_request_with_observation_and_alias(scanner, false)
.await
.expect("second merge");
assert_eq!(merged_again.result, HealAdmissionResult::Merged);
assert_eq!(merged_again.observation_id, second.observation_id);
process_manager_queue_once(&manager).await;
let second_terminal = tokio::time::timeout(Duration::from_secs(5), async {
loop {
let terminal = receiver.recv().await.expect("terminal channel");
if terminal.task_id == second.task_id && terminal.observation_id == second.observation_id {
break terminal;
}
}
})
.await
.expect("second terminal");
assert_eq!(second_terminal.observation_id, second.observation_id);
}
#[tokio::test]
async fn canonical_completion_publishes_typed_objects_only_for_terminal_untruncated_results() {
@@ -43,12 +245,12 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
error: "retry".into(),
retry_attempt: 1,
};
publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false).await;
publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), false, Some(1)).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;
publish_completed_heal(&completed_heals, &aliases, &id, completed.clone(), true, Some(1)).await;
let observed = tokio::time::timeout(Duration::from_secs(1), async {
loop {
let event = receiver.recv().await.expect("terminal broadcaster stays open");
@@ -73,7 +275,7 @@ async fn canonical_completion_publishes_typed_objects_only_for_terminal_untrunca
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;
publish_completed_heal(&completed_heals, &aliases, &truncated_id, completed, true, Some(1)).await;
let truncated = tokio::time::timeout(Duration::from_secs(1), async {
loop {
let event = receiver.recv().await.expect("terminal broadcaster stays open");
@@ -0,0 +1,653 @@
// 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.
use super::*;
fn object_request(source: HealRequestSource, priority: HealPriority, force_start: bool) -> HealRequest {
let mut request = HealRequest::new(
HealType::Object {
bucket: "canonical-owner".to_owned(),
object: "object".to_owned(),
version_id: None,
},
HealOptions::default(),
priority,
);
request.source = source;
request.force_start = force_start;
request
}
fn scoped_object_request(
source: HealRequestSource,
version_id: &str,
pool_index: Option<usize>,
set_index: Option<usize>,
force_start: bool,
) -> HealRequest {
let mut request = object_request(source, HealPriority::Low, force_start);
let HealType::Object { version_id: version, .. } = &mut request.heal_type else {
unreachable!("object_request always constructs an object heal")
};
*version = Some(version_id.to_owned());
request.options.pool_index = pool_index;
request.options.set_index = set_index;
request.options.recreate_missing = source != HealRequestSource::Scanner;
request
}
#[tokio::test]
async fn scanner_receipt_tracks_scoped_read_repair_forced_holders() {
let manager = HealManager::new_without_root_recovery_for_test(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 16,
low_priority_merge_enable: true,
..HealConfig::default()
}),
);
let version = "00000000-0000-4000-8000-000000000001";
let original = scoped_object_request(HealRequestSource::ReadRepair, version, Some(0), Some(1), false);
let original_id = original.id.clone();
let key = PriorityHealQueue::make_dedup_key(&original);
let admitted = manager
.submit_heal_request_with_receipt(original)
.await
.expect("admit read-repair owner");
assert_eq!(admitted.result, HealAdmissionResult::Accepted);
assert_eq!(admitted.task_id, original_id);
let receipt = manager
.submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
.await
.expect("scanner merges into the scoped read-repair owner");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, original_id);
let first_forced = scoped_object_request(HealRequestSource::Internal, version, Some(0), Some(1), true);
let first_forced_id = first_forced.id.clone();
let second_forced = scoped_object_request(HealRequestSource::Internal, version, Some(0), Some(1), true);
let second_forced_id = second_forced.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
assert_eq!(queue.push(first_forced), QueuePushOutcome::Accepted);
assert_eq!(queue.push(second_forced), QueuePushOutcome::Accepted);
let (selected, skipped) = queue.pop_runnable_with_skips(|_| true, |_| None);
assert!(skipped.is_empty());
assert_eq!(selected.expect("read-repair owner should leave first").id, original_id);
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(original_id.as_str()));
}
let receipt = manager
.submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
.await
.expect("scanner merges into the remaining owner");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, first_forced_id);
{
let mut queue = manager.heal_queue.lock().await;
let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_forced_id, |_| None);
assert_eq!(selected.expect("first forced holder should leave").id, first_forced_id);
assert!(queue.contains_request_id(&second_forced_id));
}
let receipt = manager
.submit_heal_request_with_receipt(scoped_object_request(HealRequestSource::Scanner, version, Some(0), Some(1), false))
.await
.expect("scanner merges into the last queued holder");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, second_forced_id);
}
#[tokio::test]
async fn scanner_receipt_does_not_merge_read_repair_with_a_different_scope_or_version() {
let version = "00000000-0000-4000-8000-000000000001";
let other_version = "00000000-0000-4000-8000-000000000002";
for (candidate_version, pool, set) in [
(version, None, None),
(version, Some(1), Some(1)),
(version, Some(0), Some(2)),
(other_version, Some(0), Some(1)),
] {
let manager = HealManager::new_without_root_recovery_for_test(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 16,
low_priority_merge_enable: true,
..HealConfig::default()
}),
);
let original = scoped_object_request(HealRequestSource::ReadRepair, version, Some(0), Some(1), false);
let original_id = original.id.clone();
let initial = manager
.submit_heal_request_with_receipt(original)
.await
.expect("admit read-repair owner");
assert_eq!(initial.result, HealAdmissionResult::Accepted);
assert_eq!(initial.task_id, original_id);
let candidate = scoped_object_request(HealRequestSource::Scanner, candidate_version, pool, set, false);
let candidate_id = candidate.id.clone();
let receipt = manager
.submit_heal_request_with_receipt(candidate)
.await
.expect("admit scanner with a distinct scope or version");
assert_eq!(
receipt.result,
HealAdmissionResult::Accepted,
"version={candidate_version}, pool={pool:?}, set={set:?}"
);
assert_eq!(receipt.task_id, candidate_id);
let queue = manager.heal_queue.lock().await;
assert!(queue.contains_request_id(&original_id));
assert!(queue.contains_request_id(&candidate_id));
}
}
async fn scanner_receipt_tracks_queued_owner(use_fairness: bool) {
let manager = HealManager::new_without_root_recovery_for_test(
Arc::new(MockStorage),
Some(HealConfig {
queue_size: 16,
low_priority_merge_enable: true,
overlap_policy: HealOverlapPolicy::Merge,
..HealConfig::default()
}),
);
if use_fairness {
let mut queue = manager.heal_queue.lock().await;
for index in 0..4 {
let request = bucket_request(&format!("internal-{index}"), HealPriority::Normal, HealRequestSource::Internal);
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
for _ in 0..4 {
let (request, _) = queue.pop_runnable_with_fairness(|_| true, |_| None);
assert_eq!(request.expect("internal work should run").source, HealRequestSource::Internal);
}
}
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
let second_id = second.id.clone();
for request in [first, second] {
assert_eq!(
manager
.submit_heal_request_with_receipt(request)
.await
.expect("admit task")
.result,
HealAdmissionResult::Accepted
);
}
let first = {
let mut queue = manager.heal_queue.lock().await;
let (selected, skipped) = if use_fairness {
queue.pop_runnable_with_fairness(|_| true, |_| None)
} else {
queue.pop_runnable_with_skips(|_| true, |_| None)
};
assert!(skipped.is_empty());
let selected = selected.expect("the first same-priority task should run");
assert_eq!(selected.id, first_id);
assert!(queue.contains_request_id(&second_id));
assert_eq!(
queue.queued_request_id_for_dedup_key(&key),
Some(first_id.as_str()),
"the existing representative for other sources must remain unchanged"
);
selected
};
let QueuedHeal { request, observation_id } = first;
manager.active_heals.lock().await.insert(
first_id.clone(),
Arc::new(HealTask::from_request(request, Arc::new(MockStorage)).with_observation_id(observation_id)),
);
let active = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
.await
.expect("scanner admission while the first task is active");
assert_eq!(active.result, HealAdmissionResult::Merged);
assert_eq!(active.task_id, first_id, "active ownership keeps precedence");
let completed_task = manager
.active_heals
.lock()
.await
.remove(&first_id)
.expect("first task should leave active ownership");
let mut retained = CompletedHealStatus::snapshot(&completed_task, HealTaskStatus::Completed).await;
Arc::make_mut(retained.outcome.as_mut().expect("retained task outcome")).finish(None);
manager
.completed_heals
.lock()
.await
.insert(first_id.clone(), Arc::new(retained));
let queued = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::Low, false))
.await
.expect("scanner admission after the first task leaves active state");
assert_eq!(queued.result, HealAdmissionResult::Merged);
assert_eq!(queued.task_id, second_id, "the receipt must name the remaining queued task");
}
#[tokio::test]
async fn scanner_receipt_tracks_queued_owner_after_normal_dequeue() {
scanner_receipt_tracks_queued_owner(false).await;
}
#[tokio::test]
async fn scanner_receipt_tracks_queued_owner_after_fair_dequeue() {
scanner_receipt_tracks_queued_owner(true).await;
}
#[tokio::test]
async fn scanner_receipt_keeps_queued_owner_when_a_different_forced_task_runs() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
let first_id = first.id.clone();
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
let second_id = second.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
assert_eq!(selected.expect("higher-priority forced task should run").id, second_id);
}
let receipt = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
.await
.expect("scanner admission after a non-representative task leaves the queue");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, first_id);
}
#[tokio::test]
async fn scanner_receipt_reselects_after_the_queued_owner_is_cancelled() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
let third_id = third.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
for request in [first, second, third] {
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
assert_eq!(selected.expect("first task should run").id, first_id);
}
let first_receipt = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
.await
.expect("first scanner admission");
assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
let remaining = if first_receipt.task_id == second_id {
third_id
} else {
second_id
};
manager.heal_queue.lock().await.remove_request_id(&first_receipt.task_id);
let next_receipt = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
.await
.expect("scanner admission after the queued owner is removed");
assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
assert_eq!(next_receipt.task_id, remaining);
}
#[tokio::test]
async fn scanner_receipt_reselects_after_multiple_forced_dequeues() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
let third_id = third.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
for request in [first, second, third] {
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
assert_eq!(selected.expect("first task should run").id, first_id);
}
let first_receipt = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
.await
.expect("first scanner admission");
assert_eq!(first_receipt.result, HealAdmissionResult::Merged);
assert!(first_receipt.task_id == second_id || first_receipt.task_id == third_id);
let remaining = if first_receipt.task_id == second_id {
third_id
} else {
second_id
};
{
let mut queue = manager.heal_queue.lock().await;
let (selected, _) = queue.pop_runnable_with_skips(|request| request.id == first_receipt.task_id, |_| None);
assert_eq!(selected.expect("selected queued owner should run").id, first_receipt.task_id);
}
let next_receipt = manager
.submit_heal_request_with_receipt(object_request(HealRequestSource::Scanner, HealPriority::High, false))
.await
.expect("scanner admission after another queued owner starts");
assert_eq!(next_receipt.result, HealAdmissionResult::Merged);
assert_eq!(next_receipt.task_id, remaining);
}
#[test]
fn scanner_owner_index_survives_a_single_remaining_holder_and_new_forced_work() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
let third_id = third.id.clone();
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
assert_eq!(queue.pop().expect("first queued task").id, first_id);
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(first_id.as_str()));
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
assert_eq!(queue.push(third), QueuePushOutcome::Accepted);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
assert_eq!(queue.pop().expect("second queued task").id, second_id);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
}
#[test]
fn scanner_owner_index_distinguishes_holders_with_the_same_request_id() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let key = PriorityHealQueue::make_dedup_key(&first);
let shared_id = first.id.clone();
let mut second = object_request(HealRequestSource::Admin, HealPriority::High, true);
second.id.clone_from(&shared_id);
let third = object_request(HealRequestSource::Internal, HealPriority::High, true);
let third_id = third.id.clone();
for request in [first, second, third] {
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
assert_eq!(queue.pop().expect("first holder").id, shared_id);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(shared_id.as_str()));
assert_eq!(queue.remove_request_id(&shared_id).expect("second holder").id, shared_id);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
}
#[test]
fn scanner_owner_index_tracks_bulk_removal_and_key_release() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::Low, true);
let third_id = third.id.clone();
for request in [first, second, third] {
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
let removed = queue.remove_matching(|request| request.id == first_id || request.id == second_id);
assert_eq!(removed.len(), 2);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
assert_eq!(queue.remove_matching(|request| request.id == third_id).len(), 1);
assert!(!queue.dedup_keys.contains_key(&key));
let next = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let next_id = next.id.clone();
assert_eq!(queue.push(next), QueuePushOutcome::Accepted);
assert!(matches!(&queue.dedup_keys[&key].scanner_queued_owners, ScannerQueuedOwners::Single(_)));
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(next_id.as_str()));
}
#[test]
fn scanner_owner_index_tracks_displacement_without_changing_legacy_reselection() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::Normal, true);
let second_id = second.id.clone();
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
let incoming = bucket_request("displacement", HealPriority::Urgent, HealRequestSource::Internal);
let displaced = queue
.push_displacing_lower_priority(incoming)
.expect("lower priority task should be displaced");
assert_eq!(displaced.id, first_id);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
}
#[tokio::test]
async fn scanner_receipt_rejects_a_missing_or_empty_holder_index() {
for missing in [true, false] {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::High, true);
{
let mut queue = manager.heal_queue.lock().await;
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
assert_eq!(queue.push(second), QueuePushOutcome::Accepted);
let entry = queue.dedup_keys.get_mut(&key).expect("dedup key should exist");
entry.scanner_queued_owners = if missing {
ScannerQueuedOwners::Single(0)
} else {
ScannerQueuedOwners::Indexed(std::collections::BTreeMap::new())
};
}
for leave_one in [false, true] {
if leave_one {
manager.heal_queue.lock().await.pop().expect("one holder should leave");
}
let request = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let request_id = request.id.clone();
let receipt = manager
.submit_heal_request_with_receipt(request)
.await
.expect("admission should fail closed");
assert_eq!(receipt.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::AlreadyRunning));
assert_eq!(receipt.task_id, request_id);
}
}
}
#[test]
fn scanner_and_legacy_owners_keep_their_distinct_reselection_orders() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Admin, HealPriority::Low, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::Normal, true);
let third_id = third.id.clone();
for request in [first, second, third] {
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
assert_eq!(queue.remove_request_id(&first_id).expect("first holder should be removed").id, first_id);
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(second_id.as_str()));
assert_eq!(queue.queued_request_id_for_dedup_key(&key), Some(third_id.as_str()));
}
#[test]
fn scanner_owner_index_tracks_a_series_of_forced_holders() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let key = PriorityHealQueue::make_dedup_key(&first);
let mut ids = vec![first.id.clone()];
assert_eq!(queue.push(first), QueuePushOutcome::Accepted);
for _ in 0..128 {
let request = object_request(HealRequestSource::Internal, HealPriority::High, true);
ids.push(request.id.clone());
assert_eq!(queue.push(request), QueuePushOutcome::Accepted);
}
for id in &ids {
assert_eq!(queue.scanner_queued_request_id_for_dedup_key(&key), Some(id.as_str()));
assert_eq!(&queue.pop().expect("indexed holder should remain queued").id, id);
}
assert!(!queue.dedup_keys.contains_key(&key));
}
#[tokio::test]
async fn observed_scanner_and_legacy_receipts_pair_their_own_reselected_holders() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Internal, HealPriority::Low, true);
let second_id = second.id.clone();
let third = object_request(HealRequestSource::Internal, HealPriority::Normal, true);
let third_id = third.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
for (request, observation_id) in [(first, 11), (second, 12), (third, 13)] {
assert_eq!(
queue.push_with_replacement_scope(request, false, Some(observation_id)),
QueuePushOutcome::Accepted
);
}
assert_eq!(queue.remove_request_id(&first_id).expect("remove first holder").id, first_id);
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), Some((second_id.as_str(), Some(12))));
assert_eq!(queue.queued_legacy_owner_for_dedup_key(&key), Some((third_id.as_str(), Some(13))));
}
for (source, expected_id, expected_observation) in [
(HealRequestSource::Scanner, &second_id, 12),
(HealRequestSource::Internal, &third_id, 13),
] {
let receipt = manager
.submit_heal_request_with_observation_and_alias(object_request(source, HealPriority::High, false), false)
.await
.expect("merge into queued owner");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(&receipt.task_id, expected_id);
assert_eq!(receipt.observation_id, Some(expected_observation));
}
}
#[tokio::test]
async fn observed_same_id_holders_use_the_selected_sequence_not_the_id() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let mut second = object_request(HealRequestSource::Internal, HealPriority::High, true);
second.id.clone_from(&id);
{
let mut queue = manager.heal_queue.lock().await;
assert_eq!(queue.push_with_replacement_scope(first, false, Some(21)), QueuePushOutcome::Accepted);
assert_eq!(queue.push_with_replacement_scope(second, false, Some(22)), QueuePushOutcome::Accepted);
let (selected, _) = queue.pop_runnable_with_skips(|_| true, |_| None);
let selected = selected.expect("first holder should run");
assert_eq!(selected.id, id);
assert_eq!(selected.observation_id, Some(21));
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), Some((id.as_str(), Some(22))));
assert_eq!(queue.queued_legacy_owner_for_dedup_key(&key), Some((id.as_str(), None)));
}
for (source, expected_observation) in [(HealRequestSource::Scanner, Some(22)), (HealRequestSource::Internal, None)] {
let receipt = manager
.submit_heal_request_with_observation_and_alias(object_request(source, HealPriority::High, false), false)
.await
.expect("merge into the selected holder");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, id);
assert_eq!(receipt.observation_id, expected_observation);
}
}
#[test]
fn displaced_holder_reselects_the_remaining_id_and_observation_together() {
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::Low, false);
let first_id = first.id.clone();
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Internal, HealPriority::Normal, true);
let second_id = second.id.clone();
assert_eq!(queue.push_with_replacement_scope(first, false, Some(31)), QueuePushOutcome::Accepted);
assert_eq!(queue.push_with_replacement_scope(second, false, Some(32)), QueuePushOutcome::Accepted);
let incoming = bucket_request("observation-displacement", HealPriority::Urgent, HealRequestSource::Internal);
let displaced = queue
.push_displacing_lower_priority_where(incoming, |_| true, Some(33))
.expect("lower-priority holder should be displaced");
assert_eq!(displaced.id, first_id);
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), Some((second_id.as_str(), Some(32))));
assert_eq!(queue.queued_legacy_owner_for_dedup_key(&key), Some((second_id.as_str(), Some(32))));
}
#[tokio::test]
async fn missing_selected_token_or_inconsistent_index_stays_unknown() {
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Internal, HealPriority::High, true);
let second_id = second.id.clone();
{
let mut queue = manager.heal_queue.lock().await;
assert_eq!(queue.push_with_replacement_scope(first, false, Some(41)), QueuePushOutcome::Accepted);
assert_eq!(queue.push_with_replacement_scope(second, false, None), QueuePushOutcome::Accepted);
queue.pop().expect("first holder should leave");
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), Some((second_id.as_str(), None)));
}
let receipt = manager
.submit_heal_request_with_observation_and_alias(
object_request(HealRequestSource::Scanner, HealPriority::High, false),
false,
)
.await
.expect("merge into owner without an observation token");
assert_eq!(receipt.result, HealAdmissionResult::Merged);
assert_eq!(receipt.task_id, second_id);
assert_eq!(receipt.observation_id, None);
let mut queue = PriorityHealQueue::new();
let first = object_request(HealRequestSource::Scanner, HealPriority::High, false);
let key = PriorityHealQueue::make_dedup_key(&first);
let second = object_request(HealRequestSource::Internal, HealPriority::High, true);
assert_eq!(queue.push_with_replacement_scope(first, false, Some(51)), QueuePushOutcome::Accepted);
assert_eq!(queue.push_with_replacement_scope(second, false, Some(52)), QueuePushOutcome::Accepted);
let ScannerQueuedOwners::Indexed(owners) = &mut queue.dedup_keys.get_mut(&key).expect("key").scanner_queued_owners else {
panic!("forced holder should create an index");
};
let first_sequence = *owners.first_key_value().expect("first indexed holder").0;
assert!(owners.remove(&first_sequence).is_some());
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), None);
queue.pop().expect("first holder should leave");
assert_eq!(queue.queued_scanner_owner_for_dedup_key(&key), None);
assert_eq!(queue.queued_legacy_owner_for_dedup_key(&key).and_then(|(_, token)| token), None);
}
@@ -448,6 +448,7 @@ async fn root_recovery_retry_cancellation_preserves_previous_attempt_outcome() {
manager.retrying_heals.lock().await.insert(
request.id.clone(),
RetryingHeal {
observation_id: None,
request: request.clone(),
error: "retryable".to_string(),
cancel_token: CancellationToken::new(),
@@ -2153,6 +2154,30 @@ async fn root_recovery_cluster_rejects_inventory_without_checkpoint_headroom() {
);
}
#[tokio::test]
async fn root_recovery_replayed_queue_owner_gets_a_fresh_in_process_observation_identity() {
let (_temp, disk) = recovery_disk().await;
let manager = recovery_manager(vec![disk.clone()]);
let request = root_request();
let first = manager
.submit_heal_request_with_observation_and_alias(request.clone(), false)
.await
.expect("persisted admission");
assert_eq!(first.result, HealAdmissionResult::Accepted);
assert!(first.observation_id.is_some());
drop(manager);
let replayed = recovery_manager(vec![disk]);
replayed.replay_root_heals().await.expect("replay persisted root");
let recovered_id = replayed
.heal_queue
.lock()
.await
.queued_observation_for_request_id(&request.id);
assert!(recovered_id.is_some());
assert_ne!(recovered_id, first.observation_id);
}
#[tokio::test]
async fn root_recovery_queued_owner_is_not_priority_displaced() {
let (_temp, disk) = recovery_disk().await;
+11
View File
@@ -413,6 +413,8 @@ pub struct HealResultWindow {
pub struct HealTask {
/// Task ID
pub id: String,
/// Process-local diagnostic identity; never persisted or part of a request.
pub(crate) observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
/// Heal type
pub heal_type: HealType,
pub bucket_incarnation_id: Option<Uuid>,
@@ -494,6 +496,7 @@ impl HealTask {
pub fn from_request(request: HealRequest, storage: Arc<dyn HealStorageAPI>) -> Self {
Self {
id: request.id,
observation_id: None,
heal_type: request.heal_type,
bucket_incarnation_id: request.bucket_incarnation_id,
expected_mrf_bucket_incarnation_id: request.expected_mrf_bucket_incarnation_id,
@@ -529,6 +532,14 @@ impl HealTask {
}
}
pub(crate) fn with_observation_id(
mut self,
observation_id: Option<rustfs_heal_contracts::completion::HealObservationId>,
) -> Self {
self.observation_id = observation_id;
self
}
pub fn retry_request(&self) -> HealRequest {
HealRequest {
id: self.id.clone(),
+10 -7
View File
@@ -280,12 +280,12 @@ pub async fn init_heal_manager_with_workload_provider(
.is_some_and(|hook| hook.fail_channel_init.load(Ordering::SeqCst));
#[cfg(not(test))]
let force_channel_failure = false;
let channel_receiver = if force_channel_failure {
Err("forced heal channel initialization failure")
} else {
rustfs_heal_contracts::heal_channel::init_heal_channels()
};
let (receiver, receipt_receiver) = match channel_receiver {
if force_channel_failure {
stop_initializing_manager(&heal_manager).await?;
return Err(Error::Config("forced heal channel initialization failure".to_string()));
}
let channel_receiver = rustfs_heal_contracts::heal_channel::init_heal_channels_with_observations();
let (receiver, receipt_receiver, observed_receiver) = match channel_receiver {
Ok(receivers) => receivers,
Err(err) => {
stop_initializing_manager(&heal_manager).await?;
@@ -302,7 +302,10 @@ pub async fn init_heal_manager_with_workload_provider(
tokio::spawn(async move {
let mut processor = channel_processor.lock().await;
if let Err(e) = processor.start_with_receipts(receiver, receipt_receiver).await {
if let Err(e) = processor
.start_with_observed_receipts(receiver, receipt_receiver, observed_receiver)
.await
{
error!(
target: "rustfs::heal",
event = EVENT_HEAL_RUNTIME_STATE,
+195 -55
View File
@@ -17,8 +17,8 @@
use metrics::{counter, gauge};
use rustfs_heal_contracts::{
completion::{HealCompletion, MAX_COMPLETION_IDENTITY_BYTES, subscribe_heal_completions},
heal_channel::{HealAdmissionReceipt, HealChannelRequest, HealScanMode},
completion::{MAX_COMPLETION_IDENTITY_BYTES, ObservedHealCompletion, subscribe_observed_heal_completions},
heal_channel::{HealChannelRequest, HealObservationChannelError, HealScanMode, ObservedHealAdmissionReceipt},
outcome::{HealAbortReason, HealExecutionOutcome, HealObjectDisposition, HealObjectIdentity, HealObjectKind},
};
use serde::Serialize;
@@ -30,6 +30,7 @@ use std::{
use tokio::sync::{broadcast, mpsc};
const MAX_PENDING: usize = 1024;
const MAX_PENDING_GROUPS: usize = MAX_PENDING;
const MAX_RECENT: usize = 1024;
const PENDING_TTL: Duration = Duration::from_secs(60 * 60);
const RECENT_TTL: Duration = Duration::from_secs(10 * 60);
@@ -122,18 +123,46 @@ impl Candidate {
count: 1,
})
}
fn has_same_semantics(&self, other: &Self) -> bool {
self.identity == other.identity && self.mode == other.mode && self.dry_run == other.dry_run
}
}
struct Pending {
// Keep the common single-group case inline.
first: Candidate,
others: Vec<Candidate>,
}
impl Pending {
fn iter(&self) -> impl Iterator<Item = &Candidate> {
std::iter::once(&self.first).chain(self.others.iter())
}
fn into_candidates(self) -> impl Iterator<Item = Candidate> {
std::iter::once(self.first).chain(self.others)
}
}
struct Recent {
completion: Arc<HealCompletion>,
completion: Arc<ObservedHealCompletion>,
counted: bool,
}
#[derive(Clone, Debug, Hash, PartialEq, Eq)]
struct ObservationKey {
task_id: String,
observation_id: rustfs_heal_contracts::completion::HealObservationId,
}
#[derive(Default)]
struct Observations {
pending: HashMap<String, Candidate>,
recent: HashMap<String, Recent>,
recent_order: VecDeque<(Instant, String)>,
pending: HashMap<ObservationKey, Pending>,
// The group bound applies across all observation keys.
pending_groups: usize,
recent: HashMap<ObservationKey, Recent>,
recent_order: VecDeque<(Instant, ObservationKey)>,
// All keys come from fixed execution/disposition/scope/mode labels, never task or object identities.
terminal_counts: BTreeMap<(&'static str, &'static str, &'static str, &'static str), u64>,
terminal_tasks: u64,
@@ -183,17 +212,7 @@ impl Settlement {
"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"
}
(Some(expected), Some(actual)) => Self::identity_scope(expected, &actual.identity),
_ => "identity_unknown",
}
};
@@ -209,6 +228,60 @@ impl Settlement {
}
}
fn identity_scope(expected: &HealObjectIdentity, actual: &HealObjectIdentity) -> &'static str {
fn known_version(version: Option<&str>) -> Option<uuid::Uuid> {
version
.and_then(|id| uuid::Uuid::parse_str(id).ok())
.filter(|id| !id.is_nil())
}
let expected_bucket = (!expected.bucket.is_empty()).then_some(expected.bucket.as_str());
let actual_bucket = (!actual.bucket.is_empty()).then_some(actual.bucket.as_str());
let expected_object = (!expected.object.is_empty()).then_some(expected.object.as_str());
let actual_object = (!actual.object.is_empty()).then_some(actual.object.as_str());
let expected_version = known_version(expected.version_id.as_deref());
let actual_version = known_version(actual.version_id.as_deref());
let expected_incarnation = expected.bucket_incarnation_id.filter(|id| !id.is_nil());
let actual_incarnation = actual.bucket_incarnation_id.filter(|id| !id.is_nil());
if expected.kind != actual.kind
|| expected_bucket.zip(actual_bucket).is_some_and(|(left, right)| left != right)
|| expected_object.zip(actual_object).is_some_and(|(left, right)| left != right)
|| expected_version
.zip(actual_version)
.is_some_and(|(left, right)| left != right)
|| expected_incarnation
.zip(actual_incarnation)
.is_some_and(|(left, right)| left != right)
|| expected
.pool_index
.zip(actual.pool_index)
.is_some_and(|(left, right)| left != right)
|| expected
.set_index
.zip(actual.set_index)
.is_some_and(|(left, right)| left != right)
{
"identity_mismatch"
} else if expected_bucket.is_none()
|| actual_bucket.is_none()
|| expected_object.is_none()
|| actual_object.is_none()
|| expected_version.is_none()
|| actual_version.is_none()
|| expected_incarnation.is_none()
|| actual_incarnation.is_none()
|| expected.pool_index.is_none()
|| actual.pool_index.is_none()
|| expected.set_index.is_none()
|| actual.set_index.is_none()
{
"identity_unknown"
} else {
// Matching identities still do not bind content or leader epoch.
"content_and_leader_unknown"
}
}
fn record(&self) {
counter!("rustfs_scanner_heal_candidate_terminal_total",
"disposition" => self.disposition, "scope" => self.scope, "scan_mode" => self.mode.as_str())
@@ -240,50 +313,78 @@ impl Observations {
self.record_unknown(settlement.scope, settlement.candidates);
}
fn admit(&mut self, task_id: String, candidate: Candidate) -> Option<Settlement> {
if let Some(recent) = self.recent.get_mut(&task_id) {
fn admit(&mut self, key: ObservationKey, candidate: Candidate) -> Option<Settlement> {
if Instant::now().saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
self.record_unknown("expired", candidate.count);
return None;
}
if let Some(recent) = self.recent.get_mut(&key) {
let settlement = Settlement::new(candidate, recent);
self.record_settlement(&settlement);
return Some(settlement);
}
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 let Some(pending) = self.pending.get_mut(&key) {
if pending.first.has_same_semantics(&candidate) {
pending.first.count = pending.first.count.saturating_add(candidate.count);
pending.first.first_seen = pending.first.first_seen.min(candidate.first_seen);
} else if let Some(group) = pending.others.iter_mut().find(|group| group.has_same_semantics(&candidate)) {
group.count = group.count.saturating_add(candidate.count);
group.first_seen = group.first_seen.min(candidate.first_seen);
} else if self.pending_groups < MAX_PENDING_GROUPS {
pending.others.push(candidate);
self.pending_groups += 1;
} else {
self.record_unknown("pending_groups_full", candidate.count);
}
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 {
} else if self.pending.len() >= MAX_PENDING {
self.record_unknown("pending_full", candidate.count);
} else if self.pending_groups >= MAX_PENDING_GROUPS {
self.record_unknown("pending_groups_full", candidate.count);
} else {
self.pending.insert(
key,
Pending {
first: candidate,
others: Vec::new(),
},
);
self.pending_groups += 1;
}
None
}
fn complete(&mut self, completion: Arc<HealCompletion>, now: Instant) -> Option<Settlement> {
if self.recent.contains_key(&completion.task_id) {
fn complete(&mut self, completion: Arc<ObservedHealCompletion>, now: Instant) -> Option<Settlement> {
let key = ObservationKey {
task_id: completion.task_id.clone(),
observation_id: completion.observation_id?,
};
if self.recent.contains_key(&key) {
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));
let mut settled = None;
if let Some(pending) = self.pending.remove(&key) {
self.pending_groups -= 1 + pending.others.len();
for candidate in pending.into_candidates() {
if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
self.record_unknown("expired", candidate.count);
continue;
}
let settlement = Settlement::new(candidate, &mut recent);
self.record_settlement(&settlement);
if settled.is_none() {
settled = Some(settlement);
}
}
}
if self.recent.len() == MAX_RECENT {
self.evict_recent();
}
self.recent_order.push_back((now, task_id.clone()));
self.recent.insert(task_id, recent);
if let Some(settlement) = &settled {
self.record_settlement(settlement);
}
self.recent_order.push_back((now, key.clone()));
self.recent.insert(key, recent);
settled
}
@@ -297,7 +398,9 @@ impl Observations {
let candidates = self
.pending
.drain()
.fold(0u64, |count, (_, candidate)| count.saturating_add(candidate.count));
.flat_map(|(_, pending)| pending.into_candidates())
.fold(0u64, |count, candidate| count.saturating_add(candidate.count));
self.pending_groups = 0;
if candidates > 0 {
self.record_unknown("completion_lost", candidates);
}
@@ -313,14 +416,32 @@ impl Observations {
fn maintain(&mut self, now: Instant) {
let mut expired = 0u64;
self.pending.retain(|_, candidate| {
if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
expired = expired.saturating_add(candidate.count);
false
} else {
true
let mut expired_groups = 0usize;
self.pending.retain(|_, pending| {
let previous_len = pending.others.len();
pending.others.retain(|candidate| {
if now.saturating_duration_since(candidate.first_seen) >= PENDING_TTL {
expired = expired.saturating_add(candidate.count);
expired_groups += 1;
false
} else {
true
}
});
if now.saturating_duration_since(pending.first.first_seen) >= PENDING_TTL {
expired = expired.saturating_add(pending.first.count);
expired_groups += 1;
if pending.others.is_empty() {
return false;
}
pending.first = pending.others.remove(0);
}
if pending.others.len() < previous_len {
pending.others.shrink_to_fit();
}
true
});
self.pending_groups -= expired_groups;
if expired > 0 {
self.record_unknown("expired", expired);
}
@@ -334,6 +455,7 @@ impl Observations {
let oldest = self
.pending
.values()
.flat_map(Pending::iter)
.map(|candidate| now.saturating_duration_since(candidate.first_seen))
.max();
gauge!("rustfs_scanner_heal_pending_tasks").set(self.pending.len() as f64);
@@ -341,7 +463,7 @@ impl Observations {
}
}
type AdmissionSender = mpsc::Sender<(String, Candidate)>;
type AdmissionSender = mpsc::Sender<(ObservationKey, Candidate)>;
#[derive(Clone)]
struct ObserverHandle {
@@ -371,7 +493,7 @@ fn observer_from(registry: &Mutex<Option<ObserverHandle>>) -> ObserverHandle {
let state = Arc::new(Mutex::new(Observations::default()));
let handle = {
let (sender, mut admissions) = mpsc::channel(MAX_PENDING);
let mut completions = subscribe_heal_completions();
let mut completions = subscribe_observed_heal_completions();
let observations = state.clone();
tokio::spawn(async move {
let mut tick = tokio::time::interval(Duration::from_secs(5));
@@ -445,12 +567,14 @@ fn observation_status_from(registry: &Mutex<Option<ObserverHandle>>) -> ScannerH
observations
.pending
.values()
.flat_map(Pending::iter)
.fold(0u64, |count, candidate| count.saturating_add(candidate.count)),
);
let now = Instant::now();
status.oldest_pending_seconds = observations
.pending
.values()
.flat_map(Pending::iter)
.map(|candidate| now.saturating_duration_since(candidate.first_seen))
.max()
.map(|age| age.as_secs_f64());
@@ -489,11 +613,18 @@ pub(crate) fn prepare(request: &HealChannelRequest) -> Option<Candidate> {
Candidate::from_request(request)
}
pub(crate) fn observe(candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
pub(crate) fn observe(
candidate: Option<Candidate>,
receipt: Result<&ObservedHealAdmissionReceipt, &HealObservationChannelError>,
) {
observe_on(&observer(), candidate, receipt);
}
fn observe_on(observer: &ObserverHandle, candidate: Option<Candidate>, receipt: Result<&HealAdmissionReceipt, ()>) {
fn observe_on(
observer: &ObserverHandle,
candidate: Option<Candidate>,
receipt: Result<&ObservedHealAdmissionReceipt, &HealObservationChannelError>,
) {
match receipt {
Ok(receipt) if receipt.result.is_admitted() => {
let Some(candidate) = candidate else {
@@ -505,11 +636,20 @@ fn observe_on(observer: &ObserverHandle, candidate: Option<Candidate>, receipt:
observer.record_unknown("missing_task_id", 1);
return;
}
if observer.sender.try_send((task_id.clone(), candidate)).is_err() {
let Some(observation_id) = receipt.observation_id else {
observer.record_unknown("owner_generation_unknown", candidate.count);
return;
};
let key = ObservationKey {
task_id: task_id.clone(),
observation_id,
};
if observer.sender.try_send((key, candidate)).is_err() {
observer.record_unknown("observer_full_or_closed", 1);
}
}
_ => observer.record_unknown("not_admitted", 1),
Ok(_) => observer.record_unknown("not_admitted", 1),
Err(_) => observer.record_unknown("admission_unknown", 1),
}
}
File diff suppressed because it is too large Load Diff
+1 -1
View File
@@ -51,7 +51,7 @@ use rustfs_filemeta::{
};
use rustfs_heal_contracts::heal_channel::{
HEAL_DELETE_DANGLING, HealAdmissionDropReason, HealAdmissionResult, HealChannelPriority, HealChannelRequest,
HealRequestSource, HealScanMode, send_heal_request_with_receipt,
HealRequestSource, HealScanMode, send_heal_request_with_observation,
};
use rustfs_scanner_metrics::metrics::{
CloseDiskGuard, IlmAction, Metric, Metrics, ScannerReplicationRepairKind, ScannerSourceWorkUpdate, ScannerWorkSource,
@@ -481,7 +481,7 @@ pub(super) async fn send_scanner_heal_request(
let priority = request.priority;
let trace_context = scanner_heal_candidate_trace_context(&request);
let candidate = crate::heal_completion::prepare(&request);
match send_heal_request_with_receipt(request).await {
match send_heal_request_with_observation(request).await {
Ok(receipt) => {
crate::heal_completion::observe(candidate, Ok(&receipt));
let result = receipt.result;
@@ -501,7 +501,8 @@ pub(super) async fn send_scanner_heal_request(
Ok(result)
}
Err(err) => {
crate::heal_completion::observe(candidate, Err(()));
crate::heal_completion::observe(candidate, Err(&err));
let err = err.to_string();
counter!(
"rustfs_heal_candidate_enqueue_total",
"type" => candidate_type.to_string(),
+14 -5
View File
@@ -623,7 +623,7 @@ typed terminal result:
| Metric | Meaning |
|---|---|
| `rustfs_scanner_heal_pending_tasks` | Canonical tasks awaiting a terminal observation. |
| `rustfs_scanner_heal_pending_tasks` | Pending task ID and owner-generation pairs 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. |
@@ -640,16 +640,25 @@ verified-byte or full-bucket coverage counter is advanced here.
The scanner status response also exposes `heal_observation` without requiring
a metrics exporter. Reading status does not start an observer. `state` is
`not_started`, `active`, or `stopped`; unavailable pending counts are null.
`retained_pending_tasks` counts canonical IDs in the existing bounded map,
while `retained_pending_candidates` counts their associated discoveries.
`retained_pending_tasks` counts distinct canonical task ID and process-local
owner-generation pairs in the bounded map. Different generations of the same
task ID count separately. `retained_pending_candidates` counts their associated
discoveries. The JSON field names are unchanged. The pending-tasks metric uses
the same pair count when periodically refreshed, so it can lag the status.
`queued_admissions` samples the local admission channel and is not an atomic
cluster backlog. `terminal_task_observations_total` keeps the same bounded
deduplication semantics as the exporter counter.
`terminal_candidates` preserves execution, object disposition, scan mode, and
unresolved scope. `unknown_candidates` includes admission rejection, overflow,
loss, expiry, and missing terminal proof. These categories overlap: a completed
task with a repaired or verified-healthy object can still have unknown scanner
loss, expiry, and missing terminal proof. The `not_admitted` reason covers an
explicit `Full` or `Dropped` decision. The `admission_unknown` reason means no
decision was received because of a channel or processor error. A lost reply
does not establish whether work was admitted; a failure known to occur before
enqueue separately rules out submission. Both reasons remain part of the
unknown total. Monitoring that filters by reason must include the new
`admission_unknown` label to retain those errors. These categories overlap: a
completed task with a repaired or verified-healthy object can still have unknown scanner
integrity. `integrity_state` therefore remains `unknown`, including when retained
pending is zero. `lost_completion_notifications` counts skipped messages, not
objects. Counts belong to the named local observer `generation`; runtime