fix(ecstore): harden HotPath profiling boundaries (#5555)

* test(hotpath): gate mimalloc heap test by platform

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

* fix(ecstore): attribute HotPath CPU measurements to impls

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

* feat(ecstore): trace raw shard I/O with HotPath

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

* fix(obs): redact profiler and OPA endpoint diagnostics

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

* fix(ecstore): settle encoded queue accounting

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

* fix(ecstore): abort encoder producer on cancellation

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

---------

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

Cargo.lock

@@ -9813,6 +9813,7 @@ dependencies = [
  "hotpath",
  "jiff",
  "libc",
+ "log",
  "metrics",
  "num_cpus",
  "nvml-wrapper",
@@ -9848,6 +9849,7 @@ dependencies = [
  "tracing-error",
  "tracing-opentelemetry",
  "tracing-subscriber",
+ "url",
  "zstd",
 ]
 
@@ -9879,6 +9881,7 @@ dependencies = [
  "time",
  "tokio",
  "tracing",
+ "tracing-subscriber",
 ]
 
 [[package]]

crates/ecstore/src/disk/local.rs

@@ -5078,7 +5078,7 @@ impl LocalDisk {
         Ok((buf, mtime))
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "LocalDisk")]
     async fn read_metadata_with_dmtime(&self, file_path: impl AsRef<Path>) -> Result<(Vec<u8>, Option<OffsetDateTime>)> {
         check_path_length(file_path.as_ref().to_string_lossy().as_ref())?;
 
@@ -5121,7 +5121,7 @@ impl LocalDisk {
         Ok((data, modtime))
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "LocalDisk")]
     async fn read_all_data(&self, volume: &str, volume_dir: impl AsRef<Path>, file_path: impl AsRef<Path>) -> Result<Vec<u8>> {
         // TODO: timeout support
         let (data, _) = self.read_all_data_with_dmtime(volume, volume_dir, file_path).await?;

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

@@ -103,7 +103,7 @@ where
     /// or `out` is larger than one shard. On error `out`'s contents are
     /// unspecified but never contain bytes that failed the hash check — the copy
     /// happens only after verification.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "BitrotReader")]
     pub async fn read(&mut self, out: &mut [u8]) -> std::io::Result<usize> {
         let want = out.len();
         self.begin_read(want)?;
@@ -303,7 +303,7 @@ where
 
     /// Write a (hash+data) block. Returns the number of data bytes written.
     /// Returns an error if called after a short write or if data exceeds shard_size.
-    #[hotpath::measure(label = "BitrotWriter::write")]
+    #[hotpath::measure(label = "BitrotWriter::write", impl_type = "BitrotWriter")]
     pub async fn write(&mut self, buf: &[u8]) -> std::io::Result<usize> {
         if buf.is_empty() {
             return Ok(0);

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

@@ -691,7 +691,7 @@ impl<R> ParallelReader<R>
 where
     R: crate::erasure::coding::ShardSource,
 {
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "ParallelReader")]
     pub async fn read(&mut self) -> (Vec<Option<Vec<u8>>>, Vec<Option<Error>>) {
         // On the reconstruction-verifying GET path, read every live shard reader
         // in lockstep so all readers advance one block per stripe and stay
@@ -1505,7 +1505,7 @@ where
 }
 
 impl Erasure {
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub async fn decode<W, R>(
         &self,
         writer: &mut W,

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

@@ -28,8 +28,11 @@ use std::vec;
 use tokio::io::AsyncRead;
 use tokio::runtime::RuntimeFlavor;
 use tokio::sync::mpsc;
+use tokio::task::{JoinError, JoinHandle};
 use tracing::error;
 
+/// Queue-capacity input for encoded blocks awaiting shard writers; it is not a
+/// per-PUT or process-RSS memory limit.
 const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES";
 const ENV_RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS: &str = "RUSTFS_ERASURE_ENCODE_BATCH_BLOCKS";
 const ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST: &str = "RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST";
@@ -87,6 +90,33 @@ fn use_bytesmut_ingest() -> bool {
         rustfs_utils::get_env_bool(ENV_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST, DEFAULT_RUSTFS_ERASURE_ENCODE_BYTESMUT_INGEST)
     })
 }
+
+/// Keeps the encoder producer scoped to its parent future. Tokio detaches a
+/// task when its `JoinHandle` is dropped, so the producer must be aborted when
+/// an upload is cancelled before the encode pipeline finishes.
+struct AbortOnDropTask<T>(JoinHandle<T>);
+
+impl<T> AbortOnDropTask<T> {
+    fn new(task: JoinHandle<T>) -> Self {
+        Self(task)
+    }
+
+    async fn abort_and_wait(&mut self) {
+        self.0.abort();
+        let _ = (&mut self.0).await;
+    }
+
+    async fn join(&mut self) -> Result<T, JoinError> {
+        (&mut self.0).await
+    }
+}
+
+impl<T> Drop for AbortOnDropTask<T> {
+    fn drop(&mut self) {
+        self.0.abort();
+    }
+}
+
 /// Read up to `limit` bytes into `buf`'s uninitialized spare capacity, appending after its
 /// current length, and distinguish a clean EOF from a short read.
 ///
@@ -133,22 +163,65 @@ fn queued_block_bytes(block: &[Bytes]) -> usize {
     block.iter().map(Bytes::len).sum()
 }
 
-async fn drain_queued_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Bytes>>) {
-    while let Some(block) = rx.recv().await {
-        rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_block_bytes(&block));
+/// Owns an encoded queue entry's gauge contribution until its consumer takes it.
+struct QueuedInflightBytes {
+    bytes: usize,
+}
+
+impl QueuedInflightBytes {
+    fn new(bytes: usize) -> Self {
+        rustfs_io_metrics::add_ec_encode_inflight_bytes(bytes);
+        Self { bytes }
     }
+
+    fn settle(&mut self) {
+        let bytes = std::mem::take(&mut self.bytes);
+        if bytes != 0 {
+            rustfs_io_metrics::remove_ec_encode_inflight_bytes(bytes);
+        }
+    }
+}
+
+impl Drop for QueuedInflightBytes {
+    fn drop(&mut self) {
+        self.settle();
+    }
+}
+
+/// Couples an encoded block with its queue gauge contribution. Keeping the
+/// guard in the queue entry also covers Tokio sends that complete through a
+/// permit after the receiver has closed.
+struct InflightEntry<T> {
+    entry: T,
+    accounting: QueuedInflightBytes,
+}
+
+impl<T> InflightEntry<T> {
+    fn new(entry: T, bytes: usize) -> Self {
+        Self {
+            entry,
+            accounting: QueuedInflightBytes::new(bytes),
+        }
+    }
+
+    fn into_inner(mut self) -> T {
+        self.accounting.settle();
+        self.entry
+    }
+}
+
+async fn send_queued<T>(
+    sender: &mpsc::Sender<InflightEntry<T>>,
+    entry: T,
+    bytes: usize,
+) -> Result<(), mpsc::error::SendError<InflightEntry<T>>> {
+    sender.send(InflightEntry::new(entry, bytes)).await
 }
 
 fn queued_batch_bytes(batch: &[Vec<Bytes>]) -> usize {
     batch.iter().map(|block| queued_block_bytes(block)).sum()
 }
 
-async fn drain_queued_batched_inflight_bytes(rx: &mut mpsc::Receiver<Vec<Vec<Bytes>>>) {
-    while let Some(batch) = rx.recv().await {
-        rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch));
-    }
-}
-
 fn dominant_error_summary_label(summary: &WriteQuorumFailureSummary) -> &'static str {
     summary.dominant_error_label
 }
@@ -504,7 +577,7 @@ impl Erasure {
         Ok((reader, total))
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub async fn encode<R>(
         self: Arc<Self>,
         reader: R,
@@ -540,13 +613,14 @@ impl Erasure {
             ));
         }
 
-        // Bound queued encoded blocks by memory budget to avoid per-request spikes.
+        // Bound queued encoded blocks by a queue budget; this does not bound
+        // reader, encoder, writer, allocator, or process-RSS memory.
         let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count());
         let max_inflight_bytes = erasure_encode_max_inflight_bytes();
         let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
-        let (tx, mut rx) = mpsc::channel::<Vec<Bytes>>(inflight_blocks);
+        let (tx, mut rx) = mpsc::channel::<InflightEntry<Vec<Bytes>>>(inflight_blocks);
 
-        let task = tokio::spawn(async move {
+        let mut task = AbortOnDropTask::new(tokio::spawn(async move {
             let block_size = self.block_size;
             let mut total = 0;
             if use_bytesmut_ingest {
@@ -567,10 +641,8 @@ impl Erasure {
                             let res = self.clone().encode_block_bytes_mut(encode_buf, n).await?;
                             buf = BytesMut::with_capacity(ingest_capacity);
                             let queued_bytes = queued_block_bytes(&res);
-                            rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
-                            if let Err(err) = tx.send(res).await {
-                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
+                            if let Err(err) = send_queued(&tx, res, queued_bytes).await {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                             }
                             record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start);
@@ -598,10 +670,8 @@ impl Erasure {
                             let (res, returned_buf) = self.clone().encode_block(encode_buf, n).await?;
                             buf = returned_buf;
                             let queued_bytes = queued_block_bytes(&res);
-                            rustfs_io_metrics::add_ec_encode_inflight_bytes(queued_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
-                            if let Err(err) = tx.send(res).await {
-                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
+                            if let Err(err) = send_queued(&tx, res, queued_bytes).await {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                             }
                             record_internal_stage_if_enabled("erasure_encode_send_wait", send_wait_stage_start);
@@ -626,7 +696,7 @@ impl Erasure {
             }
 
             Ok((reader, total))
-        });
+        }));
 
         let mut writers = MultiWriter::new(writers, quorum);
 
@@ -638,11 +708,10 @@ impl Erasure {
                 break;
             };
             record_internal_stage_if_enabled("erasure_encode_recv_wait", recv_wait_stage_start);
+            let block = block.into_inner();
             if block.is_empty() {
                 break;
             }
-            let queued_bytes = queued_block_bytes(&block);
-            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_bytes);
             let write_stage_start = stage_timer_if_enabled();
             if let Err(err) = writers.write(block).await {
                 write_err = Some(err);
@@ -652,9 +721,8 @@ impl Erasure {
         }
 
         if let Some(err) = write_err {
-            task.abort();
-            let _ = task.await;
-            drain_queued_inflight_bytes(&mut rx).await;
+            task.abort_and_wait().await;
+            drop(rx);
             let shutdown_stage_start = stage_timer_if_enabled();
             if let Err(shutdown_err) = writers.shutdown().await {
                 error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
@@ -663,14 +731,14 @@ impl Erasure {
             return Err(err);
         }
 
-        let (reader, total) = task.await??;
+        let (reader, total) = task.join().await??;
         let shutdown_stage_start = stage_timer_if_enabled();
         writers.shutdown().await?;
         record_internal_stage_if_enabled("erasure_encode_shutdown", shutdown_stage_start);
         Ok((reader, total))
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub async fn encode_batched<R>(
         self: Arc<Self>,
         mut reader: R,
@@ -692,9 +760,9 @@ impl Erasure {
         let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes);
         let batch_blocks = encode_batch_block_count().min(inflight_blocks);
         let channel_capacity = inflight_blocks.div_ceil(batch_blocks).max(1);
-        let (tx, mut rx) = mpsc::channel::<Vec<Vec<Bytes>>>(channel_capacity);
+        let (tx, mut rx) = mpsc::channel::<InflightEntry<Vec<Vec<Bytes>>>>(channel_capacity);
 
-        let task = tokio::spawn(async move {
+        let mut task = AbortOnDropTask::new(tokio::spawn(async move {
             let block_size = self.block_size;
             let mut total = 0;
             let mut buf = vec![0u8; block_size];
@@ -713,10 +781,8 @@ impl Erasure {
                         pending_batch.push(res);
 
                         if pending_batch.len() >= batch_blocks {
-                            rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
                             let send_wait_stage_start = stage_timer_if_enabled();
-                            if let Err(err) = tx.send(pending_batch).await {
-                                rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes);
+                            if let Err(err) = send_queued(&tx, pending_batch, pending_batch_bytes).await {
                                 return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                             }
                             record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
@@ -742,17 +808,15 @@ impl Erasure {
             }
 
             if !pending_batch.is_empty() {
-                rustfs_io_metrics::add_ec_encode_inflight_bytes(pending_batch_bytes);
                 let send_wait_stage_start = stage_timer_if_enabled();
-                if let Err(err) = tx.send(pending_batch).await {
-                    rustfs_io_metrics::remove_ec_encode_inflight_bytes(pending_batch_bytes);
+                if let Err(err) = send_queued(&tx, pending_batch, pending_batch_bytes).await {
                     return Err(std::io::Error::other(format!("Failed to send encoded data : {err}")));
                 }
                 record_internal_stage_if_enabled("erasure_encode_batched_send_wait", send_wait_stage_start);
             }
 
             Ok((reader, total))
-        });
+        }));
 
         let mut writers = MultiWriter::new(writers, quorum);
         let mut write_err = None;
@@ -763,7 +827,7 @@ impl Erasure {
                 break;
             };
             record_internal_stage_if_enabled("erasure_encode_batched_recv_wait", recv_wait_stage_start);
-            rustfs_io_metrics::remove_ec_encode_inflight_bytes(queued_batch_bytes(&batch));
+            let batch = batch.into_inner();
             let write_stage_start = stage_timer_if_enabled();
             for block in batch {
                 if let Err(err) = writers.write(block).await {
@@ -778,9 +842,8 @@ impl Erasure {
         }
 
         if let Some(err) = write_err {
-            task.abort();
-            let _ = task.await;
-            drain_queued_batched_inflight_bytes(&mut rx).await;
+            task.abort_and_wait().await;
+            drop(rx);
             let shutdown_stage_start = stage_timer_if_enabled();
             if let Err(shutdown_err) = writers.shutdown().await {
                 error!("failed to shutdown erasure writers after write error: {:?}", shutdown_err);
@@ -789,7 +852,7 @@ impl Erasure {
             return Err(err);
         }
 
-        let (reader, total) = task.await??;
+        let (reader, total) = task.join().await??;
         let shutdown_stage_start = stage_timer_if_enabled();
         writers.shutdown().await?;
         record_internal_stage_if_enabled("erasure_encode_batched_shutdown", shutdown_stage_start);
@@ -798,7 +861,7 @@ impl Erasure {
 
     /// Fast path for small inline objects: skip tokio::spawn + mpsc channel.
     /// Reads all data, encodes directly, writes shards sequentially.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub async fn encode_inline_small<R>(
         self: Arc<Self>,
         reader: R,
@@ -813,7 +876,7 @@ impl Erasure {
 
     /// Fast path for single-block non-inline objects: avoids the producer/consumer
     /// pipeline in `encode()` while keeping the same writer/quorum/shutdown semantics.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub async fn encode_single_block_non_inline<R>(
         self: Arc<Self>,
         reader: R,
@@ -839,7 +902,104 @@ mod tests {
     use std::sync::{Arc, Mutex};
     use std::task::{Context, Poll};
     use std::time::Duration;
-    use tokio::io::{AsyncWrite, AsyncWriteExt};
+    use tokio::io::{AsyncWrite, AsyncWriteExt, ReadBuf};
+    use tokio::sync::oneshot;
+
+    struct PendingReader {
+        entered: Option<oneshot::Sender<()>>,
+        dropped: Option<oneshot::Sender<()>>,
+    }
+
+    impl PendingReader {
+        fn new() -> (Self, oneshot::Receiver<()>, oneshot::Receiver<()>) {
+            let (entered_tx, entered_rx) = oneshot::channel();
+            let (dropped_tx, dropped_rx) = oneshot::channel();
+            (
+                Self {
+                    entered: Some(entered_tx),
+                    dropped: Some(dropped_tx),
+                },
+                entered_rx,
+                dropped_rx,
+            )
+        }
+    }
+
+    impl AsyncRead for PendingReader {
+        fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if let Some(entered) = self.entered.take() {
+                let _ = entered.send(());
+            }
+            Poll::Pending
+        }
+    }
+
+    impl Drop for PendingReader {
+        fn drop(&mut self) {
+            if let Some(dropped) = self.dropped.take() {
+                let _ = dropped.send(());
+            }
+        }
+    }
+
+    struct BlocksThenPendingReader {
+        blocks_remaining: usize,
+        block: Vec<u8>,
+        blocked: Option<oneshot::Sender<()>>,
+        dropped: Option<oneshot::Sender<()>>,
+        final_block: Option<oneshot::Sender<()>>,
+    }
+
+    impl BlocksThenPendingReader {
+        fn new(
+            blocks_remaining: usize,
+            block_size: usize,
+        ) -> (Self, oneshot::Receiver<()>, oneshot::Receiver<()>, oneshot::Receiver<()>) {
+            let (blocked_tx, blocked_rx) = oneshot::channel();
+            let (dropped_tx, dropped_rx) = oneshot::channel();
+            let (final_block_tx, final_block_rx) = oneshot::channel();
+            (
+                Self {
+                    blocks_remaining,
+                    block: vec![0x5a; block_size],
+                    blocked: Some(blocked_tx),
+                    dropped: Some(dropped_tx),
+                    final_block: Some(final_block_tx),
+                },
+                blocked_rx,
+                dropped_rx,
+                final_block_rx,
+            )
+        }
+    }
+
+    impl AsyncRead for BlocksThenPendingReader {
+        fn poll_read(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll<std::io::Result<()>> {
+            if self.blocks_remaining == 0 {
+                if let Some(blocked) = self.blocked.take() {
+                    let _ = blocked.send(());
+                }
+                return Poll::Pending;
+            }
+
+            if self.blocks_remaining == 1
+                && let Some(final_block) = self.final_block.take()
+            {
+                let _ = final_block.send(());
+            }
+            self.blocks_remaining -= 1;
+            buf.put_slice(&self.block);
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    impl Drop for BlocksThenPendingReader {
+        fn drop(&mut self) {
+            if let Some(dropped) = self.dropped.take() {
+                let _ = dropped.send(());
+            }
+        }
+    }
 
     fn erasure_with_zero_block_size() -> Erasure {
         let mut erasure = Erasure::default();
@@ -897,6 +1057,54 @@ mod tests {
         }
     }
 
+    struct FailAfterReaderBlocksWriter {
+        reader_blocked: oneshot::Receiver<()>,
+        writes: Arc<std::sync::atomic::AtomicUsize>,
+    }
+
+    impl AsyncWrite for FailAfterReaderBlocksWriter {
+        fn poll_write(mut self: Pin<&mut Self>, cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            match Pin::new(&mut self.reader_blocked).poll(cx) {
+                Poll::Pending => Poll::Pending,
+                Poll::Ready(_) => {
+                    self.writes.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+                    Poll::Ready(Err(std::io::Error::other("injected write failure after producer blocks")))
+                }
+            }
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    struct StallOnWriteWithSignal {
+        entered: Option<oneshot::Sender<()>>,
+        writes: Arc<std::sync::atomic::AtomicUsize>,
+    }
+
+    impl AsyncWrite for StallOnWriteWithSignal {
+        fn poll_write(mut self: Pin<&mut Self>, _cx: &mut Context<'_>, _buf: &[u8]) -> Poll<std::io::Result<usize>> {
+            self.writes.fetch_add(1, std::sync::atomic::Ordering::SeqCst);
+            if let Some(entered) = self.entered.take() {
+                let _ = entered.send(());
+            }
+            Poll::Pending
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<std::io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
     #[derive(Clone, Default)]
     struct ShortWriteWriter;
 
@@ -1046,6 +1254,202 @@ mod tests {
         BitrotWriterWrapper::new(CustomWriter::new_tokio_writer(writer), shard_size, HashAlgorithm::None)
     }
 
+    #[derive(Clone, Copy)]
+    enum EncodePipeline {
+        Vec,
+        BytesMut,
+        Batched,
+    }
+
+    async fn aborting_encode_drops_blocked_producer(pipeline: EncodePipeline) {
+        const BLOCK_SIZE: usize = 16;
+
+        let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let committed = Arc::new(Mutex::new(Vec::new()));
+        let mut writers = vec![Some(bitrot_writer(DeferredCommitWriter::new(committed.clone()), BLOCK_SIZE))];
+        let (reader, entered, dropped) = PendingReader::new();
+        let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
+
+        let encode = match pipeline {
+            EncodePipeline::Vec => {
+                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, false).await })
+            }
+            EncodePipeline::BytesMut => {
+                tokio::spawn(async move { erasure.encode_with_ingest_mode(reader, &mut writers, 1, true).await })
+            }
+            EncodePipeline::Batched => tokio::spawn(async move { erasure.encode_batched(reader, &mut writers, 1).await }),
+        };
+
+        tokio::time::timeout(Duration::from_secs(1), entered)
+            .await
+            .expect("producer should enter the blocked reader before cancellation")
+            .expect("blocked reader should signal entry");
+        encode.abort();
+        assert!(matches!(encode.await, Err(err) if err.is_cancelled()), "encode task should be cancelled");
+        tokio::time::timeout(Duration::from_secs(1), dropped)
+            .await
+            .expect("cancelling encode should drop the producer reader")
+            .expect("blocked reader should signal producer drop");
+        assert!(
+            committed.lock().expect("committed buffer should be lockable").is_empty(),
+            "cancelling before the first encoded block must not make data visible"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            gauge_baseline,
+            "cancelling the encode pipeline must preserve the inflight queue gauge"
+        );
+    }
+
+    async fn writer_error_aborts_blocked_producer(pipeline: EncodePipeline) {
+        const BLOCK_SIZE: usize = 16;
+
+        let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let blocks_before_pending = match pipeline {
+            EncodePipeline::Batched => encode_batch_block_count(),
+            EncodePipeline::Vec | EncodePipeline::BytesMut => 1,
+        };
+        let (reader, reader_blocked, reader_dropped, _final_block) =
+            BlocksThenPendingReader::new(blocks_before_pending, BLOCK_SIZE);
+        let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let mut writers = vec![Some(bitrot_writer(
+            FailAfterReaderBlocksWriter {
+                reader_blocked,
+                writes: writes.clone(),
+            },
+            BLOCK_SIZE,
+        ))];
+        let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
+
+        let result = match pipeline {
+            EncodePipeline::Vec => erasure.encode_with_ingest_mode(reader, &mut writers, 1, false).await,
+            EncodePipeline::BytesMut => erasure.encode_with_ingest_mode(reader, &mut writers, 1, true).await,
+            EncodePipeline::Batched => erasure.encode_batched(reader, &mut writers, 1).await,
+        };
+
+        let err = match result {
+            Ok(_) => panic!("writer quorum failure should fail the encode pipeline"),
+            Err(err) => err,
+        };
+        assert!(err.to_string().contains("Failed to write data"));
+        tokio::time::timeout(Duration::from_secs(1), reader_dropped)
+            .await
+            .expect("writer failure should abort the blocked producer")
+            .expect("blocked producer should signal reader drop");
+        assert_eq!(
+            writes.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "writer failure must stop the pipeline before any additional shard write"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            gauge_baseline,
+            "writer failure must settle all queued and pending encoded bytes"
+        );
+    }
+
+    async fn aborting_full_queue_settles_pending_send() {
+        const BLOCK_SIZE: usize = 16;
+
+        let gauge_baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let erasure = Arc::new(Erasure::new(1, 0, BLOCK_SIZE));
+        let inflight_blocks = encode_channel_capacity(
+            erasure.shard_size().saturating_mul(erasure.total_shard_count()),
+            erasure_encode_max_inflight_bytes(),
+        );
+        let (reader, _reader_blocked, reader_dropped, final_block) =
+            BlocksThenPendingReader::new(inflight_blocks + 2, BLOCK_SIZE);
+        let (writer_entered_tx, writer_entered) = oneshot::channel();
+        let writes = Arc::new(std::sync::atomic::AtomicUsize::new(0));
+        let mut writers = vec![Some(bitrot_writer(
+            StallOnWriteWithSignal {
+                entered: Some(writer_entered_tx),
+                writes: writes.clone(),
+            },
+            BLOCK_SIZE,
+        ))];
+        let erasure_for_task = erasure.clone();
+        let encode = tokio::spawn(async move { erasure_for_task.encode_with_ingest_mode(reader, &mut writers, 1, false).await });
+
+        tokio::time::timeout(Duration::from_secs(1), writer_entered)
+            .await
+            .expect("consumer should start the first writer call")
+            .expect("stalling writer should signal entry");
+        tokio::time::timeout(Duration::from_secs(1), final_block)
+            .await
+            .expect("producer should supply the block whose send fills the queue")
+            .expect("reader should signal final block");
+
+        let expected_queued_bytes =
+            u64::try_from((inflight_blocks + 1) * BLOCK_SIZE).expect("queued byte count should fit the gauge");
+        tokio::time::timeout(Duration::from_secs(1), async {
+            while rustfs_io_metrics::current_ec_encode_inflight_bytes() < gauge_baseline + expected_queued_bytes {
+                tokio::task::yield_now().await;
+            }
+        })
+        .await
+        .expect("producer should account for the pending send after the queue fills");
+
+        encode.abort();
+        assert!(matches!(encode.await, Err(err) if err.is_cancelled()), "encode task should be cancelled");
+        tokio::time::timeout(Duration::from_secs(1), reader_dropped)
+            .await
+            .expect("cancelling a full queue should abort its producer")
+            .expect("full-queue producer should signal reader drop");
+        assert_eq!(
+            writes.load(std::sync::atomic::Ordering::SeqCst),
+            1,
+            "cancellation must not resume the stalled writer"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            gauge_baseline,
+            "cancelling a full queue must settle queued and pending bytes"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn cancelling_vec_encode_drops_blocked_producer() {
+        aborting_encode_drops_blocked_producer(EncodePipeline::Vec).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn cancelling_bytesmut_encode_drops_blocked_producer() {
+        aborting_encode_drops_blocked_producer(EncodePipeline::BytesMut).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn cancelling_batched_encode_drops_blocked_producer() {
+        aborting_encode_drops_blocked_producer(EncodePipeline::Batched).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn vec_writer_error_aborts_blocked_producer() {
+        writer_error_aborts_blocked_producer(EncodePipeline::Vec).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn bytesmut_writer_error_aborts_blocked_producer() {
+        writer_error_aborts_blocked_producer(EncodePipeline::BytesMut).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn batched_writer_error_aborts_blocked_producer() {
+        writer_error_aborts_blocked_producer(EncodePipeline::Batched).await;
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn cancelling_full_queue_settles_pending_send() {
+        aborting_full_queue_settles_pending_send().await;
+    }
+
     #[tokio::test]
     async fn helper_writers_cover_flush_and_shutdown_paths() {
         let mut failing_write = FailingWriteWriter;
@@ -1329,25 +1733,99 @@ mod tests {
     }
 
     #[tokio::test]
-    async fn drain_queued_inflight_bytes_consumes_pending_blocks() {
-        let (tx, mut rx) = mpsc::channel(2);
-        tx.send(vec![Bytes::from_static(b"queued")]).await.unwrap();
-        drop(tx);
+    #[serial_test::serial]
+    async fn queued_inflight_bytes_are_settled_on_all_queue_exit_paths() {
+        let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let (tx, rx) = mpsc::channel(1);
+        let queued = vec![Bytes::from_static(b"queued")];
+        let queued_bytes = queued_block_bytes(&queued);
 
-        drain_queued_inflight_bytes(&mut rx).await;
+        send_queued(&tx, queued, queued_bytes)
+            .await
+            .expect("first queue entry should fit");
 
-        assert!(rx.recv().await.is_none());
+        let blocked = vec![Bytes::from_static(b"blocked")];
+        let blocked_bytes = queued_block_bytes(&blocked);
+        {
+            let pending_send = send_queued(&tx, blocked, blocked_bytes);
+            tokio::pin!(pending_send);
+            assert!(
+                futures::poll!(pending_send.as_mut()).is_pending(),
+                "full queue must suspend producer send"
+            );
+        }
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            baseline + u64::try_from(queued_bytes).expect("queue bytes fit the gauge"),
+            "dropping a pending producer send must compensate its bytes"
+        );
+
+        drop(rx);
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            baseline,
+            "dropping the receiver must settle every buffered entry"
+        );
+
+        let rejected = vec![Bytes::from_static(b"rejected")];
+        let rejected_bytes = queued_block_bytes(&rejected);
+        assert!(
+            send_queued(&tx, rejected, rejected_bytes).await.is_err(),
+            "closed receiver must reject a new send"
+        );
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            baseline,
+            "failed sends must compensate their bytes"
+        );
     }
 
     #[tokio::test]
-    async fn drain_queued_batched_inflight_bytes_consumes_pending_batches() {
-        let (tx, mut rx) = mpsc::channel(2);
-        tx.send(vec![vec![Bytes::from_static(b"queued")]]).await.unwrap();
-        drop(tx);
+    #[serial_test::serial]
+    async fn queued_batch_entry_settles_bytes_before_handoff() {
+        let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let (tx, rx) = mpsc::channel(2);
+        let mut rx = rx;
+        let batch = vec![vec![Bytes::from_static(b"queued")], vec![Bytes::from_static(b"batch")]];
+        let batch_bytes = queued_batch_bytes(&batch);
 
-        drain_queued_batched_inflight_bytes(&mut rx).await;
+        send_queued(&tx, batch, batch_bytes).await.expect("batch should be queued");
+        let batch = rx.recv().await.expect("queued batch should be received").into_inner();
+        assert_eq!(batch_bytes, queued_batch_bytes(&batch));
 
-        assert!(rx.recv().await.is_none());
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            baseline,
+            "receiving a batch must settle all contained block bytes before shard writes"
+        );
+    }
+
+    #[tokio::test]
+    #[serial_test::serial]
+    async fn queued_entry_settles_when_a_closed_receiver_accepts_an_outstanding_permit() {
+        let baseline = rustfs_io_metrics::current_ec_encode_inflight_bytes();
+        let (tx, mut rx) = mpsc::channel(1);
+        let permit = tx
+            .clone()
+            .reserve_owned()
+            .await
+            .expect("open receiver should reserve queue capacity");
+        rx.close();
+        assert!(
+            matches!(rx.try_recv(), Err(mpsc::error::TryRecvError::Empty)),
+            "an outstanding permit must leave the closed queue observably empty"
+        );
+
+        let block = vec![Bytes::from_static(b"late-permit")];
+        let block_bytes = queued_block_bytes(&block);
+        permit.send(InflightEntry::new(block, block_bytes));
+        drop(rx);
+
+        assert_eq!(
+            rustfs_io_metrics::current_ec_encode_inflight_bytes(),
+            baseline,
+            "a queue entry sent through an outstanding permit must settle when Tokio drops it"
+        );
     }
 
     #[tokio::test]

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

@@ -640,7 +640,7 @@ impl Erasure {
     /// # Returns
     /// A vector of encoded shards as `Bytes`.
     #[tracing::instrument(level = "debug", skip_all, fields(data_len=data.len()))]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub fn encode_data(&self, data: &[u8]) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -688,7 +688,7 @@ impl Erasure {
 
     /// Encode owned data, avoiding a copy when the caller already has a heap buffer.
     /// Falls back to copying into a new buffer if zero-copy conversion fails.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub fn encode_data_owned(&self, data: Vec<u8>) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -752,7 +752,7 @@ impl Erasure {
     /// block), the `resize(need_total_size)` below stays within capacity for every
     /// `data_len <= block_size` — both shard-size formulas are monotone in
     /// `data_len` — so this function never reallocates the buffer.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub fn encode_data_bytes_mut(&self, mut data_buffer: BytesMut, data_len: usize) -> io::Result<Vec<Bytes>> {
         let shard_size_fn = if self.uses_legacy {
             calc_shard_size_legacy
@@ -805,7 +805,7 @@ impl Erasure {
     ///
     /// # Returns
     /// Ok if reconstruction succeeds, error otherwise.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub fn decode_data(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
         if self.parity_shards > 0 {
             if self.uses_legacy {
@@ -825,7 +825,7 @@ impl Erasure {
     }
 
     /// Decode and reconstruct missing data shards, then regenerate parity shards.
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "Erasure")]
     pub fn decode_data_and_parity(&self, shards: &mut [Option<Vec<u8>>]) -> io::Result<()> {
         if self.parity_shards > 0 {
             if self.uses_legacy {

crates/ecstore/src/io_support/bitrot.rs

@@ -19,6 +19,8 @@ use crate::diagnostics::get::{
     GET_STAGE_READER_MMAP_PATH_RESOLVE, GET_STAGE_READER_OPEN_MMAP_COPY_FALLBACK, GET_STAGE_READER_OPEN_MMAP_COPY_SUCCESS,
     GET_STAGE_READER_OPEN_STREAM, GET_STAGE_READER_STREAM_FIRST_READ, record_get_stage_duration_if_enabled,
 };
+#[cfg(feature = "hotpath")]
+use crate::disk::FileWriter;
 use crate::disk::{self, DiskAPI as _, DiskStore, FileReader, MmapCopyStageMetrics, error::DiskError};
 use crate::erasure::coding::{BitrotReader, BitrotWriterWrapper, CustomWriter};
 use bytes::Bytes;
@@ -36,6 +38,11 @@ use std::time::Instant;
 use tokio::io::{AsyncRead, ReadBuf};
 use tracing::debug;
 
+#[cfg(all(test, feature = "hotpath"))]
+tokio::task_local! {
+    static FORCE_MMAP_COPY_FAILURE_FOR_TEST: ();
+}
+
 /// A shard source for the bitrot reader.
 ///
 /// `InMemory` keeps the `Bytes` concrete instead of erasing it behind
@@ -360,10 +367,21 @@ async fn open_disk_reader(
             mmap_copy_stage: GET_STAGE_READER_MMAP_COPY_BUFFER,
             direct_read_copy_stage: GET_STAGE_READER_MMAP_DIRECT_READ_COPY,
         });
-        match disk
-            .read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics)
-            .await
-        {
+        let mmap_result = {
+            #[cfg(all(test, feature = "hotpath"))]
+            if FORCE_MMAP_COPY_FAILURE_FOR_TEST.try_with(|_| ()).is_ok() {
+                Err(DiskError::other("forced mmap-copy failure for test"))
+            } else {
+                disk.read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics)
+                    .await
+            }
+            #[cfg(not(all(test, feature = "hotpath")))]
+            {
+                disk.read_file_mmap_copy_with_metrics(bucket, path, offset, length, mmap_metrics)
+                    .await
+            }
+        };
+        match mmap_result {
             Ok(bytes) => {
                 let duration_ms = zero_copy_start.elapsed().as_secs_f64() * 1000.0;
 
@@ -398,7 +416,11 @@ async fn open_disk_reader(
                 }
 
                 return match stream_result {
-                    Ok(reader) => Ok(wrap_first_read_metrics(reader, metrics_path)),
+                    Ok(reader) => {
+                        #[cfg(feature = "hotpath")]
+                        let reader = instrument_raw_shard_reader(reader, disk.is_local());
+                        Ok(wrap_first_read_metrics(reader, metrics_path))
+                    }
                     Err(_) => Err(err),
                 };
             }
@@ -407,6 +429,8 @@ async fn open_disk_reader(
 
     let stream_start = stage_metrics_enabled.then(Instant::now);
     let reader = disk.read_file_stream(bucket, path, offset, length).await?;
+    #[cfg(feature = "hotpath")]
+    let reader = instrument_raw_shard_reader(reader, disk.is_local());
     if let Some(metrics_path) = metrics_path {
         record_get_stage_duration_if_enabled(metrics_path, GET_STAGE_READER_OPEN_STREAM, stream_start);
     }
@@ -427,6 +451,37 @@ fn wrap_first_read_metrics(reader: FileReader, metrics_path: Option<&'static str
     ShardReader::Stream(reader)
 }
 
+// The labels are deliberately fixed: object keys, disk paths, and remote hosts
+// are all high-cardinality or sensitive and belong nowhere in a profiling report.
+#[cfg(feature = "hotpath")]
+const RAW_SHARD_READ_LOCAL_LABEL: &str = "EC raw shard read local";
+#[cfg(feature = "hotpath")]
+const RAW_SHARD_READ_REMOTE_LABEL: &str = "EC raw shard read remote";
+#[cfg(feature = "hotpath")]
+const RAW_SHARD_WRITE_LOCAL_LABEL: &str = "EC raw shard write local";
+#[cfg(feature = "hotpath")]
+const RAW_SHARD_WRITE_REMOTE_LABEL: &str = "EC raw shard write remote";
+
+#[cfg(feature = "hotpath")]
+fn instrument_raw_shard_reader(reader: FileReader, is_local: bool) -> FileReader {
+    // `io!` aggregates by call site, so the local and remote branches must remain distinct.
+    if is_local {
+        Box::new(hotpath::io!(reader, label = RAW_SHARD_READ_LOCAL_LABEL))
+    } else {
+        Box::new(hotpath::io!(reader, label = RAW_SHARD_READ_REMOTE_LABEL))
+    }
+}
+
+#[cfg(feature = "hotpath")]
+fn instrument_raw_shard_writer(writer: FileWriter, is_local: bool) -> FileWriter {
+    // `io!` aggregates by call site, so the local and remote branches must remain distinct.
+    if is_local {
+        Box::new(hotpath::io!(writer, label = RAW_SHARD_WRITE_LOCAL_LABEL))
+    } else {
+        Box::new(hotpath::io!(writer, label = RAW_SHARD_WRITE_REMOTE_LABEL))
+    }
+}
+
 fn bitrot_encoded_range(offset: usize, length: usize, shard_size: usize, checksum_algo: HashAlgorithm) -> (usize, usize) {
     (
         offset.div_ceil(shard_size) * checksum_algo.size() + offset,
@@ -686,6 +741,8 @@ pub async fn create_bitrot_writer(
         };
 
         let file = disk.create_file("", volume, path, length).await?;
+        #[cfg(feature = "hotpath")]
+        let file = instrument_raw_shard_writer(file, disk.is_local());
         CustomWriter::new_tokio_writer(file)
     } else {
         return Err(DiskError::DiskNotFound);
@@ -698,6 +755,182 @@ pub async fn create_bitrot_writer(
 mod tests {
     use super::*;
 
+    #[cfg(feature = "hotpath")]
+    use crate::cluster::rpc::RemoteDisk;
+    #[cfg(feature = "hotpath")]
+    use crate::cluster::rpc::internode_data_transport::{
+        InternodeDataTransport, InternodeDataTransportCapabilities, ReadStreamRequest, WalkDirStreamRequest, WriteStreamRequest,
+    };
+    #[cfg(feature = "hotpath")]
+    use crate::disk::{Disk, DiskOption, error::Result};
+
+    #[cfg(feature = "hotpath")]
+    #[derive(Debug, Clone, Default)]
+    struct TestRemoteDataTransport {
+        bytes: Arc<Mutex<Vec<u8>>>,
+    }
+
+    #[cfg(feature = "hotpath")]
+    impl TestRemoteDataTransport {
+        fn bytes(&self) -> Vec<u8> {
+            self.bytes
+                .lock()
+                .expect("test remote transport bytes lock should not be poisoned")
+                .clone()
+        }
+    }
+
+    #[cfg(feature = "hotpath")]
+    #[derive(Debug)]
+    struct TestRemoteWriter {
+        bytes: Arc<Mutex<Vec<u8>>>,
+    }
+
+    #[cfg(feature = "hotpath")]
+    impl tokio::io::AsyncWrite for TestRemoteWriter {
+        fn poll_write(self: Pin<&mut Self>, _cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
+            self.bytes
+                .lock()
+                .expect("test remote transport bytes lock should not be poisoned")
+                .extend_from_slice(buf);
+            Poll::Ready(Ok(buf.len()))
+        }
+
+        fn poll_flush(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+
+        fn poll_shutdown(self: Pin<&mut Self>, _cx: &mut Context<'_>) -> Poll<io::Result<()>> {
+            Poll::Ready(Ok(()))
+        }
+    }
+
+    #[cfg(feature = "hotpath")]
+    #[async_trait::async_trait]
+    impl InternodeDataTransport for TestRemoteDataTransport {
+        async fn open_read(&self, _request: ReadStreamRequest) -> Result<FileReader> {
+            Ok(Box::new(Cursor::new(self.bytes())))
+        }
+
+        async fn open_write(&self, _request: WriteStreamRequest) -> Result<FileWriter> {
+            Ok(Box::new(TestRemoteWriter {
+                bytes: Arc::clone(&self.bytes),
+            }))
+        }
+
+        async fn open_walk_dir(&self, _request: WalkDirStreamRequest) -> Result<FileReader> {
+            panic!("open_walk_dir must not be used by the raw shard I/O test")
+        }
+
+        fn name(&self) -> &'static str {
+            "bitrot-test-remote"
+        }
+
+        fn capabilities(&self) -> InternodeDataTransportCapabilities {
+            InternodeDataTransportCapabilities::tcp_http()
+        }
+    }
+
+    async fn local_test_disk() -> (DiskStore, tempfile::TempDir) {
+        use crate::disk::endpoint::Endpoint;
+        use crate::disk::{DiskOption, new_disk};
+
+        let dir = tempfile::tempdir().expect("tempdir should be created");
+        let mut endpoint =
+            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
+        endpoint.set_pool_index(0);
+        endpoint.set_set_index(0);
+        endpoint.set_disk_index(0);
+        let disk = new_disk(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+        )
+        .await
+        .expect("local disk should be created");
+
+        (disk, dir)
+    }
+
+    #[cfg(feature = "hotpath")]
+    async fn remote_test_disk() -> (DiskStore, TestRemoteDataTransport) {
+        use crate::disk::endpoint::Endpoint;
+
+        let endpoint = Endpoint {
+            url: url::Url::parse("http://remote-node:9000/data/rustfs0").expect("test remote endpoint should parse"),
+            is_local: false,
+            pool_idx: 0,
+            set_idx: 0,
+            disk_idx: 0,
+        };
+        let transport = TestRemoteDataTransport::default();
+        let remote = RemoteDisk::new(
+            &endpoint,
+            &DiskOption {
+                cleanup: false,
+                health_check: false,
+            },
+            Arc::new(transport.clone()),
+        )
+        .await
+        .expect("test remote disk should be created");
+
+        (Arc::new(Disk::Remote(Box::new(remote))), transport)
+    }
+
+    async fn round_trip_disk_bitrot(disk: &DiskStore, bucket: &str, path: &str, payload: &[u8], shard_size: usize) -> Vec<u8> {
+        disk.make_volume(bucket).await.expect("volume should be created");
+        let mut writer = create_bitrot_writer(
+            false,
+            Some(disk),
+            bucket,
+            path,
+            i64::try_from(payload.len()).expect("test payload length should fit i64"),
+            shard_size,
+            HashAlgorithm::None,
+        )
+        .await
+        .expect("disk bitrot writer should open the raw shard file");
+        for chunk in payload.chunks(shard_size) {
+            writer
+                .write(chunk)
+                .await
+                .expect("disk bitrot writer should preserve each shard block");
+        }
+        writer.shutdown().await.expect("disk bitrot writer should close cleanly");
+
+        let mut reader = create_bitrot_reader(
+            None,
+            Some(disk),
+            bucket,
+            path,
+            0,
+            payload.len(),
+            shard_size,
+            HashAlgorithm::None,
+            false,
+            false,
+        )
+        .await
+        .expect("disk bitrot reader should open the raw shard file")
+        .expect("disk bitrot reader should exist");
+        let mut actual = Vec::with_capacity(payload.len());
+        while actual.len() < payload.len() {
+            let remaining = payload.len() - actual.len();
+            let mut chunk = vec![0; remaining.min(shard_size)];
+            let read = reader
+                .read(&mut chunk)
+                .await
+                .expect("disk bitrot reader should return the complete shard body");
+            assert!(read > 0, "disk bitrot reader must not end before the expected shard body is complete");
+            actual.extend_from_slice(&chunk[..read]);
+        }
+
+        actual
+    }
+
     #[test]
     fn object_mmap_read_enabled_accepts_legacy_zero_copy_alias() {
         temp_env::with_vars(
@@ -724,6 +957,165 @@ mod tests {
         );
     }
 
+    #[cfg(feature = "hotpath")]
+    #[test]
+    fn raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes() {
+        const CHILD_ENV: &str = "RUSTFS_HOTPATH_RAW_SHARD_IO_TEST_CHILD";
+        if std::env::var_os(CHILD_ENV).is_none() {
+            let status = std::process::Command::new(std::env::current_exe().expect("test executable path should be available"))
+                .arg("--exact")
+                .arg("io_support::bitrot::tests::raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes")
+                .arg("--nocapture")
+                .env(CHILD_ENV, "1")
+                .status()
+                .expect("isolated HotPath I/O test process should start");
+            assert!(status.success(), "isolated HotPath I/O test process should pass");
+            return;
+        }
+
+        tokio::runtime::Builder::new_current_thread()
+            .enable_all()
+            .build()
+            .expect("test runtime should be created")
+            .block_on(raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes_in_isolated_process());
+    }
+
+    #[cfg(feature = "hotpath")]
+    async fn raw_shard_io_wrappers_report_fixed_labels_and_preserve_bytes_in_isolated_process() {
+        use hotpath::{Format, HotpathGuardBuilder, Section};
+        use tokio::io::AsyncReadExt;
+
+        let report_dir = tempfile::tempdir().expect("report tempdir should be created");
+        let report_path = report_dir.path().join("hotpath-io.json");
+        let guard = HotpathGuardBuilder::new("raw_shard_io_test")
+            .format(Format::Json)
+            .output_path(&report_path)
+            .sections(vec![Section::Io])
+            .build();
+
+        let (disk, _dir) = local_test_disk().await;
+        let bucket = "test-bucket";
+        let path = "obj/hotpath-part.1";
+        let payload = b"local shard bytes";
+        let shard_size = 4;
+        let local_read = round_trip_disk_bitrot(&disk, bucket, path, payload, shard_size).await;
+        assert_eq!(local_read, payload, "local raw shard I/O instrumentation must not alter stored bytes");
+
+        let fallback_path = "obj/hotpath-mmap-fallback-part.1";
+        let fallback_payload = b"mmap fallback shard bytes";
+        disk.write_all("test-bucket", fallback_path, Bytes::from_static(fallback_payload))
+            .await
+            .expect("fallback shard file should be written");
+        let mut fallback_reader = FORCE_MMAP_COPY_FAILURE_FOR_TEST
+            .scope((), open_disk_reader(&disk, bucket, fallback_path, 0, fallback_payload.len(), true, None))
+            .await
+            .expect("mmap-copy failure should fall back to a raw shard stream");
+        assert!(
+            matches!(fallback_reader, ShardReader::Stream(_)),
+            "forced mmap-copy failure must use the streaming fallback"
+        );
+        let mut fallback_read = Vec::new();
+        fallback_reader
+            .read_to_end(&mut fallback_read)
+            .await
+            .expect("mmap-copy fallback stream should preserve bytes");
+        assert_eq!(fallback_read, fallback_payload);
+
+        let (remote_disk, remote_transport) = remote_test_disk().await;
+        let remote_payload = b"remote shard bytes";
+        let mut remote_writer = create_bitrot_writer(
+            false,
+            Some(&remote_disk),
+            bucket,
+            "obj/hotpath-remote-part.1",
+            i64::try_from(remote_payload.len()).expect("remote payload length should fit i64"),
+            shard_size,
+            HashAlgorithm::None,
+        )
+        .await
+        .expect("remote bitrot writer should use the production raw writer path");
+        for chunk in remote_payload.chunks(shard_size) {
+            remote_writer
+                .write(chunk)
+                .await
+                .expect("remote bitrot writer should preserve bytes");
+        }
+        remote_writer
+            .shutdown()
+            .await
+            .expect("remote bitrot writer should close cleanly");
+        assert_eq!(remote_transport.bytes(), remote_payload);
+
+        let mut reader = create_bitrot_reader(
+            None,
+            Some(&remote_disk),
+            bucket,
+            "obj/hotpath-remote-part.1",
+            0,
+            remote_payload.len(),
+            shard_size,
+            HashAlgorithm::None,
+            false,
+            true,
+        )
+        .await
+        .expect("remote bitrot reader should use the production raw reader path")
+        .expect("remote bitrot reader should exist");
+        let mut remote_read = Vec::with_capacity(remote_payload.len());
+        while remote_read.len() < remote_payload.len() {
+            let remaining = remote_payload.len() - remote_read.len();
+            let mut chunk = vec![0; remaining.min(shard_size)];
+            let read = reader
+                .read(&mut chunk)
+                .await
+                .expect("remote bitrot reader should preserve bytes");
+            assert!(read > 0, "remote bitrot reader must not end before the expected shard body is complete");
+            remote_read.extend_from_slice(&chunk[..read]);
+        }
+        assert_eq!(remote_read, remote_payload);
+
+        drop(guard);
+        let report = std::fs::read_to_string(&report_path).expect("HotPath I/O report should be written");
+        let report: serde_json::Value = serde_json::from_str(&report).expect("HotPath I/O report should be valid JSON");
+        let entries = report["io"]["data"]
+            .as_array()
+            .expect("HotPath I/O report should include data rows");
+        let io_bytes = |label: &str, direction: &str| {
+            let byte_count: u64 = entries
+                .iter()
+                .filter(|entry| entry["label"].as_str().is_some_and(|entry_label| entry_label == label))
+                .filter_map(|entry| entry[direction]["bytes"].as_u64())
+                .sum();
+            assert!(byte_count > 0, "report must include fixed label {label}");
+            byte_count
+        };
+        let payload_len = u64::try_from(payload.len()).expect("test payload length should fit u64");
+        assert_eq!(
+            io_bytes(RAW_SHARD_READ_LOCAL_LABEL, "read"),
+            payload_len + u64::try_from(fallback_payload.len()).expect("fallback payload length should fit u64")
+        );
+        assert_eq!(io_bytes(RAW_SHARD_WRITE_LOCAL_LABEL, "write"), payload_len);
+        assert_eq!(
+            io_bytes(RAW_SHARD_READ_REMOTE_LABEL, "read"),
+            u64::try_from(remote_payload.len()).expect("remote payload length should fit u64")
+        );
+        assert_eq!(
+            io_bytes(RAW_SHARD_WRITE_REMOTE_LABEL, "write"),
+            u64::try_from(remote_payload.len()).expect("remote payload length should fit u64")
+        );
+        for label in [
+            RAW_SHARD_READ_LOCAL_LABEL,
+            RAW_SHARD_READ_REMOTE_LABEL,
+            RAW_SHARD_WRITE_LOCAL_LABEL,
+            RAW_SHARD_WRITE_REMOTE_LABEL,
+        ] {
+            assert!(
+                !label.contains(['/', ':', '?', '@']),
+                "raw shard I/O labels must not carry a path, host, query, or credential delimiter: {label}"
+            );
+        }
+    }
+
     #[test]
     fn object_mmap_read_max_length_defaults_and_env_override() {
         temp_env::with_var(ENV_OBJECT_MMAP_READ_MAX_LENGTH, None::<&str>, || {
@@ -741,25 +1133,9 @@ mod tests {
     // be materialized in memory by the mmap-copy path; over-cap reads stream.
     #[tokio::test]
     async fn open_disk_reader_streams_when_length_exceeds_mmap_cap() {
-        use crate::disk::endpoint::Endpoint;
-        use crate::disk::{DiskOption, new_disk};
         use tokio::io::AsyncReadExt;
 
-        let dir = tempfile::tempdir().expect("tempdir should be created");
-        let mut endpoint =
-            Endpoint::try_from(dir.path().to_str().expect("tempdir path should be utf8")).expect("endpoint should parse");
-        endpoint.set_pool_index(0);
-        endpoint.set_set_index(0);
-        endpoint.set_disk_index(0);
-        let disk = new_disk(
-            &endpoint,
-            &DiskOption {
-                cleanup: false,
-                health_check: false,
-            },
-        )
-        .await
-        .expect("local disk should be created");
+        let (disk, _dir) = local_test_disk().await;
 
         let payload = vec![7u8; 4096];
         disk.make_volume("test-bucket").await.expect("volume should be created");
@@ -814,6 +1190,17 @@ mod tests {
         .await;
     }
 
+    #[tokio::test]
+    async fn disk_bitrot_reader_and_writer_preserve_full_shard_body() {
+        let (disk, _dir) = local_test_disk().await;
+        let bucket = "test-bucket";
+        let path = "obj/wrapped-part.1";
+        let payload = b"wrapped shard body";
+        let actual = round_trip_disk_bitrot(&disk, bucket, path, payload, 4).await;
+
+        assert_eq!(actual, payload, "raw shard I/O instrumentation must not alter stored bytes");
+    }
+
     #[tokio::test]
     async fn test_create_bitrot_reader_with_inline_data() {
         let test_data = b"hello world test data";

crates/ecstore/src/set_disk/read.rs

@@ -199,7 +199,7 @@ impl SetDisks {
         );
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub async fn read_version_optimized(
         &self,
         bucket: &str,
@@ -238,7 +238,7 @@ impl SetDisks {
     }
 
     #[tracing::instrument(level = "debug", skip(self))]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_fileinfo(
         &self,
         bucket: &str,
@@ -410,7 +410,7 @@ impl SetDisks {
         Ok((fi, parts_metadata, op_online_disks))
     }
 
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_info_and_quorum(
         &self,
         bucket: &str,
@@ -605,7 +605,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_with_fileinfo<W>(
         // &self,
         bucket: &str,
@@ -1140,7 +1140,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     pub(super) async fn get_object_decode_reader_with_fileinfo(
         bucket: &str,
         object: &str,
@@ -1296,7 +1296,7 @@ impl SetDisks {
     }
 
     #[allow(clippy::too_many_arguments)]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "SetDisks")]
     async fn build_codec_streaming_part_reader(
         bucket: &str,
         object: &str,

crates/ecstore/src/store/multipart.rs

@@ -238,7 +238,7 @@ impl ECStore {
     }
 
     #[instrument(skip(self, data))]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "ECStore")]
     pub(super) async fn handle_put_object_part(
         &self,
         bucket: &str,

crates/ecstore/src/store/object.rs

@@ -815,7 +815,7 @@ impl ECStore {
     }
 
     #[instrument(level = "debug", skip(self))]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "ECStore")]
     pub(super) async fn handle_get_object_reader(
         &self,
         bucket: &str,
@@ -849,7 +849,7 @@ impl ECStore {
     }
 
     #[instrument(level = "debug", skip(self, data))]
-    #[hotpath::measure]
+    #[hotpath::measure(impl_type = "ECStore")]
     pub(super) async fn handle_put_object(
         &self,
         bucket: &str,

crates/ecstore/tests/hotpath_cpu_attribution_contract_test.rs

@@ -0,0 +1,95 @@
+// 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.
+
+const STORE_MULTIPART: &str = include_str!("../src/store/multipart.rs");
+const STORE_OBJECT: &str = include_str!("../src/store/object.rs");
+const ERASURE: &str = include_str!("../src/erasure/coding/erasure.rs");
+const DECODE: &str = include_str!("../src/erasure/coding/decode.rs");
+const ENCODE: &str = include_str!("../src/erasure/coding/encode.rs");
+const LOCAL_DISK: &str = include_str!("../src/disk/local.rs");
+const BITROT: &str = include_str!("../src/erasure/coding/bitrot.rs");
+const SET_DISK_READ: &str = include_str!("../src/set_disk/read.rs");
+
+fn assert_measured_as(source: &str, impl_type: &str, function: &str) {
+    let attribute = format!("#[hotpath::measure(impl_type = \"{impl_type}\")]\n");
+    let function_start = format!("fn {function}");
+    let mut remaining = source;
+
+    while let Some(attribute_offset) = remaining.find(&attribute) {
+        let measured_source = &remaining[attribute_offset + attribute.len()..];
+        if measured_source.find(&function_start).is_some_and(|offset| offset < 256) {
+            return;
+        }
+        remaining = measured_source;
+    }
+
+    panic!("missing HotPath impl_type={impl_type} for {function}");
+}
+
+#[test]
+fn inherent_hotpath_measurements_have_cpu_attribution_types() {
+    assert_measured_as(STORE_MULTIPART, "ECStore", "handle_put_object_part");
+    for function in ["handle_get_object_reader", "handle_put_object"] {
+        assert_measured_as(STORE_OBJECT, "ECStore", function);
+    }
+    for function in [
+        "encode_data",
+        "encode_data_owned",
+        "encode_data_bytes_mut",
+        "decode_data",
+        "decode_data_and_parity",
+    ] {
+        assert_measured_as(ERASURE, "Erasure", function);
+    }
+    assert_measured_as(DECODE, "ParallelReader", "read");
+    assert_measured_as(DECODE, "Erasure", "decode");
+    for function in [
+        "encode",
+        "encode_batched",
+        "encode_inline_small",
+        "encode_single_block_non_inline",
+    ] {
+        assert_measured_as(ENCODE, "Erasure", function);
+    }
+    for function in ["read_metadata_with_dmtime", "read_all_data"] {
+        assert_measured_as(LOCAL_DISK, "LocalDisk", function);
+    }
+    assert_measured_as(BITROT, "BitrotReader", "read");
+    assert!(
+        BITROT.contains("#[hotpath::measure(label = \"BitrotWriter::write\", impl_type = \"BitrotWriter\")]"),
+        "BitrotWriter::write must retain its stable label and CPU impl_type",
+    );
+    for function in [
+        "read_version_optimized",
+        "get_object_fileinfo",
+        "get_object_info_and_quorum",
+        "get_object_with_fileinfo",
+        "get_object_decode_reader_with_fileinfo",
+        "build_codec_streaming_part_reader",
+    ] {
+        assert_measured_as(SET_DISK_READ, "SetDisks", function);
+    }
+}
+
+#[test]
+fn module_hotpath_measurements_remain_untyped() {
+    assert!(
+        BITROT.contains("#[hotpath::measure]\npub async fn bitrot_verify"),
+        "bitrot_verify is a module function and must not claim an impl type",
+    );
+    assert!(
+        SET_DISK_READ.contains("#[hotpath::measure]\nasync fn setup_multipart_part_readers"),
+        "setup_multipart_part_readers is a module function and must not claim an impl type",
+    );
+}

crates/iam/src/sys.rs

@@ -90,7 +90,13 @@ fn get_policy_plugin_state() -> Arc<RwLock<PolicyPluginState>> {
                         }
                         Ok(_) => PolicyPluginState::Failed,
                         Err(e) => {
-                            error!("Error loading OPA configuration err:{}", e);
+                            error!(
+                                component = "iam",
+                                subsystem = "policy_plugin",
+                                result = "configuration_load_failed",
+                                error_kind = e.kind(),
+                                "OPA plugin configuration load failed"
+                            );
                             PolicyPluginState::Failed
                         }
                     };

crates/io-metrics/src/lib.rs

@@ -2195,18 +2195,22 @@ pub fn record_allocator_memory_observation(backend: &'static str, observation: A
     }
 }
 
-/// Track encoded bytes currently queued between erasure encode and disk writers.
+/// Track encoded bytes in the queue hand-off between erasure encode and disk writers.
+///
+/// This is queue occupancy, not a per-request or process-RSS memory limit. The
+/// erasure encoder settles it on failed/cancelled sends, receiver drop, and
+/// consumer hand-off before shard writes begin.
 #[inline(always)]
 pub fn add_ec_encode_inflight_bytes(bytes: usize) {
     let next = EC_ENCODE_INFLIGHT_BYTES.fetch_add(bytes as u64, Ordering::Relaxed) + bytes as u64;
-    gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64);
+    gauge_set_cached!("rustfs_ec_encode_inflight_bytes_current", next as f64);
 }
 
 /// Remove encoded bytes from the tracked erasure encode in-flight gauge.
 #[inline(always)]
 pub fn remove_ec_encode_inflight_bytes(bytes: usize) {
     let next = saturating_sub_atomic(&EC_ENCODE_INFLIGHT_BYTES, bytes as u64);
-    gauge!("rustfs_ec_encode_inflight_bytes_current").set(next as f64);
+    gauge_set_cached!("rustfs_ec_encode_inflight_bytes_current", next as f64);
 }
 
 /// Return the current tracked EC encode in-flight bytes.

crates/obs/Cargo.toml

@@ -146,6 +146,7 @@ tracing-opentelemetry = { workspace = true }
 tracing-subscriber = { workspace = true, features = ["fmt", "env-filter", "tracing-log", "local-time", "json"] }
 tokio = { workspace = true, features = ["sync", "rt-multi-thread", "time", "macros"] }
 tokio-util = { workspace = true, features = ["io", "compat"] }
+url.workspace = true
 dial9-tokio-telemetry = { workspace = true, optional = true }
 thiserror = { workspace = true }
 zstd = { workspace = true, features = ["zstdmt"] }
@@ -162,3 +163,4 @@ libc = { workspace = true }
 [dev-dependencies]
 tempfile = { workspace = true }
 temp-env = { workspace = true }
+log = "0.4"

crates/obs/src/telemetry/filter.rs

@@ -18,7 +18,26 @@
 //! used across different logging backends (stdout, file, OpenTelemetry).
 
 use std::env;
-use tracing_subscriber::{EnvFilter, filter::LevelFilter};
+use tracing::Metadata;
+use tracing_subscriber::{
+    EnvFilter,
+    filter::{FilterFn, LevelFilter, filter_fn},
+};
+
+/// Pyroscope emits raw reqwest errors from its background session manager.
+/// Those errors can include the configured endpoint, so they must never reach
+/// a RustFS logging sink even when an operator enables verbose dependency logs.
+pub(super) fn is_pyroscope_dependency_target(target: &str) -> bool {
+    target == "pyroscope" || target.starts_with("pyroscope::") || target.starts_with("Pyroscope::")
+}
+
+fn allows_non_pyroscope_dependency_event(metadata: &Metadata<'_>) -> bool {
+    !is_pyroscope_dependency_target(metadata.target())
+}
+
+pub(super) fn pyroscope_log_filter() -> FilterFn {
+    filter_fn(allows_non_pyroscope_dependency_event)
+}
 
 /// Build an `EnvFilter` from the given log level string.
 ///

crates/obs/src/telemetry/local.rs

@@ -36,7 +36,7 @@ use crate::cleaner::LogCleaner;
 use crate::cleaner::types::{CompressionAlgorithm, FileMatchMode};
 use crate::config::OtelConfig;
 use crate::global::{METRIC_LOG_CLEANER_RUN_FAILURES_TOTAL, METRIC_LOG_CLEANER_RUNS_TOTAL, set_observability_metric_enabled};
-use crate::telemetry::filter::build_env_filter;
+use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter};
 use crate::telemetry::rolling::{RollingAppender, Rotation};
 use metrics::counter;
 use rustfs_config::observability::{
@@ -319,6 +319,7 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo
 
     tracing_subscriber::registry()
         .with(env_filter)
+        .with(pyroscope_log_filter())
         .with(ErrorLayer::default())
         .with(fmt_layer)
         .try_init()
@@ -431,6 +432,7 @@ fn init_file_logging_internal(
 
     tracing_subscriber::registry()
         .with(env_filter)
+        .with(pyroscope_log_filter())
         .with(ErrorLayer::default())
         .with(file_layer)
         .with(stdout_layer)

crates/obs/src/telemetry/otel.rs

@@ -39,7 +39,7 @@
 use crate::cleaner::types::FileMatchMode;
 use crate::config::OtelConfig;
 use crate::global::set_observability_metric_enabled;
-use crate::telemetry::filter::build_env_filter;
+use crate::telemetry::filter::{build_env_filter, pyroscope_log_filter};
 use crate::telemetry::guard::{OtelGuard, ProfilingAgent};
 use crate::telemetry::local::{build_json_log_layer, spawn_cleanup_task};
 use crate::telemetry::recorder::{Recorder, install_process_global_recorder};
@@ -68,7 +68,7 @@ use std::{fs, io::IsTerminal, time::Duration};
 use tracing::{info, warn};
 use tracing_error::ErrorLayer;
 use tracing_opentelemetry::{MetricsLayer, OpenTelemetryLayer};
-use tracing_subscriber::{Layer, fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt};
+use tracing_subscriber::{fmt::format::FmtSpan, layer::SubscriberExt, util::SubscriberInitExt};
 
 const GET_OBJECT_DURATION_HISTOGRAM_METRICS: &[&str] = &[
     "rustfs_io_get_object_request_duration_seconds",
@@ -83,6 +83,26 @@ const GET_OBJECT_DURATION_HISTOGRAM_BUCKETS: &[f64] = &[
     0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
 ];
 
+#[cfg(all(
+    feature = "pyroscope",
+    any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))
+))]
+const REDACTED_PROFILING_ENDPOINT: &str = "[redacted]";
+
+#[cfg(all(
+    feature = "pyroscope",
+    any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))
+))]
+fn log_profiler_failure(result: &'static str, error_kind: &'static str) {
+    warn!(
+        backend = "pyroscope",
+        endpoint = REDACTED_PROFILING_ENDPOINT,
+        result,
+        error_kind,
+        "Profiling export agent initialization failed"
+    );
+}
+
 /// Initialize the full OpenTelemetry HTTP pipeline (traces + metrics + logs).
 ///
 /// This function is invoked when at least one OTLP endpoint has been
@@ -158,9 +178,7 @@ pub(super) fn init_observability_http(
         logger_provider = build_logger_provider(&log_ep, config, res, use_stdout)?;
 
         // Build bridge to capture `tracing` events.
-        otel_bridge = logger_provider
-            .as_ref()
-            .map(|p| OpenTelemetryTracingBridge::new(p).with_filter(build_env_filter(logger_level, None)));
+        otel_bridge = logger_provider.as_ref().map(OpenTelemetryTracingBridge::new);
 
         // No separate formatting layer is added here; when OTLP logging is
         // active, the OpenTelemetry bridge is the authoritative sink for
@@ -256,6 +274,7 @@ pub(super) fn init_observability_http(
     let filter = build_env_filter(logger_level, None);
     tracing_subscriber::registry()
         .with(filter)
+        .with(pyroscope_log_filter())
         .with(ErrorLayer::default())
         .with(file_layer_opt)
         .with(stdout_layer_opt)
@@ -531,6 +550,11 @@ pub(super) fn init_profiler(config: &OtelConfig) -> Option<ProfilingAgent> {
         return None;
     };
 
+    if url::Url::parse(endpoint).is_err() {
+        log_profiler_failure("profiling_endpoint_invalid", "invalid_endpoint");
+        return None;
+    }
+
     // Configure Pyroscope Agent
     let backend = pprof_backend(PprofConfig::default(), BackendConfig::default());
     let service_name = config.service_name.as_deref().unwrap_or(APP_NAME);
@@ -542,28 +566,16 @@ pub(super) fn init_profiler(config: &OtelConfig) -> Option<ProfilingAgent> {
         .build()
     {
         Ok(agent) => agent,
-        Err(err) => {
-            warn!(
-                backend = "pyroscope",
-                endpoint,
-                result = "profiling_agent_build_failed",
-                error = %err,
-                "Profiling export agent initialization failed"
-            );
+        Err(_) => {
+            log_profiler_failure("profiling_agent_build_failed", "build");
             return None;
         }
     };
 
     match agent.start() {
         Ok(agent) => Some(agent),
-        Err(err) => {
-            warn!(
-                backend = "pyroscope",
-                endpoint,
-                result = "profiling_agent_start_failed",
-                error = ?err,
-                "Profiling export agent failed to start"
-            );
+        Err(_) => {
+            log_profiler_failure("profiling_agent_start_failed", "start");
             None
         }
     }
@@ -650,6 +662,34 @@ fn resolve_signal_timeout(common_timeout_millis: Option<u64>, signal_timeout_mil
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::env;
+    use std::io::{self, Write};
+    use std::process::Command;
+    use std::sync::{Arc, Mutex};
+
+    const LOG_TRACER_CHILD_ENV: &str = "RUSTFS_OBS_LOG_TRACER_CHILD";
+
+    #[derive(Clone)]
+    struct TestWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl Write for TestWriter {
+        fn write(&mut self, bytes: &[u8]) -> io::Result<usize> {
+            self.0.lock().expect("lock test log buffer").extend_from_slice(bytes);
+            Ok(bytes.len())
+        }
+
+        fn flush(&mut self) -> io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestWriter {
+        type Writer = Self;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            self.clone()
+        }
+    }
 
     #[test]
     /// Valid ratios should produce trace-id-ratio sampling.
@@ -766,4 +806,107 @@ mod tests {
         assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.first(), Some(&0.0001));
         assert_eq!(GET_OBJECT_DURATION_HISTOGRAM_BUCKETS.last(), Some(&10.0));
     }
+
+    #[cfg(all(
+        feature = "pyroscope",
+        any(target_os = "macos", all(target_os = "linux", target_env = "gnu", target_arch = "x86_64"))
+    ))]
+    #[test]
+    fn test_init_profiler_invalid_endpoint_redacts_sensitive_components() {
+        let endpoint = "https://profile-user:profile-token@10.24.0.5:invalid/private/profiles?access_token=query-secret";
+        let buffer = Arc::new(Mutex::new(Vec::new()));
+        let writer = TestWriter(Arc::clone(&buffer));
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_writer(writer)
+            .finish();
+
+        let config = OtelConfig {
+            profiling_export_enabled: Some(true),
+            profiling_endpoint: Some(endpoint.to_string()),
+            ..OtelConfig::default()
+        };
+
+        tracing::subscriber::with_default(subscriber, || assert!(init_profiler(&config).is_none()));
+
+        let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log");
+        assert!(rendered.contains(REDACTED_PROFILING_ENDPOINT));
+        assert!(rendered.contains("error_kind=\"invalid_endpoint\""));
+        for leaked in [
+            "profile-user",
+            "profile-token",
+            "10.24.0.5",
+            "private/profiles",
+            "query-secret",
+        ] {
+            assert!(!rendered.contains(leaked), "profiling log leaked {leaked}: {rendered}");
+        }
+    }
+
+    #[test]
+    fn test_pyroscope_upload_failure_targets_stay_filtered_when_env_filter_enables_them() {
+        let endpoint = "https://profile-user:profile-token@10.24.0.5/private/profiles?access_token=query-secret";
+        let buffer = Arc::new(Mutex::new(Vec::new()));
+        let writer = TestWriter(Arc::clone(&buffer));
+        let env_filter = tracing_subscriber::EnvFilter::new("pyroscope=trace")
+            .add_directive("trace".parse().expect("parse trace filter directive"))
+            .add_directive("Pyroscope::Session=trace".parse().expect("parse Pyroscope filter directive"));
+        let subscriber = tracing_subscriber::registry()
+            .with(env_filter)
+            .with(pyroscope_log_filter())
+            .with(
+                tracing_subscriber::fmt::layer()
+                    .with_ansi(false)
+                    .without_time()
+                    .with_writer(writer),
+            );
+
+        tracing::subscriber::with_default(subscriber, || {
+            tracing::error!(target: "pyroscope::session", "SessionManager - Failed to send session: {endpoint}");
+            tracing::error!(target: "Pyroscope::Session", "SessionManager - Failed to send session: {endpoint}");
+        });
+
+        let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log");
+        assert!(rendered.is_empty(), "filtered Pyroscope upload failure reached a log sink: {rendered}");
+    }
+
+    #[test]
+    fn test_pyroscope_log_facade_upload_failure_is_filtered() {
+        let endpoint = "https://profile-user:profile-token@10.24.0.5/private/profiles?access_token=query-secret";
+        if env::var_os(LOG_TRACER_CHILD_ENV).is_some() {
+            tracing_subscriber::registry()
+                .with(build_env_filter("info", None))
+                .with(pyroscope_log_filter())
+                .with(tracing_subscriber::fmt::layer().with_ansi(false).without_time())
+                .try_init()
+                .expect("install isolated LogTracer subscriber");
+            log::error!(target: "pyroscope::session", "SessionManager - Failed to send session: {endpoint}");
+            log::error!(target: "Pyroscope::Session", "SessionManager - Failed to send session: {endpoint}");
+            return;
+        }
+
+        let output = Command::new(env::current_exe().expect("resolve test executable"))
+            .args([
+                "--exact",
+                "telemetry::otel::tests::test_pyroscope_log_facade_upload_failure_is_filtered",
+                "--nocapture",
+            ])
+            .env(LOG_TRACER_CHILD_ENV, "1")
+            .env("RUST_LOG", "trace,pyroscope=trace,Pyroscope::Session=trace")
+            .output()
+            .expect("run isolated LogTracer test process");
+        assert!(output.status.success(), "isolated LogTracer test failed: {output:?}");
+
+        let rendered = format!("{}{}", String::from_utf8_lossy(&output.stdout), String::from_utf8_lossy(&output.stderr));
+        for leaked in [
+            "profile-user",
+            "profile-token",
+            "10.24.0.5",
+            "private/profiles",
+            "query-secret",
+        ] {
+            assert!(!rendered.contains(leaked), "LogTracer path leaked {leaked}: {rendered}");
+        }
+    }
 }

crates/policy/Cargo.toml

@@ -32,7 +32,6 @@ workspace = true
 default = []
 hotpath = [
     "hotpath/hotpath",
-    "hotpath/reqwest-0-13",
     "rustfs-config/hotpath",
     "rustfs-credentials/hotpath",
     "rustfs-crypto/hotpath",
@@ -80,6 +79,7 @@ pollster.workspace = true
 proptest = "1"
 test-case.workspace = true
 temp-env = { workspace = true }
+tracing-subscriber = { workspace = true, features = ["fmt"] }
 
 [lib]
 doctest = false

crates/policy/src/policy/opa.rs

@@ -36,15 +36,10 @@ pub fn is_configured() -> bool {
 
 #[derive(Debug, Clone)]
 pub struct AuthZPlugin {
-    client: OpaHttpClient,
+    client: reqwest::Client,
     args: Args,
 }
 
-#[cfg(feature = "hotpath")]
-type OpaHttpClient = hotpath::wrap::reqwest::Client;
-#[cfg(not(feature = "hotpath"))]
-type OpaHttpClient = reqwest::Client;
-
 #[derive(Debug, thiserror::Error)]
 pub enum OpaConfigError {
     #[error("Missing required env var: {0}")]
@@ -63,6 +58,44 @@ pub enum OpaConfigError {
     Connection(reqwest::Error),
 }
 
+impl OpaConfigError {
+    pub fn kind(&self) -> &'static str {
+        match self {
+            Self::MissingRequiredEnv(_) => "missing_required_env",
+            Self::InvalidEnvVars(_) => "invalid_env_vars",
+            Self::EnvRead { .. } => "env_read_failed",
+            Self::InvalidStatus(_) => "invalid_status",
+            Self::Connection(_) => "connection_failed",
+        }
+    }
+}
+
+fn redact_opa_connection_error(error: reqwest::Error) -> OpaConfigError {
+    OpaConfigError::Connection(error.without_url())
+}
+
+fn opa_request_error_kind(error: &reqwest::Error) -> &'static str {
+    if error.is_timeout() {
+        "timeout"
+    } else if error.is_connect() {
+        "connect"
+    } else if error.is_request() {
+        "request"
+    } else if error.is_body() {
+        "body"
+    } else if error.is_decode() {
+        "decode"
+    } else if error.is_status() {
+        "status"
+    } else if error.is_builder() {
+        "builder"
+    } else if error.is_redirect() {
+        "redirect"
+    } else {
+        "other"
+    }
+}
+
 fn check() -> Result<(), OpaConfigError> {
     let env_list = env::vars();
     let mut candidate = HashMap::new();
@@ -90,7 +123,7 @@ async fn validate(config: &Args) -> Result<(), OpaConfigError> {
         .timeout(Duration::from_secs(5))
         .connect_timeout(Duration::from_secs(1))
         .build()
-        .map_err(OpaConfigError::Connection)?;
+        .map_err(redact_opa_connection_error)?;
 
     let mut request = client.post(&config.url);
     if !config.auth_token.is_empty() {
@@ -109,7 +142,7 @@ async fn validate(config: &Args) -> Result<(), OpaConfigError> {
             };
         }
         Err(err) => {
-            return Err(OpaConfigError::Connection(err));
+            return Err(redact_opa_connection_error(err));
         }
     };
     Ok(())
@@ -155,9 +188,6 @@ impl AuthZPlugin {
                 reqwest::Client::new()
             });
 
-        #[cfg(feature = "hotpath")]
-        let client = hotpath::http!(client, label = "Policy::OPA");
-
         Self { client, args: config }
     }
 
@@ -173,7 +203,13 @@ impl AuthZPlugin {
             Ok(resp) => {
                 let status = resp.status();
                 if !status.is_success() {
-                    error!("OPA returned non-success status: {}", status);
+                    error!(
+                        component = "policy",
+                        subsystem = "opa",
+                        result = "request_rejected",
+                        status = %status,
+                        "OPA request rejected"
+                    );
                     return false;
                 }
 
@@ -183,13 +219,25 @@ impl AuthZPlugin {
                         OpaResponseEnum::AllowResult(response) => response.result.allow,
                     },
                     Err(err) => {
-                        error!("Error parsing OPA response: {:?}", err);
+                        error!(
+                            component = "policy",
+                            subsystem = "opa",
+                            result = "response_decode_failed",
+                            error_kind = opa_request_error_kind(&err),
+                            "OPA response decoding failed"
+                        );
                         false
                     }
                 }
             }
             Err(err) => {
-                error!("Error sending request to OPA: {:?}", err);
+                error!(
+                    component = "policy",
+                    subsystem = "opa",
+                    result = "request_failed",
+                    error_kind = opa_request_error_kind(&err),
+                    "OPA request failed"
+                );
                 false
             }
         }
@@ -258,8 +306,35 @@ enum OpaResponseEnum {
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::io::{Read, Write};
+    use std::net::TcpListener;
+    use std::sync::{Arc, Mutex};
     use temp_env;
 
+    #[derive(Clone)]
+    struct TestWriter(Arc<Mutex<Vec<u8>>>);
+
+    impl Write for TestWriter {
+        fn write(&mut self, bytes: &[u8]) -> std::io::Result<usize> {
+            self.0.lock().expect("lock test log buffer").extend_from_slice(bytes);
+            Ok(bytes.len())
+        }
+
+        fn flush(&mut self) -> std::io::Result<()> {
+            Ok(())
+        }
+    }
+
+    impl<'a> tracing_subscriber::fmt::MakeWriter<'a> for TestWriter {
+        type Writer = Self;
+
+        fn make_writer(&'a self) -> Self::Writer {
+            self.clone()
+        }
+    }
+
+    fn assert_reqwest_client(_: &reqwest::Client) {}
+
     #[test]
     fn test_check_valid_config() {
         // Use temp_env to temporarily set environment variables
@@ -345,4 +420,130 @@ mod tests {
         };
         assert!(!args_disabled.enable());
     }
+
+    #[test]
+    fn test_sensitive_opa_endpoints_never_use_hotpath_http_wrapper() {
+        let endpoints = [
+            "https://profile-user:profile-token@10.24.0.5/private/opa?access_token=query-secret",
+            "https://service.internal.example/private/path?token=another-secret",
+        ];
+
+        for endpoint in endpoints {
+            let plugin = AuthZPlugin::new(Args {
+                url: endpoint.to_string(),
+                auth_token: "opa-auth-token".to_string(),
+            });
+
+            assert_reqwest_client(&plugin.client);
+            assert_eq!(plugin.args.url, endpoint);
+        }
+    }
+
+    #[test]
+    fn test_opa_connection_error_removes_sensitive_endpoint() {
+        let listener = TcpListener::bind("127.0.0.1:0").expect("bind local test listener");
+        let port = listener.local_addr().expect("read local test listener address").port();
+        let server = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().expect("accept local test request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).expect("read local test request");
+            assert!(read > 0, "local test client must send an HTTP request");
+            stream
+                .write_all(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("write local test response");
+        });
+        let endpoint = format!("http://profile-user:profile-token@127.0.0.1:{port}/private/opa?access_token=query-secret");
+        let runtime = tokio::runtime::Runtime::new().expect("build tokio test runtime");
+        let error = runtime.block_on(async {
+            reqwest::Client::builder()
+                .no_proxy()
+                .build()
+                .expect("build local test client")
+                .get(&endpoint)
+                .send()
+                .await
+                .expect("send local test request")
+                .error_for_status()
+                .expect_err("test response should be an HTTP error")
+        });
+        server.join().expect("join local test server");
+        assert!(error.url().is_some(), "fixture must carry the endpoint before redaction");
+
+        let OpaConfigError::Connection(error) = redact_opa_connection_error(error) else {
+            panic!("expected a redacted OPA connection error");
+        };
+        let rendered = error.to_string();
+        assert!(error.url().is_none());
+        for leaked in ["profile-user", "profile-token", "127.0.0.1", "private/opa", "query-secret"] {
+            assert!(!rendered.contains(leaked), "redacted error leaked {leaked}: {rendered}");
+        }
+    }
+
+    #[test]
+    fn test_is_allowed_rejection_log_redacts_sensitive_endpoint() {
+        let listener = TcpListener::bind("127.0.0.1:0").expect("bind local test listener");
+        let port = listener.local_addr().expect("read local test listener address").port();
+        let server = std::thread::spawn(move || {
+            let (mut stream, _) = listener.accept().expect("accept local test request");
+            let mut request = [0_u8; 1024];
+            let read = stream.read(&mut request).expect("read local test request");
+            assert!(read > 0, "local test client must send an HTTP request");
+            stream
+                .write_all(b"HTTP/1.1 500 Internal Server Error\r\nContent-Length: 0\r\nConnection: close\r\n\r\n")
+                .expect("write local test response");
+        });
+        let endpoint = format!("http://opa-user:opa-token@127.0.0.1:{port}/private/allow?access_token=query-secret");
+        let plugin = AuthZPlugin::new(Args {
+            url: endpoint,
+            auth_token: "authorization-secret".to_string(),
+        });
+        let groups = None;
+        let conditions = HashMap::new();
+        let claims = HashMap::new();
+        let args = PArgs {
+            account: "account",
+            groups: &groups,
+            action: crate::policy::action::Action::None,
+            bucket: "bucket",
+            conditions: &conditions,
+            is_owner: false,
+            object: "object",
+            claims: &claims,
+            deny_only: false,
+        };
+        let buffer = Arc::new(Mutex::new(Vec::new()));
+        let writer = TestWriter(Arc::clone(&buffer));
+        let subscriber = tracing_subscriber::fmt()
+            .with_ansi(false)
+            .without_time()
+            .with_writer(writer)
+            .finish();
+
+        tracing::subscriber::with_default(subscriber, || {
+            let runtime = tokio::runtime::Builder::new_current_thread()
+                .enable_all()
+                .build()
+                .expect("build tokio test runtime");
+            assert!(!runtime.block_on(plugin.is_allowed(&args)));
+        });
+        server.join().expect("join local test server");
+
+        let rendered = String::from_utf8(buffer.lock().expect("lock test log buffer").clone()).expect("decode test log");
+        assert!(!rendered.is_empty(), "OPA request failure did not reach the test log sink");
+        assert!(
+            rendered.contains("request_rejected"),
+            "OPA request failure did not retain its stable result category"
+        );
+        assert!(rendered.contains("500 Internal Server Error"));
+        for leaked in [
+            "opa-user",
+            "opa-token",
+            "127.0.0.1",
+            "private/allow",
+            "query-secret",
+            "authorization-secret",
+        ] {
+            assert!(!rendered.contains(leaked), "OPA request log leaked {leaked}: {rendered}");
+        }
+    }
 }