feat(heal): wire MRF intents with durable repair journal (HS-01) (#6189)

* feat(common): add MRF intent channel and Mrf request source (HS-01)

Introduce the producer-facing half of the mission repair feed: a global
bounded (8192) channel carrying lightweight MrfIntent values from IO
error paths, plus the RUSTFS_HEAL_MRF_ENABLE delivery kill-switch and
config constants for queue/journal sizing. Delivery is strictly
non-blocking (try_send, drop-on-full) so it can sit on decode-failure
and partial-write paths without adding latency. HealRequestSource grows
a 'mrf' variant so admission accounting can attribute replayed intents.

Part of backlog#1865 (option a: wire HealEvent-style intents with a
durable retry ledger).

Co-Authored-By: heihutu <heihutu@gmail.com>

* 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 <heihutu@gmail.com>

* feat(ecstore,scanner): deliver MRF intents from error paths (HS-01)

Wire the three production delivery points, each a single non-blocking
try_send next to the existing in-memory heal paths, which stay as the
fast path:

- read.rs decode-error branch: DecodeFailure intent beside the existing
  read-repair submit, so an Urgent ECDecode request survives restarts
  even when the Low-priority read-repair request was dropped or lost.
- add_partial: PartialWrite intent, giving partial-write recovery a
  durable Normal-priority object heal across restarts.
- scanner_folder metadata-corruption classification: MetadataCorruption
  intent beside the existing High-priority scanner heal request.

All three are on error paths only: zero cost on healthy IO.

Part of backlog#1865 (option a).

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix: include mrf heal source counts

Co-Authored-By: heihutu <heihutu@gmail.com>

* fix: keep node heal status wire compatibility

Co-Authored-By: heihutu <heihutu@gmail.com>

---------

Co-authored-by: heihutu <heihutu@gmail.com>

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,