fix(ecstore): retain diagnostic IDs across GET producer tasks

Capture request context before spawning the legacy producer. Keep the existing producer, cancellation and read-lock lifetime intact.

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

crates/ecstore/Cargo.toml

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

crates/ecstore/src/api/mod.rs

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

crates/ecstore/src/core/pools.rs

@@ -28300,167 +28300,6 @@ mod tests {
         );
     }
 
-    /// One version commits, then every in-place attempt to copy the next one
-    /// times out on the target namespace lock. The entry is deferred, not
-    /// failed, with both source versions retained; the resumed entry finds the
-    /// committed version and moves both with their exact bytes.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_defers_target_lookup_lock_timeout_and_retries_without_losing_source() {
-        use tokio::io::AsyncReadExt as _;
-
-        let fixture = NullVersionDecommission::new(NullVersionTop::NewerVersion).await;
-        let object = fixture.object;
-        let injected = Arc::new(AtomicUsize::new(0));
-        let attempted = Arc::new(AtomicUsize::new(0));
-        let fault = DecommissionTestFaultGuard::install_with_error(
-            Arc::new({
-                let injected = Arc::clone(&injected);
-                let attempted = Arc::clone(&attempted);
-                move |stage, _, faulted, _, success| {
-                    let target = stage == "before_target_commit" && faulted == object && success;
-                    let should_inject = target && attempted.fetch_add(1, Ordering::SeqCst) > 0;
-                    if should_inject {
-                        injected.fetch_add(1, Ordering::SeqCst);
-                    }
-                    should_inject
-                }
-            }),
-            Some(Arc::new(|_, bucket, object| {
-                data_movement::data_movement_stage_error_for_test(
-                    "decommission_object",
-                    "prepare_put_object",
-                    bucket,
-                    object,
-                    Error::Lock(rustfs_lock::LockError::timeout(
-                        format!("{bucket}/{object}@latest"),
-                        std::time::Duration::from_secs(5),
-                    )),
-                )
-            })),
-        );
-        let (outcome, _) = fixture.run_entry().await;
-        assert_eq!(injected.load(Ordering::SeqCst), DECOMMISSION_VERSION_COPY_ATTEMPTS);
-        fixture.assert_source_retained().await;
-        assert_eq!(
-            fixture
-                .target_versions()
-                .await
-                .expect("the first version was copied")
-                .versions
-                .len(),
-            1,
-            "one version must commit before the next version times out"
-        );
-        assert!(
-            matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
-            "{outcome:?}"
-        );
-        assert_eq!(
-            fixture.store.pool_meta.read().await.pools[0]
-                .decommission
-                .as_ref()
-                .expect("active decommission")
-                .items_decommission_failed,
-            0,
-            "a deferred entry must not be counted as a permanent failure"
-        );
-        drop(fault);
-        let (outcome, _) = fixture.run_entry().await;
-        assert!(matches!(outcome, DecommissionEntryOutcome::Complete), "{outcome:?}");
-        fixture.assert_moved().await;
-        let mut reader = fixture.store.pools[1]
-            .get_object_reader(
-                fixture.bucket,
-                fixture.object,
-                None,
-                HeaderMap::new(),
-                &ObjectOptions {
-                    version_id: Some(fixture.seeded.versions[0].version_id.expect("newer version").to_string()),
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("the newer version must remain readable");
-        let mut actual = Vec::new();
-        reader.stream.read_to_end(&mut actual).await.expect("read the newer version");
-        assert_eq!(actual, b"written after versioning was enabled");
-    }
-
-    /// A namespace lock still held on the target key, as by a copy whose node
-    /// was killed mid-write, makes target selection fail with a typed
-    /// object-lock timeout. A real held lock, not an injected error, must defer
-    /// the entry rather than fail it; once the lock is released selection
-    /// recovers and the resumed entry moves both versions.
-    #[tokio::test]
-    #[serial_test::serial]
-    async fn decommission_target_selection_returns_timeout_for_a_held_object_lock() {
-        temp_env::async_with_vars([(rustfs_config::ENV_OBJECT_LOCK_ACQUIRE_TIMEOUT, Some("1"))], async {
-            let fixture = NullVersionDecommission::new(NullVersionTop::NewerVersion).await;
-            let target_set = fixture.store.pools[1].get_disks_by_key(fixture.object);
-            let lock = target_set
-                .new_ns_lock(fixture.bucket, fixture.object)
-                .await
-                .expect("target lock");
-            let guard = lock
-                .get_write_lock(std::time::Duration::from_secs(30))
-                .await
-                .expect("hold target lock");
-            let opts = ObjectOptions {
-                data_movement: true,
-                src_pool_idx: 0,
-                version_id: Some(uuid::Uuid::nil().to_string()),
-                ..Default::default()
-            };
-            let select = || {
-                fixture
-                    .store
-                    .select_data_movement_pool_idx(fixture.bucket, fixture.object, 1, &opts, false)
-            };
-            let result = tokio::time::timeout(get_lock_acquire_timeout() + std::time::Duration::from_secs(15), select())
-                .await
-                .expect("target selection must terminate");
-            match result {
-                Err(Error::Lock(rustfs_lock::LockError::Timeout { resource, .. })) => {
-                    assert!(resource.contains(fixture.bucket) && resource.contains(fixture.object), "{resource}");
-                }
-                other => panic!("expected an object lock timeout, got {other:?}"),
-            }
-
-            let (outcome, _) = fixture.run_entry().await;
-            assert!(
-                matches!(&outcome, DecommissionEntryOutcome::Deferred(err) if is_decommission_copy_lock_timeout_deferral(err)),
-                "{outcome:?}"
-            );
-            assert_eq!(
-                fixture.store.pool_meta.read().await.pools[0]
-                    .decommission
-                    .as_ref()
-                    .expect("active decommission")
-                    .items_decommission_failed,
-                0,
-                "a held target lock must not be counted as a permanent failure"
-            );
-            fixture.assert_source_retained().await;
-            assert!(
-                fixture.target_versions().await.is_none(),
-                "no version may commit while the target lock is held"
-            );
-
-            drop(guard);
-            assert_eq!(
-                select()
-                    .await
-                    .expect("target selection must recover once the lock is released"),
-                1
-            );
-            let (outcome, _) = fixture.run_entry().await;
-            assert!(matches!(outcome, DecommissionEntryOutcome::Complete), "{outcome:?}");
-            fixture.assert_moved().await;
-        })
-        .await;
-    }
-
     #[test]
     fn decommission_object_attempt_metric_keeps_the_peak() {
         let mut attempts = 0;

crates/ecstore/src/diagnostics/batch_read_diagnostics.rs

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

crates/ecstore/src/diagnostics/mod.rs

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

crates/ecstore/src/disk/local.rs

@@ -11309,34 +11309,81 @@ impl LocalDisk {
 #[cfg(unix)]
 pub(crate) async fn batch_shard_pread(requests: Vec<(std::path::PathBuf, usize, usize)>) -> Vec<Result<Bytes>> {
     let n = requests.len();
+    #[cfg(feature = "diagnostic-local-batch")]
+    let diagnostic = crate::diagnostics::batch_read_diagnostics::Sample::capture();
     tokio::task::spawn_blocking(move || {
         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 mut results = Vec::with_capacity(n);
-        for (file_path, offset, length) in requests {
+        for (_index, (file_path, offset, length)) in requests.into_iter().enumerate() {
+            let mut total = 0usize;
             let r = (|| -> Result<Bytes> {
-                let meta = std::fs::metadata(&file_path).map_err(DiskError::from)?;
+                let meta = diagnostic_stage!(Metadata, _index, || 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 = std::fs::File::open(&file_path).map_err(DiskError::from)?;
-                let mut buf = vec![0u8; length];
-                let mut total = 0usize;
-                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;
+                let file = diagnostic_stage!(Open, _index, || 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 {
+                    diagnostic_stage!(Read, _index, || {
+                        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;
+                        }
+                        Ok::<_, DiskError>(())
+                    })?;
                 }
                 Ok(Bytes::from(buf))
             })();
+            #[cfg(feature = "diagnostic-local-batch")]
+            if let Some(sample) = &diagnostic {
+                sample.shard_result(_index, length, total, r.is_ok());
+            }
             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),
+            );
+        }
         results
     })
     .await

crates/ecstore/src/set_disk/mod.rs

@@ -1024,19 +1024,6 @@ impl GetObjectFileInfo {
         self.as_parts().0
     }
 
-    /// Marks an exact-version selection as the object's latest version after
-    /// the versions above it were skipped as dangling.
-    fn mark_selected_as_latest(&mut self) {
-        if let Some(OwnedGetObjectFileInfo {
-            payload: OwnedGetObjectFileInfoPayload::Direct { fi, .. },
-            ..
-        }) = self.owned.as_mut()
-        {
-            fi.is_latest = true;
-            fi.successor_mod_time = None;
-        }
-    }
-
     fn parts_metadata(&self) -> &[FileInfo] {
         self.as_parts().1
     }

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/ecstore/src/set_disk/read.rs

@@ -264,11 +264,6 @@ enum MetadataReadPurpose {
     GetObjectReader,
 }
 
-/// Upper bound on exact-version reads a latest read spends looking past
-/// dangling versions before it keeps the read-quorum error; each read is one
-/// metadata fan-out.
-const MAX_DANGLING_LATEST_VERSIONS: usize = 8;
-
 /// How a metadata read-quorum failure is reported when the replies prove the
 /// object or version dangling. Only the object read entry points
 /// (`get_object_info`, `get_object_reader` and prepared GET metadata) report
@@ -976,18 +971,6 @@ impl SetDisks {
             {
                 Ok(selection) => selection,
                 Err(err) => {
-                    if err == DiskError::ErasureReadQuorum
-                        && vid.is_empty()
-                        && dangling_read == DanglingRead::ReportNotFound
-                        && metadata_fanout_complete
-                        && !crate::bucket::utils::is_meta_bucketname(bucket)
-                        && let Some(snapshot) =
-                            Box::pin(self.read_latest_past_dangling_versions(bucket, object, opts, read_data, purpose, &disks))
-                                .await
-                    {
-                        rustfs_io_metrics::record_get_object_metadata_quorum_result(metadata_metrics_path, "reached");
-                        return Ok(snapshot);
-                    }
                     let err = StorageError::from(err);
                     rustfs_io_metrics::record_get_object_metadata_quorum_result(
                         metadata_metrics_path,
@@ -1068,168 +1051,6 @@ impl SetDisks {
         Ok(snapshot)
     }
 
-    /// A latest read whose split cannot reach read quorum may still have a
-    /// readable older version under newer versions that never committed;
-    /// MinIO resolves this in `pickLatestQuorumFilesInfo`. With every drive
-    /// answering, try versions newest first through exact-version reads and
-    /// serve the readable one with the newest selected modification time. A
-    /// version whose exact read proves it dangling (the NoSuchVersion an
-    /// explicit GET reports) is skipped and queued for heal, which owns its
-    /// cleanup; any other outcome keeps the read-quorum error.
-    async fn read_latest_past_dangling_versions(
-        &self,
-        bucket: &str,
-        object: &str,
-        opts: &ObjectOptions,
-        read_data: bool,
-        purpose: MetadataReadPurpose,
-        disks: &[Option<DiskStore>],
-    ) -> Option<GetObjectFileInfo> {
-        let disk_object = rustfs_utils::path::encode_dir_object(object);
-        let (raw, errs) = Self::read_all_raw_file_info(disks, bucket, &disk_object, false).await;
-        if errs
-            .iter()
-            .any(|err| err.as_ref().is_some_and(|err| !matches!(err, DiskError::FileNotFound)))
-        {
-            return None;
-        }
-        // Stacks are newest first, so only each drive's first few headers can
-        // matter. Keep a version's newest modification time across drives: a
-        // rewritten null version can carry different times on different drives.
-        let mut newest = std::collections::HashMap::<Option<uuid::Uuid>, Option<time::OffsetDateTime>>::new();
-        let mut truncated = false;
-        let mut metadata = Vec::with_capacity(raw.len());
-        for raw in raw.into_iter().flatten() {
-            let mut meta = rustfs_filemeta::FileMeta::load(&raw.buf).ok()?;
-            let mut versions = meta
-                .versions
-                .iter()
-                .enumerate()
-                .filter(|(_, version)| !version.header.free_version());
-            let mut retained_end = 0;
-            for (index, version) in versions.by_ref().take(MAX_DANGLING_LATEST_VERSIONS) {
-                let mod_time = newest.entry(version.header.version_id).or_insert(version.header.mod_time);
-                *mod_time = (*mod_time).max(version.header.mod_time);
-                retained_end = index + 1;
-            }
-            truncated |= versions.next().is_some();
-            meta.versions.truncate(retained_end);
-            metadata.push(meta);
-        }
-        let mut candidates = newest.into_iter().collect::<Vec<_>>();
-        candidates.sort_by(|left, right| right.1.cmp(&left.1).then_with(|| left.0.cmp(&right.0)));
-
-        // Use the stored header for equal-time ordering: re-encoding a selected
-        // FileInfo can change legacy signatures or omit part-checksum metadata.
-        let selected_header = |snapshot: &GetObjectFileInfo| {
-            let mut selected = snapshot.fi().clone();
-            selected.data = None;
-            let version_id = selected.version_id.unwrap_or_else(uuid::Uuid::nil).to_string();
-            let include_part_checksums =
-                opts.include_part_checksums || opts.part_number.is_some() || opts.data_movement || opts.raw_data_movement_read;
-            metadata.iter().find_map(|meta| {
-                let version = meta
-                    .versions
-                    .iter()
-                    .filter(|version| !version.header.free_version())
-                    .take(MAX_DANGLING_LATEST_VERSIONS)
-                    .find(|version| {
-                        version.header.version_id.unwrap_or_else(uuid::Uuid::nil)
-                            == selected.version_id.unwrap_or_else(uuid::Uuid::nil)
-                            && version.header.mod_time == selected.mod_time
-                    })?;
-                let mut original = if include_part_checksums {
-                    meta.into_fileinfo(&selected.volume, &selected.name, &version_id, false, opts.incl_free_versions, true)
-                } else {
-                    meta.into_fileinfo_without_part_checksums(
-                        &selected.volume,
-                        &selected.name,
-                        &version_id,
-                        false,
-                        opts.incl_free_versions,
-                    )
-                }
-                .ok()?;
-                if !include_part_checksums {
-                    rustfs_utils::http::remove_str(&mut original.metadata, rustfs_utils::http::SUFFIX_PART_CHECKSUMS);
-                }
-                original.data = None;
-                original.is_latest = selected.is_latest;
-                original.successor_mod_time = selected.successor_mod_time;
-                original.num_versions = selected.num_versions;
-                (original == selected).then_some(version.header.clone())
-            })
-        };
-
-        let mut best: Option<(Option<time::OffsetDateTime>, GetObjectFileInfo)> = None;
-        let mut skipped = Vec::new();
-        let mut settled = false;
-        for (reads, (version_id, newest_mod_time)) in candidates.into_iter().enumerate() {
-            if best.as_ref().is_some_and(|(selected, _)| newest_mod_time < *selected) {
-                settled = true;
-                break;
-            }
-            if reads == MAX_DANGLING_LATEST_VERSIONS {
-                return None;
-            }
-            // The nil UUID addresses the null version slot at this layer.
-            let version_id = version_id.unwrap_or_else(uuid::Uuid::nil).to_string();
-            let exact = ObjectOptions {
-                version_id: Some(version_id.clone()),
-                ..opts.clone()
-            };
-            match Box::pin(self.get_object_fileinfo_gated_inner(
-                bucket,
-                object,
-                &exact,
-                read_data,
-                false,
-                purpose,
-                DanglingRead::ReportNotFound,
-            ))
-            .await
-            {
-                Ok(snapshot) => {
-                    let selected = snapshot.fi().mod_time;
-                    let replace = match best.as_ref() {
-                        Some((current, previous)) if selected == *current => {
-                            selected_header(&snapshot)?.sorts_before(&selected_header(previous)?)
-                        }
-                        Some((current, _)) => selected > *current,
-                        None => true,
-                    };
-                    if replace {
-                        best = Some((selected, snapshot));
-                    }
-                }
-                Err(StorageError::VersionNotFound(..)) => skipped.push(version_id),
-                Err(_) => return None,
-            }
-        }
-        // Headers past each drive's first few were not compared; only a walk
-        // that proved no remaining candidate can be newer may stop early.
-        if !settled && truncated {
-            return None;
-        }
-        let (_, mut snapshot) = best?;
-        snapshot.mark_selected_as_latest();
-        if !opts.suppress_read_repair {
-            for version_id in &skipped {
-                submit_read_repair_heal(
-                    bucket,
-                    object,
-                    Some(version_id),
-                    self.pool_index,
-                    self.set_index,
-                    None,
-                    "dangling_latest_version",
-                )
-                .await;
-            }
-        }
-        Some(snapshot)
-    }
-
     #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_info_fileinfo_and_quorum(
         &self,
@@ -9940,13 +9761,9 @@ mod dangling_read_tests {
     const PARITY: usize = 2;
 
     async fn dangling_test_set(bucket: &str) -> (Vec<TempDir>, Arc<SetDisks>) {
-        dangling_test_set_with(bucket, DRIVES, PARITY).await
-    }
-
-    async fn dangling_test_set_with(bucket: &str, drives: usize, parity: usize) -> (Vec<TempDir>, Arc<SetDisks>) {
         let ctx = Arc::new(crate::runtime::instance::InstanceContext::new());
         ctx.update_erasure_type(crate::layout::endpoints::SetupType::Erasure).await;
-        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(drives, parity, ctx).await;
+        let (dirs, set) = crate::ecstore_validation_blackbox::make_local_set_disks_with_ctx(DRIVES, PARITY, ctx).await;
         set.make_bucket(bucket, &MakeBucketOptions::default())
             .await
             .expect("test bucket should be created on every drive");
@@ -9960,9 +9777,6 @@ mod dangling_read_tests {
             &mut PutObjReader::from_vec(body.to_vec()),
             &ObjectOptions {
                 versioned,
-                // Wait for every drive so tests that edit one drive's metadata
-                // do not race the PUT's background tail writes.
-                write_completion: crate::object_api::WriteCompletion::TailDrained,
                 ..Default::default()
             },
         )
@@ -9982,25 +9796,6 @@ mod dangling_read_tests {
         }
     }
 
-    /// Drops `version_id` from `disks`, modelling drives a failed PUT never
-    /// reached while the older versions below it stay intact.
-    async fn drop_version_from(set: &SetDisks, bucket: &str, object: &str, version_id: Uuid, disks: std::ops::Range<usize>) {
-        let all = set.disks.read().await.clone();
-        for disk in &all[disks] {
-            let disk = disk.as_ref().expect("every drive should be online");
-            let mut remove = FileInfo {
-                volume: bucket.to_string(),
-                name: object.to_string(),
-                version_id: Some(version_id),
-                ..Default::default()
-            };
-            remove.set_skip_tier_free_version();
-            disk.delete_version(bucket, object, remove, false, DeleteOptions::default())
-                .await
-                .expect("dropping the version should model a drive that never received it");
-        }
-    }
-
     fn read_opts(version_id: Option<&str>) -> ObjectOptions {
         ObjectOptions {
             version_id: version_id.map(str::to_owned),
@@ -10160,295 +9955,12 @@ mod dangling_read_tests {
         head(&set, bucket, object, Some(&v1))
             .await
             .expect("HEAD of the committed version must succeed");
-        // The latest read sees version two on three drives and version one on
-        // the rest. Version two's exact read proves it dangling, so the latest
-        // read serves version one, the newest version an exact read can serve.
-        let latest = head(&set, bucket, object, None)
-            .await
-            .expect("the latest read must serve the newest readable version");
-        assert_eq!(latest.version_id.map(|id| id.to_string()).as_deref(), Some(v1.as_str()));
-        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"version one");
-    }
-
-    #[tokio::test]
-    async fn latest_read_skips_dangling_versions_above_a_committed_version() {
-        // EC 12+4: a failed PUT that left version two on 5..=11 drives keeps
-        // version one, present in every stack, below its read quorum of twelve
-        // when only the top of each stack is read. Up to seven copies the
-        // listing quorum already shows version one; from eight it lists the
-        // dangling version two.
-        let bucket = "dangling-latest-12-4";
-        let (_dirs, set) = dangling_test_set_with(bucket, 16, 4).await;
-        for copies in 5..=11 {
-            let object = format!("torn/latest-{copies:02}");
-            let v1 = put(&set, bucket, &object, b"committed version", true)
-                .await
-                .version_id
-                .expect("versioned PUT should assign a version id");
-            let v2 = put(&set, bucket, &object, b"failed overwrite", true)
-                .await
-                .version_id
-                .expect("versioned PUT should assign a version id");
-            drop_version_from(&set, bucket, &object, v2, copies..16).await;
-
-            let context = format!("version two on {copies} of 16 drives");
-            let latest = head(&set, bucket, &object, None)
-                .await
-                .unwrap_or_else(|err| panic!("{context}: HEAD must serve version one: {err:?}"));
-            assert_eq!(latest.version_id, Some(v1), "{context}");
-            assert!(latest.is_latest, "{context}: the served version is the latest one");
-            assert_eq!(
-                get(&set, bucket, &object, None)
-                    .await
-                    .unwrap_or_else(|err| panic!("{context}: GET must serve version one: {err:?}")),
-                b"committed version",
-                "{context}"
-            );
-            let Err(err) = head(&set, bucket, &object, Some(&v2.to_string())).await else {
-                panic!("{context}: the dangling version must not be served");
-            };
-            assert!(matches!(err, StorageError::VersionNotFound(..)), "{context}: {err:?}");
-        }
-    }
-
-    #[tokio::test]
-    async fn latest_read_keeps_quorum_error_when_the_newest_version_is_uncertain() {
-        // Version two is on every drive, but a torn metadata update left two
-        // identities at one modification time, three copies each. Its exact
-        // read fails with a quorum error rather than NoSuchVersion, so the
-        // walk must not fall through to version one.
-        let bucket = "dangling-latest-uncertain";
-        let object = "torn/latest";
-        let (_dirs, set) = dangling_test_set(bucket).await;
-        put(&set, bucket, object, b"version one", true).await;
-        let v2 = put(&set, bucket, object, b"version two", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id")
-            .to_string();
-        let disks = set.disks.read().await.clone();
-        for disk in disks.iter().take(3) {
-            let disk = disk.as_ref().expect("every drive should be online");
-            let mut fi = disk
-                .read_version("", bucket, object, &v2, &ReadOptions::default())
-                .await
-                .expect("version two should be on every drive");
-            fi.metadata.insert("x-amz-meta-torn".to_string(), "half-applied".to_string());
-            disk.update_metadata(bucket, object, fi, &crate::disk::UpdateMetadataOpts::default())
-                .await
-                .expect("a half-applied metadata update should be stored");
-        }
-
-        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "torn metadata update on version two").await;
-    }
-
-    #[tokio::test]
-    async fn latest_read_serves_a_committed_null_version_below_a_dangling_version() {
-        // A null version written while versioning was suspended sits between
-        // version one and a dangling version three. The walk reaches the null
-        // slot through the nil UUID and serves it, not version one.
-        let bucket = "dangling-latest-null";
-        let object = "torn/latest";
-        let (_dirs, set) = dangling_test_set(bucket).await;
-        let v1 = put(&set, bucket, object, b"version one", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id");
-        set.put_object(
-            bucket,
-            object,
-            &mut PutObjReader::from_vec(b"null version".to_vec()),
-            &ObjectOptions {
-                version_suspended: true,
-                write_completion: crate::object_api::WriteCompletion::TailDrained,
-                ..Default::default()
-            },
-        )
-        .await
-        .expect("a suspended-versioning PUT should commit the null version");
-        let v3 = put(&set, bucket, object, b"version three", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id");
-        drop_version_from(&set, bucket, object, v3, 3..DRIVES).await;
-
-        let latest = head(&set, bucket, object, None)
-            .await
-            .expect("the committed null version must be served");
-        assert_ne!(latest.version_id, Some(v1));
-        assert_ne!(latest.version_id, Some(v3));
-        assert!(latest.is_latest);
-        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"null version");
-    }
-
-    #[tokio::test]
-    async fn latest_read_orders_versions_by_their_newest_modification_time() {
-        // Drive 0 missed the suspended-versioning rewrite of the null version,
-        // so its null copy predates version one while every other drive holds
-        // the committed rewrite. Ordering by the first drive's time would try
-        // version one before the null version and serve it.
-        let bucket = "dangling-latest-order";
-        let object = "torn/latest";
-        let (dirs, set) = dangling_test_set(bucket).await;
-        let suspended = ObjectOptions {
-            version_suspended: true,
-            write_completion: crate::object_api::WriteCompletion::TailDrained,
-            ..Default::default()
+        // The latest read sees two generations (version two on three drives,
+        // version one on the rest), which is not proof of absence.
+        let Err(err) = head(&set, bucket, object, None).await else {
+            panic!("a split latest version must not resolve without quorum");
         };
-        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"stale null".to_vec()), &suspended)
-            .await
-            .expect("the first null version should commit");
-        let v1 = put(&set, bucket, object, b"version one", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id");
-        let drive_zero = object_dir(&dirs, 0, bucket, object).join(STORAGE_FORMAT_FILE);
-        let stale = tokio::fs::read(&drive_zero)
-            .await
-            .expect("drive zero metadata should be readable");
-        set.put_object(bucket, object, &mut PutObjReader::from_vec(b"committed null".to_vec()), &suspended)
-            .await
-            .expect("the null rewrite should commit");
-        tokio::fs::write(&drive_zero, stale)
-            .await
-            .expect("drive zero should keep the pre-rewrite stack");
-        let v5 = put(&set, bucket, object, b"version five", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id");
-        drop_version_from(&set, bucket, object, v5, 3..DRIVES).await;
-
-        let latest = head(&set, bucket, object, None)
-            .await
-            .expect("the committed null rewrite must be served");
-        assert_ne!(latest.version_id, Some(v1));
-        assert_eq!(get(&set, bucket, object, None).await.expect("latest GET"), b"committed null");
-    }
-
-    #[tokio::test]
-    async fn latest_read_preserves_object_before_equal_time_marker() {
-        let bucket = "dangling-latest-equal-time-marker";
-        let (_dirs, set) = dangling_test_set(bucket).await;
-        let same_time = time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed source mtime");
-        for suspended in [false, true] {
-            let object = if suspended {
-                "torn/equal-time-null"
-            } else {
-                "torn/equal-time-versioned"
-            };
-            let object_id = if suspended { Uuid::nil() } else { Uuid::from_u128(2) };
-            set.put_object(
-                bucket,
-                object,
-                &mut PutObjReader::from_vec(b"committed object".to_vec()),
-                &ObjectOptions {
-                    versioned: !suspended,
-                    version_suspended: suspended,
-                    replication_request: true,
-                    version_id: Some(object_id.to_string()),
-                    mod_time: Some(same_time),
-                    write_completion: crate::object_api::WriteCompletion::TailDrained,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("commit object version");
-            set.delete_object(
-                bucket,
-                object,
-                ObjectOptions {
-                    versioned: true,
-                    version_id: Some(Uuid::from_u128(1).to_string()),
-                    mod_time: Some(same_time),
-                    data_movement: true,
-                    delete_marker: true,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("recreate source delete marker during data movement");
-            let expected = head(&set, bucket, object, None)
-                .await
-                .expect("canonical latest before dangling write");
-            assert_eq!(expected.version_id, (!suspended).then_some(object_id));
-            assert!(!expected.delete_marker, "canonical header order prefers object to equal-time marker");
-            assert_eq!(get(&set, bucket, object, None).await.expect("canonical GET"), b"committed object");
-            let dangling_id = Uuid::from_u128(3);
-            set.put_object(
-                bucket,
-                object,
-                &mut PutObjReader::from_vec(b"failed newer version".to_vec()),
-                &ObjectOptions {
-                    versioned: true,
-                    replication_request: true,
-                    version_id: Some(dangling_id.to_string()),
-                    mod_time: Some(same_time + time::Duration::seconds(1)),
-                    write_completion: crate::object_api::WriteCompletion::TailDrained,
-                    ..Default::default()
-                },
-            )
-            .await
-            .expect("commit newer version before modeling failed rollback");
-            drop_version_from(&set, bucket, object, dangling_id, 3..DRIVES).await;
-            let actual = head(&set, bucket, object, None)
-                .await
-                .expect("latest HEAD past dangling version");
-            let actual_get = get(&set, bucket, object, None).await;
-            assert_eq!(
-                actual.version_id, expected.version_id,
-                "{object}: unreadable newer version must preserve canonical equal-time latest; GET result: {actual_get:?}"
-            );
-            assert!(
-                !actual.delete_marker,
-                "{object}: fallback must preserve the object before its equal-time marker"
-            );
-            assert_eq!(actual_get.expect("latest GET"), b"committed object");
-
-            let checksum_opts = ObjectOptions {
-                include_part_checksums: true,
-                ..Default::default()
-            };
-            let checksum_head = set
-                .get_object_info(bucket, object, &checksum_opts)
-                .await
-                .expect("latest HEAD with part checksums");
-            assert_eq!(checksum_head.version_id, expected.version_id, "{object}: checksum HEAD version");
-            assert!(!checksum_head.delete_marker, "{object}: checksum HEAD must return the object");
-            let mut checksum_reader = set
-                .get_object_reader(bucket, object, None, HeaderMap::new(), &checksum_opts)
-                .await
-                .expect("latest GET with part checksums");
-            let mut checksum_body = Vec::new();
-            checksum_reader
-                .stream
-                .read_to_end(&mut checksum_body)
-                .await
-                .expect("checksum GET should stream the committed bytes");
-            assert_eq!(checksum_body, b"committed object", "{object}: checksum GET bytes");
-        }
-    }
-
-    #[tokio::test]
-    async fn latest_read_past_dangling_versions_requires_every_drive_to_answer() {
-        let bucket = "dangling-latest-offline";
-        let object = "torn/latest";
-        let (_dirs, set) = dangling_test_set(bucket).await;
-        put(&set, bucket, object, b"version one", true).await;
-        let v2 = put(&set, bucket, object, b"version two", true)
-            .await
-            .version_id
-            .expect("versioned PUT should assign a version id");
-        drop_version_from(&set, bucket, object, v2, 3..DRIVES).await;
-        // The offline drive may hold the copy that makes version two readable.
-        let offline = set.disks.write().await[5].take().expect("drive five should start online");
-
-        assert_get_and_head_keep_read_quorum_error(&set, bucket, object, "one drive offline").await;
-
-        set.disks.write().await[5] = Some(offline);
-        let latest = head(&set, bucket, object, None)
-            .await
-            .expect("with every drive answering the dangling version is skipped");
-        assert_ne!(latest.version_id, Some(v2));
+        assert!(is_read_quorum_error(&err), "a split latest version keeps the quorum error: {err:?}");
     }
 
     #[tokio::test]

crates/lock/src/fast_lock/shard.rs

@@ -39,37 +39,7 @@ struct GuardHolderKey {
     owner: Arc<str>,
 }
 
-#[cfg(test)]
-#[derive(Clone, Copy, Debug, Eq, Hash, PartialEq)]
-enum TestPausePoint {
-    FastExisting,
-    BeforeFastWrite,
-    FastWriteExisting,
-    SlowLookup,
-}
-
-#[cfg(test)]
-struct TestPause {
-    entered: std::sync::mpsc::Sender<()>,
-    resume: std::sync::mpsc::Receiver<()>,
-}
-
-#[cfg(test)]
-static TEST_PAUSES: std::sync::LazyLock<parking_lot::Mutex<HashMap<(TestPausePoint, ObjectKey), TestPause>>> =
-    std::sync::LazyLock::new(|| parking_lot::Mutex::new(HashMap::new()));
-
-#[cfg(test)]
-fn test_pause(point: TestPausePoint, key: &ObjectKey) {
-    let pause = TEST_PAUSES.lock().remove(&(point, key.clone()));
-    if let Some(pause) = pause {
-        pause.entered.send(()).expect("pause receiver");
-        pause.resume.recv_timeout(Duration::from_secs(5)).expect("resume signal");
-    }
-}
-
-/// Lock shard to reduce global contention.
-///
-/// When both are needed, lock `objects` before a state's owner metadata.
+/// Lock shard to reduce global contention
 #[derive(Debug)]
 pub struct LockShard {
     /// Object lock states - using parking_lot for better performance
@@ -163,16 +133,13 @@ impl LockShard {
             let objects = self.objects.read();
             if let Some(state) = objects.get(&request.key) {
                 let state = state.clone();
-
-                #[cfg(test)]
-                test_pause(TestPausePoint::FastExisting, &request.key);
+                drop(objects);
 
                 // Try atomic acquisition
                 let success = match request.mode {
                     LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
                     LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
                 };
-                drop(objects);
 
                 if success {
                     return Some(state);
@@ -183,21 +150,14 @@ impl LockShard {
         // If object doesn't exist and we're requesting exclusive lock,
         // try to create and acquire atomically
         if request.mode == LockMode::Exclusive {
-            #[cfg(test)]
-            test_pause(TestPausePoint::BeforeFastWrite, &request.key);
             let mut objects = self.objects.write();
 
             // Double-check after acquiring write lock
             if let Some(state) = objects.get(&request.key) {
                 let state = state.clone();
-                let objects = parking_lot::RwLockWriteGuard::downgrade(objects);
-
-                #[cfg(test)]
-                test_pause(TestPausePoint::FastWriteExisting, &request.key);
-
-                let acquired = state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout);
                 drop(objects);
-                if acquired {
+
+                if state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout) {
                     return Some(state);
                 }
             } else {
@@ -233,10 +193,10 @@ impl LockShard {
         const NOTIFY_WAIT_CAP: Duration = Duration::from_millis(50);
 
         loop {
-            // Keep the state in the map until its acquisition attempt completes.
-            let (state, success) = {
+            // Get or create object state
+            let state = {
                 let mut objects = self.objects.write();
-                let state = match objects.get(&request.key) {
+                match objects.get(&request.key) {
                     Some(state) => state.clone(),
                     None => {
                         let state_box = self.object_pool.acquire();
@@ -244,17 +204,13 @@ impl LockShard {
                         objects.insert(request.key.clone(), state.clone());
                         state
                     }
-                };
-                let objects = parking_lot::RwLockWriteGuard::downgrade(objects);
+                }
+            };
 
-                #[cfg(test)]
-                test_pause(TestPausePoint::SlowLookup, &request.key);
-                let success = match request.mode {
-                    LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
-                    LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
-                };
-                drop(objects);
-                (state, success)
+            // Try acquisition again
+            let success = match request.mode {
+                LockMode::Shared => state.try_acquire_shared_fast(&request.owner, request.lock_timeout),
+                LockMode::Exclusive => state.try_acquire_exclusive_fast(&request.owner, request.lock_timeout),
             };
 
             if success {
@@ -949,184 +905,6 @@ impl LockShard {
 mod tests {
     use super::*;
     use crate::fast_lock::types::{LockPriority, ObjectKey};
-    use crate::fast_lock::{FastLockGuard, FastObjectLockManager};
-
-    fn install_pause(point: TestPausePoint, key: &ObjectKey) -> (std::sync::mpsc::Receiver<()>, std::sync::mpsc::Sender<()>) {
-        let (entered_tx, entered_rx) = std::sync::mpsc::channel();
-        let (resume_tx, resume_rx) = std::sync::mpsc::channel();
-        let previous = TEST_PAUSES.lock().insert(
-            (point, key.clone()),
-            TestPause {
-                entered: entered_tx,
-                resume: resume_rx,
-            },
-        );
-        assert!(previous.is_none());
-        (entered_rx, resume_tx)
-    }
-
-    fn test_manager() -> Arc<FastObjectLockManager> {
-        // Construct outside a runtime so no background cleanup races the test.
-        Arc::new(
-            std::thread::spawn(|| {
-                FastObjectLockManager::with_config(crate::fast_lock::types::LockConfig {
-                    shard_count: 4,
-                    ..Default::default()
-                })
-            })
-            .join()
-            .expect("manager construction"),
-        )
-    }
-
-    fn start_acquisition(
-        manager: &Arc<FastObjectLockManager>,
-        request: ObjectLockRequest,
-    ) -> tokio::task::JoinHandle<Result<FastLockGuard, LockResult>> {
-        let manager = manager.clone();
-        tokio::spawn(async move { manager.acquire_lock(request).await })
-    }
-
-    async fn assert_cleanup_cannot_split_state(
-        manager: Arc<FastObjectLockManager>,
-        key: ObjectKey,
-        entered: std::sync::mpsc::Receiver<()>,
-        resume: std::sync::mpsc::Sender<()>,
-        pending: tokio::task::JoinHandle<Result<FastLockGuard, LockResult>>,
-        seed: Option<FastLockGuard>,
-    ) {
-        entered.recv_timeout(Duration::from_secs(5)).expect("state lookup");
-        drop(seed);
-
-        // Idle timestamps have one-second precision, while cleanup compares milliseconds.
-        let initial_millis = SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis();
-        while SystemTime::now().duration_since(SystemTime::UNIX_EPOCH).unwrap().as_millis() == initial_millis {
-            std::thread::yield_now();
-        }
-
-        let can_write_map = manager.get_shard(&key).objects.try_write().is_some();
-        let (started_tx, started_rx) = std::sync::mpsc::channel();
-        let mut cleanup = {
-            let manager = manager.clone();
-            let key = key.clone();
-            tokio::task::spawn_blocking(move || {
-                started_tx.send(()).expect("cleanup start receiver");
-                manager.get_shard(&key).cleanup_expired_millis(0)
-            })
-        };
-        started_rx.recv_timeout(Duration::from_secs(5)).expect("cleanup started");
-        if can_write_map {
-            assert_eq!(
-                tokio::time::timeout(Duration::from_secs(5), &mut cleanup)
-                    .await
-                    .expect("cleanup completed")
-                    .expect("cleanup task"),
-                1
-            );
-            let successor = manager
-                .acquire_write_lock(key.clone(), "second")
-                .await
-                .expect("successor should acquire after cleanup");
-            resume.send(()).expect("resume acquisition");
-            let result = tokio::time::timeout(Duration::from_secs(5), pending)
-                .await
-                .expect("acquisition completed")
-                .expect("acquisition task");
-            assert!(result.is_err(), "a removed state must not grant a conflicting lock");
-            drop(successor);
-        } else {
-            resume.send(()).expect("resume acquisition");
-            let first = tokio::time::timeout(Duration::from_secs(5), pending)
-                .await
-                .expect("acquisition completed")
-                .expect("acquisition task")
-                .expect("first acquisition");
-            assert_eq!(
-                tokio::time::timeout(Duration::from_secs(5), cleanup)
-                    .await
-                    .expect("cleanup completed")
-                    .expect("cleanup task"),
-                0
-            );
-            let second = ObjectLockRequest::new_write(key.clone(), "second").with_acquire_timeout(Duration::from_millis(100));
-            assert!(
-                manager.acquire_lock(second).await.is_err(),
-                "conflicting locks must remain mutually exclusive"
-            );
-            drop(first);
-            let successor = manager
-                .acquire_write_lock(key, "second")
-                .await
-                .expect("successor after release");
-            drop(successor);
-        }
-        manager.shutdown().await;
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn test_cleanup_during_fast_existing_state_acquisition() {
-        let manager = test_manager();
-        let key = ObjectKey::new("bucket", "fast-existing-cleanup");
-        drop(manager.acquire_write_lock(key.clone(), "seed").await.expect("seed"));
-        let (entered, resume) = install_pause(TestPausePoint::FastExisting, &key);
-        let pending = start_acquisition(&manager, ObjectLockRequest::new_write(key.clone(), "first"));
-        assert_cleanup_cannot_split_state(manager, key, entered, resume, pending, None).await;
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn test_cleanup_during_fast_write_recheck_acquisition() {
-        let manager = test_manager();
-        let key = ObjectKey::new("bucket", "fast-recheck-cleanup");
-        let (first_entered, first_resume) = install_pause(TestPausePoint::BeforeFastWrite, &key);
-        let pending = start_acquisition(&manager, ObjectLockRequest::new_write(key.clone(), "first"));
-        first_entered
-            .recv_timeout(Duration::from_secs(5))
-            .expect("before write recheck");
-        drop(manager.acquire_write_lock(key.clone(), "seed").await.expect("seed"));
-        let (entered, resume) = install_pause(TestPausePoint::FastWriteExisting, &key);
-        first_resume.send(()).expect("resume before write recheck");
-        assert_cleanup_cannot_split_state(manager, key, entered, resume, pending, None).await;
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn test_cleanup_during_slow_existing_state_acquisition() {
-        let manager = test_manager();
-        let key = ObjectKey::new("bucket", "slow-existing-cleanup");
-        let seed = manager.acquire_write_lock(key.clone(), "seed").await.expect("seed");
-        let (entered, resume) = install_pause(TestPausePoint::SlowLookup, &key);
-        let pending = start_acquisition(&manager, ObjectLockRequest::new_write(key.clone(), "first"));
-        assert_cleanup_cannot_split_state(manager, key, entered, resume, pending, Some(seed)).await;
-    }
-
-    #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
-    async fn test_cleanup_during_slow_new_state_acquisition() {
-        let manager = test_manager();
-        let key = ObjectKey::new("bucket", "slow-new-cleanup");
-        let (entered, resume) = install_pause(TestPausePoint::SlowLookup, &key);
-        let pending = start_acquisition(&manager, ObjectLockRequest::new_read(key.clone(), "first"));
-        assert_cleanup_cannot_split_state(manager, key, entered, resume, pending, None).await;
-    }
-
-    #[tokio::test]
-    async fn test_cleanup_keeps_new_fast_state_while_held() {
-        let manager = test_manager();
-        let key = ObjectKey::new("bucket", "fast-new-cleanup");
-        let first = manager
-            .acquire_write_lock(key.clone(), "first")
-            .await
-            .expect("first acquisition");
-        assert_eq!(manager.get_shard(&key).cleanup_expired_millis(0), 0);
-        let second = ObjectLockRequest::new_write(key.clone(), "second").with_acquire_timeout(Duration::from_millis(100));
-        assert!(manager.acquire_lock(second).await.is_err());
-        drop(first);
-        drop(
-            manager
-                .acquire_write_lock(key, "second")
-                .await
-                .expect("successor after release"),
-        );
-        manager.shutdown().await;
-    }
 
     #[tokio::test]
     async fn test_shard_fast_path() {

docs/README.md

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

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

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

docs/architecture/erasure-coding.md

@@ -313,7 +313,7 @@ Small objects store their payload inline after the container CRC ([filemeta_inli
 ## 8. Read path and read quorum
 
 - **INVARIANT — read quorum = `data_blocks`.** `object_quorum_from_meta` returns `(read_quorum = data_blocks, write_quorum)` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)); `parity_blocks = common_parity(...)` is the parity value held by the most disks that still reaches its own read quorum. When `default_parity_count == 0`, read = write = all shards.
-- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. S3 reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set. Reads in the system bucket (`is_meta_bucketname`) keep the commit-proof rule in `select_committed_fileinfo`: when enough copies answer to prove write quorum, the latest version must hold it, and a degraded read never serves a newer version below write quorum. Pool metadata's PREPARE/COMMIT recovery and identity epochs rely on an ordinary read implying a committed record, so a split stays unreadable until native recovery resolves it. When every drive answered and an S3 latest read still cannot select a version, `read_latest_past_dangling_versions` walks the version stack newest first through exact-version reads, as MinIO's `pickLatestQuorumFilesInfo` resolves stacks: it serves the readable version with the newest selected mod_time, skips versions whose exact read proves them dangling (the NoSuchVersion an explicit GET reports) and queues heal for them, and keeps the read-quorum error on any other outcome. A failed versioned PUT that left the new version on fewer drives than its data blocks therefore no longer hides the committed version below it; ListObjectVersions can still show the dangling version as latest until heal removes it.
+- Authoritative FileInfo selection — `find_file_info_in_quorum` ([set_disk/metadata.rs](../../crates/ecstore/src/set_disk/metadata.rs)) groups valid metas by a content-identity SHA-256 (`file_info_quorum_hash`) that hashes size/flags/mod_time/transition/version_id/data_dir/parts and, for real objects, data/parity/distribution — **excluding replication-status keys** so replication noise never splits quorum. A meta counts only if its mod_time equals the common mod_time (or etag matches when mod_time is absent). The winning hash must reach quorum, else `ErasureReadQuorum`. Reads select at read quorum through `select_valid_fileinfo`, the same rule heal uses and MinIO's `getObjectFileInfo`: the most common mod_time wins and a tie goes to the newer one, so a write that failed after reaching exactly `data_blocks` copies is served as listing shows it, instead of `ErasureReadQuorum` until heal runs. Two identities tied at the selected mod_time (a data = parity set split in half, such as a metadata update that failed after reaching half the drives) resolve to the one on the lowest-index disk, in reads and heal alike, so a retried metadata write can repair the split. A selected version with local shards also needs as many copies as its own `data_blocks`, since delete markers and empty versions can lower the common parity's read quorum. A metadata update whose target disks cannot reach write quorum is refused before any write. A version replaced by a committed write keeps fewer than `data_blocks` copies and can never be selected again; the newer side of a failed write can be served and then lose to the older side if one of its disks fails before heal converges the set.
 - **INVARIANT — decode needs ≥ `data_blocks` shards.** The stripe reader requires `available_shards ≥ data_shards`; below that the read fails closed with a read-quorum error (never silent truncation) ([set_disk/read.rs](../../crates/ecstore/src/set_disk/read.rs), [set_disk/shard_source.rs](../../crates/ecstore/src/set_disk/shard_source.rs)). Codec geometry is validated at construction: `Erasure::try_new` / `try_new_with_options` ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)) return `ErasureConstructionError` for `data_shards == 0`, `block_size == 0`, shard-count overflow, or an unsupported shard configuration, so a read never reaches a `block_size` / `data_shards` division with invalid geometry (§13).
 - If `available ≥ data_blocks` but some shards are missing, the read is served **and** a background read-repair heal is enqueued.
 - **INVARIANT — cross-stripe read verification.** When a data shard is missing and `available > data_blocks`, reconstruction regenerates parity and compares it to the surviving parity; a mismatch is `InvalidData "inconsistent read source shards"` (backlog#832), catching corruption that passed per-shard bitrot but disagrees across the stripe ([erasure.rs](../../crates/ecstore/src/erasure/coding/erasure.rs)).

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

@@ -0,0 +1,60 @@
+# Sampled local batch CPU diagnostics
+
+**Use this when:** separating sampled local EC batch allocation/initialization
+and filesystem wall time from worker CPU time in a default-off diagnostic build.
+
+This diagnostic build separates allocation/initialization, metadata, open and
+positioned-read elapsed time from Linux thread CPU time. It preserves the
+single blocking dispatch, buffer initialization, offsets, integrity checks and
+error handling. It does not implement a rustix optimization or qualify
+observation overhead.
+
+Build separately from any frozen performance binary:
+
+```sh
+RUSTUP_TOOLCHAIN=stable cargo build --release --locked -p rustfs --bin rustfs \
+  --features diagnostic-local-batch
+```
+
+The feature is absent from default builds. Even this diagnostic build records
+nothing until both the sample rate and the target filter are enabled:
+
+```sh
+export RUSTFS_DIAGNOSTIC_LOCAL_BATCH_SAMPLE_EVERY=64
+export RUST_LOG=warn,rustfs_ecstore::batch_read_diagnostics=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.
+
+Events contain a process-local batch ID, the bounded server-owned canonical
+request ID when propagated, worker PID/TID, shard index, sizes and timing.
+They explicitly have no parent span, so URI-bearing HTTP span fields are not
+copied. There are no variable-ID metric labels. Missing request context is
+reported as `request_context=missing`, never treated as a successful join.
+
+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.
+
+`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 paths,
+credentials, request bodies or arbitrary client-supplied identifiers to add
+correlation. Request-level client association must use the server's canonical
+response ID. Diagnostic results locate hypotheses; they do not regrade failed
+A/A or ABBA runs.

rustfs/Cargo.toml

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

rustfs/src/server/layer.rs

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

rustfs/src/storage/storage_api.rs

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

rustfs/src/storage_api.rs

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

scripts/check_logging_guardrails.sh

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

scripts/error-other-format-baseline.txt

@@ -45,9 +45,9 @@
 3|crates/ecstore/src/runtime/sources.rs
 4|crates/ecstore/src/services/batch_processor.rs
 12|crates/ecstore/src/services/notification_sys.rs
-15|crates/ecstore/src/services/rebalance/control.rs
+16|crates/ecstore/src/services/rebalance/control.rs
 1|crates/ecstore/src/services/rebalance/entry.rs
-6|crates/ecstore/src/services/rebalance/meta.rs
+8|crates/ecstore/src/services/rebalance/meta.rs
 8|crates/ecstore/src/services/rebalance/runtime.rs
 17|crates/ecstore/src/services/rebalance/worker.rs
 33|crates/ecstore/src/services/tier/tier.rs