From a08de9229b69115dc1c4142d04480ec33cded2d7 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 18 Aug 2026 12:43:27 +0800 Subject: [PATCH] feat(heal): wire MRF intents with durable repair journal (HS-01) (#6189) * feat(common): add MRF intent channel and Mrf request source (HS-01) Introduce the producer-facing half of the mission repair feed: a global bounded (8192) channel carrying lightweight MrfIntent values from IO error paths, plus the RUSTFS_HEAL_MRF_ENABLE delivery kill-switch and config constants for queue/journal sizing. Delivery is strictly non-blocking (try_send, drop-on-full) so it can sit on decode-failure and partial-write paths without adding latency. HealRequestSource grows a 'mrf' variant so admission accounting can attribute replayed intents. Part of backlog#1865 (option a: wire HealEvent-style intents with a durable retry ledger). Co-Authored-By: heihutu * feat(heal): add MRF queue, durable journal, and intent consumer (HS-01) Consumer half of the mission repair feed: a bounded pending queue (100k intents / 8 MiB dual ceiling, drop-newest on overflow), a durable journal at buckets/.heal/mrf/journal.bin holding the unaccepted pending snapshot, and a consumer task that batches intents off the global channel, translates them into prioritized heal requests (decode failure -> Urgent ECDecode, metadata corruption -> High Metadata, partial write -> Normal object heal), and retries full admissions with a 5s backoff and a 3-attempt ceiling. Durability: every journal record carries its own CRC32 and a format/version header, so a torn tail truncates cleanly at replay; the journal is deleted after a successful replay and when the pending set drains (mirroring MinIO's post-replay list.bin unlink). Losing the last 500 ms flush window is acceptable: replayed duplicates merge via the manager dedup key and read-repair remains the safety net. Metrics: rustfs_heal_mrf_queue_depth/_queue_bytes, _dropped_total {reason}, _replayed_total, _journal_bytes, _journal_fsync_total. The consumer is wired at heal runtime bootstrap right after manager start, honoring RUSTFS_HEAL_MRF_ENABLE (default on, rollback = off). Tests: unit tests for the dual ceiling, record roundtrip, torn-tail truncation, and the priority mapping; integration tests against a real 4-disk ECStore proving channel intents reach the manager queue as Urgent/mrf-attributed requests and journal replay arms intents, drops torn tails, and removes the file. Part of backlog#1865 (option a). Co-Authored-By: heihutu * feat(ecstore,scanner): deliver MRF intents from error paths (HS-01) Wire the three production delivery points, each a single non-blocking try_send next to the existing in-memory heal paths, which stay as the fast path: - read.rs decode-error branch: DecodeFailure intent beside the existing read-repair submit, so an Urgent ECDecode request survives restarts even when the Low-priority read-repair request was dropped or lost. - add_partial: PartialWrite intent, giving partial-write recovery a durable Normal-priority object heal across restarts. - scanner_folder metadata-corruption classification: MetadataCorruption intent beside the existing High-priority scanner heal request. All three are on error paths only: zero cost on healthy IO. Part of backlog#1865 (option a). Co-Authored-By: heihutu * fix: include mrf heal source counts Co-Authored-By: heihutu * fix: keep node heal status wire compatibility Co-Authored-By: heihutu --------- Co-authored-by: heihutu --- crates/common/src/heal_channel.rs | 4 + crates/common/src/lib.rs | 1 + crates/common/src/mrf_channel.rs | 203 ++++++ crates/config/src/constants/heal.rs | 28 + crates/ecstore/src/set_disk/ops/object.rs | 8 + crates/ecstore/src/set_disk/read.rs | 9 + crates/heal/Cargo.toml | 2 + crates/heal/src/heal/channel.rs | 3 +- crates/heal/src/heal/manager.rs | 3 + crates/heal/src/heal/mod.rs | 1 + crates/heal/src/heal/mrf_queue.rs | 682 ++++++++++++++++++++ crates/heal/src/lib.rs | 4 + crates/heal/tests/mrf_pipeline_test.rs | 189 ++++++ crates/scanner/src/scanner_folder.rs | 9 + rustfs/src/admin/handlers/heal.rs | 2 + rustfs/src/storage/rpc/node_service/heal.rs | 1 + 16 files changed, 1148 insertions(+), 1 deletion(-) create mode 100644 crates/common/src/mrf_channel.rs create mode 100644 crates/heal/src/heal/mrf_queue.rs create mode 100644 crates/heal/tests/mrf_pipeline_test.rs diff --git a/crates/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index f0ea0530e..b5a907e4e 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -287,6 +287,9 @@ pub enum HealRequestSource { Scanner, AutoHeal, ReadRepair, + /// Mission Repair Feed: intents delivered by error paths and replayed + /// from the durable MRF journal. + Mrf, } impl HealRequestSource { @@ -297,6 +300,7 @@ impl HealRequestSource { Self::Scanner => "scanner", Self::AutoHeal => "auto_heal", Self::ReadRepair => "read_repair", + Self::Mrf => "mrf", } } } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 1ae200f24..68f5f5c68 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -17,6 +17,7 @@ pub mod globals; pub mod heal_channel; pub mod last_minute; pub mod metrics; +pub mod mrf_channel; mod readiness; pub mod table_catalog; pub mod trace_bus; diff --git a/crates/common/src/mrf_channel.rs b/crates/common/src/mrf_channel.rs new file mode 100644 index 000000000..f0a91a238 --- /dev/null +++ b/crates/common/src/mrf_channel.rs @@ -0,0 +1,203 @@ +// Copyright 2024 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. + +//! Mission Repair Feed (MRF) intent channel. +//! +//! Producers on error paths (read decode failure, scanner metadata +//! corruption, partial-write recovery) hand a lightweight [`MrfIntent`] to the +//! heal crate through a global bounded channel. Delivery is strictly +//! non-blocking: `try_send_mrf_intent` never awaits and drops the intent +//! (counting it) when the channel is full or uninitialized — losing one heal +//! hint is always preferred over stalling an IO path. Durable replay of +//! unconsumed intents is the consumer's job (see `rustfs-heal` +//! `heal::mrf_queue`), mirroring MinIO's `.heal/mrf/list.bin`. + +use std::sync::{ + Arc, OnceLock, + atomic::{AtomicBool, Ordering}, +}; +use tokio::sync::mpsc; +use uuid::Uuid; + +/// Bounded capacity of the global MRF channel. Backpressure is resolved by +/// dropping (and counting) intents, never by blocking the producer. +const MRF_CHANNEL_CAPACITY: usize = 8192; + +/// Why an intent was produced. Drives the heal priority mapping on the +/// consumer side (DecodeFailure -> Urgent, MetadataCorruption -> High, +/// PartialWrite -> Normal). +#[derive(Clone, Copy, Debug, PartialEq, Eq)] +pub enum MrfKind { + /// Erasure decode failed while serving a read (read path). + DecodeFailure, + /// Scanner classified object metadata as corrupt. + MetadataCorruption, + /// A write left the object with fewer committed shards than the set size. + PartialWrite, +} + +impl MrfKind { + pub const fn as_str(self) -> &'static str { + match self { + MrfKind::DecodeFailure => "decode-failure", + MrfKind::MetadataCorruption => "metadata-corruption", + MrfKind::PartialWrite => "partial-write", + } + } +} + +/// One repair intent. Kept deliberately small so the in-memory queue and the +/// journal stay bounded; `bucket`/`object` are `Arc` so re-arming an +/// intent never re-allocates the strings. +#[derive(Clone, Debug)] +pub struct MrfIntent { + pub bucket: Arc, + pub object: Arc, + /// Version the intent targets, as raw UUID bytes. + pub version_id: Option<[u8; 16]>, + pub kind: MrfKind, + pub enqueued_at_ms: u64, + /// Times this intent has already been offered to the heal manager. + /// Dropped by the consumer once it reaches `MRF_MAX_ATTEMPTS`. + pub attempts: u8, +} + +/// Consumer-side retry ceiling before an intent is given up on. +pub const MRF_MAX_ATTEMPTS: u8 = 3; + +impl MrfIntent { + /// Rough in-memory footprint used by the queue's byte budget. + pub fn estimated_bytes(&self) -> usize { + // Struct + strings + version bytes; buckets and objects are usually + // far below this bound, so rounding up keeps the budget conservative. + 64 + self.bucket.len() + self.object.len() + } +} + +static GLOBAL_MRF_SENDER: OnceLock> = OnceLock::new(); + +/// Delivery kill-switch, set from `RUSTFS_HEAL_MRF_ENABLE`. Producers check +/// this before touching the channel so the disabled path stays allocation- and +/// sync-free. +static MRF_DELIVERY_ENABLED: AtomicBool = AtomicBool::new(true); + +/// Override delivery (used at heal-runtime startup from configuration). +pub fn set_mrf_delivery_enabled(enabled: bool) { + MRF_DELIVERY_ENABLED.store(enabled, Ordering::Relaxed); +} + +/// Whether producers currently deliver intents. +pub fn mrf_delivery_enabled() -> bool { + MRF_DELIVERY_ENABLED.load(Ordering::Relaxed) +} + +/// Create the global MRF channel and return the consumer half. Fails if the +/// channel is already initialized (the heal runtime is a singleton). +pub fn init_mrf_channel() -> Result, &'static str> { + let (sender, receiver) = mpsc::channel(MRF_CHANNEL_CAPACITY); + GLOBAL_MRF_SENDER + .set(sender) + .map_err(|_| "MRF channel sender already initialized")?; + Ok(receiver) +} + +/// Best-effort, non-blocking intent delivery from an error path. +/// +/// Returns `true` when the intent was accepted into the channel. `false` +/// means the intent was dropped (feature disabled, channel not yet +/// initialized, or channel full) — callers must not retry or await; the +/// existing read-repair / scanner heal paths remain the safety net. +/// +/// This runs on IO error paths, so it stays synchronous and cheap: one +/// bounded allocation for the two `Arc` handles plus the channel slot. +pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option) -> bool { + if !mrf_delivery_enabled() { + return false; + } + let Some(sender) = GLOBAL_MRF_SENDER.get() else { + return false; + }; + let intent = MrfIntent { + bucket: Arc::from(bucket), + object: Arc::from(object), + version_id: version_id.map(|vid| *vid.as_bytes()), + kind, + enqueued_at_ms: unix_now_ms(), + attempts: 0, + }; + sender.try_send(intent).is_ok() +} + +fn unix_now_ms() -> u64 { + // Kept trivial: the timestamp is diagnostic metadata only; wall-clock + // failure would be a bug rather than something to handle here. + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map(|d| d.as_millis() as u64) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn intents_estimate_is_conservative() { + let intent = MrfIntent { + bucket: Arc::from("bucket"), + object: Arc::from("object"), + version_id: Some([0u8; 16]), + kind: MrfKind::DecodeFailure, + enqueued_at_ms: 0, + attempts: 0, + }; + assert!(intent.estimated_bytes() >= intent.bucket.len() + intent.object.len()); + } + + #[tokio::test] + async fn try_send_delivers_and_respects_capacity() { + let mut receiver = init_mrf_channel().expect("first initialization should succeed"); + assert!(init_mrf_channel().is_err(), "double initialization must fail"); + + assert!(try_send_mrf_intent(MrfKind::DecodeFailure, "b", "o", Some(Uuid::nil()))); + let intent = receiver.recv().await.expect("intent should arrive"); + assert_eq!(intent.kind, MrfKind::DecodeFailure); + assert_eq!(intent.bucket.as_ref(), "b"); + + // Disable delivery: producers become no-ops. + set_mrf_delivery_enabled(false); + assert!(!try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None)); + set_mrf_delivery_enabled(true); + + // Fill the bounded channel past capacity: excess intents are dropped, + // never blocking. + let mut accepted = 0; + for _ in 0..(MRF_CHANNEL_CAPACITY + 64) { + if try_send_mrf_intent(MrfKind::PartialWrite, "b", "o", None) { + accepted += 1; + } + } + assert_eq!(accepted, MRF_CHANNEL_CAPACITY); + } + + #[test] + fn try_send_without_channel_is_false() { + // This test may run after the tokio test above in the same process; + // the singleton semantics make a clean "uninitialized" case hard, so + // assert the flag-off behavior only. + set_mrf_delivery_enabled(false); + assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None)); + set_mrf_delivery_enabled(true); + } +} diff --git a/crates/config/src/constants/heal.rs b/crates/config/src/constants/heal.rs index 647ca8533..b8cf3630a 100644 --- a/crates/config/src/constants/heal.rs +++ b/crates/config/src/constants/heal.rs @@ -177,3 +177,31 @@ pub const DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: usize = 80; /// Default foreground pressure recheck delay for heal scheduler, in milliseconds. pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250; + +/// Environment variable that toggles the MRF (mission repair feed) intent +/// pipeline: error paths deliver repair intents to the heal runtime, and +/// unconsumed intents are replayed from the durable journal after a restart. +pub const ENV_HEAL_MRF_ENABLE: &str = "RUSTFS_HEAL_MRF_ENABLE"; + +/// Environment variable for the MRF in-memory queue capacity (intent count). +pub const ENV_HEAL_MRF_QUEUE_SIZE: &str = "RUSTFS_HEAL_MRF_QUEUE_SIZE"; + +/// Environment variable for the MRF journal byte budget. The journal is +/// compacted once its on-disk size crosses this bound. +pub const ENV_HEAL_MRF_JOURNAL_MAX_BYTES: &str = "RUSTFS_HEAL_MRF_JOURNAL_MAX_BYTES"; + +/// Environment variable for the MRF journal replay batch size (intents per +/// replay push round). +pub const ENV_HEAL_MRF_REPLAY_BATCH: &str = "RUSTFS_HEAL_MRF_REPLAY_BATCH"; + +/// Default behavior keeps the MRF intent pipeline enabled. +pub const DEFAULT_HEAL_MRF_ENABLE: bool = true; + +/// Default MRF queue capacity (matches MinIO's 100k MRF list ceiling). +pub const DEFAULT_HEAL_MRF_QUEUE_SIZE: usize = 100_000; + +/// Default MRF journal byte budget (8 MiB), mirroring the channel payload cap. +pub const DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES: usize = 8 * 1024 * 1024; + +/// Default MRF replay batch size. +pub const DEFAULT_HEAL_MRF_REPLAY_BATCH: usize = 256; diff --git a/crates/ecstore/src/set_disk/ops/object.rs b/crates/ecstore/src/set_disk/ops/object.rs index e7498ee9d..929ac4d5c 100644 --- a/crates/ecstore/src/set_disk/ops/object.rs +++ b/crates/ecstore/src/set_disk/ops/object.rs @@ -5845,6 +5845,14 @@ impl crate::storage_api_contracts::object::ObjectOperations for SetDisks { #[tracing::instrument(skip(self))] async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { + // MRF journal intent: partial-write recovery must survive a restart + // (HS-01); the heal request below remains the in-memory fast path. + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::PartialWrite, + bucket, + object, + uuid::Uuid::try_parse(version_id).ok(), + ); let mut request = rustfs_common::heal_channel::create_heal_request_with_options( bucket.to_string(), Some(object.to_string()), diff --git a/crates/ecstore/src/set_disk/read.rs b/crates/ecstore/src/set_disk/read.rs index d59fa3e7e..07bbcf5f9 100644 --- a/crates/ecstore/src/set_disk/read.rs +++ b/crates/ecstore/src/set_disk/read.rs @@ -1077,6 +1077,15 @@ impl SetDisks { "Recoverable decode error triggered read repair" ); let version_id = fi.version_id.as_ref().map(ToString::to_string); + // MRF journal intent: keeps a durable Urgent ECDecode + // request alive across restarts even when the in-memory + // read-repair request is dropped or lost (HS-01). + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::DecodeFailure, + bucket, + object, + fi.version_id, + ); submit_read_repair_heal( bucket, object, diff --git a/crates/heal/Cargo.toml b/crates/heal/Cargo.toml index 668c4af24..4d4fb355b 100644 --- a/crates/heal/Cargo.toml +++ b/crates/heal/Cargo.toml @@ -89,6 +89,8 @@ async-trait = { workspace = true } futures = { workspace = true } metrics = { workspace = true } base64 = { workspace = true } +bytes = { workspace = true } +crc-fast = { workspace = true } [dev-dependencies] serde_json = { workspace = true, features = ["raw_value"] } diff --git a/crates/heal/src/heal/channel.rs b/crates/heal/src/heal/channel.rs index 23cf2f168..1e31056b2 100644 --- a/crates/heal/src/heal/channel.rs +++ b/crates/heal/src/heal/channel.rs @@ -612,7 +612,8 @@ impl HealChannelProcessor { HealRequestSource::Admin | HealRequestSource::AutoHeal | HealRequestSource::Internal - | HealRequestSource::ReadRepair => true, + | HealRequestSource::ReadRepair + | HealRequestSource::Mrf => true, }); // Build HealOptions with all available fields diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index 66b8f637f..216e17068 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -270,6 +270,8 @@ pub struct HealSourceCounts { pub auto_heal: u64, pub internal: u64, pub read_repair: u64, + #[serde(default)] + pub mrf: u64, } impl HealSourceCounts { @@ -280,6 +282,7 @@ impl HealSourceCounts { HealRequestSource::AutoHeal => self.auto_heal += 1, HealRequestSource::Internal => self.internal += 1, HealRequestSource::ReadRepair => self.read_repair += 1, + HealRequestSource::Mrf => self.mrf += 1, } } } diff --git a/crates/heal/src/heal/mod.rs b/crates/heal/src/heal/mod.rs index 0881e5853..ff910ed70 100644 --- a/crates/heal/src/heal/mod.rs +++ b/crates/heal/src/heal/mod.rs @@ -16,6 +16,7 @@ pub mod channel; pub mod erasure_healer; pub mod event; pub mod manager; +pub mod mrf_queue; pub mod progress; pub(crate) mod replacement_readiness; pub mod resume; diff --git a/crates/heal/src/heal/mrf_queue.rs b/crates/heal/src/heal/mrf_queue.rs new file mode 100644 index 000000000..c1144435e --- /dev/null +++ b/crates/heal/src/heal/mrf_queue.rs @@ -0,0 +1,682 @@ +// Copyright 2024 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. + +//! Mission Repair Feed (MRF) queue, journal, and consumer. +//! +//! Intents arriving on the global channel (see `rustfs_common::mrf_channel`) +//! are buffered in a bounded in-memory queue, translated into prioritized +//! heal requests, and — while they are not yet accepted by the heal manager — +//! mirrored into a durable journal so a crash or restart can replay them. +//! This is the RustFS counterpart of MinIO's `.heal/mrf/list.bin` replay, +//! layered on top of (not replacing) read-repair and scanner heal. +//! +//! Durability model: the journal is a snapshot of the *unaccepted* pending +//! set, rewritten on a group-commit cadence (every flush interval or flush +//! threshold new intents). A rewrite is atomic at the record level only — a +//! torn tail simply truncates during replay because every record carries its +//! own CRC32. Losing the last flush window (≤500 ms) is acceptable: replayed +//! duplicates are merged by the manager's dedup key, and read-repair remains +//! the safety net. + +use super::{DiskStore, HealDiskExt as _, local_disk_map_read}; +use crate::heal::manager::HealManager; +use metrics::{counter, gauge}; +use rustfs_common::heal_channel::{HealAdmissionDropReason, HealAdmissionResult}; +use rustfs_common::mrf_channel::{MRF_MAX_ATTEMPTS, MrfIntent}; +use std::collections::VecDeque; +use std::sync::Arc; +use std::time::Duration; +use tokio::sync::mpsc; +use uuid::Uuid; + +use crate::heal::task::{HealOptions, HealPriority, HealRequest, HealType}; + +/// Journal location inside the metadata bucket, following the resume-state +/// layout. +pub(crate) const MRF_JOURNAL_PATH: &str = "buckets/.heal/mrf/journal.bin"; + +/// Record format tag. +const MRF_JOURNAL_FORMAT: u8 = 1; +/// Record layout version. +const MRF_JOURNAL_VERSION: u8 = 1; + +/// Fixed header size: format, version, kind, attempts, enqueued_at_ms, +/// has_version flag. +const MRF_RECORD_FIXED_HEAD: usize = 1 + 1 + 1 + 1 + 8 + 1; + +#[derive(Debug, Clone)] +pub(crate) struct MrfConsumerConfig { + /// In-memory queue capacity in intents. + pub queue_capacity: usize, + /// Journal byte budget; a pending snapshot above this bound is rejected + /// oldest-first so the journal can never grow unbounded. + pub journal_max_bytes: usize, + /// How many journal intents to re-arm per replay round. + pub replay_batch: usize, + /// Group-commit cadence for the journal snapshot. + pub flush_interval: Duration, + /// New intents between flushes that force an early snapshot. + pub flush_threshold: usize, + /// Backoff after the heal manager reports a full admission. + pub admission_backoff: Duration, +} + +impl Default for MrfConsumerConfig { + fn default() -> Self { + Self { + queue_capacity: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_QUEUE_SIZE, + rustfs_config::DEFAULT_HEAL_MRF_QUEUE_SIZE, + ), + journal_max_bytes: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_JOURNAL_MAX_BYTES, + rustfs_config::DEFAULT_HEAL_MRF_JOURNAL_MAX_BYTES, + ), + replay_batch: rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MRF_REPLAY_BATCH, + rustfs_config::DEFAULT_HEAL_MRF_REPLAY_BATCH, + ), + flush_interval: Duration::from_millis(500), + flush_threshold: 1000, + admission_backoff: Duration::from_secs(5), + } + } +} + +/// Bounded pending set with count and byte ceilings. Overflow drops the +/// incoming intent (never a resident one) and counts the loss. +pub(crate) struct MrfQueue { + pending: VecDeque, + bytes: usize, + capacity: usize, + byte_budget: usize, +} + +impl MrfQueue { + pub(crate) fn new(capacity: usize, byte_budget: usize) -> Self { + Self { + pending: VecDeque::new(), + bytes: 0, + capacity, + byte_budget, + } + } + + /// Returns `false` (after counting) when either ceiling would be crossed. + pub(crate) fn try_push(&mut self, intent: MrfIntent) -> bool { + let cost = intent.estimated_bytes(); + if self.pending.len() >= self.capacity || self.bytes + cost > self.byte_budget { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "queue_overflow").increment(1); + return false; + } + self.bytes += cost; + self.pending.push_back(intent); + true + } + + pub(crate) fn pop_front(&mut self) -> Option { + let intent = self.pending.pop_front()?; + self.bytes = self.bytes.saturating_sub(intent.estimated_bytes()); + Some(intent) + } + + pub(crate) fn push_back(&mut self, intent: MrfIntent) { + self.bytes += intent.estimated_bytes(); + self.pending.push_back(intent); + } + + pub(crate) fn depth(&self) -> usize { + self.pending.len() + } + + pub(crate) fn bytes(&self) -> usize { + self.bytes + } + + pub(crate) fn intents(&self) -> impl Iterator { + self.pending.iter() + } +} + +// --------------------------------------------------------------------------- +// Journal record codec +// --------------------------------------------------------------------------- + +/// Append one encoded record to `out`. +pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec) { + let start = out.len(); + out.push(MRF_JOURNAL_FORMAT); + out.push(MRF_JOURNAL_VERSION); + out.push(match intent.kind { + rustfs_common::mrf_channel::MrfKind::DecodeFailure => 1, + rustfs_common::mrf_channel::MrfKind::MetadataCorruption => 2, + rustfs_common::mrf_channel::MrfKind::PartialWrite => 3, + }); + out.push(intent.attempts); + out.extend_from_slice(&intent.enqueued_at_ms.to_le_bytes()); + match intent.version_id { + Some(bytes) => { + out.push(1); + out.extend_from_slice(&bytes); + } + None => out.push(0), + } + out.extend_from_slice(&(intent.bucket.len() as u32).to_le_bytes()); + out.extend_from_slice(&(intent.object.len() as u32).to_le_bytes()); + out.extend_from_slice(intent.bucket.as_bytes()); + out.extend_from_slice(intent.object.as_bytes()); + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&out[start..]); + out.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes()); +} + +fn decode_one(data: &[u8]) -> Option<(MrfIntent, usize)> { + if data.len() < MRF_RECORD_FIXED_HEAD + 8 { + return None; + } + if data[0] != MRF_JOURNAL_FORMAT || data[1] != MRF_JOURNAL_VERSION { + return None; + } + let kind = match data[2] { + 1 => rustfs_common::mrf_channel::MrfKind::DecodeFailure, + 2 => rustfs_common::mrf_channel::MrfKind::MetadataCorruption, + 3 => rustfs_common::mrf_channel::MrfKind::PartialWrite, + _ => return None, + }; + let attempts = data[3]; + let enqueued_at_ms = u64::from_le_bytes(data[4..12].try_into().expect("slice length checked")); + let has_version = data[12] != 0; + let mut cursor = MRF_RECORD_FIXED_HEAD; + let version_id = if has_version { + if data.len() < cursor + 16 { + return None; + } + let bytes: [u8; 16] = data[cursor..cursor + 16].try_into().expect("slice length checked"); + cursor += 16; + Some(bytes) + } else { + None + }; + if data.len() < cursor + 8 { + return None; + } + let bucket_len = u32::from_le_bytes(data[cursor..cursor + 4].try_into().expect("slice length checked")) as usize; + let object_len = u32::from_le_bytes(data[cursor + 4..cursor + 8].try_into().expect("slice length checked")) as usize; + cursor += 8; + let body_end = cursor.checked_add(bucket_len)?.checked_add(object_len)?; + let record_end = body_end.checked_add(4)?; + if data.len() < record_end { + return None; + } + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&data[..body_end]); + if (hasher.finalize() as u32) != u32::from_le_bytes(data[body_end..record_end].try_into().expect("slice length checked")) { + return None; + } + let bucket = std::sync::Arc::from(std::str::from_utf8(&data[cursor..cursor + bucket_len]).ok()?); + let object = std::sync::Arc::from(std::str::from_utf8(&data[cursor + bucket_len..body_end]).ok()?); + Some(( + MrfIntent { + bucket, + object, + version_id, + kind, + enqueued_at_ms, + attempts, + }, + record_end, + )) +} + +/// Decode a whole journal, stopping at the first torn or corrupt record. +/// Returns the decoded intents and the number of trailing bytes discarded. +pub(crate) fn decode_journal(data: &[u8]) -> (Vec, usize) { + let mut intents = Vec::new(); + let mut cursor = 0usize; + while cursor < data.len() { + match decode_one(&data[cursor..]) { + Some((intent, consumed)) => { + intents.push(intent); + cursor += consumed; + } + None => break, + } + } + let truncated = data.len() - cursor; + (intents, truncated) +} + +// --------------------------------------------------------------------------- +// Journal disk IO (all local disks, first successful read wins) +// --------------------------------------------------------------------------- + +async fn journal_disks() -> Vec { + let map = local_disk_map_read().await; + map.values().flatten().cloned().collect() +} + +async fn read_journal() -> Option> { + for disk in journal_disks().await { + match disk.read_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH).await { + Ok(bytes) => return Some(bytes.to_vec()), + Err(_) => continue, + } + } + None +} + +async fn write_journal(data: &[u8]) { + let payload = bytes::Bytes::copy_from_slice(data); + for disk in journal_disks().await { + if let Err(err) = disk + .write_all(super::RUSTFS_META_BUCKET, MRF_JOURNAL_PATH, payload.clone()) + .await + { + warn_mrf_journal_write(&err); + } + } + if !data.is_empty() { + counter!("rustfs_heal_mrf_journal_fsync_total").increment(1); + } + gauge!("rustfs_heal_mrf_journal_bytes").set(data.len() as f64); +} + +async fn delete_journal() { + for disk in journal_disks().await { + let _ = disk + .delete( + super::RUSTFS_META_BUCKET, + MRF_JOURNAL_PATH, + crate::heal::storage_api::owner::EcstoreDeleteOptions::default(), + ) + .await; + } +} + +fn warn_mrf_journal_write(err: &super::DiskError) { + tracing::warn!( + target: "rustfs::heal::mrf", + error = %err, + "MRF journal write failed; unconsumed intents may be lost on restart" + ); +} + +// --------------------------------------------------------------------------- +// Consumer +// --------------------------------------------------------------------------- + +/// Translate an intent into the prioritized heal request the issue specifies: +/// decode failures go Urgent ECDecode, metadata corruption goes High +/// Metadata, partial writes go Normal object heal. +pub(crate) fn build_heal_request(intent: &MrfIntent) -> HealRequest { + let bucket = intent.bucket.to_string(); + let object = intent.object.to_string(); + let version_id = intent.version_id.map(|bytes| Uuid::from_bytes(bytes).to_string()); + let (heal_type, priority) = match intent.kind { + rustfs_common::mrf_channel::MrfKind::DecodeFailure => ( + HealType::ECDecode { + bucket, + object, + version_id, + }, + HealPriority::Urgent, + ), + rustfs_common::mrf_channel::MrfKind::MetadataCorruption => (HealType::Metadata { bucket, object }, HealPriority::High), + rustfs_common::mrf_channel::MrfKind::PartialWrite => ( + HealType::Object { + bucket, + object, + version_id, + }, + HealPriority::Normal, + ), + }; + let mut request = HealRequest::new(heal_type, HealOptions::default(), priority); + request.source = rustfs_common::heal_channel::HealRequestSource::Mrf; + request +} + +struct MrfRuntime { + queue: MrfQueue, + config: MrfConsumerConfig, + new_since_flush: usize, + /// True while a journal snapshot exists on disk that no longer reflects + /// an all-consumed pending set; the next idle tick removes it (MinIO + /// deletes its `list.bin` after replay for the same reason). + journal_on_disk: bool, + /// Earliest instant a full-admission retry may proceed. + backoff_until: Option, +} + +impl MrfRuntime { + fn record_accept(&mut self) { + // Accepted intents leave the pending set; the next flush persists the + // smaller snapshot, which is the journal's compaction. + } + + fn snapshot(&self) -> Vec { + let mut buf = Vec::new(); + for intent in self.queue.intents() { + encode_intent(intent, &mut buf); + } + buf + } + + async fn flush(&mut self) { + write_journal(&self.snapshot()).await; + self.new_since_flush = 0; + self.journal_on_disk = true; + } + + /// Drain pending intents into the heal manager until it is full, the + /// queue empties, or attempts are exhausted. + async fn dispatch(&mut self, manager: &HealManager) { + if let Some(until) = self.backoff_until { + if tokio::time::Instant::now() < until { + return; + } + self.backoff_until = None; + } + while let Some(mut intent) = self.queue.pop_front() { + let request = build_heal_request(&intent); + match manager.submit_heal_request(request).await { + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => self.record_accept(), + Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts >= MRF_MAX_ATTEMPTS { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + continue; + } + self.queue.push_back(intent); + self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff); + break; + } + Ok(HealAdmissionResult::Dropped(_)) => { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "admission_policy").increment(1); + } + Err(_) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts >= MRF_MAX_ATTEMPTS { + counter!("rustfs_heal_mrf_dropped_total", "reason" => "attempts_exhausted").increment(1); + continue; + } + self.queue.push_back(intent); + self.backoff_until = Some(tokio::time::Instant::now() + self.config.admission_backoff); + break; + } + } + } + gauge!("rustfs_heal_mrf_queue_depth").set(self.queue.depth() as f64); + gauge!("rustfs_heal_mrf_queue_bytes").set(self.queue.bytes() as f64); + } +} + +/// Initialize the global MRF channel (honoring `RUSTFS_HEAL_MRF_ENABLE`) and +/// spawn the consumer task. Called once from the heal runtime bootstrap right +/// after the manager started; a disabled feature or a double call is a no-op. +/// Public for integration tests that drive the real consumer loop. +pub fn spawn_mrf_consumer(manager: Arc) { + let enabled = rustfs_utils::get_env_bool(rustfs_config::ENV_HEAL_MRF_ENABLE, rustfs_config::DEFAULT_HEAL_MRF_ENABLE); + rustfs_common::mrf_channel::set_mrf_delivery_enabled(enabled); + if !enabled { + tracing::info!( + target: "rustfs::heal::mrf", + "MRF intent pipeline disabled by configuration; producers will not deliver" + ); + return; + } + let receiver = match rustfs_common::mrf_channel::init_mrf_channel() { + Ok(receiver) => receiver, + Err(err) => { + tracing::warn!( + target: "rustfs::heal::mrf", + error = err, + "MRF channel initialization failed; intents will be dropped at producers" + ); + return; + } + }; + tokio::spawn(async move { + run_mrf_consumer(manager, receiver).await; + }); + tracing::info!(target: "rustfs::heal::mrf", "MRF intent consumer started"); +} + +/// Replay the durable journal into a fresh pending queue and submit whatever +/// it armed. Returns the number of intact intents replayed. Duplicates are +/// merged by the manager's dedup key; the journal file is removed once read +/// (torn tails truncate via the per-record CRC). Public for integration tests; +/// the live consumer invokes this through [`replay_into`] at startup. +pub async fn replay_journal_once(manager: &Arc) -> usize { + let config = MrfConsumerConfig::default(); + let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes); + let mut backoff_until: Option = None; + replay_into(manager, &mut queue, &mut backoff_until).await +} + +/// Shared replay core: read + decode + re-arm + delete, then drain what fits. +async fn replay_into( + manager: &Arc, + queue: &mut MrfQueue, + backoff_until: &mut Option, +) -> usize { + let Some(data) = read_journal().await else { + return 0; + }; + let (intents, truncated) = decode_journal(&data); + if truncated > 0 { + tracing::warn!( + target: "rustfs::heal::mrf", + truncated_bytes = truncated, + "MRF journal had a torn tail; truncated records were discarded" + ); + } + counter!("rustfs_heal_mrf_replayed_total").increment(intents.len() as u64); + let replayed = intents.len(); + for intent in intents { + queue.try_push(intent); + } + delete_journal().await; + + // Drain the replayed intents immediately; whatever the manager refuses + // stays armed in `queue` for the consumer's retry loop. + if backoff_until.is_none() { + while let Some(mut intent) = queue.pop_front() { + let request = build_heal_request(&intent); + match manager.submit_heal_request(request).await { + Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {} + Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => { + intent.attempts = intent.attempts.saturating_add(1); + if intent.attempts < MRF_MAX_ATTEMPTS { + queue.push_back(intent); + *backoff_until = Some(tokio::time::Instant::now()); + } + break; + } + Ok(HealAdmissionResult::Dropped(_)) | Err(_) => {} + } + } + } + replayed +} + +/// Replay the journal, then keep draining the channel into the heal manager +/// while persisting the pending snapshot. +async fn run_mrf_consumer(manager: Arc, mut receiver: mpsc::Receiver) { + let config = MrfConsumerConfig::default(); + let mut runtime = MrfRuntime { + queue: MrfQueue::new(config.queue_capacity, config.journal_max_bytes), + config: config.clone(), + new_since_flush: 0, + journal_on_disk: false, + backoff_until: None, + }; + + // Replay: read the journal, re-arm intents (duplicates are merged by the + // manager's dedup key), then drop the file so the next flush starts clean. + replay_into(&manager, &mut runtime.queue, &mut runtime.backoff_until).await; + + let mut flush_tick = tokio::time::interval(runtime.config.flush_interval); + flush_tick.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + let mut batch: Vec = Vec::with_capacity(runtime.config.replay_batch); + + loop { + tokio::select! { + received = receiver.recv_many(&mut batch, runtime.config.replay_batch) => { + if received == 0 { + // Channel closed: flush once more and stop. + runtime.flush().await; + tracing::info!( + target: "rustfs::heal::mrf", + "MRF channel closed; consumer stopped after final flush" + ); + return; + } + for intent in batch.drain(..) { + runtime.queue.try_push(intent); + runtime.new_since_flush += 1; + } + runtime.dispatch(manager.as_ref()).await; + if runtime.new_since_flush >= runtime.config.flush_threshold { + runtime.flush().await; + } + } + _ = flush_tick.tick() => { + if runtime.new_since_flush > 0 || runtime.queue.depth() > 0 { + runtime.flush().await; + runtime.dispatch(manager.as_ref()).await; + } else if runtime.journal_on_disk { + // All intents consumed: remove the journal so a restart + // replays nothing (mirrors MinIO's post-replay unlink). + delete_journal().await; + runtime.journal_on_disk = false; + gauge!("rustfs_heal_mrf_journal_bytes").set(0.0); + } + gauge!("rustfs_heal_mrf_queue_depth").set(runtime.queue.depth() as f64); + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_common::mrf_channel::{MrfIntent, MrfKind}; + use std::sync::Arc as StdArc; + + fn intent(bucket: &str, object: &str, attempts: u8) -> MrfIntent { + MrfIntent { + bucket: StdArc::from(bucket), + object: StdArc::from(object), + version_id: Some([7u8; 16]), + kind: MrfKind::DecodeFailure, + enqueued_at_ms: 1_700_000_000_000, + attempts, + } + } + + #[test] + fn queue_enforces_count_and_byte_ceilings() { + let mut queue = MrfQueue::new(2, usize::MAX); + assert!(queue.try_push(intent("b", "o", 0))); + assert!(queue.try_push(intent("b", "o", 0))); + assert!(!queue.try_push(intent("b", "o", 0)), "count ceiling must drop"); + + let mut tiny = MrfQueue::new(usize::MAX, intent("bucket", "object", 0).estimated_bytes()); + assert!(tiny.try_push(intent("bucket", "object", 0))); + assert!( + !tiny.try_push(intent("bucket", "object", 0)), + "byte budget must drop before the second intent fits" + ); + } + + #[test] + fn journal_roundtrip_preserves_intents() { + let intents = vec![ + intent("bucket-a", "object/a", 0), + intent("bucket-b", "object/b", 2), + MrfIntent { + bucket: StdArc::from("bucket-c"), + object: StdArc::from("object/c"), + version_id: None, + kind: MrfKind::MetadataCorruption, + enqueued_at_ms: 5, + attempts: 1, + }, + ]; + let mut buf = Vec::new(); + for intent in &intents { + encode_intent(intent, &mut buf); + } + let (decoded, truncated) = decode_journal(&buf); + assert_eq!(truncated, 0); + assert_eq!(decoded.len(), intents.len()); + for (left, right) in decoded.iter().zip(intents.iter()) { + assert_eq!(left.bucket, right.bucket); + assert_eq!(left.object, right.object); + assert_eq!(left.version_id, right.version_id); + assert_eq!(left.kind, right.kind); + assert_eq!(left.attempts, right.attempts); + } + } + + #[test] + fn journal_torn_tail_is_truncated() { + let mut buf = Vec::new(); + encode_intent(&intent("b", "o", 0), &mut buf); + let mut torn = buf.clone(); + torn.extend_from_slice(&buf[..buf.len() / 2]); + + let (decoded, truncated) = decode_journal(&torn); + assert_eq!(decoded.len(), 1, "the intact record must survive"); + assert!(truncated > 0, "the partial tail must be discarded"); + + // A corrupted body (CRC mismatch) also truncates from that record on. + let mut corrupt = buf.clone(); + let mid = MRF_RECORD_FIXED_HEAD + 4; + corrupt[mid] ^= 0xff; + let (decoded, truncated) = decode_journal(&corrupt); + assert!(decoded.is_empty()); + assert_eq!(truncated, corrupt.len()); + } + + #[test] + fn heal_request_mapping_follows_priority_matrix() { + let decode = build_heal_request(&intent("b", "o", 0)); + assert!(matches!(decode.heal_type, HealType::ECDecode { .. })); + assert_eq!(decode.priority, HealPriority::Urgent); + + let metadata = build_heal_request(&MrfIntent { + bucket: StdArc::from("b"), + object: StdArc::from("o"), + version_id: None, + kind: MrfKind::MetadataCorruption, + enqueued_at_ms: 0, + attempts: 0, + }); + assert!(matches!(metadata.heal_type, HealType::Metadata { .. })); + assert_eq!(metadata.priority, HealPriority::High); + + let partial = build_heal_request(&MrfIntent { + bucket: StdArc::from("b"), + object: StdArc::from("o"), + version_id: None, + kind: MrfKind::PartialWrite, + enqueued_at_ms: 0, + attempts: 0, + }); + assert!(matches!(partial.heal_type, HealType::Object { .. })); + assert_eq!(partial.priority, HealPriority::Normal); + } +} diff --git a/crates/heal/src/lib.rs b/crates/heal/src/lib.rs index 3dd29b064..f1ec4cebe 100644 --- a/crates/heal/src/lib.rs +++ b/crates/heal/src/lib.rs @@ -158,6 +158,10 @@ pub async fn init_heal_manager_with_workload_provider( return Err(err); } + // Start the MRF intent consumer (error-path repair intents + durable + // journal replay) now that the manager can accept submissions. + heal::mrf_queue::spawn_mrf_consumer(heal_manager.clone()); + #[cfg(test)] test_hook_after_manager_start().await; diff --git a/crates/heal/tests/mrf_pipeline_test.rs b/crates/heal/tests/mrf_pipeline_test.rs new file mode 100644 index 000000000..36f07a5a7 --- /dev/null +++ b/crates/heal/tests/mrf_pipeline_test.rs @@ -0,0 +1,189 @@ +// Copyright 2024 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. + +//! HS-01 (rustfs/backlog#1865): MRF intent pipeline integration tests. +//! +//! Drives the real consumer loop (`spawn_mrf_consumer`) against a real +//! 4-disk `ECStore` heal storage and a `HealManager` that has not started its +//! scheduler, so submitted intents stay observable in the admission queue. +//! Under `cargo nextest` each test runs in its own process, which keeps the +//! process-global MRF channel singleton safe. + +use rustfs_common::mrf_channel::{self, MrfKind}; +use rustfs_heal::heal::{ + manager::{HealConfig, HealManager}, + mrf_queue, + storage::{ECStoreHealStorage, HealStorageAPI}, +}; +use serial_test::serial; +use std::{path::Path, sync::Arc, time::Duration}; + +mod storage_api; + +use storage_api::endpoint_index::{Endpoint, EndpointServerPools, Endpoints, PoolEndpoints, init_local_disks}; + +const META_BUCKET: &str = ".rustfs.sys"; +const JOURNAL_REL: &str = "buckets/.heal/mrf/journal.bin"; + +async fn heal_env() -> (Vec, Arc) { + let env = rustfs_test_utils::TestECStoreEnv::builder() + .prefix("rustfs_heal_mrf_test") + .build() + .await; + let heal_storage: Arc = Arc::new(ECStoreHealStorage::new(env.ecstore.clone())); + (env.disk_paths, heal_storage) +} + +fn make_manager(storage: Arc) -> Arc { + Arc::new(HealManager::new( + storage, + Some(HealConfig { + // Keep the scheduler from draining the queue before assertions. + heal_interval: Duration::from_secs(3600), + enable_auto_heal: false, + ..Default::default() + }), + )) +} + +/// Encode one journal record independently of the implementation, so a format +/// drift between writer and this fixture fails loudly here. +fn journal_record(kind: u8, bucket: &str, object: &str, version: Option<[u8; 16]>, attempts: u8) -> Vec { + let mut body = vec![1u8, 1, kind, attempts]; + body.extend_from_slice(&1_700_000_000_000u64.to_le_bytes()); + match version { + Some(bytes) => { + body.push(1); + body.extend_from_slice(&bytes); + } + None => body.push(0), + } + body.extend_from_slice(&(bucket.len() as u32).to_le_bytes()); + body.extend_from_slice(&(object.len() as u32).to_le_bytes()); + body.extend_from_slice(bucket.as_bytes()); + body.extend_from_slice(object.as_bytes()); + let mut hasher = crc_fast::Digest::new(crc_fast::CrcAlgorithm::Crc32IsoHdlc); + hasher.update(&body); + body.extend_from_slice(&(hasher.finalize() as u32).to_le_bytes()); + body +} + +fn write_journal_to_disks(disk_paths: &[std::path::PathBuf], data: &[u8]) { + for path in disk_paths { + let journal = path.join(META_BUCKET).join(JOURNAL_REL); + std::fs::create_dir_all(journal.parent().expect("journal parent")).expect("create journal dir"); + std::fs::write(&journal, data).expect("write journal fixture"); + } +} + +async fn wait_until(deadline: Duration, mut probe: F) -> bool +where + F: FnMut() -> Fut, + Fut: std::future::Future, +{ + let start = std::time::Instant::now(); + while start.elapsed() < deadline { + if probe().await { + return true; + } + tokio::time::sleep(Duration::from_millis(50)).await; + } + false +} + +/// A decode-failure intent delivered on the global channel must surface in the +/// heal manager as an Urgent request attributed to the MRF source. +#[tokio::test] +#[serial] +async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() { + let (_disk_paths, storage) = heal_env().await; + let manager = make_manager(storage); + + mrf_queue::spawn_mrf_consumer(manager.clone()); + + assert!( + mrf_channel::try_send_mrf_intent(MrfKind::DecodeFailure, "mrf-bucket", "mrf-object", None), + "intent should be accepted while the consumer holds the channel" + ); + + let appeared = wait_until(Duration::from_secs(10), || async { + let snapshot = manager.operations_snapshot().await; + snapshot.queued_by_source.mrf >= 1 && snapshot.queued_by_priority.urgent >= 1 + }) + .await; + assert!( + appeared, + "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})", + manager.operations_snapshot().await + ); +} + +/// A journal left behind by a previous process must be replayed into the +/// manager queue and then removed, and a torn tail must not block replay of +/// the intact records. +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[serial] +async fn journal_replay_arms_intents_and_deletes_the_file() { + let (disk_paths, storage) = heal_env().await; + + // The journal reader resolves disks through the process-local disk map; + // register the environment's disks the same way server startup does. + let mut endpoints: Vec = disk_paths + .iter() + .map(|p| Endpoint::try_from(p.to_string_lossy().as_ref()).expect("endpoint from disk path")) + .collect(); + for (i, endpoint) in endpoints.iter_mut().enumerate() { + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + } + let pool = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: endpoints.len(), + endpoints: Endpoints::from(endpoints), + cmd_line: "mrf-test".to_string(), + platform: String::new(), + }; + init_local_disks(EndpointServerPools::from(vec![pool])) + .await + .expect("local disks should register"); + + let mut journal = journal_record(1, "replay-bucket", "replay-object", Some([9u8; 16]), 0); + journal.extend(journal_record(3, "replay-bucket", "partial-object", None, 1)); + // Torn tail: a third record truncated mid-way must not block the two + // intact records above. + journal.extend_from_slice(&journal_record(2, "replay-bucket", "metadata-object", None, 0)[..8]); + write_journal_to_disks(&disk_paths, &journal); + + let manager = make_manager(storage); + // Replay directly (not via the process-global channel consumer, which the + // sibling test already claimed in this process under plain `cargo test`). + let replayed = mrf_queue::replay_journal_once(&manager).await; + assert_eq!(replayed, 2, "the two intact records must be replayed"); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_source.mrf, 2, "replayed intents must be attributed to the MRF source"); + + assert!( + disk_paths + .iter() + .all(|path| !Path::new(path).join(META_BUCKET).join(JOURNAL_REL).exists()), + "the journal file must be removed after a successful replay" + ); + + let snapshot = manager.operations_snapshot().await; + assert_eq!(snapshot.queued_by_priority.urgent, 1, "the decode-failure record must replay as Urgent"); + assert!(snapshot.queued_by_priority.normal >= 1, "the partial-write record must replay as Normal"); +} diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index 15a01507c..030c55fc9 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -2478,6 +2478,15 @@ impl FolderScanner { } if let GetSizeFailureAction::HealMetadata { object } = failure_action { + // MRF journal intent: durable High-priority Metadata + // heal across restarts (HS-01); the scanner heal + // request below stays as the immediate path. + rustfs_common::mrf_channel::try_send_mrf_intent( + rustfs_common::mrf_channel::MrfKind::MetadataCorruption, + &item.bucket, + &object, + None, + ); self.send_required_scanner_heal_request( PendingScannerHealKind::Object, item.bucket.clone(), diff --git a/rustfs/src/admin/handlers/heal.rs b/rustfs/src/admin/handlers/heal.rs index d9e7d1f7e..52e9fc152 100644 --- a/rustfs/src/admin/handlers/heal.rs +++ b/rustfs/src/admin/handlers/heal.rs @@ -317,6 +317,7 @@ fn add_source_counts(total: &mut rustfs_heal::HealSourceCounts, next: rustfs_hea total.auto_heal = total.auto_heal.saturating_add(next.auto_heal); total.internal = total.internal.saturating_add(next.internal); total.read_repair = total.read_repair.saturating_add(next.read_repair); + total.mrf = total.mrf.saturating_add(next.mrf); } fn add_operations(total: &mut rustfs_heal::HealOperationsSnapshot, next: rustfs_heal::HealOperationsSnapshot) { @@ -2353,6 +2354,7 @@ mod tests { auto_heal: value, internal: value, read_repair: value, + mrf: value, }; let operations = |value| rustfs_heal::HealOperationsSnapshot { queue_length: value, diff --git a/rustfs/src/storage/rpc/node_service/heal.rs b/rustfs/src/storage/rpc/node_service/heal.rs index f9b990bf3..8f796726c 100644 --- a/rustfs/src/storage/rpc/node_service/heal.rs +++ b/rustfs/src/storage/rpc/node_service/heal.rs @@ -585,6 +585,7 @@ mod tests { let decoded = decode_node_heal_status(&encoded).expect("fixed v1 fixture should decode"); assert_eq!(decoded.info().bitrot_start_cycle, 9); assert_eq!(decoded.operations.queue_length, 2); + assert_eq!(decoded.operations.queued_by_source.mrf, 0); } #[test]