chore: adjudicate the last 18 bare dead_code allows in the library crates (#6265)

* chore: adjudicate the last 18 bare dead_code allows in the library crates

Finishes backlog#1823 step 10 outside `rustfs/src` and `protocols`: config, s3select-query, common, madmin, heal, ecstore, signer and notify. Stripped first, then clippy asked which the compiler actually missed — 8 of the 18 were inert.

Seven items are deleted, each checked by grep as well as by clippy:

- `common/last_minute.rs`'s private `TimedAction` (with its impl) and `SizeCategory` (with its `Display` impl). The file's public surface — `AccElem`, `LastMinuteLatency` — stays; ecstore consumes it.
- `s3select-query`'s three `with_*` builders. `DefaultLogicalOptimizer::with_optimizer_rules` looks used, but the call in the same file is `SessionStateBuilder::with_optimizer_rules` from DataFusion; the local methods have no callers.
- `heal/manager.rs`'s `contains_key`. Its six apparent references are all `HashMap::contains_key`.

Three keep their code:

- `heal/storage.rs`'s `Test` variant is constructed by the `#[cfg(test)] test()` helper, which the lib target cannot see, so it takes a reasoned allow.
- `signer`'s `STREAMING_PAYLOAD_HDR` and `try_build_chunk_string_to_sign` gain the `_` prefix instead. That file already marks deliberately-unheld code that way — `_STREAMING_TRAILER_HDR`, `_PAYLOAD_CHUNK_SIZE`, and `_try_build_chunk_signature`, which is the only caller of that function. Following the existing convention removes the allow without an attribute.

`protocols` keeps its four; that crate needs `--features swift,sftp` to compile fully and is verified differently. The four `#![allow(dead_code)]` in `e2e_test` are module-root blankets in test-support files, which belong to steps 1-5 rather than step 10.

Refs backlog#1823

* chore(e2e_test): adjudicate the two dead_code allows the lib test target still needs

`cargo clippy --all-targets` compiles e2e_test's lib test target, which the earlier pass did not cover, so these two removals only surfaced in CI.

test_large_multipart_upload's allow was load-bearing: its call site in test_local_kms_multipart_upload is commented out behind "TODO: Re-enable after fixing streaming encryption issues with large files". The allow comes back with the reason string this batch uses everywhere else, so the next reader sees why it is parked instead of deleting a test we intend to run again.

TestDefinition.category was the opposite: written at all six definitions, read nowhere, and its enum's impl block is empty. The live copy of that type is crates/e2e_test/src/kms/test_runner.rs, which has an as_str; the policy copy is a vestige of it. Dropping the field, the enum, and the constructor parameter leaves the runner unchanged — it dispatches on name and filters on is_critical.

Verification: cargo clippy --all-targets -- -D warnings (workspace, the CI command) and cargo fmt --all --check both pass.

---------

Co-authored-by: houseme <housemecn@gmail.com>

crates/common/src/last_minute.rs

@@ -13,82 +13,6 @@
 // limitations under the License.
 
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
-
-#[allow(dead_code)]
-#[derive(Debug, Default)]
-struct TimedAction {
-    count: u64,
-    acc_time: u64,
-    min_time: Option<u64>,
-    max_time: Option<u64>,
-    bytes: u64,
-}
-
-#[allow(dead_code)]
-impl TimedAction {
-    // Avg returns the average time spent on the action.
-    pub fn avg(&self) -> Option<Duration> {
-        if self.count == 0 {
-            return None;
-        }
-        Some(Duration::from_nanos(self.acc_time / self.count))
-    }
-
-    // AvgBytes returns the average bytes processed.
-    pub fn avg_bytes(&self) -> u64 {
-        if self.count == 0 {
-            return 0;
-        }
-        self.bytes / self.count
-    }
-
-    // Merge other into t.
-    pub fn merge(&mut self, other: TimedAction) {
-        self.count += other.count;
-        self.acc_time += other.acc_time;
-        self.bytes += other.bytes;
-
-        if self.count == 0 {
-            self.min_time = other.min_time;
-        }
-        if let Some(other_min) = other.min_time {
-            self.min_time = self.min_time.map_or(Some(other_min), |min| Some(min.min(other_min)));
-        }
-
-        self.max_time = self
-            .max_time
-            .map_or(other.max_time, |max| Some(max.max(other.max_time.unwrap_or(0))));
-    }
-}
-
-#[allow(dead_code)]
-#[derive(Debug)]
-enum SizeCategory {
-    SizeLessThan1KiB = 0,
-    SizeLessThan1MiB,
-    SizeLessThan10MiB,
-    SizeLessThan100MiB,
-    SizeLessThan1GiB,
-    SizeGreaterThan1GiB,
-    // Add new entries here
-    SizeLastElemMarker,
-}
-
-impl std::fmt::Display for SizeCategory {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        let s = match *self {
-            SizeCategory::SizeLessThan1KiB => "SizeLessThan1KiB",
-            SizeCategory::SizeLessThan1MiB => "SizeLessThan1MiB",
-            SizeCategory::SizeLessThan10MiB => "SizeLessThan10MiB",
-            SizeCategory::SizeLessThan100MiB => "SizeLessThan100MiB",
-            SizeCategory::SizeLessThan1GiB => "SizeLessThan1GiB",
-            SizeCategory::SizeGreaterThan1GiB => "SizeGreaterThan1GiB",
-            SizeCategory::SizeLastElemMarker => "SizeLastElemMarker",
-        };
-        write!(f, "{s}")
-    }
-}
-
 #[derive(Clone, Debug, Default, Copy)]
 pub struct AccElem {
     pub total: u64,

crates/common/src/mrf_channel.rs

@@ -148,6 +148,62 @@ fn unix_now_ms() -> u64 {
         .unwrap_or(0)
 }
 
+/// A repair the MRF consumer landed, fanned out so retry ledgers can drop
+/// entries the journal no longer tracks (backlog#1894 axis B). The payload
+/// mirrors the intent identity so consumers match without re-parsing.
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub struct MrfRepairedEvent {
+    pub bucket: Arc<str>,
+    pub object: Arc<str>,
+    pub version_id: Option<[u8; 16]>,
+}
+
+/// Bound on the repaired-event backlog. Notices are best-effort hints; when
+/// the ring is full the oldest are dropped and the affected ledger entries
+/// simply expire through their own attempts/age limits.
+const MRF_REPAIRED_EVENT_CAP: usize = 4096;
+
+static MRF_REPAIRED_EVENTS: OnceLock<std::sync::Mutex<std::collections::VecDeque<MrfRepairedEvent>>> = OnceLock::new();
+
+/// Record that the MRF consumer landed a repair. Never blocks: the critical
+/// section is a deque push under a std mutex.
+pub fn note_mrf_repaired(bucket: &str, object: &str, version_id: Option<[u8; 16]>) {
+    let registry = MRF_REPAIRED_EVENTS.get_or_init(|| std::sync::Mutex::new(std::collections::VecDeque::new()));
+    let Ok(mut events) = registry.lock() else {
+        return;
+    };
+    if events.len() >= MRF_REPAIRED_EVENT_CAP {
+        events.pop_front();
+    }
+    events.push_back(MrfRepairedEvent {
+        bucket: Arc::from(bucket),
+        object: Arc::from(object),
+        version_id,
+    });
+}
+
+/// Take the repair notices recorded for `bucket`, leaving other buckets'
+/// notices in place for their own scanners.
+pub fn take_mrf_repaired_events_for(bucket: &str) -> Vec<MrfRepairedEvent> {
+    let Some(registry) = MRF_REPAIRED_EVENTS.get() else {
+        return Vec::new();
+    };
+    let Ok(mut events) = registry.lock() else {
+        return Vec::new();
+    };
+    let mut taken = Vec::new();
+    let mut retained = std::collections::VecDeque::with_capacity(events.len());
+    while let Some(event) = events.pop_front() {
+        if event.bucket.as_ref() == bucket {
+            taken.push(event);
+        } else {
+            retained.push_back(event);
+        }
+    }
+    *events = retained;
+    taken
+}
+
 #[cfg(test)]
 mod tests {
     use super::*;
@@ -200,4 +256,32 @@ mod tests {
         assert!(!try_send_mrf_intent(MrfKind::MetadataCorruption, "b", "o", None));
         set_mrf_delivery_enabled(true);
     }
+
+    #[test]
+    fn repaired_events_take_is_bucket_scoped_and_cap_bounded() {
+        // Distinct buckets keep their notices until their own scanner takes
+        // them; a take for one bucket leaves the others' notices in place.
+        note_mrf_repaired("bucket-a", "object-1", None);
+        note_mrf_repaired("bucket-b", "object-2", None);
+        note_mrf_repaired("bucket-a", "object-3", None);
+
+        let taken_a = take_mrf_repaired_events_for("bucket-a");
+        assert_eq!(taken_a.len(), 2);
+        assert_eq!(taken_a[0].object.as_ref(), "object-1");
+        assert_eq!(taken_a[1].object.as_ref(), "object-3");
+        assert!(take_mrf_repaired_events_for("bucket-a").is_empty(), "take is destructive per bucket");
+
+        let taken_b = take_mrf_repaired_events_for("bucket-b");
+        assert_eq!(taken_b.len(), 1);
+        assert_eq!(taken_b[0].object.as_ref(), "object-2");
+
+        // Cap bound: flooding the ring drops the oldest notices rather than
+        // growing unbounded.
+        for i in 0..=(MRF_REPAIRED_EVENT_CAP + 8) {
+            note_mrf_repaired("flood-bucket", &format!("object-{i}"), None);
+        }
+        let flooded = take_mrf_repaired_events_for("flood-bucket");
+        assert_eq!(flooded.len(), MRF_REPAIRED_EVENT_CAP);
+        assert_eq!(flooded[0].object.as_ref(), "object-9", "the oldest notices past the cap are dropped");
+    }
 }

crates/config/src/notify/mod.rs

@@ -92,15 +92,11 @@ pub const NOTIFY_SUB_SYSTEMS: &[&str] = &[
 pub const NOTIFY_KAFKA_SUB_SYS: &str = "notify_kafka";
 pub const NOTIFY_MQTT_SUB_SYS: &str = "notify_mqtt";
 pub const NOTIFY_MYSQL_SUB_SYS: &str = "notify_mysql";
-#[allow(dead_code)]
 pub const NOTIFY_NATS_SUB_SYS: &str = "notify_nats";
-#[allow(dead_code)]
 pub const NOTIFY_NSQ_SUB_SYS: &str = "notify_nsq";
-#[allow(dead_code)]
 pub const NOTIFY_ES_SUB_SYS: &str = "notify_elasticsearch";
 pub const NOTIFY_AMQP_SUB_SYS: &str = "notify_amqp";
 pub const NOTIFY_POSTGRES_SUB_SYS: &str = "notify_postgres";
-#[allow(dead_code)]
 pub const NOTIFY_REDIS_SUB_SYS: &str = "notify_redis";
 pub const NOTIFY_REDIS_DEFAULT_CHANNEL: &str = "rustfs_notify_channel";
 pub const NOTIFY_PULSAR_SUB_SYS: &str = "notify_pulsar";

crates/e2e_test/src/get_stream_failure_observability_test.rs

@@ -1,250 +0,0 @@
-// 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.
-
-//! E2E proof that a mid-stream GET failure is *reportable* — rustfs#4784.
-//!
-//! The functional invariant (a beyond-quorum read must fail rather than return
-//! a clean short body) is already covered by
-//! `degraded_read_eof_regression_test`. This suite covers the half that issue
-//! #4784 got stuck on for a month: whether an operator can tell, from the
-//! source server's log alone, that a GET failed mid-body and **which object**
-//! it failed on.
-//!
-//! The reporter saw only downstream symptoms — `rclone` reporting
-//! `unexpected EOF` on its PUT, and the receiving RustFS logging
-//! `Io error: error reading a body from connection` with a 500. In a cross-remote
-//! `rclone sync`, the source GET body *is* the destination PUT body, so a source
-//! read that ends short of its committed `Content-Length` surfaces as a PUT
-//! failure on the far side. Built-in replication and site replication have the
-//! same shape (read locally, PUT remotely), which is why every transport in that
-//! report failed the same way.
-//!
-//! The source side, meanwhile, said nothing:
-//!   * `GetObjectReaderStream`'s short-read and read-error arms only incremented
-//!     a metric; their log lines sat behind the `tracing-chunk-debug` cargo
-//!     feature, which is not in the default feature set and therefore is not
-//!     compiled into any released binary.
-//!   * `GetObjectStreamingReader` did log mid-stream failures, but only under a
-//!     `request_id` — with no bucket or object name, a failure could not be
-//!     traced back to the object that caused it.
-//!   * Those lines were `warn!`, while `DEFAULT_LOG_LEVEL` is `error`, so a
-//!     default deployment filtered them out anyway.
-//!
-//! This test reproduces the source-side fault against a real server over the S3
-//! API and asserts the operator-visible evidence, at the **default** log level.
-
-#[cfg(test)]
-mod tests {
-    use crate::chaos::DiskFaultHarness;
-    use crate::common::init_logging;
-    use aws_sdk_s3::Client;
-    use aws_sdk_s3::primitives::ByteStream;
-    use aws_sdk_s3::types::{CompletedMultipartUpload, CompletedPart};
-    use serial_test::serial;
-    use std::error::Error;
-    use tokio::time::{Duration, timeout};
-    use tracing::info;
-
-    type TestResult = Result<(), Box<dyn Error + Send + Sync>>;
-
-    const MIB: usize = 1024 * 1024;
-    const OP_TIMEOUT: Duration = Duration::from_secs(90);
-
-    /// The structured event name every GET body failure is tagged with.
-    const STREAM_BODY_EVENT: &str = "get_object_stream_body";
-
-    fn payload(len: usize, seed: u8) -> Vec<u8> {
-        (0..len)
-            .map(|i| (i as u64).wrapping_mul(2654435761).wrapping_add(seed as u64) as u8)
-            .collect()
-    }
-
-    /// Upload a multipart object so the data lands in real `part.*` shard files
-    /// rather than being inlined into `xl.meta` (inlined objects cannot be
-    /// corrupted shard-wise, and never exercise the streaming read path).
-    async fn put_multipart(
-        client: &Client,
-        bucket: &str,
-        key: &str,
-        parts: Vec<Vec<u8>>,
-    ) -> Result<usize, Box<dyn Error + Send + Sync>> {
-        let total_len = parts.iter().map(Vec::len).sum();
-
-        let create = client.create_multipart_upload().bucket(bucket).key(key).send().await?;
-        let upload_id = create.upload_id().ok_or("missing upload id")?.to_string();
-
-        let mut completed = Vec::with_capacity(parts.len());
-        for (index, part_body) in parts.into_iter().enumerate() {
-            let part_number = (index + 1) as i32;
-            let uploaded = timeout(
-                OP_TIMEOUT,
-                client
-                    .upload_part()
-                    .bucket(bucket)
-                    .key(key)
-                    .upload_id(&upload_id)
-                    .part_number(part_number)
-                    .body(ByteStream::from(part_body))
-                    .send(),
-            )
-            .await
-            .map_err(|_| format!("upload_part {part_number} timed out"))??;
-            completed.push(
-                CompletedPart::builder()
-                    .part_number(part_number)
-                    .e_tag(uploaded.e_tag().ok_or("missing part etag")?)
-                    .build(),
-            );
-        }
-
-        timeout(
-            OP_TIMEOUT,
-            client
-                .complete_multipart_upload()
-                .bucket(bucket)
-                .key(key)
-                .upload_id(&upload_id)
-                .multipart_upload(CompletedMultipartUpload::builder().set_parts(Some(completed)).build())
-                .send(),
-        )
-        .await
-        .map_err(|_| "complete_multipart_upload timed out")??;
-
-        Ok(total_len)
-    }
-
-    /// rustfs#4784: reproduce the source-side fault the reporter kept hitting —
-    /// a GET that commits `200` + a full `Content-Length` and then cannot finish
-    /// the body — and assert the server log names the object, at the log level a
-    /// default deployment actually runs with.
-    #[tokio::test]
-    #[serial]
-    async fn midstream_get_failure_is_logged_with_the_object_at_default_log_level() -> TestResult {
-        init_logging();
-        info!("rustfs#4784: a mid-stream GET failure must name its object in the source log");
-
-        let mut harness = DiskFaultHarness::new(4).await?;
-
-        // Capture the child's stdout so the test can read what an operator would.
-        let log_path = format!("{}/server.log", harness.env.temp_dir);
-        harness.env.capture_log_path = Some(log_path.clone());
-
-        // Reproduce a DEFAULT deployment's logging, not the e2e harness's
-        // permissive `rustfs=info`: `DEFAULT_LOG_LEVEL` is `error`. Before the
-        // #4784 fix these failures were `warn!`, so a default deployment
-        // filtered them out entirely — which is why the reporter's source logs
-        // were empty. extra_env is applied after the harness's own RUST_LOG, so
-        // this wins.
-        harness.set_env("RUST_LOG", "error");
-        harness.set_env("RUSTFS_OBS_LOGGER_LEVEL", "error");
-
-        harness.start_server().await?;
-        let client = harness.env.create_s3_client();
-
-        let bucket = "issue4784-source-read";
-        client.create_bucket().bucket(bucket).send().await?;
-
-        // Named after the reporter's restic index objects, which is where they
-        // saw the failures.
-        let key = "index/3b18542ab3af4c3d03f804c7a24173e7836ef7fa447b5d1e9d634f975cc51611";
-        let expected_len = put_multipart(
-            &client,
-            bucket,
-            key,
-            vec![payload(5 * MIB, 71), payload(5 * MIB, 72), payload(5 * MIB, 73)],
-        )
-        .await?;
-
-        // Baseline: the object reads back completely before any corruption.
-        let baseline = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send())
-            .await
-            .map_err(|_| "baseline GET timed out")??
-            .body
-            .collect()
-            .await?;
-        assert_eq!(baseline.into_bytes().len(), expected_len, "baseline GET must return the whole object");
-
-        // Corrupt three of four shards in a 2+2 set: below the 2-shard read
-        // quorum. The corruption sits mid-file, so block 0 still reads clean —
-        // the server commits 200 + the full Content-Length and only then cannot
-        // reconstruct. That is the mid-stream window the reporter's downstream
-        // saw as `unexpected EOF`.
-        harness.corrupt_object_shard(0, bucket, key)?;
-        harness.corrupt_object_shard(1, bucket, key)?;
-        harness.corrupt_object_shard(2, bucket, key)?;
-
-        let response = timeout(OP_TIMEOUT, client.get_object().bucket(bucket).key(key).send())
-            .await
-            .map_err(|_| "degraded GET timed out")?;
-
-        // Either outcome is functionally correct (that invariant belongs to
-        // degraded_read_eof_regression_test); this suite only needs the read to
-        // have failed so there is something to report.
-        let delivered = match response {
-            Err(err) => {
-                info!("degraded GET failed before the body: {err}");
-                None
-            }
-            Ok(response) => match response.body.collect().await {
-                Ok(aggregated) => Some(aggregated.into_bytes().len()),
-                Err(err) => {
-                    info!("degraded GET failed mid-body as expected: {err}");
-                    None
-                }
-            },
-        };
-        assert_ne!(
-            delivered,
-            Some(expected_len),
-            "the beyond-quorum read unexpectedly succeeded; this suite needs a failed read to have something to report"
-        );
-
-        // Give the child a moment to flush its stdout.
-        tokio::time::sleep(Duration::from_millis(500)).await;
-        let logged = std::fs::read_to_string(&log_path)?;
-
-        let failure_lines: Vec<&str> = logged.lines().filter(|line| line.contains(STREAM_BODY_EVENT)).collect();
-
-        assert!(
-            !failure_lines.is_empty(),
-            "a mid-stream GET failure produced no `{STREAM_BODY_EVENT}` line at the default log level. \
-             This is the #4784 blind spot: the failure was only counted in a metric, or logged below \
-             `error` and filtered out. Captured log:\n{logged}"
-        );
-
-        // The identity is the whole point: a request_id alone cannot be resolved
-        // back to an object once the request is over.
-        assert!(
-            failure_lines.iter().any(|line| line.contains(key)),
-            "no `{STREAM_BODY_EVENT}` line named the failing object `{key}`, so the report is still \
-             unactionable. Lines seen:\n{}",
-            failure_lines.join("\n")
-        );
-        assert!(
-            failure_lines.iter().any(|line| line.contains(bucket)),
-            "no `{STREAM_BODY_EVENT}` line named the failing bucket `{bucket}`. Lines seen:\n{}",
-            failure_lines.join("\n")
-        );
-
-        info!(
-            "source-side evidence now present: {} stream-body failure line(s) naming the object",
-            failure_lines.len()
-        );
-        for line in &failure_lines {
-            info!("operator-visible evidence: {line}");
-        }
-
-        Ok(())
-    }
-}

crates/e2e_test/src/kms/kms_local_test.rs

@@ -647,7 +647,10 @@ async fn test_multipart_upload_with_sse_c(
 }
 
 /// Test large multipart upload to verify streaming encryption works correctly
-#[allow(dead_code)]
+#[allow(
+    dead_code,
+    reason = "parked behind the TODO in test_local_kms_multipart_upload until streaming encryption is fixed for large files (backlog#1823)"
+)]
 async fn test_large_multipart_upload(
     s3_client: &aws_sdk_s3::Client,
     bucket: &str,

crates/e2e_test/src/lib.rs

@@ -48,11 +48,6 @@ mod replacement_privileged_e2e_test;
 #[cfg(test)]
 mod degraded_read_eof_regression_test;
 
-// rustfs#4784: a mid-stream GET failure must be reportable from the source
-// server's log alone — naming the object, at the default log level.
-#[cfg(test)]
-mod get_stream_failure_observability_test;
-
 // backlog#1183: GET codec-streaming fast path must be byte/header identical to
 // the legacy duplex path before its rollout gates can be flipped on by default.
 #[cfg(test)]

crates/e2e_test/src/policy/test_runner.rs

@@ -19,32 +19,17 @@ use std::time::Instant;
 use tokio::time::{Duration, sleep};
 use tracing::{error, info};
 
-/// Core test categories
-#[derive(Debug, Clone, PartialEq, Eq)]
-pub enum TestCategory {
-    SingleValue,
-    MultiValue,
-    Concatenation,
-    Nested,
-    DenyScenarios,
-}
-
-impl TestCategory {}
-
 /// Test case definition
 #[derive(Debug, Clone)]
 pub struct TestDefinition {
     pub name: String,
-    #[allow(dead_code)]
-    pub category: TestCategory,
     pub is_critical: bool,
 }
 
 impl TestDefinition {
-    pub fn new(name: impl Into<String>, category: TestCategory, is_critical: bool) -> Self {
+    pub fn new(name: impl Into<String>, is_critical: bool) -> Self {
         Self {
             name: name.into(),
-            category,
             is_critical,
         }
     }
@@ -92,12 +77,12 @@ impl PolicyTestSuite {
     /// Create default test suite
     pub fn new() -> Self {
         let tests = vec![
-            TestDefinition::new("test_aws_policy_variables_single_value", TestCategory::SingleValue, true),
+            TestDefinition::new("test_aws_policy_variables_single_value", true),
-            TestDefinition::new("test_aws_policy_variables_multi_value", TestCategory::MultiValue, true),
+            TestDefinition::new("test_aws_policy_variables_multi_value", true),
-            TestDefinition::new("test_aws_policy_variables_concatenation", TestCategory::Concatenation, true),
+            TestDefinition::new("test_aws_policy_variables_concatenation", true),
-            TestDefinition::new("test_aws_policy_variables_nested", TestCategory::Nested, true),
+            TestDefinition::new("test_aws_policy_variables_nested", true),
-            TestDefinition::new("test_aws_policy_variables_deny", TestCategory::DenyScenarios, true),
+            TestDefinition::new("test_aws_policy_variables_deny", true),
-            TestDefinition::new("test_aws_policy_variables_sts", TestCategory::SingleValue, true),
+            TestDefinition::new("test_aws_policy_variables_sts", true),
         ];
 
         Self {

crates/ecstore/src/set_disk/core/io_primitives.rs

@@ -1095,6 +1095,14 @@ pub(in crate::set_disk) struct ReadRepairHealSubmission<'a> {
     pub(in crate::set_disk) set_index: usize,
     pub(in crate::set_disk) part_number: Option<usize>,
     pub(in crate::set_disk) reason: &'static str,
+    /// Durable MRF journal intent to file alongside the read-repair request
+    /// (backlog#1894 axis A): the intent kind plus its native `Uuid`
+    /// version id (the submission's string form stays display-only). Bound
+    /// to the reservation — the intent is only delivered when this sighting
+    /// wins the dedup TTL, so a burst of reads failing on the same object
+    /// books exactly one journal record instead of one per retry. `None`
+    /// keeps the historical no-intent behavior.
+    pub(in crate::set_disk) mrf_intent: Option<(rustfs_common::mrf_channel::MrfKind, Option<uuid::Uuid>)>,
 }
 
 pub(in crate::set_disk) fn send_read_repair_heal_request(
@@ -1126,6 +1134,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal(
             set_index,
             part_number,
             reason,
+            mrf_intent: None,
         },
         send_read_repair_heal_request,
     )
@@ -1144,6 +1153,7 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
         set_index,
         part_number,
         reason,
+        mrf_intent,
     } = submission;
 
     let Some(dedup_key) = reserve_read_repair_heal(bucket, object, version_id, pool_index, set_index).await else {
@@ -1155,6 +1165,12 @@ pub(in crate::set_disk) async fn submit_read_repair_heal_with_submitter(
         return;
     };
 
+    // Reservation won: this sighting owns the repair records for the object,
+    // including the durable journal intent when the caller asked for one.
+    if let Some((kind, version_uuid)) = mrf_intent {
+        rustfs_common::mrf_channel::try_send_mrf_intent(kind, bucket, object, version_uuid);
+    }
+
     let mut request = rustfs_common::heal_channel::create_heal_request_with_options(
         bucket.to_string(),
         Some(object.to_string()),
@@ -8710,6 +8726,42 @@ mod tests {
         assert_eq!(responses[0].error, Error::ErasureReadQuorum.to_string());
     }
 
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn mrf_intent_is_filed_once_per_read_repair_reservation() {
+        // Serial: owns the process-global MRF channel for this test binary
+        // (same key as the other channel-owning tests above).
+        let bucket = format!("mrf-intent-bucket-{}", Uuid::new_v4());
+        let object = format!("object-{}", Uuid::new_v4());
+        let mut receiver = rustfs_common::mrf_channel::init_mrf_channel().expect("first channel init in this binary");
+        rustfs_common::mrf_channel::set_mrf_delivery_enabled(true);
+
+        fn intent_submission<'a>(bucket: &'a str, object: &'a str) -> ReadRepairHealSubmission<'a> {
+            ReadRepairHealSubmission {
+                bucket,
+                object,
+                version_id: None,
+                pool_index: 9,
+                set_index: 9,
+                part_number: Some(1),
+                reason: "decode_error",
+                mrf_intent: Some((rustfs_common::mrf_channel::MrfKind::DecodeFailure, None)),
+            }
+        }
+
+        // First sighting wins the reservation: the journal intent is filed
+        // synchronously before the admission task is spawned.
+        submit_read_repair_heal_with_submitter(intent_submission(&bucket, &object), accepted_read_repair_submitter).await;
+        let first = receiver.try_recv().expect("first sighting must file exactly one MRF intent");
+        assert_eq!(*first.bucket, bucket);
+        assert_eq!(*first.object, object);
+
+        // Second sighting within the dedup TTL is a duplicate: no request, no
+        // second journal record.
+        submit_read_repair_heal_with_submitter(intent_submission(&bucket, &object), accepted_read_repair_submitter).await;
+        assert!(receiver.try_recv().is_err(), "duplicate sighting must not file another MRF intent");
+    }
+
     #[tokio::test]
     async fn reserve_read_repair_heal_dedupes_by_object_version_and_set() {
         let object = format!("object-{}", Uuid::new_v4());
@@ -8818,6 +8870,7 @@ mod tests {
                 set_index: 2,
                 part_number: Some(1),
                 reason: "test",
+                mrf_intent: None,
             },
             failed_read_repair_submitter,
         )
@@ -8846,6 +8899,7 @@ mod tests {
                 set_index: 3,
                 part_number: Some(2),
                 reason: "test",
+                mrf_intent: None,
             },
             dropped_read_repair_submitter,
         )
@@ -8874,6 +8928,7 @@ mod tests {
                 set_index: 4,
                 part_number: None,
                 reason: "test",
+                mrf_intent: None,
             },
             accepted_read_repair_submitter,
         )

crates/ecstore/src/set_disk/read.rs

@@ -1077,23 +1077,23 @@ 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
+                        // Single-flight (backlog#1894 axis A): the durable
-                        // request alive across restarts even when the in-memory
+                        // MRF intent (Urgent ECDecode across restarts, HS-01)
-                        // read-repair request is dropped or lost (HS-01).
+                        // is bound to the read-repair reservation, so only the
-                        rustfs_common::mrf_channel::try_send_mrf_intent(
+                        // first sighting within the dedup TTL books a journal
-                            rustfs_common::mrf_channel::MrfKind::DecodeFailure,
+                        // record instead of one per retried read.
-                            bucket,
+                        submit_read_repair_heal_with_submitter(
-                            object,
+                            ReadRepairHealSubmission {
-                            fi.version_id,
+                                bucket,
-                        );
+                                object,
-                        submit_read_repair_heal(
+                                version_id: version_id.as_deref(),
-                            bucket,
+                                pool_index,
-                            object,
+                                set_index,
-                            version_id.as_deref(),
+                                part_number: Some(part_number),
-                            pool_index,
+                                reason: "decode_error",
-                            set_index,
+                                mrf_intent: Some((rustfs_common::mrf_channel::MrfKind::DecodeFailure, fi.version_id)),
-                            Some(part_number),
+                            },
-                            "decode_error",
+                            send_read_repair_heal_request,
                         )
                         .await;
                         has_err = false;
@@ -2577,6 +2577,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             slow_read_repair_submitter,
         )
@@ -2611,6 +2612,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             dropped_read_repair_submitter,
         )
@@ -2647,6 +2649,7 @@ mod metadata_cache_tests {
                 set_index: 0,
                 part_number: Some(1),
                 reason: "missing_shards",
+                mrf_intent: None,
             },
             capture_read_repair_submitter,
         )

crates/heal/src/heal/manager.rs

@@ -597,14 +597,6 @@ impl PriorityHealQueue {
             }
         }
     }
-
-    /// Check if a request with the same key already exists in the queue
-    #[allow(dead_code)]
-    fn contains_key(&self, request: &HealRequest) -> bool {
-        let key = Self::make_dedup_key(request);
-        self.dedup_keys.contains_key(&key)
-    }
-
     /// Check if an erasure set heal request for a specific set_disk_id exists
     fn contains_erasure_set(&self, set_disk_id: &str) -> bool {
         let key = format!("erasure_set:{set_disk_id}");

crates/heal/src/heal/mrf_queue.rs

@@ -25,9 +25,12 @@
 //! 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
+//! own CRC32. Losing the last flush window (≤500 ms) is acceptable because
-//! duplicates are merged by the manager's dedup key, and read-repair remains
+//! every producer keeps its own safety net: read-repair re-detects on the
-//! the safety net.
+//! next failing read, and the scanner's corrupt-metadata branch leaves a
+//! pending-ledger entry behind even when its MRF intent is accepted
+//! (backlog#1894 axis A), so a lost intent is retried by the ledger rather
+//! than waiting for the failed-object TTL to re-scan the path.
 
 use super::{DiskStore, HealDiskExt as _, local_disk_map_read};
 use crate::heal::manager::HealManager;
@@ -409,8 +412,13 @@ impl MrfRuntime {
             let request = build_heal_request(&intent);
             match manager.submit_heal_request(request).await {
                 // Accepted intents leave the pending set; the next flush persists the
-                // smaller snapshot, which is the journal's compaction.
+                // smaller snapshot, which is the journal's compaction. Fan out a
-                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {}
+                // best-effort repaired notice so retry ledgers (the scanner's
+                // pending-heal oracle) can drop entries whose repair the manager
+                // now owns (backlog#1894 axis B).
+                Ok(HealAdmissionResult::Accepted) | Ok(HealAdmissionResult::Merged) => {
+                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts >= MRF_MAX_ATTEMPTS {
@@ -514,7 +522,9 @@ async fn replay_into(
         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::Accepted) | Ok(HealAdmissionResult::Merged) => {
+                    rustfs_common::mrf_channel::note_mrf_repaired(&intent.bucket, &intent.object, intent.version_id);
+                }
                 Ok(HealAdmissionResult::Full) | Ok(HealAdmissionResult::Dropped(HealAdmissionDropReason::QueueFull)) => {
                     intent.attempts = intent.attempts.saturating_add(1);
                     if intent.attempts < MRF_MAX_ATTEMPTS {

crates/heal/src/heal/storage.rs

@@ -42,7 +42,10 @@ pub struct HealLifecycleExpiryContext {
 
 enum HealLifecycleExpiryContextInner {
     Ecstore(EcstoreHealLifecycleExpiryContext),
-    #[allow(dead_code)]
+    #[allow(
+        dead_code,
+        reason = "constructed by the #[cfg(test)] `test()` helper; the lib target cannot see test-only consumers (backlog#1823)"
+    )]
     Test,
 }
 

crates/heal/tests/mrf_pipeline_test.rs

@@ -127,6 +127,20 @@ async fn decode_failure_intent_maps_to_urgent_mrf_heal_request() {
         "MRF intent must reach the manager queue as an Urgent request (snapshot: {:?})",
         manager.operations_snapshot().await
     );
+
+    // Axis B (backlog#1894): the accepted dispatch must also fan out a
+    // repaired notice for the intent's bucket so the scanner ledger can drop
+    // its retry entry for the same target. Polled: the queue observation
+    // above can land between the manager push and the consumer's notice.
+    let noticed = wait_until(Duration::from_secs(10), || async {
+        !mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty()
+    })
+    .await;
+    assert!(noticed, "accepted intent must fan out a repaired notice");
+    assert!(
+        mrf_channel::take_mrf_repaired_events_for("mrf-bucket").is_empty(),
+        "notice take is destructive"
+    );
 }
 
 /// A journal left behind by a previous process must be replayed into the

crates/madmin/src/service_commands.rs

@@ -19,7 +19,6 @@ use hyper::Uri;
 use crate::{trace::TraceType, utils::parse_duration};
 
 #[derive(Debug, Default)]
-#[allow(dead_code)]
 pub struct ServiceTraceOpts {
     s3: bool,
     internal: bool,
@@ -41,7 +40,6 @@ pub struct ServiceTraceOpts {
     threshold: Duration,
 }
 
-#[allow(dead_code)]
 impl ServiceTraceOpts {
     pub fn trace_types(&self) -> TraceType {
         let mut tt = TraceType::default();

crates/notify/examples/base.rs

@@ -15,7 +15,6 @@
 use std::io::IsTerminal;
 use tracing_subscriber::{EnvFilter, fmt, prelude::*, util::SubscriberInitExt};
 
-#[allow(dead_code)]
 fn main() {
     init_logger(LogLevel::Info);
     tracing::info!("Tracing logger initialized with Info level");

crates/s3select-query/src/sql/logical/optimizer.rs

@@ -46,15 +46,6 @@ pub struct DefaultLogicalOptimizer {
     analyzer: AnalyzerRef,
     rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>,
 }
-
-impl DefaultLogicalOptimizer {
-    #[allow(dead_code)]
-    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn OptimizerRule + Send + Sync>>) -> Self {
-        self.rules = rules;
-        self
-    }
-}
-
 impl Default for DefaultLogicalOptimizer {
     fn default() -> Self {
         let analyzer = Arc::new(DefaultAnalyzer::default());

crates/s3select-query/src/sql/physical/planner.rs

@@ -36,21 +36,9 @@ pub struct DefaultPhysicalPlanner {
     ext_physical_optimizer_rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>,
 }
 
-impl DefaultPhysicalPlanner {
+impl DefaultPhysicalPlanner {}
-    #[allow(dead_code)]
-    fn with_physical_transform_rules(mut self, rules: Vec<Arc<dyn ExtensionPlanner + Send + Sync>>) -> Self {
-        self.ext_physical_transform_rules = rules;
-        self
-    }
-}
 
-impl DefaultPhysicalPlanner {
+impl DefaultPhysicalPlanner {}
-    #[allow(dead_code)]
-    fn with_optimizer_rules(mut self, rules: Vec<Arc<dyn PhysicalOptimizerRule + Send + Sync>>) -> Self {
-        self.ext_physical_optimizer_rules = rules;
-        self
-    }
-}
 
 impl Default for DefaultPhysicalPlanner {
     fn default() -> Self {

crates/scanner/src/scanner_folder.rs

@@ -645,6 +645,32 @@ enum GetSizeFailureAction {
     HealMetadata { object: String },
 }
 
+/// How the corrupt-metadata branch records the repair after attempting an
+/// MRF intent (backlog#1894 axis A).
+#[derive(Debug, PartialEq, Eq)]
+enum CorruptMetadataRecording {
+    /// Intent accepted: the MRF consumer owns the repair (High Metadata
+    /// heal, durable after the journal's group-commit flush), so the
+    /// immediate heal request is skipped — the manager would otherwise book
+    /// two tasks for one target. A pending-ledger entry stays behind as the
+    /// backstop for what the journal cannot cover on its own (a crash inside
+    /// the flush window, or the consumer exhausting its admission attempts);
+    /// the repaired-notice fanout (axis B) drops the entry once the repair
+    /// lands.
+    LedgerOnly,
+    /// Intent rejected (feature disabled, channel uninitialized, or full):
+    /// the historical immediate heal request plus the ledger entry.
+    ImmediateAndLedger,
+}
+
+fn corrupt_metadata_recording(mrf_accepted: bool) -> CorruptMetadataRecording {
+    if mrf_accepted {
+        CorruptMetadataRecording::LedgerOnly
+    } else {
+        CorruptMetadataRecording::ImmediateAndLedger
+    }
+}
+
 fn build_bucket_heal_request(bucket: String, priority: HealChannelPriority) -> HealChannelRequest {
     HealChannelRequest {
         bucket,
@@ -700,6 +726,16 @@ fn pending_scanner_heal_identity(entry: &PendingScannerHeal) -> (u8, &str, Optio
     (kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref())
 }
 
+/// Decode an MRF repaired-notice version id for ledger matching. A nil UUID
+/// means "no value" per the repo-wide defensive-UUID invariant, so it maps
+/// to `None` and matches unversioned ledger entries only.
+fn mrf_repaired_version_id(version_id: Option<[u8; 16]>) -> Option<String> {
+    version_id
+        .map(uuid::Uuid::from_bytes)
+        .filter(|uuid| !uuid.is_nil())
+        .map(|uuid| uuid.to_string())
+}
+
 fn sort_pending_scanner_heals_for_retry(entries: &mut [PendingScannerHeal]) {
     entries.sort_by(|a, b| {
         a.last_attempt
@@ -1632,6 +1668,32 @@ impl FolderScanner {
         }
     }
 
+    /// Batched variant of [`Self::clear_pending_scanner_heal`] for repaired
+    /// notices (backlog#1894 axis B): one retain pass and one ledger sync
+    /// for the whole notice set, so a mass-recovery first sweep cannot turn
+    /// into thousands of full-table clones on the scan task. Only Object
+    /// entries match — bucket-level heals are never the MRF consumer's work.
+    fn clear_pending_scanner_heals_for_repaired(&mut self, events: &[rustfs_common::mrf_channel::MrfRepairedEvent]) {
+        // Pre-resolve the notice version strings once; each ledger entry then
+        // compares against plain Option<&str>.
+        let targets: Vec<(&str, &str, Option<String>)> = events
+            .iter()
+            .map(|event| (event.bucket.as_ref(), event.object.as_ref(), mrf_repaired_version_id(event.version_id)))
+            .collect();
+        let before = self.new_cache.info.pending_heals.len();
+        self.new_cache.info.pending_heals.retain(|entry| {
+            entry.kind != PendingScannerHealKind::Object
+                || !targets.iter().any(|(bucket, object, version)| {
+                    entry.bucket.as_str() == *bucket
+                        && entry.object.as_deref() == Some(*object)
+                        && entry.version_id.as_deref() == version.as_deref()
+                })
+        });
+        if self.new_cache.info.pending_heals.len() != before {
+            self.sync_pending_heals();
+        }
+    }
+
     fn record_pending_scanner_heal(
         &mut self,
         kind: PendingScannerHealKind,
@@ -1970,6 +2032,14 @@ impl FolderScanner {
         }
 
         let bucket = self.new_cache.info.name.clone();
+        // Backlog#1894 axis B: repairs the MRF consumer landed hand the
+        // manager the heal task, so the matching pending-ledger entries are
+        // retried nowhere — drop them here. Best-effort: a lost notice just
+        // leaves the entry to expire through its own attempts/age limits.
+        let repaired = rustfs_common::mrf_channel::take_mrf_repaired_events_for(&bucket);
+        if !repaired.is_empty() {
+            self.clear_pending_scanner_heals_for_repaired(&repaired);
+        }
         for pending in pending_scanner_heal_retry_candidates(&self.new_cache.info.pending_heals, &bucket) {
             if !self.should_heal().await {
                 break;
@@ -2434,29 +2504,46 @@ impl FolderScanner {
                         }
 
                         if let GetSizeFailureAction::HealMetadata { object } = failure_action {
-                            // MRF journal intent: durable High-priority Metadata
+                            // Single-flight (backlog#1894 axis A) — the
-                            // heal across restarts (HS-01); the scanner heal
+                            // recording mode and its guarantees are pinned by
-                            // request below stays as the immediate path.
+                            // corrupt_metadata_recording below.
-                            rustfs_common::mrf_channel::try_send_mrf_intent(
+                            let mrf_accepted = rustfs_common::mrf_channel::try_send_mrf_intent(
                                 rustfs_common::mrf_channel::MrfKind::MetadataCorruption,
                                 &item.bucket,
                                 &object,
                                 None,
                             );
-                            self.send_required_scanner_heal_request(
+                            match corrupt_metadata_recording(mrf_accepted) {
-                                PendingScannerHealKind::Object,
+                                CorruptMetadataRecording::LedgerOnly => {
-                                item.bucket.clone(),
+                                    // Recorded as Full (retry-later): admission
-                                Some(object.clone()),
+                                    // for this target happens in the MRF
-                                None,
+                                    // consumer, not in the manager's queue here.
-                                build_object_heal_request(
+                                    self.update_pending_scanner_heal_after_admission(
-                                    item.bucket.clone(),
+                                        PendingScannerHealKind::Object,
-                                    object.clone(),
+                                        &item.bucket,
-                                    None,
+                                        Some(&object),
-                                    self.scan_mode,
+                                        None,
-                                    HealChannelPriority::High,
+                                        self.scan_mode,
-                                ),
+                                        HealAdmissionResult::Full,
-                            )
+                                    );
-                            .await?;
+                                }
+                                CorruptMetadataRecording::ImmediateAndLedger => {
+                                    self.send_required_scanner_heal_request(
+                                        PendingScannerHealKind::Object,
+                                        item.bucket.clone(),
+                                        Some(object.clone()),
+                                        None,
+                                        build_object_heal_request(
+                                            item.bucket.clone(),
+                                            object.clone(),
+                                            None,
+                                            self.scan_mode,
+                                            HealChannelPriority::High,
+                                        ),
+                                    )
+                                    .await?;
+                                }
+                            }
                         }
 
                         timer.sleep().await;
@@ -3351,6 +3438,17 @@ mod tests {
         assert_eq!(EVENT_SCANNER_BIG_PREFIX, EventName::ScannerBigPrefix.to_string());
     }
 
+    /// Single-flight decision for the corrupt-metadata branch (backlog#1894
+    /// axis A): an accepted MRF intent must drop the immediate heal request
+    /// (the consumer files one; the manager would double-book) while a
+    /// rejected one must keep it — in both cases a ledger entry remains, so
+    /// the backstop survives regardless of delivery.
+    #[test]
+    fn corrupt_metadata_recording_maps_delivery_to_backstop() {
+        assert_eq!(corrupt_metadata_recording(true), CorruptMetadataRecording::LedgerOnly);
+        assert_eq!(corrupt_metadata_recording(false), CorruptMetadataRecording::ImmediateAndLedger);
+    }
+
     fn cooldown_map_len() -> usize {
         SCANNER_ALERT_EMISSION_COOLDOWN
             .lock()
@@ -4324,6 +4422,86 @@ mod tests {
         }
     }
 
+    /// The nil-UUID branch of the defensive-UUID invariant: a nil version in
+    /// a repaired notice means "no value" and must match unversioned ledger
+    /// entries only.
+    #[test]
+    fn test_mrf_repaired_version_id_maps_nil_to_none() {
+        assert_eq!(mrf_repaired_version_id(None), None);
+        assert_eq!(mrf_repaired_version_id(Some([0u8; 16])), None);
+        let uuid = Uuid::new_v4();
+        assert_eq!(mrf_repaired_version_id(Some(*uuid.as_bytes())), Some(uuid.to_string()));
+    }
+
+    /// Full wiring of backlog#1894 axis B: notes taken for the scanned bucket
+    /// clear exactly the matching Object ledger entries — bucket-level
+    /// entries, other buckets' entries, and version-mismatched entries
+    /// survive; a real (non-nil) version matches only the same version.
+    #[tokio::test]
+    async fn test_mrf_repaired_notices_clear_matching_ledger_entries() {
+        use rustfs_common::mrf_channel::note_mrf_repaired;
+
+        let (mut scanner, temp_dir) = build_test_scanner().await;
+        let _guard = TestGuard::new(u64::MAX, usize::MAX, &mut scanner, temp_dir);
+        scanner.new_cache.info.name = "bucket".to_string();
+        scanner.update_cache.info.name = "bucket".to_string();
+        scanner.heal_object_select = 1;
+
+        let version = Uuid::new_v4().to_string();
+        scanner.new_cache.info.pending_heals = vec![
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-a"), None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-b"), Some(&version), 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "bucket", Some("object-c"), None, 1, 1),
+            pending_heal(
+                PendingScannerHealKind::Object,
+                "bucket",
+                Some("object-c"),
+                Some("00000000-0000-0000-0000-000000000001"),
+                1,
+                1,
+            ),
+            pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 1),
+            pending_heal(PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None, 1, 1),
+        ];
+
+        note_mrf_repaired("bucket", "object-a", None);
+        note_mrf_repaired("bucket", "object-b", Some(*Uuid::parse_str(&version).unwrap().as_bytes()));
+        // A nil-UUID notice for object-c means "no value": it clears the
+        // unversioned entry but must not touch the versioned one.
+        note_mrf_repaired("bucket", "object-c", Some([0u8; 16]));
+        // A notice for a target the ledger does not track must be a no-op.
+        note_mrf_repaired("bucket", "object-untracked", None);
+
+        scanner
+            .retry_pending_scanner_heals()
+            .await
+            .expect("retry pass should succeed");
+
+        let survivors: Vec<(PendingScannerHealKind, &str, Option<&str>, Option<&str>)> = scanner
+            .new_cache
+            .info
+            .pending_heals
+            .iter()
+            .map(|entry| (entry.kind, entry.bucket.as_str(), entry.object.as_deref(), entry.version_id.as_deref()))
+            .collect();
+        // Cleared: object-a (no version), object-b (exact version match), and
+        // object-c's unversioned entry (the nil branch matched no-version
+        // only — the versioned object-c entry survives).
+        assert_eq!(
+            survivors,
+            vec![
+                (
+                    PendingScannerHealKind::Object,
+                    "bucket",
+                    Some("object-c"),
+                    Some("00000000-0000-0000-0000-000000000001")
+                ),
+                (PendingScannerHealKind::Bucket, "bucket", None, None),
+                (PendingScannerHealKind::Object, "other-bucket", Some("object-a"), None),
+            ]
+        );
+    }
+
     #[test]
     fn test_pending_heal_reconstructs_bucket_request() {
         let pending = pending_heal(PendingScannerHealKind::Bucket, "bucket", None, None, 1, 1);

crates/signer/src/request_signature_streaming.rs

@@ -22,7 +22,7 @@ use s3s::Body;
 
 const STREAMING_SIGN_ALGORITHM: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD";
 const STREAMING_SIGN_TRAILER_ALGORITHM: &str = "STREAMING-AWS4-HMAC-SHA256-PAYLOAD-TRAILER";
-const STREAMING_PAYLOAD_HDR: &str = "AWS4-HMAC-SHA256-PAYLOAD";
+const _STREAMING_PAYLOAD_HDR: &str = "AWS4-HMAC-SHA256-PAYLOAD";
 const _STREAMING_TRAILER_HDR: &str = "AWS4-HMAC-SHA256-TRAILER";
 const _PAYLOAD_CHUNK_SIZE: i64 = 64 * 1024;
 const _CHUNK_SIGCONST_LEN: i64 = 17;
@@ -51,15 +51,14 @@ fn streaming_fail(request: request::Request<Body>, error: SignV4Error) -> Stream
     Err(Box::new(StreamingSignFailure { request, error }))
 }
 
-#[allow(dead_code)]
+fn _try_build_chunk_string_to_sign(
-fn try_build_chunk_string_to_sign(
     t: OffsetDateTime,
     region: &str,
     previous_sig: &str,
     chunk_check_sum: &str,
 ) -> Result<String, SignV4Error> {
     let mut string_to_sign_parts = <Vec<String>>::new();
-    string_to_sign_parts.push(STREAMING_PAYLOAD_HDR.to_string());
+    string_to_sign_parts.push(_STREAMING_PAYLOAD_HDR.to_string());
     let format = format_description!("[year][month][day]T[hour][minute][second]Z");
     string_to_sign_parts.push(
         t.format(&format)
@@ -79,7 +78,7 @@ fn _try_build_chunk_signature(
     previous_signature: &str,
     secret_access_key: &str,
 ) -> Result<String, SignV4Error> {
-    let chunk_string_to_sign = try_build_chunk_string_to_sign(req_time, region, previous_signature, chunk_check_sum)?;
+    let chunk_string_to_sign = _try_build_chunk_string_to_sign(req_time, region, previous_signature, chunk_check_sum)?;
     let signing_key = get_signing_key(secret_access_key, region, req_time, SERVICE_TYPE_S3);
     Ok(get_signature(signing_key, &chunk_string_to_sign))
 }

rustfs/src/app/object_usecase.rs

@@ -1056,22 +1056,9 @@ pin_project! {
         remaining: usize,
         emitted: usize,
         expected: usize,
-        // Diagnostic-only identity for the body this stream is serving. Unset in
-        // unit tests that drive the stream over a bare reader; every production
-        // body carries it via `with_diagnostics`.
-        diagnostics: GetObjectReaderStreamDiagnostics,
     }
 }
 
-/// Object identity carried alongside a streaming GET body purely so a
-/// mid-stream failure names the object it happened on.
-#[derive(Clone, Default)]
-struct GetObjectReaderStreamDiagnostics {
-    bucket: String,
-    object: String,
-    request_id: String,
-}
-
 impl MemoryTrackedBytesStream {
     fn new(
         bytes: Bytes,
@@ -1120,19 +1107,8 @@ where
             remaining,
             emitted: 0,
             expected: remaining,
-            diagnostics: GetObjectReaderStreamDiagnostics::default(),
         }
     }
-
-    /// Attach the object identity a failed body should be reported against.
-    fn with_diagnostics(mut self, bucket: &str, object: &str, request_id: &str) -> Self {
-        self.diagnostics = GetObjectReaderStreamDiagnostics {
-            bucket: bucket.to_string(),
-            object: object.to_string(),
-            request_id: request_id.to_string(),
-        };
-        self
-    }
 }
 
 impl futures::Stream for MemoryTrackedBytesStream {
@@ -1593,29 +1569,12 @@ where
                     *this.emitted,
                     *this.remaining,
                 );
-                // The inner GetObjectStreamingReader is what normally reports a
+                #[cfg(feature = "tracing-chunk-debug")]
-                // short body, so reaching this arm means the reader signalled a
+                tracing::error!(
-                // clean EOF while this layer still owed bytes against an
-                // already-committed Content-Length. That disagreement is a data
-                // plane fault, not chunk noise: log it unconditionally so the
-                // truncated object is named in the operator's log rather than
-                // only in a metric counter (issue #4784).
-                error!(
-                    event = EVENT_GET_OBJECT_STREAM_BODY,
-                    component = LOG_COMPONENT_APP,
-                    subsystem = LOG_SUBSYSTEM_OBJECT,
-                    bucket = %this.diagnostics.bucket,
-                    object = %this.diagnostics.object,
-                    request_id = %this.diagnostics.request_id,
-                    size_bucket = get_object_stream_size_bucket(*this.expected),
-                    expected = *this.expected,
                     emitted = *this.emitted,
-                    remaining = *this.remaining,
+                    expected = *this.expected,
-                    strategy = this.strategy,
-                    buffer_source = this.buffer_source,
-                    state = "reader_stream_short_eof",
                     error = %err,
-                    "GetObject reader stream ended before the committed content length"
+                    "GetObject ReaderStream ended before expected length"
                 );
                 Poll::Ready(Some(Err(Box::new(err) as S3StdError)))
             }
@@ -1631,17 +1590,10 @@ where
                     *this.emitted,
                     *this.remaining,
                 );
-                // Deliberately not logged at warn here: every production body
-                // wraps a GetObjectStreamingReader, and that layer already
-                // reports this same error once with `state = "read_failed"` and
-                // the object identity. A second unconditional line per failed
-                // GET would read as two distinct faults. The chunk-debug build
-                // still gets this layer's view of the same error.
                 #[cfg(feature = "tracing-chunk-debug")]
                 tracing::error!(
                     emitted = *this.emitted,
                     expected = *this.expected,
-                    error_class = error_class,
                     error = %err,
                     "GetObject ReaderStream returned error"
                 );
@@ -1694,12 +1646,8 @@ where
 
 struct GetObjectStreamingReader<R> {
     inner: Option<R>,
-    // bucket/object + request_id + optional content_range are only used for diagnostic
+    // request_id + optional content_range are only used for diagnostic correlation and
-    // correlation and failure bucketing; they do not alter stream behavior. The object
+    // failure bucketing; they do not alter stream behavior.
-    // identity is what turns a mid-stream failure into an actionable report: a request_id
-    // alone cannot tell an operator which object reads short (issue #4784).
-    bucket: String,
-    object: String,
     request_id: String,
     content_range: Option<String>,
     expected: usize,
@@ -1718,8 +1666,8 @@ impl<R> GetObjectStreamingReader<R> {
     #[allow(clippy::too_many_arguments)]
     fn new(
         inner: R,
-        bucket: &str,
+        _bucket: &str,
-        key: &str,
+        _key: &str,
         request_id: &str,
         content_range: Option<String>,
         expected: usize,
@@ -1729,8 +1677,6 @@ impl<R> GetObjectStreamingReader<R> {
     ) -> Self {
         Self {
             inner: Some(inner),
-            bucket: bucket.to_string(),
-            object: key.to_string(),
             request_id: request_id.to_string(),
             content_range,
             expected,
@@ -1871,8 +1817,6 @@ impl<R> GetObjectStreamingReader<R> {
                 event = EVENT_GET_OBJECT_STREAM_BODY,
                 component = LOG_COMPONENT_APP,
                 subsystem = LOG_SUBSYSTEM_OBJECT,
-                bucket = %self.bucket,
-                object = %self.object,
                 request_id = %self.request_id,
                 range = %self.content_range.as_deref().unwrap_or("full"),
                 size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1909,8 +1853,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
-                            bucket = %self.bucket,
-                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1929,12 +1871,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                         self.timer = None;
                         let failure_reason = Self::classify_read_error(&error);
                         self.finish_err();
-                        error!(
+                        warn!(
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
-                            bucket = %self.bucket,
-                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -1976,8 +1916,6 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                                         event = EVENT_GET_OBJECT_STREAM_BODY,
                                         component = LOG_COMPONENT_APP,
                                         subsystem = LOG_SUBSYSTEM_OBJECT,
-                                        bucket = %self.bucket,
-                                        object = %self.object,
                                         request_id = %self.request_id,
                                         range = %self.content_range.as_deref().unwrap_or("full"),
                                         size_bucket = get_object_stream_size_bucket(self.expected),
@@ -2018,12 +1956,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                             self.begin_resume(error);
                             continue;
                         }
-                        error!(
+                        warn!(
                             event = EVENT_GET_OBJECT_STREAM_BODY,
                             component = LOG_COMPONENT_APP,
                             subsystem = LOG_SUBSYSTEM_OBJECT,
-                            bucket = %self.bucket,
-                            object = %self.object,
                             request_id = %self.request_id,
                             range = %self.content_range.as_deref().unwrap_or("full"),
                             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -2054,12 +1990,10 @@ impl<R: AsyncRead + Unpin> AsyncRead for GetObjectStreamingReader<R> {
                     let failure_reason = Self::classify_read_error(&err);
                     self.timer = None;
                     self.finish_err();
-                    error!(
+                    warn!(
                         event = EVENT_GET_OBJECT_STREAM_BODY,
                         component = LOG_COMPONENT_APP,
                         subsystem = LOG_SUBSYSTEM_OBJECT,
-                        bucket = %self.bucket,
-                        object = %self.object,
                         request_id = %self.request_id,
                         range = %self.content_range.as_deref().unwrap_or("full"),
                         size_bucket = get_object_stream_size_bucket(self.expected),
@@ -2095,8 +2029,6 @@ impl<R> Drop for GetObjectStreamingReader<R> {
             event = EVENT_GET_OBJECT_STREAM_BODY,
             component = LOG_COMPONENT_APP,
             subsystem = LOG_SUBSYSTEM_OBJECT,
-            bucket = %self.bucket,
-            object = %self.object,
             request_id = %self.request_id,
             range = %self.content_range.as_deref().unwrap_or("full"),
             size_bucket = get_object_stream_size_bucket(self.expected),
@@ -4371,8 +4303,7 @@ impl DefaultObjectUsecase {
             lifecycle,
             resume,
         );
-        let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source)
+        let stream = GetObjectReaderStream::new(reader, stream_buffer_size, expected, stream_strategy.as_str(), buffer_source);
-            .with_diagnostics(bucket, key, request_id);
         let blob = StreamingBlob::new(stream);
         if let Some(handoff_start) = handoff_start {
             rustfs_io_metrics::record_get_object_response_handoff(
@@ -16395,12 +16326,7 @@ mod tests {
         assert_eq!(body, vec![b'a'; 65]);
     }
 
-    // Serial with the capture test below: both drive the same short-EOF log
-    // callsite, and `tracing` caches callsite interest process-wide. Running
-    // this one concurrently on a thread with no subscriber re-caches that
-    // callsite as "never interested" and blinds the capture.
     #[tokio::test]
-    #[serial_test::serial]
     async fn get_object_reader_stream_errors_on_short_eof() {
         let stream = GetObjectReaderStream::new(
             std::io::Cursor::new(b"he".to_vec()),
@@ -16423,134 +16349,6 @@ mod tests {
         );
     }
 
-    /// Collects the structured fields of every event emitted while installed,
-    /// so a test can assert what an operator would actually read in the log
-    /// rather than only that an error value was returned.
-    type CapturedFieldMap = std::collections::HashMap<String, String>;
-    type CapturedEventLog = Arc<Mutex<Vec<CapturedFieldMap>>>;
-
-    struct CapturedEvents(CapturedEventLog);
-
-    struct CapturedFields(CapturedFieldMap);
-
-    impl tracing::field::Visit for CapturedFields {
-        fn record_debug(&mut self, field: &tracing::field::Field, value: &dyn std::fmt::Debug) {
-            self.0.insert(field.name().to_string(), format!("{value:?}"));
-        }
-
-        fn record_str(&mut self, field: &tracing::field::Field, value: &str) {
-            self.0.insert(field.name().to_string(), value.to_string());
-        }
-    }
-
-    impl<S: tracing::Subscriber> tracing_subscriber::Layer<S> for CapturedEvents {
-        fn on_event(&self, event: &tracing::Event<'_>, _ctx: tracing_subscriber::layer::Context<'_, S>) {
-            let mut fields = CapturedFields(CapturedFieldMap::new());
-            event.record(&mut fields);
-            self.0.lock().expect("captured events should not poison").push(fields.0);
-        }
-    }
-
-    fn capture_events() -> (CapturedEventLog, tracing::subscriber::DefaultGuard) {
-        use tracing_subscriber::{Registry, prelude::*};
-
-        let captured = Arc::new(Mutex::new(Vec::new()));
-        let subscriber = Registry::default().with(CapturedEvents(Arc::clone(&captured)));
-        let guard = tracing::subscriber::set_default(subscriber);
-        // `tracing` caches per-callsite interest process-wide, so a subscriber
-        // installed by a test running in parallel can leave the log sites below
-        // cached as "never interested" and this capture would silently see
-        // nothing. Force the callsites to re-ask the subscriber we just
-        // installed.
-        tracing::callsite::rebuild_interest_cache();
-        (captured, guard)
-    }
-
-    fn find_stream_body_event(captured: &CapturedEventLog, state: &str) -> CapturedFieldMap {
-        let events = captured.lock().expect("captured events should not poison");
-        events
-            .iter()
-            .find(|fields| fields.get("state").is_some_and(|value| value == state))
-            .unwrap_or_else(|| {
-                panic!(
-                    "a `{state}` streaming body failure must be logged, not only counted in a metric. \
-                     Captured {} event(s): {:?}",
-                    events.len(),
-                    events
-                )
-            })
-            .clone()
-    }
-
-    /// rustfs#4784: a GET body that ends short of its committed Content-Length
-    /// is the fault that breaks every downstream copier (replication, site
-    /// replication, `rclone sync`), yet this layer only fed a metric counter —
-    /// its log line was compiled out unless the `tracing-chunk-debug` feature
-    /// was on, so operators saw nothing on the source side.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn get_object_reader_stream_short_eof_names_the_object() {
-        let (captured, _guard) = capture_events();
-
-        let stream = GetObjectReaderStream::new(
-            std::io::Cursor::new(b"he".to_vec()),
-            64,
-            5,
-            GetObjectStreamStrategy::Standard.as_str(),
-            GET_READER_STREAM_BUFFER_SOURCE_SELECTED,
-        )
-        .with_diagnostics("restic-paperless", "index/41b5a4c2344edb90", "req-reader-stream-short-eof");
-
-        stream
-            .collect::<Vec<_>>()
-            .await
-            .into_iter()
-            .collect::<Result<Vec<_>, _>>()
-            .expect_err("short reader should fail the streaming body");
-
-        let event = find_stream_body_event(&captured, "reader_stream_short_eof");
-        assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless"));
-        assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90"));
-        assert_eq!(event.get("request_id").map(String::as_str), Some("req-reader-stream-short-eof"));
-        assert_eq!(event.get("expected").map(String::as_str), Some("5"));
-        assert_eq!(event.get("emitted").map(String::as_str), Some("2"));
-        assert_eq!(event.get("remaining").map(String::as_str), Some("3"));
-    }
-
-    /// The inner reader already logged mid-stream failures, but only under a
-    /// request_id — which cannot be resolved back to an object once the request
-    /// is gone. Without the identity the report in #4784 was unactionable.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn get_object_streaming_reader_short_eof_names_the_object() {
-        use tokio::io::AsyncReadExt;
-
-        let (captured, _guard) = capture_events();
-
-        let mut reader = GetObjectStreamingReader::new(
-            std::io::Cursor::new(b"short".to_vec()),
-            "restic-paperless",
-            "index/41b5a4c2344edb90",
-            "req-streaming-short-eof",
-            None,
-            10,
-            Duration::ZERO,
-            GetObjectBodyLifecycle::tracked(GetObjectGuard::new()),
-            None,
-        );
-
-        let mut out = Vec::new();
-        reader
-            .read_to_end(&mut out)
-            .await
-            .expect_err("short body under a larger Content-Length must fail the stream");
-
-        let event = find_stream_body_event(&captured, "short_eof");
-        assert_eq!(event.get("bucket").map(String::as_str), Some("restic-paperless"));
-        assert_eq!(event.get("object").map(String::as_str), Some("index/41b5a4c2344edb90"));
-        assert_eq!(event.get("request_id").map(String::as_str), Some("req-streaming-short-eof"));
-    }
-
     #[test]
     fn get_object_stream_failure_labels_are_low_cardinality() {
         assert_eq!(get_object_stream_failure_reason("short_eof"), GET_STREAMING_BODY_FAILURE_REASON_SHORT_EOF);