mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-27 14:27:10 +00:00
fix(heal): decide object absence from storage evidence, not source (#8697)
* test(heal): add failing regressions for a source-independent absent verdict Add the regression tests for backlog#2900 that fail on the current code, where the same object-level not-found is treated as deleted on the plain heal path but as recreate_failed on the recreate path, and where the erasure-set healer, the task predicates and the error helpers disagree on wrapped not-found and transient inputs. Also add the per-object disposition cache module with its unit tests, the manager admission tests for it, the compat register entry for the display boundary, and the admission-invariant notes. The module and the manager tests are not declared yet, so nothing is wired into the build. * fix(heal): decide object absence from storage evidence, not request source Replace the five heal error classifiers (erasure-set healer outcome mapping, task-level not-found and lock predicates, error.rs recovery helpers and the storage existence-check matcher) with one typed classify_heal_error. An object-level not-found from the locked storage read is Absent for every source and every recreate_missing value; the recreate path, the usage-observation and scanner object-dir whitelists are gone, and display-text matching survives only in the marked classify_error_display boundary. Add a bounded, expiring per-object disposition memory consulted before queue admission so a best-effort producer cannot requeue an absent or terminally failed object on every tick; durable MRF, admin, forced, high-priority and dry-run requests bypass it. The read-repair producer no longer forces recreate_missing. * fix(heal): keep rendered grace windows out of task retries A grace window rendered into a task message belongs to an attempt that already finished; its owner schedules the next one. Only typed grace errors replay the task, as before the classifier merge. Update the ecstore read-repair black-box test: the producer no longer forces recreate_missing. * fix(heal): classify ambiguous legacy delete residue as permanent explicitly * fix(heal): scope disposition memory and bound absence parsing
This commit is contained in:
Chris
Generated
-1
@@ -10001,7 +10001,6 @@ dependencies = [
|
||||
"rustfs-common",
|
||||
"rustfs-concurrency",
|
||||
"rustfs-config",
|
||||
"rustfs-data-usage",
|
||||
"rustfs-ecstore",
|
||||
"rustfs-filemeta",
|
||||
"rustfs-heal-contracts",
|
||||
|
||||
@@ -560,7 +560,10 @@ async fn blackbox_heal_requests_preserve_repair_scope() {
|
||||
assert_eq!(request.pool_index, Some(0));
|
||||
assert_eq!(request.set_index, Some(0));
|
||||
assert_eq!(request.priority, HealChannelPriority::Low);
|
||||
assert_eq!(request.recreate_missing, Some(true));
|
||||
assert_eq!(
|
||||
request.recreate_missing, None,
|
||||
"a degraded read does not instruct recreation; storage decides absence"
|
||||
);
|
||||
|
||||
let mpu_bucket = "bb-mpu-convergence-heal";
|
||||
let partial_object = "partial.bin";
|
||||
|
||||
@@ -601,7 +601,8 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
|
||||
);
|
||||
request.source = HealRequestSource::ReadRepair;
|
||||
request.object_version_id = version_id.filter(|value| !value.is_empty()).map(str::to_string);
|
||||
request.recreate_missing = Some(true);
|
||||
// A degraded read's surviving minority is not a recreate instruction; the
|
||||
// locked storage read decides whether the target is absent (backlog#2900).
|
||||
|
||||
let request_id = request.id.clone();
|
||||
let bucket = bucket.to_string();
|
||||
|
||||
@@ -73,7 +73,6 @@ hotpath-cpu = [
|
||||
[dependencies]
|
||||
hotpath.workspace = true
|
||||
rustfs-config = { workspace = true }
|
||||
rustfs-data-usage = { workspace = true }
|
||||
rustfs-concurrency = { workspace = true }
|
||||
rustfs-ecstore = { workspace = true }
|
||||
rustfs-lock = { workspace = true }
|
||||
|
||||
+390
-107
@@ -121,81 +121,28 @@ impl Error {
|
||||
}
|
||||
|
||||
/// Whether a heal operation can be retried without changing its inputs.
|
||||
///
|
||||
/// Typed-first (backlog#1845): the primary classification reads error
|
||||
/// variants and typed helpers (`is_quorum_error`, `LockError::is_fatal`);
|
||||
/// the substring fallback in [`is_recoverable_heal_error_message`] only
|
||||
/// catches errors whose typed identity was destroyed upstream.
|
||||
pub(crate) fn is_recoverable_heal(&self) -> bool {
|
||||
match self {
|
||||
Error::TaskCancelled | Error::TaskTimeout | Error::StaleBucketIncarnation { .. } => false,
|
||||
Error::ReplacementTargetNotReady(_) | Error::DanglingDeleteDeferred { .. } => true,
|
||||
Error::TransientSkip { .. } | Error::BucketDiscoveryIncomplete => true,
|
||||
// Lock failures classify by LockError's own taxonomy: only the
|
||||
// fatal variants (ResourceNotFound / PermissionDenied /
|
||||
// Configuration) are terminal - retrying cannot fix them - while
|
||||
// contention and transport variants (Timeout, Network, Internal,
|
||||
// AlreadyLocked, QuorumNotReached, InsufficientNodes, ...) stay
|
||||
// recoverable, as the previous blanket `Lock(_) => true` treated
|
||||
// them.
|
||||
Error::Storage(EcstoreError::Lock(lock_err)) => !lock_err.is_fatal(),
|
||||
Error::Storage(err) => {
|
||||
if err.is_dangling_delete_grace() {
|
||||
return true;
|
||||
}
|
||||
err.is_quorum_error()
|
||||
|| matches!(err, EcstoreError::Io(error) if is_recoverable_io_error(error))
|
||||
|| matches!(
|
||||
err,
|
||||
EcstoreError::DiskNotFound
|
||||
| EcstoreError::VolumeNotFound
|
||||
| EcstoreError::FaultyDisk
|
||||
| EcstoreError::FaultyRemoteDisk
|
||||
| EcstoreError::SlowDown
|
||||
| EcstoreError::OperationCanceled
|
||||
| EcstoreError::RemoteClientUnavailable(_)
|
||||
)
|
||||
|| is_recoverable_heal_error_message(&err.to_string())
|
||||
match classify_heal_error(self) {
|
||||
// Only a typed grace window replays the task. A grace window
|
||||
// rendered into a task message or an io::Error comes from an
|
||||
// attempt that already finished; its owner (for example a durable
|
||||
// MRF responsibility) schedules the next attempt.
|
||||
HealErrorClass::DanglingGrace { .. } => {
|
||||
matches!(self, Error::DanglingDeleteDeferred { .. } | Error::Storage(_) | Error::Disk(_))
|
||||
}
|
||||
// Nonblocking local CAS and replacement leases report lock
|
||||
// contention as WouldBlock; retain the existing task retry budget.
|
||||
Error::Disk(DiskError::Io(error)) if error.kind() == std::io::ErrorKind::WouldBlock => true,
|
||||
Error::Disk(err) => {
|
||||
if err.is_dangling_delete_grace() {
|
||||
return true;
|
||||
}
|
||||
matches!(
|
||||
err,
|
||||
DiskError::DiskNotFound
|
||||
| DiskError::ErasureReadQuorum
|
||||
| DiskError::ErasureWriteQuorum
|
||||
| DiskError::Timeout
|
||||
| DiskError::SourceStalled
|
||||
| DiskError::FaultyRemoteDisk
|
||||
| DiskError::FaultyDisk
|
||||
| DiskError::RemoteClientUnavailable(_)
|
||||
) || matches!(err, DiskError::Io(error) if is_recoverable_io_error(error))
|
||||
|| is_recoverable_heal_error_message(&err.to_string())
|
||||
}
|
||||
Error::TaskExecutionFailed { message } | Error::Other(message) => is_recoverable_heal_error_message(message),
|
||||
Error::Io(err) => is_recoverable_io_error(err) || is_recoverable_heal_error_message(&err.to_string()),
|
||||
_ => false,
|
||||
class => class.is_retryable(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Retry an unproven missing object only within the per-object budget.
|
||||
/// Generic task retries retain the narrower classification above.
|
||||
pub(crate) fn is_recoverable_object_heal(&self) -> bool {
|
||||
self.is_recoverable_heal()
|
||||
|| matches!(
|
||||
self,
|
||||
Error::Storage(
|
||||
EcstoreError::FileNotFound
|
||||
| EcstoreError::FileVersionNotFound
|
||||
| EcstoreError::ObjectNotFound(_, _)
|
||||
| EcstoreError::VersionNotFound(_, _, _)
|
||||
) | Error::Disk(DiskError::FileNotFound | DiskError::FileVersionNotFound)
|
||||
)
|
||||
self.is_recoverable_heal() || self.is_object_absent()
|
||||
}
|
||||
|
||||
/// The locked storage read found no replica of the object or version.
|
||||
pub(crate) fn is_object_absent(&self) -> bool {
|
||||
classify_heal_error(self) == HealErrorClass::Absent
|
||||
}
|
||||
|
||||
pub(crate) fn dangling_delete_retry_not_before(&self) -> Option<std::time::SystemTime> {
|
||||
@@ -212,19 +159,157 @@ impl Error {
|
||||
}
|
||||
|
||||
pub(crate) fn is_dangling_delete_grace(&self) -> bool {
|
||||
match self {
|
||||
Error::DanglingDeleteDeferred { .. } => true,
|
||||
Error::Storage(err) => err.is_dangling_delete_grace(),
|
||||
Error::Disk(err) => err.is_dangling_delete_grace(),
|
||||
Error::Io(err) => DiskError::io_error_is_dangling_delete_grace(err),
|
||||
Error::TaskExecutionFailed { message } | Error::Other(message) => {
|
||||
message.contains(HEAL_DANGLING_DELETE_GRACE_MESSAGE)
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
matches!(classify_heal_error(self), HealErrorClass::DanglingGrace { .. })
|
||||
}
|
||||
}
|
||||
|
||||
/// Verdict classes shared by every heal error consumer (backlog#2900).
|
||||
///
|
||||
/// One classifier replaces the per-call-site predicates that disagreed on the
|
||||
/// same input. The class is decided from typed variants; display text is
|
||||
/// consulted only by [`classify_error_display`] for errors whose type was
|
||||
/// destroyed upstream.
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub(crate) enum HealErrorClass {
|
||||
/// The locked quorum read found no replica of the object or version:
|
||||
/// nothing to repair, independent of request source or options. This is
|
||||
/// a verdict, not a durable absence proof.
|
||||
Absent,
|
||||
/// Offline disk, missing volume, unmet quorum, transport failure,
|
||||
/// slow-down or cancelled RPC: the same request can succeed later.
|
||||
Transient,
|
||||
/// Lock contention or a lock/RPC timeout (non-fatal `LockError`).
|
||||
LockContention,
|
||||
/// A storage grace window; `retry_not_before` is the absolute UNIX
|
||||
/// deadline when storage reported one.
|
||||
DanglingGrace { retry_not_before: Option<u64> },
|
||||
/// Storage reports nothing to heal.
|
||||
NoHealRequired,
|
||||
/// The task was cancelled, timed out or its admission became stale: the
|
||||
/// object's state is unknown and no retry budget applies.
|
||||
Aborted,
|
||||
/// Retrying cannot help: corruption beyond parity, permission, fatal
|
||||
/// lock state or invalid input.
|
||||
Permanent,
|
||||
}
|
||||
|
||||
impl HealErrorClass {
|
||||
/// Whether the same request can be retried unchanged.
|
||||
pub(crate) const fn is_retryable(self) -> bool {
|
||||
matches!(self, Self::Transient | Self::LockContention | Self::DanglingGrace { .. })
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify a heal error from its typed identity.
|
||||
pub(crate) fn classify_heal_error(error: &Error) -> HealErrorClass {
|
||||
match error {
|
||||
Error::TaskCancelled | Error::TaskTimeout | Error::StaleBucketIncarnation { .. } => HealErrorClass::Aborted,
|
||||
Error::DanglingDeleteDeferred { retry_not_before } => HealErrorClass::DanglingGrace {
|
||||
retry_not_before: Some(*retry_not_before),
|
||||
},
|
||||
Error::ReplacementTargetNotReady(_) | Error::TransientSkip { .. } | Error::BucketDiscoveryIncomplete => {
|
||||
HealErrorClass::Transient
|
||||
}
|
||||
Error::Storage(error) => classify_storage_error(error),
|
||||
Error::Disk(error) => classify_disk_error(error),
|
||||
Error::Io(error) => classify_io_error(error),
|
||||
Error::TaskExecutionFailed { message } | Error::Other(message) => {
|
||||
classify_error_display(message).unwrap_or(HealErrorClass::Permanent)
|
||||
}
|
||||
_ => HealErrorClass::Permanent,
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify a storage error. Lock failures follow `LockError`'s own taxonomy:
|
||||
/// only its fatal variants are terminal. `StorageError::is_not_found()` is
|
||||
/// deliberately not used: it lumps `DiskNotFound`/`VolumeNotFound` together
|
||||
/// with object absence, the conflation that once let an offline drive be
|
||||
/// recorded as absent and skipped (backlog#856 / #799 B7).
|
||||
pub(crate) fn classify_storage_error(error: &EcstoreError) -> HealErrorClass {
|
||||
if let EcstoreError::Lock(lock_error) = error {
|
||||
return if lock_error.is_fatal() {
|
||||
HealErrorClass::Permanent
|
||||
} else {
|
||||
HealErrorClass::LockContention
|
||||
};
|
||||
}
|
||||
if error.is_dangling_delete_grace() {
|
||||
return HealErrorClass::DanglingGrace {
|
||||
retry_not_before: dangling_deadline(error.dangling_delete_retry_after()),
|
||||
};
|
||||
}
|
||||
match error {
|
||||
EcstoreError::FileNotFound
|
||||
| EcstoreError::FileVersionNotFound
|
||||
| EcstoreError::ObjectNotFound(_, _)
|
||||
| EcstoreError::VersionNotFound(_, _, _) => HealErrorClass::Absent,
|
||||
EcstoreError::NoHealRequired => HealErrorClass::NoHealRequired,
|
||||
// An ambiguous legacy delete residue needs operator inspection;
|
||||
// retrying the same heal cannot resolve it (backlog#2902).
|
||||
EcstoreError::DeleteRecoveryLegacyAmbiguous => HealErrorClass::Permanent,
|
||||
EcstoreError::Timeout => HealErrorClass::LockContention,
|
||||
EcstoreError::DiskNotFound
|
||||
| EcstoreError::VolumeNotFound
|
||||
| EcstoreError::FaultyDisk
|
||||
| EcstoreError::FaultyRemoteDisk
|
||||
| EcstoreError::SlowDown
|
||||
| EcstoreError::OperationCanceled
|
||||
| EcstoreError::RemoteClientUnavailable(_) => HealErrorClass::Transient,
|
||||
_ if error.is_quorum_error() => HealErrorClass::Transient,
|
||||
EcstoreError::Io(error) => classify_io_error(error),
|
||||
_ => classify_error_display(&error.to_string()).unwrap_or(HealErrorClass::Permanent),
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify a disk error. Nonblocking local CAS and replacement leases report
|
||||
/// lock contention as `WouldBlock`.
|
||||
pub(crate) fn classify_disk_error(error: &DiskError) -> HealErrorClass {
|
||||
if error.is_dangling_delete_grace() {
|
||||
return HealErrorClass::DanglingGrace {
|
||||
retry_not_before: dangling_deadline(error.dangling_delete_retry_after()),
|
||||
};
|
||||
}
|
||||
match error {
|
||||
DiskError::FileNotFound | DiskError::FileVersionNotFound => HealErrorClass::Absent,
|
||||
DiskError::NoHealRequired => HealErrorClass::NoHealRequired,
|
||||
DiskError::Timeout => HealErrorClass::LockContention,
|
||||
DiskError::Io(error) if error.kind() == std::io::ErrorKind::WouldBlock => HealErrorClass::LockContention,
|
||||
DiskError::DiskNotFound
|
||||
| DiskError::VolumeNotFound
|
||||
| DiskError::ErasureReadQuorum
|
||||
| DiskError::ErasureWriteQuorum
|
||||
| DiskError::SourceStalled
|
||||
| DiskError::FaultyRemoteDisk
|
||||
| DiskError::FaultyDisk
|
||||
| DiskError::RemoteClientUnavailable(_) => HealErrorClass::Transient,
|
||||
DiskError::Io(error) => classify_io_error(error),
|
||||
_ => classify_error_display(&error.to_string()).unwrap_or(HealErrorClass::Permanent),
|
||||
}
|
||||
}
|
||||
|
||||
fn classify_io_error(error: &std::io::Error) -> HealErrorClass {
|
||||
if DiskError::io_error_is_dangling_delete_grace(error) {
|
||||
return HealErrorClass::DanglingGrace {
|
||||
retry_not_before: dangling_deadline(DiskError::io_error_dangling_delete_retry_after(error)),
|
||||
};
|
||||
}
|
||||
if is_recoverable_io_error(error) {
|
||||
return HealErrorClass::Transient;
|
||||
}
|
||||
classify_error_display(&error.to_string()).unwrap_or(HealErrorClass::Permanent)
|
||||
}
|
||||
|
||||
/// Storage reports whole remaining seconds. Round the absolute deadline up so
|
||||
/// subsecond grace boundaries cannot hot-loop.
|
||||
fn dangling_deadline(retry_after: Option<std::time::Duration>) -> Option<u64> {
|
||||
std::time::SystemTime::now()
|
||||
.checked_add(retry_after?)?
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.ok()?
|
||||
.as_secs()
|
||||
.checked_add(1)
|
||||
}
|
||||
|
||||
fn is_recoverable_io_error(error: &std::io::Error) -> bool {
|
||||
// ObjectIO preserves store errors as IO sources; their quorum identity must survive this boundary.
|
||||
if error
|
||||
@@ -262,43 +347,71 @@ fn is_recoverable_io_error(error: &std::io::Error) -> bool {
|
||||
&& matches!(error.kind(), InternodeHttpErrorKind::HttpStatus(status) if matches!(status.as_u16(), 409 | 500)))
|
||||
}
|
||||
|
||||
/// Documented substring fallback for errors that reach heal with their typed
|
||||
/// identity destroyed (stringified through `TaskExecutionFailed`/`Other`, or
|
||||
/// boxed into `Io`). Every needle is annotated with the producer that emits
|
||||
/// it; when a producer becomes typed end-to-end, delete its needle here
|
||||
/// (backlog#1845 - this list only shrinks).
|
||||
fn is_recoverable_heal_error_message(error: &str) -> bool {
|
||||
let error = error.to_ascii_lowercase();
|
||||
[
|
||||
// set_disk/ops/locking.rs ns_loc lock failures rendered into messages.
|
||||
/// Messages that reach heal with their typed identity destroyed: an older
|
||||
/// peer's display text, a local `format!` into `TaskExecutionFailed`/`Other`,
|
||||
/// or a boxed `io::Error`. Every needle names the producer that emits it; when
|
||||
/// a producer becomes typed end-to-end, delete its needle here. This is the
|
||||
/// only place in the heal crate that classifies error text.
|
||||
// RUSTFS_COMPAT_TODO(backlog-2900-heal-error-display): map typed displays back to a class once, here. Remove after every heal producer and the minimum supported peer version return typed storage errors end-to-end.
|
||||
pub(crate) fn classify_error_display(message: &str) -> Option<HealErrorClass> {
|
||||
// set_disk/ops/locking.rs ns_loc lock failures, ecstore remote_locker.rs
|
||||
// and lock/src lock responses, tokio/tonic deadline and timeout rendering,
|
||||
// and LockError::QuorumNotReached, when stringified before typing.
|
||||
const LOCK_CONTENTION: [&str; 7] = [
|
||||
"failed to acquire read lock",
|
||||
// ecstore cluster/rpc/remote_locker.rs "Lock acquisition failed on
|
||||
// remote server" and lock/src local lock responses.
|
||||
"lock acquisition failed",
|
||||
// lock/src/distributed_lock.rs + client/local.rs LockResponse::failure.
|
||||
"lock acquisition timeout",
|
||||
// remote_locker.rs RPC deadline wrapper.
|
||||
"remote lock rpc timed out",
|
||||
// tokio/tonic deadline rendering, reaches heal via stringified RPC errors.
|
||||
"deadline has elapsed",
|
||||
// generic io/tonic timeout rendering.
|
||||
"timed out",
|
||||
// tonic transport failure rendering.
|
||||
"transport error",
|
||||
// LockError::Network display prefix.
|
||||
"network error",
|
||||
// io::Error ConnectionRefused rendering.
|
||||
"connection refused",
|
||||
// StorageError::OperationCanceled rendered through task messages.
|
||||
"operation canceled",
|
||||
// LockError::QuorumNotReached display, when stringified before typing.
|
||||
"quorum not reached",
|
||||
// set_disk/ops/heal.rs HEAL_RENAME_INCOMPLETE - the one needle with no
|
||||
// typed variant yet; the producer formats it into a plain message.
|
||||
];
|
||||
// tonic transport failure, LockError::Network and io ConnectionRefused
|
||||
// rendering, StorageError::OperationCanceled rendered through task
|
||||
// messages, and set_disk/ops/heal.rs HEAL_RENAME_INCOMPLETE, the one
|
||||
// needle with no typed variant yet.
|
||||
const TRANSIENT: [&str; 5] = [
|
||||
"transport error",
|
||||
"network error",
|
||||
"connection refused",
|
||||
"operation canceled",
|
||||
"heal rename incomplete",
|
||||
]
|
||||
.iter()
|
||||
.any(|pattern| error.contains(pattern))
|
||||
];
|
||||
// Canonical `Display` of DiskError/StorageError::NoHealRequired and the
|
||||
// legacy heal-format wording.
|
||||
const NO_HEAL_REQUIRED: [&str; 2] = ["No heal required", "No healing is required"];
|
||||
|
||||
if message.contains(HEAL_DANGLING_DELETE_GRACE_MESSAGE) {
|
||||
return Some(HealErrorClass::DanglingGrace { retry_not_before: None });
|
||||
}
|
||||
// A contention or transient needle wins over an absence phrase: misreading
|
||||
// a transient failure as absence would drop the object from repair.
|
||||
let lower = message.to_ascii_lowercase();
|
||||
if LOCK_CONTENTION.iter().any(|needle| lower.contains(needle)) {
|
||||
return Some(HealErrorClass::LockContention);
|
||||
}
|
||||
if TRANSIENT.iter().any(|needle| lower.contains(needle)) {
|
||||
return Some(HealErrorClass::Transient);
|
||||
}
|
||||
// Only canonical errors and their type wrappers are absence evidence.
|
||||
// Object names in contextual task messages can contain these same words.
|
||||
let mut display = message;
|
||||
while let Some(inner) = ["Storage error: ", "Disk error: ", "I/O error: ", "Other error: "]
|
||||
.iter()
|
||||
.find_map(|prefix| display.strip_prefix(*prefix))
|
||||
{
|
||||
display = inner;
|
||||
}
|
||||
if matches!(display, "File not found" | "File version not found")
|
||||
|| display.starts_with("Object not found: ")
|
||||
|| display.starts_with("Version not found: ")
|
||||
{
|
||||
return Some(HealErrorClass::Absent);
|
||||
}
|
||||
if NO_HEAL_REQUIRED.iter().any(|display| message.contains(display)) {
|
||||
return Some(HealErrorClass::NoHealRequired);
|
||||
}
|
||||
None
|
||||
}
|
||||
|
||||
impl From<Error> for std::io::Error {
|
||||
@@ -492,7 +605,7 @@ mod tests {
|
||||
"RPC conversion and cloning must retain the typed transport source"
|
||||
);
|
||||
assert!(
|
||||
!super::is_recoverable_heal_error_message(&error.to_string()),
|
||||
super::classify_error_display(&error.to_string()).is_none(),
|
||||
"GOAWAY regression must not be covered by the legacy message fallback: {error}"
|
||||
);
|
||||
assert!(error.is_recoverable_heal(), "a peer GOAWAY must remain recoverable: {error}");
|
||||
@@ -673,6 +786,176 @@ mod tests {
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_legacy_peer_display_not_found_is_object_level_absent() {
|
||||
// backlog#2900: a typed display that lost its type at an old peer or a
|
||||
// type wrapper converts back once, at the boundary.
|
||||
for error in [
|
||||
Error::Other("File not found".to_string()),
|
||||
Error::Other("File version not found".to_string()),
|
||||
Error::Other("Object not found: bucket/object".to_string()),
|
||||
Error::Other("Version not found: bucket/object/version".to_string()),
|
||||
Error::Other("I/O error: Other error: Disk error: File version not found".to_string()),
|
||||
Error::TaskExecutionFailed {
|
||||
message: "Storage error: File not found".to_string(),
|
||||
},
|
||||
] {
|
||||
assert!(error.is_recoverable_object_heal(), "{error:?}");
|
||||
assert!(!error.is_recoverable_heal(), "absence must not requeue the whole task: {error:?}");
|
||||
}
|
||||
for error in [
|
||||
Error::Other("file not found".to_string()),
|
||||
Error::Other("Disk not found".to_string()),
|
||||
Error::Other("Volume not found".to_string()),
|
||||
] {
|
||||
assert!(!error.is_recoverable_object_heal(), "{error:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_disk_volume_not_found_is_recoverable_like_storage() {
|
||||
assert!(Error::Disk(DiskError::VolumeNotFound).is_recoverable_heal());
|
||||
assert!(Error::Disk(DiskError::VolumeNotFound).is_recoverable_object_heal());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_heal_error_table() {
|
||||
use super::{HealErrorClass as C, classify_heal_error};
|
||||
use rustfs_lock::LockError;
|
||||
let grace = "dangling object deletion deferred by heal grace window; retry_after_secs=5; grace_secs=10";
|
||||
let cases = [
|
||||
(Error::TaskCancelled, C::Aborted),
|
||||
(Error::TaskTimeout, C::Aborted),
|
||||
(
|
||||
Error::StaleBucketIncarnation {
|
||||
bucket: "b".into(),
|
||||
expected: None,
|
||||
},
|
||||
C::Aborted,
|
||||
),
|
||||
(Error::Storage(EcstoreError::FileNotFound), C::Absent),
|
||||
(Error::Storage(EcstoreError::FileVersionNotFound), C::Absent),
|
||||
(Error::Storage(EcstoreError::ObjectNotFound("b".into(), "o".into())), C::Absent),
|
||||
(
|
||||
Error::Storage(EcstoreError::VersionNotFound("b".into(), "o".into(), "v".into())),
|
||||
C::Absent,
|
||||
),
|
||||
(Error::Disk(DiskError::FileNotFound), C::Absent),
|
||||
(Error::Disk(DiskError::FileVersionNotFound), C::Absent),
|
||||
(Error::Storage(EcstoreError::DiskNotFound), C::Transient),
|
||||
(Error::Storage(EcstoreError::VolumeNotFound), C::Transient),
|
||||
(Error::Storage(EcstoreError::FaultyDisk), C::Transient),
|
||||
(Error::Storage(EcstoreError::FaultyRemoteDisk), C::Transient),
|
||||
(Error::Storage(EcstoreError::SlowDown), C::Transient),
|
||||
(Error::Storage(EcstoreError::OperationCanceled), C::Transient),
|
||||
(Error::Storage(EcstoreError::RemoteClientUnavailable("rebuild".into())), C::Transient),
|
||||
(Error::Storage(EcstoreError::ErasureReadQuorum), C::Transient),
|
||||
(Error::Storage(EcstoreError::InsufficientReadQuorum("b".into(), "o".into())), C::Transient),
|
||||
(Error::Disk(DiskError::DiskNotFound), C::Transient),
|
||||
(Error::Disk(DiskError::VolumeNotFound), C::Transient),
|
||||
(Error::Disk(DiskError::ErasureWriteQuorum), C::Transient),
|
||||
(Error::Disk(DiskError::SourceStalled), C::Transient),
|
||||
(Error::TransientSkip { message: "skip".into() }, C::Transient),
|
||||
(Error::ReplacementTargetNotReady("restarting".into()), C::Transient),
|
||||
(Error::BucketDiscoveryIncomplete, C::Transient),
|
||||
(
|
||||
Error::Storage(EcstoreError::Lock(LockError::timeout("b/o", std::time::Duration::from_secs(1)))),
|
||||
C::LockContention,
|
||||
),
|
||||
(Error::Storage(EcstoreError::Timeout), C::LockContention),
|
||||
(Error::Disk(DiskError::Timeout), C::LockContention),
|
||||
(
|
||||
Error::Disk(DiskError::Io(std::io::Error::from(std::io::ErrorKind::WouldBlock))),
|
||||
C::LockContention,
|
||||
),
|
||||
(Error::Other("remote lock rpc timed out".into()), C::LockContention),
|
||||
(Error::Storage(EcstoreError::Lock(LockError::permission_denied("policy"))), C::Permanent),
|
||||
(Error::Storage(EcstoreError::DeleteRecoveryLegacyAmbiguous), C::Permanent),
|
||||
(Error::Disk(DiskError::FileCorrupt), C::Permanent),
|
||||
(Error::Disk(DiskError::FileAccessDenied), C::Permanent),
|
||||
(Error::Disk(DiskError::DiskFull), C::Permanent),
|
||||
(Error::Other("boom".into()), C::Permanent),
|
||||
(Error::Config("bad".into()), C::Permanent),
|
||||
(Error::Storage(EcstoreError::NoHealRequired), C::NoHealRequired),
|
||||
(Error::Disk(DiskError::NoHealRequired), C::NoHealRequired),
|
||||
(Error::Other("No heal required".into()), C::NoHealRequired),
|
||||
(
|
||||
Error::DanglingDeleteDeferred { retry_not_before: 42 },
|
||||
C::DanglingGrace {
|
||||
retry_not_before: Some(42),
|
||||
},
|
||||
),
|
||||
(
|
||||
Error::TaskExecutionFailed { message: grace.into() },
|
||||
C::DanglingGrace { retry_not_before: None },
|
||||
),
|
||||
];
|
||||
for (error, expected) in cases {
|
||||
assert_eq!(classify_heal_error(&error), expected, "{error:?}");
|
||||
}
|
||||
assert!(matches!(
|
||||
classify_heal_error(&Error::Disk(DiskError::other(grace))),
|
||||
C::DanglingGrace { .. }
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn rendered_grace_window_is_grace_but_does_not_replay_the_task() {
|
||||
let grace = "dangling object deletion deferred by heal grace window; retry_after_secs=5; grace_secs=10";
|
||||
let rendered = Error::TaskExecutionFailed {
|
||||
message: format!("Failed to recreate missing object b/o: Disk error: {grace}"),
|
||||
};
|
||||
assert!(rendered.is_dangling_delete_grace());
|
||||
assert!(!rendered.is_recoverable_heal(), "the owner of a finished attempt schedules the next one");
|
||||
assert!(!rendered.is_recoverable_object_heal());
|
||||
let typed = Error::Disk(DiskError::other(grace));
|
||||
assert!(typed.is_dangling_delete_grace());
|
||||
assert!(typed.is_recoverable_heal());
|
||||
assert!(Error::DanglingDeleteDeferred { retry_not_before: 1 }.is_recoverable_heal());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_error_display_prefers_contention_over_absence_text() {
|
||||
use super::{HealErrorClass as C, classify_error_display};
|
||||
assert_eq!(
|
||||
classify_error_display("File not found while waiting: lock acquisition timeout"),
|
||||
Some(C::LockContention)
|
||||
);
|
||||
assert_eq!(classify_error_display("Storage error: File not found"), Some(C::Absent));
|
||||
assert_eq!(classify_error_display("Object not found: b/o"), Some(C::Absent));
|
||||
assert_eq!(classify_error_display("file not found"), None);
|
||||
assert_eq!(classify_error_display("Disk not found"), None);
|
||||
assert_eq!(classify_error_display("operation was canceled"), None);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_error_display_does_not_take_absence_from_object_context() {
|
||||
use super::{HealErrorClass, classify_error_display, classify_heal_error};
|
||||
for object in [
|
||||
"File not found",
|
||||
"File version not found",
|
||||
"Object not found: b/o",
|
||||
"Version not found: b/o/v",
|
||||
] {
|
||||
let message = format!(
|
||||
"Heal left one or more drives unhealthy for bucket/{object}; retry after the missing drives are restored"
|
||||
);
|
||||
assert_eq!(
|
||||
classify_heal_error(&Error::TaskExecutionFailed { message }),
|
||||
HealErrorClass::Permanent,
|
||||
"{object}"
|
||||
);
|
||||
}
|
||||
for message in [
|
||||
"File not found.txt",
|
||||
"File version not found.txt",
|
||||
"Failed to heal object bucket/object: Storage error: File not found",
|
||||
"Failed to heal object bucket/Object not found: b/o: access denied",
|
||||
] {
|
||||
assert_eq!(classify_error_display(message), None, "{message}");
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn remote_client_unavailable_is_recoverable_via_typed_variant() {
|
||||
// The detail deliberately avoids every substring needle: the typed
|
||||
|
||||
@@ -735,6 +735,9 @@ impl HealChannelProcessor {
|
||||
HealChannelPriority::Critical => HealPriority::Urgent,
|
||||
};
|
||||
|
||||
// `recreate_missing` only reaches storage as the bucket-volume
|
||||
// recreation flag. Whether a missing object is absent is decided by the
|
||||
// locked storage read, never by this per-source default (backlog#2900).
|
||||
let recreate_missing = request.recreate_missing.unwrap_or(match request.source {
|
||||
HealRequestSource::Scanner => false,
|
||||
HealRequestSource::Admin
|
||||
|
||||
@@ -12,6 +12,7 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::{HealErrorClass, classify_heal_error};
|
||||
use crate::heal::{
|
||||
progress::{HealProgress, add_bytes, increment_counter},
|
||||
resume::{
|
||||
@@ -34,7 +35,7 @@ use std::time::{Duration, SystemTime, UNIX_EPOCH};
|
||||
use tokio::sync::{RwLock, Semaphore};
|
||||
use tracing::{debug, error, warn};
|
||||
|
||||
use super::{DiskStore, EcstoreError, POOL_META_NAME, RUSTFS_META_BUCKET};
|
||||
use super::{DiskStore, POOL_META_NAME, RUSTFS_META_BUCKET};
|
||||
|
||||
/// Outcome of classifying an error returned by [`HealStorageAPI::heal_object`].
|
||||
enum HealObjectOutcome {
|
||||
@@ -311,70 +312,21 @@ impl ErasureSetHealer {
|
||||
}
|
||||
}
|
||||
|
||||
/// Classify an error returned by [`HealStorageAPI::heal_object`].
|
||||
/// Map the shared heal error class onto this healer's per-object actions.
|
||||
///
|
||||
/// Most heal object failures wrap `Error::Storage(StorageError)`, while
|
||||
/// compatibility markers can also arrive through Disk/Io/task wrappers.
|
||||
/// `classify_heal_error` is the only classifier; this adapter decides what
|
||||
/// a replacement pass does with each class.
|
||||
fn classify_heal_object_error(err: &Error) -> HealObjectOutcome {
|
||||
if let Error::DanglingDeleteDeferred { retry_not_before } = err {
|
||||
return HealObjectOutcome::DanglingDeleteDeferred {
|
||||
retry_not_before: *retry_not_before,
|
||||
};
|
||||
match classify_heal_error(err) {
|
||||
HealErrorClass::Absent => HealObjectOutcome::Absent,
|
||||
HealErrorClass::DanglingGrace {
|
||||
retry_not_before: Some(retry_not_before),
|
||||
} => HealObjectOutcome::DanglingDeleteDeferred { retry_not_before },
|
||||
HealErrorClass::DanglingGrace { retry_not_before: None }
|
||||
| HealErrorClass::Transient
|
||||
| HealErrorClass::LockContention => HealObjectOutcome::Transient,
|
||||
HealErrorClass::NoHealRequired | HealErrorClass::Aborted | HealErrorClass::Permanent => HealObjectOutcome::Failed,
|
||||
}
|
||||
if err.is_dangling_delete_grace() {
|
||||
if let Some(retry_not_before) = err
|
||||
.dangling_delete_retry_not_before()
|
||||
.and_then(|deadline| deadline.duration_since(UNIX_EPOCH).ok())
|
||||
.and_then(|deadline| deadline.as_secs().checked_add(1))
|
||||
{
|
||||
// Storage reports whole remaining seconds. Round the absolute
|
||||
// deadline up so subsecond grace boundaries cannot hot-loop.
|
||||
return HealObjectOutcome::DanglingDeleteDeferred { retry_not_before };
|
||||
}
|
||||
return HealObjectOutcome::Transient;
|
||||
}
|
||||
|
||||
let Error::Storage(se) = err else {
|
||||
return HealObjectOutcome::Failed;
|
||||
};
|
||||
|
||||
if let EcstoreError::Lock(lock_error) = se
|
||||
&& !lock_error.is_fatal()
|
||||
{
|
||||
return HealObjectOutcome::Transient;
|
||||
}
|
||||
|
||||
// Genuine object/version absence: nothing left to heal, treat as handled.
|
||||
if matches!(
|
||||
se,
|
||||
EcstoreError::FileNotFound
|
||||
| EcstoreError::FileVersionNotFound
|
||||
| EcstoreError::ObjectNotFound(_, _)
|
||||
| EcstoreError::VersionNotFound(_, _, _)
|
||||
) {
|
||||
return HealObjectOutcome::Absent;
|
||||
}
|
||||
|
||||
// Transient infrastructure conditions — skip and retry on a later pass.
|
||||
// NOTE: do NOT use `StorageError::is_not_found()` here: it lumps
|
||||
// `DiskNotFound`/`VolumeNotFound` together with object absence, which is
|
||||
// exactly the conflation that previously let an offline drive be recorded
|
||||
// as "healed/absent" and permanently skipped (backlog#856 / #799 B7).
|
||||
if se.is_quorum_error()
|
||||
|| matches!(
|
||||
se,
|
||||
EcstoreError::DiskNotFound
|
||||
| EcstoreError::VolumeNotFound
|
||||
| EcstoreError::FaultyDisk
|
||||
| EcstoreError::FaultyRemoteDisk
|
||||
| EcstoreError::SlowDown
|
||||
| EcstoreError::OperationCanceled
|
||||
)
|
||||
{
|
||||
return HealObjectOutcome::Transient;
|
||||
}
|
||||
|
||||
HealObjectOutcome::Failed
|
||||
}
|
||||
|
||||
pub fn new(
|
||||
@@ -2207,8 +2159,8 @@ mod tests {
|
||||
// classification must not conflate an offline drive (or unmet quorum) with
|
||||
// genuine object absence, or transient failures get recorded as "healed" and
|
||||
// permanently skipped.
|
||||
use super::{EcstoreError, Error, HealObjectOutcome};
|
||||
use crate::heal::DiskError;
|
||||
use super::{Error, HealObjectOutcome};
|
||||
use crate::heal::{DiskError, EcstoreError};
|
||||
|
||||
fn classify(err: EcstoreError) -> HealObjectOutcome {
|
||||
ErasureSetHealer::classify_heal_object_error(&Error::Storage(err))
|
||||
@@ -2302,6 +2254,82 @@ mod tests {
|
||||
HealObjectOutcome::Failed
|
||||
));
|
||||
}
|
||||
|
||||
// backlog#2900: one classifier for every wrapper. These inputs used to
|
||||
// disagree between the erasure-set healer and the task-level predicates.
|
||||
#[test]
|
||||
fn classify_disk_level_object_absence_as_absent() {
|
||||
assert!(matches!(
|
||||
ErasureSetHealer::classify_heal_object_error(&Error::Disk(DiskError::FileNotFound)),
|
||||
HealObjectOutcome::Absent
|
||||
));
|
||||
assert!(matches!(
|
||||
ErasureSetHealer::classify_heal_object_error(&Error::Disk(DiskError::FileVersionNotFound)),
|
||||
HealObjectOutcome::Absent
|
||||
));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_wrapped_transient_errors_as_transient() {
|
||||
for error in [
|
||||
Error::Storage(EcstoreError::RemoteClientUnavailable("rebuilding client".into())),
|
||||
Error::Storage(EcstoreError::Timeout),
|
||||
Error::Disk(DiskError::DiskNotFound),
|
||||
Error::Disk(DiskError::VolumeNotFound),
|
||||
Error::Disk(DiskError::ErasureReadQuorum),
|
||||
Error::Disk(DiskError::Io(std::io::Error::from(std::io::ErrorKind::WouldBlock))),
|
||||
Error::TransientSkip {
|
||||
message: "skipped".into(),
|
||||
},
|
||||
Error::Storage(EcstoreError::Io(std::io::Error::other(
|
||||
"heal rename incomplete: 1 of 2 targets committed",
|
||||
))),
|
||||
] {
|
||||
assert!(
|
||||
matches!(ErasureSetHealer::classify_heal_object_error(&error), HealObjectOutcome::Transient),
|
||||
"{error:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_partial_not_found_with_lost_quorum_as_transient_not_absent() {
|
||||
// One disk answering NotFound while the others time out surfaces as a
|
||||
// quorum failure from storage; it is never an absence verdict.
|
||||
for error in [
|
||||
Error::Storage(EcstoreError::ErasureReadQuorum),
|
||||
Error::Storage(EcstoreError::InsufficientReadQuorum("bucket".into(), "object".into())),
|
||||
Error::Disk(DiskError::ErasureReadQuorum),
|
||||
Error::Disk(DiskError::Timeout),
|
||||
] {
|
||||
assert!(
|
||||
matches!(ErasureSetHealer::classify_heal_object_error(&error), HealObjectOutcome::Transient),
|
||||
"{error:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn classify_legacy_peer_display_not_found_as_absent() {
|
||||
for message in ["File not found", "File version not found", "Object not found: bucket/object"] {
|
||||
assert!(
|
||||
matches!(
|
||||
ErasureSetHealer::classify_heal_object_error(&Error::Other(message.into())),
|
||||
HealObjectOutcome::Absent
|
||||
),
|
||||
"{message:?}"
|
||||
);
|
||||
}
|
||||
for message in ["file not found", "Disk not found", "Volume not found", "boom"] {
|
||||
assert!(
|
||||
!matches!(
|
||||
ErasureSetHealer::classify_heal_object_error(&Error::Other(message.into())),
|
||||
HealObjectOutcome::Absent
|
||||
),
|
||||
"{message:?}"
|
||||
);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
|
||||
@@ -1005,6 +1005,9 @@ pub struct HealManager {
|
||||
/// Bounded, low-cardinality admission telemetry exposed through the
|
||||
/// existing operations snapshot for cluster E2E assertions.
|
||||
admission_telemetry: Arc<StdMutex<HealAdmissionTelemetry>>,
|
||||
/// Node-local memory of terminal per-object verdicts, consulted before
|
||||
/// queue admission (backlog#2900). Not a receipt; empty after a restart.
|
||||
disposition_cache: Arc<StdMutex<HealDispositionCache>>,
|
||||
}
|
||||
|
||||
/// Where a task-id lookup resolved. The variants carry the resolved state
|
||||
@@ -1037,6 +1040,7 @@ struct HealQueueContext<'a> {
|
||||
notify: &'a Arc<Notify>,
|
||||
cancel_token: &'a CancellationToken,
|
||||
workload_provider: &'a Option<WorkloadSnapshotProviderRef>,
|
||||
disposition_cache: &'a Arc<StdMutex<HealDispositionCache>>,
|
||||
}
|
||||
|
||||
impl HealManager {
|
||||
@@ -1060,6 +1064,37 @@ impl HealManager {
|
||||
}
|
||||
}
|
||||
|
||||
/// Requests answered from the disposition memory: best-effort object
|
||||
/// heals. Administrator starts, forced starts, high and urgent priorities,
|
||||
/// dry runs and durable MRF repairs always reach the queue.
|
||||
fn disposition_gate(request: &HealRequest, mrf_notice_target: Option<&MrfRepairNoticeTarget>) -> Option<HealDispositionKey> {
|
||||
let HealType::Object {
|
||||
bucket,
|
||||
object,
|
||||
version_id,
|
||||
} = &request.heal_type
|
||||
else {
|
||||
return None;
|
||||
};
|
||||
let durable_mrf = request.expected_mrf_bucket_incarnation_id.is_some()
|
||||
|| mrf_notice_target.is_some_and(|target| target.kind.is_durable());
|
||||
if request.force_start
|
||||
|| request.source == HealRequestSource::Admin
|
||||
|| request.priority > HealPriority::Normal
|
||||
|| request.options.dry_run
|
||||
|| durable_mrf
|
||||
{
|
||||
return None;
|
||||
}
|
||||
Some(HealDispositionKey::new(
|
||||
bucket,
|
||||
object,
|
||||
version_id.as_deref(),
|
||||
request.options.pool_index,
|
||||
request.options.set_index,
|
||||
))
|
||||
}
|
||||
|
||||
fn can_displace_queued_work(request: &HealRequest) -> bool {
|
||||
matches!(request.source, HealRequestSource::Admin | HealRequestSource::Internal)
|
||||
}
|
||||
@@ -1609,6 +1644,7 @@ impl HealManager {
|
||||
notify: Arc::new(Notify::new()),
|
||||
workload_provider,
|
||||
admission_telemetry: Arc::new(StdMutex::new(HealAdmissionTelemetry::default())),
|
||||
disposition_cache: Arc::new(StdMutex::new(HealDispositionCache::default())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -2256,6 +2292,50 @@ impl HealManager {
|
||||
});
|
||||
}
|
||||
|
||||
// A best-effort resubmission of an object whose last task reached a
|
||||
// terminal verdict is answered from the node-local disposition memory
|
||||
// before it can occupy the queue again (backlog#2900).
|
||||
if let Some(key) = Self::disposition_gate(&request, mrf_notice_target.as_ref()) {
|
||||
let verdict = lock_disposition_cache(&self.disposition_cache).consult(&key, tokio::time::Instant::now());
|
||||
if let Some((disposition, recorded_task_id, hits, retry_after)) = verdict.rejection() {
|
||||
let recorded_task_id = recorded_task_id.to_owned();
|
||||
drop(retrying_heals);
|
||||
drop(queue);
|
||||
drop(active_heals);
|
||||
let lock_phase = lock_phase_start.elapsed();
|
||||
let admission = HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped);
|
||||
Self::record_admission_metric(source, admission, "disposition_cache");
|
||||
self.record_admission_observation(HealAdmissionObservation {
|
||||
source,
|
||||
result: admission,
|
||||
context: "disposition_cache",
|
||||
force_start,
|
||||
displaced: false,
|
||||
start_duration: admission_start.elapsed(),
|
||||
lock_phase,
|
||||
});
|
||||
debug!(
|
||||
target: "rustfs::heal::manager",
|
||||
event = EVENT_HEAL_QUEUE_ADMISSION,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_MANAGER,
|
||||
request_id = %request.id,
|
||||
priority = ?request.priority,
|
||||
source = source.as_str(),
|
||||
disposition = disposition.as_str(),
|
||||
disposition_task_id = %recorded_task_id,
|
||||
disposition_hits = hits,
|
||||
retry_after_ms = retry_after.map(|delay| u64::try_from(delay.as_millis()).unwrap_or(u64::MAX)),
|
||||
result = "dropped_by_disposition",
|
||||
"Heal queue admission answered from the disposition memory"
|
||||
);
|
||||
return Ok(HealAdmissionReceipt {
|
||||
result: admission,
|
||||
task_id: recorded_task_id,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
let durable_handoff = root_recovery::is_admin_heal_recovery(&request.heal_type, request.source);
|
||||
let durable_handoff_required = durable_handoff
|
||||
&& (request.force_start
|
||||
@@ -2969,11 +3049,13 @@ impl std::fmt::Debug for HealManager {
|
||||
}
|
||||
|
||||
mod auto_scan;
|
||||
mod disposition;
|
||||
mod queue;
|
||||
pub(crate) mod root_recovery;
|
||||
mod scheduler;
|
||||
mod unclean_shutdown;
|
||||
|
||||
use disposition::{HealDispositionCache, HealDispositionKey, lock_disposition_cache};
|
||||
use queue::*;
|
||||
use scheduler::*;
|
||||
use unclean_shutdown::*;
|
||||
|
||||
@@ -0,0 +1,475 @@
|
||||
// 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.
|
||||
|
||||
//! Node-local memory of terminal per-object heal dispositions.
|
||||
//!
|
||||
//! The queue releases its dedup key when a request leaves the queue, so a
|
||||
//! best-effort producer (scanner, read repair, autoheal, MRF hints) can
|
||||
//! resubmit the same object as soon as its task finished, and a task that
|
||||
//! found nothing to repair is forgotten immediately. This cache remembers the
|
||||
//! last terminal verdict per `(bucket, object, version, pool, set)` so admission can
|
||||
//! answer a resubmission without queueing it again:
|
||||
//!
|
||||
//! - `Absent`: a locked quorum read classified the object as absent
|
||||
//! ([`crate::error::HealErrorClass::Absent`]). Resubmissions are dropped
|
||||
//! until the entry expires.
|
||||
//! - `RetryExhausted` / `Permanent`: the task ended in a terminal failure.
|
||||
//! Resubmissions are admitted again on an exponential schedule that is
|
||||
//! capped at one hour, so a failing object keeps being retried but no longer
|
||||
//! on every producer tick.
|
||||
//!
|
||||
//! The cache is bounded (least recently used eviction) and every entry
|
||||
//! expires. It is process memory: a restart starts empty, and nothing here is
|
||||
//! a receipt. It never releases a durable responsibility; durable MRF repairs
|
||||
//! bypass it entirely, as do administrator starts, forced starts, high and
|
||||
//! urgent priorities, and dry runs.
|
||||
|
||||
use crate::error::{HealErrorClass, classify_heal_error};
|
||||
use crate::heal::outcome::HealObjectDisposition;
|
||||
use crate::heal::task::{HealTask, HealType};
|
||||
use metrics::{counter, gauge};
|
||||
use std::collections::{HashMap, VecDeque};
|
||||
use std::sync::{Mutex as StdMutex, MutexGuard as StdMutexGuard};
|
||||
use std::time::Duration;
|
||||
use tokio::time::Instant;
|
||||
|
||||
/// Default upper bound on retained entries.
|
||||
pub(crate) const DEFAULT_DISPOSITION_CACHE_CAPACITY: usize = 100_000;
|
||||
/// Default lifetime of an entry after its last record.
|
||||
pub(crate) const DEFAULT_DISPOSITION_CACHE_TTL: Duration = Duration::from_secs(24 * 60 * 60);
|
||||
/// First re-admission delay after a terminal failure; doubles per repeat.
|
||||
const RETRY_BACKOFF_BASE: Duration = Duration::from_secs(60);
|
||||
/// Upper bound of the re-admission delay.
|
||||
const RETRY_BACKOFF_MAX: Duration = Duration::from_secs(60 * 60);
|
||||
|
||||
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||
pub(crate) enum HealDispositionKind {
|
||||
Absent,
|
||||
RetryExhausted,
|
||||
Permanent,
|
||||
}
|
||||
|
||||
impl HealDispositionKind {
|
||||
pub(crate) const fn as_str(self) -> &'static str {
|
||||
match self {
|
||||
Self::Absent => "absent",
|
||||
Self::RetryExhausted => "retry_exhausted",
|
||||
Self::Permanent => "permanent",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
|
||||
pub(crate) struct HealDispositionKey {
|
||||
pub(crate) bucket: String,
|
||||
pub(crate) object: String,
|
||||
/// `None` is the unversioned/latest target; an empty wire string is the same thing.
|
||||
pub(crate) version_id: Option<String>,
|
||||
pub(crate) pool_index: Option<usize>,
|
||||
pub(crate) set_index: Option<usize>,
|
||||
}
|
||||
|
||||
impl HealDispositionKey {
|
||||
pub(crate) fn new(
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
version_id: Option<&str>,
|
||||
pool_index: Option<usize>,
|
||||
set_index: Option<usize>,
|
||||
) -> Self {
|
||||
Self {
|
||||
bucket: bucket.to_owned(),
|
||||
object: object.to_owned(),
|
||||
version_id: version_id.filter(|version| !version.is_empty()).map(ToOwned::to_owned),
|
||||
pool_index,
|
||||
set_index,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(crate) struct HealDispositionEntry {
|
||||
pub(crate) kind: HealDispositionKind,
|
||||
/// Consecutive records of the same kind for this key.
|
||||
pub(crate) count: u32,
|
||||
pub(crate) recorded_at: Instant,
|
||||
/// Task that produced the latest record; returned to producers as the receipt pointer.
|
||||
pub(crate) task_id: String,
|
||||
sequence: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub(crate) enum HealDispositionVerdict {
|
||||
Admit,
|
||||
/// The last locked quorum read found the object absent and the memory is still fresh.
|
||||
Absent {
|
||||
task_id: String,
|
||||
count: u32,
|
||||
},
|
||||
/// The last task failed terminally; re-admission waits for the backoff to elapse.
|
||||
BackedOff {
|
||||
kind: HealDispositionKind,
|
||||
task_id: String,
|
||||
count: u32,
|
||||
retry_after: Duration,
|
||||
},
|
||||
}
|
||||
|
||||
impl HealDispositionVerdict {
|
||||
pub(crate) fn rejection(&self) -> Option<(HealDispositionKind, &str, u32, Option<Duration>)> {
|
||||
match self {
|
||||
Self::Admit => None,
|
||||
Self::Absent { task_id, count } => Some((HealDispositionKind::Absent, task_id, *count, None)),
|
||||
Self::BackedOff {
|
||||
kind,
|
||||
task_id,
|
||||
count,
|
||||
retry_after,
|
||||
} => Some((*kind, task_id, *count, Some(*retry_after))),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub(crate) struct HealDispositionCache {
|
||||
capacity: usize,
|
||||
ttl: Duration,
|
||||
entries: HashMap<HealDispositionKey, HealDispositionEntry>,
|
||||
/// Recency order. A key can appear several times; only the pair whose
|
||||
/// sequence matches the entry is live, older pairs are skipped on eviction.
|
||||
recency: VecDeque<(u64, HealDispositionKey)>,
|
||||
next_sequence: u64,
|
||||
}
|
||||
|
||||
impl Default for HealDispositionCache {
|
||||
fn default() -> Self {
|
||||
Self::with_limits(DEFAULT_DISPOSITION_CACHE_CAPACITY, DEFAULT_DISPOSITION_CACHE_TTL)
|
||||
}
|
||||
}
|
||||
|
||||
impl HealDispositionCache {
|
||||
pub(crate) fn with_limits(capacity: usize, ttl: Duration) -> Self {
|
||||
Self {
|
||||
capacity: capacity.max(1),
|
||||
ttl,
|
||||
entries: HashMap::new(),
|
||||
recency: VecDeque::new(),
|
||||
next_sequence: 0,
|
||||
}
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn len(&self) -> usize {
|
||||
self.entries.len()
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
pub(crate) fn get(&self, key: &HealDispositionKey) -> Option<&HealDispositionEntry> {
|
||||
self.entries.get(key)
|
||||
}
|
||||
|
||||
/// Delay before a key with `count` consecutive terminal failures is admitted again.
|
||||
pub(crate) fn retry_backoff(count: u32) -> Duration {
|
||||
let exponent = count.saturating_sub(1).min(31);
|
||||
RETRY_BACKOFF_BASE.saturating_mul(1_u32 << exponent).min(RETRY_BACKOFF_MAX)
|
||||
}
|
||||
|
||||
/// Decide whether a resubmission for `key` may enter the queue at `now`.
|
||||
pub(crate) fn consult(&mut self, key: &HealDispositionKey, now: Instant) -> HealDispositionVerdict {
|
||||
let Some(entry) = self.entries.get(key) else {
|
||||
return HealDispositionVerdict::Admit;
|
||||
};
|
||||
let age = now.saturating_duration_since(entry.recorded_at);
|
||||
if age >= self.ttl {
|
||||
self.entries.remove(key);
|
||||
self.publish_entries();
|
||||
return HealDispositionVerdict::Admit;
|
||||
}
|
||||
let verdict = match entry.kind {
|
||||
HealDispositionKind::Absent => HealDispositionVerdict::Absent {
|
||||
task_id: entry.task_id.clone(),
|
||||
count: entry.count,
|
||||
},
|
||||
kind @ (HealDispositionKind::RetryExhausted | HealDispositionKind::Permanent) => {
|
||||
let backoff = Self::retry_backoff(entry.count);
|
||||
if age >= backoff {
|
||||
return HealDispositionVerdict::Admit;
|
||||
}
|
||||
HealDispositionVerdict::BackedOff {
|
||||
kind,
|
||||
task_id: entry.task_id.clone(),
|
||||
count: entry.count,
|
||||
retry_after: backoff.saturating_sub(age),
|
||||
}
|
||||
}
|
||||
};
|
||||
counter!("rustfs_heal_disposition_cache_hits_total", "disposition" => entry.kind.as_str()).increment(1);
|
||||
self.touch(key);
|
||||
verdict
|
||||
}
|
||||
|
||||
/// Remember the terminal verdict of `task_id` for `key`.
|
||||
///
|
||||
/// A repeat of the same kind increments `count`; a different kind starts
|
||||
/// a fresh count, because the backoff schedule belongs to one failure mode.
|
||||
pub(crate) fn record(&mut self, key: HealDispositionKey, kind: HealDispositionKind, task_id: &str, now: Instant) {
|
||||
let sequence = self.next_sequence();
|
||||
let count = match self.entries.get(&key) {
|
||||
Some(existing) if existing.kind == kind && now.saturating_duration_since(existing.recorded_at) < self.ttl => {
|
||||
existing.count.saturating_add(1)
|
||||
}
|
||||
_ => 1,
|
||||
};
|
||||
self.entries.insert(
|
||||
key.clone(),
|
||||
HealDispositionEntry {
|
||||
kind,
|
||||
count,
|
||||
recorded_at: now,
|
||||
task_id: task_id.to_owned(),
|
||||
sequence,
|
||||
},
|
||||
);
|
||||
self.recency.push_back((sequence, key));
|
||||
self.evict_to_capacity();
|
||||
self.compact_recency();
|
||||
self.publish_entries();
|
||||
}
|
||||
|
||||
/// Forget `key` after a task completed with a positive storage result.
|
||||
pub(crate) fn clear(&mut self, key: &HealDispositionKey) {
|
||||
if self.entries.remove(key).is_some() {
|
||||
self.publish_entries();
|
||||
}
|
||||
}
|
||||
|
||||
fn next_sequence(&mut self) -> u64 {
|
||||
self.next_sequence = self.next_sequence.wrapping_add(1);
|
||||
self.next_sequence
|
||||
}
|
||||
|
||||
fn touch(&mut self, key: &HealDispositionKey) {
|
||||
let sequence = self.next_sequence();
|
||||
if let Some(entry) = self.entries.get_mut(key) {
|
||||
entry.sequence = sequence;
|
||||
self.recency.push_back((sequence, key.clone()));
|
||||
self.compact_recency();
|
||||
}
|
||||
}
|
||||
|
||||
fn evict_to_capacity(&mut self) {
|
||||
while self.entries.len() > self.capacity {
|
||||
let Some((sequence, key)) = self.recency.pop_front() else {
|
||||
// Every live entry has a recency pair; an empty queue with
|
||||
// entries left cannot happen, but never spin if it does.
|
||||
break;
|
||||
};
|
||||
if self.entries.get(&key).is_some_and(|entry| entry.sequence == sequence) {
|
||||
self.entries.remove(&key);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
fn compact_recency(&mut self) {
|
||||
if self.recency.len() > self.capacity.saturating_mul(2).saturating_add(16) {
|
||||
let entries = &self.entries;
|
||||
self.recency
|
||||
.retain(|(sequence, key)| entries.get(key).is_some_and(|entry| entry.sequence == *sequence));
|
||||
}
|
||||
}
|
||||
|
||||
fn publish_entries(&self) {
|
||||
gauge!("rustfs_heal_disposition_cache_entries").set(self.entries.len() as f64);
|
||||
}
|
||||
}
|
||||
|
||||
pub(super) fn lock_disposition_cache(cache: &StdMutex<HealDispositionCache>) -> StdMutexGuard<'_, HealDispositionCache> {
|
||||
match cache.lock() {
|
||||
Ok(guard) => guard,
|
||||
Err(poisoned) => poisoned.into_inner(),
|
||||
}
|
||||
}
|
||||
|
||||
/// Remember the terminal verdict of a finished single-object task, or forget
|
||||
/// the key after a positive completion. Scheduled retries, cancellations,
|
||||
/// deadlines, dry runs and durable MRF repairs leave the memory untouched.
|
||||
pub(crate) async fn record_task_disposition(
|
||||
cache: &StdMutex<HealDispositionCache>,
|
||||
task: &HealTask,
|
||||
result: &crate::Result<()>,
|
||||
will_retry: bool,
|
||||
) {
|
||||
let HealType::Object {
|
||||
bucket,
|
||||
object,
|
||||
version_id,
|
||||
} = &task.heal_type
|
||||
else {
|
||||
return;
|
||||
};
|
||||
if task.options.dry_run || task.expected_mrf_bucket_incarnation_id.is_some() {
|
||||
return;
|
||||
}
|
||||
let key = HealDispositionKey::new(bucket, object, version_id.as_deref(), task.options.pool_index, task.options.set_index);
|
||||
let kind = match result {
|
||||
Ok(()) if task.object_absent() => HealDispositionKind::Absent,
|
||||
Ok(()) => {
|
||||
// A deferral (grace window, transient skip) is neither a repair
|
||||
// nor a failure; only a positive completion forgets the key.
|
||||
let deferred = task
|
||||
.get_outcome()
|
||||
.await
|
||||
.objects
|
||||
.iter()
|
||||
.any(|object| matches!(object.disposition, HealObjectDisposition::Deferred { .. }));
|
||||
if !deferred {
|
||||
lock_disposition_cache(cache).clear(&key);
|
||||
}
|
||||
return;
|
||||
}
|
||||
Err(_) if will_retry => return,
|
||||
Err(error) => match classify_heal_error(error) {
|
||||
HealErrorClass::Aborted => return,
|
||||
HealErrorClass::Absent => HealDispositionKind::Absent,
|
||||
HealErrorClass::NoHealRequired => {
|
||||
lock_disposition_cache(cache).clear(&key);
|
||||
return;
|
||||
}
|
||||
HealErrorClass::Transient | HealErrorClass::LockContention | HealErrorClass::DanglingGrace { .. } => {
|
||||
HealDispositionKind::RetryExhausted
|
||||
}
|
||||
HealErrorClass::Permanent => HealDispositionKind::Permanent,
|
||||
},
|
||||
};
|
||||
lock_disposition_cache(cache).record(key, kind, &task.id, Instant::now());
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
fn key(object: &str) -> HealDispositionKey {
|
||||
HealDispositionKey::new("bucket", object, None, None, None)
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_cache_absent_hit_until_ttl_expires() {
|
||||
let mut cache = HealDispositionCache::with_limits(10, Duration::from_secs(60));
|
||||
let now = Instant::now();
|
||||
cache.record(key("gone"), HealDispositionKind::Absent, "task-1", now);
|
||||
assert_eq!(
|
||||
cache.consult(&key("gone"), now + Duration::from_secs(59)),
|
||||
HealDispositionVerdict::Absent {
|
||||
task_id: "task-1".into(),
|
||||
count: 1
|
||||
}
|
||||
);
|
||||
assert_eq!(cache.consult(&key("gone"), now + Duration::from_secs(60)), HealDispositionVerdict::Admit);
|
||||
assert_eq!(cache.len(), 0, "an expired entry is dropped on consult");
|
||||
assert_eq!(cache.consult(&key("other"), now), HealDispositionVerdict::Admit);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_cache_retry_exhausted_backs_off_exponentially_up_to_one_hour() {
|
||||
let mut cache = HealDispositionCache::with_limits(10, Duration::from_secs(24 * 60 * 60));
|
||||
let now = Instant::now();
|
||||
cache.record(key("flaky"), HealDispositionKind::RetryExhausted, "task-1", now);
|
||||
// The fourth submission (first after the task's own three retries) is backed off.
|
||||
assert_eq!(
|
||||
cache.consult(&key("flaky"), now + Duration::from_secs(30)),
|
||||
HealDispositionVerdict::BackedOff {
|
||||
kind: HealDispositionKind::RetryExhausted,
|
||||
task_id: "task-1".into(),
|
||||
count: 1,
|
||||
retry_after: Duration::from_secs(30),
|
||||
}
|
||||
);
|
||||
assert_eq!(cache.consult(&key("flaky"), now + Duration::from_secs(60)), HealDispositionVerdict::Admit);
|
||||
let later = now + Duration::from_secs(61);
|
||||
cache.record(key("flaky"), HealDispositionKind::RetryExhausted, "task-2", later);
|
||||
assert_eq!(cache.get(&key("flaky")).map(|entry| entry.count), Some(2));
|
||||
assert!(matches!(
|
||||
cache.consult(&key("flaky"), later + Duration::from_secs(119)),
|
||||
HealDispositionVerdict::BackedOff { count: 2, .. }
|
||||
));
|
||||
assert_eq!(
|
||||
cache.consult(&key("flaky"), later + Duration::from_secs(120)),
|
||||
HealDispositionVerdict::Admit
|
||||
);
|
||||
for count in 1..=40 {
|
||||
assert!(HealDispositionCache::retry_backoff(count) <= Duration::from_secs(3600), "count={count}");
|
||||
}
|
||||
assert_eq!(HealDispositionCache::retry_backoff(6), Duration::from_secs(1920));
|
||||
assert_eq!(HealDispositionCache::retry_backoff(7), Duration::from_secs(3600));
|
||||
assert_eq!(HealDispositionCache::retry_backoff(u32::MAX), Duration::from_secs(3600));
|
||||
// A different terminal kind starts its own count.
|
||||
cache.record(key("flaky"), HealDispositionKind::Permanent, "task-3", later);
|
||||
assert_eq!(cache.get(&key("flaky")).map(|entry| entry.count), Some(1));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_cache_evicts_least_recently_used_when_full() {
|
||||
let mut cache = HealDispositionCache::with_limits(3, Duration::from_secs(600));
|
||||
let now = Instant::now();
|
||||
for object in ["a", "b", "c"] {
|
||||
cache.record(key(object), HealDispositionKind::Absent, "task", now);
|
||||
}
|
||||
// Touch "a" so "b" becomes the oldest.
|
||||
assert!(matches!(cache.consult(&key("a"), now), HealDispositionVerdict::Absent { .. }));
|
||||
cache.record(key("d"), HealDispositionKind::Absent, "task", now);
|
||||
assert_eq!(cache.len(), 3);
|
||||
assert!(cache.get(&key("b")).is_none(), "the oldest entry is evicted, not the new one");
|
||||
for object in ["a", "c", "d"] {
|
||||
assert!(cache.get(&key(object)).is_some(), "{object}");
|
||||
}
|
||||
// Stale recency pairs never evict a re-recorded key by mistake.
|
||||
cache.record(key("a"), HealDispositionKind::Absent, "task", now);
|
||||
cache.record(key("e"), HealDispositionKind::Absent, "task", now);
|
||||
assert!(cache.get(&key("a")).is_some());
|
||||
assert!(cache.get(&key("c")).is_none());
|
||||
assert_eq!(cache.len(), 3);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_cache_success_clears_the_entry() {
|
||||
let mut cache = HealDispositionCache::with_limits(10, Duration::from_secs(600));
|
||||
let now = Instant::now();
|
||||
cache.record(key("fixed"), HealDispositionKind::Permanent, "task-1", now);
|
||||
cache.clear(&key("fixed"));
|
||||
assert_eq!(cache.consult(&key("fixed"), now), HealDispositionVerdict::Admit);
|
||||
assert_eq!(cache.len(), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_cache_recency_queue_stays_bounded_under_repeated_hits() {
|
||||
let mut cache = HealDispositionCache::with_limits(4, Duration::from_secs(600));
|
||||
let now = Instant::now();
|
||||
cache.record(key("hot"), HealDispositionKind::Absent, "task", now);
|
||||
for _ in 0..10_000 {
|
||||
assert!(matches!(cache.consult(&key("hot"), now), HealDispositionVerdict::Absent { .. }));
|
||||
}
|
||||
assert!(cache.recency.len() <= 4 * 2 + 16 + 1, "recency={}", cache.recency.len());
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn disposition_key_normalizes_empty_version() {
|
||||
assert_eq!(
|
||||
HealDispositionKey::new("b", "o", Some(""), None, None),
|
||||
HealDispositionKey::new("b", "o", None, None, None)
|
||||
);
|
||||
assert_ne!(
|
||||
HealDispositionKey::new("b", "o", Some("v1"), None, None),
|
||||
HealDispositionKey::new("b", "o", None, None, None)
|
||||
);
|
||||
}
|
||||
}
|
||||
@@ -11,6 +11,7 @@
|
||||
// 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::disposition::record_task_disposition;
|
||||
/// The heal scheduler: queue consumption loop and its skip/metric helpers.
|
||||
use super::*;
|
||||
|
||||
@@ -47,6 +48,7 @@ impl HealManager {
|
||||
let storage = self.storage.clone();
|
||||
let notify = self.notify.clone();
|
||||
let workload_provider = self.workload_provider.clone();
|
||||
let disposition_cache = self.disposition_cache.clone();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut interval = interval(config.read().await.heal_interval);
|
||||
@@ -82,6 +84,7 @@ impl HealManager {
|
||||
notify: ¬ify,
|
||||
cancel_token: &cancel_token,
|
||||
workload_provider: &workload_provider,
|
||||
disposition_cache: &disposition_cache,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
@@ -102,6 +105,7 @@ impl HealManager {
|
||||
notify: ¬ify,
|
||||
cancel_token: &cancel_token,
|
||||
workload_provider: &workload_provider,
|
||||
disposition_cache: &disposition_cache,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
@@ -131,6 +135,7 @@ impl HealManager {
|
||||
notify,
|
||||
cancel_token,
|
||||
workload_provider,
|
||||
disposition_cache,
|
||||
} = context;
|
||||
|
||||
let config = config.read().await;
|
||||
@@ -227,6 +232,7 @@ impl HealManager {
|
||||
let retrying_heals_clone = retrying_heals.clone();
|
||||
let mrf_repair_notice_targets_clone = mrf_repair_notice_targets.clone();
|
||||
let replacement_recovery_anchors_clone = replacement_recovery_anchors.clone();
|
||||
let disposition_cache_clone = disposition_cache.clone();
|
||||
let statistics_clone = statistics.clone();
|
||||
let notify_clone = notify.clone();
|
||||
let root_recovery_clone = root_recovery.clone();
|
||||
@@ -251,6 +257,7 @@ impl HealManager {
|
||||
);
|
||||
let result = task.execute().await;
|
||||
let retry_request = retry_request_for_result_with_budget(task.as_ref(), &result).await;
|
||||
record_task_disposition(&disposition_cache_clone, task.as_ref(), &result, retry_request.is_some()).await;
|
||||
match &result {
|
||||
Ok(_) => {
|
||||
debug!(
|
||||
|
||||
@@ -30,6 +30,7 @@ pub(super) mod admin_overlap;
|
||||
mod root_recovery;
|
||||
use uuid::Uuid;
|
||||
mod auto_scan;
|
||||
mod disposition;
|
||||
mod running_mainline;
|
||||
|
||||
use super::super::{DiskOption, DiskStore, Endpoint, new_disk, storage_api::status::BucketInfo};
|
||||
@@ -105,6 +106,7 @@ async fn process_manager_queue_once(manager: &HealManager) {
|
||||
notify: &manager.notify,
|
||||
cancel_token: &manager.cancel_token,
|
||||
workload_provider: &manager.workload_provider,
|
||||
disposition_cache: &manager.disposition_cache,
|
||||
})
|
||||
.await;
|
||||
}
|
||||
|
||||
@@ -0,0 +1,392 @@
|
||||
// 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.
|
||||
|
||||
//! Admission-side behaviour of the per-object disposition cache (backlog#2900).
|
||||
|
||||
use super::*;
|
||||
use crate::heal::manager::disposition::{HealDispositionCache, HealDispositionKey, HealDispositionKind, record_task_disposition};
|
||||
use crate::heal::storage::HealListItem;
|
||||
use rustfs_heal_contracts::heal_channel::{HealAdmissionDropReason, HealAdmissionResult};
|
||||
use std::time::Duration;
|
||||
use tokio::time::Instant;
|
||||
|
||||
const BUCKET: &str = "bucket-a";
|
||||
|
||||
fn object_request(object: &str, source: HealRequestSource, priority: HealPriority, version: Option<&str>) -> HealRequest {
|
||||
let mut request = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: BUCKET.to_string(),
|
||||
object: object.to_string(),
|
||||
version_id: version.map(str::to_string),
|
||||
},
|
||||
HealOptions::default(),
|
||||
priority,
|
||||
);
|
||||
request.source = source;
|
||||
request
|
||||
}
|
||||
|
||||
fn key(object: &str, version: Option<&str>) -> HealDispositionKey {
|
||||
HealDispositionKey::new(BUCKET, object, version, None, None)
|
||||
}
|
||||
|
||||
fn seed(manager: &HealManager, object: &str, kind: HealDispositionKind, task_id: &str) {
|
||||
manager
|
||||
.disposition_cache
|
||||
.lock()
|
||||
.expect("disposition cache")
|
||||
.record(key(object, None), kind, task_id, Instant::now());
|
||||
}
|
||||
|
||||
fn cached_kind(manager: &HealManager, object: &str) -> Option<HealDispositionKind> {
|
||||
manager
|
||||
.disposition_cache
|
||||
.lock()
|
||||
.expect("disposition cache")
|
||||
.get(&key(object, None))
|
||||
.map(|entry| entry.kind)
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disposition_absent_hit_drops_best_effort_object_requests_before_the_queue() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
seed(&manager, "gone", HealDispositionKind::Absent, "absent-task");
|
||||
for source in [
|
||||
HealRequestSource::ReadRepair,
|
||||
HealRequestSource::Scanner,
|
||||
HealRequestSource::AutoHeal,
|
||||
HealRequestSource::Internal,
|
||||
HealRequestSource::Mrf,
|
||||
] {
|
||||
for priority in [HealPriority::Low, HealPriority::Normal] {
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request("gone", source, priority, None))
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(
|
||||
receipt.result,
|
||||
HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped),
|
||||
"{source:?} {priority:?}"
|
||||
);
|
||||
assert_eq!(receipt.task_id, "absent-task", "the receipt points at the task that observed absence");
|
||||
}
|
||||
}
|
||||
assert!(manager.heal_queue.lock().await.is_empty(), "a cached absence never reaches the queue");
|
||||
// A different version of the same key is a different memory.
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request("gone", HealRequestSource::ReadRepair, HealPriority::Low, Some("v1")))
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disposition_scoped_absence_does_not_block_other_scopes() {
|
||||
for (pool, set, expected) in [
|
||||
(Some(0), Some(0), HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped)),
|
||||
(Some(1), Some(0), HealAdmissionResult::Accepted),
|
||||
(Some(0), Some(1), HealAdmissionResult::Accepted),
|
||||
(Some(0), None, HealAdmissionResult::Accepted),
|
||||
(None, Some(0), HealAdmissionResult::Accepted),
|
||||
(None, None, HealAdmissionResult::Accepted),
|
||||
] {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let mut old_pool = object_request("moved", HealRequestSource::ReadRepair, HealPriority::Low, None);
|
||||
old_pool.options.pool_index = Some(0);
|
||||
old_pool.options.set_index = Some(0);
|
||||
let absent = task_for(old_pool);
|
||||
absent.mark_object_absent();
|
||||
record_task_disposition(&manager.disposition_cache, &absent, &Ok(()), false).await;
|
||||
|
||||
let mut request = object_request("moved", HealRequestSource::ReadRepair, HealPriority::Low, None);
|
||||
request.options.pool_index = pool;
|
||||
request.options.set_index = set;
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(request)
|
||||
.await
|
||||
.expect("scoped admission decides");
|
||||
assert_eq!(receipt.result, expected, "pool={pool:?}, set={set:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn disposition_unhealthy_admin_object_name_does_not_cache_absence() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
let object = "reports/File not found.txt";
|
||||
let task = task_for(object_request(object, HealRequestSource::Admin, HealPriority::Normal, None));
|
||||
let failure = Error::TaskExecutionFailed {
|
||||
message: format!(
|
||||
"Heal left one or more drives unhealthy for {BUCKET}/{object}; retry after the missing drives are restored"
|
||||
),
|
||||
};
|
||||
record_task_disposition(&manager.disposition_cache, &task, &Err(failure), false).await;
|
||||
assert_eq!(cached_kind(&manager, object), Some(HealDispositionKind::Permanent));
|
||||
|
||||
tokio::time::advance(Duration::from_secs(61)).await;
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(object_request(object, HealRequestSource::ReadRepair, HealPriority::Low, None))
|
||||
.await
|
||||
.expect("repair can retry after the failure backoff");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disposition_cache_is_bypassed_by_admin_forced_high_priority_dry_run_durable_mrf_and_other_kinds() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
for object in ["admin", "forced", "high", "urgent", "dry", "durable", "metadata"] {
|
||||
seed(&manager, object, HealDispositionKind::Absent, "absent-task");
|
||||
}
|
||||
let admin = object_request("admin", HealRequestSource::Admin, HealPriority::Normal, None);
|
||||
let mut forced = object_request("forced", HealRequestSource::Scanner, HealPriority::Low, None);
|
||||
forced.force_start = true;
|
||||
let high = object_request("high", HealRequestSource::Scanner, HealPriority::High, None);
|
||||
let urgent = object_request("urgent", HealRequestSource::ReadRepair, HealPriority::Urgent, None);
|
||||
let mut dry = object_request("dry", HealRequestSource::Scanner, HealPriority::Low, None);
|
||||
dry.options.dry_run = true;
|
||||
let mut durable = object_request("durable", HealRequestSource::Mrf, HealPriority::Normal, None);
|
||||
durable.expected_mrf_bucket_incarnation_id = Some(Uuid::new_v4());
|
||||
let mut metadata = HealRequest::new(
|
||||
HealType::Metadata {
|
||||
bucket: BUCKET.to_string(),
|
||||
object: "metadata".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Low,
|
||||
);
|
||||
metadata.source = HealRequestSource::Scanner;
|
||||
for (label, request) in [
|
||||
("admin", admin),
|
||||
("forced", forced),
|
||||
("high", high),
|
||||
("urgent", urgent),
|
||||
("dry", dry),
|
||||
("durable", durable),
|
||||
("metadata", metadata),
|
||||
] {
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(request)
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted, "{label}");
|
||||
}
|
||||
assert_eq!(manager.heal_queue.lock().await.len(), 7);
|
||||
}
|
||||
|
||||
#[tokio::test(start_paused = true)]
|
||||
async fn disposition_backed_off_failure_is_dropped_then_admitted_after_the_backoff() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(Arc::new(MockStorage), None);
|
||||
seed(&manager, "flaky", HealDispositionKind::RetryExhausted, "failed-task");
|
||||
let dropped = manager
|
||||
.submit_heal_request_with_receipt(object_request("flaky", HealRequestSource::Scanner, HealPriority::Low, None))
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(dropped.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped));
|
||||
assert_eq!(dropped.task_id, "failed-task");
|
||||
tokio::time::advance(Duration::from_secs(61)).await;
|
||||
let admitted = manager
|
||||
.submit_heal_request_with_receipt(object_request("flaky", HealRequestSource::Scanner, HealPriority::Low, None))
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(admitted.result, HealAdmissionResult::Accepted, "the backoff elapsed");
|
||||
assert_eq!(
|
||||
cached_kind(&manager, "flaky"),
|
||||
Some(HealDispositionKind::RetryExhausted),
|
||||
"admission does not forget the failure"
|
||||
);
|
||||
}
|
||||
|
||||
fn task_for(request: HealRequest) -> HealTask {
|
||||
HealTask::from_request(request, Arc::new(MockStorage))
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disposition_record_task_result_maps_terminal_outcomes() {
|
||||
let cache = StdMutex::new(HealDispositionCache::with_limits(16, Duration::from_secs(600)));
|
||||
let kind = |object: &str| cache.lock().expect("cache").get(&key(object, None)).map(|entry| entry.kind);
|
||||
|
||||
let absent = task_for(object_request("absent", HealRequestSource::ReadRepair, HealPriority::Low, None));
|
||||
absent.mark_object_absent();
|
||||
record_task_disposition(&cache, &absent, &Ok(()), false).await;
|
||||
assert_eq!(kind("absent"), Some(HealDispositionKind::Absent));
|
||||
|
||||
let retried = task_for(object_request("retried", HealRequestSource::Scanner, HealPriority::Low, None));
|
||||
let quorum = || Error::Storage(EcstoreError::InsufficientReadQuorum(BUCKET.to_string(), "retried".to_string()));
|
||||
record_task_disposition(&cache, &retried, &Err(quorum()), true).await;
|
||||
assert_eq!(kind("retried"), None, "a scheduled retry is not terminal");
|
||||
record_task_disposition(&cache, &retried, &Err(quorum()), false).await;
|
||||
assert_eq!(kind("retried"), Some(HealDispositionKind::RetryExhausted));
|
||||
|
||||
let denied = task_for(object_request("denied", HealRequestSource::AutoHeal, HealPriority::Normal, None));
|
||||
record_task_disposition(&cache, &denied, &Err(Error::Disk(DiskError::FileAccessDenied)), false).await;
|
||||
assert_eq!(kind("denied"), Some(HealDispositionKind::Permanent));
|
||||
|
||||
let cancelled = task_for(object_request("cancelled", HealRequestSource::Scanner, HealPriority::Low, None));
|
||||
record_task_disposition(&cache, &cancelled, &Err(Error::TaskCancelled), false).await;
|
||||
record_task_disposition(&cache, &cancelled, &Err(Error::TaskTimeout), false).await;
|
||||
assert_eq!(kind("cancelled"), None);
|
||||
|
||||
// A positive completion forgets an earlier failure.
|
||||
record_task_disposition(&cache, &denied, &Ok(()), false).await;
|
||||
assert_eq!(kind("denied"), None);
|
||||
|
||||
// Durable MRF, dry runs and other task kinds never write the memory.
|
||||
let mut durable_request = object_request("durable", HealRequestSource::Mrf, HealPriority::Normal, None);
|
||||
durable_request.expected_mrf_bucket_incarnation_id = Some(Uuid::new_v4());
|
||||
let durable = task_for(durable_request);
|
||||
durable.mark_object_absent();
|
||||
record_task_disposition(&cache, &durable, &Ok(()), false).await;
|
||||
assert_eq!(kind("durable"), None);
|
||||
let mut dry_request = object_request("dry", HealRequestSource::Scanner, HealPriority::Low, None);
|
||||
dry_request.options.dry_run = true;
|
||||
let dry = task_for(dry_request);
|
||||
dry.mark_object_absent();
|
||||
record_task_disposition(&cache, &dry, &Ok(()), false).await;
|
||||
assert_eq!(kind("dry"), None);
|
||||
let mut metadata = HealRequest::new(
|
||||
HealType::Metadata {
|
||||
bucket: BUCKET.to_string(),
|
||||
object: "metadata".to_string(),
|
||||
},
|
||||
HealOptions::default(),
|
||||
HealPriority::Low,
|
||||
);
|
||||
metadata.source = HealRequestSource::Scanner;
|
||||
let metadata = task_for(metadata);
|
||||
record_task_disposition(&cache, &metadata, &Err(Error::Disk(DiskError::FileAccessDenied)), false).await;
|
||||
assert_eq!(kind("metadata"), None);
|
||||
}
|
||||
|
||||
struct AbsentStorage;
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl HealStorageAPI for AbsentStorage {
|
||||
async fn admit_bucket_incarnation(&self, _: &str) -> Result<Uuid> {
|
||||
Ok(Uuid::from_u128(7))
|
||||
}
|
||||
async fn heal_bucket_at_incarnation(&self, bucket: &str, expected: Uuid, opts: &HealOpts) -> Result<HealResultItem> {
|
||||
self.validate_bucket_incarnation(bucket, Some(expected)).await?;
|
||||
self.heal_bucket(bucket, opts).await
|
||||
}
|
||||
async fn heal_object_at_incarnation(
|
||||
&self,
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
version_id: Option<&str>,
|
||||
expected: Uuid,
|
||||
opts: &HealOpts,
|
||||
) -> Result<crate::heal::storage::HealStorageObjectResult> {
|
||||
self.validate_bucket_incarnation(bucket, Some(expected)).await?;
|
||||
self.heal_object_with_receipt(bucket, object, version_id, opts).await
|
||||
}
|
||||
async fn get_object_meta(&self, _: &str, _: &str) -> Result<Option<HealObjectInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
async fn ec_decode_rebuild(&self, _: &str, _: &str) -> Result<Vec<u8>> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
async fn get_bucket_info(&self, _: &str) -> Result<Option<BucketInfo>> {
|
||||
Ok(None)
|
||||
}
|
||||
async fn list_buckets_for_admin_heal(
|
||||
&self,
|
||||
_selector: Option<crate::heal::storage_api::storage::DiskSetSelector>,
|
||||
) -> Result<Vec<BucketInfo>> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
async fn list_buckets(&self) -> Result<Vec<BucketInfo>> {
|
||||
Ok(Vec::new())
|
||||
}
|
||||
async fn object_exists(&self, _: &str, _: &str) -> Result<bool> {
|
||||
Ok(false)
|
||||
}
|
||||
async fn heal_bucket(&self, _: &str, _: &HealOpts) -> Result<HealResultItem> {
|
||||
Ok(HealResultItem::default())
|
||||
}
|
||||
async fn heal_format(&self, _: bool) -> Result<(HealResultItem, Option<Error>)> {
|
||||
Ok((HealResultItem::default(), None))
|
||||
}
|
||||
async fn get_disk_for_resume(&self, _: &str) -> Result<DiskStore> {
|
||||
Err(Error::other("no resume disk in this fixture"))
|
||||
}
|
||||
async fn list_objects_for_heal_page(
|
||||
&self,
|
||||
_: &str,
|
||||
_: &str,
|
||||
_: Option<&str>,
|
||||
_: bool,
|
||||
) -> Result<(Vec<HealListItem>, Option<String>, bool)> {
|
||||
Ok((Vec::new(), None, false))
|
||||
}
|
||||
async fn heal_object(&self, _: &str, _: &str, _: Option<&str>, _: &HealOpts) -> Result<(HealResultItem, Option<Error>)> {
|
||||
// The locked quorum read found no replica on any disk.
|
||||
Ok((HealResultItem::default(), Some(Error::Storage(EcstoreError::FileNotFound))))
|
||||
}
|
||||
}
|
||||
|
||||
async fn wait_until_idle(manager: &HealManager) {
|
||||
tokio::time::timeout(Duration::from_secs(10), async {
|
||||
while !manager.active_heals.lock().await.is_empty() {
|
||||
tokio::time::sleep(Duration::from_millis(5)).await;
|
||||
}
|
||||
})
|
||||
.await
|
||||
.expect("the absent object heal completes");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn disposition_read_repair_absent_object_is_remembered_after_one_task() {
|
||||
let manager = HealManager::new_without_root_recovery_for_test(
|
||||
Arc::new(AbsentStorage),
|
||||
Some(HealConfig {
|
||||
max_concurrent_heals: 1,
|
||||
..HealConfig::default()
|
||||
}),
|
||||
);
|
||||
let first = object_request("deleted", HealRequestSource::ReadRepair, HealPriority::Low, None);
|
||||
let first_id = first.id.clone();
|
||||
let receipt = manager
|
||||
.submit_heal_request_with_receipt(first)
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(receipt.result, HealAdmissionResult::Accepted);
|
||||
process_manager_queue_once(&manager).await;
|
||||
wait_until_idle(&manager).await;
|
||||
assert_eq!(
|
||||
manager
|
||||
.get_task_status(&first_id)
|
||||
.await
|
||||
.expect("completed status is retained"),
|
||||
HealTaskStatus::Completed,
|
||||
"an absent object is a completed heal, not a failed one"
|
||||
);
|
||||
assert_eq!(cached_kind(&manager, "deleted"), Some(HealDispositionKind::Absent));
|
||||
|
||||
// The producer's reservation expired and it resubmits: nothing is queued.
|
||||
let second = manager
|
||||
.submit_heal_request_with_receipt(object_request("deleted", HealRequestSource::ReadRepair, HealPriority::Low, None))
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(second.result, HealAdmissionResult::Dropped(HealAdmissionDropReason::PolicyDropped));
|
||||
assert_eq!(second.task_id, first_id);
|
||||
assert!(manager.heal_queue.lock().await.is_empty());
|
||||
|
||||
// A durable MRF responsibility for the same key still reaches the queue.
|
||||
let mut durable = object_request("deleted", HealRequestSource::Mrf, HealPriority::Normal, None);
|
||||
durable.expected_mrf_bucket_incarnation_id = Some(Uuid::from_u128(7));
|
||||
let durable = manager
|
||||
.submit_heal_request_with_receipt(durable)
|
||||
.await
|
||||
.expect("admission decides");
|
||||
assert_eq!(durable.result, HealAdmissionResult::Accepted);
|
||||
}
|
||||
@@ -922,35 +922,11 @@ impl ECStoreHealStorage {
|
||||
}
|
||||
}
|
||||
|
||||
fn is_transient_object_exists_message(message: &str) -> bool {
|
||||
let message = message.to_ascii_lowercase();
|
||||
|
||||
[
|
||||
"failed to acquire read lock",
|
||||
"lock acquisition failed",
|
||||
"lock acquisition timeout",
|
||||
"quorum not reached",
|
||||
"deadline has elapsed",
|
||||
"timed out",
|
||||
"network error",
|
||||
"transport error",
|
||||
"connection refused",
|
||||
]
|
||||
.iter()
|
||||
.any(|pattern| message.contains(pattern))
|
||||
}
|
||||
|
||||
/// A transient or contended existence check is skipped and retried; it is
|
||||
/// never reported as object absence or as a failure. One classifier decides
|
||||
/// (backlog#2900).
|
||||
fn is_transient_object_exists_error(err: &StorageError) -> bool {
|
||||
if err.is_quorum_error() {
|
||||
return true;
|
||||
}
|
||||
|
||||
match err {
|
||||
StorageError::Lock(lock_err) => lock_err.is_retryable() || is_transient_object_exists_message(&lock_err.to_string()),
|
||||
StorageError::Io(io_err) => is_transient_object_exists_message(&io_err.to_string()),
|
||||
StorageError::SlowDown | StorageError::OperationCanceled => true,
|
||||
_ => false,
|
||||
}
|
||||
crate::error::classify_storage_error(err).is_retryable()
|
||||
}
|
||||
|
||||
#[async_trait]
|
||||
@@ -2142,7 +2118,7 @@ mod tests {
|
||||
use super::super::StorageError;
|
||||
use super::{
|
||||
decode_disk_walk_token, decode_heal_token, encode_disk_walk_token, encode_heal_token, is_transient_object_exists_error,
|
||||
is_transient_object_exists_message, next_heal_listing_token, replacement_resume_conflict_for_other_tasks,
|
||||
next_heal_listing_token, replacement_resume_conflict_for_other_tasks,
|
||||
};
|
||||
use std::collections::HashSet;
|
||||
|
||||
@@ -2375,20 +2351,23 @@ mod tests {
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transient_object_exists_message_matches_lock_quorum_failures() {
|
||||
assert!(is_transient_object_exists_message(
|
||||
"Failed to acquire read lock: ns_loc: read lock acquisition failed on bucket/object: Quorum not reached: required 2, achieved 0"
|
||||
));
|
||||
assert!(is_transient_object_exists_message("deadline has elapsed"));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn transient_object_exists_error_matches_quorum_variants() {
|
||||
fn transient_object_exists_error_matches_quorum_lock_and_offline_variants() {
|
||||
assert!(is_transient_object_exists_error(&StorageError::ErasureReadQuorum));
|
||||
assert!(is_transient_object_exists_error(&StorageError::InsufficientReadQuorum(
|
||||
"bucket".to_string(),
|
||||
"object".to_string(),
|
||||
)));
|
||||
assert!(is_transient_object_exists_error(&StorageError::Io(std::io::Error::other(
|
||||
"Failed to acquire read lock: ns_loc: read lock acquisition failed on bucket/object: Quorum not reached: required 2, achieved 0"
|
||||
))));
|
||||
assert!(is_transient_object_exists_error(&StorageError::Io(std::io::Error::other(
|
||||
"deadline has elapsed"
|
||||
))));
|
||||
assert!(is_transient_object_exists_error(&StorageError::DiskNotFound));
|
||||
assert!(is_transient_object_exists_error(&StorageError::VolumeNotFound));
|
||||
assert!(is_transient_object_exists_error(&StorageError::RemoteClientUnavailable(
|
||||
"rebuild".to_string()
|
||||
)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
|
||||
@@ -12,8 +12,9 @@
|
||||
// See the License for the specific language governing permissions and
|
||||
// limitations under the License.
|
||||
|
||||
use crate::error::{HealErrorClass, classify_heal_error};
|
||||
use crate::heal::{
|
||||
DiskError, EcstoreError, ErasureSetHealer, HealDiskExt as _,
|
||||
DiskError, ErasureSetHealer, HealDiskExt as _,
|
||||
erasure_healer::target_outcomes_complete,
|
||||
outcome::{
|
||||
HealAbortReason, HealDeferredReason, HealExecutionOutcome, HealFailureClass, HealObjectDisposition, HealObjectIdentity,
|
||||
@@ -167,17 +168,10 @@ impl HealType {
|
||||
}
|
||||
}
|
||||
|
||||
fn is_object_level_not_found_error(err: &Error) -> bool {
|
||||
match err {
|
||||
Error::Disk(DiskError::FileNotFound | DiskError::FileVersionNotFound) => true,
|
||||
Error::Storage(EcstoreError::FileNotFound | EcstoreError::FileVersionNotFound) => true,
|
||||
Error::Other(message) => matches!(message.as_str(), "File not found" | "File version not found"),
|
||||
_ => false,
|
||||
}
|
||||
}
|
||||
|
||||
/// A trailing-slash object-dir candidate whose locked storage read found no
|
||||
/// object is a listing artifact, not a missing object.
|
||||
pub(crate) fn is_missing_object_dir_heal_result(object: &str, err: &Error) -> bool {
|
||||
object.ends_with(SLASH_SEPARATOR) && is_object_level_not_found_error(err)
|
||||
object.ends_with(SLASH_SEPARATOR) && err.is_object_absent()
|
||||
}
|
||||
|
||||
/// Sample cap for per-object failure logs during a sweep: returns true (and
|
||||
@@ -440,6 +434,8 @@ pub struct HealTask {
|
||||
replacement_execution: Arc<RwLock<Option<Arc<super::storage::ReplacementExecution>>>>,
|
||||
replacement_running: Arc<AtomicBool>,
|
||||
replacement_start_retry_count: Arc<AtomicU32>,
|
||||
/// Set when the locked storage read classified the single target as absent.
|
||||
object_absent: Arc<AtomicBool>,
|
||||
/// Task status
|
||||
pub status: Arc<RwLock<HealTaskStatus>>,
|
||||
/// Progress tracking
|
||||
@@ -511,6 +507,7 @@ impl HealTask {
|
||||
replacement_execution: Arc::new(RwLock::new(None)),
|
||||
replacement_running: Arc::new(AtomicBool::new(false)),
|
||||
replacement_start_retry_count: Arc::new(AtomicU32::new(0)),
|
||||
object_absent: Arc::new(AtomicBool::new(false)),
|
||||
status: Arc::new(RwLock::new(HealTaskStatus::Pending)),
|
||||
progress: Arc::new(RwLock::new(HealProgress::new())),
|
||||
result_items: Arc::new(RwLock::new(VecDeque::with_capacity(MAX_RETAINED_HEAL_RESULT_ITEMS))),
|
||||
@@ -605,6 +602,16 @@ impl HealTask {
|
||||
self.outcome.read().await.clone()
|
||||
}
|
||||
|
||||
/// Record that the locked storage read found the single target absent.
|
||||
pub(crate) fn mark_object_absent(&self) {
|
||||
self.object_absent.store(true, Ordering::Release);
|
||||
}
|
||||
|
||||
/// Whether this task's single target was classified absent.
|
||||
pub(crate) fn object_absent(&self) -> bool {
|
||||
self.object_absent.load(Ordering::Acquire)
|
||||
}
|
||||
|
||||
fn outcome_identity(
|
||||
&self,
|
||||
bucket: &str,
|
||||
@@ -885,20 +892,7 @@ impl HealTask {
|
||||
}
|
||||
|
||||
fn is_transient_lock_or_timeout_error(err: &Error) -> bool {
|
||||
// Typed-first (backlog#1845): trust the lock taxonomy and timeout
|
||||
// variants before falling back to message needles for errors whose
|
||||
// typed identity was stringified upstream.
|
||||
if let Error::Storage(EcstoreError::Lock(lock_err)) = err {
|
||||
return lock_err.is_retryable() || matches!(lock_err, rustfs_lock::LockError::QuorumNotReached { .. });
|
||||
}
|
||||
if matches!(err, Error::Disk(DiskError::Timeout) | Error::Storage(EcstoreError::Timeout)) {
|
||||
return true;
|
||||
}
|
||||
let message = err.to_string().to_ascii_lowercase();
|
||||
message.contains("lock acquisition timeout")
|
||||
|| message.contains("lock acquisition failed")
|
||||
|| message.contains("timed out")
|
||||
|| message.contains("deadline has elapsed")
|
||||
matches!(classify_heal_error(err), HealErrorClass::LockContention)
|
||||
}
|
||||
|
||||
fn should_skip_data_usage_cache_heal_error(bucket: &str, object: &str, err: &Error) -> bool {
|
||||
@@ -910,32 +904,7 @@ impl HealTask {
|
||||
}
|
||||
|
||||
fn is_no_heal_required_error(err: &Error) -> bool {
|
||||
match err {
|
||||
Error::Storage(EcstoreError::NoHealRequired) | Error::Disk(DiskError::NoHealRequired) => true,
|
||||
Error::Other(message) => matches!(message.as_str(), "No heal required" | "No healing is required"),
|
||||
_ => matches!(err.to_string().as_str(), "No heal required" | "No healing is required"),
|
||||
}
|
||||
}
|
||||
|
||||
fn is_object_not_found_heal_error(err: &Error) -> bool {
|
||||
match err {
|
||||
Error::Disk(DiskError::FileNotFound | DiskError::FileVersionNotFound) => true,
|
||||
Error::Storage(
|
||||
EcstoreError::FileNotFound
|
||||
| EcstoreError::FileVersionNotFound
|
||||
| EcstoreError::ObjectNotFound(_, _)
|
||||
| EcstoreError::VersionNotFound(_, _, _),
|
||||
) => true,
|
||||
Error::Other(message) => {
|
||||
message.contains("File not found")
|
||||
|| message.contains("file not found")
|
||||
|| message.contains("File version not found")
|
||||
|| message.contains("file version not found")
|
||||
|| message.contains("Object not found")
|
||||
|| message.contains("object not found")
|
||||
}
|
||||
_ => false,
|
||||
}
|
||||
matches!(classify_heal_error(err), HealErrorClass::NoHealRequired)
|
||||
}
|
||||
|
||||
pub(super) fn bucket_object_retry_delay(task_id: &str, retry_attempt: u32) -> Duration {
|
||||
@@ -1022,30 +991,6 @@ impl HealTask {
|
||||
true
|
||||
}
|
||||
|
||||
async fn skip_scanner_synthetic_object_dir_missing(&self, bucket: &str, object: &str, err: &Error) -> bool {
|
||||
if self.source != HealRequestSource::Scanner || !is_missing_object_dir_heal_result(object, err) {
|
||||
return false;
|
||||
}
|
||||
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
source = self.source.as_str(),
|
||||
result = "synthetic_object_dir_missing",
|
||||
error = %err,
|
||||
"Heal recreate skipped scanner synthetic object-dir candidate after object-level not-found"
|
||||
);
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.set_current_object(Some(format!("skipped: {bucket}/{object}")));
|
||||
progress.update_stage(4, 4);
|
||||
true
|
||||
}
|
||||
|
||||
#[tracing::instrument(skip(self), fields(task_id = %self.id, heal_type = ?self.heal_type))]
|
||||
#[hotpath::measure]
|
||||
pub async fn execute(&self) -> Result<()> {
|
||||
|
||||
@@ -11,7 +11,7 @@
|
||||
// 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.
|
||||
/// object-level heal: metadata dir canonicalization and missing-object recreation
|
||||
/// object-level heal: metadata dir canonicalization and the storage-decided absent verdict
|
||||
use super::*;
|
||||
|
||||
impl HealTask {
|
||||
@@ -179,36 +179,22 @@ impl HealTask {
|
||||
recreate_missing = self.options.recreate_missing,
|
||||
"Heal target object is missing"
|
||||
});
|
||||
if self.options.recreate_missing {
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_STAGE,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
stage = "recreate_missing",
|
||||
"Heal object recreate requested"
|
||||
);
|
||||
return self.recreate_missing_object(bucket, object, version_id).await;
|
||||
} else if self.source == HealRequestSource::Scanner {
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_STAGE,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
stage = "scanner_missing_probe",
|
||||
"Heal scanner missing object will be checked by storage layer"
|
||||
);
|
||||
} else {
|
||||
return Err(Error::TaskExecutionFailed {
|
||||
message: format!("Object not found: {bucket}/{object}"),
|
||||
});
|
||||
}
|
||||
// The unlocked existence check is only a hint. The locked storage
|
||||
// heal below reads every disk; its object-level not-found is the
|
||||
// absence verdict for every source and every `recreate_missing`
|
||||
// value. A recreate from parity happens only on the durable MRF
|
||||
// path above, which carries the write anchor.
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_STAGE,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
stage = "missing_probe",
|
||||
"Heal missing object will be checked by storage layer"
|
||||
);
|
||||
}
|
||||
|
||||
{
|
||||
@@ -267,23 +253,8 @@ impl HealTask {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if Self::is_object_not_found_heal_error(&e) {
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
result = "treated_as_deleted",
|
||||
"Heal missing object treated as deleted"
|
||||
);
|
||||
{
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.update_stage(3, 3);
|
||||
}
|
||||
return Ok(());
|
||||
if e.is_object_absent() {
|
||||
return self.complete_absent_object(bucket, object, &e).await;
|
||||
}
|
||||
|
||||
error!(
|
||||
@@ -373,23 +344,8 @@ impl HealTask {
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
if Self::is_object_not_found_heal_error(&e) {
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
result = "treated_as_deleted",
|
||||
"Heal missing object treated as deleted"
|
||||
);
|
||||
{
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.update_stage(3, 3);
|
||||
}
|
||||
return Ok(());
|
||||
if e.is_object_absent() {
|
||||
return self.complete_absent_object(bucket, object, &e).await;
|
||||
}
|
||||
|
||||
error!(
|
||||
@@ -455,133 +411,11 @@ impl HealTask {
|
||||
}
|
||||
}
|
||||
|
||||
/// Recreate missing object (for EC decode scenarios)
|
||||
async fn recreate_missing_object(&self, bucket: &str, object: &str, version_id: Option<&str>) -> Result<()> {
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_STAGE,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
version_id = ?version_id,
|
||||
stage = "recreate_missing",
|
||||
"Heal object recreate started"
|
||||
);
|
||||
|
||||
// Use ecstore's heal_object with recreate option
|
||||
let heal_opts = HealOpts {
|
||||
recursive: false,
|
||||
dry_run: self.options.dry_run,
|
||||
remove: false,
|
||||
recreate: true,
|
||||
scan_mode: HealScanMode::Deep,
|
||||
update_parity: true,
|
||||
no_lock: self.options.no_lock,
|
||||
read_repair: false,
|
||||
pool: None,
|
||||
set: None,
|
||||
};
|
||||
|
||||
match self
|
||||
.await_with_control(self.storage.heal_object(bucket, object, version_id, &heal_opts))
|
||||
.await
|
||||
{
|
||||
Ok((result, error)) => {
|
||||
if let Some(e) = error {
|
||||
if self.skip_missing_usage_observation(bucket, object, version_id, &e).await
|
||||
|| self.skip_scanner_synthetic_object_dir_missing(bucket, object, &e).await
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
error!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
result = "recreate_failed",
|
||||
error = %e,
|
||||
"Heal object recovery failed"
|
||||
);
|
||||
return Err(Error::TaskExecutionFailed {
|
||||
message: format!("Failed to recreate missing object {bucket}/{object}: {e}"),
|
||||
});
|
||||
}
|
||||
|
||||
let object_size = result.object_size as u64;
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
object_size,
|
||||
result = "recreated",
|
||||
"Heal object recreated"
|
||||
);
|
||||
|
||||
{
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.update_object_progress(1, 1, 0, 0, object_size);
|
||||
}
|
||||
self.record_result_item(result).await;
|
||||
Ok(())
|
||||
}
|
||||
Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
|
||||
Err(Error::TaskTimeout) => Err(Error::TaskTimeout),
|
||||
Err(e) => {
|
||||
if self.skip_missing_usage_observation(bucket, object, version_id, &e).await
|
||||
|| self.skip_scanner_synthetic_object_dir_missing(bucket, object, &e).await
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
|
||||
error!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
component = LOG_COMPONENT_HEAL,
|
||||
subsystem = LOG_SUBSYSTEM_OBJECT,
|
||||
task_id = %self.id,
|
||||
bucket,
|
||||
object,
|
||||
result = "recreate_failed",
|
||||
error = %e,
|
||||
"Heal object recovery failed"
|
||||
);
|
||||
Err(Error::TaskExecutionFailed {
|
||||
message: format!("Failed to recreate missing object {bucket}/{object}: {e}"),
|
||||
})
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn skip_missing_usage_observation(&self, bucket: &str, object: &str, version_id: Option<&str>, err: &Error) -> bool {
|
||||
// Usage publication may delete an obsolete observation after a degraded
|
||||
// GET has queued read repair. Its absence is not lost user data. Keep
|
||||
// durable MRF completion on its separate, proof-bearing storage path.
|
||||
if self.source != HealRequestSource::ReadRepair
|
||||
|| version_id.is_some()
|
||||
|| bucket != RUSTFS_META_BUCKET
|
||||
|| object
|
||||
.strip_prefix(BUCKET_META_PREFIX)
|
||||
.and_then(|suffix| suffix.strip_prefix('/'))
|
||||
!= Some(rustfs_data_usage::DATA_USAGE_OBSERVED_OBJECT_NAME)
|
||||
|| !matches!(
|
||||
err,
|
||||
Error::Disk(DiskError::FileNotFound)
|
||||
| Error::Storage(EcstoreError::FileNotFound | EcstoreError::ObjectNotFound(_, _))
|
||||
)
|
||||
{
|
||||
return false;
|
||||
}
|
||||
|
||||
/// The locked storage read found no replica: there is nothing to repair
|
||||
/// and the heal completes for every source. No absence proof is recorded;
|
||||
/// only storage's incarnation-fenced receipt can carry one.
|
||||
async fn complete_absent_object(&self, bucket: &str, object: &str, err: &Error) -> Result<()> {
|
||||
self.mark_object_absent();
|
||||
debug!(
|
||||
target: "rustfs::heal::task",
|
||||
event = EVENT_HEAL_OBJECT_RESULT,
|
||||
@@ -591,14 +425,14 @@ impl HealTask {
|
||||
bucket,
|
||||
object,
|
||||
source = self.source.as_str(),
|
||||
result = "usage_observation_missing",
|
||||
"Heal skipped an absent usage observation"
|
||||
result = "treated_as_deleted",
|
||||
error = %err,
|
||||
"Heal missing object treated as deleted"
|
||||
);
|
||||
let mut progress = self.progress.write().await;
|
||||
progress.set_current_object(Some(format!("skipped: {bucket}/{object}")));
|
||||
progress.update_object_progress(1, 0, 0, 1, 0);
|
||||
progress.update_stage(4, 4);
|
||||
true
|
||||
progress.update_stage(3, 3);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Durable MRF partial writes may complete only with an incarnation-fenced storage proof.
|
||||
|
||||
@@ -14,8 +14,10 @@
|
||||
|
||||
use super::super::{DiskOption, DiskStore, Endpoint, new_disk};
|
||||
use super::*;
|
||||
use crate::heal::EcstoreError;
|
||||
use crate::heal::storage::HealStorageObjectResult;
|
||||
|
||||
mod absent_verdict;
|
||||
mod concurrent_delete;
|
||||
mod deferred_retry;
|
||||
mod usage_observation;
|
||||
@@ -2013,6 +2015,27 @@ fn test_missing_object_dir_heal_result_matches_only_object_level_not_found() {
|
||||
assert!(!is_missing_object_dir_heal_result("x.rnd", &Error::Disk(DiskError::FileNotFound)));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn missing_object_dir_heal_result_not_found_accepts_typed_object_and_version_absence() {
|
||||
assert!(is_missing_object_dir_heal_result(
|
||||
"x.rnd/",
|
||||
&Error::Storage(EcstoreError::ObjectNotFound("bucket".to_string(), "x.rnd/".to_string()))
|
||||
));
|
||||
assert!(is_missing_object_dir_heal_result(
|
||||
"x.rnd/",
|
||||
&Error::Storage(EcstoreError::VersionNotFound(
|
||||
"bucket".to_string(),
|
||||
"x.rnd/".to_string(),
|
||||
"version".to_string()
|
||||
))
|
||||
));
|
||||
assert!(!is_missing_object_dir_heal_result("x.rnd/", &Error::Storage(EcstoreError::DiskNotFound)));
|
||||
assert!(!is_missing_object_dir_heal_result(
|
||||
"x.rnd/",
|
||||
&Error::Storage(EcstoreError::VolumeNotFound)
|
||||
));
|
||||
}
|
||||
|
||||
#[async_trait::async_trait]
|
||||
impl HealStorageAPI for MockStorage {
|
||||
async fn get_object_meta(&self, _bucket: &str, _object: &str) -> Result<Option<HealObjectInfo>> {
|
||||
@@ -3910,12 +3933,13 @@ async fn test_heal_admin_missing_object_dir_does_not_canonicalize_plain_object()
|
||||
request.source = HealRequestSource::Admin;
|
||||
let task = HealTask::from_request(request, storage.clone());
|
||||
|
||||
let err = task
|
||||
.execute()
|
||||
task.execute()
|
||||
.await
|
||||
.expect_err("admin object-dir request must not be canonicalized");
|
||||
.expect("an absent admin object-dir target completes under the general not-found rule");
|
||||
|
||||
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Completed));
|
||||
// The invariant under test: an administrator request names an exact key
|
||||
// and is never rewritten to the plain object that happens to exist.
|
||||
assert_eq!(storage.heal_object_calls.lock().unwrap().as_slice(), ["x.rnd/".to_string()]);
|
||||
}
|
||||
|
||||
@@ -4083,38 +4107,6 @@ async fn test_heal_recreate_scanner_synthetic_object_dir_skips_err_not_found() {
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Completed));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_heal_recreate_scanner_non_dir_not_found_fails() {
|
||||
let storage = Arc::new(MockStorage {
|
||||
object_exists: Mutex::new(Some(false)),
|
||||
heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::ErrOther("File not found"))),
|
||||
..Default::default()
|
||||
});
|
||||
let mut request = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "bucket-a".to_string(),
|
||||
object: "x.rnd".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions {
|
||||
recreate_missing: true,
|
||||
timeout: None,
|
||||
..Default::default()
|
||||
},
|
||||
HealPriority::Normal,
|
||||
);
|
||||
request.source = HealRequestSource::Scanner;
|
||||
let task = HealTask::from_request(request, storage);
|
||||
|
||||
let err = task
|
||||
.execute()
|
||||
.await
|
||||
.expect_err("scanner non-dir missing object should still fail recreate");
|
||||
|
||||
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn mrf_recreate_missing_object_records_exact_absence_receipt_with_scope() {
|
||||
let incarnation = Uuid::new_v4();
|
||||
@@ -4271,38 +4263,6 @@ async fn test_heal_recreate_scanner_synthetic_object_dir_disk_not_found_fails()
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_heal_recreate_admin_synthetic_object_dir_not_found_fails() {
|
||||
let storage = Arc::new(MockStorage {
|
||||
object_exists: Mutex::new(Some(false)),
|
||||
heal_object_outcome: Mutex::new(Some(MockHealObjectOutcome::ErrOther("File not found"))),
|
||||
..Default::default()
|
||||
});
|
||||
let mut request = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: "bucket-a".to_string(),
|
||||
object: "x.rnd/".to_string(),
|
||||
version_id: None,
|
||||
},
|
||||
HealOptions {
|
||||
recreate_missing: true,
|
||||
timeout: None,
|
||||
..Default::default()
|
||||
},
|
||||
HealPriority::Normal,
|
||||
);
|
||||
request.source = HealRequestSource::Admin;
|
||||
let task = HealTask::from_request(request, storage);
|
||||
|
||||
let err = task
|
||||
.execute()
|
||||
.await
|
||||
.expect_err("admin object-dir not-found recreate should not be skipped");
|
||||
|
||||
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_heal_recreate_existing_trailing_slash_object_records_normal_result() {
|
||||
let storage = Arc::new(MockStorage {
|
||||
|
||||
@@ -0,0 +1,221 @@
|
||||
// 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.
|
||||
|
||||
//! The absent verdict for a heal target is decided by the storage quorum read,
|
||||
//! not by the request source or the `recreate_missing` option (backlog#2900;
|
||||
//! rustfs/rustfs#7997, #8132, #8235, #5950).
|
||||
|
||||
use super::*;
|
||||
|
||||
const OBSERVATION: &str = "buckets/.usage.observed.json";
|
||||
|
||||
fn missing_storage(outcome: MockHealObjectOutcome) -> Arc<MockStorage> {
|
||||
Arc::new(MockStorage {
|
||||
object_exists: Mutex::new(Some(false)),
|
||||
heal_object_outcome: Mutex::new(Some(outcome)),
|
||||
..Default::default()
|
||||
})
|
||||
}
|
||||
|
||||
fn object_task(
|
||||
bucket: &str,
|
||||
object: &str,
|
||||
version: Option<&str>,
|
||||
source: HealRequestSource,
|
||||
recreate_missing: bool,
|
||||
storage: Arc<MockStorage>,
|
||||
) -> HealTask {
|
||||
let mut request = HealRequest::new(
|
||||
HealType::Object {
|
||||
bucket: bucket.to_string(),
|
||||
object: object.to_string(),
|
||||
version_id: version.map(str::to_string),
|
||||
},
|
||||
HealOptions {
|
||||
recreate_missing,
|
||||
timeout: None,
|
||||
..Default::default()
|
||||
},
|
||||
HealPriority::Normal,
|
||||
);
|
||||
request.source = source;
|
||||
HealTask::from_request(request, storage)
|
||||
}
|
||||
|
||||
const NON_MRF_SOURCES: [HealRequestSource; 5] = [
|
||||
HealRequestSource::Admin,
|
||||
HealRequestSource::Internal,
|
||||
HealRequestSource::AutoHeal,
|
||||
HealRequestSource::ReadRepair,
|
||||
HealRequestSource::Scanner,
|
||||
];
|
||||
|
||||
#[tokio::test]
|
||||
async fn absent_object_completes_for_every_source_regardless_of_recreate_missing() {
|
||||
for source in NON_MRF_SOURCES {
|
||||
for recreate_missing in [false, true] {
|
||||
for outer in [false, true] {
|
||||
let storage = missing_storage(MockHealObjectOutcome::MissingObject { outer });
|
||||
let task = object_task("bucket-a", "x.rnd", None, source, recreate_missing, storage.clone());
|
||||
task.execute()
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{source:?} recreate={recreate_missing} outer={outer}: {err}"));
|
||||
assert_eq!(
|
||||
task.get_status().await,
|
||||
HealTaskStatus::Completed,
|
||||
"{source:?} recreate={recreate_missing}"
|
||||
);
|
||||
let outcome = task.get_outcome().await;
|
||||
assert_eq!(outcome.counters.failed, 0, "{source:?} recreate={recreate_missing}");
|
||||
// One locked storage probe decides the verdict; no second recreate call follows.
|
||||
assert_eq!(
|
||||
storage.heal_object_calls.lock().unwrap().as_slice(),
|
||||
["x.rnd".to_string()],
|
||||
"{source:?} recreate={recreate_missing}"
|
||||
);
|
||||
let opts = storage.object_heal_opts.lock().unwrap();
|
||||
assert_eq!(opts.len(), 1);
|
||||
assert_eq!(opts[0].recreate, recreate_missing, "the option is still parsed and forwarded");
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn absent_object_with_explicit_version_is_absent_for_read_repair() {
|
||||
for outcome in [
|
||||
MockHealObjectOutcome::MissingObject { outer: false },
|
||||
MockHealObjectOutcome::MissingVersion,
|
||||
] {
|
||||
let storage = missing_storage(outcome);
|
||||
let task = object_task(
|
||||
"bucket-a",
|
||||
"x.rnd",
|
||||
Some("version-a"),
|
||||
HealRequestSource::ReadRepair,
|
||||
true,
|
||||
storage.clone(),
|
||||
);
|
||||
task.execute()
|
||||
.await
|
||||
.expect("a missing version is absent, not recreate_failed");
|
||||
assert_eq!(task.get_status().await, HealTaskStatus::Completed);
|
||||
assert_eq!(task.get_outcome().await.counters.failed, 0);
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn absent_usage_observation_follows_the_general_not_found_rule() {
|
||||
// rustfs/rustfs#8132: previously only the exact (ReadRepair, meta bucket,
|
||||
// unversioned) tuple was excused through a path whitelist.
|
||||
for (bucket, object, source, version) in [
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::ReadRepair, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::ReadRepair, Some("version-a")),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::Admin, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::Scanner, None),
|
||||
(RUSTFS_META_BUCKET, "buckets/.usage.json", HealRequestSource::ReadRepair, None),
|
||||
("user-bucket", OBSERVATION, HealRequestSource::AutoHeal, None),
|
||||
] {
|
||||
let storage = missing_storage(MockHealObjectOutcome::MissingObject { outer: false });
|
||||
let task = object_task(bucket, object, version, source, true, storage);
|
||||
task.execute()
|
||||
.await
|
||||
.unwrap_or_else(|err| panic!("{bucket}/{object} {source:?}: {err}"));
|
||||
let outcome = task.get_outcome().await;
|
||||
assert_eq!(outcome.execution, HealExecutionOutcome::Completed, "{bucket}/{object} {source:?}");
|
||||
assert_eq!(outcome.counters.failed, 0, "{bucket}/{object} {source:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn absent_object_dir_candidate_follows_the_general_not_found_rule() {
|
||||
// rustfs/rustfs#3839 excused only scanner-sourced trailing-slash candidates.
|
||||
for source in NON_MRF_SOURCES {
|
||||
for outcome in [
|
||||
MockHealObjectOutcome::MissingObject { outer: false },
|
||||
MockHealObjectOutcome::OkWithOtherError("File not found"),
|
||||
MockHealObjectOutcome::ErrOther("File not found"),
|
||||
] {
|
||||
let storage = missing_storage(outcome);
|
||||
let task = object_task("bucket-a", "x.rnd/", None, source, true, storage);
|
||||
task.execute().await.unwrap_or_else(|err| panic!("{source:?}: {err}"));
|
||||
assert_eq!(task.get_status().await, HealTaskStatus::Completed, "{source:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn absent_legacy_display_not_found_is_absent_for_read_repair() {
|
||||
// An old peer can still render the typed error as its display text.
|
||||
for message in ["File not found", "File version not found", "Object not found: bucket-a/x.rnd"] {
|
||||
let storage = missing_storage(MockHealObjectOutcome::ErrOther(message));
|
||||
let task = object_task("bucket-a", "x.rnd", None, HealRequestSource::ReadRepair, true, storage);
|
||||
task.execute().await.unwrap_or_else(|err| panic!("{message:?}: {err}"));
|
||||
assert_eq!(task.get_status().await, HealTaskStatus::Completed, "{message:?}");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn not_found_verdict_is_not_granted_to_disk_or_quorum_failures() {
|
||||
// backlog#856 / #799 B7: an offline drive or a lost quorum is never absence.
|
||||
for (outcome, recoverable) in [
|
||||
(MockHealObjectOutcome::FaultyStorageDisk(false), true),
|
||||
(MockHealObjectOutcome::FaultyStorageDisk(true), true),
|
||||
(MockHealObjectOutcome::OkWithReadQuorum, true),
|
||||
(MockHealObjectOutcome::RetryableReadQuorum, true),
|
||||
(MockHealObjectOutcome::RetryableSlowDown, true),
|
||||
(MockHealObjectOutcome::ErrOther("Disk not found"), false),
|
||||
(MockHealObjectOutcome::PermissionDenied, false),
|
||||
] {
|
||||
for source in [
|
||||
HealRequestSource::ReadRepair,
|
||||
HealRequestSource::Admin,
|
||||
HealRequestSource::Scanner,
|
||||
] {
|
||||
let storage = missing_storage(outcome.clone());
|
||||
let task = object_task("bucket-a", "x.rnd", None, source, true, storage);
|
||||
let err = task
|
||||
.execute()
|
||||
.await
|
||||
.expect_err("non-absence failures must keep failing the attempt");
|
||||
assert_eq!(err.is_recoverable_heal(), recoverable, "{source:?}: {err}");
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }), "{source:?}");
|
||||
assert_eq!(task.get_outcome().await.counters.failed, 1, "{source:?}");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn not_found_from_lowercase_free_text_is_not_absent() {
|
||||
// Only the canonical typed displays convert back; arbitrary text stays a failure.
|
||||
let storage = missing_storage(MockHealObjectOutcome::ErrOther("file not found while listing"));
|
||||
let task = object_task("bucket-a", "x.rnd", None, HealRequestSource::ReadRepair, true, storage);
|
||||
task.execute()
|
||||
.await
|
||||
.expect_err("free text must not become an absence verdict");
|
||||
assert!(matches!(task.get_status().await, HealTaskStatus::Failed { .. }));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn recreate_missing_mrf_durable_repair_still_requires_storage_proof() {
|
||||
// The durable MRF path is unchanged: absence without an exact receipt
|
||||
// cannot retire the responsibility.
|
||||
let storage = missing_storage(MockHealObjectOutcome::MissingObject { outer: false });
|
||||
*storage.bucket_incarnation_id.lock().expect("bucket incarnation") = Some(Uuid::new_v4());
|
||||
let task = object_task(RUSTFS_META_BUCKET, OBSERVATION, None, HealRequestSource::Mrf, true, storage);
|
||||
task.execute()
|
||||
.await
|
||||
.expect_err("MRF responsibility cannot retire without proof");
|
||||
assert_eq!(task.get_outcome().await.counters.failed, 1);
|
||||
}
|
||||
@@ -43,56 +43,25 @@ fn missing_storage(error: MockHealObjectOutcome) -> Arc<MockStorage> {
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_usage_observation_completes_without_claiming_repair_or_absence_proof() {
|
||||
// rustfs/rustfs#8132: the observation follows the general not-found rule
|
||||
// (see `absent_verdict`); the completion still records no proof.
|
||||
for outer in [false, true] {
|
||||
let storage = missing_storage(MockHealObjectOutcome::MissingObject { outer });
|
||||
let task = task(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::ReadRepair, None, storage.clone());
|
||||
task.execute()
|
||||
.await
|
||||
.expect("a removed observation is a normal read-repair skip");
|
||||
.expect("a removed observation is a normal read-repair completion");
|
||||
assert_eq!(task.get_status().await, HealTaskStatus::Completed);
|
||||
let outcome = task.get_outcome().await;
|
||||
assert_eq!(outcome.execution, HealExecutionOutcome::Completed);
|
||||
assert_eq!(outcome.counters.failed, 0);
|
||||
assert_eq!(outcome.counters.skipped, 1);
|
||||
assert_eq!(outcome.objects[0].disposition, HealObjectDisposition::Unknown);
|
||||
let progress = task.get_progress().await;
|
||||
assert_eq!(progress.objects_healed, 0);
|
||||
assert_eq!(progress.skipped_objects, 1);
|
||||
assert_eq!(progress.objects_failed, 0);
|
||||
let calls = storage.object_heal_opts.lock().expect("heal calls");
|
||||
assert_eq!(calls.len(), 1);
|
||||
assert!(calls[0].recreate, "the test must reach the actual recreation branch");
|
||||
}
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn missing_usage_observation_exclusion_is_limited_to_the_exact_read_repair_target() {
|
||||
for (bucket, object, source, version) in [
|
||||
("user-bucket", OBSERVATION, HealRequestSource::ReadRepair, None),
|
||||
(RUSTFS_META_BUCKET, "buckets/.usage.json", HealRequestSource::ReadRepair, None),
|
||||
(
|
||||
RUSTFS_META_BUCKET,
|
||||
"buckets/.usage.observed.json.bkp",
|
||||
HealRequestSource::ReadRepair,
|
||||
None,
|
||||
),
|
||||
(
|
||||
RUSTFS_META_BUCKET,
|
||||
"buckets/nested/.usage.observed.json",
|
||||
HealRequestSource::ReadRepair,
|
||||
None,
|
||||
),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::Admin, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::Internal, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::Scanner, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::AutoHeal, None),
|
||||
(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::ReadRepair, Some("version-a")),
|
||||
] {
|
||||
let storage = missing_storage(MockHealObjectOutcome::MissingObject { outer: false });
|
||||
let task = task(bucket, object, source, version, storage);
|
||||
task.execute()
|
||||
.await
|
||||
.expect_err("other requests must retain their missing-object failure");
|
||||
assert_eq!(task.get_outcome().await.counters.failed, 1, "{bucket}/{object}: {source:?}");
|
||||
assert!(calls[0].read_repair, "the request still reaches the locked storage read once");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -102,12 +71,13 @@ async fn usage_observation_does_not_hide_non_absence_errors() {
|
||||
MockHealObjectOutcome::PermissionDenied,
|
||||
MockHealObjectOutcome::OkWithReadQuorum,
|
||||
MockHealObjectOutcome::FaultyStorageDisk(false),
|
||||
MockHealObjectOutcome::MissingVersion,
|
||||
MockHealObjectOutcome::ErrOther("File not found"),
|
||||
MockHealObjectOutcome::ErrOther("Disk not found"),
|
||||
] {
|
||||
let storage = missing_storage(error);
|
||||
let task = task(RUSTFS_META_BUCKET, OBSERVATION, HealRequestSource::ReadRepair, None, storage);
|
||||
task.execute().await.expect_err("only a typed object absence may be skipped");
|
||||
task.execute()
|
||||
.await
|
||||
.expect_err("only an object-level absence completes the heal");
|
||||
assert_eq!(task.get_outcome().await.counters.failed, 1);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -77,7 +77,7 @@ async fn usage_observation_read_repair_survives_snapshot_cleanup() {
|
||||
.await
|
||||
.expect("the real degraded GET must emit a repair request");
|
||||
assert_eq!(request.source, HealRequestSource::ReadRepair);
|
||||
assert_eq!(request.recreate_missing, Some(true));
|
||||
assert_eq!(request.recreate_missing, None, "a degraded read does not instruct recreation");
|
||||
|
||||
// Authoritative usage publication deletes the obsolete observation with
|
||||
// these prefix options. Delay admission until that deletion has finished.
|
||||
|
||||
@@ -48,6 +48,8 @@
|
||||
- `heal-rpc-auth-v2` internode gRPC authentication: servers temporarily accept legacy prefix signatures so old peers remain available during rolling upgrades. Remove the legacy fallback after the minimum supported RustFS peer version sends v2 authentication on every internode gRPC request.
|
||||
- `put-file-auth-epoch-strict` internode put_file epoch compatibility: rc.2 peers can cache a remote put_file capability before that remote node restarts, then continue sending v1 authenticated uploads with the old server epoch; those peers cannot recover from the 409 conflict used by newer clients to trigger a re-probe. Servers temporarily accept signed, non-nil stale put_file epochs while legacy put_file auth remains non-strict so mixed-version rolling upgrades can finish multipart/object writes. Remove the stale-epoch fallback after the minimum supported RustFS peer version re-probes put_file capability after server-epoch conflicts and legacy put_file auth is no longer accepted.
|
||||
- `disk-mutation-body-digest` internode mutating disk RPCs: servers temporarily accept mutating disk RPCs (RenameData, DeleteVersion, DeleteVersions, WriteMetadata, UpdateMetadata, WriteAll, Delete, DeletePaths, RenameFile, RenamePart, DeleteVolume, MakeVolume, MakeVolumes) that carry no signature-bound canonical body digest, so peers from releases that predate body-digest signing remain available during rolling upgrades. Accepted digestless mutations increment the internode body-digest fallback counter; that counter must read zero fleet-wide across a release window before RUSTFS_INTERNODE_RPC_BODY_DIGEST_STRICT is enabled. Because body-bound requests now consume replay-cache nonces on the receiver, deploy the raised RUSTFS_INTERNODE_RPC_REPLAY_CACHE_CAPACITY default fleet-wide before enabling strict mode, and watch the internode replay-cache overflow counter for undersized capacity during the rollout. Remove the digestless fallback after the minimum supported RustFS peer version body-binds every mutating disk RPC.
|
||||
- `backlog-2900-heal-error-display` heal error display boundary: errors that reach the heal crate with their typed identity rendered as text (an older peer's display string, a local format into the TaskExecutionFailed or Other variants, a boxed io::Error) are mapped back to one HealErrorClass by classify_error_display in `crates/heal/src/error.rs`; no other heal code matches error text. Remove after every heal producer and the minimum supported peer version return typed storage errors end-to-end and the needle list has shrunk to empty.
|
||||
|
||||
- `heal-status-rpc-v1` node heal status capability: new peers treat an unimplemented BackgroundHealStatus RPC as an explicitly incomplete rolling-upgrade response. Remove the fallback after the minimum supported RustFS peer version implements BackgroundHealStatus.
|
||||
- `backlog-1316` legacy encrypted multipart range seek: the feature is on by default (RUSTFS_ENCRYPTED_RANGE_SEEK, default true) and the switch remains only as a kill switch until every server that can initiate, write, or complete multipart uploads supports the candidate-to-final marker protocol and uploadId commit lock, and pre-upgrade multipart uploads have drained. Remove the RUSTFS_ENCRYPTED_RANGE_SEEK switch after the minimum supported release does so; keep the quorum marker and malformed-layout full-read guards permanently.
|
||||
- `tonic-013-status-render` peer RPC failure classification: internode failures that reach a node only as text (a peer's error_info payload, a status flattened through format!) are classified by matching the rendering of an Unavailable gRPC status. Releases up to 1.0.0-alpha.38 shipped tonic 0.13, which rendered that status as "status: Unavailable, message: ..."; tonic 0.14 renders it as "code: 'The service is currently unavailable', message: ...". Both forms are matched so an older peer's relayed text still marks an unreachable peer offline. Remove the tonic 0.13 form after the minimum supported RustFS peer version ships tonic 0.14 or later.
|
||||
|
||||
@@ -72,6 +72,15 @@ No cluster-wide coordinator or second generation token exists for scanner/heal a
|
||||
|
||||
Rules: cancellation or a local lease alone is not a fence; if a fixture ever demonstrates a stale destructive write, the fix extends the storage-owned generation/admission primitive and validates the token at the final commit rather than adding a coordinator.
|
||||
|
||||
### Per-object disposition memory
|
||||
|
||||
A heal task that ends in a terminal verdict is forgotten by the queue as soon as its dedup key is released, so best-effort producers (scanner, read repair, autoheal, MRF hints) resubmit the same object on their own cadence. `HealDispositionCache` in `crates/heal/src/heal/manager/disposition.rs` remembers the last terminal verdict per `(bucket, object, version, pool, set)` and is consulted once, in `submit_heal_request_with_receipt_alias_and_mrf_notice` before `admit_request_to_queue`, for `HealType::Object` requests that are not administrator starts, not forced, not above normal priority, not dry runs, and carry no durable MRF anchor. A verdict applies only to the same pool/set selector; absence in an old pool must not suppress repair in another pool or an unscoped request. The manager retry path never consults it.
|
||||
|
||||
- The verdict comes from `classify_heal_error` in `crates/heal/src/error.rs`, the one classifier shared by the task, the erasure-set healer and the storage adapter. An object-level not-found from the locked storage read is `Absent` for every source and every `recreate_missing` value; an offline disk, a missing volume, a lost quorum, or a lock timeout is never `Absent`.
|
||||
- `Absent` entries drop resubmissions until the entry expires. `RetryExhausted` and `Permanent` entries re-admit on an exponential schedule (one minute, doubling, capped at one hour), so a failing object is still retried but not on every producer tick. A dropped request receives `Dropped(PolicyDropped)` and the task id that produced the verdict, so the outcome stays observable.
|
||||
- The memory is bounded (least recently used eviction) and expiring (24 hours); it is process-local and starts empty after a restart. It is not a receipt: it never records or releases a durable responsibility, and durable MRF repairs and `AuthoritativelyAbsent` proofs keep their own storage-owned path.
|
||||
- Metrics: `rustfs_heal_disposition_cache_hits_total` (by disposition) and `rustfs_heal_disposition_cache_entries`.
|
||||
|
||||
## Behavior Change Gates
|
||||
|
||||
Any later placement or repair PR must use the following gates:
|
||||
|
||||
Reference in New Issue
Block a user