feat(ecstore): add request-coherent GET diagnostics

Sample canonical GET requests once and retain their context through batch
workers, the legacy producer and response completion. Aggregate bounded
verification, copy, reconstruction, emit and body-poll diagnostics without
changing the I/O backend or integrity checks.

Keep thread CPU snapshots inside synchronous operations, preserve explicit
cancellation boundaries and leave the feature disabled by default.

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,10 @@ 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"]
+# Request-coherent diagnostics; never enables an alternative I/O backend.
+diagnostic-get-pipeline = ["diagnostic-local-batch"]
 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,20 @@
 
 //! 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-get-pipeline")]
+    pub use crate::diagnostics::get_pipeline::{
+        RequestSample, cancel_response, capture_request, fail_service, finish_response, in_context, response_poll, response_ready,
+    };
+    #[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,599 @@
+// 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>>;
+}
+
+pub(super) 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)
+}
+
+pub(super) 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>> {
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    if super::get_pipeline::mode_enabled() {
+        return super::get_pipeline::current().and_then(|sample| {
+            let selected_id = sample.id();
+            (selected_id.as_ref() == id).then_some(selected_id)
+        });
+    }
+    (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> {
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    let future = super::get_pipeline::with_captured_context(future);
+    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")]
+pub(super) 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"))]
+pub(super) fn cpu_ns() -> Option<u64> {
+    None
+}
+
+pub(super) 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,
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    pipeline: Option<Arc<super::get_pipeline::RequestSample>>,
+}
+
+impl Sample {
+    pub(crate) fn capture() -> Option<Self> {
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        if super::get_pipeline::mode_enabled() {
+            // Request mode uses the external request's one decision. It never
+            // intersects an independent per-batch sampler with that cohort.
+            let pipeline = super::get_pipeline::current()?;
+            let mut sample = Self::capture_with(1, true, &NEXT_BATCH)?;
+            sample.request_id = Some(pipeline.id());
+            sample.pipeline = Some(pipeline);
+            return Some(sample);
+        }
+        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),
+            #[cfg(feature = "diagnostic-get-pipeline")]
+            pipeline: None,
+        })
+    }
+
+    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) {
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        if self.pipeline.as_ref().is_some_and(|sample| !sample.allow_batch_event()) {
+            return;
+        }
+        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) {
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        if self.pipeline.as_ref().is_some_and(|sample| !sample.allow_batch_event()) {
+            return;
+        }
+        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/get_pipeline.rs

@@ -0,0 +1,719 @@
+// 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.
+
+//! Bounded, request-coherent diagnostics for a separate build. Overlapping
+//! stage totals are work sums, not an additive request critical path.
+
+use super::batch_read_diagnostics::{cpu_ns, parse_rate, thread_id, valid_request_id};
+use serde::Serialize;
+use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, LazyLock, Mutex, OnceLock};
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
+use tracing::Dispatch;
+
+const TARGET: &str = rustfs_config::LOCAL_BATCH_DIAGNOSTIC_LOG_TARGET;
+const EVENT_GET_PIPELINE_DIAGNOSTIC: &str = "get_pipeline_diagnostic";
+const LOG_COMPONENT_ECSTORE: &str = "ecstore";
+const LOG_SUBSYSTEM_GET_PIPELINE: &str = "get_pipeline_diagnostic";
+const ENV_SAMPLE_EVERY: &str = "RUSTFS_DIAGNOSTIC_GET_STAGE_SAMPLE_EVERY";
+const MAX_REQUESTS: u64 = 4096;
+// Each fixed-schema event reserves 8 KiB, including its bounded ASCII ID.
+// Reserve the final summary before accepting any batch events. Exhaustion
+// censors further batch events and remains visible in that summary.
+const MAX_EVENT_BYTES: u64 = 8192;
+const MAX_EVENTS: u64 = (1024 * 1024 * 1024) / MAX_EVENT_BYTES;
+static RATE: OnceLock<u64> = OnceLock::new();
+static BUDGET: LazyLock<Arc<Budget>> = LazyLock::new(|| Arc::new(Budget::default()));
+
+tokio::task_local! {
+    static REQUEST: Option<Arc<RequestSample>>;
+}
+
+#[derive(Default)]
+struct Budget {
+    sequence: AtomicU64,
+    requests: AtomicU64,
+    events: AtomicU64,
+}
+
+fn reserve(counter: &AtomicU64, limit: u64) -> bool {
+    counter
+        .try_update(Ordering::Relaxed, Ordering::Relaxed, |n| (n < limit).then(|| n + 1))
+        .is_ok()
+}
+
+pub(crate) fn mode_enabled() -> bool {
+    *RATE.get_or_init(|| parse_rate(std::env::var(ENV_SAMPLE_EVERY).ok().as_deref())) != 0
+}
+
+/// Capture once at the external canonical-ID boundary, never from a client header.
+pub fn capture_request(id: &str) -> Option<Arc<RequestSample>> {
+    let rate = *RATE.get_or_init(|| parse_rate(std::env::var(ENV_SAMPLE_EVERY).ok().as_deref()));
+    if rate == 0 || !tracing::enabled!(target: TARGET, tracing::Level::TRACE) {
+        return None;
+    }
+    capture(id, rate, true, &BUDGET)
+}
+
+fn capture(id: &str, rate: u64, enabled: bool, budget: &Arc<Budget>) -> Option<Arc<RequestSample>> {
+    if rate == 0 || !enabled || !valid_request_id(id) {
+        return None;
+    }
+    let request_sequence = budget.sequence.fetch_add(1, Ordering::Relaxed);
+    if !request_sequence.is_multiple_of(rate) || !reserve(&budget.requests, MAX_REQUESTS) || !reserve(&budget.events, MAX_EVENTS)
+    {
+        return None;
+    }
+    Some(Arc::new(RequestSample {
+        id: Arc::from(id),
+        request_sequence,
+        sample_every: rate,
+        start: Instant::now(),
+        start_unix_ns: SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .ok()
+            .and_then(|d| d.as_nanos().try_into().ok()),
+        dispatch: tracing::dispatcher::get_default(Clone::clone),
+        budget: Arc::clone(budget),
+        truncated: AtomicBool::new(false),
+        stats: Mutex::new(Stats::default()),
+    }))
+}
+
+/// Scope only the synchronous operation/poll. Task locals are explicitly restored.
+pub fn in_context<T>(sample: Option<&Arc<RequestSample>>, operation: impl FnOnce() -> T) -> T {
+    REQUEST.sync_scope(sample.cloned(), operation)
+}
+
+pub(crate) fn current() -> Option<Arc<RequestSample>> {
+    REQUEST.try_with(Clone::clone).ok().flatten()
+}
+
+pub(crate) fn with_captured_context<F: std::future::Future>(future: F) -> impl std::future::Future<Output = F::Output> {
+    REQUEST.scope(current(), future)
+}
+
+#[derive(Clone, Copy)]
+pub(crate) enum Stage {
+    Verify,
+    Copy,
+    Reconstruct,
+    Emit,
+    ResponsePoll,
+}
+
+impl Stage {
+    fn index(self) -> usize {
+        self as usize
+    }
+}
+
+#[derive(Clone, Copy, Default, Serialize)]
+struct Total {
+    calls: u64,
+    bytes: u64,
+    wall_ns: u64,
+    cpu_ns: u64,
+    cpu_unavailable: u64,
+    failures: u64,
+    cancelled: u64,
+    first_tid: Option<i32>,
+    last_tid: Option<i32>,
+    thread_changes: u64,
+}
+
+#[derive(Clone, Copy, Default)]
+struct Stats {
+    totals: [Total; 5],
+    terminal: Option<&'static str>,
+    response_ready_ns: Option<u64>,
+    first_data_poll_end_ns: Option<u64>,
+    body_end_ns: Option<u64>,
+    overflowed: bool,
+    poisoned: bool,
+}
+
+/// Owned by the request, producer and selected blocking work. The last owner
+/// emits after cancellation too; no mutex is held across await or event output.
+pub struct RequestSample {
+    id: Arc<str>,
+    request_sequence: u64,
+    sample_every: u64,
+    start: Instant,
+    start_unix_ns: Option<u64>,
+    dispatch: Dispatch,
+    budget: Arc<Budget>,
+    truncated: AtomicBool,
+    stats: Mutex<Stats>,
+}
+
+impl RequestSample {
+    pub(crate) fn id(&self) -> Arc<str> {
+        Arc::clone(&self.id)
+    }
+
+    pub(crate) fn allow_batch_event(&self) -> bool {
+        if reserve(&self.budget.events, MAX_EVENTS) {
+            return true;
+        }
+        self.truncated.store(true, Ordering::Relaxed);
+        false
+    }
+
+    fn update(&self, operation: impl FnOnce(&mut Stats)) {
+        let mut stats = self.stats.lock().unwrap_or_else(|poison| {
+            let mut stats = poison.into_inner();
+            stats.poisoned = true;
+            stats
+        });
+        operation(&mut stats);
+    }
+
+    fn record(&self, stage: Stage, bytes: usize, wall: u128, cpu: Option<u64>, outcome: Option<bool>, tid: Option<i32>) {
+        self.update(|stats| {
+            if matches!(stage, Stage::ResponsePoll) && bytes != 0 && stats.first_data_poll_end_ns.is_none() {
+                stats.first_data_poll_end_ns = self.start.elapsed().as_nanos().try_into().ok();
+            }
+            let total = &mut stats.totals[stage.index()];
+            if let Some(tid) = tid {
+                if total.last_tid.is_some_and(|previous| previous != tid) {
+                    if let Some(changes) = total.thread_changes.checked_add(1) {
+                        total.thread_changes = changes;
+                    } else {
+                        total.thread_changes = u64::MAX;
+                        stats.overflowed = true;
+                    }
+                }
+                total.first_tid.get_or_insert(tid);
+                total.last_tid = Some(tid);
+            }
+            let mut add = |destination: &mut u64, value: u128| {
+                if let Some(sum) = u64::try_from(value).ok().and_then(|v| destination.checked_add(v)) {
+                    *destination = sum;
+                } else {
+                    *destination = u64::MAX;
+                    stats.overflowed = true;
+                }
+            };
+            add(&mut total.calls, 1);
+            add(&mut total.bytes, u128::try_from(bytes).unwrap_or(u128::MAX));
+            add(&mut total.wall_ns, wall);
+            if let Some(cpu) = cpu {
+                add(&mut total.cpu_ns, u128::from(cpu));
+            } else {
+                add(&mut total.cpu_unavailable, 1);
+            }
+            if outcome == Some(false) {
+                add(&mut total.failures, 1);
+            }
+            if outcome.is_none() {
+                add(&mut total.cancelled, 1);
+            }
+        });
+    }
+
+    fn finish_body(&self, outcome: &'static str) {
+        self.update(|stats| {
+            if stats.terminal.is_none() {
+                stats.terminal = Some(outcome);
+                stats.body_end_ns = self.start.elapsed().as_nanos().try_into().ok();
+            }
+        });
+    }
+}
+
+impl Drop for RequestSample {
+    fn drop(&mut self) {
+        let stats = *self.stats.get_mut().unwrap_or_else(|p| {
+            let s = p.into_inner();
+            s.poisoned = true;
+            s
+        });
+        let Ok(totals) = serde_json::to_string(&stats.totals) else {
+            return;
+        };
+        tracing::dispatcher::with_default(&self.dispatch, || {
+            tracing::trace!(target: TARGET, parent: None,
+                event = EVENT_GET_PIPELINE_DIAGNOSTIC, component = LOG_COMPONENT_ECSTORE,
+                subsystem = LOG_SUBSYSTEM_GET_PIPELINE, result = stats.terminal.unwrap_or("cancelled"),
+                request_id = self.id.as_ref(), request_sequence = self.request_sequence, sample_every = self.sample_every,
+                pid = std::process::id(), emitter_tid = thread_id(),
+                start_unix_ns = self.start_unix_ns,
+                lifetime_wall_ns = u64::try_from(self.start.elapsed().as_nanos()).ok(),
+                response_ready_ns = stats.response_ready_ns, first_data_poll_end_ns = stats.first_data_poll_end_ns,
+                body_end_ns = stats.body_end_ns,
+                stages = "verify,copy,reconstruct,emit,response_poll", stage_totals_json = totals,
+                truncated = self.truncated.load(Ordering::Relaxed), overflowed = stats.overflowed,
+                poisoned = stats.poisoned,
+                "Sampled GET pipeline totals");
+        });
+    }
+}
+
+fn cpu_delta(before: Option<(i32, u64)>, after: Option<(i32, u64)>) -> Option<u64> {
+    let ((before_tid, before_cpu), (after_tid, after_cpu)) = before.zip(after)?;
+    (before_tid == after_tid).then(|| after_cpu.checked_sub(before_cpu)).flatten()
+}
+
+fn cpu_snapshot() -> Option<(i32, u64)> {
+    thread_id().zip(cpu_ns())
+}
+
+// Private synchronous guard: callers cannot carry thread CPU state across await.
+struct SyncMeasurement {
+    sample: Arc<RequestSample>,
+    stage: Stage,
+    bytes: usize,
+    start: Instant,
+    cpu: Option<(i32, u64)>,
+    outcome: Option<bool>,
+}
+
+impl Drop for SyncMeasurement {
+    fn drop(&mut self) {
+        let cpu = cpu_delta(self.cpu, cpu_snapshot());
+        self.sample
+            .record(self.stage, self.bytes, self.start.elapsed().as_nanos(), cpu, self.outcome, thread_id());
+    }
+}
+
+pub(crate) fn measure_sync<T>(stage: Stage, bytes: usize, operation: impl FnOnce() -> T, success: impl FnOnce(&T) -> bool) -> T {
+    let Some(sample) = current() else {
+        return operation();
+    };
+    let mut guard = SyncMeasurement {
+        sample,
+        stage,
+        bytes,
+        start: Instant::now(),
+        cpu: cpu_snapshot(),
+        outcome: None,
+    };
+    let result = operation();
+    guard.outcome = Some(success(&result));
+    result
+}
+
+/// Async scopes intentionally carry no CPU snapshot. Drop records cancellation.
+pub(crate) struct ElapsedMeasurement {
+    sample: Arc<RequestSample>,
+    stage: Stage,
+    start: Instant,
+    outcome: Option<bool>,
+    bytes: usize,
+}
+
+impl ElapsedMeasurement {
+    pub(crate) fn emit() -> Option<Self> {
+        current().map(|sample| Self {
+            sample,
+            stage: Stage::Emit,
+            start: Instant::now(),
+            outcome: None,
+            bytes: 0,
+        })
+    }
+    pub(crate) fn finish(mut self, success: bool, bytes: usize) {
+        self.outcome = Some(success);
+        self.bytes = bytes;
+    }
+}
+
+impl Drop for ElapsedMeasurement {
+    fn drop(&mut self) {
+        self.sample
+            .record(self.stage, self.bytes, self.start.elapsed().as_nanos(), None, self.outcome, None);
+    }
+}
+
+/// Complete once at server body EOF/error; this is not a client/network ACK.
+pub fn finish_response(sample: Option<&Arc<RequestSample>>, success: bool) {
+    if let Some(sample) = sample {
+        sample.finish_body(if success { "body_complete" } else { "body_error" });
+    }
+}
+
+/// Mark the server service's response-ready boundary, not a network send.
+pub fn response_ready(sample: Option<&Arc<RequestSample>>) {
+    if let Some(sample) = sample {
+        sample.update(|stats| {
+            if stats.response_ready_ns.is_none() {
+                stats.response_ready_ns = sample.start.elapsed().as_nanos().try_into().ok();
+            }
+        });
+    }
+}
+
+/// The request/body owner calls this on drop, before any surviving producer drains.
+pub fn cancel_response(sample: Option<&Arc<RequestSample>>) {
+    if let Some(sample) = sample {
+        sample.finish_body("cancelled");
+    }
+}
+
+pub fn fail_service(sample: Option<&Arc<RequestSample>>) {
+    if let Some(sample) = sample {
+        sample.finish_body("service_error");
+    }
+}
+
+/// Observe exactly one synchronous body poll, preserving its return value.
+pub fn response_poll<T>(sample: Option<&Arc<RequestSample>>, operation: impl FnOnce() -> (T, usize, bool)) -> T {
+    in_context(sample, || {
+        let Some(sample) = current() else {
+            return operation().0;
+        };
+        let mut guard = SyncMeasurement {
+            sample,
+            stage: Stage::ResponsePoll,
+            bytes: 0,
+            start: Instant::now(),
+            cpu: cpu_snapshot(),
+            outcome: None,
+        };
+        let (result, bytes, success) = operation();
+        guard.bytes = bytes;
+        guard.outcome = Some(success);
+        result
+    })
+}
+
+#[cfg(test)]
+pub(crate) fn test_request() -> Arc<RequestSample> {
+    capture("canonical-stage-test", 1, true, &Arc::new(Budget::default())).expect("select test request")
+}
+
+#[cfg(test)]
+pub(crate) fn test_total(sample: &RequestSample, stage: Stage) -> (u64, u64, u64, u64, u64, u64) {
+    let total = sample.stats.lock().expect("snapshot diagnostic total").totals[stage.index()];
+    (
+        total.calls,
+        total.bytes,
+        total.cpu_ns,
+        total.cpu_unavailable,
+        total.failures,
+        total.cancelled,
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use serde_json::{Value, json};
+    use std::collections::BTreeMap;
+    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, f: &Field, v: u64) {
+            self.0.insert(f.name().into(), json!(v));
+        }
+        fn record_bool(&mut self, f: &Field, v: bool) {
+            self.0.insert(f.name().into(), json!(v));
+        }
+        fn record_str(&mut self, f: &Field, v: &str) {
+            self.0.insert(f.name().into(), json!(v));
+        }
+        fn record_debug(&mut self, f: &Field, v: &dyn std::fmt::Debug) {
+            self.0.insert(f.name().into(), json!(format!("{v:?}")));
+        }
+    }
+    type Records = Arc<Mutex<Vec<BTreeMap<String, Value>>>>;
+    struct Recorder(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 must never inherit URI-bearing spans");
+            let mut fields = Values::default();
+            event.record(&mut fields);
+            self.0.lock().expect("record bounded event").push(fields.0);
+        }
+    }
+    fn sample() -> Arc<RequestSample> {
+        capture("canonical-123", 1, true, &Arc::new(Budget::default())).expect("select canonical request")
+    }
+    fn stats(sample: &RequestSample) -> Stats {
+        *sample.stats.lock().expect("snapshot stage totals")
+    }
+
+    #[test]
+    fn selection_is_disabled_filtered_validated_and_bounded() {
+        let budget = Arc::new(Budget::default());
+        assert!(capture("canonical", 0, true, &budget).is_none());
+        assert!(capture("canonical", 64, false, &budget).is_none());
+        for id in ["", "invalid/id", "invalid\nid", &"a".repeat(129)] {
+            assert!(capture(id, 1, true, &budget).is_none());
+        }
+        assert_eq!(budget.sequence.load(Ordering::Relaxed), 0);
+        let selected: Vec<_> = (0..130)
+            .filter(|_| capture("canonical", 64, true, &budget).is_some())
+            .collect();
+        assert_eq!(selected, [0, 64, 128]);
+        budget.requests.store(MAX_REQUESTS, Ordering::Relaxed);
+        assert!(capture("canonical", 1, true, &budget).is_none());
+        budget.requests.store(0, Ordering::Relaxed);
+        budget.events.store(MAX_EVENTS, Ordering::Relaxed);
+        assert!(capture("canonical", 1, true, &budget).is_none());
+    }
+
+    #[test]
+    fn sync_errors_and_unwinds_preserve_behavior_without_payloads() {
+        let sample = sample();
+        let result: Result<(), _> = in_context(Some(&sample), || {
+            measure_sync(Stage::Verify, 123, || Err("private-error-payload"), Result::is_ok)
+        });
+        assert_eq!(result, Err("private-error-payload"));
+        let panic = std::panic::catch_unwind(std::panic::AssertUnwindSafe(|| {
+            in_context(Some(&sample), || measure_sync(Stage::Copy, 9, || panic!("original panic"), |_: &()| true));
+        }));
+        assert!(panic.is_err(), "diagnostics must not swallow panics");
+        let totals = stats(&sample).totals;
+        assert_eq!(totals[Stage::Verify.index()].failures, 1);
+        assert_eq!(totals[Stage::Verify.index()].bytes, 123);
+        assert_eq!(totals[Stage::Copy.index()].cancelled, 1);
+        assert!(current().is_none(), "task-local must restore after unwind");
+    }
+
+    #[test]
+    fn cpu_requires_same_thread_available_monotonic_snapshots() {
+        assert_eq!(cpu_delta(Some((1, 100)), Some((1, 145))), Some(45));
+        assert_eq!(cpu_delta(Some((1, 100)), Some((2, 145))), None);
+        assert_eq!(cpu_delta(Some((1, 100)), Some((1, 99))), None);
+        assert_eq!(cpu_delta(None, Some((1, 100))), None);
+        assert_eq!(cpu_delta(Some((1, 100)), None), None);
+    }
+
+    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
+    async fn captured_context_survives_spawn_and_pending_without_leaking() {
+        let sample = sample();
+        let cloned = Arc::clone(&sample);
+        let work = in_context(Some(&sample), || {
+            with_captured_context(async move {
+                assert!(Arc::ptr_eq(&current().expect("producer sample"), &cloned));
+                tokio::task::yield_now().await;
+                assert!(Arc::ptr_eq(&current().expect("resumed producer sample"), &cloned));
+                measure_sync(Stage::Copy, 17, || 42, |_| true)
+            })
+        });
+        assert_eq!(tokio::spawn(work).await.expect("join producer"), 42);
+        assert!(current().is_none());
+        assert_eq!(stats(&sample).totals[Stage::Copy.index()].bytes, 17);
+    }
+
+    #[test]
+    fn elapsed_scope_can_move_threads_and_has_no_cpu_delta() {
+        let sample = sample();
+        let guard = in_context(Some(&sample), ElapsedMeasurement::emit).expect("sample emit");
+        std::thread::spawn(move || guard.finish(true, 81))
+            .join()
+            .expect("finish on different thread");
+        let cancelled = in_context(Some(&sample), ElapsedMeasurement::emit).expect("sample cancelled emit");
+        drop(cancelled);
+        let total = stats(&sample).totals[Stage::Emit.index()];
+        assert_eq!(
+            (total.calls, total.bytes, total.cpu_ns, total.cpu_unavailable, total.cancelled),
+            (2, 81, 0, 2, 1)
+        );
+    }
+
+    #[test]
+    fn summary_waits_for_producer_and_emits_once_with_reserved_budget() {
+        let records = Records::default();
+        let subscriber = tracing_subscriber::registry().with(Recorder(Arc::clone(&records)));
+        let producer = tracing::subscriber::with_default(subscriber, || {
+            let span = tracing::info_span!("untrusted-request", uri = "/private?signature=secret");
+            let _entered = span.enter();
+            let sample = sample();
+            finish_response(Some(&sample), false);
+            finish_response(Some(&sample), true);
+            sample.budget.events.store(MAX_EVENTS, Ordering::Relaxed);
+            assert!(!sample.allow_batch_event());
+            sample
+        });
+        assert!(records.lock().expect("before producer drop").is_empty());
+        drop(producer);
+        let records = records.lock().expect("last owner emitted");
+        assert_eq!(records.len(), 1);
+        assert_eq!(records[0]["result"], "body_error");
+        assert_eq!(records[0]["truncated"], true);
+        assert_eq!(records[0]["request_id"], "canonical-123");
+        assert!(
+            !serde_json::to_string(&*records)
+                .expect("serialize fixed fields")
+                .contains("signature")
+        );
+    }
+
+    #[test]
+    fn terminal_states_and_poll_values_are_preserved() {
+        let sample = sample();
+        response_ready(Some(&sample));
+        let result = response_poll(Some(&sample), || (std::task::Poll::<Result<(), &str>>::Pending, 0, true));
+        assert_eq!(result, std::task::Poll::Pending);
+        let result = response_poll(Some(&sample), || (Err::<(), _>("original"), 512, false));
+        assert_eq!(result, Err("original"));
+        response_ready(Some(&sample));
+        fail_service(Some(&sample));
+        finish_response(Some(&sample), true);
+        let stats = stats(&sample);
+        assert_eq!(stats.terminal, Some("service_error"));
+        assert!(stats.response_ready_ns.is_some());
+        assert!(stats.body_end_ns.is_some());
+        assert_eq!(stats.totals[Stage::ResponsePoll.index()].calls, 2);
+        assert_eq!(stats.totals[Stage::ResponsePoll.index()].bytes, 512);
+        assert_eq!(stats.totals[Stage::ResponsePoll.index()].failures, 1);
+        assert!(current().is_none());
+    }
+
+    #[test]
+    fn cancelled_owner_boundary_precedes_last_producer_drain() {
+        let sample = sample();
+        cancel_response(Some(&sample));
+        let boundary = stats(&sample).body_end_ns.expect("cancelled body boundary");
+        let producer = Arc::clone(&sample);
+        in_context(Some(&producer), || measure_sync(Stage::Copy, 7, || (), |_| true));
+        finish_response(Some(&sample), true);
+        assert_eq!(stats(&sample).terminal, Some("cancelled"));
+        assert_eq!(
+            stats(&sample).body_end_ns,
+            Some(boundary),
+            "late work must not overwrite cancellation time"
+        );
+        assert_eq!(stats(&sample).totals[Stage::Copy.index()].bytes, 7);
+    }
+
+    #[test]
+    fn overflow_is_censored_and_owned_schema_fits_reserved_bytes() {
+        let sample = sample();
+        sample.record(Stage::Copy, 0, u128::MAX, Some(u64::MAX), Some(true), Some(1));
+        sample.record(Stage::Copy, 1, 1, Some(1), Some(true), Some(2));
+        assert!(stats(&sample).overflowed);
+        let total = stats(&sample).totals[Stage::Copy.index()];
+        assert_eq!((total.first_tid, total.last_tid, total.thread_changes), (Some(1), Some(2), 1));
+        let maximum = Total {
+            calls: u64::MAX,
+            bytes: u64::MAX,
+            wall_ns: u64::MAX,
+            cpu_ns: u64::MAX,
+            cpu_unavailable: u64::MAX,
+            failures: u64::MAX,
+            cancelled: u64::MAX,
+            first_tid: Some(i32::MAX),
+            last_tid: Some(i32::MAX),
+            thread_changes: u64::MAX,
+        };
+        let payload = serde_json::to_string(&[maximum; 5]).expect("serialize maximum fixed schema");
+        let escaped = serde_json::to_string(&payload).expect("serialize JSON field");
+        assert!(escaped.len() + 128 + 2048 < usize::try_from(MAX_EVENT_BYTES).expect("bounded event size"));
+    }
+
+    #[tokio::test]
+    async fn production_bitrot_copy_follows_successful_verification_only() {
+        use crate::erasure::coding::BitrotReader;
+        use rustfs_utils::HashAlgorithm;
+        let payload = b"verified payload";
+        let algo = HashAlgorithm::HighwayHash256S;
+        let valid = [algo.hash_encode(payload).as_ref(), payload].concat();
+        for scenario in ["valid", "corrupt", "short"] {
+            let mut bytes = valid.clone();
+            if scenario == "corrupt" {
+                bytes[0] ^= 1;
+            }
+            if scenario == "short" {
+                bytes.pop();
+            }
+            let sample = sample();
+            let mut reader =
+                BitrotReader::new(std::io::Cursor::new(bytes::Bytes::from(bytes)), payload.len(), algo.clone(), false);
+            let mut out = vec![0x23];
+            let result =
+                in_context(Some(&sample), || with_captured_context(reader.read_appending(&mut out, payload.len()))).await;
+            let totals = stats(&sample).totals;
+            if scenario == "valid" {
+                assert_eq!(result.expect("valid shard"), payload.len());
+                assert_eq!(&out[1..], payload);
+                assert_eq!(totals[Stage::Verify.index()].calls, 1);
+                assert_eq!(totals[Stage::Copy.index()].bytes, u64::try_from(payload.len()).expect("small payload"));
+            } else {
+                assert!(result.is_err(), "corrupt and short shards must fail");
+                assert_eq!(out, [0x23], "unverified bytes must not reach output");
+                assert_eq!(totals[Stage::Copy.index()].calls, 0);
+                if scenario == "corrupt" {
+                    assert_eq!(totals[Stage::Verify.index()].failures, 1);
+                }
+            }
+        }
+    }
+}
+
+#[cfg(test)]
+mod cohort_test {
+    use super::*;
+
+    #[test]
+    fn production_cohort_uses_one_decision_for_every_batch() {
+        const CHILD: &str = "RUSTFS_TEST_PIPELINE_COHORT_CHILD";
+        if std::env::var_os(CHILD).is_none() {
+            let result = std::process::Command::new(std::env::current_exe().expect("current test executable"))
+                .args([
+                    "--exact",
+                    "diagnostics::get_pipeline::cohort_test::production_cohort_uses_one_decision_for_every_batch",
+                    "--nocapture",
+                ])
+                .env(CHILD, "1")
+                .env(ENV_SAMPLE_EVERY, "64")
+                .env("RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY", "65536")
+                .status()
+                .expect("run isolated cohort contract");
+            assert!(result.success(), "request and batch sampling contract 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 selected = capture_request("server-cohort-1").expect("first request is selected");
+            assert_eq!((selected.request_sequence, selected.sample_every), (0, 64));
+            let not_selected = capture_request("server-cohort-2");
+            assert!(not_selected.is_none());
+            let batches = in_context(Some(&selected), || {
+                let context =
+                    super::super::batch_read_diagnostics::request_context("server-cohort-1").expect("selected canonical context");
+                assert!(Arc::ptr_eq(&context, &selected.id), "selected ID must reuse the request allocation");
+                assert!(super::super::batch_read_diagnostics::request_context("different-request").is_none());
+                let first = super::super::batch_read_diagnostics::Sample::capture().expect("first selected batch");
+                let second =
+                    super::super::batch_read_diagnostics::Sample::capture().expect("same request must select its second batch");
+                (first, second)
+            });
+            assert!(in_context(not_selected.as_ref(), super::super::batch_read_diagnostics::Sample::capture).is_none());
+            assert!(
+                super::super::batch_read_diagnostics::request_context("server-cohort-2").is_none(),
+                "unselected request must not retain a diagnostic ID"
+            );
+            assert!(current().is_none());
+            assert!(
+                Arc::strong_count(&selected) >= 3,
+                "blocking batch work must retain the request until drained"
+            );
+            drop(batches);
+            assert_eq!(Arc::strong_count(&selected), 1);
+        });
+    }
+}

crates/ecstore/src/diagnostics/mod.rs

@@ -15,5 +15,56 @@
 // #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;
+#[cfg(feature = "diagnostic-get-pipeline")]
+pub(crate) mod get_pipeline;
 pub(crate) mod object_lock;
+
+// The disabled expansion contains neither diagnostic arguments nor calls.
+macro_rules! get_pipeline_sync {
+    ($stage:ident, $bytes:expr, result $operation:expr) => {{
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        {
+            $crate::diagnostics::get_pipeline::measure_sync(
+                $crate::diagnostics::get_pipeline::Stage::$stage,
+                $bytes,
+                $operation,
+                |result| result.is_ok(),
+            )
+        }
+        #[cfg(not(feature = "diagnostic-get-pipeline"))]
+        {
+            ($operation)()
+        }
+    }};
+    ($stage:ident, $bytes:expr, $operation:expr) => {{
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        {
+            $crate::diagnostics::get_pipeline::measure_sync(
+                $crate::diagnostics::get_pipeline::Stage::$stage,
+                $bytes,
+                $operation,
+                |_| true,
+            )
+        }
+        #[cfg(not(feature = "diagnostic-get-pipeline"))]
+        {
+            ($operation)()
+        }
+    }};
+}
+pub(crate) use get_pipeline_sync;
+
+#[cfg(all(test, not(feature = "diagnostic-get-pipeline")))]
+mod pipeline_disabled_tests {
+    #[test]
+    fn disabled_macro_preserves_operations_without_evaluating_diagnostic_arguments() {
+        let result = super::get_pipeline_sync!(Copy, panic!("disabled diagnostics evaluated bytes"), || 17);
+        assert_eq!(result, 17);
+        let result =
+            super::get_pipeline_sync!(Verify, panic!("disabled diagnostics evaluated bytes"), result || Err::<(), _>("original"));
+        assert_eq!(result, Err("original"));
+    }
+}

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/erasure/coding/bitrot.rs

@@ -12,6 +12,7 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 
+use crate::diagnostics::get_pipeline_sync;
 use crate::disk::error::Error as DiskError;
 use crate::erasure::codec::buffer_pool::{get_ec_buffer, return_ec_buffer};
 use pin_project_lite::pin_project;
@@ -176,7 +177,7 @@ where
         if let Some(integrity) = &mut self.integrity {
             integrity.verify(data).await?;
         }
-        out.copy_from_slice(data);
+        get_pipeline_sync!(Copy, data.len(), || out.copy_from_slice(data));
         self.last_verify_duration = verify;
         Ok(want)
     }
@@ -254,21 +255,27 @@ fn split_and_verify<'a>(hash_algo: &HashAlgorithm, skip_verify: bool, block: &'a
     if skip_verify {
         return Ok((data, Duration::ZERO));
     }
-    let verify_start = std::time::Instant::now();
-    let actual_hash = hash_algo.hash_encode(data);
-    let verify = verify_start.elapsed();
-    if actual_hash.as_ref() != hash {
-        error!(
-            event = EVENT_BITROT_HASH_MISMATCH,
-            component = LOG_COMPONENT_ECSTORE,
-            subsystem = LOG_SUBSYSTEM_ERASURE,
-            state = "failed",
-            data_len = data.len(),
-            "bitrot hash mismatch"
-        );
-        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, DiskError::FileCorrupt));
-    }
-    Ok((data, verify))
+    get_pipeline_sync!(
+        Verify,
+        data.len(),
+        result || {
+            let verify_start = std::time::Instant::now();
+            let actual_hash = hash_algo.hash_encode(data);
+            let verify = verify_start.elapsed();
+            if actual_hash.as_ref() != hash {
+                error!(
+                    event = EVENT_BITROT_HASH_MISMATCH,
+                    component = LOG_COMPONENT_ECSTORE,
+                    subsystem = LOG_SUBSYSTEM_ERASURE,
+                    state = "failed",
+                    data_len = data.len(),
+                    "bitrot hash mismatch"
+                );
+                return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, DiskError::FileCorrupt));
+            }
+            Ok((data, verify))
+        }
+    )
 }
 
 impl<R> BitrotReader<R>
@@ -329,7 +336,7 @@ where
 
         if let Some(block) = self.inner.try_take_block(need) {
             let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &block)?;
-            out.extend_from_slice(data);
+            get_pipeline_sync!(Copy, data.len(), || out.extend_from_slice(data));
             self.last_verify_duration = verify;
             return Ok(want);
         }
@@ -401,7 +408,7 @@ where
             if self.chunks.len() == 1 && self.buf.is_empty() {
                 let block = &self.chunks[0];
                 let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, block)?;
-                out.extend_from_slice(data);
+                get_pipeline_sync!(Copy, data.len(), || out.extend_from_slice(data));
                 self.last_verify_duration = verify;
                 return Ok(want);
             }
@@ -413,43 +420,49 @@ where
                     .chain((!self.buf.is_empty()).then_some(self.buf.as_slice()))
             };
             if !self.skip_verify {
-                let verify_start = std::time::Instant::now();
-                let actual_hash = self
-                    .hash_algo
-                    .hash_encode_slices(block_chunks().scan(hash_size, |skip, chunk| {
-                        let start = (*skip).min(chunk.len());
-                        *skip -= start;
-                        Some(&chunk[start..])
-                    }));
-                let verify = verify_start.elapsed();
-                let mut hash_offset = 0;
-                let mut remaining = hash_size;
-                for chunk in block_chunks() {
-                    let take = remaining.min(chunk.len());
-                    if actual_hash.as_ref()[hash_offset..hash_offset + take] != chunk[..take] {
-                        error!(
-                            event = EVENT_BITROT_HASH_MISMATCH,
-                            component = LOG_COMPONENT_ECSTORE,
-                            subsystem = LOG_SUBSYSTEM_ERASURE,
-                            state = "failed",
-                            data_len = want,
-                            "bitrot hash mismatch"
-                        );
-                        return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, DiskError::FileCorrupt));
+                self.last_verify_duration = get_pipeline_sync!(
+                    Verify,
+                    want,
+                    result || {
+                        let verify_start = std::time::Instant::now();
+                        let actual_hash = self
+                            .hash_algo
+                            .hash_encode_slices(block_chunks().scan(hash_size, |skip, chunk| {
+                                let start = (*skip).min(chunk.len());
+                                *skip -= start;
+                                Some(&chunk[start..])
+                            }));
+                        let verify = verify_start.elapsed();
+                        let mut hash_offset = 0;
+                        let mut remaining = hash_size;
+                        for chunk in block_chunks() {
+                            let take = remaining.min(chunk.len());
+                            if actual_hash.as_ref()[hash_offset..hash_offset + take] != chunk[..take] {
+                                error!(
+                                    event = EVENT_BITROT_HASH_MISMATCH,
+                                    component = LOG_COMPONENT_ECSTORE,
+                                    subsystem = LOG_SUBSYSTEM_ERASURE,
+                                    state = "failed",
+                                    data_len = want,
+                                    "bitrot hash mismatch"
+                                );
+                                return Err(std::io::Error::new(std::io::ErrorKind::InvalidData, DiskError::FileCorrupt));
+                            }
+                            hash_offset += take;
+                            remaining -= take;
+                            if remaining == 0 {
+                                break;
+                            }
+                        }
+                        Ok::<_, std::io::Error>(verify)
                     }
-                    hash_offset += take;
-                    remaining -= take;
-                    if remaining == 0 {
-                        break;
-                    }
-                }
-                self.last_verify_duration = verify;
+                )?;
             }
             let mut skip = hash_size;
             for chunk in block_chunks() {
                 let start = skip.min(chunk.len());
                 skip -= start;
-                out.extend_from_slice(&chunk[start..]);
+                get_pipeline_sync!(Copy, chunk.len() - start, || out.extend_from_slice(&chunk[start..]));
             }
             return Ok(want);
         }
@@ -459,7 +472,7 @@ where
         // `copy_from_slice` into a pre-zeroed buffer).
         self.read_scratch_block(need, want).await?;
         let (data, verify) = split_and_verify(&self.hash_algo, self.skip_verify, &self.buf[..need])?;
-        out.extend_from_slice(data);
+        get_pipeline_sync!(Copy, data.len(), || out.extend_from_slice(data));
         self.last_verify_duration = verify;
         Ok(want)
     }

crates/ecstore/src/erasure/coding/decode.rs

@@ -19,6 +19,7 @@ use crate::diagnostics::get::{
     GET_STAGE_STRIPE_READ_QUORUM, GetObjectFailureReason, classify_io_error, get_stage_timer_if_enabled,
     record_get_object_pipeline_failure, record_get_stage_duration_if_enabled,
 };
+use crate::diagnostics::get_pipeline_sync;
 use crate::disk::disk_store::get_object_disk_read_timeout;
 use crate::disk::error::{Error, is_terminal_read_error};
 use crate::disk::error_reduce::reduce_errs;
@@ -2179,7 +2180,13 @@ where
 
         let write_len = write_left.min(block_slice.len());
         let write_stage_start = get_stage_timer_if_enabled(rustfs_io_metrics::get_stage_metrics_enabled());
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let diagnostic = crate::diagnostics::get_pipeline::ElapsedMeasurement::emit();
         if let Err(e) = writer.write_all(&block_slice[..write_len]).await {
+            #[cfg(feature = "diagnostic-get-pipeline")]
+            if let Some(diagnostic) = diagnostic {
+                diagnostic.finish(false, 0);
+            }
             if let Some(write_stage_start) = write_stage_start {
                 rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
             }
@@ -2208,6 +2215,10 @@ where
             }
             return Err(e);
         }
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        if let Some(diagnostic) = diagnostic {
+            diagnostic.finish(true, write_len);
+        }
         if let Some(write_stage_start) = write_stage_start {
             rustfs_io_metrics::record_get_object_duplex_backpressure_duration(write_stage_start.elapsed().as_secs_f64());
         }
@@ -2447,11 +2458,17 @@ impl Erasure {
         } else {
             None
         };
-        let decode_result = if require_surplus_source {
-            self.decode_data_with_reconstruction_verification_for_lockstep(shards)
-        } else {
-            self.decode_data_with_reconstruction_verification(shards)
-        };
+        let decode_result = get_pipeline_sync!(
+            Reconstruct,
+            0,
+            result || {
+                if require_surplus_source {
+                    self.decode_data_with_reconstruction_verification_for_lockstep(shards)
+                } else {
+                    self.decode_data_with_reconstruction_verification(shards)
+                }
+            }
+        );
         if let Err(e) = decode_result {
             record_get_stage_duration_if_enabled(GET_OBJECT_PATH_LEGACY_DUPLEX, GET_STAGE_RECONSTRUCT, reconstruct_stage_start);
             let reason = GetObjectFailureReason::DecodeError;
@@ -3211,6 +3228,60 @@ mod tests {
         assert!(defaulted.verify_reconstruction);
     }
 
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    #[tokio::test]
+    async fn diagnostic_emit_preserves_success_error_and_pending_cancellation() {
+        use crate::diagnostics::get_pipeline::{self as diagnostic, Stage};
+        struct Downstream {
+            pending: bool,
+        }
+        impl AsyncWrite for Downstream {
+            fn poll_write(self: Pin<&mut Self>, _: &mut Context<'_>, _: &[u8]) -> Poll<io::Result<usize>> {
+                if self.pending {
+                    Poll::Pending
+                } else {
+                    Poll::Ready(Err(io::Error::from(io::ErrorKind::BrokenPipe)))
+                }
+            }
+            fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+            fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
+                Poll::Ready(Ok(()))
+            }
+        }
+        let blocks = vec![Some(vec![1, 2, 3, 4])];
+        let sample = diagnostic::test_request();
+        let mut out = Vec::new();
+        let result = diagnostic::in_context(Some(&sample), || {
+            diagnostic::with_captured_context(write_data_blocks(&mut out, &blocks, 1, 0, 4))
+        })
+        .await;
+        assert_eq!(result.expect("emit original bytes"), 4);
+        assert_eq!(out, [1, 2, 3, 4]);
+        assert_eq!(diagnostic::test_total(&sample, Stage::Emit), (1, 4, 0, 1, 0, 0));
+
+        let sample = diagnostic::test_request();
+        let mut downstream = Downstream { pending: false };
+        let error = diagnostic::in_context(Some(&sample), || {
+            diagnostic::with_captured_context(write_data_blocks(&mut downstream, &blocks, 1, 0, 4))
+        })
+        .await
+        .expect_err("closed downstream must remain an error");
+        assert_eq!(error.kind(), io::ErrorKind::BrokenPipe);
+        assert_eq!(diagnostic::test_total(&sample, Stage::Emit), (1, 0, 0, 1, 1, 0));
+
+        let sample = diagnostic::test_request();
+        let mut downstream = Downstream { pending: true };
+        let future = diagnostic::in_context(Some(&sample), || {
+            diagnostic::with_captured_context(write_data_blocks(&mut downstream, &blocks, 1, 0, 4))
+        });
+        let mut future = Box::pin(future);
+        assert!(futures::poll!(future.as_mut()).is_pending());
+        drop(future);
+        assert_eq!(diagnostic::test_total(&sample, Stage::Emit), (1, 0, 0, 1, 0, 1));
+    }
+
     #[tokio::test]
     async fn test_write_data_blocks_writes_range_across_blocks() {
         let blocks = vec![Some(vec![1, 2, 3, 4]), Some(vec![5, 6, 7]), Some(vec![8, 9])];

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 and GET pipeline diagnostics](operations/local-batch-cpu-diagnostics.md) — default-off request-coherent allocation, verification, copy and response 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, sampled local-batch context and request-coherent GET diagnostics. | 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,161 @@
+# Sampled local batch CPU diagnostics
+
+**Use this when:** separating sampled local EC batch allocation/initialization
+and filesystem wall time from worker CPU time, or associating GET verification,
+copy and response stages within one request 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.
+
+## Request-coherent GET stages
+
+`diagnostic-get-pipeline` is a separate default-off build feature. It includes
+`diagnostic-local-batch` and adds one request-level sampling decision at the
+external S3 GET boundary. Use this mode to associate verification, copies,
+reconstruction, producer emit and response-body polls with the same canonical
+response ID. List and rejected GET requests can have no storage stages; never
+classify them as complete object-read samples solely from their method.
+
+```sh
+RUSTUP_TOOLCHAIN=stable cargo build --release --locked -p rustfs --bin rustfs \
+  --features diagnostic-get-pipeline
+export RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY=0
+export RUSTFS_DIAGNOSTIC_GET_STAGE_SAMPLE_EVERY=64
+export RUST_LOG=warn,rustfs_local_batch_diagnostic=trace
+```
+
+The request rate uses the same `1..=65536` bounds and restart requirement.
+When it is nonzero, selected requests select all their batch records; the
+per-batch interval does not independently filter that cohort. Only selected
+requests retain the shared diagnostic ID in request mode; the original batch
+mode retains its existing all-request ID behavior. When the request
+rate is missing/invalid/zero, the original batch mode remains available. The
+rate and target filter must both be enabled to select a request. There are no
+request-stage symbols or calls without the new feature; synchronous probe
+macros discard diagnostic arguments in that build.
+
+A selected request retains its diagnostic context through service polling,
+the legacy producer, response-body polling and selected blocking work. Its
+last owner emits one `get_pipeline_diagnostic` summary on the same isolated
+JSON target. `request_sequence` advances for each eligible canonical GET and
+`sample_every` records the configured interval; selected sequences are multiples
+of that interval. Emission can be reordered by producer drain, so sort sequence
+values and check gaps rather than assuming log order is request order. `stage_totals_json` is a JSON-encoded array in the fixed order
+`verify,copy,reconstruct,emit,response_poll`. Every entry contains `calls`,
+`bytes`, `wall_ns`, `cpu_ns`, `cpu_unavailable`, `failures`, `cancelled`,
+`first_tid`, `last_tid` and `thread_changes`. The TID fields describe
+synchronous observations; changes compare consecutive completed observations,
+not scheduler migrations. An async emit has no thread CPU/TID attribution.
+The summary's `emitter_tid` identifies its final emitter, not the worker for
+every stage.
+Counters saturate with `overflowed=true`; a poisoned accumulator is explicitly
+marked. Neither case is usable for timing attribution.
+
+`verify` measures synchronous bitrot hash/comparison; `copy` measures the
+verified reader-to-output copies, not every copy in the service. Reconstruction
+measures the legacy duplex synchronous decode call, excluding awaits for integrity proof
+preparation. `emit` brackets each async duplex `write_all` and never reads a
+thread CPU clock. Its bytes count only fully completed writes; a failed or
+cancelled write may already have transferred an unknown partial prefix.
+Reconstruction does not report byte accounting in this version. The codec-streaming reconstruction path and copies outside these reader hooks
+are not covered by these reconstruction/copy totals. `response_poll` measures
+one synchronous inner body poll,
+including Pending polls, and counts bytes returned by data frames. It does
+not measure transport/network progress while the body is not polled. Other
+CPU unavailability is explicit; available zero and unavailable are distinct.
+Selected requests execute additional clock reads and accumulator updates.
+That work can itself move selected requests into a slow cohort; selection
+does not make them an unperturbed random control. Treat stage totals as
+diagnostic observations and qualify instrumentation cost separately.
+Stage scopes can overlap, including copies performed within a response poll.
+They must not be added to form a request critical path.
+
+`result=body_complete` means the server body reached EOF or a suppressed-body
+completion. It is not a client acknowledgement. Body errors (including unsuccessful HTTP status), service errors
+and incomplete drops use `body_error`, `service_error` and `cancelled`.
+Completion is idempotent; if cancellation leaves a producer/blocking worker
+alive, its token retains the summary until the work drains. Thus
+`lifetime_wall_ns` can extend beyond body completion and is not an HTTP latency.
+`response_ready_ns`, `first_data_poll_end_ns` and `body_end_ns` are monotonic
+offsets from the captured request start, not network timestamps. The owner
+marks an unfinished request/body cancelled when dropped; the summary can
+remain deferred while producer work drains. Missing boundaries remain absent.
+
+The process admits at most 4096 selected request summaries. It reserves a final
+summary event before accepting a request, and at most 131072 events in request
+mode, budgeting 8192 bytes per fixed-schema event (1 GiB of reserved payload
+budget). Request IDs are bounded ASCII and all other owned fields are fixed
+enums, counters or bounded numeric JSON. The logger can add its standard
+metadata and can write to more than one sink; this budget does not limit other
+logging or exporter output. Batch-event exhaustion marks `truncated=true` in
+the reserved summary. Admission exhaustion stops selecting requests. Reject
+truncated records and runs that reach an admission/event cap as evidence for a
+complete cohort; do not silently treat missing records as zeros.
+
+Keep the new feature, both rates, limits, source, binary and client in separate
+provenance. Unit/contract checks are diagnostic correctness evidence only.
+New Linux workloads still need a separately registered protocol and resource
+reservation. This mode neither selects an I/O optimization nor qualifies
+baseline repeatability or observation overhead.

rustfs/Cargo.toml

@@ -79,6 +79,10 @@ 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"]
+# Default-off request-coherent verification, copy and response diagnostics.
+diagnostic-get-pipeline = ["diagnostic-local-batch", "rustfs-ecstore/diagnostic-get-pipeline"]
 # 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,28 @@ 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 = ();
+#[cfg(feature = "diagnostic-get-pipeline")]
+type PipelineDiagnosticRequest = Option<PipelineRequestOwner>;
+#[cfg(not(feature = "diagnostic-get-pipeline"))]
+type PipelineDiagnosticRequest = ();
+#[cfg(feature = "diagnostic-get-pipeline")]
+struct PipelineRequestOwner(Arc<batch_diagnostic::RequestSample>);
+#[cfg(feature = "diagnostic-get-pipeline")]
+impl PipelineRequestOwner {
+    fn sample(&self) -> &Arc<batch_diagnostic::RequestSample> {
+        &self.0
+    }
+}
+#[cfg(feature = "diagnostic-get-pipeline")]
+impl Drop for PipelineRequestOwner {
+    fn drop(&mut self) {
+        batch_diagnostic::cancel_response(Some(&self.0));
+    }
+}
 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 +309,26 @@ where
         } else {
             req.headers().get(REQUEST_ID_HEADER).cloned()
         };
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let pipeline_request = (is_s3 && req.method() == Method::GET)
+            .then(|| batch_diagnostic::capture_request(&request_context.request_id))
+            .flatten()
+            .map(PipelineRequestOwner);
+        #[cfg(not(feature = "diagnostic-get-pipeline"))]
+        let pipeline_request = ();
+        #[cfg(feature = "diagnostic-local-batch")]
+        let diagnostic_context = || {
+            is_s3
+                .then(|| batch_diagnostic::request_context(&request_context.request_id))
+                .flatten()
+        };
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let diagnostic_context =
+            || batch_diagnostic::in_context(pipeline_request.as_ref().map(PipelineRequestOwner::sample), diagnostic_context);
+        #[cfg(feature = "diagnostic-local-batch")]
+        let diagnostic_request = diagnostic_context();
+        #[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,15 +344,23 @@ where
             .flatten();
         let head_request = req.method() == Method::HEAD;
         let mut metrics = is_s3.then(|| s3_http_request_guard(req.method().as_str()));
-        let inner = match metrics.as_mut() {
+        let call = || match metrics.as_mut() {
             Some(metrics) => metrics.in_scope(|| self.inner.call(req)),
             None => self.inner.call(req),
         };
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let call = || batch_diagnostic::in_context(pipeline_request.as_ref().map(PipelineRequestOwner::sample), call);
+        #[cfg(feature = "diagnostic-local-batch")]
+        let inner = batch_diagnostic::in_request_context(diagnostic_request.as_ref(), call);
+        #[cfg(not(feature = "diagnostic-local-batch"))]
+        let inner = call();
         if let Some(feedback) = feedback.as_mut() {
             feedback.set_class(foreground_class(metrics.as_ref().and_then(S3HttpRequestGuard::operation)));
         }
         ExternalRequestContextFuture {
             inner,
+            diagnostic_request,
+            pipeline_request,
             request_id,
             is_s3,
             metrics,
@@ -358,6 +410,8 @@ pin_project! {
     pub struct ExternalRequestContextFuture<F> {
         #[pin]
         inner: F,
+        diagnostic_request: BatchDiagnosticRequest,
+        pipeline_request: PipelineDiagnosticRequest,
         request_id: Option<HeaderValue>,
         is_s3: bool,
         metrics: Option<S3HttpRequestGuard>,
@@ -374,16 +428,24 @@ where
 
     fn poll(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Self::Output> {
         let this = self.project();
-        let result = match this.metrics.as_mut() {
+        let poll = || match this.metrics.as_mut() {
             Some(metrics) => metrics.in_scope(|| this.inner.poll(cx)),
             None => this.inner.poll(cx),
         };
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let poll = || batch_diagnostic::in_context(this.pipeline_request.as_ref().map(PipelineRequestOwner::sample), poll);
+        #[cfg(feature = "diagnostic-local-batch")]
+        let result = batch_diagnostic::in_request_context(this.diagnostic_request.as_ref(), poll);
+        #[cfg(not(feature = "diagnostic-local-batch"))]
+        let result = poll();
         if let Some(feedback) = this.feedback.as_mut() {
             feedback.set_class(foreground_class(this.metrics.as_ref().and_then(S3HttpRequestGuard::operation)));
         }
         let mut response = match result {
             Poll::Ready(Ok(response)) => response,
             Poll::Ready(Err(error)) => {
+                #[cfg(feature = "diagnostic-get-pipeline")]
+                batch_diagnostic::fail_service(this.pipeline_request.as_ref().map(PipelineRequestOwner::sample));
                 if let Some(metrics) = this.metrics.as_mut() {
                     metrics.service_error();
                 }
@@ -407,18 +469,23 @@ where
             }
         }
 
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        batch_diagnostic::response_ready(this.pipeline_request.as_ref().map(PipelineRequestOwner::sample));
         let status = response.status();
         let suppressed = *this.head_request
             || status.is_informational()
             || status == StatusCode::NO_CONTENT
             || status == StatusCode::NOT_MODIFIED;
         Poll::Ready(Ok(response.map(|body| {
-            ForegroundResponseBody::new(
+            let body = ForegroundResponseBody::new(
                 body,
                 this.feedback.take(),
                 !status.is_client_error() && !status.is_server_error(),
                 suppressed,
-            )
+            );
+            #[cfg(feature = "diagnostic-get-pipeline")]
+            let body = body.with_pipeline_request(this.pipeline_request.take(), suppressed);
+            body
         })))
     }
 }
@@ -452,6 +519,7 @@ pin_project! {
         // Body::is_end_stream only has a shared reference. Empty responses can
         // finish there without being polled; Cell adds no mutex or allocation.
         observation: Cell<Option<ForegroundObservation<'static>>>,
+        pipeline_request: PipelineDiagnosticRequest,
         succeeded: bool,
     }
 }
@@ -467,6 +535,7 @@ impl<B> ForegroundResponseBody<B> {
         let body = Self {
             inner,
             observation: Cell::new(observation),
+            pipeline_request: Default::default(),
             succeeded,
         };
         if suppressed {
@@ -475,7 +544,18 @@ impl<B> ForegroundResponseBody<B> {
         body
     }
 
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    fn with_pipeline_request(mut self, sample: Option<PipelineRequestOwner>, suppressed: bool) -> Self {
+        self.pipeline_request = sample;
+        if suppressed {
+            batch_diagnostic::finish_response(self.pipeline_request.as_ref().map(PipelineRequestOwner::sample), self.succeeded);
+        }
+        self
+    }
+
     fn complete(&self) {
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        batch_diagnostic::finish_response(self.pipeline_request.as_ref().map(PipelineRequestOwner::sample), self.succeeded);
         finish_foreground_observation(
             &self.observation,
             if self.succeeded {
@@ -493,8 +573,22 @@ impl<B: Body> Body for ForegroundResponseBody<B> {
 
     fn poll_frame(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Result<http_body::Frame<Self::Data>, Self::Error>>> {
         let mut this = self.project();
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let result = batch_diagnostic::response_poll(this.pipeline_request.as_ref().map(PipelineRequestOwner::sample), || {
+            let result = this.inner.as_mut().poll_frame(cx);
+            let bytes = match &result {
+                Poll::Ready(Some(Ok(frame))) => frame.data_ref().map_or(0, bytes::Buf::remaining),
+                _ => 0,
+            };
+            let success = !matches!(&result, Poll::Ready(Some(Err(_))));
+            (result, bytes, success)
+        });
+        #[cfg(not(feature = "diagnostic-get-pipeline"))]
         let result = this.inner.as_mut().poll_frame(cx);
-        if this.observation.get_mut().is_none() {
+        let observed = this.observation.get_mut().is_some();
+        #[cfg(feature = "diagnostic-get-pipeline")]
+        let observed = observed || this.pipeline_request.is_some();
+        if !observed {
             return result;
         }
         let outcome = match &result {
@@ -512,6 +606,11 @@ impl<B: Body> Body for ForegroundResponseBody<B> {
             _ => None,
         };
         if let Some(outcome) = outcome {
+            #[cfg(feature = "diagnostic-get-pipeline")]
+            batch_diagnostic::finish_response(
+                this.pipeline_request.as_ref().map(PipelineRequestOwner::sample),
+                matches!(outcome, CompletionOutcome::Success),
+            );
             finish_foreground_observation(this.observation, outcome);
         }
         result
@@ -2608,6 +2707,255 @@ fn rewrite_double_slash_root(uri: &Uri) -> Option<Uri> {
 
 #[cfg(test)]
 mod tests {
+    #[cfg(feature = "diagnostic-get-pipeline")]
+    mod diagnostic_pipeline {
+        use super::*;
+        use std::io::Write;
+        use std::sync::Mutex as StdMutex;
+        use tracing_subscriber::fmt::MakeWriter;
+
+        #[derive(Clone, Default)]
+        struct Output(Arc<StdMutex<Vec<u8>>>);
+        struct Writer(Output);
+        impl Write for Writer {
+            fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+                self.0.0.lock().expect("capture diagnostic JSON").extend_from_slice(bytes);
+                Ok(bytes.len())
+            }
+            fn flush(&mut self) -> std::io::Result<()> {
+                Ok(())
+            }
+        }
+        impl<'a> MakeWriter<'a> for Output {
+            type Writer = Writer;
+            fn make_writer(&'a self) -> Self::Writer {
+                Writer(self.clone())
+            }
+        }
+        struct ScriptBody {
+            step: u8,
+            error: bool,
+        }
+        impl Body for ScriptBody {
+            type Data = Bytes;
+            type Error = &'static str;
+            fn poll_frame(
+                self: Pin<&mut Self>,
+                cx: &mut Context<'_>,
+            ) -> Poll<Option<Result<http_body::Frame<Bytes>, Self::Error>>> {
+                let this = self.get_mut();
+                this.step += 1;
+                match this.step {
+                    1 => {
+                        cx.waker().wake_by_ref();
+                        Poll::Pending
+                    }
+                    2 => Poll::Ready(Some(Ok(http_body::Frame::data(Bytes::from_static(b"body-bytes"))))),
+                    3 if this.error => Poll::Ready(Some(Err("original-body-error"))),
+                    _ => Poll::Ready(None),
+                }
+            }
+            fn is_end_stream(&self) -> bool {
+                self.step >= 3
+            }
+        }
+
+        #[test]
+        fn canonical_response_completion_error_and_drop_are_preserved() {
+            const CHILD: &str = "RUSTFS_TEST_PIPELINE_HTTP_CHILD";
+            if std::env::var_os(CHILD).is_none() {
+                let status = std::process::Command::new(std::env::current_exe().expect("current test binary"))
+                    .args([
+                        "--exact",
+                        "server::layer::tests::diagnostic_pipeline::canonical_response_completion_error_and_drop_are_preserved",
+                        "--nocapture",
+                    ])
+                    .env(CHILD, "1")
+                    .env("RUSTFS_DIAGNOSTIC_GET_STAGE_SAMPLE_EVERY", "1")
+                    .env("RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY", "0")
+                    .status()
+                    .expect("run isolated diagnostic HTTP contract");
+                assert!(status.success(), "diagnostic HTTP contract failed");
+                return;
+            }
+            for scenario in ["complete", "error", "drop", "service-error", "suppressed", "status-error"] {
+                let output = Output::default();
+                let subscriber = tracing_subscriber::fmt()
+                    .json()
+                    .flatten_event(true)
+                    .with_current_span(false)
+                    .with_span_list(false)
+                    .with_max_level(tracing::Level::TRACE)
+                    .with_writer(output.clone())
+                    .finish();
+                let mut canonical_id = None;
+                tracing::subscriber::with_default(subscriber, || {
+                    let handler = tower::service_fn(move |_: HttpRequest<Full<Bytes>>| async move {
+                        if scenario == "service-error" {
+                            return Err("original-service-error");
+                        }
+                        let mut response = Response::new(ScriptBody {
+                            step: 0,
+                            error: scenario == "error",
+                        });
+                        if scenario == "suppressed" {
+                            *response.status_mut() = StatusCode::NO_CONTENT;
+                        }
+                        if scenario == "status-error" {
+                            *response.status_mut() = StatusCode::INTERNAL_SERVER_ERROR;
+                        }
+                        Ok(response)
+                    });
+                    let mut service = ExternalRequestContextLayer::new(false).layer(handler);
+                    let request = HttpRequest::builder()
+                        .method(Method::GET)
+                        .uri("/private-object?signature=private")
+                        .header(REQUEST_ID_HEADER, "client-controlled")
+                        .body(Full::new(Bytes::new()))
+                        .expect("test request");
+                    let response = futures::executor::block_on(service.call(request));
+                    if scenario == "service-error" {
+                        assert_eq!(response.err(), Some("original-service-error"));
+                    } else {
+                        let response = response.expect("original response");
+                        canonical_id = Some(response.headers()[AMZ_REQUEST_ID].to_str().expect("canonical ID").to_owned());
+                        assert_ne!(canonical_id.as_deref(), Some("client-controlled"));
+                        if matches!(scenario, "drop" | "suppressed") {
+                            drop(response);
+                        } else {
+                            assert!(output.0.lock().expect("summary before EOF").is_empty());
+                            let result = futures::executor::block_on(response.into_body().collect());
+                            if scenario == "error" {
+                                assert_eq!(result.err(), Some("original-body-error"));
+                            } else {
+                                assert_eq!(result.expect("original body bytes").to_bytes(), Bytes::from_static(b"body-bytes"));
+                            }
+                        }
+                    }
+                });
+                let bytes = output.0.lock().expect("recorded diagnostic JSON");
+                let text = std::str::from_utf8(&bytes).expect("UTF-8 diagnostic records");
+                let rows: Vec<serde_json::Value> = text
+                    .lines()
+                    .map(|l| serde_json::from_str(l).expect("diagnostic JSON"))
+                    .collect();
+                assert_eq!(rows.len(), 1, "one final summary for {scenario}: {text}");
+                let row = &rows[0];
+                assert_eq!(row["event"], "get_pipeline_diagnostic");
+                assert!(row["body_end_ns"].is_u64(), "EOF/error/drop must retain its boundary");
+                if scenario != "service-error" {
+                    assert!(row["response_ready_ns"].is_u64());
+                }
+                assert_eq!(
+                    row["result"],
+                    match scenario {
+                        "complete" | "suppressed" => "body_complete",
+                        "error" | "status-error" => "body_error",
+                        "drop" => "cancelled",
+                        _ => "service_error",
+                    }
+                );
+                if let Some(id) = canonical_id {
+                    assert_eq!(row["request_id"], id);
+                }
+                let totals: Vec<serde_json::Value> =
+                    serde_json::from_str(row["stage_totals_json"].as_str().expect("stage schema")).expect("stage totals");
+                if matches!(scenario, "complete" | "error" | "status-error") {
+                    assert_eq!(totals[4]["calls"], 3);
+                    assert_eq!(totals[4]["bytes"], 10);
+                    assert_eq!(totals[4]["failures"], u64::from(scenario == "error"));
+                }
+                for forbidden in [
+                    "private-object",
+                    "signature",
+                    "client-controlled",
+                    "original-body-error",
+                    "original-service-error",
+                ] {
+                    assert!(!text.contains(forbidden), "request and error payload must not enter diagnostic records");
+                }
+            }
+        }
+    }
+
+    #[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")),
+            pipeline_request: Default::default(),
+            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,22 @@ 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(feature = "diagnostic-get-pipeline")]
+    pub(crate) use rustfs_ecstore::api::diagnostic::{
+        RequestSample, cancel_response, capture_request, fail_service, finish_response, in_context, response_poll, response_ready,
+    };
+    #[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,18 @@ 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;
+            #[cfg(feature = "diagnostic-get-pipeline")]
+            pub(crate) use crate::storage::storage_api::ecstore_diagnostic::{
+                RequestSample, cancel_response, capture_request, fail_service, finish_response, in_context, response_poll,
+                response_ready,
+            };
+            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,8 @@ repo_root="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
 cd "$repo_root"
 
 checked_files=(
+  "crates/ecstore/src/diagnostics/batch_read_diagnostics.rs"
+  "crates/ecstore/src/diagnostics/get_pipeline.rs"
   "rustfs/src/main.rs"
   "rustfs/src/startup_entrypoint.rs"
   "rustfs/src/init.rs"