Merge branch 'main' into cursor/minio-scanner-compat-2e74

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",
         }
     }
 }

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;

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<str>` so re-arming an
+/// intent never re-allocates the strings.
+#[derive(Clone, Debug)]
+pub struct MrfIntent {
+    pub bucket: Arc<str>,
+    pub object: Arc<str>,
+    /// 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<mpsc::Sender<MrfIntent>> = 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<mpsc::Receiver<MrfIntent>, &'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<str>` handles plus the channel slot.
+pub fn try_send_mrf_intent(kind: MrfKind, bucket: &str, object: &str, version_id: Option<Uuid>) -> 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);
+    }
+}

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;

crates/config/src/constants/object.rs

@@ -234,6 +234,31 @@ pub const ENV_OBJECT_DISK_WRITE_ABSOLUTE_CAP: &str = "RUSTFS_OBJECT_DISK_WRITE_A
 /// Default absolute per-object erasure write cap in seconds (`0` = disabled).
 pub const DEFAULT_OBJECT_DISK_WRITE_ABSOLUTE_CAP: u64 = 0;
 
+/// Enable foreground PutObject request admission.
+///
+/// This is an experimental, default-off foreground write backpressure gate for
+/// strict commit tail investigations. When disabled, PUTs follow the legacy
+/// path and only the existing request counters are updated.
+pub const ENV_PUT_FOREGROUND_ADMISSION_ENABLE: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_ENABLE";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE: bool = false;
+
+/// Maximum foreground PutObject requests admitted concurrently per process.
+///
+/// The limit is used only when [`ENV_PUT_FOREGROUND_ADMISSION_ENABLE`] is true.
+/// A value of `0` disables the gate even when the enable flag is present, so a
+/// partially configured rollout cannot reject every PUT.
+pub const ENV_PUT_FOREGROUND_ADMISSION_LIMIT: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_LIMIT";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_LIMIT: usize = 0;
+
+/// Time in milliseconds a foreground PutObject waits for an admission permit.
+///
+/// Once this timeout expires the request fails before body ingest/storage
+/// mutation with S3 `SlowDown`/503. `0` means fail fast when the limit is full.
+pub const ENV_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: &str = "RUSTFS_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS";
+pub const DEFAULT_PUT_FOREGROUND_ADMISSION_WAIT_TIMEOUT_MS: u64 = 0;
+
+const _: () = assert!(!DEFAULT_PUT_FOREGROUND_ADMISSION_ENABLE);
+
 /// Environment variable for minimum GetObject timeout in seconds.
 ///
 /// When dynamic timeout calculation is enabled, this is the minimum timeout

crates/ecstore/src/set_disk/ops/heal.rs

@@ -3297,4 +3297,223 @@ mod heal_result_report_tests {
         assert!(result.detail.contains("part 1"));
         assert!(result.detail.contains("bitrot_failure=true"));
     }
+
+    // HS-12 (backlog#1874): a versioned DELETE racing an object heal must never
+    // resurrect the deleted version. The heal has real reconstruction work (a
+    // shard of the doomed version is removed), so both sides touch the same
+    // (bucket, object, data_dir); whichever order the ns write lock serializes
+    // them in, the committed delete must win.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn heal_racing_version_delete_never_resurrects_the_deleted_version() {
+        let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "heal-race-delete-no-resurrect";
+        let object = "object.bin";
+        set.make_bucket(
+            bucket,
+            &MakeBucketOptions {
+                versioning_enabled: true,
+                ..Default::default()
+            },
+        )
+        .await
+        .expect("versioned bucket should be created");
+
+        let mut first_reader = PutObjReader::from_vec(vec![0x11; 1024 * 1024]);
+        let first_info = set
+            .put_object(
+                bucket,
+                object,
+                &mut first_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("first version should be written");
+        let first_version = first_info
+            .version_id
+            .expect("versioned put should return the first version id")
+            .to_string();
+
+        let mut second_reader = PutObjReader::from_vec(vec![0x22; 1024 * 1024]);
+        let second_info = set
+            .put_object(
+                bucket,
+                object,
+                &mut second_reader,
+                &ObjectOptions {
+                    versioned: true,
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("second version should be written");
+        let second_version = second_info
+            .version_id
+            .expect("versioned put should return the second version id")
+            .to_string();
+
+        // Damage one shard of the doomed version so the racing heal performs an
+        // actual reconstruction over its data dir instead of an early exit.
+        let doomed_source = disks[0]
+            .read_version("", bucket, object, &first_version, &ReadOptions::default())
+            .await
+            .expect("doomed version metadata should be readable");
+        let doomed_data_dir = doomed_source
+            .data_dir
+            .expect("non-inline version should have a data directory");
+        tokio::fs::remove_file(
+            temp_dirs[1]
+                .path()
+                .join(bucket)
+                .join(object)
+                .join(doomed_data_dir.to_string())
+                .join("part.1"),
+        )
+        .await
+        .expect("shard damage should be injected before the race");
+
+        let delete_set = set.clone();
+        let (delete_res, heal_res) = tokio::join!(
+            async {
+                delete_set
+                    .delete_object(
+                        bucket,
+                        object,
+                        ObjectOptions {
+                            versioned: true,
+                            version_id: Some(first_version.clone()),
+                            object_lock_config_snapshot: Some(Arc::new(crate::set_disk::ObjectLockConfigSnapshot::new(
+                                crate::bucket::metadata_sys::ObjectLockConfigState::ConfirmedAbsent,
+                            ))),
+                            ..Default::default()
+                        },
+                    )
+                    .await
+            },
+            async {
+                set.heal_object(
+                    bucket,
+                    object,
+                    "",
+                    &HealOpts {
+                        scan_mode: HealScanMode::Deep,
+                        ..Default::default()
+                    },
+                )
+                .await
+            },
+        );
+        delete_res.expect("version delete must succeed under lock serialization");
+        // The heal may legitimately report a transient failure when the version
+        // it was rebuilding disappears mid-flight; only the end state matters.
+        drop(heal_res);
+
+        let resurrected = set
+            .get_object_info(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(first_version.clone()),
+                    ..Default::default()
+                },
+            )
+            .await;
+        assert!(
+            matches!(&resurrected, Err(Error::FileVersionNotFound) | Err(Error::ObjectNotFound(..))),
+            "a racing heal must not resurrect the deleted version: {resurrected:?}"
+        );
+
+        let survivor = set
+            .get_object_info(
+                bucket,
+                object,
+                &ObjectOptions {
+                    versioned: true,
+                    version_id: Some(second_version.clone()),
+                    ..Default::default()
+                },
+            )
+            .await
+            .expect("surviving version must remain readable after the race");
+        assert_eq!(survivor.size, 1024 * 1024, "survivor size must be intact");
+    }
+
+    // HS-12 (backlog#1874): unversioned overwrite commits race a Deep heal on
+    // the same object. The overwrite's post-commit tail deletes the replaced
+    // data dir without the ns lock (object.rs commit tail), which is exactly
+    // the intersection the audit flagged: the heal must tolerate the tail race
+    // (retryable outcome) and every committed overwrite must survive — the
+    // final current version is exactly the last payload written.
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn heal_racing_unversioned_overwrites_preserves_the_last_commit() {
+        let (temp_dirs, disks, set) = hermetic_set_disks_isolated(4).await;
+        let bucket = "heal-race-put-overwrite";
+        let object = "object.bin";
+        set.make_bucket(bucket, &MakeBucketOptions::default())
+            .await
+            .expect("bucket should be created");
+
+        const ROUNDS: usize = 8;
+        const PAYLOAD_SIZE: usize = 256 * 1024;
+        let mut last_etag = String::new();
+        for round in 0..ROUNDS {
+            // Give the heal something to rebuild on alternating rounds: remove a
+            // shard of the current data dir right before the race.
+            if round % 2 == 1 {
+                let current = disks[2]
+                    .read_version("", bucket, object, "", &ReadOptions::default())
+                    .await
+                    .expect("current metadata should be readable");
+                if let Some(data_dir) = current.data_dir {
+                    let shard = temp_dirs[3]
+                        .path()
+                        .join(bucket)
+                        .join(object)
+                        .join(data_dir.to_string())
+                        .join("part.1");
+                    if shard.exists() {
+                        tokio::fs::remove_file(&shard)
+                            .await
+                            .expect("shard damage should be injectable mid-race");
+                    }
+                }
+            }
+
+            let payload = vec![round as u8; PAYLOAD_SIZE];
+            let mut put_reader = PutObjReader::from_vec(payload);
+            let put_opts = ObjectOptions::default();
+            let heal_opts = HealOpts {
+                scan_mode: HealScanMode::Deep,
+                ..Default::default()
+            };
+            let (put_res, heal_res) = tokio::join!(
+                set.put_object(bucket, object, &mut put_reader, &put_opts),
+                set.heal_object(bucket, object, "", &heal_opts),
+            );
+            let put_info = put_res.expect("overwrite must succeed under lock serialization");
+            last_etag = put_info.etag.clone().unwrap_or_default();
+            // Heal outcome is unconstrained (may hit the tail race and report a
+            // retryable error); the invariant is checked on the end state.
+            drop(heal_res);
+        }
+
+        let final_info = set
+            .get_object_info(bucket, object, &ObjectOptions::default())
+            .await
+            .expect("object must remain readable after the race loop");
+        assert_eq!(
+            final_info.size, PAYLOAD_SIZE as i64,
+            "final current version must be the last committed overwrite"
+        );
+        assert_eq!(
+            final_info.etag.unwrap_or_default(),
+            last_etag,
+            "the racing heal loop must never leave a stale or resurrected current version"
+        );
+    }
 }

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()),

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,

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"] }

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

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,
         }
     }
 }

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;

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<MrfIntent>,
+    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<MrfIntent> {
+        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<Item = &MrfIntent> {
+        self.pending.iter()
+    }
+}
+
+// ---------------------------------------------------------------------------
+// Journal record codec
+// ---------------------------------------------------------------------------
+
+/// Append one encoded record to `out`.
+pub(crate) fn encode_intent(intent: &MrfIntent, out: &mut Vec<u8>) {
+    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<MrfIntent>, 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<DiskStore> {
+    let map = local_disk_map_read().await;
+    map.values().flatten().cloned().collect()
+}
+
+async fn read_journal() -> Option<Vec<u8>> {
+    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<tokio::time::Instant>,
+}
+
+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<u8> {
+        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<HealManager>) {
+    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<HealManager>) -> usize {
+    let config = MrfConsumerConfig::default();
+    let mut queue = MrfQueue::new(config.queue_capacity, config.journal_max_bytes);
+    let mut backoff_until: Option<tokio::time::Instant> = 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<HealManager>,
+    queue: &mut MrfQueue,
+    backoff_until: &mut Option<tokio::time::Instant>,
+) -> 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<HealManager>, mut receiver: mpsc::Receiver<MrfIntent>) {
+    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<MrfIntent> = 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);
+    }
+}

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;
 

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<std::path::PathBuf>, Arc<dyn HealStorageAPI>) {
+    let env = rustfs_test_utils::TestECStoreEnv::builder()
+        .prefix("rustfs_heal_mrf_test")
+        .build()
+        .await;
+    let heal_storage: Arc<dyn HealStorageAPI> = Arc::new(ECStoreHealStorage::new(env.ecstore.clone()));
+    (env.disk_paths, heal_storage)
+}
+
+fn make_manager(storage: Arc<dyn HealStorageAPI>) -> Arc<HealManager> {
+    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<u8> {
+    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<F, Fut>(deadline: Duration, mut probe: F) -> bool
+where
+    F: FnMut() -> Fut,
+    Fut: std::future::Future<Output = bool>,
+{
+    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<Endpoint> = 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");
+}

crates/io-core/src/io_profile.rs

@@ -26,7 +26,6 @@ pub enum StorageMedia {
 }
 
 impl StorageMedia {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Nvme => "nvme",
@@ -60,7 +59,6 @@ pub enum AccessPattern {
 }
 
 impl AccessPattern {
-    #[allow(dead_code)]
     pub fn as_str(&self) -> &'static str {
         match self {
             Self::Sequential => "sequential",
@@ -71,25 +69,21 @@ impl AccessPattern {
     }
 
     /// Check if this is a sequential access pattern.
-    #[allow(dead_code)]
     pub fn is_sequential(&self) -> bool {
         matches!(self, Self::Sequential)
     }
 
     /// Check if this is a random access pattern.
-    #[allow(dead_code)]
     pub fn is_random(&self) -> bool {
         matches!(self, Self::Random)
     }
 
     /// Check if this is a mixed access pattern.
-    #[allow(dead_code)]
     pub fn is_mixed(&self) -> bool {
         matches!(self, Self::Mixed)
     }
 
     /// Check if this pattern is unknown.
-    #[allow(dead_code)]
     pub fn is_unknown(&self) -> bool {
         matches!(self, Self::Unknown)
     }

crates/object-capacity/src/capacity_manager.rs

@@ -427,7 +427,6 @@ pub enum DataSource {
     /// Write triggered
     WriteTriggered,
     /// Fallback value
-    #[allow(dead_code)]
     Fallback,
 }
 
@@ -603,7 +602,6 @@ impl WriteRecord {
 
 /// Hybrid strategy configuration
 #[derive(Debug, Clone)]
-#[allow(dead_code)]
 pub struct HybridStrategyConfig {
     /// Scheduled update interval
     pub scheduled_update_interval: Duration,
@@ -998,14 +996,12 @@ impl HybridCapacityManager {
     }
 
     /// Get cache age
-    #[allow(dead_code)]
     pub async fn get_cache_age(&self) -> Option<Duration> {
         let cache = self.cache.read().await;
         cache.as_ref().map(|c| c.last_update.elapsed())
     }
 
     /// Get write frequency (writes/minute)
-    #[allow(dead_code)]
     pub async fn get_write_frequency(&self) -> usize {
         let record = &self.write_record;
         record.recent_write_count(record.monotonic_second())
@@ -1300,7 +1296,6 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
 ///     .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
 ///     .await;
 /// ```
-#[allow(dead_code)]
 pub fn create_isolated_manager(config: HybridStrategyConfig) -> Arc<HybridCapacityManager> {
     Arc::new(HybridCapacityManager::new(config))
 }

crates/rio/src/compress_index.rs

@@ -49,7 +49,6 @@ pub struct IndexInfo {
     pub uncompressed_offset: i64,
 }
 
-#[allow(dead_code)]
 impl Index {
     pub fn new() -> Self {
         Self {
@@ -60,14 +59,6 @@ impl Index {
         }
     }
 
-    #[allow(dead_code)]
-    fn reset(&mut self, max_block: usize) {
-        self.est_block_uncomp = max_block as i64;
-        self.total_compressed = -1;
-        self.total_uncompressed = -1;
-        self.info.clear();
-    }
-
     pub fn len(&self) -> usize {
         self.info.len()
     }
@@ -511,47 +502,6 @@ fn read_varint(buf: &[u8]) -> io::Result<(i64, usize)> {
     Err(io::Error::new(io::ErrorKind::UnexpectedEof, "unexpected EOF"))
 }
 
-// Helper functions for index header manipulation
-#[allow(dead_code)]
-pub fn remove_index_headers(b: &[u8]) -> Option<&[u8]> {
-    if b.len() < 4 + S2_INDEX_TRAILER.len() {
-        return None;
-    }
-
-    // Skip size
-    let b = &b[4..];
-
-    // Check trailer
-    if !b.starts_with(S2_INDEX_TRAILER) {
-        return None;
-    }
-
-    Some(&b[S2_INDEX_TRAILER.len()..])
-}
-
-#[allow(dead_code)]
-pub fn restore_index_headers(in_data: &[u8]) -> Vec<u8> {
-    if in_data.is_empty() {
-        return Vec::new();
-    }
-
-    let mut b = Vec::with_capacity(4 + S2_INDEX_HEADER.len() + in_data.len() + S2_INDEX_TRAILER.len() + 4);
-    b.extend_from_slice(&[0x50, 0x2A, 0x4D, 0x18]);
-    b.extend_from_slice(S2_INDEX_HEADER);
-    b.extend_from_slice(in_data);
-
-    let total_size = (b.len() + 4 + S2_INDEX_TRAILER.len()) as u32;
-    b.extend_from_slice(&total_size.to_le_bytes());
-    b.extend_from_slice(S2_INDEX_TRAILER);
-
-    let chunk_len = b.len() - 4;
-    b[1] = chunk_len as u8;
-    b[2] = (chunk_len >> 8) as u8;
-    b[3] = (chunk_len >> 16) as u8;
-
-    b
-}
-
 #[cfg(test)]
 mod tests {
     use super::*;

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(),

crates/scanner/tests/lifecycle_integration_test.rs

@@ -206,7 +206,6 @@ async fn setup_isolated_test_env(init_expiry: bool) -> (Vec<PathBuf>, Arc<ECStor
 }
 
 /// Test helper: Create a test bucket
-#[allow(dead_code)]
 async fn create_test_bucket(ecstore: &Arc<ECStore>, bucket_name: &str) {
     (**ecstore)
         .make_bucket(bucket_name, &Default::default())
@@ -251,7 +250,6 @@ async fn modeled_versioned_delete_opts(bucket: &str, object: &str) -> ObjectOpti
 }
 
 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
     // Create a simple lifecycle configuration XML with 0 days expiry for immediate testing
     let lifecycle_xml = r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -274,7 +272,6 @@ async fn set_bucket_lifecycle(bucket_name: &str) -> Result<(), Box<dyn std::erro
 }
 
 /// Test helper: Set bucket lifecycle configuration
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<dyn std::error::Error>> {
     // Create lifecycle rule that targets delete-marker cleanup only.
     // Keep Expiration.Days unset to avoid expiring live transitioned object versions.
@@ -297,7 +294,6 @@ async fn set_bucket_lifecycle_deletemarker(bucket_name: &str) -> Result<(), Box<
     Ok(())
 }
 
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64) -> Result<(), Box<dyn std::error::Error>> {
     let lifecycle_xml = format!(
         r#"<?xml version="1.0" encoding="UTF-8"?>
@@ -320,7 +316,6 @@ async fn set_bucket_lifecycle_delmarker_expiration(bucket_name: &str, days: i64)
     Ok(())
 }
 
-#[allow(dead_code)]
 async fn set_bucket_lifecycle_transition_with_tier(
     bucket_name: &str,
     storage_class: &str,
@@ -368,7 +363,6 @@ async fn object_exists(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bo
 }
 
 /// Test helper: Check if object exists
-#[allow(dead_code)]
 async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &str) -> bool {
     if let Ok(oi) = (**ecstore).get_object_info(bucket, object, &ObjectOptions::default()).await {
         println!("oi: {oi:?}");
@@ -379,7 +373,6 @@ async fn object_is_delete_marker(ecstore: &Arc<ECStore>, bucket: &str, object: &
     }
 }
 
-#[allow(dead_code)]
 async fn wait_for_object_absence(ecstore: &Arc<ECStore>, bucket: &str, object: &str, timeout: Duration) -> bool {
     let deadline = tokio::time::Instant::now() + timeout;
 

crates/targets/src/net.rs

@@ -428,7 +428,6 @@ pub fn parse_url(s: &str) -> Result<ParsedURL, NetError> {
     Ok(ParsedURL(uu))
 }
 
-#[allow(dead_code)]
 pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
     let u = parse_url(s)?;
     match u.0.scheme() {
@@ -437,7 +436,6 @@ pub fn parse_http_url(s: &str) -> Result<ParsedURL, NetError> {
     }
 }
 
-#[allow(dead_code)]
 pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> bool {
     if err.kind() == std::io::ErrorKind::TimedOut {
         return !expect_timeouts;
@@ -449,12 +447,10 @@ pub fn is_network_or_host_down(err: &std::io::Error, expect_timeouts: bool) -> b
         || err_str.contains("use of closed network connection")
 }
 
-#[allow(dead_code)]
 pub fn is_conn_reset_err(err: &std::io::Error) -> bool {
     err.to_string().contains("connection reset by peer") || matches!(err.raw_os_error(), Some(libc::ECONNRESET))
 }
 
-#[allow(dead_code)]
 pub fn is_conn_refused_err(err: &std::io::Error) -> bool {
     err.to_string().contains("connection refused") || matches!(err.raw_os_error(), Some(libc::ECONNREFUSED))
 }

crates/targets/src/target/mqtt.rs

@@ -32,8 +32,8 @@ use arc_swap::ArcSwap;
 use async_trait::async_trait;
 use hyper_rustls::ConfigBuilderExt;
 use rumqttc::{
-    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, PublishNoticeError, QoS,
-    Transport, mqttbytes::Error as MqttBytesError,
+    AsyncClient, Broker, ClientError, ConnectionError, EventLoop, Incoming, MqttOptions, Outgoing, ProtocolViolation,
+    PublishNoticeError, PublishOptions, QoS, Transport, mqttbytes::Error as MqttBytesError,
 };
 use rustfs_config::{
     EnableState, MQTT_TLS_CA, MQTT_TLS_CLIENT_CERT, MQTT_TLS_CLIENT_KEY, MQTT_TLS_TRUST_LEAF_AS_CA, MQTT_WS_PATH_ALLOWLIST,
@@ -791,7 +791,7 @@ where
                 .as_ref()
                 .ok_or_else(|| TargetError::Configuration("MQTT client not initialized".to_string()))?;
             let notice = client
-                .publish_tracked(&self.args.topic, self.args.qos, false, body)
+                .publish_tracked(&self.args.topic, body, PublishOptions::new(self.args.qos))
                 .await
                 .map_err(|error| classify_mqtt_client_error(&error))?;
             drop(client_guard);
@@ -1145,7 +1145,7 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
                         );
                         connected_status.store(false, Ordering::SeqCst);
                     }
-                    rumqttc::Event::Incoming(Incoming::PingResp(_)) => {
+                    rumqttc::Event::Incoming(Incoming::PingResp) => {
                         trace!(target_id = %target_id, "Received PingResp from broker. Connection is alive.");
                     }
                     rumqttc::Event::Incoming(Incoming::SubAck(suback)) => {
@@ -1257,7 +1257,11 @@ async fn run_mqtt_event_loop(mut eventloop: EventLoop, connected_status: Arc<Ato
 /// copy is preserved and replayed rather than dropped (backlog#971).
 fn classify_mqtt_client_error(err: &ClientError) -> TargetError {
     match err {
-        ClientError::Request(_) | ClientError::TryRequest(_) | ClientError::TrackingUnavailable => TargetError::NotConnected,
+        ClientError::RequestChannelFull(_) | ClientError::RequestChannelDisconnected(_) | ClientError::TrackingUnavailable => {
+            TargetError::NotConnected
+        }
+        ClientError::InvalidRequest(_) => TargetError::Request(format!("Invalid MQTT publish request: {err}")),
+        _ => TargetError::NotConnected,
     }
 }
 
@@ -1270,10 +1274,14 @@ fn classify_mqtt_notice_error(err: &PublishNoticeError) -> TargetError {
         PublishNoticeError::Recv
         | PublishNoticeError::SessionReset
         | PublishNoticeError::Qos0NotFlushed
+        | PublishNoticeError::BrokerOnlySessionResume
+        | PublishNoticeError::SessionPersistence(_)
         | PublishNoticeError::TopicAliasReplayUnavailable(_) => TargetError::NotConnected,
+        PublishNoticeError::RetainNotSupported => TargetError::Request(format!("MQTT broker rejected publish: {err}")),
         PublishNoticeError::V5PubAck(_) | PublishNoticeError::V5PubRec(_) | PublishNoticeError::V5PubComp(_) => {
             TargetError::Request(format!("MQTT broker rejected publish: {err}"))
         }
+        _ => TargetError::NotConnected,
     }
 }
 
@@ -1299,12 +1307,13 @@ fn is_fatal_mqtt_error(err: &ConnectionError) -> bool {
                         | MqttBytesError::MalformedPacket // Package format error
                         | MqttBytesError::PayloadTooLong // Too long load
                         | MqttBytesError::PayloadSizeLimitExceeded { .. } // Load size limit exceeded
-                        | MqttBytesError::TopicNotUtf8 // Topic Non-UTF-8 (Serious Agreement Violation)
+                        | MqttBytesError::TopicNotUtf8 { .. } // Topic Non-UTF-8 (Serious Agreement Violation)
                     )
                 }
                 // Others that are fatal StateError variants
                 rumqttc::StateError::InvalidState          // The internal state machine is in invalid state
-                | rumqttc::StateError::WrongPacket         // Agreement Violation: Unexpected Data Packet Received
+                | rumqttc::StateError::ProtocolViolation(ProtocolViolation::UnexpectedIncomingPacket(_)) // Agreement Violation: Unexpected Data Packet Received
+                | rumqttc::StateError::ProtocolViolation(_) // Agreement Violation
                 | rumqttc::StateError::Unsolicited(_)      // Agreement Violation: Unsolicited ACK Received
                 | rumqttc::StateError::CollisionTimeout    // Agreement Violation (if this stage occurs)
                 | rumqttc::StateError::EmptySubscription   // Agreement violation (if this stage occurs)
@@ -1727,8 +1736,8 @@ where
 mod tests {
     use super::{
         AsyncClient, ClientError, MQTT_RECONNECT_BACKOFF_MAX, MQTT_RECONNECT_BACKOFF_MIN, MQTTArgs, MQTTTarget, MQTTTlsConfig,
-        MqttOptions, PublishNoticeError, QoS, QueuedPayloadMeta, classify_mqtt_client_error, classify_mqtt_notice_error,
-        next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
+        MqttOptions, PublishNoticeError, PublishOptions, QoS, QueuedPayloadMeta, classify_mqtt_client_error,
+        classify_mqtt_notice_error, next_reconnect_backoff, reconnect_supervisor, validate_mqtt_broker_url,
     };
     use crate::error::TargetError;
     use crate::target::{REDACTED_SECRET, TargetType};
@@ -1794,7 +1803,7 @@ mod tests {
             .capacity(1)
             .build();
         client
-            .publish("fill", QoS::AtLeastOnce, false, b"fill".as_slice())
+            .publish("fill", b"fill".as_slice(), PublishOptions::new(QoS::AtLeastOnce))
             .await
             .expect("first publish should fill the local channel");
         *target.client.lock().await = Some(client);