From 06eeb886f8e85dca1e8ebb76699798b8bf66fc28 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 18 Aug 2026 09:03:11 +0800 Subject: [PATCH] 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 --- crates/heal/Cargo.toml | 2 + crates/heal/src/heal/channel.rs | 3 +- crates/heal/src/heal/manager.rs | 2 + 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 +++++++ 7 files changed, 882 insertions(+), 1 deletion(-) create mode 100644 crates/heal/src/heal/mrf_queue.rs create mode 100644 crates/heal/tests/mrf_pipeline_test.rs 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 d2be2545d..9bcfd034e 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 79ba6b169..3fd3f8065 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -270,6 +270,7 @@ pub struct HealSourceCounts { pub auto_heal: u64, pub internal: u64, pub read_repair: u64, + pub mrf: u64, } impl HealSourceCounts { @@ -280,6 +281,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 7c156304d..45e9eceae 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"); +}