chore(ecstore): integrate batch diagnostics for request stage work

Retain the reviewed diagnostic dependency while starting request-stage work
from current main. This integration is not a performance qualification.

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

crates/config/src/constants/app.rs

@@ -374,6 +374,9 @@ pub const DEFAULT_OBS_PROFILING_EXPORT_ENABLED: bool = true;
 /// Environment variable: RUSTFS_OBS_LOG_STDOUT_ENABLED
 pub const DEFAULT_OBS_LOG_STDOUT_ENABLED: bool = false;
 
+/// Fixed target for sampled local-batch diagnostics with isolated JSON output.
+pub const LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET: &str = "rustfs_local_batch_diagnostic";
+
 /// Constant representing 1 Kibibyte (1024 bytes)
 /// Default value: 1024
 pub const KI_B: usize = 1024;

crates/ecstore/Cargo.toml

@@ -32,6 +32,8 @@ workspace = true
 [features]
 default = []
 diagnostic-read-version-caller = ["rustfs-io-metrics/diagnostic-read-version-caller"]
+# Sampled wall/CPU diagnostics only; does not change the read backend.
+diagnostic-local-batch = ["rustix/time"]
 gcs = ["dep:google-cloud-storage", "dep:google-cloud-auth"]
 rio-v2 = ["dep:rustfs-rio-v2"]
 hotpath = [

crates/ecstore/src/api/mod.rs

@@ -14,12 +14,16 @@
 
 //! Explicit ECStore public facades for outer crate compatibility boundaries.
 
-#[cfg(feature = "diagnostic-read-version-caller")]
+#[cfg(any(feature = "diagnostic-read-version-caller", feature = "diagnostic-local-batch"))]
 pub mod diagnostic {
+    #[cfg(feature = "diagnostic-read-version-caller")]
     pub use crate::cluster::rpc::runtime_sources::{
         current_read_version_entry_source, observe_read_version_context_boundary, with_captured_read_version_context,
         with_read_version_entry, with_read_version_spawn_context,
     };
+    #[cfg(feature = "diagnostic-local-batch")]
+    pub use crate::diagnostics::batch_read_diagnostics::{current_request_id, in_request_context, request_context};
+    #[cfg(feature = "diagnostic-read-version-caller")]
     pub use rustfs_io_metrics::internode_metrics::{ReadVersionContextBoundary, ReadVersionEntrySource};
 }
 

crates/ecstore/src/diagnostics/batch_read_diagnostics.rs

@@ -0,0 +1,568 @@
+// 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.
+
+//! Sampled local batch diagnostics for a separate, default-off build.
+//!
+//! Event fields contain server-owned correlation IDs, sizes and timing, never
+//! object/disk paths or I/O error strings. They are not performance acceptance.
+//! Linux supplies thread CPU time; other targets report it absent.
+
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::{Arc, OnceLock};
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
+use tracing::{Dispatch, Level};
+
+const LOG_COMPONENT_ECSTORE: &str = "ecstore";
+const LOG_SUBSYSTEM_LOCAL_BATCH: &str = "local_batch_diagnostic";
+const EVENT_LOCAL_BATCH_DIAGNOSTIC: &str = "local_batch_diagnostic";
+const TARGET: &str = rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET;
+const ENV_SAMPLE_EVERY: &str = "RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY";
+const MAX_SAMPLE_EVERY: u64 = 65_536;
+static SAMPLE_EVERY: OnceLock<u64> = OnceLock::new();
+static NEXT_BATCH: AtomicU64 = AtomicU64::new(0);
+
+tokio::task_local! {
+    static REQUEST_ID: Option<Arc<str>>;
+}
+
+fn parse_rate(value: Option<&str>) -> u64 {
+    value
+        .and_then(|value| value.parse::<u64>().ok())
+        .filter(|rate| (1..=MAX_SAMPLE_EVERY).contains(rate))
+        .unwrap_or(0)
+}
+
+fn enabled_rate() -> u64 {
+    *SAMPLE_EVERY.get_or_init(|| parse_rate(std::env::var(ENV_SAMPLE_EVERY).ok().as_deref()))
+}
+
+fn recording_enabled() -> bool {
+    enabled_rate() != 0 && tracing::enabled!(target: TARGET, Level::TRACE)
+}
+
+fn valid_request_id(id: &str) -> bool {
+    !id.is_empty()
+        && id.len() <= 128
+        && id
+            .bytes()
+            .all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_'))
+}
+
+/// Retain a bounded server-owned canonical ID only while diagnostics are enabled.
+///
+/// The external S3 boundary must supply its canonical ID, not a client header.
+pub fn request_context(id: &str) -> Option<Arc<str>> {
+    (recording_enabled() && valid_request_id(id)).then(|| Arc::from(id))
+}
+
+/// Poll a request under its diagnostic ID without retaining request fields.
+pub fn in_request_context<T>(id: Option<&Arc<str>>, operation: impl FnOnce() -> T) -> T {
+    REQUEST_ID.sync_scope(id.cloned(), operation)
+}
+
+/// Return the current ID for explicit diagnostic propagation checks.
+pub fn current_request_id() -> Option<Arc<str>> {
+    REQUEST_ID.try_with(Clone::clone).ok().flatten()
+}
+
+/// Capture before spawning the legacy producer; task locals are not inherited.
+pub(crate) fn with_captured_request_context<F: std::future::Future>(future: F) -> impl std::future::Future<Output = F::Output> {
+    REQUEST_ID.scope(current_request_id(), future)
+}
+
+#[derive(Clone, Copy)]
+pub(crate) enum Stage {
+    Execution,
+    Metadata,
+    Open,
+    Allocation,
+    Read,
+}
+
+impl Stage {
+    fn as_str(self) -> &'static str {
+        match self {
+            Self::Execution => "execution",
+            Self::Metadata => "metadata",
+            Self::Open => "open",
+            Self::Allocation => "allocation",
+            Self::Read => "read",
+        }
+    }
+}
+
+fn unix_ns() -> Option<u64> {
+    SystemTime::now().duration_since(UNIX_EPOCH).ok()?.as_nanos().try_into().ok()
+}
+
+#[cfg(target_os = "linux")]
+fn cpu_ns() -> Option<u64> {
+    use rustix::time::{ClockId, DynamicClockId, clock_gettime_dynamic};
+    let time = clock_gettime_dynamic(DynamicClockId::Known(ClockId::ThreadCPUTime)).ok()?;
+    let seconds = u64::try_from(time.tv_sec).ok()?;
+    let nanos = u64::try_from(time.tv_nsec).ok().filter(|value| *value < 1_000_000_000)?;
+    seconds.checked_mul(1_000_000_000)?.checked_add(nanos)
+}
+
+#[cfg(not(target_os = "linux"))]
+fn cpu_ns() -> Option<u64> {
+    None
+}
+
+fn thread_id() -> Option<i32> {
+    #[cfg(target_os = "linux")]
+    {
+        thread_local! { static TID: i32 = rustix::thread::gettid().as_raw_nonzero().get(); }
+        Some(TID.with(|id| *id))
+    }
+    #[cfg(not(target_os = "linux"))]
+    {
+        None
+    }
+}
+
+pub(crate) struct Started {
+    wall: Instant,
+    unix_ns: Option<u64>,
+    cpu_ns: Option<u64>,
+}
+struct Timing {
+    unix_ns: Option<u64>,
+    wall_ns: Option<u64>,
+    cpu_ns: Option<u64>,
+    cpu_status: &'static str,
+}
+
+pub(crate) struct Sample {
+    batch_id: u64,
+    request_id: Option<Arc<str>>,
+    submitted: Instant,
+    submitted_unix_ns: Option<u64>,
+    dispatch: Dispatch,
+}
+
+impl Sample {
+    pub(crate) fn capture() -> Option<Self> {
+        Self::capture_with(enabled_rate(), tracing::enabled!(target: TARGET, Level::TRACE), &NEXT_BATCH)
+    }
+
+    fn capture_with(rate: u64, enabled: bool, sequence: &AtomicU64) -> Option<Self> {
+        if rate == 0 || !enabled {
+            return None;
+        }
+        let batch_id = sequence.fetch_add(1, Ordering::Relaxed);
+        if !batch_id.is_multiple_of(rate) {
+            return None;
+        }
+        Some(Self {
+            batch_id,
+            request_id: current_request_id(),
+            submitted: Instant::now(),
+            submitted_unix_ns: unix_ns(),
+            dispatch: tracing::dispatcher::get_default(Clone::clone),
+        })
+    }
+
+    pub(crate) fn dispatch(&self) -> &Dispatch {
+        &self.dispatch
+    }
+
+    pub(crate) fn start(&self) -> Started {
+        Started {
+            wall: Instant::now(),
+            unix_ns: unix_ns(),
+            cpu_ns: cpu_ns(),
+        }
+    }
+
+    pub(crate) fn worker_started(&self) {
+        self.emit(
+            "queue_wait",
+            None,
+            true,
+            Timing {
+                unix_ns: self.submitted_unix_ns,
+                wall_ns: self.submitted.elapsed().as_nanos().try_into().ok(),
+                cpu_ns: None,
+                cpu_status: "not_applicable",
+            },
+        );
+    }
+
+    pub(crate) fn finish(&self, stage: Stage, shard: Option<usize>, start: Started, success: bool) {
+        let cpu = start
+            .cpu_ns
+            .zip(cpu_ns())
+            .and_then(|(before, after)| after.checked_sub(before));
+        self.emit(
+            stage.as_str(),
+            shard,
+            success,
+            Timing {
+                unix_ns: start.unix_ns,
+                wall_ns: start.wall.elapsed().as_nanos().try_into().ok(),
+                cpu_ns: cpu,
+                cpu_status: if cpu.is_some() { "available" } else { "unavailable" },
+            },
+        );
+    }
+
+    fn emit(&self, stage: &'static str, shard: Option<usize>, success: bool, timing: Timing) {
+        tracing::trace!(target: TARGET, parent: None,
+            event = EVENT_LOCAL_BATCH_DIAGNOSTIC, component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_LOCAL_BATCH, result = if success { "success" } else { "error" },
+            stage, batch_id = self.batch_id, request_id = self.request_id.as_deref(),
+            request_context = if self.request_id.is_some() { "present" } else { "missing" },
+            pid = std::process::id(), tid = thread_id(), shard_index = shard,
+            submitted_unix_ns = self.submitted_unix_ns, start_unix_ns = timing.unix_ns,
+            wall_ns = timing.wall_ns, cpu_ns = timing.cpu_ns, cpu_clock_status = timing.cpu_status,
+            "Sampled local batch stage");
+    }
+
+    pub(crate) fn shard_result(&self, shard: usize, requested: usize, read: usize, success: bool) {
+        tracing::trace!(target: TARGET, parent: None,
+            event = EVENT_LOCAL_BATCH_DIAGNOSTIC, component = LOG_COMPONENT_ECSTORE,
+            subsystem = LOG_SUBSYSTEM_LOCAL_BATCH, result = if success { "success" } else { "error" },
+            stage = "shard_result", batch_id = self.batch_id, request_id = self.request_id.as_deref(),
+            request_context = if self.request_id.is_some() { "present" } else { "missing" },
+            pid = std::process::id(), tid = thread_id(), submitted_unix_ns = self.submitted_unix_ns,
+            shard_index = shard, requested_bytes = requested, read_bytes = read,
+            "Sampled local batch result");
+    }
+}
+
+pub(crate) fn measure<T, E>(
+    sample: Option<&Sample>,
+    stage: Stage,
+    shard: usize,
+    operation: impl FnOnce() -> Result<T, E>,
+) -> Result<T, E> {
+    let Some(sample) = sample else { return operation() };
+    let start = sample.start();
+    let result = operation();
+    sample.finish(stage, Some(shard), start, result.is_ok());
+    result
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::{Value, json};
+    use std::collections::BTreeMap;
+    use std::sync::{Barrier, Mutex, mpsc};
+    use tracing::field::{Field, Visit};
+    use tracing_subscriber::{Layer, layer::Context, prelude::*};
+
+    #[derive(Default)]
+    struct Values(BTreeMap<String, Value>);
+    impl Visit for Values {
+        fn record_u64(&mut self, field: &Field, value: u64) {
+            self.0.insert(field.name().into(), json!(value));
+        }
+        fn record_i64(&mut self, field: &Field, value: i64) {
+            self.0.insert(field.name().into(), json!(value));
+        }
+        fn record_bool(&mut self, field: &Field, value: bool) {
+            self.0.insert(field.name().into(), json!(value));
+        }
+        fn record_str(&mut self, field: &Field, value: &str) {
+            self.0.insert(field.name().into(), json!(value));
+        }
+        fn record_debug(&mut self, field: &Field, value: &dyn std::fmt::Debug) {
+            self.0.insert(field.name().into(), json!(format!("{value:?}")));
+        }
+    }
+
+    struct Records {
+        events: Mutex<Vec<BTreeMap<String, Value>>>,
+        block_queue: std::sync::atomic::AtomicBool,
+        queue_started: Mutex<Option<mpsc::Sender<()>>>,
+        release: Barrier,
+    }
+
+    struct Recorder(Arc<Records>);
+    impl<S: tracing::Subscriber> Layer<S> for Recorder {
+        fn on_event(&self, event: &tracing::Event<'_>, _: Context<'_, S>) {
+            if event.metadata().target() != TARGET {
+                return;
+            }
+            assert!(event.is_root(), "diagnostic events must not inherit URI-bearing spans");
+            let mut fields = Values::default();
+            event.record(&mut fields);
+            let queue = fields.0.get("stage") == Some(&json!("queue_wait"));
+            self.0.events.lock().expect("record diagnostic event").push(fields.0);
+            if queue && self.0.block_queue.swap(false, Ordering::SeqCst) {
+                if let Some(sender) = self.0.queue_started.lock().expect("queue sender").take() {
+                    sender.send(()).expect("signal worker start");
+                }
+                self.0.release.wait();
+            }
+        }
+    }
+
+    fn records() -> Arc<Records> {
+        Arc::new(Records {
+            events: Mutex::new(Vec::new()),
+            block_queue: false.into(),
+            queue_started: Mutex::new(None),
+            release: Barrier::new(2),
+        })
+    }
+
+    #[test]
+    fn rate_is_default_off_and_bounded() {
+        for value in [None, Some("0"), Some("invalid"), Some("-1"), Some("65537")] {
+            assert_eq!(parse_rate(value), 0);
+        }
+        assert_eq!(parse_rate(Some("1")), 1);
+        assert_eq!(parse_rate(Some("64")), 64);
+        assert_eq!(parse_rate(Some("65536")), MAX_SAMPLE_EVERY);
+    }
+
+    #[test]
+    fn disabled_or_filtered_capture_does_not_advance_sampling() {
+        let sequence = AtomicU64::new(0);
+        assert!(Sample::capture_with(0, true, &sequence).is_none());
+        assert!(Sample::capture_with(64, false, &sequence).is_none());
+        assert_eq!(sequence.load(Ordering::Relaxed), 0);
+        let result = measure::<(), _>(None, Stage::Read, 0, || Err("original error"));
+        assert_eq!(result, Err("original error"));
+    }
+
+    #[test]
+    fn sampling_keeps_the_registered_interval() {
+        let sequence = AtomicU64::new(0);
+        let selected: Vec<_> = (0..12)
+            .filter_map(|_| Sample::capture_with(4, true, &sequence).map(|sample| sample.batch_id))
+            .collect();
+        assert_eq!(selected, [0, 4, 8]);
+    }
+
+    #[test]
+    fn request_context_is_bounded_and_restored_on_unwind() {
+        assert!(valid_request_id("server-owned-uuid_123"));
+        assert!(valid_request_id(&"a".repeat(128)));
+        for value in ["", "secret/path", "line\nbreak", &"a".repeat(129)] {
+            assert!(!valid_request_id(value));
+        }
+        let outer: Arc<str> = Arc::from("server-outer");
+        in_request_context(Some(&outer), || {
+            in_request_context(None, || assert!(current_request_id().is_none()));
+            assert_eq!(current_request_id().as_deref(), Some("server-outer"));
+            let nested: Arc<str> = Arc::from("server-inner");
+            let result = std::panic::catch_unwind(|| in_request_context(Some(&nested), || panic!("test unwind")));
+            assert!(result.is_err());
+            assert_eq!(current_request_id().as_deref(), Some("server-outer"));
+        });
+        assert!(current_request_id().is_none());
+    }
+
+    #[tokio::test]
+    async fn captured_request_context_crosses_spawn_without_leaking_to_caller() {
+        let original: Arc<str> = Arc::from("server-detached-producer");
+        let task = in_request_context(Some(&original), || {
+            tokio::spawn(with_captured_request_context(async {
+                tokio::task::yield_now().await;
+                current_request_id()
+            }))
+        });
+        assert!(current_request_id().is_none());
+        assert_eq!(task.await.expect("detached diagnostic context").as_deref(), Some(original.as_ref()));
+        assert!(current_request_id().is_none());
+    }
+
+    #[test]
+    fn failed_partial_operation_preserves_error_and_initialized_byte_count() {
+        let _interest = crate::test_tracing::pin_callsite_interest_for_test();
+        let state = records();
+        let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state)));
+        tracing::subscriber::with_default(subscriber, || {
+            let parent = tracing::info_span!("request", uri = "/must-not-copy-parent-path");
+            let _entered = parent.enter();
+            let sample = Sample::capture_with(1, true, &AtomicU64::new(0)).expect("sampled diagnostic");
+            let mut read = 0;
+            let result = measure::<(), _>(Some(&sample), Stage::Read, 0, || {
+                read = 2;
+                Err("partial read failure")
+            });
+            assert_eq!(result, Err("partial read failure"));
+            sample.shard_result(0, 4, read, false);
+        });
+        let events = state.events.lock().expect("read events");
+        assert!(
+            events
+                .iter()
+                .any(|e| e.get("stage") == Some(&json!("read")) && e.get("result") == Some(&json!("error")))
+        );
+        assert!(
+            events
+                .iter()
+                .any(|e| e.get("read_bytes") == Some(&json!(2)) && e.get("requested_bytes") == Some(&json!(4)))
+        );
+    }
+
+    #[test]
+    fn unavailable_cpu_is_absent_and_not_a_zero_measurement() {
+        let _interest = crate::test_tracing::pin_callsite_interest_for_test();
+        let state = records();
+        let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state)));
+        tracing::subscriber::with_default(subscriber, || {
+            let sample = Sample::capture_with(1, true, &AtomicU64::new(0)).expect("diagnostic sample");
+            sample.emit(
+                "metadata",
+                Some(0),
+                true,
+                Timing {
+                    unix_ns: None,
+                    wall_ns: Some(1),
+                    cpu_ns: None,
+                    cpu_status: "unavailable",
+                },
+            );
+            sample.emit(
+                "metadata",
+                Some(0),
+                true,
+                Timing {
+                    unix_ns: None,
+                    wall_ns: Some(1),
+                    cpu_ns: Some(0),
+                    cpu_status: "available",
+                },
+            );
+        });
+        let events = state.events.lock().expect("CPU availability events");
+        assert!(!events[0].contains_key("cpu_ns"));
+        assert_eq!(events[0]["cpu_clock_status"], "unavailable");
+        assert_eq!(events[1]["cpu_ns"], 0);
+        assert_eq!(events[1]["cpu_clock_status"], "available");
+    }
+
+    #[cfg(target_os = "linux")]
+    #[test]
+    fn thread_cpu_clock_is_available_without_zero_substitution() {
+        let before = cpu_ns().expect("Linux thread CPU clock available");
+        for value in 0..10_000 {
+            std::hint::black_box(value);
+        }
+        let after = cpu_ns().expect("Linux thread CPU clock remains available");
+        assert!(after >= before);
+    }
+
+    #[cfg(unix)]
+    #[test]
+    fn production_batch_diagnostic_survives_caller_cancellation() {
+        const CHILD: &str = "RUSTFS_TEST_BATCH_DIAGNOSTIC_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let status = std::process::Command::new(std::env::current_exe().expect("current test executable"))
+                .args([
+                    "--exact",
+                    "diagnostics::batch_read_diagnostics::tests::production_batch_diagnostic_survives_caller_cancellation",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .env(ENV_SAMPLE_EVERY, "1")
+                .status()
+                .expect("run isolated diagnostic test");
+            assert!(status.success(), "isolated production diagnostic failed");
+            return;
+        }
+        let _interest = crate::test_tracing::pin_callsite_interest_for_test();
+        let state = records();
+        let recorder = crate::test_metrics::CapturingRecorder::default();
+        rustfs_io_metrics::set_get_stage_metrics_enabled(true);
+        let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&state)));
+        metrics::with_local_recorder(&recorder, || {
+            tracing::subscriber::with_default(subscriber, || {
+                let runtime = tokio::runtime::Builder::new_current_thread()
+                    .enable_all()
+                    .build()
+                    .expect("test runtime");
+                runtime.block_on(async {
+                    let directory = tempfile::tempdir().expect("diagnostic test directory");
+                    let path = directory.path().join("must-not-appear-in-trace");
+                    std::fs::write(&path, b"abcdef").expect("write payload");
+                    let results = REQUEST_ID
+                        .scope(
+                            Some(Arc::from("server-original")),
+                            crate::disk::local::batch_shard_pread(vec![
+                                (path.clone(), 1, 3),
+                                (directory.path().join("missing-secret-path"), 0, 3),
+                                (path.clone(), 0, 0),
+                                (path.clone(), 7, 1),
+                            ]),
+                        )
+                        .await;
+                    assert_eq!(results[0].as_ref().expect("positioned read").as_ref(), b"bcd");
+                    assert!(results[1].is_err());
+                    assert!(results[2].as_ref().expect("zero-length read").is_empty());
+                    assert!(results[3].is_err());
+                    let counter = "rustfs_io_get_object_local_batch_shards_total";
+                    assert_eq!(recorder.counter_value(counter, &[("outcome", "success")]), 2);
+                    assert_eq!(recorder.counter_value(counter, &[("outcome", "error")]), 2);
+                    assert_eq!(recorder.counter_value("rustfs_io_get_object_local_batch_read_bytes_total", &[]), 3);
+                    {
+                        let events = state.events.lock().expect("normal events");
+                        let indices: Vec<_> = events
+                            .iter()
+                            .filter(|event| event.get("stage") == Some(&json!("shard_result")))
+                            .map(|event| event["shard_index"].as_u64().expect("shard index"))
+                            .collect();
+                        assert_eq!(indices, [0, 1, 2, 3], "errors and empty reads retain their input index");
+                        for (index, result) in [(0, "success"), (1, "error"), (2, "success"), (3, "error")] {
+                            assert!(
+                                events.iter().any(|event| {
+                                    event.get("stage") == Some(&json!("shard_result"))
+                                        && event.get("shard_index") == Some(&json!(index))
+                                        && event.get("result") == Some(&json!(result))
+                                }),
+                                "missing diagnostic result for shard {index}"
+                            );
+                        }
+                        assert!(
+                            !events
+                                .iter()
+                                .any(|e| e.get("stage") == Some(&json!("read")) && e.get("shard_index") == Some(&json!(2)))
+                        );
+                        assert!(!format!("{events:?}").contains("must-not-appear"));
+                        assert!(!format!("{events:?}").contains("missing-secret"));
+                    }
+                    let (sender, receiver) = mpsc::channel();
+                    *state.queue_started.lock().expect("install queue signal") = Some(sender);
+                    state.block_queue.store(true, Ordering::SeqCst);
+                    let mut future = Box::pin(REQUEST_ID.scope(
+                        Some(Arc::from("server-cancelled")),
+                        crate::disk::local::batch_shard_pread(vec![(path, 0, 6)]),
+                    ));
+                    assert!(futures::poll!(future.as_mut()).is_pending());
+                    receiver
+                        .recv_timeout(std::time::Duration::from_secs(5))
+                        .expect("blocking worker started");
+                    drop(future);
+                    state.release.wait();
+                    let deadline = Instant::now() + std::time::Duration::from_secs(5);
+                    loop {
+                        if state.events.lock().expect("completion events").iter().any(|e| {
+                            e.get("stage") == Some(&json!("execution")) && e.get("request_id") == Some(&json!("server-cancelled"))
+                        }) {
+                            break;
+                        }
+                        assert!(Instant::now() < deadline, "detached worker must complete diagnostic");
+                        tokio::time::sleep(std::time::Duration::from_millis(1)).await;
+                    }
+                });
+            })
+        });
+    }
+}

crates/ecstore/src/diagnostics/mod.rs

@@ -15,5 +15,7 @@
 // #730: diagnostics constants are staged for request-path telemetry migration.
 
 pub(crate) mod admin_server_info;
+#[cfg(feature = "diagnostic-local-batch")]
+pub(crate) mod batch_read_diagnostics;
 pub(crate) mod get;
 pub(crate) mod object_lock;

crates/ecstore/src/disk/local.rs

@@ -11328,10 +11328,39 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
     let n = requests.len();
     let metrics = rustfs_io_metrics::LocalBatchReadMetrics::capture();
     let submitted = metrics.as_ref().map(|_| std::time::Instant::now());
+    #[cfg(feature = "diagnostic-local-batch")]
+    let diagnostic = crate::diagnostics::batch_read_diagnostics::Sample::capture();
     tokio::task::spawn_blocking(move || {
         use rustfs_io_metrics::LocalBatchReadStage;
         use std::os::unix::fs::FileExt;
 
+        // The disabled build expands directly to the original operation.
+        macro_rules! diagnostic_stage {
+            ($stage:ident, $index:expr, $operation:expr) => {{
+                #[cfg(feature = "diagnostic-local-batch")]
+                {
+                    crate::diagnostics::batch_read_diagnostics::measure(
+                        diagnostic.as_ref(),
+                        crate::diagnostics::batch_read_diagnostics::Stage::$stage,
+                        $index,
+                        $operation,
+                    )
+                }
+                #[cfg(not(feature = "diagnostic-local-batch"))]
+                {
+                    $operation()
+                }
+            }};
+        }
+        #[cfg(feature = "diagnostic-local-batch")]
+        let _dispatch = diagnostic
+            .as_ref()
+            .map(|sample| tracing::dispatcher::set_default(sample.dispatch()));
+        #[cfg(feature = "diagnostic-local-batch")]
+        let execution = diagnostic.as_ref().map(|sample| {
+            sample.worker_started();
+            sample.start()
+        });
         let execution_started = metrics.as_ref().map(|_| std::time::Instant::now());
         if let (Some(metrics), Some(submitted)) = (&metrics, submitted) {
             metrics.record_stage(LocalBatchReadStage::QueueWait, true, submitted.elapsed());
@@ -11339,41 +11368,64 @@ pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize,
         }
         let mut results = Vec::with_capacity(n);
         for (file_path, offset, length) in requests {
-            let mut bytes_read = 0;
+            #[cfg(feature = "diagnostic-local-batch")]
+            let _index = results.len();
+            let mut total = 0usize;
             let r = (|| -> Result<Bytes> {
-                let meta = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Metadata, || {
-                    std::fs::metadata(&file_path).map_err(DiskError::from)
-                })?;
+                let meta = diagnostic_stage!(Metadata, _index, || measure_local_batch_stage(
+                    metrics.as_deref(),
+                    LocalBatchReadStage::Metadata,
+                    || std::fs::metadata(&file_path).map_err(DiskError::from),
+                ))?;
                 let end = offset.checked_add(length).ok_or(DiskError::FileCorrupt)?;
                 if meta.len() < u64::try_from(end).unwrap_or(u64::MAX) {
                     return Err(DiskError::FileCorrupt);
                 }
 
-                let file = measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Open, || {
-                    std::fs::File::open(&file_path).map_err(DiskError::from)
-                })?;
-                let mut buf = vec![0u8; length];
+                let file = diagnostic_stage!(Open, _index, || measure_local_batch_stage(
+                    metrics.as_deref(),
+                    LocalBatchReadStage::Open,
+                    || std::fs::File::open(&file_path).map_err(DiskError::from),
+                ))?;
+                let mut buf = diagnostic_stage!(Allocation, _index, || Ok::<_, DiskError>(vec![0u8; length]))?;
                 if length > 0 {
-                    measure_local_batch_stage(metrics.as_deref(), LocalBatchReadStage::Read, || {
-                        while bytes_read < length {
-                            let nbytes = file
-                                .read_at(&mut buf[bytes_read..], u64::try_from(offset + bytes_read).unwrap_or(u64::MAX))
-                                .map_err(DiskError::from)?;
-                            if nbytes == 0 {
-                                return Err(DiskError::FileCorrupt);
+                    diagnostic_stage!(Read, _index, || measure_local_batch_stage(
+                        metrics.as_deref(),
+                        LocalBatchReadStage::Read,
+                        || {
+                            while total < length {
+                                let nbytes = file
+                                    .read_at(&mut buf[total..], u64::try_from(offset + total).unwrap_or(u64::MAX))
+                                    .map_err(DiskError::from)?;
+                                if nbytes == 0 {
+                                    return Err(DiskError::FileCorrupt);
+                                }
+                                total += nbytes;
                             }
-                            bytes_read += nbytes;
+                            Ok::<_, DiskError>(())
                         }
-                        Ok(())
-                    })?;
+                    ))?;
                 }
                 Ok(Bytes::from(buf))
             })();
+            #[cfg(feature = "diagnostic-local-batch")]
+            if let Some(sample) = &diagnostic {
+                sample.shard_result(_index, length, total, r.is_ok());
+            }
             if let Some(metrics) = &metrics {
-                metrics.record_shard(r.is_ok(), length, bytes_read);
+                metrics.record_shard(r.is_ok(), length, total);
             }
             results.push(r);
         }
+        #[cfg(feature = "diagnostic-local-batch")]
+        if let (Some(sample), Some(started)) = (&diagnostic, execution) {
+            sample.finish(
+                crate::diagnostics::batch_read_diagnostics::Stage::Execution,
+                None,
+                started,
+                results.iter().all(Result::is_ok),
+            );
+        }
         if let (Some(metrics), Some(started)) = (&metrics, execution_started) {
             metrics.record_stage(LocalBatchReadStage::Execution, results.iter().all(Result::is_ok), started.elapsed());
         }

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

@@ -3604,7 +3604,7 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
         let read_policy = crate::set_disk::get_object_read_policy();
         let erasure_cache = Arc::clone(&self.erasure_cache);
         let (fi, files, disks) = snapshot.into_owned();
-        tokio::spawn(async move {
+        let producer = async move {
             let _guard = read_lock_guard;
             let mut writer = GetObjectDownstreamWriter::new(wd);
             // Do not wrap the entire read+write pipeline in `disk_read_timeout`.
@@ -3687,7 +3687,10 @@ impl crate::storage_api_contracts::object::ObjectIO for SetDisks {
                 }
             };
             let _ = producer_terminal_tx.send(producer_result.map(|_| ()));
-        });
+        };
+        #[cfg(feature = "diagnostic-local-batch")]
+        let producer = crate::diagnostics::batch_read_diagnostics::with_captured_request_context(producer);
+        tokio::spawn(producer);
 
         Ok(reader)
     }

crates/obs/Cargo.toml

@@ -27,6 +27,8 @@ documentation = "https://docs.rs/rustfs-obs/latest/rustfs_obs/"
 
 [features]
 default = []
+# Default-off JSON isolation for sampled local-batch diagnostic root events.
+diagnostic-local-batch = []
 hotpath = [
     "hotpath/hotpath",
     "hotpath/tokio",

crates/obs/src/telemetry/local.rs

@@ -192,7 +192,7 @@ impl<T> RequestIdJsonFormat<T> {
 impl<S, T> FormatEvent<S, JsonFields> for RequestIdJsonFormat<T>
 where
     S: Subscriber + for<'lookup> LookupSpan<'lookup>,
-    T: tracing_subscriber::fmt::time::FormatTime,
+    T: tracing_subscriber::fmt::time::FormatTime + Clone,
 {
     fn format_event(
         &self,
@@ -200,6 +200,17 @@ where
         mut writer: Writer<'_>,
         event: &tracing::Event<'_>,
     ) -> fmt::Result {
+        #[cfg(feature = "diagnostic-local-batch")]
+        if event.is_root() && event.metadata().target() == rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET {
+            // Explicit root events must not acquire the formatter's current
+            // request span or an inferred request ID from that span.
+            return self
+                .inner
+                .clone()
+                .with_current_span(false)
+                .with_span_list(false)
+                .format_event(ctx, writer, event);
+        }
         let mut buffer = String::new();
         self.inner.format_event(ctx, Writer::new(&mut buffer), event)?;
 
@@ -813,6 +824,74 @@ mod tests {
         }
     }
 
+    #[cfg(feature = "diagnostic-local-batch")]
+    fn render_local_batch_diagnostic(request_id: Option<&str>) -> (Value, Value) {
+        let writer = SharedWriter::default();
+        let layer = build_json_log_layer(writer.clone(), false, FmtSpan::FULL, true);
+        let subscriber = Registry::default().with(layer);
+        tracing::subscriber::with_default(subscriber, || {
+            let parent = tracing::info_span!("request", request_id = "parent-request", uri = "/must-not-copy-parent-path");
+            let _entered = parent.enter();
+            tracing::trace!(target: rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET, parent: None,
+                event = "local_batch_diagnostic", component = "ecstore", subsystem = "local_batch_diagnostic",
+                result = "success", request_id,
+                request_context = if request_id.is_some() { "present" } else { "missing" },
+                "diagnostic root event");
+            tracing::info!(
+                event = EVENT_LOCAL_LOGGING_STATE,
+                component = LOG_COMPONENT_OBS,
+                subsystem = LOG_SUBSYSTEM_LOCAL_LOGGING,
+                result = "test",
+                "ordinary context event"
+            );
+        });
+        let bytes = writer.inner.lock().expect("formatted diagnostic records");
+        let records: Vec<Value> = std::str::from_utf8(&bytes)
+            .expect("UTF-8 JSON")
+            .lines()
+            .map(|line| serde_json::from_str(line).expect("JSON record"))
+            .collect();
+        let diagnostic = records
+            .iter()
+            .find(|record| record["target"] == rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET)
+            .expect("diagnostic event")
+            .clone();
+        let ordinary = records
+            .into_iter()
+            .find(|record| record["message"] == "ordinary context event")
+            .expect("ordinary event");
+        (diagnostic, ordinary)
+    }
+
+    #[cfg(feature = "diagnostic-local-batch")]
+    #[test]
+    fn local_batch_diagnostic_root_omits_parent_span_without_changing_ordinary_logs() {
+        let (diagnostic, ordinary) = render_local_batch_diagnostic(Some("server-owned"));
+        assert_eq!(diagnostic["request_id"], "server-owned");
+        assert!(diagnostic.get("request-id").is_none());
+        assert!(diagnostic.get("span").is_none());
+        assert!(diagnostic.get("spans").is_none());
+        assert!(
+            !serde_json::to_string(&diagnostic)
+                .expect("diagnostic JSON")
+                .contains("must-not-copy")
+        );
+        assert_eq!(ordinary["span"]["uri"], "/must-not-copy-parent-path");
+        assert_eq!(ordinary["request_id"], "parent-request");
+        assert_eq!(ordinary["request-id"], "parent-request");
+    }
+
+    #[cfg(feature = "diagnostic-local-batch")]
+    #[test]
+    fn local_batch_diagnostic_missing_context_does_not_infer_parent_request_id() {
+        let (diagnostic, _) = render_local_batch_diagnostic(None);
+        assert_eq!(diagnostic["request_context"], "missing");
+        assert!(diagnostic.get("request_id").is_none());
+        assert!(diagnostic.get("request-id").is_none());
+        assert!(diagnostic.get("span").is_none());
+        assert!(diagnostic.get("spans").is_none());
+    }
+
     fn render_json_log(enable_ansi: bool) -> (String, Value) {
         let writer = SharedWriter::default();
         let layer = build_json_log_layer(writer.clone(), enable_ansi, FmtSpan::NONE, false);

docs/README.md

@@ -8,6 +8,8 @@ collection:
 
 ## Operations
 
+- [Sampled local batch CPU diagnostics](operations/local-batch-cpu-diagnostics.md) — default-off request-correlated allocation and worker wall/CPU diagnosis.
+
 - [Retained MRF responsibilities](operations/mrf-responsibilities.md) — unknown bucket identities, bounded probes, and unreadable checkpoint recovery.
 - [Read-only BTF declaration inspection](operations/btf-prototype-inspection.md) — bounded prototypes and scalar member layouts, explicit void handling, and independent native references.
 

docs/architecture/ecstore-api-facade-inventory.md

@@ -15,6 +15,7 @@ The broad `rustfs_ecstore::api` facade is a compatibility boundary, not an archi
 | `data_usage`, `capacity`, `notification`, `metrics`, `rebalance`, `integrity` | Domain and service facades consumed through owner-local boundaries. | Narrow one group at a time after explicit aliases or wrappers exist. |
 | `set_disk`, `object`, `object_api_utils`, `rio`, `bitrot`, `erasure`, `compression`, `cache`, `store_list` | Low-level object IO, reader, erasure, cache, and migration helper compatibility. | Keep stable while `SetDisks` remains the shared state carrier. |
 | `admin`, `event`, `global` | Admin, event hook, and bootstrap-global compatibility. | `global` is limited to bootstrap writes and lifecycle controls; read-only runtime access goes through `runtime`. |
+| `diagnostic` | Feature-gated read-version attribution and sampled local-batch request context. | Keep explicit exports and default-off gates; consumers use the storage-owner and root-local diagnostic boundaries. |
 
 The S3 client is no longer a facade group: it lives in `crates/s3-client` (`rustfs_s3_client`). Regenerate the group list with:
 

docs/operations/local-batch-cpu-diagnostics.md

@@ -0,0 +1,69 @@
+# Sampled local batch CPU diagnostics
+
+**Use this when:** separating sampled local EC batch allocation/initialization
+and filesystem wall time from worker CPU time in a default-off diagnostic build.
+
+This diagnostic build separates allocation/initialization, metadata, open and
+positioned-read elapsed time from Linux thread CPU time. It preserves the
+single blocking dispatch, buffer initialization, offsets, integrity checks and
+error handling. It does not implement a rustix optimization or qualify
+observation overhead.
+
+Build separately from any frozen performance binary:
+
+```sh
+RUSTUP_TOOLCHAIN=stable cargo build --release --locked -p rustfs --bin rustfs \
+  --features diagnostic-local-batch
+```
+
+The feature is absent from default builds. Even this diagnostic build records
+nothing until both the sample rate and the target filter are enabled:
+
+```sh
+export RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY=64
+export RUST_LOG=warn,rustfs_local_batch_diagnostic=trace
+```
+
+The rate is an integer in `1..=65536`. Missing, zero or invalid values disable
+capture. Rate 64 selects one in every 64 submitted batches; rate 1 is intended
+only for small correctness probes. Restart to change the rate. Disabled or
+unselected batches create no timing records and read no diagnostic clocks.
+While recording is enabled, each external S3 request retains a bounded shared
+ID, including requests whose batches are not selected. This is diagnostic work
+and must be accounted for before any observation-cost claim.
+
+Business fields contain a process-local batch ID, the bounded server-owned canonical
+request ID when propagated, worker PID/TID, shard index, sizes and timing.
+The RustFS feature also enables a matching JSON formatter policy: explicit
+root events on this fixed diagnostic target omit current/ancestor spans and
+parent-derived request ID aliases. No additional spans are allocated on the
+sampled I/O path. There are no variable-ID metric labels. Missing request
+context is reported as `request_context=missing`, never treated as a successful
+join.
+The standard logger may attach source-location and thread metadata. Its JSON
+format flattens business fields; readers must distinguish these fields from
+formatter metadata and reject inherited request spans.
+
+Queue wait is elapsed time across the async/blocking boundary and has no
+thread CPU delta. Execution and per-shard stages run synchronously on one
+worker thread. CPU clock failure or unsupported platforms omit `cpu_ns` and
+report `cpu_clock_status=unavailable`; this is different from an available
+zero-duration measurement. Clock and logging work perturb diagnostic samples.
+Elapsed and CPU are measured separately and must not be assumed equal or
+subtracted without accounting for measurement overhead. Stage durations can
+overlap and per-shard samples differ from per-batch samples.
+When production batch-stage metrics are enabled, these timings also include
+their recording work. Keep that state in the provenance; they are not bare
+system-call durations.
+
+`shard_result` reports requested and actually read bytes, including partial
+reads on error. Unattempted stages are absent; a zero-length shard has no read
+sample. Captured correlation and dispatch stay with a blocking worker even
+after its caller is cancelled.
+
+Keep the diagnostic binary, source revision, feature, rate, filters and raw
+events distinct from baseline or acceptance artifacts. Do not log credentials,
+object/disk paths, request bodies or arbitrary client-supplied identifiers to add
+correlation. Request-level client association must use the server's canonical
+response ID. Diagnostic results locate hypotheses; they do not regrade failed
+A/A or ABBA runs.

rustfs/Cargo.toml

@@ -79,6 +79,8 @@ pyroscope = ["rustfs-obs/pyroscope", "dep:pyroscope", "dep:object", "dep:symboli
 dial9 = ["rustfs-obs/dial9"]
 # Diagnostic-only Rust GlobalAlloc live-byte accounting; adds per-allocation atomic updates.
 diagnostic-allocator-live = []
+# Default-off sampled local batch wall/CPU and request correlation diagnostics.
+diagnostic-local-batch = ["rustfs-ecstore/diagnostic-local-batch", "rustfs-obs/diagnostic-local-batch"]
 # Diagnostic-only S3-operation attribution for internode ReadVersion and list-walker metrics.
 diagnostic-read-version-caller = ["rustfs-ecstore/diagnostic-read-version-caller", "rustfs-io-metrics/diagnostic-read-version-caller", "rustfs-scanner/diagnostic-read-version-caller"]
 http3 = ["dep:h3", "dep:h3-quinn", "dep:quinn"]

rustfs/src/server/layer.rs

@@ -29,6 +29,8 @@ use crate::server::{
 };
 use crate::shared_types::ReadinessDegradedReason;
 use crate::storage_api::server::layer::apply_cors_headers;
+#[cfg(feature = "diagnostic-local-batch")]
+use crate::storage_api::server::layer::batch_diagnostic;
 use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced};
 use bytes::{Bytes, BytesMut};
 use futures::future::Either;
@@ -73,6 +75,10 @@ const HTTP_REQUEST_INFLIGHT_SLOW_EVENT: &str = "http_request_inflight_slow";
 const LOG_COMPONENT_SERVER: &str = "server";
 const LOG_SUBSYSTEM_HTTP: &str = "http";
 const REDACTED_QUERY_VALUE: &str = "redacted";
+#[cfg(feature = "diagnostic-local-batch")]
+type BatchDiagnosticRequest = Option<Arc<str>>;
+#[cfg(not(feature = "diagnostic-local-batch"))]
+type BatchDiagnosticRequest = ();
 const OBJECT_ZIP_DOWNLOADS_PATH: &str = "/v3/object-zip-downloads/";
 const HTTP_REQUEST_INFLIGHT_WARN_THRESHOLD: Duration = Duration::from_secs(5);
 static HTTP_SERVER_ERROR_LOGS: [rustfs_utils::LogThrottle; 100] = [const { rustfs_utils::LogThrottle::new(5_000) }; 100];
@@ -285,6 +291,12 @@ where
         } else {
             req.headers().get(REQUEST_ID_HEADER).cloned()
         };
+        #[cfg(feature = "diagnostic-local-batch")]
+        let diagnostic_request = is_s3
+            .then(|| batch_diagnostic::request_context(&request_context.request_id))
+            .flatten();
+        #[cfg(not(feature = "diagnostic-local-batch"))]
+        let diagnostic_request = ();
         req.extensions_mut().insert(request_context);
 
         // This outer boundary includes readiness, rate-limit and auth
@@ -300,6 +312,12 @@ where
             .flatten();
         let head_request = req.method() == Method::HEAD;
         let mut metrics = is_s3.then(|| s3_http_request_guard(req.method().as_str()));
+        #[cfg(feature = "diagnostic-local-batch")]
+        let inner = batch_diagnostic::in_request_context(diagnostic_request.as_ref(), || match metrics.as_mut() {
+            Some(metrics) => metrics.in_scope(|| self.inner.call(req)),
+            None => self.inner.call(req),
+        });
+        #[cfg(not(feature = "diagnostic-local-batch"))]
         let inner = match metrics.as_mut() {
             Some(metrics) => metrics.in_scope(|| self.inner.call(req)),
             None => self.inner.call(req),
@@ -309,6 +327,7 @@ where
         }
         ExternalRequestContextFuture {
             inner,
+            diagnostic_request,
             request_id,
             is_s3,
             metrics,
@@ -358,6 +377,7 @@ pin_project! {
     pub struct ExternalRequestContextFuture<F> {
         #[pin]
         inner: F,
+        diagnostic_request: BatchDiagnosticRequest,
         request_id: Option<HeaderValue>,
         is_s3: bool,
         metrics: Option<S3HttpRequestGuard>,
@@ -374,6 +394,12 @@ where
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
+        #[cfg(feature = "diagnostic-local-batch")]
+        let result = batch_diagnostic::in_request_context(this.diagnostic_request.as_ref(), || match this.metrics.as_mut() {
+            Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)),
+            None => this.inner.poll(cx),
+        });
+        #[cfg(not(feature = "diagnostic-local-batch"))]
         let result = match this.metrics.as_mut() {
             Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)),
             None => this.inner.poll(cx),
@@ -2608,6 +2634,83 @@ fn rewrite_double_slash_root(uri: &Uri) -> Option<Uri> {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(feature = "diagnostic-local-batch")]
+    #[test]
+    fn diagnostic_external_request_uses_server_id_not_client_header() {
+        const CHILD: &str = "RUSTFS_TEST_BATCH_REQUEST_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let status = std::process::Command::new(std::env::current_exe().expect("current test executable"))
+                .args([
+                    "--exact",
+                    "server::layer::tests::diagnostic_external_request_uses_server_id_not_client_header",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .env("RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY", "1")
+                .status()
+                .expect("run isolated request correlation test");
+            assert!(status.success(), "isolated request correlation test failed");
+            return;
+        }
+        let subscriber = tracing_subscriber::fmt()
+            .with_max_level(tracing::Level::TRACE)
+            .with_writer(std::io::sink)
+            .finish();
+        tracing::subscriber::with_default(subscriber, || {
+            let inner = tower::service_fn(|request: HttpRequest<Full<Bytes>>| {
+                let context = request
+                    .extensions()
+                    .get::<RequestContext>()
+                    .expect("canonical request context");
+                let expected = context.request_id.clone();
+                assert_ne!(expected, "client-controlled-id");
+                assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some(expected.as_str()));
+                async move {
+                    assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some(expected.as_str()));
+                    Ok::<_, std::convert::Infallible>(Response::new(Full::new(Bytes::new())))
+                }
+            });
+            let mut service = ExternalRequestContextLayer::new(false).layer(inner);
+            let request = HttpRequest::builder()
+                .method(Method::GET)
+                .uri("/diagnostic-bucket/object")
+                .header(REQUEST_ID_HEADER, "client-controlled-id")
+                .body(Full::new(Bytes::new()))
+                .expect("test request");
+            let response = futures::executor::block_on(service.call(request)).expect("correlated response");
+            assert_eq!(response.status(), StatusCode::OK);
+            assert!(batch_diagnostic::current_request_id().is_none());
+        });
+    }
+
+    #[cfg(feature = "diagnostic-local-batch")]
+    #[test]
+    fn diagnostic_request_scope_is_present_on_every_poll_and_restored() {
+        let mut polls = 0;
+        let inner = std::future::poll_fn(|cx| {
+            assert_eq!(batch_diagnostic::current_request_id().as_deref(), Some("server-owned-test"));
+            polls += 1;
+            if polls == 1 {
+                cx.waker().wake_by_ref();
+                Poll::Pending
+            } else {
+                Poll::Ready(Ok::<_, std::convert::Infallible>(Response::new(http_body_util::Full::new(Bytes::new()))))
+            }
+        });
+        let future = ExternalRequestContextFuture {
+            inner,
+            diagnostic_request: Some(Arc::from("server-owned-test")),
+            request_id: None,
+            is_s3: false,
+            metrics: None,
+            feedback: None,
+            head_request: false,
+        };
+        let response = futures::executor::block_on(future).expect("diagnostic request future");
+        assert_eq!(response.status(), StatusCode::OK);
+        assert_eq!(polls, 2);
+        assert!(batch_diagnostic::current_request_id().is_none());
+    }
     #[tokio::test]
     async fn foreground_feedback_unobserved_body_is_a_direct_passthrough() {
         struct EndProbe(Full<Bytes>);

rustfs/src/storage/storage_api.rs

@@ -28,13 +28,18 @@ pub(crate) mod ecstore_integrity {
     };
 }
 
-#[cfg(feature = "diagnostic-read-version-caller")]
+#[cfg(any(feature = "diagnostic-read-version-caller", feature = "diagnostic-local-batch"))]
 pub(crate) mod ecstore_diagnostic {
+    #[cfg(all(test, feature = "diagnostic-local-batch"))]
+    pub(crate) use rustfs_ecstore::api::diagnostic::current_request_id;
+    #[cfg(feature = "diagnostic-read-version-caller")]
     pub(crate) use rustfs_ecstore::api::diagnostic::{
         ReadVersionContextBoundary, ReadVersionEntrySource, with_captured_read_version_context, with_read_version_spawn_context,
     };
-    #[cfg(test)]
+    #[cfg(all(test, feature = "diagnostic-read-version-caller"))]
     pub(crate) use rustfs_ecstore::api::diagnostic::{current_read_version_entry_source, with_read_version_entry};
+    #[cfg(feature = "diagnostic-local-batch")]
+    pub(crate) use rustfs_ecstore::api::diagnostic::{in_request_context, request_context};
 }
 use tokio::sync::{Mutex, OwnedMutexGuard};
 use tokio_util::sync::CancellationToken;

rustfs/src/storage_api.rs

@@ -245,6 +245,13 @@ pub(crate) mod server {
     pub(crate) mod layer {
         pub(crate) use crate::storage::storage_api::apply_cors_headers;
 
+        #[cfg(feature = "diagnostic-local-batch")]
+        pub(crate) mod batch_diagnostic {
+            #[cfg(test)]
+            pub(crate) use crate::storage::storage_api::ecstore_diagnostic::current_request_id;
+            pub(crate) use crate::storage::storage_api::ecstore_diagnostic::{in_request_context, request_context};
+        }
+
         pub(crate) mod request_context {
             pub(crate) use crate::storage::storage_api::request_context_consumer::{
                 RequestContext, extract_request_id_from_headers, spawn_traced,

scripts/check_logging_guardrails.sh

@@ -5,6 +5,7 @@ repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 cd "$repo_root"
 
 checked_files=(
+  "crates/ecstore/src/diagnostics/batch_read_diagnostics.rs"
   "rustfs/src/main.rs"
   "rustfs/src/startup_entrypoint.rs"
   "rustfs/src/init.rs"