From f477d27861ae12f0ccfe92d9eec045211b61d698 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 18 Aug 2026 09:03:02 +0800 Subject: [PATCH] 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 --- 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 ++++ 4 files changed, 236 insertions(+) create mode 100644 crates/common/src/mrf_channel.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 09240e25b..bcbbf0ab1 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; 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;